changing move order to move, then release, to have the messages already in the target folder, when release might fail.

Removed delete operation in Release & Report option.
Improved queueing to not block further jobs, when wait is active in job.
updated docs

AGENTS.md

@@ -35,7 +35,7 @@ venv/bin/python -m pytest       # 57 tests, no Okta tenant or real config needed
 | `app.py` | `create_app(store, queue, prefs)` — Flask routes. **No import-time side effects.** |
 | `config_store.py` | `ConfigStore`: tomlkit load/validate/atomic-write, immutable snapshots, PPSClient rebuild. The heart of the hot-reload design. |
 | `auth.py` | Okta OIDC + static dev mode; `login_required`/`admin_required`; `is_allowed`/`safe_next` (pure); CSRF. |
-| `pipeline.py` | Data-driven Report & Release step registry (release → move → delete). No Flask. |
+| `pipeline.py` | Data-driven Report & Release step registry (move → release). No Flask. |
 | `worker.py` | `JobQueue`: SQLite-backed background job thread; ops audit log. |
 | `prefs.py` | `PrefStore`: per-user preferences in jobs.db (key-value). |
 | `admin.py` | `/admin` page + `/api/admin/*` (config, pps-test, jobs, logs). |
@@ -59,16 +59,25 @@ venv/bin/python -m pytest       # 57 tests, no Okta tenant or real config needed
    message is moved/deleted. The `guid` is global and stable. To act on a message after a
    relocating step, re-find its current `localguid` by matching `guid` (`pipeline.refind_localguids`).
 
-4. **Report & Release: release FIRST, with NO `deletedfolder`.** `release` must run before
-   `move`, and must not pass `deletedfolder`, so the message stays in place and its
-   `localguid` survives for the move. Passing `deletedfolder` on release relocates the
-   message and invalidates every localguid — the exact bug the guid-refind fallback exists
-   to paper over. See the comment in `pipeline._release`.
+4. **Report & Release: move FIRST, then re-read, then release — and `release` never gets
+   a `deletedfolder`.** The order is `move → release` (`pipeline.ALLOWED_STEPS`); the
+   pipeline has no delete step — Delete is a separate action, so Report & Release can
+   never destroy mail as a side effect.
+   Handles are never carried across steps: after the configured delay the worker re-reads
+   each message's current `localguid` by its stable `guid` (`pipeline.locate_messages`),
+   because PPS's backend is eventually consistent and acting on a cached handle after a
+   relocation is the race this ordering exists to kill. `release` must still not pass
+   `deletedfolder` — that would relocate the message a second time. See `pipeline._release`.
+
+   Configs written against the old `release → move` order are re-ordered on load, and a
+   retired `delete` step is dropped (`config_store._migrate`) — neither is rejected.
 
 5. **Snapshot per job / per request.** Config is an immutable `Snapshot`; `apply()` builds a
    new one and rebinds one attribute (atomic under the GIL). A worker reads one snapshot at
    job start and threads it through, so a mid-job admin edit affects the *next* job, never a
-   job in flight. Never re-read `store.snapshot()` inside a running job.
+   job in flight. Never re-read `store.snapshot()` inside a running job. A job that resumes
+   after a step delay necessarily takes a fresh snapshot, so the load-bearing part of its
+   plan (`steps`, `delay`) is frozen into `jobs.state` on the first run and reused.
 
 6. **WAL + busy_timeout are mandatory.** Two connections (queue + prefs) share `jobs.db`.
    Without `PRAGMA journal_mode=WAL` and `busy_timeout`, a prefs write concurrent with the
@@ -94,6 +103,12 @@ venv/bin/python -m pytest       # 57 tests, no Okta tenant or real config needed
 10. **Log source is an enum, never a path.** `/api/admin/logs?source=ops|app` resolves to a
     path server-side. A `?path=` parameter would be arbitrary file read — never add one.
 
+11. **The worker thread never sleeps.** There is exactly one, so a `time.sleep` in it stalls
+    every queued job. Work that has to wait (the Report & Release step delay) persists its
+    progress and goes back to `status='pending'` with a future `jobs.run_after`
+    (`JobQueue._defer`); `_claim_next` skips such rows until the timestamp passes. Anything
+    else that needs to wait must use the same mechanism, not a sleep.
+
 ## Config zones
 
 `config.toml` is one file with two zones. The admin panel writes only the admin zone and

README.md

@@ -34,9 +34,12 @@ job queue, and the per-user prefs store, starts one background worker thread, an
 waitress. The browser fires an action and gets an immediate `202`; the worker drains the
 queue and calls PPS. Jobs survive a restart. **Run as a single process only.**
 
-Report & Release is a configurable pipeline of steps (`release → move → delete`, pick any
-subset) — release delivers the mail in place, move copies it to a report folder, delete
-removes it. Configured with checkboxes in the admin panel.
+Report & Release is a configurable pipeline of steps (`move → release`, pick either or
+both) — move files the mail into a report folder, release delivers it from there. It never
+deletes; Delete is its own action. Configured with checkboxes in the admin panel. Between
+steps the job waits
+`step_delay_seconds` for PPS's backend to settle and then re-reads where the messages
+actually are; while it waits it is parked back on the queue, so other jobs keep running.
 
 ## Documentation
 

config.example.toml

@@ -37,13 +37,16 @@ default_sort_field = "subject"                 # subject|date|from|rcpt (per-use
 default_sort_dir = "asc"                       # asc|desc (per-user overridable)
 
 [quarantine.report_release]                    # (A) "Report & Release" pipeline
-# Steps run in the FIXED order release -> move -> delete; pick any subsequence.
-# release delivers the mail in place; move relocates a copy to move_target; delete removes.
-steps = ["release", "move"]
+# Steps run in the FIXED order move -> release; pick either or both.
+# move relocates the mail to move_target; release then delivers it from there.
+# Deleting is NOT part of this pipeline — Delete is its own action/button.
+steps = ["move", "release"]
 move_target = "Debugging - Josef"              # must be one of `folders`
 # Seconds to wait between steps on the same message so PPS's eventually-consistent
-# backend settles (e.g. before moving a just-released message). 0..3600; 0 disables.
-# The wait blocks the background worker, not the HTTP request.
+# backend settles. After the wait the worker re-reads where the messages actually are
+# (by their stable guid) instead of reusing handles from before the move. 0..3600;
+# 0 disables both the wait and the re-read. A waiting job is parked back on the queue,
+# so it blocks neither the HTTP request nor other queued jobs.
 step_delay_seconds = 60
 
 [app]                                          # (S) restart required for changes here

config_store.py

@@ -71,6 +71,13 @@ _PPS_FINGERPRINT = (
     "base_url", "username", "password", "verify_tls", "timeout", "client_cert", "client_key",
 )
 
+# Report & Release step order before it flipped to move-first (`pipeline.ALLOWED_STEPS`).
+# Configs written against it are migrated, not rejected — see `_migrate`.
+_LEGACY_STEP_ORDER: tuple[str, ...] = ("release", "move", "delete")
+
+# Steps the pipeline used to offer and no longer does; dropped from old configs on load.
+_RETIRED_STEPS: frozenset[str] = frozenset({"delete"})
+
 _SORT_FIELDS = frozenset({"subject", "date", "from", "rcpt"})
 _SORT_DIRS = frozenset({"asc", "desc"})
 _ROLES = frozenset({"admin", "user"})
@@ -230,6 +237,9 @@ class ConfigStore:
             if not applied:
                 return ApplyResult(self._version, [], [])
 
+            # Normalise the merged result (e.g. a step list still in the pre-flip
+            # release-before-move order) before validating and writing it.
+            _migrate(merged)
             _validate(merged)
 
             for dotted in applied:
