[Phase 4] Reconciliation: State Checker Engine

[frankbria]

Summary

Create a reconciliation engine that checks the current state of tasks against external sources and reports discrepancies.

Parent issue: #403

Scope

New module: core/reconciler.py

1. ReconciliationResult dataclass:

   @dataclass
   class ReconciliationResult:
       task_id: str
       current_status: TaskStatus
       external_status: TaskStatus | None
       action: ReconcileAction  # SKIP, KILL, REQUEUE, NONE
       reason: str

2. Reconciler class:

   class Reconciler:
       def __init__(self, workspace_path: Path, check_interval_ms: int = 30_000):
           ...
       
       def check_tasks(self, task_ids: list[str]) -> list[ReconciliationResult]:
           """Check current state of given tasks against stored state."""
           ...
       
       def check_blockers(self, task_ids: list[str]) -> list[ReconciliationResult]:
           """Check if any blocked tasks have had their blockers resolved."""
           ...

3. State check logic:

   • Read task status from durable state (SQLite/filesystem)
   • Compare against in-memory batch state
   • If task externally transitioned to DONE/FAILED → recommend SKIP
   • If task externally transitioned to BLOCKED → recommend KILL
   • If blocked task has all blockers resolved → recommend REQUEUE

Acceptance Criteria

• Reconciler checks task state from durable storage
• Detects external status transitions (DONE, FAILED, BLOCKED)
• Detects blocker resolutions for blocked tasks
• Returns structured ReconciliationResult with recommended action
• Unit tests for all state transition scenarios

Dependencies

• None — standalone module using existing core/tasks.py and core/blockers.py

[traycerai]

Plan

Observations

• TaskStatus lives in codeframe/core/state_machine.py (BACKLOG, READY, IN_PROGRESS, BLOCKED, FAILED, DONE, MERGED).
• tasks.get() and tasks.list_tasks() are the durable-state read primitives; blockers are queried via blockers.list_for_task() and blockers.list_all().
• All v2 core modules are headless, use @dataclass, and avoid FastAPI/HTTP.
• The test convention for tests/core/ uses plain pytest class groups, a tmp_path-backed workspace fixture, and pytestmark = pytest.mark.v2.

Approach

Create codeframe/core/reconciler.py as a pure, standalone module — no new dependencies. ReconcileAction is defined as a str Enum (consistent with TaskStatus and BlockerStatus). ReconciliationResult is a @dataclass. Reconciler receives a Workspace and an optional check_interval_ms (stored but not used for scheduling — keeping scope minimal). The two public methods each call existing tasks.get() / blockers.list_for_task() APIs. A matching test file is placed at tests/core/test_reconciler.py.

---

Implementation Steps

1. Define ReconcileAction enum — file:codeframe/core/reconciler.py

Add a ReconcileAction(str, Enum) at the top of the new file with four members:

Member	Meaning
NONE	No discrepancy detected
SKIP	Task externally transitioned to DONE or FAILED — skip it
KILL	Task externally transitioned to BLOCKED — stop it
REQUEUE	Blocked task has all blockers resolved — re-queue it

Pattern-match TaskStatus and BlockerStatus (same str mixin style as the rest of the codebase).

---

2. Define ReconciliationResult dataclass — file:codeframe/core/reconciler.py

@dataclass
class ReconciliationResult:
    task_id: str
    current_status: TaskStatus          # status in durable store (SQLite)
    external_status: TaskStatus | None  # status passed in as in-memory/batch state
    action: ReconcileAction
    reason: str

external_status is None when the task is not found in the supplied batch state dict.

---

3. Implement Reconciler class — file:codeframe/core/reconciler.py

Constructor

• Accept workspace: Workspace and check_interval_ms: int = 30_000.
• Store both on self; no background thread is started.

check_tasks(self, task_ids: list[str], batch_state: dict[str, TaskStatus] | None = None) -> list[ReconciliationResult]

The ticket specifies "compare against in-memory batch state". Since Reconciler is standalone, pass the in-memory state as an optional dict argument (keyed by task_id → TaskStatus).

