fix(scheduler): serve in_progress SHORTEST-REMAINING first — turn effort into completions

Within a stage the matrix workers took the STALEST project. But working on a
project REFRESHES its `updated_at` and sends it to the back of the queue — so
in_progress was a strict ROUND-ROBIN over all 435 projects. Draining one takes
several ticks (~38 tasks at ~10/tick), so each project had to wait for the other
434 between touches. Every project crept forward and NONE ever finished: classic
processor-sharing starvation, and a direct reason zero projects had EVER crossed
research_complete even when the pipeline was healthy.

Meanwhile 68 projects were already close: 8 with <=5 tasks left, 54 with 16-30.
They were never prioritised, so they never landed.

FIX: implement stages (in_progress / paper_in_progress) are served
shortest-remaining-first — the project closest to done goes first (optimal for
mean completion time). Staleness stays the tiebreak. It cannot starve the long
projects: a finished project LEAVES the stage, so the queue steadily advances to
them. Everywhere else keeps oldest-waiting-first (no "remaining work" to be short
of). tasks.md is resolved via the SSoT `feature_dir_for`; an unreadable one sorts
LAST, so "no tasks file" can never masquerade as "almost finished".

Live effect (12 workers): the 8 in_progress workers now land on the projects that
can actually cross — five with 0 tasks remaining (straight to the execution gate),
then 1, 1, 2 remaining — instead of three arbitrary 60-task projects.

Also fixes a bug in the new apportionment: MAX_STAGE_WORKER_SHARE could IDLE a
worker when only one stage was occupied (nothing else to spend the capped slot
on). The cap exists to stop one stage crowding out others, never to waste a
worker — leftover slots are now filled ignoring the cap, bounded by pile size.

Full offline suite: 6147 passed.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: github-actions[bot]

src/llmxive/pipeline/scheduler.py

