[Phase 4] Reconciliation: Conductor Batch Loop Integration

[frankbria]

Summary

Wire the reconciliation engine into the conductor's batch execution loop so running batches respond to external state changes.

Parent issue: #403

Scope

Changes to core/conductor.py

1. Add reconciliation check between task dispatches:

   • After each task completes (or on a timer for parallel execution)
   • Call reconciler.check_tasks() for all queued and running tasks
   • Apply recommended actions:
     • SKIP: Remove task from queue, log skip reason
     • KILL: Cancel running agent for that task (if parallel)
     • REQUEUE: Move blocked task back to ready queue
     • NONE: No action needed

2. Configurable reconciliation interval:

   • --reconcile-interval CLI flag (default: 30s)
   • 0 disables reconciliation (current behavior)
   • For serial execution: check between each task
   • For parallel execution: check on timer + after each task completion

3. Batch status reporting:

   • Track reconciliation events in batch status
   • Show "3 tasks skipped (externally completed)" in cf work batch status

Changes to cli/app.py

4. Add --reconcile-interval flag to cf work batch run

Acceptance Criteria

• Reconciliation runs between task dispatches in serial mode
• Reconciliation runs on timer in parallel mode
• Externally completed tasks are skipped with log message
• Externally blocked tasks have their agents stopped
• Resolved blockers cause tasks to be re-queued
• cf work batch status reflects reconciliation events
• Integration test: modify task state during batch, verify batch adjusts

Dependencies

• Depends on [Phase 4] Reconciliation: State Checker Engine #404 (reconciliation engine)

[traycerai]

Plan

Observations

• file:codeframe/core/conductor.py contains start_batch(), _execute_serial(), _execute_parallel(), and _execute_group_parallel() as the core batch loop, with BatchRun using a results: dict[str, str] field backed by JSON in SQLite.
• _active_processes: dict[str, dict[str, subprocess.Popen]] is already tracked with _active_processes_lock — KILL support requires only SIGTERM calls into this dict.
• file:codeframe/cli/app.py has batch_run() (around line 2808) which calls conductor.start_batch() and batch_status() (line 3021) which renders results.
• The reconciliation engine (reconciler.py) is expected to be delivered by issue [Phase 4] Reconciliation: State Checker Engine #404 and is not yet present — this plan assumes it exposes reconciler.check_tasks(task_ids) -> list[ReconcileRecommendation] with .task_id and .action (one of SKIP, KILL, REQUEUE, NONE) and .reason.

Approach

Integrate the #404 reconciler into the existing conductor loop with minimal invasiveness: add reconcile_interval as a new parameter threaded from CLI → start_batch() → execution functions, insert a _run_reconciliation() helper called between serial task dispatches and on a background timer for parallel execution, track reconciliation events in a new BatchRun field, and surface them in batch_status.

---

Implementation Steps

1. Extend BatchRun with reconciliation tracking — file:codeframe/core/conductor.py

Add a new field to the BatchRun dataclass:

reconcile_events: list[dict] = field(default_factory=list)

Each event dict holds {"task_id": ..., "action": ..., "reason": ..., "timestamp": ...}.

---

2. Update _save_batch() and _row_to_batch() — file:codeframe/core/conductor.py

• In _save_batch(): serialize reconcile_events to JSON and include it in the INSERT OR REPLACE statement. This requires adding a reconcile_events column to the batch_runs table. Add a CREATE TABLE IF NOT EXISTS migration or ALTER TABLE ... ADD COLUMN reconcile_events TEXT DEFAULT NULL guard at batch creation time using get_db_connection.
• In _row_to_batch(): deserialize reconcile_events from the new column (index 11), defaulting to [] if null/absent.

---

3. Add _run_reconciliation() helper — file:codeframe/core/conductor.py

Create a new private function with this signature:

def _run_reconciliation(
    workspace: Workspace,
    batch: BatchRun,
    queued_task_ids: list[str],
    on_event: Optional[Callable[[str, dict], None]] = None,
) -> set[str]:  # Returns set of task_ids to skip from queue

Inside:

1. Import from codeframe.core import reconciler (the [Phase 4] Reconciliation: State Checker Engine #404 module).
2. Call reconciler.check_tasks(workspace, queued_task_ids) → list of recommendations.
3. Iterate recommendations:
   • SKIP: Log "[Reconciler] Skipping {task_id}: {reason}", record event in batch.reconcile_events, mark batch.results[task_id] = "SKIPPED", emit a BATCH_TASK_SKIPPED event (or reuse BATCH_TASK_COMPLETED with metadata), add to returned skip-set.
   • KILL: If task_id is in _active_processes[batch.id], send SIGTERM to the process (mirror pattern from stop_batch()), log "[Reconciler] Killing {task_id}: {reason}", record event.
   • REQUEUE: Log "[Reconciler] Re-queueing {task_id}: {reason}", record event. Return it as not in the skip-set (caller keeps it in queue).
   • NONE: No-op.
4. Call _save_batch(workspace, batch) after all actions are applied.
5. Return the set of task IDs to skip from the serial queue.

---

4. Wire reconciliation into _execute_serial() — file:codeframe/core/conductor.py

Add reconcile_interval: int = 0 parameter. At the top of the for i, task_id in enumerate(batch.task_ids) loop, before emitting BATCH_TASK_QUEUED:

1. If reconcile_interval > 0, build the list of remaining (not-yet-dispatched) task IDs from batch.task_ids[i:].
2. Call _run_reconciliation(workspace, batch, remaining_ids, on_event) → skip_set.
3. If task_id in skip_set: print(f"[Reconciler] Skipping {task_id}"), continue to next iteration.

---

5. Wire reconciliation into _execute_parallel() — file:codeframe/core/conductor.py

Add reconcile_interval: int = 0 parameter.

Between groups (serial loop at line 1396): After each group completes, if reconcile_interval > 0, build remaining task IDs for subsequent groups and call _run_reconciliation(). Filter out any skipped tasks from subsequent groups before dispatching.

Background timer for within-group parallel execution: Inside _execute_group_parallel(), add reconcile_interval: int = 0 parameter. When reconcile_interval > 0, use threading.Timer to fire _run_reconciliation() periodically while the ThreadPoolExecutor runs. The KILL action uses _active_processes[batch.id][task_id].terminate() directly. Use a threading.Event to cancel the timer when the group finishes.

---

6. Wire reconciliation into _execute_serial_resume() — file:codeframe/core/conductor.py

Add reconcile_interval: int = 0 parameter, applying the same between-task reconciliation check as step 4.

---

7. Thread reconcile_interval through start_batch() — file:codeframe/core/conductor.py

• Add reconcile_interval: int = 0 parameter to start_batch().
• Forward it to _execute_serial(), _execute_parallel(), and _execute_serial_resume() at their call sites (lines ~577–594, ~864).

---

8. Add --reconcile-interval flag to batch_run() — file:codeframe/cli/app.py

In the @batch_app.command("run") function (batch_run()), add a new Typer option after the existing engine option:

reconcile_interval: int = typer.Option(
    30,
    "--reconcile-interval",
    help="Seconds between reconciliation checks (0 = disabled)",
)

Pass it through to conductor.start_batch(... reconcile_interval=reconcile_interval).

Also update the execution plan console output block (around line 2932) to show:

console.print(f"  Reconcile interval: {'disabled' if reconcile_interval == 0 else f'{reconcile_interval}s'}")

---

9. Show reconciliation events in batch_status — file:codeframe/cli/app.py

In batch_status(), after the existing task result loop (around line 3081), add a new section that, if batch.reconcile_events is non-empty:

1. Prints a [bold]Reconciliation Events:[/bold] section.
2. Iterates events and prints " [dim]{timestamp}[/dim] {action}: {task_id[:8]} - {reason}".
3. Adds a summary line, e.g. " 3 tasks skipped (externally completed)", " 1 task killed (externally blocked)".

Also update the batch list table columns to include a Skipped count column sourced from reconcile_events.

---

Data / Control Flow

sequenceDiagram
    participant CLI as cli/app.py (batch_run)
    participant Conductor as conductor.py (start_batch)
    participant Serial as _execute_serial
    participant Parallel as _execute_parallel
    participant Rec as _run_reconciliation
    participant Reconciler as reconciler.py (#404)
    participant Timer as threading.Timer

    CLI->>Conductor: start_batch(..., reconcile_interval=30)
    alt serial strategy
        Conductor->>Serial: _execute_serial(..., reconcile_interval=30)
        loop per task
            Serial->>Rec: _run_reconciliation(remaining_ids)
            Rec->>Reconciler: check_tasks(ids)
            Reconciler-->>Rec: [SKIP task_a, NONE task_b]
            Rec-->>Serial: skip_set={task_a}
            Serial->>Serial: skip task_a, continue
            Serial->>Serial: dispatch task_b via subprocess
        end
    else parallel strategy
        Conductor->>Parallel: _execute_parallel(..., reconcile_interval=30)
        loop per dependency group
            Parallel->>Rec: _run_reconciliation(remaining_ids)
            Parallel->>Parallel: _execute_group_parallel (ThreadPoolExecutor)
            Timer-->>Rec: periodic _run_reconciliation (KILL active tasks)
        end
    end
    Rec->>Conductor: batch.reconcile_events updated + _save_batch()
    CLI->>CLI: batch_status → show reconcile_events summary

Loading

Import In IDE

🤖 Prompt for AI Agents

## Observations

- `file:codeframe/core/conductor.py` contains `start_batch()`, `_execute_serial()`, `_execute_parallel()`, and `_execute_group_parallel()` as the core batch loop, with `BatchRun` using a `results: dict[str, str]` field backed by JSON in SQLite.
- `_active_processes: dict[str, dict[str, subprocess.Popen]]` is already tracked with `_active_processes_lock` — KILL support requires only SIGTERM calls into this dict.
- `file:codeframe/cli/app.py` has `batch_run()` (around line 2808) which calls `conductor.start_batch()` and `batch_status()` (line 3021) which renders results.
- The reconciliation engine (`reconciler.py`) is expected to be delivered by issue #404 and is not yet present — this plan assumes it exposes `reconciler.check_tasks(task_ids) -> list[ReconcileRecommendation]` with `.task_id` and `.action` (one of `SKIP`, `KILL`, `REQUEUE`, `NONE`) and `.reason`.

## Approach

Integrate the #404 reconciler into the existing conductor loop with minimal invasiveness: add `reconcile_interval` as a new parameter threaded from CLI → `start_batch()` → execution functions, insert a `_run_reconciliation()` helper called between serial task dispatches and on a background timer for parallel execution, track reconciliation events in a new `BatchRun` field, and surface them in `batch_status`.

---

## Implementation Steps

### 1. Extend `BatchRun` with reconciliation tracking — `file:codeframe/core/conductor.py`

Add a new field to the `BatchRun` dataclass:

```
reconcile_events: list[dict] = field(default_factory=list)
```

Each event dict holds `{"task_id": ..., "action": ..., "reason": ..., "timestamp": ...}`.

---

### 2. Update `_save_batch()` and `_row_to_batch()` — `file:codeframe/core/conductor.py`

- In `_save_batch()`: serialize `reconcile_events` to JSON and include it in the `INSERT OR REPLACE` statement. This requires adding a `reconcile_events` column to the `batch_runs` table. Add a `CREATE TABLE IF NOT EXISTS` migration or `ALTER TABLE ... ADD COLUMN reconcile_events TEXT DEFAULT NULL` guard at batch creation time using `get_db_connection`.
- In `_row_to_batch()`: deserialize `reconcile_events` from the new column (index 11), defaulting to `[]` if null/absent.

---

### 3. Add `_run_reconciliation()` helper — `file:codeframe/core/conductor.py`

Create a new private function with this signature:

```
def _run_reconciliation(
    workspace: Workspace,
    batch: BatchRun,
    queued_task_ids: list[str],
    on_event: Optional[Callable[[str, dict], None]] = None,
) -> set[str]:  # Returns set of task_ids to skip from queue
```

Inside:
1. Import `from codeframe.core import reconciler` (the #404 module).
2. Call `reconciler.check_tasks(workspace, queued_task_ids)` → list of recommendations.
3. Iterate recommendations:
   - **`SKIP`**: Log `"[Reconciler] Skipping {task_id}: {reason}"`, record event in `batch.reconcile_events`, mark `batch.results[task_id] = "SKIPPED"`, emit a `BATCH_TASK_SKIPPED` event (or reuse `BATCH_TASK_COMPLETED` with metadata), add to returned skip-set.
   - **`KILL`**: If `task_id` is in `_active_processes[batch.id]`, send SIGTERM to the process (mirror pattern from `stop_batch()`), log `"[Reconciler] Killing {task_id}: {reason}"`, record event.
   - **`REQUEUE`**: Log `"[Reconciler] Re-queueing {task_id}: {reason}"`, record event. Return it as _not_ in the skip-set (caller keeps it in queue).
   - **`NONE`**: No-op.
4. Call `_save_batch(workspace, batch)` after all actions are applied.
5. Return the set of task IDs to skip from the serial queue.

---

### 4. Wire reconciliation into `_execute_serial()` — `file:codeframe/core/conductor.py`

Add `reconcile_interval: int = 0` parameter. At the top of the `for i, task_id in enumerate(batch.task_ids)` loop, **before** emitting `BATCH_TASK_QUEUED`:

1. If `reconcile_interval > 0`, build the list of remaining (not-yet-dispatched) task IDs from `batch.task_ids[i:]`.
2. Call `_run_reconciliation(workspace, batch, remaining_ids, on_event)` → `skip_set`.
3. If `task_id in skip_set`: `print(f"[Reconciler] Skipping {task_id}")`, `continue` to next iteration.

---

### 5. Wire reconciliation into `_execute_parallel()` — `file:codeframe/core/conductor.py`

Add `reconcile_interval: int = 0` parameter.

**Between groups (serial loop at line 1396):** After each group completes, if `reconcile_interval > 0`, build remaining task IDs for subsequent groups and call `_run_reconciliation()`. Filter out any skipped tasks from subsequent groups before dispatching.

**Background timer for within-group parallel execution:** Inside `_execute_group_parallel()`, add `reconcile_interval: int = 0` parameter. When `reconcile_interval > 0`, use `threading.Timer` to fire `_run_reconciliation()` periodically while the `ThreadPoolExecutor` runs. The KILL action uses `_active_processes[batch.id][task_id].terminate()` directly. Use a `threading.Event` to cancel the timer when the group finishes.

---

### 6. Wire reconciliation into `_execute_serial_resume()` — `file:codeframe/core/conductor.py`

Add `reconcile_interval: int = 0` parameter, applying the same between-task reconciliation check as step 4.

---

### 7. Thread `reconcile_interval` through `start_batch()` — `file:codeframe/core/conductor.py`

- Add `reconcile_interval: int = 0` parameter to `start_batch()`.
- Forward it to `_execute_serial()`, `_execute_parallel()`, and `_execute_serial_resume()` at their call sites (lines ~577–594, ~864).

---

### 8. Add `--reconcile-interval` flag to `batch_run()` — `file:codeframe/cli/app.py`

In the `@batch_app.command("run")` function (`batch_run()`), add a new Typer option after the existing `engine` option:

```
reconcile_interval: int = typer.Option(
    30,
    "--reconcile-interval",
    help="Seconds between reconciliation checks (0 = disabled)",
)
```

Pass it through to `conductor.start_batch(... reconcile_interval=reconcile_interval)`.

Also update the execution plan console output block (around line 2932) to show:
```
console.print(f"  Reconcile interval: {'disabled' if reconcile_interval == 0 else f'{reconcile_interval}s'}")
```

---

### 9. Show reconciliation events in `batch_status` — `file:codeframe/cli/app.py`

In `batch_status()`, after the existing task result loop (around line 3081), add a new section that, if `batch.reconcile_events` is non-empty:

1. Prints a `[bold]Reconciliation Events:[/bold]` section.
2. Iterates events and prints `"  [dim]{timestamp}[/dim] {action}: {task_id[:8]} - {reason}"`.
3. Adds a summary line, e.g. `"  3 tasks skipped (externally completed)"`, `"  1 task killed (externally blocked)"`.

Also update the batch list table columns to include a `Skipped` count column sourced from `reconcile_events`.

---

### Data / Control Flow

```mermaid
sequenceDiagram
    participant CLI as cli/app.py (batch_run)
    participant Conductor as conductor.py (start_batch)
    participant Serial as _execute_serial
    participant Parallel as _execute_parallel
    participant Rec as _run_reconciliation
    participant Reconciler as reconciler.py (#404)
    participant Timer as threading.Timer

    CLI->>Conductor: start_batch(..., reconcile_interval=30)
    alt serial strategy
        Conductor->>Serial: _execute_serial(..., reconcile_interval=30)
        loop per task
            Serial->>Rec: _run_reconciliation(remaining_ids)
            Rec->>Reconciler: check_tasks(ids)
            Reconciler-->>Rec: [SKIP task_a, NONE task_b]
            Rec-->>Serial: skip_set={task_a}
            Serial->>Serial: skip task_a, continue
            Serial->>Serial: dispatch task_b via subprocess
        end
    else parallel strategy
        Conductor->>Parallel: _execute_parallel(..., reconcile_interval=30)
        loop per dependency group
            Parallel->>Rec: _run_reconciliation(remaining_ids)
            Parallel->>Parallel: _execute_group_parallel (ThreadPoolExecutor)
            Timer-->>Rec: periodic _run_reconciliation (KILL active tasks)
        end
    end
    Rec->>Conductor: batch.reconcile_events updated + _save_batch()
    CLI->>CLI: batch_status → show reconcile_events summary
```

---

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

#128 - Implement bottleneck detection for workflow monitoring [merged]
#363 - feat: wire ReactAgent into runtime with --engine react CLI flag [merged]
#369 - test: add CLI integration tests for ReactAgent runtime parameters [merged]
#382 - fix: return non-zero exit codes from work retry and batch run [merged]
#387 - [Phase 2.5-F] Wire engine parameter through v2 API routes (#354) [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!

[frankbria]

Implemented in PR #439 (merged). All reconciliation components (engine, conductor integration, GitHub sync) delivered as a single PR implementing parent issue #403.