<!-- task=dispatcher_parallel_limit plan=0 tag=API -->

# Configurable Global Dispatcher Parallel Limit

## For the Implementing Agent

> **[IMPLEMENTING AGENT — READ FIRST]** Filling implementation-owned sections in `CODE_REVIEW-*-G??.md` is mandatory. Run every verification command, paste actual notes/output into the review file, keep both active files in place, and report ready for review. Finalization is code-review-skill only. If blocked, record only the exact blocker, attempted commands/output, and resume condition in implementation-owned evidence fields. Do not ask the user, call user-input tools, create control-plane stop files, classify the next state, archive logs, or write `complete.log`.

## Background

The Python agent-task dispatcher currently admits every dependency-ready task whose workspace write claim is disjoint, so an operator cannot bound total concurrent worker, self-check, and official-review attempts. A small invocation-scoped global limit is needed to reduce host/provider pressure while preserving the current unlimited behavior by default. Capacity waiting must integrate with the dispatcher's restricted rescan contract so a deferred ready task is not lost after another attempt changes stage.

## Analysis

### Files Read

- `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`
- `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`
- `agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md`
- `agent-test/local/rules.md`
- `agent-test/local/testing-smoke.md`
- `agent-roadmap/current.md`
- `agent-roadmap/priority-queue.md`
- `agent-roadmap/phase/automation-runtime-bridge/PHASE.md`
- `agent-roadmap/phase/automation-runtime-bridge/milestones/iop-agent-cli-runtime.md`

### SDD Criteria

Not applicable. This is a non-roadmap compatibility patch for the current Python dispatcher and does not complete an `IOP Agent CLI Runtime` Milestone Task.

### Verification Context

- Handoff: no formal `verification_context` was supplied. Repository-native source, tests, rules, roadmap context, and read-only baseline commands were used.
- Environment: local checkout `/config/workspace/iop`, branch `dev`, HEAD `0f4619ba`, clean against `origin/dev`.
- Source paths: the three dispatcher source/test/skill paths listed under `Files Read`.
- Baseline commands and results:
  - `python3 -m py_compile agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py` passed.
  - `python3 agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py --help` passed and confirmed that no numeric parallel-limit option exists.
  - `python3 agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py ReviewSchedulingTest WriteSetTest` passed 16 tests.
  - `python3 agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py DispatcherConvergenceSimulationTest BlockerDrainTest` passed 8 tests.
  - `python3 -m unittest discover -s agent-ops/skills/project/orchestrate-agent-task-loop/tests -p 'test_*.py'` passed 287 tests.
- Preconditions: Python standard library only; no dependency manifest change, credentials, provider CLI, network, deployment, or long-running external runtime is required.
- Constraints: preserve `0=unlimited`, review-before-worker ordering, disjoint write-claim admission, initial/verified-completion full-scan discipline, task-only reclassification after an ordinary attempt, and non-terminal adoption of external active attempts.
- Gap: current tests prove unlimited review selection and write-claim behavior but do not cover a global cap, capacity-deferred re-admission, external-active slot accounting, or negative CLI input.
- Confidence: high; the admission and lifecycle paths are directly covered by deterministic fake-runner tests.
- External Verification Preflight: not applicable because required verification stays inside this checkout and must not invoke real providers.

### Test Coverage Gaps

- Default unlimited admission: covered by `ReviewSchedulingTest.test_all_ready_reviews_are_selected_without_numeric_cap`; retain it as a compatibility regression.
- Positive global cap across worker/self-check/review: not covered; add normal cases for limits `1` and `2`.
- Capacity wait without a premature new write claim, while preserving an already-owned lifecycle claim: not covered; add assertions against `StateStore.write_claims`.
- Slot release and re-admission without a forbidden full rescan: not covered; add an async convergence regression.
- Review-preflight failure freeing a slot for an independent worker: existing unlimited drain coverage exists, but capped behavior is not covered.
- Restart/external active attempt consuming capacity: external-active detection is covered, but its interaction with a configured cap is not covered.
- Negative CLI value: not covered; add parser boundary coverage.

### Symbol References

None. No symbol is renamed or removed. The new option and internal capacity parameter are additive.

### Split Judgment

Keep one plan. CLI parsing, active-slot calculation, capacity-wait bookkeeping, claim timing, review-preflight refill, tests, and skill text form one compact scheduling invariant; splitting them would permit an intermediate state that either loses ready tasks or documents behavior the runtime does not enforce.

