adeelahmad · Nov 5, 2026
diff --git a/changelog.d/fix-queue-readiness.md b/changelog.d/fix-queue-readiness.md
@@ -0,0 +1,3 @@
+Queued pipelines now wait for deferred components before later steps, not just before their first step, and resume
+without repeating completed work. Live updates keep the Queued view in worker priority order before applying the
+page limit, so new arrivals no longer displace runs that should execute sooner.
diff --git a/docs/processing.md b/docs/processing.md
@@ -629,7 +629,9 @@ with examples.
 Imports, pipeline runs and folder scans return at once and run as jobs stored in SurrealDB. The server runs
 `workers.inline` workers itself; more can run anywhere that reaches the database, each limited to the steps it can do.
 A job whose next step a worker can't run goes back on the queue for one that can, so a Mac can transcribe with mlx while
-the container does the rest. Jobs can be cancelled and retried from the failed step; a job whose worker stops
+the container does the rest. This also applies to later steps waiting for components to be fetched: completed steps
+stay completed, and the job resumes at the waiting step once its component is ready. Jobs can be cancelled and
+retried from the failed step; a job whose worker stops
 responding is retried. In Activity the queue is handled as a whole too: **Clear queue** cancels every queued and held
 run you may process (in the namespace in view), asking first, and its message can undo it; under **Queued** the runs
 show in the order workers take them, and dragging one by its grip (or Alt+↑ and Alt+↓ on the grip, or Move to top

diff --git a/fastapi_backend/app/domain/jobs.py b/fastapi_backend/app/domain/jobs.py
@@ -1248,7 +1248,9 @@ def run_once(self):
             "lens.pipeline.version": pipe.get("version"),
         }
         with telemetry.span("job", attrs) as sp:
-            outcome = run_job(self.db, self.cfg_fn, job, self.name, self.can, self.log)
+            # Keep the readiness gate for every step, not just the first one we claimed. A later
+            # step whose component is still deferred must be queued so the keeper sees its demand.
+            outcome = run_job(self.db, self.cfg_fn, job, self.name, can, self.log)
             sp.set_attribute("lens.job.outcome", outcome)
         telemetry.record("lens.jobs", 1, {"lens.job.outcome": outcome})
         self.register()

diff --git a/fastapi_backend/tests/api/test_jobs.py b/fastapi_backend/tests/api/test_jobs.py
@@ -518,3 +518,26 @@ def test_agents_work_the_queue_over_mcp(client, db, cfg, folder):
     db.q("UPDATE $j SET status = 'failed'", j=store.R("job", ja))
     assert [r["job_id"] for r in tool(client, he, "runs", status="failed")["runs"]] == [ja]
     assert tool(client, he, "retry_runs")["jobs"] == [ja]
