[Phase 4] Continuous Reconciliation During Batch Execution

[frankbria]

Summary

During batch execution, periodically check if tasks have been externally modified (GitHub issue closed, task manually completed, dependency resolved) and adjust the running batch accordingly. Inspired by Symphony's per-tick reconciliation loop.

Motivation

Currently cf work batch run operates on a snapshot of task states taken at batch start. If someone closes a GitHub issue, manually marks a task DONE, or resolves a blocker while the batch is running, CodeFrame won't notice until the agent tries to work on the (now irrelevant) task.

Symphony handles this with reconciliation on every poll tick — checking tracker state for all running issues and killing/adjusting as needed.

Sub-Issues

• [Phase 4] Reconciliation: State Checker Engine #404 — Reconciliation state checker engine (standalone)
• [Phase 4] Reconciliation: Conductor Batch Loop Integration #405 — Conductor batch loop integration (depends on [Phase 4] Reconciliation: State Checker Engine #404)
• [Phase 4] Reconciliation: GitHub Issue State Sync #406 — GitHub issue state sync (depends on [Phase 4] Reconciliation: State Checker Engine #404)

Dependency Graph

#404 (engine) ──┬──> #405 (conductor integration)
                └──> #406 (GitHub sync)

#405 and #406 are independent of each other and can be done in parallel.

Dependencies

• Benefits from Phase 4 multi-agent work (long-running batches)
• No hard prerequisites

Acceptance Criteria

• Running/queued tasks checked for external state changes every N seconds (configurable)
• Tasks closed externally → agent killed, task skipped in batch
• Tasks completed externally → agent killed, task marked DONE
• Blockers resolved externally → blocked tasks re-queued
• GitHub issue state changes reflected in batch execution
• Reconciliation failures logged but don't crash the batch
• Integration tests for external state change → batch adjustment

[traycerai]

Plan

Observations

The batch execution engine lives in file:codeframe/core/conductor.py with three main loops: _execute_serial, _execute_parallel, and _execute_serial_resume. Each loop iterates over tasks with no polling between steps — task state is a snapshot taken at batch start. tasks.py has a Task dataclass with no github_issue_number field. github_integration.py covers PRs only (no issue-state methods). The blockers module has list_for_task and answer/resolve functions. GlobalConfig in config.py is BaseSettings-backed, making it the right place for a configurable RECONCILIATION_INTERVAL_SECONDS env var.

Approach