@@ -440,12 +440,75 @@ def pick_for_worker(
         return None
     stage = slots[worker_index]
     depth = slots[:worker_index].count(stage)
-    in_stage = sorted(by_stage[stage], key=lambda p: (p.updated_at, p.id))
+    in_stage = _order_within_stage(stage, by_stage[stage], repo_root=repo_root)
     if depth >= len(in_stage):
         return None
     return in_stage[depth]
 
 
+#: Stages whose completion needs MANY sequential ticks (the implementer drains a
+#: batch of tasks per tick, and a project carries 30-70 tasks).
+_IMPLEMENT_STAGES: frozenset[Stage] = frozenset({
+    Stage.IN_PROGRESS, Stage.PAPER_IN_PROGRESS,
+})
+
+
+def _remaining_tasks(project: Project, *, repo_root: Path | None) -> int:
+    """Open (``[ ]``) + under-review (``[~]``) task count for an implement project.
+
+    Resolved through the SSoT ``feature_dir_for`` (honours the ``speckit_*_dir``
+    pointer) so a stale lower-numbered ``specs/*`` dir can never shadow the real one.
+    An unreadable/absent tasks.md sorts LAST (a large sentinel) rather than first —
+    "no tasks file" must not masquerade as "almost finished".
+    """
+    rr = Path(repo_root) if repo_root is not None else Path.cwd()
+    track = "paper" if project.current_stage == Stage.PAPER_IN_PROGRESS else "research"
+    try:
+        fdir = project_store.feature_dir_for(
+            rr / "projects" / project.id, track=track
+        )
+        if fdir is None:
+            return _NO_TASKS_SENTINEL
+        text = (fdir / "tasks.md").read_text(encoding="utf-8")
+    except OSError:
+        return _NO_TASKS_SENTINEL
+    return text.count("- [ ]") + text.count("- [~]")
+
+
+#: Sorts a project with no readable tasks.md to the BACK of its implement queue.
+_NO_TASKS_SENTINEL = 1 << 30
+
+
+def _order_within_stage(
+    stage: Stage, projects: list[Project], *, repo_root: Path | None
+) -> list[Project]:
+    """Order a stage's queue for the matrix workers.
+
+    Implement stages are served SHORTEST-REMAINING-FIRST: the project closest to
+    done goes first. Everywhere else, oldest-waiting first (staleness) — there is no
+    "remaining work" to be short of.
+
+    Why implement is different: draining a project takes SEVERAL ticks (~38 tasks at
+    ~10/tick). Ordering by staleness made that a strict ROUND-ROBIN — working on a
+    project refreshes its ``updated_at`` and sends it to the back of the queue, so
+    each project waited for all 434 others between touches. Every project crept
+    forward and NONE ever finished: classic processor-sharing starvation, and the
+    reason zero projects had EVER crossed research_complete. Serving the shortest
+    remaining queue first turns the same compute into COMPLETED projects (it is
+    optimal for mean completion time), and it cannot starve the long ones: a finished
+    project LEAVES the stage, so the queue steadily advances to them.
+
+    Staleness remains the tiebreak, so equal-remaining projects still go
+    oldest-first.
+    """
+    if stage in _IMPLEMENT_STAGES:
+        return sorted(
+            projects,
+            key=lambda p: (_remaining_tasks(p, repo_root=repo_root), p.updated_at, p.id),
+        )
+    return sorted(projects, key=lambda p: (p.updated_at, p.id))
+
+
 def _apportion_workers(
     ranked_stages: list[Stage],
     by_stage: dict[Stage, list[Project]],
@@ -531,6 +594,22 @@ def _take(stage: Stage) -> None:
         if best is None:
             break  # every pile exhausted / capped
         _take(best)
+    # The share cap exists to stop one stage crowding OUT the others — never to idle
+    # a worker. If slots remain because every stage is at its cap (e.g. in_progress is
+    # the only occupied stage), hand them out ignoring the cap, bounded only by pile
+    # size. Otherwise a 1-stage repo would leave a third of the matrix doing nothing.
+    while len(slots) < n_workers:
+        best = None
+        best_score = -1.0
+        for s in ranked_stages:
+            if counts[s] >= len(by_stage[s]):
+                continue  # a worker here would pick nothing
+            score = weights[s] / (counts[s] + 1)
+            if score > best_score:
+                best_score, best = score, s
+        if best is None:
+            break  # genuinely fewer projects than workers
+        _take(best)
     return slots
 
 

tests/unit/test_scheduler_worker.py

@@ -241,3 +241,54 @@ def test_sparse_transient_stage_is_always_drained(tmp_path: Path) -> None:
     picks = [scheduler.pick_for_worker(i, 6, repo_root=repo) for i in range(6)]
     ids = [p.id for p in picks if p is not None]
     assert len(ids) == len(set(ids)), f"workers double-picked: {ids}"
+
+
+# --- in_progress is served SHORTEST-REMAINING first (completions, not sharing) ---
+#
+# Within a stage, projects were ordered stalest-`updated_at`-first. But working on a
+# project REFRESHES its updated_at and sends it to the back — so in_progress became a
+# strict round-robin across all 435 projects. Draining a project takes several ticks
+# (~38 tasks at ~10/tick), so each one had to wait for the other 434 between touches
+# and NOTHING ever completed: processor-sharing starvation, and the reason 0 projects
+# had ever crossed research_complete. Implement stages now serve the project CLOSEST
+# to done first, which converts the same effort into finished projects. It cannot
+# starve the big ones: a completed project LEAVES the stage, so the queue advances.
+
+
+def _impl_project(repo: Path, pid: str, *, open_tasks: int, age_days: float) -> None:
+    _make(repo, pid, Stage.IN_PROGRESS, age_days=age_days)
+    d = repo / "projects" / pid / "specs" / "001-t"
+    d.mkdir(parents=True, exist_ok=True)
+    lines = ["# Tasks", ""]
+    lines += [f"- [X] T{i:03d} done work" for i in range(3)]
+    lines += [f"- [ ] T{100 + i:03d} remaining work" for i in range(open_tasks)]
+    (d / "tasks.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def test_in_progress_serves_the_closest_to_done_first(tmp_path: Path) -> None:
+    for sub in ("projects", "run-log", "locks"):
+        (tmp_path / "state" / sub).mkdir(parents=True, exist_ok=True)
+    # The near-done project is the FRESHEST (worked on most recently) — under the old
+    # stalest-first order it would be picked LAST, which is exactly the bug.
+    _impl_project(tmp_path, "PROJ-9001-almost", open_tasks=2, age_days=1)
+    _impl_project(tmp_path, "PROJ-9002-midway", open_tasks=20, age_days=10)
+    _impl_project(tmp_path, "PROJ-9003-barely", open_tasks=60, age_days=30)
+
+    p0 = scheduler.pick_for_worker(0, 1, repo_root=tmp_path)
+    assert p0 is not None and p0.id == "PROJ-9001-almost", p0
+
+    # Successive workers walk outward: fewest-remaining first, then the rest.
+    picks = [scheduler.pick_for_worker(i, 3, repo_root=tmp_path) for i in range(3)]
+    got = [p.id for p in picks if p is not None]
+    assert got == ["PROJ-9001-almost", "PROJ-9002-midway", "PROJ-9003-barely"], got
+
+
+def test_non_implement_stages_still_ordered_by_staleness(tmp_path: Path) -> None:
+    """The change is scoped to the implement stages — elsewhere oldest-waiting still
+    wins (there is no 'remaining work' to be short of)."""
+    for sub in ("projects", "run-log", "locks"):
+        (tmp_path / "state" / sub).mkdir(parents=True, exist_ok=True)
+    _make(tmp_path, "PROJ-9101-fresh", Stage.BRAINSTORMED, age_days=1)
+    _make(tmp_path, "PROJ-9102-stale", Stage.BRAINSTORMED, age_days=40)
+    p0 = scheduler.pick_for_worker(0, 1, repo_root=tmp_path)
+    assert p0 is not None and p0.id == "PROJ-9102-stale", p0