fix(core): sweep before the batch starts, and check GitHub for blocked tasks (#1032)

Seventh codex pass.

Before-first-tick: the reconciliation loop waits a full interval before its
first pass, so 'the issue was already closed when the batch started' — the
likeliest case there is — was missed for every task the serial loop reached
inside the first 30 seconds, i.e. the first one. _start_reconciliation_thread
now runs one sweep synchronously before spawning the thread, so the executor's
skip check cannot race it. Costs one call per linked task, nothing when no
task carries an issue.

Blocked tasks: the GitHub check hung off an elif after the blocker branch, so
a BLOCKED task with UNANSWERED blockers fell through both and was never asked
about — despite being exactly the task worth asking about, since the human may
have resolved the work on GitHub instead of answering. Now guarded on
'no local signal already fired and not DONE' rather than chained, with a test
that a resolved blocker still wins and costs no API call.

Author: frankbria

codeframe/core/conductor.py

@@ -2205,46 +2205,59 @@ def _start_reconciliation_thread(
     stop_event = threading.Event()
     engine = ReconciliationEngine(workspace)
 
+    def _pass() -> None:
+        """One reconciliation sweep. Raises nothing the caller must handle."""
+        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:
+                return
+
+            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)
+
     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)
+            _pass()
+
+    # One sweep before the executor launches anything (#1032). The loop waits a
+    # full interval before its first pass, so without this the likeliest case
+    # of all — the linked issue was ALREADY closed when the batch started —
+    # was missed for every task reached inside the first 30 seconds, which in a
+    # serial batch means the first one. Synchronous rather than an early tick
+    # on the thread, so the executor's skip check cannot race it. Costs one
+    # call per linked task, and nothing at all when no task carries an issue.
+    _pass()
 
     thread = threading.Thread(target=_loop, daemon=True, name=f"reconcile-{batch.id[:8]}")
     thread.start()

codeframe/core/reconciliation.py

@@ -256,12 +256,19 @@ def check_task(self, task_id: str) -> list[ExternalStateChange]:
 
         # The task's linked GitHub issue was closed by someone outside this
         # batch — an external completion, exactly like a task marked DONE in