@@ -336,18 +346,32 @@ def _set_dotted_doc(doc, dotted: str, value) -> None:
 def _migrate(data: dict) -> dict:
     """In-memory back-compat. Never rewrites the user's file on load.
 
-    `quarantine.report_release_folder` (PoC) -> `[quarantine.report_release]` with
-    steps=["release","move"], which reproduces the old hardcoded behaviour exactly.
+    `quarantine.report_release_folder` (PoC) -> `[quarantine.report_release]` with the
+    move+release pipeline. Two shapes of retired config are rewritten rather than
+    rejected: a `delete` step (Report & Release no longer deletes — Delete is its own
+    action) and a step list in the pre-flip order (release before move).
     """
+    import pipeline  # local import: pipeline imports nothing from us
+
     q = data.setdefault("quarantine", {})
     if "report_release" not in q:
         legacy = q.get("report_release_folder")
         q["report_release"] = {
-            "steps": ["release", "move"] if legacy else ["release"],
+            "steps": ["move", "release"] if legacy else ["release"],
             "move_target": legacy or "",
         }
     rr = q["report_release"]
-    rr.setdefault("steps", ["release", "move"])
+    rr.setdefault("steps", ["move", "release"])
+    steps = rr.get("steps")
+    if isinstance(steps, (list, tuple)) and any(s in _RETIRED_STEPS for s in steps):
+        steps = [s for s in steps if s not in _RETIRED_STEPS] or ["release"]
+        log.warning("dropped retired Report & Release step(s) %s — Delete is a separate "
+                    "action; steps are now %s", sorted(_RETIRED_STEPS), steps)
+        rr["steps"] = steps
+    if (isinstance(steps, (list, tuple))
+            and _is_subsequence(steps, _LEGACY_STEP_ORDER)
+            and not _is_subsequence(steps, pipeline.ALLOWED_STEPS)):
+        rr["steps"] = sorted(steps, key=pipeline.ALLOWED_STEPS.index)
     rr.setdefault("move_target", q.get("report_release_folder", "") or "")
     rr.setdefault("step_delay_seconds", 60)
     q.setdefault("default_sort_field", "subject")
@@ -466,8 +490,6 @@ def _validate(data: dict) -> None:
                     "quarantine.report_release.move_target",
                     f"{target!r} is not in the folder list",
                 )
-        if "delete" in steps and not q.get("deleted_folder"):
-            bad("quarantine.deleted_folder", "required when 'delete' is selected")
     try:
         d = int(rr.get("step_delay_seconds", 60))
         if not 0 <= d <= 3600:

docs/architecture.md

@@ -40,12 +40,18 @@ snapshot. Readers are lock-free (attribute rebind is atomic under the GIL). See
 
 ## Background jobs
 
-- Tables: `jobs(id, action, folder, user, extra, status, processed, total, error, …)` and
-  `job_items(job_id, localguid, guid, folder, subject, sender, recipient)`.
-- The worker claims the lowest `pending` job, snapshots config once, chunks the items
-  (`chunk_size`), calls `pipeline.run_pipeline` (Report & Release) or a single PPS action,
-  writes one **audit line per message** to `worker.log`, and marks the job done/failed.
-- `retry(job_id)` requeues a failed job.
+- Tables: `jobs(id, action, folder, user, extra, status, processed, total, error,
+  run_after, state, …)` and `job_items(job_id, localguid, guid, folder, subject, sender,
+  recipient)`.
+- The worker claims the lowest `pending` job whose `run_after` has passed, snapshots config
+  once, chunks the items (`chunk_size`), runs one PPS action — or, for Report & Release,
+  one `pipeline.run_step` per configured step — writes one **audit line per message per
+  step** to `worker.log`, and marks the job done/failed.
+- **Waiting doesn't block the queue.** Between two Report & Release steps the job is parked
+  (`_defer`): progress goes into `jobs.state`, `run_after` is set to now + delay, and the
+  worker picks up other jobs until the wait is over. On resume it re-reads the messages'
+  current `localguid`s by `guid` — cached handles are never reused across a step.
+- `retry(job_id)` requeues a failed job from the top (clears `state`/`run_after`).
 
 ## Data stores
 

docs/configuration.md

@@ -39,12 +39,21 @@ previous save.
 
 | Key | Type | Notes |
 |-----|------|-------|
-| `steps` | list[str] | Subsequence of `["release","move","delete"]`, run in that fixed order. |
+| `steps` | list[str] | Subsequence of `["move","release"]`, run in that fixed order. |
 | `move_target` | str | Required if `move` in steps; must be in `folders`. |
+| `step_delay_seconds` | int | 0..3600 (default 60). Wait between steps, then re-read locations. |
 
-Outcomes: `["release"]` delivers in place; `["release","move"]` delivers + copies to
-`move_target`; `["release","delete"]` delivers + removes to `deleted_folder`;
-`["release","move","delete"]` delivers, moves, then removes.
+Outcomes: `["release"]` delivers in place; `["move","release"]` files the mail in
+`move_target` and delivers it from there. Report & Release never deletes — Delete is a
+separate action; a `delete` step in an old config is dropped on load.
+
+Between two steps the worker waits `step_delay_seconds` for PPS's eventually-consistent
+backend to settle and then **re-reads** each message's current `localguid` by its stable
+`guid` — handles captured before the move are never reused. `0` disables both the wait and
+the re-read. The waiting job is parked back on the queue, so other jobs keep running.
+
+A `steps` list in the old `release`-before-`move` order is migrated to the current order
+on load (and a `delete` entry is removed).
 
 ## `[app]` (S, restart — except `log_level`)
 

docs/development.md

@@ -50,7 +50,7 @@ Suites:
 | `test_config_store.py` | round-trip (comments preserved), atomic write/perms/backup, redaction, validation, restart-vs-live, client rebuild, migration, isolation guard |
 | `test_auth.py` | `is_allowed`/`safe_next` (pure), static login, CSRF, admin gate |
 | `test_oidc.py` | OIDC callback allow/deny/role/demotion (monkeypatched token) |
-| `test_report_release.py` | pipeline: no-`deletedfolder` rule, lazy refind, fallback, destinations |
+| `test_report_release.py` | pipeline: move-first order, fresh re-read per step, no-`deletedfolder` rule, fallback, destinations, deferred (non-blocking) step delay |
 | `test_paging.py` | cursor from raw batch, `has_more` boundary, `_clamp_limit` |
 | `test_prefs.py` | override/default/fallback/isolation |
 | `test_jobs_visibility.py` | user attribution, `targetfolder` validation, per-user scoping |

docs/operations.md

@@ -36,8 +36,10 @@ TLS-terminating reverse proxy (nginx) in front; set `[app] cookie_secure = true`
 - **Application log** (`app.log`): requests, job lifecycle, PPS API calls, errors. Viewable
   in the admin panel (Logs → Application) with errors in red, or `tail -f app.log`.
 - **Operations/audit log** (`worker.log`): one line per message acted on, e.g.
-  `job=#12 user=alice@x action=report_release result=ok src='Quarantine' dst='Debugging - Josef' localguid=… guid=… from=… rcpt=… subject=…`.
+  `job=#12 user=alice@x action=report_release step=move result=ok src='Quarantine' dst='Debugging - Josef' localguid=… guid=… from=… rcpt=… subject=…`.
   This is the "who released/deleted what" audit trail. Grep it: `grep result=FAILED worker.log`.
+  Report & Release writes one line per **step** (`step=move`, then `step=release`), each
+  with that step's own `src`/`dst`; single actions have no `step=` field.
 
 Both rotate at 5 MB × 5 files. The PPS `x-pps-reqid` appears in error lines for
 cross-referencing PPS's own webservices log.
@@ -47,7 +49,10 @@ cross-referencing PPS's own webservices log.
 - View in the admin panel (Jobs) — all users, with status and progress. Failed jobs have a
   **Retry** button (`POST /api/admin/jobs/<id>/retry`).
 - Jobs persist in `jobs.db` and survive a restart; a job interrupted mid-run is requeued
