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

# Configurable Workspace-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. Add an invocation-scoped physical-workspace limit that preserves unlimited behavior by default. Capacity waiting must preserve the restricted rescan contract, task lifecycle claims, and non-terminal handling of live attempts adopted from any task group.

## Archive Evidence Snapshot

- Prior planning-only pair: `agent-task/dispatcher_parallel_limit/plan_local_G05_0.log` and `agent-task/dispatcher_parallel_limit/code_review_cloud_G05_0.log`.
- Prior state: no implementation, verification output, review verdict, or runtime execution was recorded.
- Replan corrections: count same-workspace live attempts outside `--task-group`, refresh occupancy after finished futures clear active state, keep capacity-only external waits non-terminal, and define review-preflight slot refill precisely.
- The facts needed for implementation are reproduced below; do not reread the prior logs unless exact draft comparison is required.

## 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`, source unchanged against `origin/dev`; only this task's planning artifacts exist.
- Baseline 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 showed no numeric parallel-limit option.
  - `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 change, credentials, provider CLI, network, deployment, or external runtime is required.
- Constraints: `0=unlimited`; positive limits count one task-stage attempt per task across worker/self-check/review; the physical-workspace count is not narrowed by `--task-group`; internal stream-pump/quota-probe coroutines are not additional slots; reviews remain ordered before worker/self-check candidates; write claims remain workspace-global; full scans remain limited to initial entry and verified completion.
- Gaps: no current test covers a global cap, task-only capacity re-admission, cross-group live occupancy, capacity-only exit semantics, stale occupancy after a future ends, capped review-preflight refill, or invalid numeric input.
- Confidence: high; deterministic fake-runner tests can exercise the real admission loop without a provider.
- External Verification Preflight: not applicable because verification stays inside this checkout and must deny real provider subprocesses.

### Test Coverage Gaps

- Default unlimited selection: covered; retain as a compatibility regression.
- Positive limits `1` and `2` across mixed roles: not covered.
- Newly capacity-deferred task not acquiring a claim, while an existing lifecycle owner retains its claim: not covered.
- Ordinary stage completion re-admitting an already-scanned waiter without a full scan: not covered.
- Same-workspace external active attempt outside the selected task group consuming capacity: not covered.
- Capacity filled only by an external attempt returning exit `3` without persisting a blocker: not covered.
- Finished task removal from a previously sampled live set before slot calculation: not covered.
- Capped review-preflight failure refilling an independent worker slot: not covered.
- Negative and non-integer CLI values: not covered.

### Symbol References

None. No symbol is renamed or removed; the CLI option and internal helpers/parameters are additive.

### Split Judgment

Keep one plan. Argument parsing, workspace occupancy, admission order, capacity-only wait state, claim timing, review-preflight refill, tests, and documentation form one scheduler invariant. Splitting would allow an intermediate implementation to exceed the cap, strand ready tasks, or persist a live-capacity wait as a blocker.

### Scope Rationale

Include only the Python dispatcher entry point, its existing test module, and its project skill. Exclude provider output filters, tunnel buffers, provider-specific concurrency/quota policy, Go `iop-agent` configuration/parity, persistent config schema, deployment, and live provider smoke. The limit is per dispatcher invocation and per canonical physical workspace; separate clones/worktrees remain independent.

### Final Routing

- `evaluation_mode`: `isolated-reassessment`
- `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 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 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.

## Implementation Checklist

- [ ] Add the non-negative `--max-parallel` CLI contract with `0` as the backward-compatible unlimited default.
- [ ] Enforce one physical-workspace cap across worker, self-check, review, and verified external-active attempts without narrowing occupancy by `--task-group`.
- [ ] Preserve claim ownership, review ordering, task-only reclassification, review-preflight drain, and non-terminal capacity waiting.
- [ ] Add deterministic regressions for unlimited, limits `1`/`2`, cross-group external occupancy, stale-slot release, claims, dry-run, preflight refill, and invalid input.
- [ ] Update the dispatcher skill input, concurrency contract, examples, and verification checklist.
- [ ] 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 workspace-global capacity admission

#### Problem