+
+
+def test_later_steps_wait_for_their_components(db, cfg, folder, monkeypatch):
+    """A ready first step must not bypass a later step's deferred model download."""
+    a = seed(db, cfg, folder)[0]
+    ran = []
+    monkeypatch.setitem(jobs.STEPS, "analyze", lambda *args: ran.append("analyze"))
+    monkeypatch.setitem(jobs.STEPS, "report", lambda *args: ran.append("report"))
+    jid = jobs.enqueue(db, a, ["analyze", "report"])
+    blocked = {"report"}
+    worker = jobs.Worker(db, lambda: cfg, name="fetching", steps=["analyze", "report"])
+    worker.keeper = type("Fetching", (), {"blocked": lambda self: blocked.copy()})()
+
+    assert worker.run_once()
+    waiting = jobs.get(db, jid)
+    assert (waiting["status"], waiting["step_index"], waiting["next_step"]) == ("queued", 1, "report")
+    assert ran == ["analyze"]
+    assert not worker.run_once()  # nothing runnable until the component is ready
+
+    blocked.clear()
+    assert worker.run_once()
+    assert jobs.get(db, jid)["status"] == "succeeded"
+    assert ran == ["analyze", "report"]  # resume, without repeating completed steps
diff --git a/nextjs-frontend/__tests__/job-events.test.ts b/nextjs-frontend/__tests__/job-events.test.ts
@@ -10,6 +10,34 @@ const list = (jobs: Partial<JobRecord>[], counts: Record<string, number>): JobLi
   ({ jobs, counts, running: false, log: "" }) as unknown as JobList;
 
 describe("live changes to cached job lists", () => {
+  it("keeps queue priority before limiting live rows", () => {
+    const qc = new QueryClient();
+    const key = ["jobs", { status: "queued", order: "queue", limit: 2 }];
+    const first = { id: 1, status: "queued", priority: 2, created_at: "2026-01-01" } as JobRecord;
+    const second = { id: 2, status: "queued", priority: 1, created_at: "2026-01-02" } as JobRecord;
+    qc.setQueryData(key, list([first, second], { queued: 2 }));
+    const ids = () => qc.getQueryData<JobList>(key)?.jobs.map((j) => j.id);
+
+    // An arrival outside this page must not displace a run workers will take sooner.
+    patchJobLists(qc, { id: 3, status: "queued", priority: 0, created_at: "2026-01-03" } as JobRecord);
+    expect(ids()).toEqual([1, 2]);
+    // Reprioritisation moves an existing row, and can bring an off-page run to the front.
+    patchJobLists(qc, { ...second, priority: 3 });
+    expect(ids()).toEqual([2, 1]);
+    patchJobLists(qc, { id: 3, status: "queued", priority: 4 } as JobRecord);
+    expect(ids()).toEqual([3, 2]);
+    patchJobLists(qc, { ...second, status: "running" });
+    expect(ids()).toEqual([3]);
+  });
+
+  it("keeps oldest queued runs first when priorities tie", () => {
+    const qc = new QueryClient();
+    const key = ["jobs", { status: "queued", order: "queue", limit: 2 }];
+    qc.setQueryData(key, list([{ id: 1, status: "queued", created_at: "2026-01-01" }], { queued: 1 }));
+    patchJobLists(qc, { id: 2, status: "queued", created_at: "2026-01-02" } as JobRecord);
+    expect(qc.getQueryData<JobList>(key)?.jobs.map((j) => j.id)).toEqual([1, 2]);
+  });
+
   it("keeps a namespace's or a batch's list to its own jobs", () => {
     const qc = new QueryClient();
     const all = ["jobs", { limit: 200 }];

diff --git a/nextjs-frontend/components/activity/job-events.ts b/nextjs-frontend/components/activity/job-events.ts
@@ -10,7 +10,7 @@ import { useSession } from "next-auth/react";
 import { useEffect, useMemo, useRef, useSyncExternalStore } from "react";
 
 import type { JobList } from "@/app/openapi-client/types.gen";
-import { applyCount, applyJobEvent, type JobRecord } from "@/components/activity/job-model";
+import { applyCount, applyJobEvent, byQueue, type JobRecord } from "@/components/activity/job-model";
 import { streamSSE } from "@/lib/api/sse";
 import { useArchive } from "@/lib/hooks/session";
 
@@ -146,7 +146,8 @@ export function useJobEvents(onJob?: Listener, onOpen?: () => void): FeedStatus
 }
 
 type JobsKeyOpts =
-  { status?: string; recording?: number; limit?: number; namespace?: string; batch?: number } | undefined;
+  | { status?: string; recording?: number; limit?: number; namespace?: string; batch?: number; order?: "queue" }
+  | undefined;
 
 /**
  * Patch every cached job list (query keys ["jobs", {status, recording, limit, namespace, batch}]) with one change.
@@ -165,6 +166,8 @@ export function patchJobLists(qc: QueryClient, job: JobRecord, spaceIds?: Record
       if (!old || !Array.isArray(old.jobs)) return old;
       const before = old.jobs.find((j) => j.id === job.id)?.status;
       let jobs = applyJobEvent(old.jobs as JobRecord[], job, statuses);
+      // Sort before truncating: a new run joins the back, not the front of a full queue page.
+      if (opts?.order === "queue") jobs = [...jobs].sort(byQueue);
       if (opts?.limit && jobs.length > opts.limit) jobs = jobs.slice(0, opts.limit);
       const counts = opts?.recording != null ? old.counts : applyCount(old.counts ?? {}, before, job.status);
       const running = Boolean((counts.queued ?? 0) + (counts.running ?? 0));