-  automatically on boot.
+  automatically on boot and resumes at the pipeline step it had reached.
+- A Report & Release job shows as **waiting** while it sits out
+  `step_delay_seconds` between two steps. It is not stuck: other jobs run during the wait,
+  and it is re-claimed within a second of the delay expiring.
 
 ## Backup
 

pipeline.py

@@ -1,30 +1,37 @@
 """Configurable Report & Release pipeline.
 
 "Report & Release" is a sequence of steps run per message chunk, chosen in the admin
-panel by checkboxes. The order is FIXED: release → move → delete. Config only toggles
-which steps are included (a subsequence of ALLOWED_STEPS).
-
-Why the order is load-bearing (see AGENTS.md):
-
-* `release` MUST run first and MUST NOT pass `deletedfolder`. Releasing in place leaves
-  the message in its folder so its `localguid` stays valid for a following `move`.
-  Passing `deletedfolder` would relocate the message and invalidate every localguid —
-  the exact bug the guid-refind fallback exists to paper over.
-* `move`/`delete` relocate the message, so afterwards the localguids are stale. A step
-  returns `localguids=None` to signal that; the runner re-finds them by the stable
-  `guid` — but LAZILY, only when another step actually follows. So the common
-  `["release", "move"]` still issues exactly two PPS calls, identical to the PoC.
-
-Each step acts on the message's CURRENT location, threading (folder, localguids)
-forward. Adding a future action = one `@step(...)` function + an entry in ALLOWED_STEPS;
-no worker or frontend change.
+panel by checkboxes. The order is FIXED: move → release. Config only toggles which steps
+are included (a subsequence of ALLOWED_STEPS).
+
+Deleting is deliberately NOT part of this pipeline — Delete is its own action/button, so
+Report & Release can never destroy mail as a side effect.
+
+Why the order and the re-reads are load-bearing (see AGENTS.md):
+
+* `move` runs FIRST, straight out of the quarantine folder, using the localguids the job
+  was enqueued with. The message reaches its destination folder before anything else
+  touches it.
+* `release` then runs where the message NOW is, and MUST NOT pass `deletedfolder` —
+  that would relocate it again and invalidate every localguid.
+* localguids are NEVER carried from one step to the next. They are folder-local, and
+  PPS's backend is eventually consistent: after a relocating step the cached handles are
+  stale, and acting on stale handles is exactly the race this pipeline kept hitting.
+  Every step after the first waits `step_delay_seconds` and then re-reads where the
+  messages actually are, by their stable `guid` (`locate_messages`).
+
+The worker drives one step at a time via `run_step` and parks the job on the queue
+during the delay, so a waiting pipeline never blocks jobs queued behind it
+(`worker.JobQueue._process`).
+
+Each step acts on the message's CURRENT location. Adding a future action = one
+`@step(...)` function + an entry in ALLOWED_STEPS; no worker or frontend change.
 """
 
 from __future__ import annotations
 
 import logging
-import time
-from dataclasses import dataclass, replace
+from dataclasses import dataclass
 from typing import Any, Callable, Mapping
 
 from pps_client import PPSError
@@ -32,11 +39,12 @@ from pps_client import PPSError
 log = logging.getLogger("pps.pipeline")
 
 # Canonical order. Also the validation whitelist (config_store imports this).
-ALLOWED_STEPS: tuple[str, ...] = ("release", "move", "delete")
+ALLOWED_STEPS: tuple[str, ...] = ("move", "release")
 
 # Seconds to wait between two steps on the same message when config omits the setting.
-# PPS's backend is eventually-consistent: acting again too soon (e.g. move right after
-# release) can hit a stale view. The delay lets the backend catch up.
+# PPS's backend is eventually-consistent: acting again too soon (e.g. releasing a
+# just-moved message) can hit a stale view. The delay lets the backend catch up — and
+# only after it has passed do we re-read where the messages are.
 DEFAULT_STEP_DELAY = 60
 
 
@@ -45,14 +53,13 @@ class StepContext:
     pps: Any
     cfg: Mapping[str, Any]           # the quarantine config section
     folder: str                      # where the messages are RIGHT NOW
-    localguids: list[str] | None     # None => stale; runner must re-find by guid
+    localguids: list[str]            # freshly read handles, valid in `folder`
     chunk: list[dict]                # job_items rows: localguid, guid, subject, ...
 
 
 @dataclass(frozen=True)
 class StepResult:
-    folder: str
-    localguids: list[str] | None
+    folder: str                      # where the messages are after this step
     dst: str | None                  # folder this step lands messages in (for ops log)
 
 
@@ -67,13 +74,6 @@ def step(name: str):
     return register
 
 
-@step("release")
-def _release(ctx: StepContext) -> StepResult:
-    # NO deletedfolder — this is the localguid-stability rule. Never add it here.
-    ctx.pps.act("release", ctx.folder, ctx.localguids)
-    return StepResult(folder=ctx.folder, localguids=ctx.localguids, dst=None)
-
-
 @step("move")
 def _move(ctx: StepContext) -> StepResult:
     target = ctx.cfg["report_release"]["move_target"]
@@ -81,60 +81,88 @@ def _move(ctx: StepContext) -> StepResult:
         raise ValueError("report_release 'move' step requires move_target")
     try:
         ctx.pps.act("move", ctx.folder, ctx.localguids, targetfolder=target)