The ticket decomposes into three sub-issues: a standalone engine (#404), conductor integration (#405), and GitHub sync (#406). Following that dependency order: build a self-contained ReconciliationEngine first, integrate it as a background thread in the conductor loops, then add GitHub issue state fetching as a pluggable check. This keeps each layer independently testable and avoids touching the existing batch execution logic heavily.

---

Implementation Plan

Step 1 — Add reconciliation_interval to GlobalConfig (file:codeframe/core/config.py)

In the GlobalConfig class (which uses pydantic_settings.BaseSettings), add a new field:

• reconciliation_interval_seconds: int = Field(30, alias="RECONCILIATION_INTERVAL_SECONDS") with a sensible default of 30 seconds.

Also add github_issue_number: Optional[int] = None to the Task dataclass in file:codeframe/core/tasks.py and its DB read/write paths in file:codeframe/core/workspace.py (the CREATE TABLE tasks statement and _row_to_task), so tasks can be linked to a GitHub issue for sync purposes.

---

Step 2 — Create ReconciliationEngine (#404) as file:codeframe/core/reconciliation.py

This is a headless, standalone module (no FastAPI). It should contain:

ExternalStateChange dataclass:

Field	Type	Meaning
task_id	str	Which task changed
change_type	str	"closed", "completed", "blocker_resolved"
source	str	"github", "manual"
details	dict	Extra info

ReconciliationResult dataclass:

• changes_detected: list[ExternalStateChange]
• tasks_skipped: list[str]
• tasks_requeued: list[str]
• errors: list[str]

ReconciliationEngine class:

• Constructor accepts workspace: Workspace, a list of task_ids: list[str] (the batch's active task set), and an optional github_checker: Optional[Callable] (for injecting the GitHub check — keeps GitHub as optional).

• check_task(task_id: str) -> list[ExternalStateChange]: Re-fetches the task from tasks.get(workspace, task_id), checks for:

  • TaskStatus.DONE → emit "completed" change
  • TaskStatus.BLOCKED with all open blockers now answered → emit "blocker_resolved" change
  • If a GitHub checker is registered, call it and merge its results

• check_all_active(active_task_ids: list[str]) -> ReconciliationResult: Iterates active_task_ids, calls check_task for each, accumulates results, catches and logs all exceptions individually so a single failure never propagates.

• apply_changes(result: ReconciliationResult, batch: BatchRun, active_processes: dict) -> None: For each change:

  • "completed" or "closed" → terminate the task's subprocess from active_processes via SIGTERM, set batch.results[task_id] = RunStatus.COMPLETED.value or RunStatus.FAILED.value as appropriate, log with events.emit_for_workspace.
  • "blocker_resolved" → mark task as re-queued (batch.results[task_id] = "READY_REQUEUE" or just remove from blocked tracking so the conductor will retry it).

All exceptions caught in apply_changes are appended to result.errors and logged — they do not raise.

---

Step 3 — Add GitHub Issue State Checker (#406) as file:codeframe/git/github_issue_sync.py

This is a lightweight extension to file:codeframe/git/github_integration.py following its same httpx.AsyncClient pattern. Create a new module github_issue_sync.py rather than modifying the existing PR-focused module:

get_issue_state(token: str, repo: str, issue_number: int) -> str: A synchronous wrapper (using httpx.Client, not async, since it's called from a background thread in the conductor) that hits GET /repos/{owner}/{repo}/issues/{issue_number} and returns the issue state field ("open" or "closed").

build_github_task_checker(token: str, repo: str) -> Callable[[str, Task], list[ExternalStateChange]]: A factory that returns a closure suitable for injecting into ReconciliationEngine. The closure:

1. Checks task.github_issue_number — if None, returns [].
2. Calls get_issue_state and, if the issue is "closed", returns an ExternalStateChange(change_type="closed", source="github", ...).
3. Catches httpx.RequestError and any HTTP errors, returns [] and logs the failure.

---

Step 4 — Integrate reconciliation loop into conductor (#405) (file:codeframe/core/conductor.py)

The reconciliation loop runs as a background daemon thread during batch execution. This avoids blocking the main task-dispatch loop and requires no restructuring of _execute_serial or _execute_parallel.

Add _start_reconciliation_thread(workspace, batch, engine, interval) -> threading.Event:

• Spawns a threading.Thread(daemon=True) that:
  1. Sleeps for interval seconds.
  2. Calls engine.check_all_active(batch.task_ids).
  3. Calls engine.apply_changes(result, batch, _active_processes.get(batch.id, {})).
  4. Repeats until a stop event is set.
• Returns the threading.Event used to stop the thread.

Wire into _execute_serial and _execute_parallel:

At the top of each function:
  - Instantiate ReconciliationEngine(workspace, batch.task_ids, github_checker)
  - Start the reconciliation thread with the configured interval
  - Store the stop_event

At the bottom (in a try/finally):
  - Set stop_event to terminate the background thread

The reconciliation thread uses the existing _active_processes_lock and _active_processes dict (already in conductor.py) to terminate processes safely — no new locking primitives needed.

How to get the github_checker:

Inside start_batch, optionally build the GitHub checker if GITHUB_TOKEN and GITHUB_REPO are set in GlobalConfig. Pass it down to _execute_serial/_execute_parallel as a parameter. If the env vars are absent, pass None — the engine gracefully skips GitHub checks.

---

Step 5 — Add new EventType constants (file:codeframe/core/events.py)

Add to the EventType class:

RECONCILIATION_STARTED       = "RECONCILIATION_STARTED"
RECONCILIATION_TASK_SKIPPED  = "RECONCILIATION_TASK_SKIPPED"
RECONCILIATION_TASK_REQUEUED = "RECONCILIATION_TASK_REQUEUED"
RECONCILIATION_ERROR         = "RECONCILIATION_ERROR"

These are emitted by ReconciliationEngine.apply_changes using the existing events.emit_for_workspace helper.

---

Data Flow

sequenceDiagram
    participant S as start_batch()
    participant ES as _execute_serial / _execute_parallel
    participant RT as Reconciliation Thread (daemon)
    participant RE as ReconciliationEngine
    participant GH as GitHubIssueSync
    participant DB as SQLite (tasks/blockers)
    participant AP as _active_processes

    S->>ES: call with workspace, batch
    ES->>RT: start daemon thread (interval N sec)
    loop Every N seconds
        RT->>RE: check_all_active(active_task_ids)
        RE->>DB: tasks.get(workspace, task_id) × N
        RE->>GH: get_issue_state (if github_issue_number set)
        RE-->>RT: ReconciliationResult
        RT->>RE: apply_changes(result, batch, active_processes)
        RE->>AP: SIGTERM process if task closed/completed
        RE->>DB: update batch.results + _save_batch
        RE->>DB: events.emit_for_workspace(RECONCILIATION_*)
    end
    ES->>RT: set stop_event (finally block)

Loading

---

Configuration Summary

Env Var	Default	Effect
RECONCILIATION_INTERVAL_SECONDS	30	How often the engine polls task/GitHub state
GITHUB_TOKEN	(existing)	Enables GitHub issue sync; absent = sync skipped
GITHUB_REPO	(existing)	Repo for issue state lookup

Import In IDE

🤖 Prompt for AI Agents

## Observations

The batch execution engine lives in `file:codeframe/core/conductor.py` with three main loops: `_execute_serial`, `_execute_parallel`, and `_execute_serial_resume`. Each loop iterates over tasks with **no polling between steps** — task state is a snapshot taken at batch start. `tasks.py` has a `Task` dataclass with no `github_issue_number` field. `github_integration.py` covers PRs only (no issue-state methods). The `blockers` module has `list_for_task` and `answer`/`resolve` functions. `GlobalConfig` in `config.py` is `BaseSettings`-backed, making it the right place for a configurable `RECONCILIATION_INTERVAL_SECONDS` env var.

## Approach

The ticket decomposes into three sub-issues: a standalone engine (#404), conductor integration (#405), and GitHub sync (#406). Following that dependency order: build a self-contained `ReconciliationEngine` first, integrate it as a background thread in the conductor loops, then add GitHub issue state fetching as a pluggable check. This keeps each layer independently testable and avoids touching the existing batch execution logic heavily.

---

## Implementation Plan

### Step 1 — Add `reconciliation_interval` to `GlobalConfig` (`file:codeframe/core/config.py`)

In the `GlobalConfig` class (which uses `pydantic_settings.BaseSettings`), add a new field:

- `reconciliation_interval_seconds: int = Field(30, alias="RECONCILIATION_INTERVAL_SECONDS")` with a sensible default of 30 seconds.

Also add `github_issue_number: Optional[int] = None` to the `Task` dataclass in `file:codeframe/core/tasks.py` and its DB read/write paths in `file:codeframe/core/workspace.py` (the `CREATE TABLE tasks` statement and `_row_to_task`), so tasks can be linked to a GitHub issue for sync purposes.

---

### Step 2 — Create `ReconciliationEngine` (#404) as `file:codeframe/core/reconciliation.py`

This is a **headless, standalone** module (no FastAPI). It should contain:

**`ExternalStateChange` dataclass:**
| Field | Type | Meaning |
|---|---|---|
| `task_id` | `str` | Which task changed |
| `change_type` | `str` | `"closed"`, `"completed"`, `"blocker_resolved"` |
| `source` | `str` | `"github"`, `"manual"` |
| `details` | `dict` | Extra info |

**`ReconciliationResult` dataclass:**
- `changes_detected: list[ExternalStateChange]`
- `tasks_skipped: list[str]`
- `tasks_requeued: list[str]`
- `errors: list[str]`

**`ReconciliationEngine` class:**

- Constructor accepts `workspace: Workspace`, a list of `task_ids: list[str]` (the batch's active task set), and an optional `github_checker: Optional[Callable]` (for injecting the GitHub check — keeps GitHub as optional).

- `check_task(task_id: str) -> list[ExternalStateChange]`: Re-fetches the task from `tasks.get(workspace, task_id)`, checks for:
  - `TaskStatus.DONE` → emit `"completed"` change
  - `TaskStatus.BLOCKED` with all open blockers now answered → emit `"blocker_resolved"` change
  - If a GitHub checker is registered, call it and merge its results

- `check_all_active(active_task_ids: list[str]) -> ReconciliationResult`: Iterates `active_task_ids`, calls `check_task` for each, accumulates results, catches and logs all exceptions individually so a single failure never propagates.

- `apply_changes(result: ReconciliationResult, batch: BatchRun, active_processes: dict) -> None`: For each change:
  - `"completed"` or `"closed"` → terminate the task's subprocess from `active_processes` via `SIGTERM`, set `batch.results[task_id] = RunStatus.COMPLETED.value` or `RunStatus.FAILED.value` as appropriate, log with `events.emit_for_workspace`.
  - `"blocker_resolved"` → mark task as re-queued (`batch.results[task_id] = "READY_REQUEUE"` or just remove from blocked tracking so the conductor will retry it).

All exceptions caught in `apply_changes` are appended to `result.errors` and logged — they do **not** raise.

---

### Step 3 — Add GitHub Issue State Checker (#406) as `file:codeframe/git/github_issue_sync.py`

This is a lightweight extension to `file:codeframe/git/github_integration.py` following its same `httpx.AsyncClient` pattern. Create a new module `github_issue_sync.py` rather than modifying the existing PR-focused module:

**`get_issue_state(token: str, repo: str, issue_number: int) -> str`**: A synchronous wrapper (using `httpx.Client`, not async, since it's called from a background thread in the conductor) that hits `GET /repos/{owner}/{repo}/issues/{issue_number}` and returns the issue `state` field (`"open"` or `"closed"`).

**`build_github_task_checker(token: str, repo: str) -> Callable[[str, Task], list[ExternalStateChange]]`**: A factory that returns a closure suitable for injecting into `ReconciliationEngine`. The closure:
1. Checks `task.github_issue_number` — if `None`, returns `[]`.
2. Calls `get_issue_state` and, if the issue is `"closed"`, returns an `ExternalStateChange(change_type="closed", source="github", ...)`.
3. Catches `httpx.RequestError` and any HTTP errors, returns `[]` and logs the failure.

---

### Step 4 — Integrate reconciliation loop into conductor (#405) (`file:codeframe/core/conductor.py`)

The reconciliation loop runs as a **background daemon thread** during batch execution. This avoids blocking the main task-dispatch loop and requires no restructuring of `_execute_serial` or `_execute_parallel`.

**Add `_start_reconciliation_thread(workspace, batch, engine, interval) -> threading.Event`:**
- Spawns a `threading.Thread(daemon=True)` that:
  1. Sleeps for `interval` seconds.
  2. Calls `engine.check_all_active(batch.task_ids)`.
  3. Calls `engine.apply_changes(result, batch, _active_processes.get(batch.id, {}))`.
  4. Repeats until a stop event is set.
- Returns the `threading.Event` used to stop the thread.

**Wire into `_execute_serial` and `_execute_parallel`:**

```
At the top of each function:
  - Instantiate ReconciliationEngine(workspace, batch.task_ids, github_checker)
  - Start the reconciliation thread with the configured interval
  - Store the stop_event

At the bottom (in a try/finally):
  - Set stop_event to terminate the background thread
```

The reconciliation thread uses the existing `_active_processes_lock` and `_active_processes` dict (already in `conductor.py`) to terminate processes safely — no new locking primitives needed.

**How to get the `github_checker`:**

Inside `start_batch`, optionally build the GitHub checker if `GITHUB_TOKEN` and `GITHUB_REPO` are set in `GlobalConfig`. Pass it down to `_execute_serial`/`_execute_parallel` as a parameter. If the env vars are absent, pass `None` — the engine gracefully skips GitHub checks.

---

### Step 5 — Add new `EventType` constants (`file:codeframe/core/events.py`)

Add to the `EventType` class:

```
RECONCILIATION_STARTED       = "RECONCILIATION_STARTED"
RECONCILIATION_TASK_SKIPPED  = "RECONCILIATION_TASK_SKIPPED"
RECONCILIATION_TASK_REQUEUED = "RECONCILIATION_TASK_REQUEUED"
RECONCILIATION_ERROR         = "RECONCILIATION_ERROR"
```

These are emitted by `ReconciliationEngine.apply_changes` using the existing `events.emit_for_workspace` helper.

---

### Data Flow

```mermaid
sequenceDiagram
    participant S as start_batch()
    participant ES as _execute_serial / _execute_parallel
    participant RT as Reconciliation Thread (daemon)
    participant RE as ReconciliationEngine
    participant GH as GitHubIssueSync
    participant DB as SQLite (tasks/blockers)
    participant AP as _active_processes

    S->>ES: call with workspace, batch
    ES->>RT: start daemon thread (interval N sec)
    loop Every N seconds
        RT->>RE: check_all_active(active_task_ids)
        RE->>DB: tasks.get(workspace, task_id) × N
        RE->>GH: get_issue_state (if github_issue_number set)
        RE-->>RT: ReconciliationResult
        RT->>RE: apply_changes(result, batch, active_processes)
        RE->>AP: SIGTERM process if task closed/completed
        RE->>DB: update batch.results + _save_batch
        RE->>DB: events.emit_for_workspace(RECONCILIATION_*)
    end
    ES->>RT: set stop_event (finally block)
```

---

### Configuration Summary

| Env Var | Default | Effect |
|---|---|---|
| `RECONCILIATION_INTERVAL_SECONDS` | `30` | How often the engine polls task/GitHub state |
| `GITHUB_TOKEN` | (existing) | Enables GitHub issue sync; absent = sync skipped |
| `GITHUB_REPO` | (existing) | Repo for issue state lookup |

---

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]
#229 - feat(e2e): add state reconciliation tests for late-joining users [merged]
#234 - fix(ui): make TaskStats phase-aware to fix late-joining user bug [merged]
#363 - feat: wire ReactAgent into runtime with --engine react CLI flag [merged]
#382 - fix: return non-zero exit codes from work retry and batch run [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!