feat(core): continuous reconciliation during batch execution

codeframe/core/conductor.py

@@ -8,6 +8,7 @@
 
 import hashlib
 import json
+import logging
 import re
 import subprocess
 import sys
@@ -26,6 +27,8 @@
 from codeframe.core.runtime import RunStatus, get_active_run, reset_blocked_run
 
 
+logger = logging.getLogger(__name__)
+
 # Tactical patterns that the supervisor should auto-resolve
 SUPERVISOR_TACTICAL_PATTERNS = [
     # Virtual environment / package management
@@ -1226,6 +1229,12 @@ def _execute_serial(
 
     Updates batch.results and batch.status as tasks complete.
     """
+    # Start reconciliation thread for continuous state checking
+    from codeframe.core.config import load_environment_config
+    env_config = load_environment_config(workspace.repo_path)
+    interval = env_config.reconciliation_interval_seconds if env_config else 30
+    reconcile_stop = _start_reconciliation_thread(workspace, batch, interval_seconds=interval)
+
     completed_count = 0
     failed_count = 0
     blocked_count = 0
@@ -1359,6 +1368,9 @@ def _execute_serial(
         print_event=True,
     )
 
+    # Stop reconciliation thread
+    reconcile_stop.set()
+
     # Print summary
     print(f"\nBatch {batch.status.value.lower()}: {completed_count}/{total} tasks completed")
     if failed_count > 0:
@@ -1394,6 +1406,12 @@ def _execute_parallel(
     else:
         print("All tasks are sequential (chain dependencies)")
 
+    # Start reconciliation thread for continuous state checking
+    from codeframe.core.config import load_environment_config as _load_env_config
+    _env_config_p = _load_env_config(workspace.repo_path)
+    _interval_p = _env_config_p.reconciliation_interval_seconds if _env_config_p else 30
+    _reconcile_stop_p = _start_reconciliation_thread(workspace, batch, interval_seconds=_interval_p)
+
     completed_count = 0
     failed_count = 0
     blocked_count = 0
@@ -1500,6 +1518,9 @@ def _execute_parallel(
         print_event=True,
     )
 
+    # Stop reconciliation thread
+    _reconcile_stop_p.set()
+
     # Print summary
     print(f"\nBatch {batch.status.value.lower()}: {completed_count}/{total} tasks completed")
     print(f"  Execution: {plan.num_groups} groups (parallel strategy)")
@@ -1509,6 +1530,71 @@ def _execute_parallel(
         print(f"  Blocked: {blocked_count}")
 
 
+def _start_reconciliation_thread(
+    workspace: Workspace,
+    batch: BatchRun,
+    interval_seconds: int = 30,
+    github_checker: Optional[Callable] = None,
+) -> threading.Event:
+    """Start a daemon thread that periodically reconciles batch state.
+
+    The thread checks all active tasks for external state changes every
+    ``interval_seconds`` and applies adjustments (skip completed tasks,
+    re-queue unblocked tasks).
+
+    Returns a threading.Event that can be set to stop the thread.
+    """
+    from codeframe.core.reconciliation import ReconciliationEngine
+
+    stop_event = threading.Event()
+    engine = ReconciliationEngine(workspace, github_checker=github_checker)
+
+    def _loop() -> None:
+        while not stop_event.wait(timeout=interval_seconds):
+            try:
+                # Get currently active task IDs from the batch
+                active_ids = [
+                    tid for tid in batch.task_ids
+                    if batch.results.get(tid) is None
+                    or batch.results.get(tid) == "RUNNING"
+                ]
+                if not active_ids:
+                    continue
+
+                result = engine.check_all_active(active_ids)
+                if result.changes_detected:
+                    with _active_processes_lock:
+                        procs = _active_processes.get(batch.id, {})
+                        engine.apply_changes(result, batch, procs)
+
+                    # Emit events for changes
+                    for tid in result.tasks_skipped:
+                        events.emit_for_workspace(
+                            workspace,
+                            events.EventType.RECONCILIATION_TASK_SKIPPED,
+                            {"batch_id": batch.id, "task_id": tid},
+                        )
+                    for tid in result.tasks_requeued:
+                        events.emit_for_workspace(
+                            workspace,
+                            events.EventType.RECONCILIATION_TASK_REQUEUED,
+                            {"batch_id": batch.id, "task_id": tid},
+                        )
+                if result.errors:
+                    for err in result.errors:
+                        events.emit_for_workspace(
+                            workspace,
+                            events.EventType.RECONCILIATION_ERROR,
+                            {"batch_id": batch.id, "error": err},
+                        )
+            except Exception as exc:
+                logger.warning("Reconciliation loop error: %s", exc)
+
+    thread = threading.Thread(target=_loop, daemon=True, name=f"reconcile-{batch.id[:8]}")
+    thread.start()
+    return stop_event
+
+
 def _execute_single_task(
     workspace: Workspace,
     batch: BatchRun,

codeframe/core/config.py

@@ -127,6 +127,9 @@ class EnvironmentConfig:
     # Workspace lifecycle hooks
     hooks: HooksConfig = dataclass_field(default_factory=HooksConfig)
 
+    # Reconciliation during batch execution
+    reconciliation_interval_seconds: int = 30
+
     # Execution engine
     engine: str = "react"
 

codeframe/core/events.py

@@ -108,6 +108,12 @@ class EventType:
     HOOK_EXECUTED = "HOOK_EXECUTED"
     HOOK_FAILED = "HOOK_FAILED"
 
+    # Reconciliation events
+    RECONCILIATION_STARTED = "RECONCILIATION_STARTED"
+    RECONCILIATION_TASK_SKIPPED = "RECONCILIATION_TASK_SKIPPED"
+    RECONCILIATION_TASK_REQUEUED = "RECONCILIATION_TASK_REQUEUED"
+    RECONCILIATION_ERROR = "RECONCILIATION_ERROR"
+
 
 @dataclass
 class Event:

codeframe/core/reconciliation.py

@@ -0,0 +1,183 @@
+"""Continuous reconciliation engine for batch execution.
+
+Periodically checks if tasks have been externally modified (GitHub issue
+closed, task manually completed, blocker resolved) and adjusts the running
+batch accordingly.
+
+This module is headless (no FastAPI, no HTTP). It exposes a standalone
+ReconciliationEngine that can be driven by a background thread in the
+conductor.
+"""
+
+from __future__ import annotations
+
+import logging
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING, Callable, Optional
+
+from codeframe.core import blockers, tasks
+from codeframe.core.state_machine import TaskStatus
+
+if TYPE_CHECKING:
+    from codeframe.core.workspace import Workspace
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass
+class ExternalStateChange:
+    """A detected external change to a task's state."""
+
+    task_id: str
+    change_type: str  # "completed", "closed", "blocker_resolved"
+    source: str  # "manual", "github"
+    details: dict = field(default_factory=dict)
+
+
+@dataclass
+class ReconciliationResult:
+    """Accumulated result from a reconciliation check."""
+
+    changes_detected: list[ExternalStateChange] = field(default_factory=list)
+    tasks_skipped: list[str] = field(default_factory=list)
+    tasks_requeued: list[str] = field(default_factory=list)
+    errors: list[str] = field(default_factory=list)
+
+
+class ReconciliationEngine:
+    """Checks tasks for external state changes and applies adjustments.
+
+    The engine is stateless per invocation — call check_all_active() to scan,
+    then apply_changes() to act on the results.
+
+    Args:
+        workspace: The workspace to check tasks in.
+        github_checker: Optional callable(task_id, task) -> list[ExternalStateChange]
+            for GitHub issue state sync. If None, GitHub checks are skipped.
+    """
+
+    def __init__(
+        self,
+        workspace: Workspace,
+        *,
+        github_checker: Optional[Callable] = None,
+    ) -> None:
+        self._workspace = workspace
+        self._github_checker = github_checker
+
+    def check_task(self, task_id: str) -> list[ExternalStateChange]:
+        """Check a single task for external state changes.
+
+        Returns a list of detected changes (may be empty).
+        """
+        task = tasks.get(self._workspace, task_id)
+        if task is None:
+            return []
+
+        changes: list[ExternalStateChange] = []
+
+        # Task was completed externally (e.g., manually marked DONE)
+        if task.status == TaskStatus.DONE:
+            changes.append(ExternalStateChange(
+                task_id=task_id,
+                change_type="completed",
+                source="manual",
+                details={"status": task.status.value},
+            ))
+
+        # Task is blocked but all blockers have been answered
+        elif task.status == TaskStatus.BLOCKED:
+            task_blockers = blockers.list_for_task(self._workspace, task_id)
+            if task_blockers and all(
+                b.status.value in ("ANSWERED", "RESOLVED") for b in task_blockers
+            ):
+                changes.append(ExternalStateChange(
+                    task_id=task_id,
+                    change_type="blocker_resolved",
+                    source="manual",
+                    details={"blockers_resolved": len(task_blockers)},
+                ))
+
+        # Check GitHub issue state if checker is available
+        if self._github_checker:
+            try:
+                gh_changes = self._github_checker(task_id, task)
+                changes.extend(gh_changes)
+            except Exception as exc:
+                logger.warning("GitHub check failed for task %s: %s", task_id, exc)
+
+        return changes
+
+    def check_all_active(
+        self, active_task_ids: list[str]
+    ) -> ReconciliationResult:
+        """Check all active tasks for external state changes.
+
+        Individual task check failures are caught and logged — a single
+        failure never crashes the entire reconciliation pass.
+        """
+        result = ReconciliationResult()
+
+        for task_id in active_task_ids:
+            try:
+                changes = self.check_task(task_id)
+                result.changes_detected.extend(changes)
+            except Exception as exc:
+                error_msg = f"Reconciliation check failed for {task_id}: {exc}"
+                result.errors.append(error_msg)
+                logger.warning(error_msg)
+
+        return result
+
+    def apply_changes(
+        self,
+        result: ReconciliationResult,
+        batch: object,
+        active_processes: dict,
+    ) -> None:
+        """Apply detected changes to the batch and running processes.
+
+        For completed/closed tasks: terminate the subprocess and skip.
+        For blocker_resolved tasks: mark for re-queue.
+
+        All exceptions are caught and appended to result.errors.
+        """
+        for change in result.changes_detected:
+            try:
+                if change.change_type in ("completed", "closed"):
+                    # Terminate subprocess if running
+                    proc = active_processes.get(change.task_id)
+                    if proc is not None:
+                        try:
+                            proc.terminate()
+                        except OSError:
+                            pass  # Process already dead
+
+                    result.tasks_skipped.append(change.task_id)
+
+                    # Update batch results
+                    if hasattr(batch, "results"):
+                        status = "COMPLETED" if change.change_type == "completed" else "FAILED"
+                        batch.results[change.task_id] = status
+
+                    logger.info(
+                        "Task %s %s externally (%s) — skipped in batch",
+                        change.task_id, change.change_type, change.source,
+                    )
+
+                elif change.change_type == "blocker_resolved":
+                    result.tasks_requeued.append(change.task_id)
+
+                    # Update batch results to signal re-queue
+                    if hasattr(batch, "results"):
+                        batch.results[change.task_id] = "READY"
+
+                    logger.info(
+                        "Task %s blocker resolved — re-queued",
+                        change.task_id,
+                    )
+
+            except Exception as exc:
+                error_msg = f"Failed to apply change for {change.task_id}: {exc}"
+                result.errors.append(error_msg)
+                logger.warning(error_msg)

codeframe/core/tasks.py

@@ -55,6 +55,7 @@ class Task:
     estimated_hours: Optional[float] = None
     complexity_score: Optional[int] = None
     uncertainty_level: Optional[str] = None
+    github_issue_number: Optional[int] = None
 
 
 def create(
@@ -669,4 +670,5 @@ def _row_to_task(row: tuple) -> Task:
         uncertainty_level=row[10],
         created_at=datetime.fromisoformat(row[11]),
         updated_at=datetime.fromisoformat(row[12]),
+        github_issue_number=row[13] if len(row) > 13 else None,
     )

codeframe/core/workspace.py

@@ -133,6 +133,8 @@ def _init_database(db_path: Path) -> None:
         cursor.execute("ALTER TABLE tasks ADD COLUMN complexity_score INTEGER")
     if "uncertainty_level" not in columns:
         cursor.execute("ALTER TABLE tasks ADD COLUMN uncertainty_level TEXT")
+    if "github_issue_number" not in columns:
+        cursor.execute("ALTER TABLE tasks ADD COLUMN github_issue_number INTEGER")
 
     # Append-only event log
     cursor.execute("""