`select_dispatch_candidates` at `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py:5662-5755` claims every disjoint ready task. `dispatch_with_store` at lines `5827-5838`, `6006-6103`, `6269-6272`, and `6353-6473` has no capacity-only waiter state and launches every selected candidate. `orchestration_live_agent_processes` at lines `1062-1081` is scoped to one orchestration, so it cannot by itself enforce a physical-workspace-global cap when `--task-group` is used. `parse_args` at lines `6544-6555` exposes no numeric option.

#### Solution

Add `--max-parallel` with a type validator that accepts integers `>=0`; default `0` means unlimited. Validate direct programmatic namespaces too rather than silently clamping a negative value. Use `getattr(args, "max_parallel", 0)` so existing test namespaces without the additive field preserve compatibility.

Count occupied slots by unique task name:

- current invocation's `running` futures;
- every task state in the same canonical workspace for which `external_active_is_live` verifies current-workspace live/conservative evidence, regardless of orchestration scope or `--task-group`.

Do not count internal pump, heartbeat, selector, or quota-probe coroutines as extra slots. Union the two name sets to avoid double-counting a current future that also has live locator evidence. After finished futures call `store.clear_active`, refresh the workspace live set (or explicitly remove `finished_names`) before calculating the next admission budget so a stale snapshot cannot consume a returned slot.

Extend candidate selection with `available_slots: int | None`. Preserve review-before-worker ordering and write-set validation/conflict checks. Admit and acquire/replace a claim only while a slot is available. A newly capacity-deferred task gets a stable wait reason and no new claim; a task that already owns its lifecycle claim keeps it unchanged while waiting.

Keep `capacity_waiting` as an in-memory task-name set. After an ordinary attempt finishes without `complete.log`, reclassify `finished_names | capacity_waiting` from `task_cache`; do not perform a full scan. Rebuild the set from capacity-only deferrals after each pass. Dependency, blocker, invalid-write-set, and claim-collision deferrals are not capacity waiters and rely on their existing wake-up event.

When the first selected review batch fails `ensure_review_shared_state`, mark every currently ready review with the same shared preflight blocker. Reviews that had received slots retain their existing claims; reviews that never received slots do not synthesize claims. Remove review candidates, then refill the freed runtime slots from disjoint non-review capacity waiters before building the admission/quota snapshot and launching futures. If claim collision prevents refill, preserve the existing blocker/wait semantics.

Apply the same capacity calculation in dry-run without persisting state. If a live external attempt alone fills the cap, report capacity waiting as non-terminal tracking state and return `3`; never call `mark_orchestration_blocked` for a capacity-only wait. Do not terminate already-active attempts when a restart uses a smaller limit—admit nothing until occupancy drops.

Before:

```python
# 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
```

```python
# dispatch.py:6101-6103
else:
    tasks = sorted(task_cache.values(), key=lambda task: (task.index, task.name))
    candidate_scope = finished_names
```

```python
# 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)
```

After:

```python
max_parallel = validated_max_parallel(getattr(args, "max_parallel", 0))
capacity_waiting: set[str] = set()
...
workspace_live = workspace_live_agent_processes(store)
workspace_live.difference_update(finished_names)
occupied_names = set(running) | set(workspace_live)
available_slots = (
    None if max_parallel == 0 else max(0, max_parallel - len(occupied_names))
)
candidate_scope = finished_names | capacity_waiting
```

```python
def select_dispatch_candidates(
    store: StateStore,
    ready: list[tuple[Task, str]],
    *,
    persist: bool,
    available_slots: int | None = None,
) -> tuple[...]:
    # Validate claim eligibility, then admit only while a slot remains.
    # Capacity-only deferral does not create or replace a task claim.
```

#### Modified Files and Checklist

- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`: add CLI/programmatic validation and default.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`: derive unscoped same-workspace live occupancy and refresh it after future completion.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`: add bounded candidate admission, capacity-wait wake-up, and claim preservation.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/scripts/dispatch.py`: refill after shared review preflight failure and keep external-capacity-only waits non-terminal.

#### Test Strategy

Write API-2 regressions only; do not invoke real providers.

#### Verification