-    except PPSError:
-        # Some deployments' release relocates the message to the deleted folder (like the
-        # admin GUI). The move from `folder` then fails; re-find by stable guid and move
-        # from the deleted folder instead.
-        deleted = ctx.cfg.get("deleted_folder")
-        if not deleted:
-            raise PPSError("release relocated messages but no deleted_folder configured")
-        found = refind_localguids(ctx.pps, ctx.cfg, deleted, ctx.chunk)
-        ctx.pps.act("move", deleted, found, targetfolder=target)
-    return StepResult(folder=target, localguids=None, dst=target)
-
-
-@step("delete")
-def _delete(ctx: StepContext) -> StepResult:
-    deleted = ctx.cfg.get("deleted_folder")
-    ctx.pps.act("delete", ctx.folder, ctx.localguids, deletedfolder=deleted)
-    return StepResult(folder=deleted, localguids=None, dst=deleted)
-
-
-def run_pipeline(pps, cfg: Mapping[str, Any], folder: str, chunk: list[dict],
-                 sleep: Callable[[float], None] = time.sleep) -> None:
-    """Execute the configured steps in order for one chunk of messages.
-
-    Between two steps we wait `report_release.step_delay_seconds` (default 60) so PPS's
-    eventually-consistent backend settles before the next action on the same message —
-    without this, e.g. a move right after a release can act on a stale view. The wait
-    blocks the worker thread; it does NOT block the HTTP request (actions are queued and
-    return 202 immediately). `sleep` is injectable for tests.
+    except PPSError as exc:
+        # As the first step, `move` runs on the localguids captured when the job was
+        # enqueued — which may have gone stale in the meantime (or the message may
+        # already have been relocated). Re-read the messages' real location and retry
+        # once; a genuine failure raises again from `locate_messages` or the retry.
+        log.warning("move from %r failed (%s) — re-reading message locations", ctx.folder, exc)
+        folder, localguids = locate_messages(ctx.pps, ctx.cfg, ctx.folder, ctx.chunk)
+        ctx.pps.act("move", folder, localguids, targetfolder=target)
+    return StepResult(folder=target, dst=target)
+
+
+@step("release")
+def _release(ctx: StepContext) -> StepResult:
+    # NO deletedfolder — this is the localguid-stability rule. Never add it here.
+    ctx.pps.act("release", ctx.folder, ctx.localguids)
+    return StepResult(folder=ctx.folder, dst=None)
+
+
+# ---------------------------------------------------------------- driving the steps
+
+def job_steps(cfg: Mapping[str, Any]) -> list[str]:
+    """The configured steps, in canonical order."""
+    return list(cfg["report_release"]["steps"])
+
+
+def step_delay(cfg: Mapping[str, Any]) -> int:
+    """Seconds to wait between steps (0 disables the wait)."""
+    return int(cfg["report_release"].get("step_delay_seconds", DEFAULT_STEP_DELAY))
+
+
+def run_step(pps, cfg: Mapping[str, Any], name: str, folder: str, chunk: list[dict],
+             *, refresh: bool) -> str:
+    """Run one pipeline step for one chunk; return the folder the messages end up in.
+
+    `refresh=True` (every step after the first) discards the job's cached localguids and
+    re-reads the messages' current handles in `folder` by their stable guid. The caller
+    is responsible for having waited `step_delay` before asking for a refresh.
     """
-    steps = cfg["report_release"]["steps"]
-    delay = int(cfg["report_release"].get("step_delay_seconds", DEFAULT_STEP_DELAY))
-    ctx = StepContext(
-        pps=pps,
-        cfg=cfg,
-        folder=folder,
-        localguids=[it["localguid"] for it in chunk],
-        chunk=chunk,
-    )
-    for i, name in enumerate(steps):
-        # Delay before every step except the first — also lets the backend settle before
-        # the refind search below finds a just-relocated message.
-        if i > 0 and delay > 0:
-            log.info("waiting %ds before %r step (PPS backend consistency)", delay, name)
-            sleep(delay)
-        if ctx.localguids is None:  # a prior relocating step left them stale
-            ctx = replace(ctx, localguids=refind_localguids(pps, cfg, ctx.folder, chunk))
-        result = STEPS[name](ctx)
-        ctx = replace(ctx, folder=result.folder, localguids=result.localguids)
+    if refresh:
+        folder, localguids = locate_messages(pps, cfg, folder, chunk)
+    else:
+        localguids = [it["localguid"] for it in chunk]
+    result = STEPS[name](StepContext(pps=pps, cfg=cfg, folder=folder,
+                                     localguids=localguids, chunk=chunk))
+    return result.folder
+
+
+def step_destination(cfg: Mapping[str, Any], name: str, folder: str) -> str | None:
+    """Folder the messages are in after `name`, computed WITHOUT touching PPS."""
+    if name == "move":
+        return cfg["report_release"]["move_target"]
+    return folder
+
+
+def locate_messages(pps, cfg: Mapping[str, Any], folder: str,
+                    chunk: list[dict]) -> tuple[str, list[str]]:
+    """Re-read where `chunk`'s messages are right now: `(folder, localguids)`.
+
+    Looks in `folder` first; if nothing matches there and a `deleted_folder` is
+    configured, looks there too — some deployments' actions relocate messages to the
+    deleted folder behind our back. Raises PPSError if the messages are nowhere.
+    """
+    if not any(it.get("guid") for it in chunk):
+        raise PPSError("cannot locate messages: no guids stored for this chunk")
+    candidates = [folder]
+    deleted = cfg.get("deleted_folder")
+    if deleted and deleted != folder:
+        candidates.append(deleted)
+    for i, candidate in enumerate(candidates):
+        try:
+            return candidate, refind_localguids(pps, cfg, candidate, chunk)
+        except PPSError:
+            if i == len(candidates) - 1:
+                raise
+            log.warning("messages not in %r — looking in %r", candidate, candidates[i + 1])
+    raise AssertionError("unreachable")
 
 
 def refind_localguids(pps, cfg: Mapping[str, Any], folder: str, chunk: list[dict]) -> list[str]:
     """Recover current localguids in `folder` by matching the stable guid.
 
-    Generalises the PoC's _move_from_deleted_by_guid. Raises PPSError if nothing matches.
+    Raises PPSError if nothing matches.
     """
     wanted = {it["guid"]: it for it in chunk if it.get("guid")}
     if not wanted:
@@ -151,6 +179,7 @@ def refind_localguids(pps, cfg: Mapping[str, Any], folder: str, chunk: list[dict
     ]
     if not found:
         raise PPSError(f"messages not found in {folder!r} to continue the pipeline")
+    log.info("re-read %d/%d message(s) in %r", len(found), len(wanted), folder)
     return found
 
 
@@ -160,21 +189,17 @@ def pipeline_destination(cfg: Mapping[str, Any]) -> str | None:
     for name in cfg["report_release"]["steps"]:
         if name == "move":
             folder = cfg["report_release"]["move_target"]
-        elif name == "delete":
-            folder = cfg.get("deleted_folder")
     return folder
 
 
 def describe_pipeline(cfg: Mapping[str, Any]) -> str:
     """Human sentence for the admin panel, e.g.
-    'released in place → moved to "Debugging - Josef" → deleted to "Deleted"'."""
+    'moved to "Debugging - Josef" → released in place'."""
     rr = cfg.get("report_release", {})
     phrases = []
     for name in rr.get("steps", []):
-        if name == "release":
-            phrases.append("released in place")
-        elif name == "move":
+        if name == "move":
             phrases.append(f'moved to "{rr.get("move_target", "?")}"')
-        elif name == "delete":
-            phrases.append(f'deleted to "{cfg.get("deleted_folder", "?")}"')
+        elif name == "release":
+            phrases.append("released in place")
     return " → ".join(phrases) if phrases else "(no steps configured)"

static/admin.js

@@ -158,11 +158,9 @@ function currentSteps() {
 function updateRrSentence() {
   const steps = currentSteps();
   const move = $("rr-move_target").value;
-  const deleted = CONFIG.quarantine.deleted_folder;
   const parts = steps.map((s) =>
     s === "release" ? "released in place" :
-    s === "move" ? `moved to "${move}"` :
-    s === "delete" ? `deleted to "${deleted}"` : s);
+    s === "move" ? `moved to "${move}"` : s);
   $("rr-sentence").textContent = parts.length ? parts.join(" → ") : "(no steps selected)";
 }
 
@@ -247,7 +245,9 @@ async function refreshJobs() {
       const tr = document.createElement("tr");
       tr.innerHTML =
         `<td>${j.id}</td><td>${esc(j.user || "-")}</td><td>${esc(j.action)}</td>` +
-        `<td>${esc(j.folder)}</td><td class="st-${j.status}">${j.status}</td>` +
+        // A pending job with run_after is waiting out a Report & Release step delay.
+        `<td>${esc(j.folder)}</td><td class="st-${j.status}" title="${esc(j.run_after || "")}">` +
+        `${j.status === "pending" && j.run_after ? "waiting" : j.status}</td>` +
         `<td>${j.processed}/${j.total}</td>` +
         `<td>${j.status === "failed" ? `<button class="btn small" data-retry="${j.id}">Retry</button>` : ""}</td>`;
       body.append(tr);

static/app.js

@@ -422,11 +422,17 @@ function renderStatus(jobs, active) {
   }
   statusBar.className = "status-bar";
   statusBar.textContent = active
-    .map((j) => `${labelFor(j.action)}: ${j.processed}/${j.total} (${j.status})`)
+    .map((j) => `${labelFor(j.action)}: ${j.processed}/${j.total} (${jobState(j)})`)
     .join("   •   ");
   statusBar.classList.remove("hidden");
 }
 
+// A pending job with run_after set is waiting out a Report & Release step delay —
+// it isn't stuck behind the queue, it's the queue waiting for PPS to settle.
+function jobState(j) {
+  return j.status === "pending" && j.run_after ? "waiting" : j.status;
+}
+
 // --- misc helpers ----------------------------------------------------------
 function hideMoveMenu() { moveMenu.classList.add("hidden"); }
 

templates/admin.html

@@ -88,7 +88,7 @@
       <label class="field-block">Delay between steps (seconds)
         <input id="rr-step_delay_seconds" type="number" min="0" max="3600" step="1">
       </label>