### Scope Rationale

Include only the current Python dispatcher entry point, its existing test module, and its project skill documentation. Exclude provider output validation/filtering, tunnel buffers, provider-specific quota/concurrency rules, Go `iop-agent` configuration or parity work, persistent config schemas, deployment, and live provider smoke because the requested control is an invocation-scoped global scheduler cap.

### Final Routing

- `evaluation_mode`: `first-pass`
- `finalizer`: `finalize-task-policy.sh` (`pair`)
- Build closures: `scope_closed=true`, `context_closed=true`, `verification_closed=true`, `evidence_trusted=true`, `ownership_closed=true`, `decision_closed=true`
- Build grade scores: scope `1`, state/concurrency `2`, blast/irreversibility `1`, evidence/diagnosis `0`, verification `1`; grade `G05`
- Build base/route: `local-fit` / `local`; canonical filename `PLAN-local-G05.md`
- Review closures: `scope_closed=true`, `context_closed=true`, `verification_closed=true`, `evidence_trusted=true`, `ownership_closed=true`, `decision_closed=true`
- Review grade scores: scope `1`, state/concurrency `2`, blast/irreversibility `1`, evidence/diagnosis `0`, verification `1`; grade `G05`
- Review route: `official-review`, `cloud`, Codex `gpt-5.6-sol` xhigh; canonical filename `CODE_REVIEW-cloud-G05.md`
- `large_indivisible_context=false`
- Positive loop risks: `temporal_state`, `concurrent_consistency`; count `2`
- Recovery signals: `review_rework_count=0`, `evidence_integrity_failure=false`
- Capability-gap evidence: none; all implementation and verification context is locally closed.

## Implementation Checklist

- [ ] Add the non-negative `--max-parallel` CLI contract with `0` as the backward-compatible unlimited default.
- [ ] Enforce the global cap across worker, self-check, official review, and adopted external-active attempts while preserving review ordering, claim safety, and capacity-wait re-admission.
- [ ] Add deterministic unit and async regressions for unlimited, limits `1`/`2`, mixed roles, slot refill, external-active accounting, claim timing, review-preflight failure, and negative input.
- [ ] Update the dispatcher skill inputs, concurrency contract, examples, and verification checklist for the global limit.
- [ ] Run the focused and full final verification commands and record fresh output.
- [ ] Fill implementation-owned sections in CODE_REVIEW-*-G??.md with actual implementation notes and verification output.

### [API-1] Add invocation-scoped global admission capacity

#### Problem

`select_dispatch_candidates` at `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py:5662-5755` selects and claims every disjoint ready task. `dispatch_with_store` at lines `5827-5838`, `6041-6100`, `6269-6272`, and `6353-6473` has no capacity-wait state and creates one asyncio task per selected candidate. `parse_args` at lines `6544-6555` exposes no numeric limit. Merely truncating `candidates` would strand deferred tasks because an ordinary attempt completion narrows `candidate_scope` to `finished_names`.

#### Solution

Add `--max-parallel` as a non-negative integer with default `0`; `0` means unlimited and positive values cap the union of current `running` tasks and same-workspace `live_external_processes`. Use `getattr(args, "max_parallel", 0)` inside the scheduler so existing direct test namespaces remain compatible.

Extend candidate admission with an optional available-slot budget. Preserve review-before-worker order and validate write sets/conflicts before admission, but acquire or replace a new task's write claim only when that task receives a slot. A task that already owns a lifecycle claim keeps it while capacity-deferred. Return a stable capacity-wait reason for otherwise-ready tasks.

Maintain a task-name set for candidates deferred only by capacity. After an attempt ends without `complete.log`, set the next `candidate_scope` to `finished_names` plus that set, allowing slot refill from the existing task snapshot without a complete rescan. Rebuild the set after each admission pass; a dependency, blocker, or write-claim defer must be removed from the capacity set and rely on the existing lifecycle event that normally makes it eligible again.

If capped review candidates fail the shared-state preflight, keep their task-local blocker/claim semantics, then fill the now-unused admission slots from capacity-deferred independent candidates before creating futures. Build the quota/admission batch snapshot only from the final candidates that will actually launch. Apply the same cap in dry-run as a stateless admission-wave preview. Never terminate excess already-active attempts when a restarted dispatcher is invoked with a smaller limit; admit no new attempt until occupied slots fall below the limit.

