"""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. """ from __future__ import annotations import logging import time from dataclasses import dataclass, replace from typing import Any, Callable, Mapping 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") # 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. DEFAULT_STEP_DELAY = 60 @dataclass(frozen=True) 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 chunk: list[dict] # job_items rows: localguid, guid, subject, ... @dataclass(frozen=True) class StepResult: folder: str localguids: list[str] | None dst: str | None # folder this step lands messages in (for ops log) StepFn = Callable[[StepContext], StepResult] STEPS: dict[str, StepFn] = {} def step(name: str): def register(fn: StepFn) -> StepFn: STEPS[name] = fn return fn 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"] if not target: 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. """ 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) 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. """ wanted = {it["guid"]: it for it in chunk if it.get("guid")} if not wanted: raise PPSError("cannot recover messages: no guids stored for this chunk") records = pps.search( folder, cfg.get("list_query", "from=*"), limit=1000, days_back=int(cfg.get("default_days_back", 7)), ) found = [ r["localguid"] for r in records if r.get("guid") in wanted and r.get("localguid") ] if not found: raise PPSError(f"messages not found in {folder!r} to continue the pipeline") return found def pipeline_destination(cfg: Mapping[str, Any]) -> str | None: """Final folder after all steps, computed WITHOUT touching PPS (for the ops log).""" folder = 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"'.""" rr = cfg.get("report_release", {}) phrases = [] for name in rr.get("steps", []): if name == "release": phrases.append("released in place") elif name == "move": phrases.append(f'moved to "{rr.get("move_target", "?")}"') elif name == "delete": phrases.append(f'deleted to "{cfg.get("deleted_folder", "?")}"') return " → ".join(phrases) if phrases else "(no steps configured)"