feat(core): continuous reconciliation during batch execution - #439
Conversation
Implement periodic checking of tasks for external state changes during batch execution, with automatic adjustment of running batches. - ReconciliationEngine: standalone checker for task state changes - Detects: completed externally, blocker resolved, GitHub issue closed - All errors caught per-task (never crashes the batch) - GitHub issue state sync via synchronous httpx client - build_github_task_checker() factory for injection into engine - Skips tasks without github_issue_number - Conductor integration: daemon reconciliation thread in serial/parallel - Configurable interval (default 30s) via reconciliation_interval_seconds - Uses existing _active_processes dict for subprocess termination - Task.github_issue_number field + DB migration - RECONCILIATION_* event types for observability - 27 unit tests covering engine, sync, config, and apply_changes Closes #403, #404, #405, #406
WalkthroughImplements continuous background reconciliation during batch execution. Periodically checks active task states for external changes—GitHub issues closed, manual task completion, blockers resolved—and adjusts running batches accordingly. Introduces reconciliation engine, event types, database schema updates, and GitHub integration. Changes
Sequence DiagramsequenceDiagram
actor Conductor
participant Thread as Reconciliation<br/>Thread
participant Engine as Reconciliation<br/>Engine
participant GHChecker as GitHub<br/>Checker
participant Batch as Batch<br/>Process
participant OS as OS<br/>Process
Conductor->>Thread: Start daemon thread<br/>(interval=30s)
loop Every 30 seconds
Thread->>Engine: check_all_active(active_ids)
par Check Task 1
Engine->>Engine: check_task(id)
alt Task status = DONE
Engine->>Engine: Record "completed"
else Task status = BLOCKED
Engine->>Engine: Record "blocker_resolved"
end
and Check Task 2 (GitHub)
Engine->>GHChecker: invoke_github_checker(task)
GHChecker-->>Engine: ExternalStateChange<br/>(if closed)
end
Engine-->>Thread: ReconciliationResult
Thread->>Engine: apply_changes(result, batch,<br/>active_processes)
alt "completed" or "closed"
Engine->>OS: Terminate process
Engine->>Batch: Mark COMPLETED/FAILED
else "blocker_resolved"
Engine->>Batch: Re-queue task (READY)
end
Thread->>Batch: Emit reconciliation events
end
Conductor->>Thread: Signal stop event
Thread-->>Conductor: Thread exits
Estimated code review effort🎯 4 (Complex) | ⏱️ ~60 minutes Poem
🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✏️ Tip: You can configure your own custom pre-merge checks in the settings. ✨ Finishing Touches
🧪 Generate unit tests (beta)
📝 Coding Plan
Comment |
Code Review — PR #439: Continuous Reconciliation During Batch ExecutionThe overall shape of this feature is solid. The architecture is clean and headless, the injection-based 1. Thread safety:
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Caution
Some comments are outside the diff and can’t be posted inline due to platform limitations.
⚠️ Outside diff range comments (2)
codeframe/core/workspace.py (1)
379-381:⚠️ Potential issue | 🟠 MajorMissing
github_issue_numbermigration in_ensure_schema_upgrades.The
github_issue_numbercolumn is added in_init_database(line 137), but_ensure_schema_upgradesdoes not include this column. Existing workspaces that upgrade will not have this column, causing the_row_to_taskfunction to fail when checkinglen(row) > 13since the column won't exist in the database.🐛 Proposed fix to add migration
Add after line 381:
if "uncertainty_level" not in task_columns: cursor.execute("ALTER TABLE tasks ADD COLUMN uncertainty_level TEXT") conn.commit() + if "github_issue_number" not in task_columns: + cursor.execute("ALTER TABLE tasks ADD COLUMN github_issue_number INTEGER") + conn.commit()🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@codeframe/core/workspace.py` around lines 379 - 381, The schema upgrade function _ensure_schema_upgrades is missing the migration that adds the github_issue_number column (added in _init_database), so existing databases will lack this column and _row_to_task will break; update _ensure_schema_upgrades to check task_columns for "github_issue_number" and, if absent, execute an ALTER TABLE tasks ADD COLUMN github_issue_number TEXT (similar to the existing uncertainty_level migration) and commit the connection so the tasks table matches what _row_to_task expects.codeframe/core/tasks.py (1)
138-145:⚠️ Potential issue | 🟠 MajorSELECT queries do not include
github_issue_numbercolumn.The
_row_to_taskfunction attempts to readrow[13]forgithub_issue_number, but the SELECT queries throughout the file (e.g., lines 140-141, 176-177, 187-188) only select 13 columns (indices 0-12). This meanslen(row) > 13will always be false, andgithub_issue_numberwill always beNoneeven when the column has a value in the database.Update all SELECT queries to include
github_issue_number:🐛 Proposed fix for SELECT queries
cursor.execute( """ - SELECT id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, created_at, updated_at + SELECT id, workspace_id, prd_id, title, description, status, priority, depends_on, estimated_hours, complexity_score, uncertainty_level, created_at, updated_at, github_issue_number FROM tasks WHERE workspace_id = ? AND id = ? """,Apply similarly to all other SELECT queries in
list_tasks(lines 176-177 and 187-188).Also applies to: 673-673
🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@codeframe/core/tasks.py` around lines 138 - 145, The SELECT statements that build Task rows (e.g., the query in the get-by-id block and those used by list_tasks) are missing the github_issue_number column, so _row_to_task (which expects row[13] for github_issue_number) always sees None; update every SELECT in codeframe/core/tasks.py that selects task columns (including the query around cursor.execute in the single-task fetch and the queries used in list_tasks) to include github_issue_number in the column list in the same position as _row_to_task expects, ensuring the row length and indices align with _row_to_task's mapping.
🧹 Nitpick comments (4)
codeframe/git/github_issue_sync.py (1)
53-53: Consider adding a return type annotation for the checker.The inner
checkerfunction returnslistbut could specifylist[ExternalStateChange]for better type safety and IDE support.♻️ Proposed type annotation
- def checker(task_id: str, task) -> list: + def checker(task_id: str, task) -> list["ExternalStateChange"]: from codeframe.core.reconciliation import ExternalStateChangeNote: Use a string literal for forward reference since
ExternalStateChangeis imported inside the function.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@codeframe/git/github_issue_sync.py` at line 53, Add a precise return annotation to the inner function checker by changing its signature to return list["ExternalStateChange"]; specifically update def checker(task_id: str, task) -> list["ExternalStateChange"]: so the return type is a forward-referenced list of ExternalStateChange (keep the existing import of ExternalStateChange inside the function and use the string literal to avoid forward-reference issues).codeframe/core/reconciliation.py (1)
132-137: Consider stronger typing forbatchparameter.The
batch: objecttype hint is intentionally weak to avoid circular imports withconductor.py. This is acceptable, but consider using aProtocolorTYPE_CHECKINGimport for better type safety if the interface stabilizes.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@codeframe/core/reconciliation.py` around lines 132 - 137, The apply_changes method's batch parameter is currently typed as object which is too weak; update apply_changes(self, result: ReconciliationResult, batch: object, active_processes: dict) to use a stricter interface by defining a lightweight Protocol describing the attributes/methods the code expects (or use a TYPE_CHECKING import from conductor to import the concrete Batch type only for type checking), then annotate batch with that Protocol or the TYPE_CHECKING-only type; reference the apply_changes function and ReconciliationResult so callers and implementers get stronger static checks without introducing circular runtime imports.codeframe/core/conductor.py (1)
878-1030:_execute_serial_resumedoes not start a reconciliation thread.Unlike
_execute_serialand_execute_parallel, the_execute_serial_resumefunction does not start/stop a reconciliation thread. This means resumed batches won't benefit from continuous reconciliation for external state changes.Consider adding reconciliation thread lifecycle to
_execute_serial_resumefor consistency.🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@codeframe/core/conductor.py` around lines 878 - 1030, _execute_serial_resume currently omits the reconciliation thread lifecycle: mirror the pattern used in _execute_serial/_execute_parallel by starting the reconciliation thread at the top of _execute_serial_resume (e.g., call _start_reconciliation_thread(workspace, batch.id) and capture the handle) and ensure it's stopped in a finally block (e.g., call _stop_reconciliation_thread(handle) or equivalent) so the thread runs during task retries and is cleaned up on exit/cancel; wrap the main loop in try/finally to guarantee _stop_reconciliation_thread is invoked even on errors or cancellation.tests/core/test_reconciliation.py (1)
324-326: Weak assertion does not verify error handling behavior.The assertion
assert len(result.errors) >= 0is always true and doesn't verify that the error was actually caught or logged. Consider asserting that no exception propagates (which is implicitly tested) or explicitly check that errors are recorded.♻️ Proposed stronger assertion
# Should not raise engine.apply_changes(result, batch, active_processes) - assert len(result.errors) >= 0 # Error may or may not be logged + # Verify the method completed without raising + # The OSError from terminate() is silently caught per implementation + assert mock_proc.terminate.called🤖 Prompt for AI Agents
Verify each finding against the current code and only fix it if needed. In `@tests/core/test_reconciliation.py` around lines 324 - 326, Replace the no-op assertion after engine.apply_changes(result, batch, active_processes) with an explicit check on the error container: assert that result.errors is a list (e.g., assert isinstance(result.errors, list)) and then assert the expected semantic outcome—either assert len(result.errors) > 0 if you expect an error to be recorded or assert result.errors == [] if you expect no errors; this makes the test verify actual error-handling behavior rather than a tautology.
🤖 Prompt for all review comments with AI agents
Verify each finding against the current code and only fix it if needed.
Outside diff comments:
In `@codeframe/core/tasks.py`:
- Around line 138-145: The SELECT statements that build Task rows (e.g., the
query in the get-by-id block and those used by list_tasks) are missing the
github_issue_number column, so _row_to_task (which expects row[13] for
github_issue_number) always sees None; update every SELECT in
codeframe/core/tasks.py that selects task columns (including the query around
cursor.execute in the single-task fetch and the queries used in list_tasks) to
include github_issue_number in the column list in the same position as
_row_to_task expects, ensuring the row length and indices align with
_row_to_task's mapping.
In `@codeframe/core/workspace.py`:
- Around line 379-381: The schema upgrade function _ensure_schema_upgrades is
missing the migration that adds the github_issue_number column (added in
_init_database), so existing databases will lack this column and _row_to_task
will break; update _ensure_schema_upgrades to check task_columns for
"github_issue_number" and, if absent, execute an ALTER TABLE tasks ADD COLUMN
github_issue_number TEXT (similar to the existing uncertainty_level migration)
and commit the connection so the tasks table matches what _row_to_task expects.
---
Nitpick comments:
In `@codeframe/core/conductor.py`:
- Around line 878-1030: _execute_serial_resume currently omits the
reconciliation thread lifecycle: mirror the pattern used in
_execute_serial/_execute_parallel by starting the reconciliation thread at the
top of _execute_serial_resume (e.g., call
_start_reconciliation_thread(workspace, batch.id) and capture the handle) and
ensure it's stopped in a finally block (e.g., call
_stop_reconciliation_thread(handle) or equivalent) so the thread runs during
task retries and is cleaned up on exit/cancel; wrap the main loop in try/finally
to guarantee _stop_reconciliation_thread is invoked even on errors or
cancellation.
In `@codeframe/core/reconciliation.py`:
- Around line 132-137: The apply_changes method's batch parameter is currently
typed as object which is too weak; update apply_changes(self, result:
ReconciliationResult, batch: object, active_processes: dict) to use a stricter
interface by defining a lightweight Protocol describing the attributes/methods
the code expects (or use a TYPE_CHECKING import from conductor to import the
concrete Batch type only for type checking), then annotate batch with that
Protocol or the TYPE_CHECKING-only type; reference the apply_changes function
and ReconciliationResult so callers and implementers get stronger static checks
without introducing circular runtime imports.
In `@codeframe/git/github_issue_sync.py`:
- Line 53: Add a precise return annotation to the inner function checker by
changing its signature to return list["ExternalStateChange"]; specifically
update def checker(task_id: str, task) -> list["ExternalStateChange"]: so the
return type is a forward-referenced list of ExternalStateChange (keep the
existing import of ExternalStateChange inside the function and use the string
literal to avoid forward-reference issues).
In `@tests/core/test_reconciliation.py`:
- Around line 324-326: Replace the no-op assertion after
engine.apply_changes(result, batch, active_processes) with an explicit check on
the error container: assert that result.errors is a list (e.g., assert
isinstance(result.errors, list)) and then assert the expected semantic
outcome—either assert len(result.errors) > 0 if you expect an error to be
recorded or assert result.errors == [] if you expect no errors; this makes the
test verify actual error-handling behavior rather than a tautology.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro
Run ID: 464b6443-9d20-486b-b37f-d0fb1151ade5
📒 Files selected for processing (8)
codeframe/core/conductor.pycodeframe/core/config.pycodeframe/core/events.pycodeframe/core/reconciliation.pycodeframe/core/tasks.pycodeframe/core/workspace.pycodeframe/git/github_issue_sync.pytests/core/test_reconciliation.py
1 participant
Summary
Implements #403: Continuous Reconciliation During Batch Execution (+ sub-issues #404, #405, #406)
Adds periodic checking of tasks for external state changes during batch execution with automatic adjustments.
codeframe/core/reconciliation.py): Standalone checker that detects completed tasks, resolved blockers, and GitHub issue closurescodeframe/git/github_issue_sync.py): Synchronous httpx client for checking issue state, injectable into the engine_execute_serialand_execute_parallel, configurable interval (default 30s)Acceptance Criteria
Test Plan
Implementation Notes
_active_processes_lockin conductor for safe process terminationgithub_issue_number INTEGERcolumn added via existing_init_database()migration patternCloses #403, #404, #405, #406
Summary by CodeRabbit
Release Notes