Before:

```python
# agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py:5662-5679
def select_dispatch_candidates(
    store: StateStore,
    ready: list[tuple[Task, str]],
    *,
    persist: bool,
) -> tuple[...]:
    ready_reviews = [(task, stage) for task, stage in ready if stage == "review"]
    ready_workers = [(task, stage) for task, stage in ready if stage in {"worker", "selfcheck"}]
    ordered = ready_reviews + ready_workers
    ...
    for task, stage in ordered:
```

```python
# agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py:6353-6473
candidates, deferred, _ = select_dispatch_candidates(
    store,
    ready,
    persist=True,
)
...
for task, stage in candidates:
    ...
    running[task.name] = future
...
if running:
    await asyncio.wait(running.values(), return_when=asyncio.FIRST_COMPLETED)
```

```python
# agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py:6544-6555
def parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description=__doc__)
    parser.add_argument("--workspace", default=".", help="repository root (default: current directory)")
    parser.add_argument("--task-group", help="run only agent-task/<task_group>")
    parser.add_argument("--dry-run", action="store_true", help="classify and print without launching CLIs")
    parser.add_argument("--retry-blocked", action="store_true", help="clear dispatcher-local blocked state")
```

After:

```python
def non_negative_int(raw: str) -> int:
    value = int(raw)
    if value < 0:
        raise argparse.ArgumentTypeError("must be >= 0")
    return value


def select_dispatch_candidates(
    store: StateStore,
    ready: list[tuple[Task, str]],
    *,
    persist: bool,
    available_slots: int | None = None,
) -> tuple[...]:
    ...
    # Validate claim safety first. When no slot remains, defer without
    # adding this task to the persisted claim snapshot.
```

```python
max_parallel = getattr(args, "max_parallel", 0)
capacity_waiting: set[str] = set()
...
occupied = len(set(running) | set(live_external_processes))
available_slots = (
    None if max_parallel == 0 else max(0, max_parallel - occupied)
)
...
# Re-admit capacity_waiting from task_cache when any slot is returned,
# without broadening the full-scan triggers.
```

#### Modified Files and Checklist

- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`: add argument validation and default.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`: add slot calculation, capacity-only waiting state, slot refill, and external-active accounting.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`: preserve claim ownership and review-preflight drain behavior.

#### Test Strategy

Write regression tests in API-2. Do not invoke real provider CLIs; use the existing `Task`, `StateStore`, fake worker/self-check/review coroutines, and subprocess-deny patterns.

#### Verification

Run:

```bash
python3 -m py_compile agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py
python3 agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py --help | rg --fixed-strings -- '--max-parallel MAX_PARALLEL'
```

Expected: compilation succeeds and help lists the new option.

### [API-2] Add capacity, ordering, and lifecycle regressions

#### Problem

`ReviewSchedulingTest` at `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py:4863-4939` asserts only unlimited selection. `WriteSetTest` at lines `5180-5345` covers claim collision and stateless preview but not capacity-deferred claims. `DispatcherConvergenceSimulationTest` at lines `7341-7528` proves unrestricted parallel convergence but not a bounded admission wave or task-only re-admission.

#### Solution

Keep the existing unlimited test and add focused candidate-selection cases showing `available_slots=None` (or the default) selects every disjoint candidate, limit `2` keeps review priority across mixed roles, newly capacity-deferred tasks do not appear in persistent write claims, and an already-owned lifecycle claim remains intact.

Add async scheduler cases with fake agents:

- limit `1` never exceeds one active attempt and starts a capacity-deferred sibling after the first task's ordinary stage completion without calling a forbidden full scan;
- limit `2` never exceeds two attempts across mixed worker/self-check/review roles and refills a returned slot;
- capped dry-run previews only the first admission wave, reports overflow as waiting, and leaves state unchanged;
- a same-workspace external-active locator consumes a slot;
- a failed capped review preflight does not consume its released runtime slot and an independent worker still drains;
- default `0` retains the existing unrestricted convergence result;
- `--max-parallel -1` is rejected by argparse with exit code `2`.

All concurrency observations must use deterministic events/barriers and active counters with short bounded `asyncio.wait_for` only inside tests.

Before:

```python
# agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py:4892-4901
selected, deferred, reason = dispatch.select_dispatch_candidates(
    store,
    ready,
    persist=False,
)
...
self.assertEqual(selected, ready)
self.assertEqual(deferred, [])
```

After:

```python
selected, deferred, _ = dispatch.select_dispatch_candidates(
    store,
    ready,
    persist=True,
    available_slots=2,
)
self.assertEqual([stage for _, stage in selected], ["review", "review"])
self.assertTrue(all(task.name not in store.data["write_claims"] for task, _, _ in newly_deferred))
```

#### Modified Files and Checklist

- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`: retain default-unlimited coverage and add parser/candidate boundary cases.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`: add deterministic capped convergence and slot-refill tests.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`: assert external-active accounting and capped review-preflight drain behavior.