```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 additive option.

### [API-2] Add admission and lifecycle regressions

#### Problem

`ReviewSchedulingTest` at `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py:4863-4939` covers only unlimited selection. `WriteSetTest` at lines `5180-5345` does not cover capacity claim timing. `BlockerDrainTest` contains same-group external-active and unlimited preflight-drain cases but not cross-group/global-cap behavior. `DispatcherConvergenceSimulationTest` at lines `7341-7528` proves unrestricted convergence only.

#### Solution

Add `ParallelLimitSchedulingTest` in the existing module using `Task`, `StateStore`, deterministic events/counters, fake role coroutines, and subprocess-deny guards:

- default/explicit `0` selects all disjoint ready tasks;
- limit `2` selects reviews before worker/self-check, and total observed role attempts never exceeds two;
- limit `1` serializes attempts and re-admits a capacity waiter after an ordinary stage transition without an extra full scan;
- a newly deferred task has no claim, while a capacity-deferred task with an existing lifecycle claim retains it;
- dry-run applies the cap, reports overflow as waiting, and leaves dispatcher state unchanged;
- a verified external-active task from another task group consumes the workspace slot even with `--task-group`;
- external-only saturation returns `3` and does not persist the selected orchestration as blocked;
- finishing a current future releases its slot even if it appeared in the prior live snapshot;
- capped review preflight failure blocks ready reviews but still fills the released slot with a disjoint worker;
- negative and non-integer values are rejected with exit code `2`;
- the existing unlimited convergence test remains unchanged.

Use `asyncio.Event`/active counters and bounded `asyncio.wait_for` only inside tests. Patch every real subprocess entry point to fail if invoked.

Before:

```python
# 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.assertNotIn(new_waiter.name, store.data["write_claims"])
self.assertEqual(store.data["write_claims"][existing_owner.name], prior_claim)
```

#### Modified Files and Checklist

- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`: add CLI and candidate normal/boundary tests.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`: add bounded mixed-role, task-only wake-up, stale-slot release, and claim tests.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/tests/test_dispatch.py`: add cross-group external saturation, dry-run, and review-preflight refill tests with provider subprocess denial.

#### Test Strategy

Write all listed regressions in the existing test file. The pass oracle is the configured maximum active task-attempt count, deterministic start order, correct claim snapshot, correct scan count, correct exit/status state, and zero provider invocations.

#### Verification

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

Expected: all focused tests pass with no real provider process.

### [API-3] Document exact scope and restart behavior

#### Problem

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

#### Solution

Document `max_parallel`/`--max-parallel` as invocation-scoped: `0` is unlimited and a positive value caps unique active task-stage attempts across the physical workspace. State that `--task-group` does not narrow occupancy, adopted external attempts count, internal helper coroutines do not count separately, and the flag must be supplied again on restart. Keep provider-specific rules intact. Revise official-review wording to “no separate review-only limit; subject to the global cap.” Add all-task and task-group examples with `--max-parallel 2`, plus dry-run preview wording.

Before:

```markdown
<!-- SKILL.md:82-85 -->
- Pi `ornith:35b`: 3.
- agy: 1.
- Official Codex review: no separate numeric limit.
```

After:

```markdown
- Global physical-workspace limit: `max_parallel=0` is unlimited; a positive
  value caps unique active task-stage attempts and is not narrowed by
  `task_group`.
- Official Codex review: no separate review-only limit; subject to the global limit.
```

#### Modified Files and Checklist

- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md`: add input, scope, counting, restart, and dry-run semantics.
- [ ] `agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md`: add capped invocation examples and align review verification wording.

#### Test Strategy

No separate documentation test file; use deterministic searches plus API-2 runtime tests.

#### Verification

```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
rg -n --sort path --fixed-strings 'not narrowed by `task_group`' agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md
```

Expected: input, CLI examples, and non-narrowing task-group semantics are 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
rg -n --sort path --fixed-strings 'not narrowed by `task_group`' agent-ops/skills/project/orchestrate-agent-task-loop/SKILL.md
git diff --check
```

Expected: compilation/help checks pass; focused fake-runner tests pass; the full dispatcher suite passes fresh; documentation checks find the exact scope; `git diff --check` prints nothing. Python test cache is not accepted.

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