[Phase 4] Per-State Concurrency Limits for Batch Execution

[frankbria]

Summary

Add per-state concurrency limits to batch execution, allowing different parallelism levels for tasks in different states. Inspired by Symphony's max_concurrent_agents_by_state configuration.

Motivation

Currently cf work batch run --max-parallel N applies a single global concurrency limit. In practice, teams may want different limits based on task state or type — for example, allowing more agents on READY tasks but limiting how many IN_PROGRESS tasks run simultaneously to avoid resource contention.

This becomes more important in Phase 4 multi-agent scenarios where different agent roles may have different resource requirements.

Design (single session)

Changes to core/conductor.py

1. Add ConcurrencyConfig dataclass:

   @dataclass
   class ConcurrencyConfig:
       max_parallel: int = 4              # global limit
       by_status: dict[str, int] = {}     # per-status limits
       # e.g., {"READY": 3, "IN_PROGRESS": 2, "BLOCKED": 1}

2. Update _get_available_slots() to check both global and per-status limits:

   • Global slots = max_parallel - running_count
   • Per-status slots = by_status[status] - running_count_for_status
   • Available = min(global_slots, per_status_slots)

3. Update dispatch logic to count running tasks by status

Changes to cli/app.py

4. Add --max-parallel-by-status flag:

   cf work batch run --all-ready --max-parallel 4 --max-parallel-by-status "READY=3,IN_PROGRESS=2"

Changes to config

5. Support in AGENTS.md/CODEFRAME.md:

   batch:
     max_parallel: 4
     max_parallel_by_status:
       READY: 3
       IN_PROGRESS: 2

Acceptance Criteria

• Per-status limits respected during parallel batch execution
• Global limit still enforced as upper bound
• --max-parallel-by-status CLI flag works
• Config file loading works
• Status counts updated correctly as tasks transition
• Unspecified statuses fall back to global limit
• Unit tests for slot calculation with mixed status limits
• Integration test: batch with per-status limits

Dependencies

• No hard prerequisites
• Benefits from [Phase 4] Continuous Reconciliation During Batch Execution #403 (reconciliation) for accurate status tracking during execution

[traycerai]

Plan

Observations

• conductor.py uses @dataclass + field (already imported from dataclasses) throughout; BatchRun already holds max_parallel: int. The parallel execution path lives in _execute_parallel → _execute_group_parallel, which calls min(group_size, batch.max_parallel) to pick worker count.
• The CLI uses Typer (typer.Option) for all flags. The batch_run command in file:codeframe/cli/app.py passes max_parallel directly to conductor.start_batch().
• Config lives in file:codeframe/core/config.py as EnvironmentConfig (YAML dataclass). There is no batch section yet; the pattern used for nested config is shown by ContextConfig and AgentBudgetConfig.
• Task statuses are BACKLOG | READY | IN_PROGRESS | BLOCKED | FAILED | DONE | MERGED defined in file:codeframe/core/state_machine.py as TaskStatus.
• tasks.get(workspace, task_id) returns a Task whose .status is a TaskStatus. This is already available in _execute_single_task and _execute_group_parallel — no extra DB calls needed to look up the task's original status.

Approach

Introduce ConcurrencyConfig as a new @dataclass in conductor.py (matching the existing dataclass style). Thread it through BatchRun, start_batch(), and _execute_parallel. The per-status enforcement happens at the _execute_group_parallel call site by inspecting each task's initial status (the status when it enters the batch) and applying the per-status cap as an additional upper bound on max_workers. Expose the config via a new --max-parallel-by-status Typer option and as a BatchConfig section in EnvironmentConfig / config.yaml.

---

Implementation Steps

1. Add ConcurrencyConfig dataclass in file:codeframe/core/conductor.py

Place it immediately before the BatchStatus enum (around line 408), following the same @dataclass style as existing classes:

• Fields: max_parallel: int = 4 and by_status: dict[str, int] = field(default_factory=dict).
• Add a method get_limit_for_status(status: str) -> int that returns by_status[status] when the status key exists, or falls back to max_parallel.
• Add a method effective_workers(status: str, group_size: int, global_running: int) -> int that computes min(global_slots, per_status_slots, group_size) where:
  • global_slots = max_parallel - global_running
  • per_status_slots = get_limit_for_status(status) - running_count_for_status

2. Extend BatchRun dataclass in file:codeframe/core/conductor.py