#### Test Strategy

Write all listed tests in the existing module. The assertion goals are: maximum observed active count never exceeds the configured positive limit; `0` preserves current behavior; review ordering remains deterministic; a newly capacity-deferred task is not persisted as a blocker or new write claim; an existing lifecycle owner retains its claim; and slot return re-admits an already-scanned task.

#### Verification

Run:

```bash
python3 agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py ParallelLimitSchedulingTest ReviewSchedulingTest WriteSetTest BlockerDrainTest DispatcherConvergenceSimulationTest
```

Expected: every focused test passes with no real subprocess/provider invocation.

### [API-3] Document the global limit without changing provider policies

#### Problem

`agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md:51-56` has no parallel-limit input, lines `80-90` describe provider and claim concurrency without a global cap, lines `182-203` show only unlimited invocations, and line `230` states official reviews run without a numeric limit.

#### Solution

Document `max_parallel` as an optional invocation-scoped global cap, with `0` as unlimited. Clarify that it applies after dependency/write-claim eligibility across worker, self-check, and review roles; counts adopted same-workspace active attempts; and does not replace provider-specific limits. Update examples to show `--max-parallel 2`, and revise “no numeric limit” wording to mean “no separate review-only limit, subject to the configured global cap.”

Before:

```markdown
<!-- agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md:82-85 -->
- Pi `ornith:35b`: 3.
- agy: 1.
- Official Codex review: no separate numeric limit.
- Run worker/self-check and official review in parallel only when ...
```

After:

```markdown
- Global dispatcher limit: `max_parallel=0` is unlimited; a positive value
  caps all active worker, self-check, and official-review attempts.
- Pi `ornith:35b`: 3.
- agy: 1.
- Official Codex review: no separate review-only numeric limit; it remains
  subject to the global dispatcher limit.
```

#### Modified Files and Checklist

- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md`: add the input and invocation example.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md`: align concurrency and verification wording with the global cap.

#### Test Strategy

No separate documentation test file. Use deterministic `rg` checks plus the runtime tests from API-2.

#### Verification

Run:

```bash
rg -n --sort path --fixed-strings 'max_parallel' agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md
rg -n --sort path --fixed-strings -- '--max-parallel' agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md
```

Expected: the input/contract and CLI example are both present.

## Modified Files Summary

| File | Items |
|---|---|
| `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py` | API-1 |
| `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py` | API-2 |
| `agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md` | API-3 |
| `agent-task/dispatcher_parallel_limit/CODE_REVIEW-cloud-G05.md` | API-1, API-2, API-3 |

## Final Verification

Run from `/config/workspace/iop`:

```bash
python3 -m py_compile agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py
python3 agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py --help | rg --fixed-strings -- '--max-parallel MAX_PARALLEL'
python3 agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py ParallelLimitSchedulingTest ReviewSchedulingTest WriteSetTest BlockerDrainTest DispatcherConvergenceSimulationTest
python3 -m unittest discover -s agent-ops/skills/project/orchestrate-agent-task-loop/tests -p 'test_*.py'
rg -n --sort path --fixed-strings 'max_parallel' agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md
rg -n --sort path --fixed-strings -- '--max-parallel' agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md
git diff --check
```

Expected: compilation and help checks pass; focused fake-runner regressions pass; the full dispatcher suite passes without a real provider process; documentation checks find the input and example; `git diff --check` prints nothing. Python test cache is not accepted as verification evidence; the unittest commands execute fresh processes.

After completing all code changes, fill implementation-owned sections in `CODE_REVIEW-*-G??.md`.