-      <p class="hint">Waits this long between actions on the same message so PPS's backend settles (e.g. before moving a just-released message). 0 disables.</p>
+      <p class="hint">Waits this long between actions on the same message so PPS's backend settles, then re-reads where the messages actually are before the next step. The job is parked meanwhile — other jobs keep running. 0 disables.</p>
       <p class="rr-sentence" id="rr-sentence"></p>
       <button class="btn primary" data-save="rr">Save Report &amp; Release</button>
     </section>

tests/test_config_store.py

@@ -45,8 +45,6 @@ def test_blank_password_leaves_value_unchanged(store):
     ({"quarantine": {"default_folder": "Nope"}}, "quarantine.default_folder"),
     ({"quarantine": {"folders": ["a,b"]}}, "quarantine.folders"),
     ({"pps": {"base_url": "ftp://x"}}, "pps.base_url"),
-    ({"quarantine": {"report_release": {"steps": ["move", "release"]}}},
-     "quarantine.report_release.steps"),
     ({"quarantine": {"report_release": {"steps": ["bogus"]}}},
      "quarantine.report_release.steps"),
     ({"pps": {"timeout": 0}}, "pps.timeout"),
@@ -99,10 +97,26 @@ def test_migration_report_release_folder(tmp_path):
     )
     snap = ConfigStore(cfg).snapshot()
     rr = snap.quarantine["report_release"]
-    assert list(rr["steps"]) == ["release", "move"]
+    assert list(rr["steps"]) == ["move", "release"]
     assert rr["move_target"] == "Rep"
 
 
+def test_migration_reorders_pre_flip_steps(store):
+    """Configs written when release ran before move are re-ordered, not rejected."""
+    store.apply({"quarantine": {"report_release": {"steps": ["release", "move"]}}},
+                actor="t")
+    rr = store.snapshot().quarantine["report_release"]
+    assert list(rr["steps"]) == ["move", "release"]
+
+
+def test_migration_drops_the_retired_delete_step(store):
+    """Report & Release no longer deletes; an old config with the step still boots."""
+    store.apply({"quarantine": {"report_release": {"steps": ["release", "move", "delete"]}}},
+                actor="t")
+    rr = store.snapshot().quarantine["report_release"]
+    assert list(rr["steps"]) == ["move", "release"]
+
+
 def test_ppsq_config_guard_under_pytest(monkeypatch):
     monkeypatch.delenv("PPSQ_CONFIG", raising=False)
     monkeypatch.setenv("PYTEST_CURRENT_TEST", "x")

tests/test_report_release.py

@@ -1,8 +1,10 @@
 """The Report & Release pipeline — the highest-risk behaviour.
 
-Asserts on exact PPS call kwargs, not just counts: the localguid-stability rule (release
-must NOT pass deletedfolder) and the lazy guid-refind are both invisible to a call-count
-check but critical to correctness.
+Asserts on exact PPS call kwargs, not just counts: the move-first order, the re-read of
+localguids before every later step (never the cached ones), and the localguid-stability
+rule (release must NOT pass deletedfolder) are all invisible to a call-count check but
+critical to correctness. The last group covers the other half of the design — waiting
+between steps must not block the job queue.
 """
 
 import pytest
@@ -12,7 +14,7 @@ from pipeline import (
     ALLOWED_STEPS,
     describe_pipeline,
     pipeline_destination,
-    run_pipeline,
+    run_step,
 )
 from pps_client import PPSError
 
@@ -22,6 +24,7 @@ def _cfg(steps, move_target="Rep", deleted="Deleted", step_delay=0):
         "deleted_folder": deleted,
         "list_query": "from=*",
         "default_days_back": 7,
+        "chunk_size": 25,
         "report_release": {
             "steps": steps,
             "move_target": move_target,
@@ -37,111 +40,211 @@ def _chunk():
     ]
 
 
-def test_release_only_has_no_deletedfolder():
-    pps = FakePPS()
-    run_pipeline(pps, _cfg(["release"]), "Quarantine", _chunk())
-    acts = pps.acts()
-    assert len(acts) == 1
-    assert acts[0]["action"] == "release"
-    assert acts[0]["deletedfolder"] is None    # THE localguid-stability rule
-    assert acts[0]["folder"] == "Quarantine"
+def _moved_records():
+    """What a search of the move target returns once the messages have landed there."""
+    return [{"guid": "g1", "localguid": "R:1"}, {"guid": "g2", "localguid": "R:2"}]
+
 
+# ---------------------------------------------------------------- single steps
 
-def test_release_move_is_two_calls_zero_searches():
+def test_first_step_uses_the_job_localguids_without_searching():
+    pps = FakePPS()
+    folder = run_step(pps, _cfg(["move", "release"]), "move", "Quarantine", _chunk(),
+                      refresh=False)
+    assert folder == "Rep"
+    assert pps.searches() == []
+    move = pps.acts()[0]
+    assert move["action"] == "move" and move["folder"] == "Quarantine"
+    assert move["localguids"] == ["6:6:1", "6:6:2"]
+    assert move["targetfolder"] == "Rep"
+
+
+def test_later_step_rereads_and_never_uses_cached_localguids():
+    pps = FakePPS(records=_moved_records())
+    folder = run_step(pps, _cfg(["move", "release"]), "release", "Rep", _chunk(),
+                      refresh=True)
+    assert folder == "Rep"
+    assert [s["folder"] for s in pps.searches()] == ["Rep"]   # re-read in the NEW folder
+    release = pps.acts()[0]
+    assert release["action"] == "release"
+    assert release["folder"] == "Rep"
+    assert release["localguids"] == ["R:1", "R:2"]            # fresh, not "6:6:*"
+    assert release["deletedfolder"] is None                   # THE stability rule
+
+
+def test_release_carries_no_deletedfolder_even_as_first_step():
     pps = FakePPS()
-    run_pipeline(pps, _cfg(["release", "move"]), "Quarantine", _chunk())
-    assert [a["action"] for a in pps.acts()] == ["release", "move"]
+    run_step(pps, _cfg(["release"]), "release", "Quarantine", _chunk(), refresh=False)
     assert pps.acts()[0]["deletedfolder"] is None
-    assert pps.acts()[1]["targetfolder"] == "Rep"
-    assert pps.acts()[1]["folder"] == "Quarantine"
-    assert pps.searches() == []                 # lazy refind: no search needed
-
-
-def test_move_fallback_refinds_in_deleted_folder():
-    class MovingPPS(FakePPS):
-        def act(self, action, folder, localguids, *, targetfolder=None, deletedfolder=None, scan=False):
-            if action == "move" and folder == "Quarantine":
-                self.calls.append({"m": "act", "action": action, "folder": folder,
-                                   "localguids": list(localguids), "targetfolder": targetfolder,
-                                   "deletedfolder": deletedfolder, "scan": scan})
-                raise PPSError("release relocated it")
-            return super().act(action, folder, localguids, targetfolder=targetfolder,
-                               deletedfolder=deletedfolder, scan=scan)
-
-    pps = MovingPPS(records=[{"guid": "g1", "localguid": "D:1"}, {"guid": "g2", "localguid": "D:2"}])
-    run_pipeline(pps, _cfg(["release", "move"]), "Quarantine", _chunk())
-    # release, failed move on Quarantine, search Deleted, move from Deleted
-    assert [s["folder"] for s in pps.searches()] == ["Deleted"]
-    moves = [a for a in pps.acts() if a["action"] == "move"]
-    assert moves[-1]["folder"] == "Deleted"
-    assert sorted(moves[-1]["localguids"]) == ["D:1", "D:2"]
-
-
-def test_release_move_delete_refinds_between_move_and_delete():
-    pps = FakePPS(records=[{"guid": "g1", "localguid": "R:1"}, {"guid": "g2", "localguid": "R:2"}])
-    run_pipeline(pps, _cfg(["release", "move", "delete"]), "Quarantine", _chunk())
-    actions = [a["action"] for a in pps.acts()]
-    assert actions == ["release", "move", "delete"]
-    # delete happens in the move_target after a refind there
-    assert len(pps.searches()) == 1 and pps.searches()[0]["folder"] == "Rep"
-    delete = [a for a in pps.acts() if a["action"] == "delete"][0]
-    assert delete["folder"] == "Rep"
-    assert delete["deletedfolder"] == "Deleted"
-
-
-def test_refind_no_guids_raises():
+
+
+def test_reread_falls_back_to_the_deleted_folder():
+    """Some deployments relocate messages to the deleted folder behind our back."""
+    class OnlyDeleted(FakePPS):
+        def search(self, folder, query, limit=200, days_back=7, enddate=None):
+            self.calls.append({"m": "search", "folder": folder, "enddate": enddate})
+            return _moved_records() if folder == "Deleted" else []
+
+    pps = OnlyDeleted()
+    run_step(pps, _cfg(["move", "release"]), "release", "Rep", _chunk(), refresh=True)
+    assert [s["folder"] for s in pps.searches()] == ["Rep", "Deleted"]
+    assert pps.acts()[0]["folder"] == "Deleted"
+    assert pps.acts()[0]["localguids"] == ["R:1", "R:2"]
+
+
+def test_move_retries_once_after_rereading_when_the_stored_handles_are_stale():
+    class StaleHandles(FakePPS):
+        def act(self, action, folder, localguids, **kw):
+            result = super().act(action, folder, localguids, **kw)   # record the attempt
+            if action == "move" and "6:6:1" in localguids:
+                raise PPSError("no such localguid")
+            return result
+
+    pps = StaleHandles(records=_moved_records())
+    run_step(pps, _cfg(["move"]), "move", "Quarantine", _chunk(), refresh=False)
+    assert [s["folder"] for s in pps.searches()] == ["Quarantine"]
+    assert [a["localguids"] for a in pps.acts()] == [["6:6:1", "6:6:2"], ["R:1", "R:2"]]
+    assert pps.acts()[-1]["targetfolder"] == "Rep"
+
+
+def test_reread_without_guids_raises():
     pps = FakePPS()