Add a field concurrency: ConcurrencyConfig = field(default_factory=ConcurrencyConfig). This is the single source of truth for concurrency limits during the run. Keep max_parallel: int for backward compat / DB persistence — set it from concurrency.max_parallel at construction.

3. Update start_batch() signature and body in file:codeframe/core/conductor.py

• Add parameter concurrency_by_status: dict[str, int] | None = None.
• Construct a ConcurrencyConfig(max_parallel=max_parallel, by_status=concurrency_by_status or {}) and attach it to the BatchRun.

4. Add per-status running-count tracking in _execute_parallel() in file:codeframe/core/conductor.py

Introduce a threading.Lock-protected counter dict _running_by_status: dict[str, int] scoped to the batch execution. Before dispatching each task within _execute_group_parallel, look up the task's current status (tasks.get(workspace, task_id).status.value) and use it as the "initial status" key. Compute effective_workers using batch.concurrency.effective_workers(...) and pass this value to ThreadPoolExecutor(max_workers=...).

sequenceDiagram
    participant EP as _execute_parallel
    participant EGP as _execute_group_parallel
    participant CC as ConcurrencyConfig
    participant DB as SQLite/tasks

    EP->>DB: look up task statuses for group
    EP->>CC: effective_workers(status, group_size, global_running)
    CC-->>EP: capped worker count
    EP->>EGP: max_workers = capped count
    EGP->>DB: execute tasks via subprocess (ThreadPoolExecutor)
    EGP-->>EP: results dict
    EP->>EP: update _running_by_status counters

Key subtlety: Since all tasks in a single dependency group are dispatched together in _execute_group_parallel, the simplest correct strategy is to group them by their initial status and apply the cap per-status sub-group within the group. Tasks in a group that belong to different statuses should each have their own sub-pool of workers bounded by their status cap.

5. Update _execute_group_parallel() signature in file:codeframe/core/conductor.py

• Accept an optional concurrency: ConcurrencyConfig | None = None parameter.
• When concurrency is provided (and by_status is non-empty), partition the task group by each task's initial status, then for each status partition compute min(partition_size, concurrency.get_limit_for_status(status)) and run partitions with their own sub-ThreadPoolExecutor. The global max_parallel outer bound is honoured by the caller at the group level.
• When by_status is empty, behaviour is identical to today (no regression).

6. Add --max-parallel-by-status Typer option in file:codeframe/cli/app.py

Add to batch_run():

max_parallel_by_status: Optional[str] = typer.Option(
    None,
    "--max-parallel-by-status",
    help='Per-status concurrency limits, e.g. "READY=3,IN_PROGRESS=2"',
)

• Parse the string into dict[str, int] (split on ,, then =). Validate keys against TaskStatus values; raise a user-friendly error on bad input.
• Pass the resulting dict as concurrency_by_status to conductor.start_batch().

Also update the "Batch Execution Plan" print block to show per-status limits when set.

7. Add BatchConfig dataclass and wire into EnvironmentConfig in file:codeframe/core/config.py

Following the ContextConfig / AgentBudgetConfig pattern:

• New @dataclass BatchConfig with fields max_parallel: int = 4 and max_parallel_by_status: dict[str, int] = field(default_factory=dict).
• Add batch: BatchConfig = field(default_factory=BatchConfig) to EnvironmentConfig.
• Update EnvironmentConfig.from_dict() to deserialize the nested batch dict into a BatchConfig instance.

This enables config.yaml to support:

batch:
  max_parallel: 4
  max_parallel_by_status:
    READY: 3
    IN_PROGRESS: 2

8. Load BatchConfig from config.yaml in the CLI batch_run() command in file:codeframe/cli/app.py

After loading the workspace:

• Call load_environment_config(path) (already importable from codeframe.core.config).
• If config exists and has batch.max_parallel_by_status, merge it with the CLI-supplied values (CLI flag takes precedence over config file).
• Pass the merged concurrency_by_status to conductor.start_batch().

9. Validate per-status keys in start_batch() in file:codeframe/core/conductor.py

At the start of start_batch(), validate that all keys in concurrency_by_status are valid TaskStatus values. Raise ValueError with a clear message if not.

---

Summary of touched files

File	Changes
file:codeframe/core/conductor.py	Add ConcurrencyConfig, extend BatchRun, update start_batch(), _execute_parallel(), _execute_group_parallel()
file:codeframe/core/config.py	Add BatchConfig, add batch field to EnvironmentConfig, update from_dict()
file:codeframe/cli/app.py	Add --max-parallel-by-status option to batch_run, parse + merge config, forward to start_batch()