Logic per task_id:

1. Call tasks.get(self.workspace, task_id) → stored_task.
2. If not found → ReconciliationResult(action=NONE, reason="task not found", external_status=None, current_status=None) — skip silently.
3. current_status = stored_task.status (durable).
4. external_status = batch_state.get(task_id) if batch_state provided, else None.
5. Apply decision table:

Stored status	Condition	action	reason
DONE or FAILED	—	SKIP	"task already terminal in durable state"
BLOCKED	external is DONE/FAILED	SKIP	"task externally completed while blocked"
IN_PROGRESS	external is BLOCKED	KILL	"task externally transitioned to BLOCKED"
DONE/FAILED	external differs	SKIP	"external state confirms terminal"
any	no discrepancy	NONE	"no action required"

check_blockers(self, task_ids: list[str]) -> list[ReconciliationResult]

Logic per task_id:

1. stored_task = tasks.get(self.workspace, task_id) — skip if None.
2. If stored_task.status != TaskStatus.BLOCKED → emit NONE result.
3. Fetch task_blockers = blockers.list_for_task(self.workspace, task_id).
4. A blocker is resolved when its status is BlockerStatus.RESOLVED or BlockerStatus.ANSWERED.
5. If all task_blockers are resolved (including the case of an empty list after checking len > 0) → REQUEUE, else NONE.

---

4. Imports for file:codeframe/core/reconciler.py

The file only needs:

from __future__ import annotations
import dataclasses
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Optional

from codeframe.core.workspace import Workspace
from codeframe.core import tasks, blockers
from codeframe.core.state_machine import TaskStatus
from codeframe.core.blockers import BlockerStatus

No new third-party packages; all are already present.

---

5. Unit tests — file:tests/core/test_reconciler.py

Follow the exact pattern of file:tests/core/test_task_dependencies.py:

• Module-level pytestmark = pytest.mark.v2
• A workspace fixture using create_or_load_workspace(tmp_path)
• Plain pytest classes (no asyncio)

Test classes and scenarios to cover:

TestCheckTasksNoDiscrepancy

• Task in READY, batch state also READY → NONE
• Task in IN_PROGRESS, no batch state provided → NONE

TestCheckTasksSkipAction

• Task stored as DONE → SKIP regardless of batch state
• Task stored as FAILED → SKIP
• Task IN_PROGRESS, external/batch state is DONE → SKIP

TestCheckTasksKillAction

• Task IN_PROGRESS, external/batch state is BLOCKED → KILL

TestCheckBlockersRequeue

• Task is BLOCKED, all associated blockers have status ANSWERED or RESOLVED → REQUEUE

TestCheckBlockersNoRequeue

• Task is BLOCKED, at least one blocker is still OPEN → NONE
• Task is IN_PROGRESS (not BLOCKED) → NONE

TestCheckTasksMissingTask

• task_ids contains an unknown ID → result is silently skipped (no crash)

Each test creates tasks and blockers via the existing tasks.create(), tasks.update_status(), and blockers.create() / blockers.answer() / blockers.resolve() helpers — no mocking needed since tests run against a real in-memory SQLite workspace.

---

6. Data-flow summary

sequenceDiagram
    participant Caller
    participant Reconciler
    participant tasks_module as core/tasks
    participant blockers_module as core/blockers
    participant SQLite

    Caller->>Reconciler: check_tasks(task_ids, batch_state)
    loop per task_id
        Reconciler->>tasks_module: get(workspace, task_id)
        tasks_module->>SQLite: SELECT status FROM tasks WHERE id=?
        SQLite-->>tasks_module: row
        tasks_module-->>Reconciler: Task | None
        Reconciler->>Reconciler: compare durable vs batch_state
        Reconciler-->>Caller: ReconciliationResult(action, reason)
    end

    Caller->>Reconciler: check_blockers(task_ids)
    loop per task_id
        Reconciler->>tasks_module: get(workspace, task_id)
        tasks_module-->>Reconciler: Task
        Reconciler->>blockers_module: list_for_task(workspace, task_id)
        blockers_module->>SQLite: SELECT * FROM blockers WHERE task_id=?
        SQLite-->>blockers_module: rows
        blockers_module-->>Reconciler: list[Blocker]
        Reconciler->>Reconciler: check all blockers resolved?
        Reconciler-->>Caller: ReconciliationResult(REQUEUE | NONE)
    end