-    chunk = [{"localguid": "6:6:1"}]   # no guid
     with pytest.raises(PPSError):
-        # force a refind by making move fail
-        class P(FakePPS):
-            def act(self, *a, **k):
-                if a[0] == "move":
-                    raise PPSError("relocated")
-                return super().act(*a, **k)
-        run_pipeline(P(), _cfg(["release", "move"]), "Quarantine", chunk)
+        run_step(pps, _cfg(["move", "release"]), "release", "Rep",
+                 [{"localguid": "6:6:1"}], refresh=True)   # no guid stored
 
 
+# ---------------------------------------------------------------- descriptions
+
 def test_pipeline_destination():
     assert pipeline_destination(_cfg(["release"])) is None
-    assert pipeline_destination(_cfg(["release", "move"])) == "Rep"
-    assert pipeline_destination(_cfg(["release", "delete"])) == "Deleted"
-    assert pipeline_destination(_cfg(["release", "move", "delete"])) == "Deleted"
+    assert pipeline_destination(_cfg(["move"])) == "Rep"
+    assert pipeline_destination(_cfg(["move", "release"])) == "Rep"
 
 
-def test_describe_pipeline():
-    text = describe_pipeline(_cfg(["release", "move", "delete"]))
-    assert "released in place" in text and "Rep" in text and "Deleted" in text
-    assert "→" in text
+def test_describe_pipeline_reads_in_execution_order():
+    text = describe_pipeline(_cfg(["move", "release"]))
+    assert text.index("Rep") < text.index("released")
 
 
-def test_allowed_steps_is_canonical_order():
-    assert ALLOWED_STEPS == ("release", "move", "delete")
+def test_allowed_steps_is_canonical_order_and_never_deletes():
+    assert ALLOWED_STEPS == ("move", "release")
 
 
-def test_step_delay_waits_between_steps():
-    pps = FakePPS()
-    waits = []
-    run_pipeline(pps, _cfg(["release", "move"], step_delay=60), "Quarantine", _chunk(),
-                 sleep=waits.append)
-    # One wait, of the configured length, between the two steps.
-    assert waits == [60]
+# ---------------------------------------------------------------- worker driving
 
+def _enqueue(queue):
+    return queue.enqueue("report_release", "Quarantine", _chunk(), {}, user="dev@x")
 
-def test_no_delay_before_first_or_when_zero():
-    pps = FakePPS()
-    waits = []
-    run_pipeline(pps, _cfg(["release"], step_delay=60), "Quarantine", _chunk(), sleep=waits.append)
-    assert waits == []                      # single step -> no wait
-    waits.clear()
-    run_pipeline(pps, _cfg(["release", "move"], step_delay=0), "Quarantine", _chunk(),
-                 sleep=waits.append)
-    assert waits == []                      # delay 0 -> disabled
-
-
-def test_delay_scales_with_step_count():
-    pps = FakePPS(records=[{"guid": "g1", "localguid": "R:1"}, {"guid": "g2", "localguid": "R:2"}])
-    waits = []
-    run_pipeline(pps, _cfg(["release", "move", "delete"], step_delay=30), "Quarantine",
-                 _chunk(), sleep=waits.append)
-    assert waits == [30, 30]                # before move, before delete
+
+def _run_next(queue):
+    """Claim and process one job; returns the claimed job row (None if none is due)."""
+    job = queue._claim_next()
+    if job is not None:
+        queue._process(job)
+    return job
+
+
+def _job(queue, job_id):
+    return next(j for j in queue.recent_jobs(limit=50) if j["id"] == job_id)
+
+
+def test_zero_delay_runs_every_step_in_one_pass(make_queue):
+    pps = FakePPS(records=_moved_records())
+    q = make_queue(pps, _cfg(["move", "release"], step_delay=0))
+    job_id = _enqueue(q)
+
+    _run_next(q)
+    assert [a["action"] for a in pps.acts()] == ["move", "release"]
+    assert _job(q, job_id)["status"] == "done"
+    # Even with no wait, the release step re-reads rather than reusing the move's handles.
+    assert pps.acts()[1]["localguids"] == ["R:1", "R:2"]
+
+
+def test_delay_parks_the_job_instead_of_blocking_the_queue(make_queue):
+    pps = FakePPS(records=_moved_records())
+    q = make_queue(pps, _cfg(["move", "release"], step_delay=60))
+    rr_id = _enqueue(q)
+    other_id = q.enqueue("delete", "Quarantine", [{"localguid": "x"}], {}, user="dev@x")
+
+    # First pass: the move, then the job parks itself.
+    assert _run_next(q)["id"] == rr_id
+    assert [a["action"] for a in pps.acts()] == ["move"]
+    parked = _job(q, rr_id)
+    assert parked["status"] == "pending" and parked["run_after"]
+
+    # The job queued behind it is NOT blocked by the wait — it runs now.
+    assert _run_next(q)["id"] == other_id
+    assert [a["action"] for a in pps.acts()] == ["move", "delete"]
+    assert _job(q, other_id)["status"] == "done"
+
+    # Nothing else is due while the wait is on.
+    assert q._claim_next() is None
+
+    # Once the delay has passed the job is re-claimed and finishes on fresh handles.
+    q._conn.execute("UPDATE jobs SET run_after='2000-01-01 00:00:00' WHERE id=?", (rr_id,))
+    q._conn.commit()
+    assert _run_next(q)["id"] == rr_id
+    release = [a for a in pps.acts() if a["action"] == "release"][0]
+    assert [s["folder"] for s in pps.searches()] == ["Rep"]
+    assert release["folder"] == "Rep" and release["localguids"] == ["R:1", "R:2"]
+    assert release["deletedfolder"] is None
+    assert _job(q, rr_id)["status"] == "done"
+
+
+def test_resume_keeps_the_steps_it_started_with(make_queue):
+    """An admin edit during the wait must not change a job already in flight."""
+    cfg = _cfg(["move", "release"], step_delay=60)
+    pps = FakePPS(records=_moved_records())
+    q = make_queue(pps, cfg)
+    rr_id = _enqueue(q)
+    _run_next(q)
+
+    cfg["report_release"]["steps"] = ["move"]        # live config change during the wait
+    q._conn.execute("UPDATE jobs SET run_after=NULL WHERE id=?", (rr_id,))
+    q._conn.commit()
+    _run_next(q)
+
+    # Still released: the plan came from the job's state, not the edited config.
+    assert [a["action"] for a in pps.acts()] == ["move", "release"]
+    assert _job(q, rr_id)["status"] == "done"
+
+
+def test_a_chunk_that_failed_a_step_is_skipped_by_later_steps(make_queue):
+    class OneBadMove(FakePPS):
+        def act(self, action, folder, localguids, **kw):
+            result = super().act(action, folder, localguids, **kw)   # record the attempt
+            if action == "move" and localguids == ["6:6:2"]:
+                raise PPSError("boom")
+            return result
+
+    cfg = _cfg(["move", "release"], step_delay=0)
+    cfg["chunk_size"] = 1                                   # one message per chunk
+    pps = OneBadMove(records=[{"guid": "g1", "localguid": "R:1"}])
+    q = make_queue(pps, cfg)
+    rr_id = _enqueue(q)
+    _run_next(q)
+
+    # The message whose move failed is never released; the other one is.
+    assert [(a["action"], a["localguids"]) for a in pps.acts()] == [
+        ("move", ["6:6:1"]), ("move", ["6:6:2"]), ("release", ["R:1"])]
+    row = _job(q, rr_id)
+    # The failed move re-read first (see the retry above) and couldn't locate it either.
+    assert row["status"] == "failed" and "not found" in row["error"]
+    assert row["processed"] == 1
+
+
+def test_retry_clears_the_parked_state(make_queue):
+    pps = FakePPS(records=_moved_records())
+    q = make_queue(pps, _cfg(["move", "release"], step_delay=60))
+    rr_id = _enqueue(q)
+    _run_next(q)
+    q._conn.execute("UPDATE jobs SET status='failed' WHERE id=?", (rr_id,))
+    q._conn.commit()
+
+    assert q.retry(rr_id) is True
+    row = q._conn.execute("SELECT run_after, state FROM jobs WHERE id=?", (rr_id,)).fetchone()
+    assert row["run_after"] is None and row["state"] == "{}"
+    # ...so it starts over at the first step, on the job's own localguids.
+    _run_next(q)
+    assert [a["action"] for a in pps.acts()] == ["move", "move"]