Import In IDE

🤖 Prompt for AI Agents

## Observations

- `conductor.py` uses `@dataclass` + `field` (already imported from `dataclasses`) throughout; `BatchRun` already holds `max_parallel: int`. The parallel execution path lives in `_execute_parallel` → `_execute_group_parallel`, which calls `min(group_size, batch.max_parallel)` to pick worker count.
- The CLI uses **Typer** (`typer.Option`) for all flags. The `batch_run` command in `file:codeframe/cli/app.py` passes `max_parallel` directly to `conductor.start_batch()`.
- Config lives in `file:codeframe/core/config.py` as `EnvironmentConfig` (YAML dataclass). There is no `batch` section yet; the pattern used for nested config is shown by `ContextConfig` and `AgentBudgetConfig`.
- Task statuses are `BACKLOG | READY | IN_PROGRESS | BLOCKED | FAILED | DONE | MERGED` defined in `file:codeframe/core/state_machine.py` as `TaskStatus`.
- `tasks.get(workspace, task_id)` returns a `Task` whose `.status` is a `TaskStatus`. This is already available in `_execute_single_task` and `_execute_group_parallel` — no extra DB calls needed to look up the task's original status.

## Approach

Introduce `ConcurrencyConfig` as a new `@dataclass` in `conductor.py` (matching the existing dataclass style). Thread it through `BatchRun`, `start_batch()`, and `_execute_parallel`. The per-status enforcement happens at the `_execute_group_parallel` call site by inspecting each task's **initial** status (the status when it enters the batch) and applying the per-status cap as an additional upper bound on `max_workers`. Expose the config via a new `--max-parallel-by-status` Typer option and as a `BatchConfig` section in `EnvironmentConfig` / `config.yaml`.

---

## Implementation Steps

### 1. Add `ConcurrencyConfig` dataclass in `file:codeframe/core/conductor.py`

Place it immediately before the `BatchStatus` enum (around line 408), following the same `@dataclass` style as existing classes:

- Fields: `max_parallel: int = 4` and `by_status: dict[str, int] = field(default_factory=dict)`.
- Add a method `get_limit_for_status(status: str) -> int` that returns `by_status[status]` when the status key exists, or falls back to `max_parallel`.
- Add a method `effective_workers(status: str, group_size: int, global_running: int) -> int` that computes `min(global_slots, per_status_slots, group_size)` where:
  - `global_slots = max_parallel - global_running`
  - `per_status_slots = get_limit_for_status(status) - running_count_for_status`

### 2. Extend `BatchRun` dataclass in `file:codeframe/core/conductor.py`

Add a field `concurrency: ConcurrencyConfig = field(default_factory=ConcurrencyConfig)`. This is the single source of truth for concurrency limits during the run. Keep `max_parallel: int` for backward compat / DB persistence — set it from `concurrency.max_parallel` at construction.

### 3. Update `start_batch()` signature and body in `file:codeframe/core/conductor.py`

- Add parameter `concurrency_by_status: dict[str, int] | None = None`.
- Construct a `ConcurrencyConfig(max_parallel=max_parallel, by_status=concurrency_by_status or {})` and attach it to the `BatchRun`.

### 4. Add per-status running-count tracking in `_execute_parallel()` in `file:codeframe/core/conductor.py`

Introduce a `threading.Lock`-protected counter dict `_running_by_status: dict[str, int]` scoped to the batch execution. Before dispatching each task within `_execute_group_parallel`, look up the task's current status (`tasks.get(workspace, task_id).status.value`) and use it as the "initial status" key. Compute `effective_workers` using `batch.concurrency.effective_workers(...)` and pass this value to `ThreadPoolExecutor(max_workers=...)`.

```
sequenceDiagram
    participant EP as _execute_parallel
    participant EGP as _execute_group_parallel
    participant CC as ConcurrencyConfig
    participant DB as SQLite/tasks

    EP->>DB: look up task statuses for group
    EP->>CC: effective_workers(status, group_size, global_running)
    CC-->>EP: capped worker count
    EP->>EGP: max_workers = capped count
    EGP->>DB: execute tasks via subprocess (ThreadPoolExecutor)
    EGP-->>EP: results dict
    EP->>EP: update _running_by_status counters
```

> **Key subtlety**: Since all tasks in a single dependency group are dispatched together in `_execute_group_parallel`, the simplest correct strategy is to **group them by their initial status** and apply the cap per-status sub-group within the group. Tasks in a group that belong to different statuses should each have their own sub-pool of workers bounded by their status cap.