-        # the UI (#1032). Only asked for tasks that are not already finished
-        # locally: the DONE branch above has already fired for those, so a
-        # lookup would be a wasted call on every tick for the rest of the run.
-        # GitHubIssueState never raises and answers False when it does not
-        # know, so a GitHub outage leaves the batch exactly as it was.
-        elif self._issue_state.is_closed(task):
+        # the UI (#1032).
+        #
+        # Deliberately NOT an `elif` off the blocker branch: a BLOCKED task
+        # whose blockers are still unanswered falls through both branches
+        # above, and that task is exactly the one worth asking about — the
+        # human may have resolved the work on GitHub instead of answering.
+        #
+        # Guarded on `not changes` so it costs nothing when a local signal
+        # already fired, and on `not DONE` because the first branch has already
+        # handled those — otherwise it would be a wasted call every tick for
+        # the rest of the run. GitHubIssueState never raises and answers False
+        # when it does not know, so an outage leaves the batch as it was.
+        if not changes and task.status != TaskStatus.DONE and self._issue_state.is_closed(task):
             changes.append(ExternalStateChange(
                 task_id=task_id,
                 change_type="completed",

tests/core/test_external_completion_honored_1032.py

@@ -290,6 +290,62 @@ def fake_subprocess(ws, task_id, batch_id, **kwargs):
         )
 
 
+class TestTheFirstTaskIsCheckedBeforeItStarts:
+    """The reconciliation thread waits a full interval before its first pass.
+
+    So "the issue was already closed when the batch started" — the likeliest
+    case of all — was missed for every task the serial loop reached inside the
+    first 30 seconds, which in practice means the first one.
+    """
+
+    def test_a_pass_runs_before_the_loop_launches_anything(
+        self, workspace, three_tasks, monkeypatch
+    ):
+        executed = []
+        monkeypatch.setattr(
+            conductor, "_execute_task_subprocess",
+            lambda ws, tid, bid, **kw: executed.append(tid) or "COMPLETED",
+        )
+
+        # A reconciler that would find the first task's issue closed. Real
+        # thread start is left intact except for the interval, so this pins the
+        # *synchronous* pass, not a lucky race with the daemon.
+        from codeframe.core.reconciliation import ExternalStateChange
+
+        target = three_tasks[0]
+
+        class StubEngine:
+            def __init__(self, workspace):
+                pass
+
+            def check_all_active(self, ids):
+                from codeframe.core.reconciliation import ReconciliationResult
+
+                r = ReconciliationResult()
+                if target in ids:
+                    r.changes_detected.append(
+                        ExternalStateChange(target, "completed", "github", {})
+                    )
+                return r
+
+            def apply_changes(self, result, batch, procs):
+                for c in result.changes_detected:
+                    batch.results[c.task_id] = "COMPLETED"
+                    result.tasks_skipped.append(c.task_id)
+
+        monkeypatch.setattr(
+            "codeframe.core.reconciliation.ReconciliationEngine", StubEngine
+        )
+
+        batch = _make_batch(workspace, three_tasks)
+        conductor._execute_serial(workspace, batch)
+
+        assert target not in executed, (
+            "the first task ran before reconciliation ever looked at it"
+        )
+        assert batch.results[target] == "COMPLETED"
+
+
 class TestEveryWriteSiteUsesTheGuard:
     def test_no_executor_writes_batch_results_directly(self):
         """A guard that four of five call sites bypass guards nothing.

tests/core/test_github_issue_reconciliation_1032.py

@@ -167,6 +167,43 @@ class Batch:
         assert batch.results["t1"] == "COMPLETED"
         assert result.tasks_skipped == ["t1"]
 
+    def test_a_blocked_task_with_open_blockers_still_checks_github(self, monkeypatch):
+        """The blocker branch used to swallow this case: a BLOCKED task whose
+        blockers are unanswered never reached the GitHub check, so closing the
+        issue left the task stuck and re-runnable forever."""
+        task = make_task(status=TaskStatus.BLOCKED)
+        unanswered = type("B", (), {"status": type("S", (), {"value": "OPEN"})()})()
+        monkeypatch.setattr("codeframe.core.tasks.get", lambda ws, tid: task)
+        monkeypatch.setattr(
+            "codeframe.core.blockers.list_for_task", lambda ws, tid: [unanswered]
+        )
+        engine = ReconciliationEngine(
+            workspace=object(),
+            issue_state=GitHubIssueState(fetch=FakeGitHub(state="closed"), pat="tok"),
+        )
+
+        changes = engine.check_task("t1")
+
+        assert [(c.change_type, c.source) for c in changes] == [("completed", "github")]
+
+    def test_a_resolved_blocker_still_wins_over_the_github_check(self, monkeypatch):
+        """Don't let the new check displace the existing signal."""
+        task = make_task(status=TaskStatus.BLOCKED)
+        answered = type("B", (), {"status": type("S", (), {"value": "ANSWERED"})()})()
+        monkeypatch.setattr("codeframe.core.tasks.get", lambda ws, tid: task)
+        monkeypatch.setattr(
+            "codeframe.core.blockers.list_for_task", lambda ws, tid: [answered]
+        )
+        gh = FakeGitHub(state="closed")
+        engine = ReconciliationEngine(
+            workspace=object(), issue_state=GitHubIssueState(fetch=gh, pat="tok")
+        )
+
+        changes = engine.check_task("t1")
+
+        assert [c.change_type for c in changes] == ["blocker_resolved"]
+        assert gh.calls == [], "spent a call on a task that already had a change"
+
     def test_already_done_task_is_not_charged_a_github_call(self, monkeypatch):
         """The local DONE check already fires; asking GitHub would be a wasted
         call on every tick for every finished task."""