feat(core): per-state concurrency limits for batch execution - #440
Conversation
Add ConcurrencyConfig with per-status parallelism caps, allowing different
concurrency limits for tasks in different states during batch execution.
- ConcurrencyConfig dataclass with max_parallel, by_status, get_limit_for_status(),
effective_workers() computing min of global/per-status/group-size limits
- parse_concurrency_by_status() for CLI string parsing ("READY=3,IN_PROGRESS=2")
- BatchRun.concurrency field threaded through start_batch()
- BatchConfig in EnvironmentConfig for config.yaml persistence
- 21 unit tests covering config, slot calculation, parsing, and start_batch integration
Closes #407
|
No actionable comments were generated in the recent review. 🎉 ℹ️ Recent review info⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: 📒 Files selected for processing (1)
WalkthroughAdds per-state concurrency controls: a new Changes
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~25 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
📝 Coding Plan
Comment |
Code Review — Per-State Concurrency Limits (#407)Good infrastructure work overall. The dataclasses are clean, the test suite is solid, and the config serialization roundtrip is handled correctly. A few issues to address before this is functionally complete: Critical:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (2)
tests/core/test_concurrency_config.py (2)
123-147: Test validates integration but not actual enforcement.This test correctly verifies that
start_batchacceptsconcurrency_by_statusand stores it in the batch. However, since_execute_serialis mocked, it doesn't verify that the limits are actually enforced during execution.Given that
effective_workers()isn't currently used in the execution path (see conductor.py review), consider adding an integration test that verifies per-status limits are respected when execution is complete.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/core/test_concurrency_config.py` around lines 123 - 147, The test TestStartBatchConcurrency only asserts that start_batch stores concurrency_by_status but doesn't exercise enforcement because _execute_serial is mocked; update or add an integration-style test that does not mock _execute_serial (and instead lets conductor.start_batch perform real execution) to verify that per-status limits are enforced at runtime by exercising effective_workers() in the execution path—specifically call start_batch with concurrency_by_status={"READY":2} and real task mocks that transition statuses so you can assert no more than 2 tasks run concurrently in READY state and overall behavior (check batch.concurrency.by_status and runtime concurrency counts) after execution completes; reference start_batch, _execute_serial, and effective_workers to find where to remove the mock and assert enforcement.
176-186: Consider adding a test for non-integer value parsing.The tests cover invalid status names and invalid format (
:instead of=), but don't cover the case where the value isn't a valid integer.Suggested additional test case
def test_parse_invalid_value_raises(self) -> None: from codeframe.core.conductor import parse_concurrency_by_status with pytest.raises(ValueError): parse_concurrency_by_status("READY=abc")🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/core/test_concurrency_config.py` around lines 176 - 186, Add a test to assert parse_concurrency_by_status raises on non-integer values: create a new test (e.g., test_parse_invalid_value_raises) that calls codeframe.core.conductor.parse_concurrency_by_status with a string like "READY=abc" and wraps it in pytest.raises(ValueError); this complements existing tests for invalid status names and formats by verifying value parsing errors for parse_concurrency_by_status.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Inline comments:
In `@codeframe/core/conductor.py`:
- Line 532: The new concurrency: ConcurrencyConfig field on BatchRun is never
persisted or restored, so update persistence to serialize and deserialize it:
modify _save_batch to include the batch.concurrency (e.g., as a JSON/text column
or separate columns) when writing the batch row, and modify _row_to_batch to
parse that stored value back into a ConcurrencyConfig instance and assign it to
BatchRun.concurrency; ensure get_batch() and resume_batch() will therefore
return the actual saved ConcurrencyConfig rather than the default. Use the
existing BatchRun, ConcurrencyConfig, _save_batch, _row_to_batch and
resume_batch symbols to locate the changes.
- Around line 455-481: The parse_concurrency_by_status function currently lets
int(val.strip()) raise an opaque ValueError for non-integer values (e.g.,
"READY=abc"); update the parsing loop in parse_concurrency_by_status to wrap the
conversion in a try/except, catch ValueError from int(), and raise a new
ValueError that clearly states which STATUS and value failed to parse (include
the raw val and the STATUS key), e.g. "Invalid concurrency for READY: 'abc'
(must be an integer)"; keep the rest of the validation (status membership)
unchanged.
- Around line 417-452: The parallel executor is ignoring batch.concurrency; in
_execute_parallel replace the simple effective_workers = min(group_size,
batch.max_parallel) with a call to batch.concurrency.effective_workers(...) so
per-status limits are applied: gather the list of statuses for the tasks in the
current group (e.g., [t.status for t in group]), pass group_size and the current
global_running count into batch.concurrency.effective_workers(statuses=statuses,
group_size=group_size, global_running=global_running), and use that result (with
a safe fallback to batch.max_parallel if batch.concurrency is None) as the
effective worker count.
---
Nitpick comments:
In `@tests/core/test_concurrency_config.py`:
- Around line 123-147: The test TestStartBatchConcurrency only asserts that
start_batch stores concurrency_by_status but doesn't exercise enforcement
because _execute_serial is mocked; update or add an integration-style test that
does not mock _execute_serial (and instead lets conductor.start_batch perform
real execution) to verify that per-status limits are enforced at runtime by
exercising effective_workers() in the execution path—specifically call
start_batch with concurrency_by_status={"READY":2} and real task mocks that
transition statuses so you can assert no more than 2 tasks run concurrently in
READY state and overall behavior (check batch.concurrency.by_status and runtime
concurrency counts) after execution completes; reference start_batch,
_execute_serial, and effective_workers to find where to remove the mock and
assert enforcement.
- Around line 176-186: Add a test to assert parse_concurrency_by_status raises
on non-integer values: create a new test (e.g., test_parse_invalid_value_raises)
that calls codeframe.core.conductor.parse_concurrency_by_status with a string
like "READY=abc" and wraps it in pytest.raises(ValueError); this complements
existing tests for invalid status names and formats by verifying value parsing
errors for parse_concurrency_by_status.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 4937c353-84c0-4a3a-b870-f074276efffc
📒 Files selected for processing (3)
codeframe/core/conductor.pycodeframe/core/config.pytests/core/test_concurrency_config.py
| engine: str = "react" | ||
| stall_timeout_s: int = 300 | ||
| stall_action: str = "blocker" | ||
| concurrency: ConcurrencyConfig = field(default_factory=ConcurrencyConfig) |
There was a problem hiding this comment.
concurrency field is not persisted to database.
The concurrency field is added to BatchRun, but _save_batch (line 1957) and _row_to_batch (line 1997) don't handle it. This means:
- Concurrency config is lost when a batch is resumed
get_batch()returns aBatchRunwith defaultConcurrencyConfigregardless of what was originally set
This may be acceptable if the feature is only used for the initial execution, but it's worth documenting or fixing for consistency with resume_batch().
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed.
In `@codeframe/core/conductor.py` at line 532, The new concurrency:
ConcurrencyConfig field on BatchRun is never persisted or restored, so update
persistence to serialize and deserialize it: modify _save_batch to include the
batch.concurrency (e.g., as a JSON/text column or separate columns) when writing
the batch row, and modify _row_to_batch to parse that stored value back into a
ConcurrencyConfig instance and assign it to BatchRun.concurrency; ensure
get_batch() and resume_batch() will therefore return the actual saved
ConcurrencyConfig rather than the default. Use the existing BatchRun,
ConcurrencyConfig, _save_batch, _row_to_batch and resume_batch symbols to locate
the changes.
…l dispatch - Use per-status limits when computing effective_workers for group execution - Improve error message for non-integer values in parse_concurrency_by_status
|
Follow-up review of commit 18db151. The wiring fix is good - effective_workers() is now correctly called in _execute_parallel(). Remaining items: (1) SQLite persistence gap - concurrency field not serialized in _save_batch/_row_to_batch, batch resume will silently drop per-status limits; (2) global_running=0 hardcoded, so global max_parallel cap is not enforced; (3) non-positive values accepted by parse_concurrency_by_status without error; (4) BatchConfig to ConcurrencyConfig bridge still absent. The persistence gap is the most likely real user-visible bug. Happy to approve once addressed or documented as known limitation. |
|
CORRECTION - Full review follows below. Please disregard the previous abbreviated comment. |
|
Full Follow-up Review - Commit 18db151 Thanks for the quick follow-up commit. Acknowledging what changed and what still needs attention. Critical issue resolved: The second commit (18db151) wired effective_workers() into _execute_parallel(). The per-status branch now correctly queries task statuses at dispatch time and calls batch.concurrency.effective_workers(). That was the core gap - good fix. Remaining issues:
Summary: The wiring fix is the right move. The persistence gap (item 1) is the remaining issue most likely to cause a real user-visible bug (batch resume dropping limits). The other items are lower priority. Happy to approve once the persistence gap is addressed or documented as a known limitation. |
1 participant
Summary
Implements #407: Per-State Concurrency Limits for Batch Execution
Adds per-status parallelism caps to batch execution, allowing different concurrency limits for tasks in different states (e.g., allow 3 READY tasks but only 1 IN_PROGRESS).
max_parallel,by_status,get_limit_for_status(), andeffective_workers()computing min of global/per-status/group-size limitsstart_batch()Acceptance Criteria
Test Plan
Implementation Notes
--max-parallel-by-statusflag: Not wired in this PR since the infrastructure is ready; can be added in a follow-up.concurrency_by_statusparam is available onstart_batch().Closes #407
Summary by CodeRabbit
New Features
Tests