Loading

Import In IDE

🤖 Prompt for AI Agents

## Observations

- `TaskStatus` lives in `codeframe/core/state_machine.py` (BACKLOG, READY, IN_PROGRESS, BLOCKED, FAILED, DONE, MERGED).
- `tasks.get()` and `tasks.list_tasks()` are the durable-state read primitives; blockers are queried via `blockers.list_for_task()` and `blockers.list_all()`.
- All v2 core modules are headless, use `@dataclass`, and avoid FastAPI/HTTP.
- The test convention for `tests/core/` uses plain `pytest` class groups, a `tmp_path`-backed `workspace` fixture, and `pytestmark = pytest.mark.v2`.

## Approach

Create `codeframe/core/reconciler.py` as a pure, standalone module — no new dependencies. `ReconcileAction` is defined as a `str Enum` (consistent with `TaskStatus` and `BlockerStatus`). `ReconciliationResult` is a `@dataclass`. `Reconciler` receives a `Workspace` and an optional `check_interval_ms` (stored but not used for scheduling — keeping scope minimal). The two public methods each call existing `tasks.get()` / `blockers.list_for_task()` APIs. A matching test file is placed at `tests/core/test_reconciler.py`.

---

## Implementation Steps

### 1. Define `ReconcileAction` enum — `file:codeframe/core/reconciler.py`

Add a `ReconcileAction(str, Enum)` at the top of the new file with four members:

| Member | Meaning |
|--------|---------|
| `NONE` | No discrepancy detected |
| `SKIP` | Task externally transitioned to DONE or FAILED — skip it |
| `KILL` | Task externally transitioned to BLOCKED — stop it |
| `REQUEUE` | Blocked task has all blockers resolved — re-queue it |

Pattern-match `TaskStatus` and `BlockerStatus` (same `str` mixin style as the rest of the codebase).

---

### 2. Define `ReconciliationResult` dataclass — `file:codeframe/core/reconciler.py`

```
@dataclass
class ReconciliationResult:
    task_id: str
    current_status: TaskStatus          # status in durable store (SQLite)
    external_status: TaskStatus | None  # status passed in as in-memory/batch state
    action: ReconcileAction
    reason: str
```

`external_status` is `None` when the task is not found in the supplied batch state dict.

---

### 3. Implement `Reconciler` class — `file:codeframe/core/reconciler.py`

**Constructor**
- Accept `workspace: Workspace` and `check_interval_ms: int = 30_000`.
- Store both on `self`; no background thread is started.

**`check_tasks(self, task_ids: list[str], batch_state: dict[str, TaskStatus] | None = None) -> list[ReconciliationResult]`**

> The ticket specifies "compare against in-memory batch state". Since `Reconciler` is standalone, pass the in-memory state as an optional dict argument (keyed by `task_id → TaskStatus`).

Logic per `task_id`:
1. Call `tasks.get(self.workspace, task_id)` → `stored_task`.
2. If not found → `ReconciliationResult(action=NONE, reason="task not found", external_status=None, current_status=None)` — skip silently.
3. `current_status = stored_task.status` (durable).
4. `external_status = batch_state.get(task_id)` if `batch_state` provided, else `None`.
5. Apply decision table:

| Stored status | Condition | `action` | `reason` |
|---|---|---|---|
| `DONE` or `FAILED` | — | `SKIP` | "task already terminal in durable state" |
| `BLOCKED` | external is `DONE`/`FAILED` | `SKIP` | "task externally completed while blocked" |
| `IN_PROGRESS` | external is `BLOCKED` | `KILL` | "task externally transitioned to BLOCKED" |
| `DONE`/`FAILED` | external differs | `SKIP` | "external state confirms terminal" |
| any | no discrepancy | `NONE` | "no action required" |