worker.py

@@ -6,6 +6,11 @@ SQLite so queued work survives an app restart.
 
 One JobQueue instance owns a single SQLite connection (guarded by a lock) shared
 between Flask request threads and the worker thread.
+
+Because that worker thread is single, nothing may sleep in it. A Report & Release job
+that has to wait between two pipeline steps saves its progress and goes back to
+'pending' with a future `run_after` (`_defer`); the loop runs everything else that is
+queued and re-claims the job when the wait is over.
 """
 
 from __future__ import annotations
@@ -16,7 +21,7 @@ import sqlite3
 import threading
 import time
 import traceback
-from datetime import datetime, timezone
+from datetime import datetime, timedelta, timezone
 from logging.handlers import RotatingFileHandler
 from typing import Callable, Mapping
 
@@ -51,6 +56,22 @@ def _now() -> str:
     return datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S")
 
 
+def _plus_seconds(seconds: int) -> str:
+    """A `_now()`-comparable UTC timestamp `seconds` in the future (sorts lexically)."""
+    return (datetime.now(timezone.utc) + timedelta(seconds=seconds)).strftime(
+        "%Y-%m-%d %H:%M:%S"
+    )
+
+
+def _progress(total_items: int, steps: int, step_index: int, done_in_step: int) -> int:
+    """Job progress in item units, spread evenly over the pipeline's steps.
+
+    Keeps `processed` monotonic and comparable to `total` (the message count) even
+    though a multi-step job walks every message several times.
+    """
+    return (step_index * total_items + done_in_step) // max(steps, 1)
+
+
 class JobQueue:
     def __init__(self, db_path: str,
                  pps_provider: Callable[[], PPSClient],
@@ -91,6 +112,10 @@ class JobQueue:
                     processed  INTEGER NOT NULL DEFAULT 0,
                     total      INTEGER NOT NULL DEFAULT 0,
                     error      TEXT,
+                    -- A pending job with run_after in the future is waiting out a
+                    -- pipeline step delay; `state` holds where it got to (see _defer).
+                    run_after  TEXT,
+                    state      TEXT NOT NULL DEFAULT '{}',
                     created_at TEXT NOT NULL,
                     updated_at TEXT NOT NULL
                 );
@@ -117,10 +142,17 @@ class JobQueue:
             job_cols = {r["name"] for r in self._conn.execute("PRAGMA table_info(jobs)")}
             if "user" not in job_cols:
                 self._conn.execute("ALTER TABLE jobs ADD COLUMN user TEXT")
+            if "run_after" not in job_cols:
+                self._conn.execute("ALTER TABLE jobs ADD COLUMN run_after TEXT")
+            if "state" not in job_cols:
+                self._conn.execute(
+                    "ALTER TABLE jobs ADD COLUMN state TEXT NOT NULL DEFAULT '{}'"
+                )
             self._conn.commit()
 
     def _recover(self) -> None:
-        # Any job left 'running' when the process died is requeued.
+        # Any job left 'running' when the process died is requeued. Its `state` survives,
+        # so a multi-step pipeline resumes at the step it had reached.
         with self._lock:
             cur = self._conn.execute(
                 "UPDATE jobs SET status='pending', updated_at=? WHERE status='running'",
@@ -173,7 +205,7 @@ class JobQueue:
 
     def recent_jobs(self, limit: int = 25, user: str | None = None) -> list[dict]:
         """Recent jobs, newest first. Pass `user` to scope to one owner (None = all)."""
-        cols = ("id, action, folder, user, status, processed, total, error,"
+        cols = ("id, action, folder, user, status, processed, total, error, run_after,"
                 " created_at, updated_at")
         with self._lock:
             if user is None:
@@ -195,10 +227,11 @@ class JobQueue:
         return row is not None
 
     def retry(self, job_id: int) -> bool:
-        """Requeue a failed job. Returns False if it isn't a failed job."""
+        """Requeue a failed job — from the top: progress state and any wait are cleared."""
         with self._lock:
             cur = self._conn.execute(
-                "UPDATE jobs SET status='pending', processed=0, error=NULL, updated_at=?"
+                "UPDATE jobs SET status='pending', processed=0, error=NULL,"
+                " run_after=NULL, state='{}', updated_at=?"
                 " WHERE id=? AND status='failed'",
                 (_now(), job_id),
             )
@@ -226,9 +259,14 @@ class JobQueue:
                 self._mark_failed(job["id"], traceback.format_exc())
 
     def _claim_next(self) -> dict | None:
+        # A job waiting out a step delay carries a future `run_after` and is simply
+        # skipped, so jobs queued behind it run instead of being blocked by the wait.
         with self._lock:
             row = self._conn.execute(
-                "SELECT * FROM jobs WHERE status='pending' ORDER BY id ASC LIMIT 1"
+                "SELECT * FROM jobs WHERE status='pending'"
+                " AND (run_after IS NULL OR run_after <= ?)"
+                " ORDER BY id ASC LIMIT 1",
+                (_now(),),
             ).fetchone()
             if row is None:
                 return None
@@ -240,42 +278,91 @@ class JobQueue:
             return dict(row)
 
     def _process(self, job: dict) -> None:
+        """Run a job, or the next step of one, and either finish or re-park it.
+
+        A Report & Release job is a sequence of steps (move → release → delete). Each
+        step is run for every chunk, then — if another step follows — the job is put back
+        on the queue with `run_after` set (`_defer`) instead of sleeping here: the worker
+        thread is single, so sleeping would stall every job queued behind this one.
+        Resuming re-reads the messages' current location, never the cached localguids.
+        """
         job_id = job["id"]
         action = job["action"]