### 5. Update `_execute_group_parallel()` signature in `file:codeframe/core/conductor.py`

- Accept an optional `concurrency: ConcurrencyConfig | None = None` parameter.
- When `concurrency` is provided (and `by_status` is non-empty), partition the task group by each task's initial status, then for each status partition compute `min(partition_size, concurrency.get_limit_for_status(status))` and run partitions with their own sub-`ThreadPoolExecutor`. The global `max_parallel` outer bound is honoured by the caller at the group level.
- When `by_status` is empty, behaviour is identical to today (no regression).

### 6. Add `--max-parallel-by-status` Typer option in `file:codeframe/cli/app.py`

Add to `batch_run()`:

```
max_parallel_by_status: Optional[str] = typer.Option(
    None,
    "--max-parallel-by-status",
    help='Per-status concurrency limits, e.g. "READY=3,IN_PROGRESS=2"',
)
```

- Parse the string into `dict[str, int]` (split on `,`, then `=`). Validate keys against `TaskStatus` values; raise a user-friendly error on bad input.
- Pass the resulting dict as `concurrency_by_status` to `conductor.start_batch()`.

Also update the "Batch Execution Plan" print block to show per-status limits when set.

### 7. Add `BatchConfig` dataclass and wire into `EnvironmentConfig` in `file:codeframe/core/config.py`

Following the `ContextConfig` / `AgentBudgetConfig` pattern:

- New `@dataclass BatchConfig` with fields `max_parallel: int = 4` and `max_parallel_by_status: dict[str, int] = field(default_factory=dict)`.
- Add `batch: BatchConfig = field(default_factory=BatchConfig)` to `EnvironmentConfig`.
- Update `EnvironmentConfig.from_dict()` to deserialize the nested `batch` dict into a `BatchConfig` instance.

This enables `config.yaml` to support:

```yaml
batch:
  max_parallel: 4
  max_parallel_by_status:
    READY: 3
    IN_PROGRESS: 2
```

### 8. Load `BatchConfig` from `config.yaml` in the CLI `batch_run()` command in `file:codeframe/cli/app.py`

After loading the workspace:

- Call `load_environment_config(path)` (already importable from `codeframe.core.config`).
- If config exists and has `batch.max_parallel_by_status`, merge it with the CLI-supplied values (CLI flag takes precedence over config file).
- Pass the merged `concurrency_by_status` to `conductor.start_batch()`.

### 9. Validate per-status keys in `start_batch()` in `file:codeframe/core/conductor.py`

At the start of `start_batch()`, validate that all keys in `concurrency_by_status` are valid `TaskStatus` values. Raise `ValueError` with a clear message if not.

---

## Summary of touched files

| File | Changes |
|---|---|
| `file:codeframe/core/conductor.py` | Add `ConcurrencyConfig`, extend `BatchRun`, update `start_batch()`, `_execute_parallel()`, `_execute_group_parallel()` |
| `file:codeframe/core/config.py` | Add `BatchConfig`, add `batch` field to `EnvironmentConfig`, update `from_dict()` |
| `file:codeframe/cli/app.py` | Add `--max-parallel-by-status` option to `batch_run`, parse + merge config, forward to `start_batch()` |

---

Execution Information

Branch: main
Commit: 705b160

---

💡 Tips

Supported Commands (Inside Comments)

• Use @traycerai generate to iterate on the previous version of the implementation plan.

Supported Commands (Inside Description)

• Add @traycerai ignore anywhere in the ticket description to prevent this ticket from being processed.
• Add @traycerai branch:<branch-name> anywhere in the ticket description to specify the target branch for the implementation plan.

Community

• Join our Discord Community to get help, request features, and share feedback.
• Follow us on X/Twitter for updates and announcements.

[coderabbitai]

🔗 Related PRs

#363 - feat: wire ReactAgent into runtime with --engine react CLI flag [merged]

---

📝 Issue Planner

_{Check the box below or use the @coderabbitai plan command to generate an implementation plan and prompts that you can use with your favorite coding assistant.}

• Create Plan

---

🧪 Issue enrichment is currently in open beta.

You can configure auto-planning by selecting labels in the issue_enrichment configuration.

To disable automatic issue enrichment, add the following to your .coderabbit.yaml:

issue_enrichment:
  auto_enrich:
    enabled: false

💬 Have feedback or questions? Drop into our discord!