**`check_blockers(self, task_ids: list[str]) -> list[ReconciliationResult]`**

Logic per `task_id`:
1. `stored_task = tasks.get(self.workspace, task_id)` — skip if `None`.
2. If `stored_task.status != TaskStatus.BLOCKED` → emit `NONE` result.
3. Fetch `task_blockers = blockers.list_for_task(self.workspace, task_id)`.
4. A blocker is resolved when its `status` is `BlockerStatus.RESOLVED` or `BlockerStatus.ANSWERED`.
5. If all `task_blockers` are resolved (including the case of an empty list after checking `len > 0`) → `REQUEUE`, else `NONE`.

---

### 4. Imports for `file:codeframe/core/reconciler.py`

The file only needs:
```
from __future__ import annotations
import dataclasses
from dataclasses import dataclass
from enum import Enum
from pathlib import Path
from typing import Optional

from codeframe.core.workspace import Workspace
from codeframe.core import tasks, blockers
from codeframe.core.state_machine import TaskStatus
from codeframe.core.blockers import BlockerStatus
```
No new third-party packages; all are already present.

---

### 5. Unit tests — `file:tests/core/test_reconciler.py`

Follow the exact pattern of `file:tests/core/test_task_dependencies.py`:
- Module-level `pytestmark = pytest.mark.v2`
- A `workspace` fixture using `create_or_load_workspace(tmp_path)`
- Plain `pytest` classes (no `asyncio`)

Test classes and scenarios to cover:

**`TestCheckTasksNoDiscrepancy`**
- Task in READY, batch state also READY → `NONE`
- Task in IN_PROGRESS, no batch state provided → `NONE`

**`TestCheckTasksSkipAction`**
- Task stored as DONE → `SKIP` regardless of batch state
- Task stored as FAILED → `SKIP`
- Task IN_PROGRESS, external/batch state is DONE → `SKIP`

**`TestCheckTasksKillAction`**
- Task IN_PROGRESS, external/batch state is BLOCKED → `KILL`

**`TestCheckBlockersRequeue`**
- Task is BLOCKED, all associated blockers have status ANSWERED or RESOLVED → `REQUEUE`

**`TestCheckBlockersNoRequeue`**
- Task is BLOCKED, at least one blocker is still OPEN → `NONE`
- Task is IN_PROGRESS (not BLOCKED) → `NONE`

**`TestCheckTasksMissingTask`**
- `task_ids` contains an unknown ID → result is silently skipped (no crash)

Each test creates tasks and blockers via the existing `tasks.create()`, `tasks.update_status()`, and `blockers.create()` / `blockers.answer()` / `blockers.resolve()` helpers — no mocking needed since tests run against a real in-memory SQLite workspace.

---

### 6. Data-flow summary

```mermaid
sequenceDiagram
    participant Caller
    participant Reconciler
    participant tasks_module as core/tasks
    participant blockers_module as core/blockers
    participant SQLite

    Caller->>Reconciler: check_tasks(task_ids, batch_state)
    loop per task_id
        Reconciler->>tasks_module: get(workspace, task_id)
        tasks_module->>SQLite: SELECT status FROM tasks WHERE id=?
        SQLite-->>tasks_module: row
        tasks_module-->>Reconciler: Task | None
        Reconciler->>Reconciler: compare durable vs batch_state
        Reconciler-->>Caller: ReconciliationResult(action, reason)
    end

    Caller->>Reconciler: check_blockers(task_ids)
    loop per task_id
        Reconciler->>tasks_module: get(workspace, task_id)
        tasks_module-->>Reconciler: Task
        Reconciler->>blockers_module: list_for_task(workspace, task_id)
        blockers_module->>SQLite: SELECT * FROM blockers WHERE task_id=?
        SQLite-->>blockers_module: rows
        blockers_module-->>Reconciler: list[Blocker]
        Reconciler->>Reconciler: check all blockers resolved?
        Reconciler-->>Caller: ReconciliationResult(REQUEUE | NONE)
    end
```

---

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

#229 - feat(e2e): add state reconciliation tests for late-joining users [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.