fix(core): retire a window whose remaining rows FIFO refuses

The event-time window asked only whether its ten-minute range still held
*some* pending row before deciding the anchor was still active. A pending
row that FIFO holds back is not claimable: it may be blocked by an earlier
sibling of the same Harness whose own event time sits outside the range, and
that sibling can never be claimed from inside this window. The anchor then
stayed pinned forever and the whole queue starved.

Live shape on one machine: the anchor sat on 2026-10-07T03:24Z while the four
pending rows inside it were blocked by 929/930/1243/2347 earlier workbuddy
jobs, the earliest of them dated 2026-03-12. Completions stopped at 09:59 and
3.8k queued turns sat at zero output; restarting does not help because the
anchor is persisted.

The contract (docs/contracts/memory-event-time-scheduling-v1.md) retires a
window once every *eligible* pending/running row in the range has settled or
parked, and re-anchors on the next oldest eligible job. Reuse the claim's own
predicate — free lease, no busy Harness, no earlier unsent sibling — so "the
window is busy" and "the window can still be worked" finally agree. Anchor
selection and in-window selection already used it; only the liveness check
did not.

memhub/apps/core/src/one_memory/memory_server/durable_queue.py

@@ -2217,18 +2217,43 @@ class DurableWriteDrain:
                         f"AND active.source_key IN ({placeholders}) "
                     )
                     active_source_params = tuple(sorted(enabled_sources))
+                # The contract retires a window when every *eligible*
+                # pending/running job in it has settled or parked. A pending row
+                # that FIFO already holds back (an unexpired lease, a busy
+                # Harness, or an earlier unsent sibling) is not eligible, so it
+                # must not keep the anchor alive: those blockers usually sit
+                # outside this ten-minute range and can never be claimed from
+                # inside it, which pinned the anchor and starved the whole
+                # queue. Reuse the claim's own predicate so "the window is
+                # busy" and "the window can still be worked" agree.
                 active = connection.execute(
                     "SELECT 1 FROM memory_write_jobs AS active "
                     "WHERE active.event_at>=? AND active.event_at<? AND ("
                     "active.state=? OR (active.state=? "
+                    "AND (active.lease_expires_at IS NULL "
+                    "OR active.lease_expires_at < ?) "
                     + active_source_clause
-                    + ")) LIMIT 1",
+                    + "AND NOT EXISTS ("
+                    "SELECT 1 FROM memory_write_jobs AS held "
+                    "WHERE held.source_key=active.source_key AND held.state=? "
+                    "AND held.lease_owner IS NOT NULL "
+                    "AND held.lease_expires_at>=?) "
+                    "AND NOT EXISTS ("
+                    "SELECT 1 FROM memory_write_jobs AS earlier "
+                    "WHERE earlier.source_key=active.source_key "
+                    f"AND earlier.state IN ({blocking}) "
+                    "AND earlier.queue_seq<active.queue_seq) "
+                    ")) LIMIT 1",
                     (
                         anchor_text,
                         end_text,
                         RUNNING,
                         PENDING,
+                        now.isoformat(),
                         *active_source_params,
+                        RUNNING,
+                        now.isoformat(),
+                        *_FIFO_BLOCKING_STATES,
                     ),
                 ).fetchone()
                 if active is None:

memhub/apps/core/tests/test_memory_server_durable_queue.py

@@ -4812,3 +4812,83 @@ def test_readmit_paths_never_requeue_a_possibly_sent_row(tmp_path) -> None:
     assert rows["parked"]["state"] == WAITING_FOR_PERMISSION
     assert rows["model"]["state"] == WAITING_FOR_MODEL
     assert DurableWriteDrain(queue, execute=_never_execute).claim(limit=5) == ()
+
+
+def test_a_window_holding_only_blocked_rows_advances_instead_of_starving(
+    tmp_path,
+) -> None:
+    """A window whose remaining rows FIFO holds back must be retired.
+
+    Live shape: the anchor sat on 2026-10-07T03:24Z while the four pending rows
+    inside it were each blocked by 900+ earlier siblings of the same Harness
+    whose own event times sat outside the range (the earliest was 2026-03-12).
+    None of them could ever be claimed from inside that window, and because the
+    window only asked whether the range still held *some* pending row it never
+    retired: 3.8k queued turns sat at zero completions until restart, which
+    does not help because the anchor is persisted.
+    """
+    path = tmp_path / "one-memory.sqlite3"
+    queue = DurableWriteQueue(path)
+    anchor = datetime(2026, 10, 18, 8, 0, tzinfo=UTC)
+
+    # The anchor itself: another Harness, claimable, inside the range.
+    opener = queue.submit(
+        _request(key="starve-opener", source="codex", event_at=anchor)
+    )
+    # Same Harness as the blocked row but an hour later: outside the range, and
+    # it sorts earlier in queue_seq, so FIFO holds the row below behind it.
+    older_seq_newer_time = queue.submit(
+        _request(
+            key="starve-head",
+            source="workbuddy",
+            event_at=anchor + timedelta(hours=1),
+        )
+    )
+    blocked = queue.submit(
+        _request(
+            key="starve-blocked",
+            source="workbuddy",
+            event_at=anchor + timedelta(minutes=5),
+        )
+    )
+
+    drain = DurableWriteDrain(
+        queue,
+        execute=_never_execute,
+        owner="starve-owner",
+        max_concurrent_sources=10,
+    )
+
+    # The blocked row is inside the window but FIFO-refused, so it is not
+    # claimed alongside the opener.
+    assert drain.claim(limit=10) == (opener.job_id,)
+    queue.settle(
+        opener.job_id,
+        state=COMPLETED,
+        hy_invoked=True,
+        owner="starve-owner",
+    )
+
+    # The window now holds exactly one row and that row can never be claimed
+    # from inside it. Retire the window and re-anchor on the blocker.
+    assert drain.claim(limit=10) == (older_seq_newer_time.job_id,)
+
+    with sqlite3.connect(path.as_uri() + "?mode=ro", uri=True) as db:
+        window = db.execute(
+            "SELECT window_anchor FROM memory_write_schedule_state "
+            "WHERE singleton=1"
+        ).fetchone()
+    assert window is not None
+    assert str(window[0]) == (
+        anchor + timedelta(hours=1)
+    ).astimezone(UTC).isoformat()
+
+    queue.settle(
+        older_seq_newer_time.job_id,
+        state=COMPLETED,
+        hy_invoked=True,
+        owner="starve-owner",
+    )
+    # Once the blocker is gone the deferred row keeps its original event time
+    # and is reached on the next pass.
+    assert drain.claim(limit=10) == (blocked.job_id,)