-        folder = job["folder"]
         user = job["user"] if "user" in job.keys() else None
         extra = json.loads(job["extra"] or "{}")
+        state = json.loads((job["state"] if "state" in job.keys() else None) or "{}")
         items = self._items_for(job_id)
 
         # Snapshot config + client ONCE at job start — a mid-job admin edit affects the
-        # next job, not this one (see AGENTS.md "snapshot-per-job").
+        # next job, not this one (see AGENTS.md "snapshot-per-job"). A job that resumes
+        # after a wait re-reads the snapshot, so the plan it started with (steps, delay)
+        # is carried in `state` rather than re-derived from the new config.
         cfg = self._cfg()
         pps = self._pps()
         chunk_size = int(cfg.get("chunk_size", 25))
+        chunks = list(_chunks(items, chunk_size))
+
+        if action == "report_release":
+            steps: list[str | None] = list(state.get("steps") or pipeline.job_steps(cfg))
+            delay = int(state.get("delay", pipeline.step_delay(cfg)))
+        else:
+            steps, delay = [None], 0
+        index = int(state.get("step_index", 0))
+        folder = state.get("folder") or job["folder"]
+        errors: list[str] = list(state.get("errors") or [])
+        failed: set[int] = set(state.get("failed") or [])
+
         log.info(
-            "job #%d start: action=%s folder=%r user=%s total=%d",
-            job_id, action, folder, user, len(items),
+            "job #%d %s: action=%s folder=%r user=%s total=%d step=%s",
+            job_id, "resume" if index else "start", action, folder, user,
+            len(items), steps[index] or "-",
         )
 
-        dst = _destination(action, extra, cfg)
-        errors: list[str] = []
-        processed = 0
         started = time.monotonic()
-        for chunk in _chunks(items, chunk_size):
-            try:
-                self._run_action(pps, cfg, action, folder, chunk, extra)
-                processed += len(chunk)
-                self._set_processed(job_id, processed)
-                self._log_ops(job_id, action, folder, dst, chunk, user, "ok", None)
-            except PPSError as exc:
-                log.error("job #%d chunk failed: %s", job_id, exc)
-                errors.append(f"chunk failed: {exc}")
-                self._log_ops(job_id, action, folder, dst, chunk, user, "FAILED", str(exc))
-            except Exception as exc:  # noqa: BLE001
-                log.exception("job #%d chunk raised", job_id)
-                errors.append(f"chunk failed: {exc}")
-                self._log_ops(job_id, action, folder, dst, chunk, user, "FAILED", str(exc))
-
+        while index < len(steps):
+            step = steps[index]
+            dst = (pipeline.step_destination(cfg, step, folder) if step
+                   else _destination(action, extra, cfg))
+            done = 0
+            for i, chunk in enumerate(chunks):
+                if i in failed:
+                    continue                      # an earlier step gave up on this chunk
+                try:
+                    if step is None:
+                        self._run_action(pps, cfg, action, folder, chunk, extra)
+                    else:
+                        # refresh after the first step: the wait has passed, so re-read
+                        # where these messages are now instead of trusting stale handles.
+                        pipeline.run_step(pps, cfg, step, folder, chunk, refresh=index > 0)
+                except PPSError as exc:
+                    log.error("job #%d chunk failed: %s", job_id, exc)
+                    errors.append(f"chunk failed: {exc}")
+                    failed.add(i)
+                    self._log_ops(job_id, action, step, folder, dst, chunk, user,
+                                  "FAILED", str(exc))
+                except Exception as exc:  # noqa: BLE001
+                    log.exception("job #%d chunk raised", job_id)
+                    errors.append(f"chunk failed: {exc}")
+                    failed.add(i)
+                    self._log_ops(job_id, action, step, folder, dst, chunk, user,
+                                  "FAILED", str(exc))
+                else:
+                    done += len(chunk)
+                    self._set_processed(job_id, _progress(len(items), len(steps), index, done))
+                    self._log_ops(job_id, action, step, folder, dst, chunk, user, "ok", None)
+
+            index += 1
+            folder = dst if step and dst else folder
+            if index < len(steps) and delay > 0:
+                self._defer(job_id, delay, {
+                    "steps": steps, "delay": delay, "step_index": index,
+                    "folder": folder, "errors": errors, "failed": sorted(failed),
+                })
+                log.info("job #%d waiting %ds before %r step — other jobs run meanwhile",
+                         job_id, delay, steps[index])
+                return
+
+        processed = len(items) - sum(len(chunks[i]) for i in failed)
         took = _fmt_duration(time.monotonic() - started)
         if errors:
             log.error("job #%d FAILED: %d/%d processed, %d chunk error(s), took %s",
@@ -286,15 +373,35 @@ class JobQueue:
                      job_id, processed, len(items), took)
             self._mark_done(job_id)
 
-    def _log_ops(self, job_id: int, action: str, src: str, dst: str | None,
-                 chunk: list[dict], user: str | None,
+    def _defer(self, job_id: int, delay: int, state: dict) -> None:
+        """Park a partly-done job back on the queue, runnable again in `delay` seconds.
+
+        This is what keeps the step delay from blocking the queue: the job goes back to
+        'pending' with a future `run_after`, `_claim_next` skips it until then, and every
+        other queued job gets its turn in the meantime.
+        """
+        with self._lock:
+            self._conn.execute(
+                "UPDATE jobs SET status='pending', run_after=?, state=?, updated_at=?"
+                " WHERE id=?",
+                (_plus_seconds(delay), json.dumps(state), _now(), job_id),
+            )
+            self._conn.commit()
+
+    def _log_ops(self, job_id: int, action: str, step: str | None, src: str,
+                 dst: str | None, chunk: list[dict], user: str | None,
                  result: str, error: str | None) -> None:
-        """Write one audit line per message in the chunk to the operations log."""
+        """Write one audit line per message in the chunk to the operations log.
+
+        Multi-step Report & Release writes one line per message PER STEP, each naming the
+        step it reports on (`step=move`) and that step's own src/dst.
+        """
         for it in chunk:
             fields = [
                 f"job=#{job_id}",
                 f"user={user or '-'}",
                 f"action={action}",
+                *([f"step={step}"] if step else []),
                 f"result={result}",
                 f"src={src!r}",
                 f"dst={dst!r}",
@@ -327,11 +434,9 @@ class JobQueue:
                 raise ValueError("move action requires targetfolder")
             pps.act("move", folder, localguids, targetfolder=target)
 
-        elif action == "report_release":
-            # Data-driven pipeline (release -> move -> delete, configurable subset).
-            pipeline.run_pipeline(pps, cfg, folder, chunk)
-
         else:
+            # report_release never lands here: `_process` drives its steps one at a time
+            # so it can yield the worker thread during the between-step delay.
             raise ValueError(f"unknown action: {action}")
 
     # -------------------------------------------------------------- db helpers
@@ -355,7 +460,8 @@ class JobQueue:
     def _mark_done(self, job_id: int) -> None:
         with self._lock:
             self._conn.execute(
-                "UPDATE jobs SET status='done', updated_at=? WHERE id=?",
+                "UPDATE jobs SET status='done', run_after=NULL, state='{}', updated_at=?"
+                " WHERE id=?",
                 (_now(), job_id),
             )
             self._conn.commit()
@@ -363,7 +469,8 @@ class JobQueue:
     def _mark_failed(self, job_id: int, error: str) -> None:
         with self._lock:
             self._conn.execute(
-                "UPDATE jobs SET status='failed', error=?, updated_at=? WHERE id=?",
+                "UPDATE jobs SET status='failed', error=?, run_after=NULL, state='{}',"
+                " updated_at=? WHERE id=?",
                 (error, _now(), job_id),
             )
             self._conn.commit()