fix(policies): poll runs to completion with progress, soft-retry when queue is full (Stirling-Tools#6690)

## What & why

Production reports of policy enforcement "hanging" traced to
large/many-page documents: the watermark step's flatten-to-image
(`convertPDFToImage`) on a 500+ page PDF takes minutes, exceeding both
the client poll cap and the backend per-step timeout. This makes the
slow case graceful instead of looking broken, and makes load-shedding
non-fatal.

### Poll runs to completion (no false "hang")
The client poll loop used a flat ~150s cap that was **shorter than the
backend's 300s per-step timeout**, so it abandoned long-but-healthy runs
mid-flight. The budget is now sized to the backend's real worst case —
`stepCount × per-step timeout + grace`, learned from the first status
report — so the client always polls long enough to surface the run's
**actual** terminal state (success or the backend's real error) rather
than a misleading client-side timeout.

### Per-step progress
The activity feed now shows `Enforcing… · step n/m` (from
`currentStep`/`stepCount`), so a slow step shows movement instead of a
dead spinner.

### Soft-retry on queue rejection
Under load the shared `JobQueue` rejects runs ("queue full"), which
previously surfaced as a hard failure needing a manual Retry. The
backend now tags that rejection with a stable `POLICY_QUEUE_FULL`
errorCode; the client treats it as transient backpressure and
**auto-retries the file in place** with exponential backoff (≈4s→64s, ~2
min), shown as a soft "Busy — retrying…" row, falling back to the manual
Retry only once the retry budget is spent.

## Testing
- **Frontend unit tests** (30 pass across the policies suite), including
a new `usePolicyAutoRun.retry.test.tsx` that drives the real controller
orchestration (poll → `POLICY_QUEUE_FULL` → relabel → backoff → in-place
re-dispatch), plus poll-budget, step-progress, and activity-feed relabel
cases.
- **Backend** `PolicyEngineTest` case asserting a queue-rejected run
carries the `POLICY_QUEUE_FULL` code.
- Typecheck clean on all three flavors (proprietary/saas/core); prettier
+ spotless clean.
- Poll-budget + progress + real-error surfacing were also verified live
end-to-end against a 599-page run (survived past the old cap, showed
step progress, reported the backend's real 300s-timeout failure,
recovered after a simulated network drop).

## Not included (follow-ups)
- The underlying flatten-to-image cost itself (bounded-memory/streaming
flatten, revisiting `convertPDFToImage` default and the 300s timeout) —
the real perf fix, deliberately out of scope here.

Author: reecebrowne

app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java

@@ -59,6 +59,10 @@ public class PolicyEngine {
     // files. See ResourceMonitor#shouldQueueJob(int).
     private static final int RUN_RESOURCE_WEIGHT = 50;
 
+    // errorCode marking a run that was never admitted (job queue full under load). Transient: the
+    // client treats it as "busy" and retries, rather than as a terminal processing failure.
+    private static final String QUEUE_FULL_CODE = "POLICY_QUEUE_FULL";
+
     private final PolicyExecutor stepExecutor;
     private final TaskManager taskManager;
     private final PolicyRunRegistry registry;
@@ -239,7 +243,8 @@ private ResponseEntity<?> failRejectedRun(
         if (!completion.isDone()) {
             String message = "Policy run could not be queued: " + ex.getMessage();
             log.error("Policy run {} was not admitted: {}", run.getRunId(), ex.getMessage());
-            run.fail(message);
+            // Transient admission rejection, not a processing failure (see QUEUE_FULL_CODE).
+            run.failWithCode(message, QUEUE_FULL_CODE, null);
             taskManager.setError(run.getRunId(), message);
             completion.complete(run);
         }

app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyEngineTest.java

@@ -336,6 +336,27 @@ void runIsQueuedUnderResourcePressure() {
         assertEquals(PolicyRunStatus.PENDING, registry.get(handle.runId()).getStatus());
     }
 
+    @Test
+    void runRejectedWhenQueueFullCarriesTransientErrorCode() {
+        when(resourceMonitor.shouldQueueJob(anyInt())).thenReturn(true);
+        // Admission rejected (queue full): the queued future completes exceptionally.
+        CompletableFuture<Object> rejected = new CompletableFuture<>();
+        rejected.completeExceptionally(
+                new RuntimeException("Job queue full, please try again later"));
+        doReturn(rejected).when(jobQueue).queueJob(anyString(), anyInt(), any(), anyLong());
+
+        PolicyRunHandle handle =
+                engine.submit(
+                        definition(new PipelineStep(ROTATE, Map.of())),
+                        PolicyInputs.of(List.of(pdf("input", "input.pdf"))),
+                        PolicyProgressListener.NOOP);
+
+        PolicyRun run = registry.get(handle.runId());
+        assertEquals(PolicyRunStatus.FAILED, run.getStatus());
+        // Tagged transient so the client backs off and retries instead of hard-failing.
+        assertEquals("POLICY_QUEUE_FULL", run.getErrorCode());
+    }
+
     @Test
     void resumeIsNotYetImplemented() {
         assertThrows(UnsupportedOperationException.class, () -> engine.resume("any", List.of()));

frontend/editor/src/proprietary/components/policies/policyRunStore.test.ts

@@ -1,9 +1,11 @@
 import { describe, it, expect, beforeEach } from "vitest";
 import {
   dispatchKey,
+  getRun,
   isDispatched,
   markDispatched,
   recordRunStart,
+  removeRun,
   updateRun,
   resetPolicyRuns,
   type PolicyRunRecord,
@@ -70,6 +72,16 @@ describe("policyRunStore", () => {
     expect(read("stirling-policy-runs").runs[0].status).toBe("PENDING");
   });
 
+  it("getRun returns the record by id, removeRun drops it but keeps the dispatched key", () => {
+    recordRunStart(rec({ runId: "abc" }));
+    expect(getRun("abc")?.fileId).toBe("f1");
+    removeRun("abc");
+    expect(getRun("abc")).toBeUndefined();
+    expect(read("stirling-policy-runs").runs).toHaveLength(0);
+    // The (policy, file) pair stays dispatched so the auto-run doesn't re-fire on its own.
+    expect(isDispatched("security", "f1")).toBe(true);
+  });
+
   it("caps stored runs at 50, newest first", () => {
     for (let i = 0; i < 55; i++) {
       recordRunStart(rec({ runId: `r${i}`, fileId: `f${i}`, startedAt: i }));

frontend/editor/src/proprietary/components/policies/policyRunStore.ts

@@ -19,6 +19,11 @@ export interface PolicyRunRecord {
   fileName: string;
   fileSize: number;
   status: PolicyRunStatus;
+  /** Pipeline progress reported by the run-status endpoint: the 1-based step
+   *  currently running, and the total step count. Drive the "step X/Y" label
+   *  while a run is in flight. Absent until the first status report. */
+  currentStep?: number;
+  stepCount?: number;
   /** Output files (downloadable via /api/v1/general/files/{id}) once done. */
   outputs: { fileId: string; fileName: string }[];
   /** True once ALL outputs have been imported into the workspace. */
@@ -33,6 +38,9 @@ export interface PolicyRunRecord {
   error: string | null;
   /** Stable backend failure code (e.g. an entitlement sentinel) when FAILED; null otherwise. */
   errorCode?: string | null;
+  /** Set while an auto-retry is pending after a transient (queue-full) rejection, so the activity
+   *  feed shows a soft "busy" row instead of a hard failure during the backoff window. */
+  retrying?: boolean;
   /** Epoch ms when the run was dispatched. */
   startedAt: number;
 }
@@ -146,6 +154,19 @@ export function updateRun(runId: string, patch: Partial<PolicyRunRecord>) {
   emit();
 }
 
+/** The current record for a run id, if any. */
+export function getRun(runId: string): PolicyRunRecord | undefined {
+  return state.runs.find((r) => r.runId === runId);
+}
+
+/** Drop a run record (leaving its dispatched key intact). Used when retrying a
+ *  queue-rejected run in place, so the replacement run doesn't stack a second row. */
+export function removeRun(runId: string) {
+  if (!state.runs.some((r) => r.runId === runId)) return;
+  state = { ...state, runs: state.runs.filter((r) => r.runId !== runId) };
+  emit();
+}
+
 /** Reset the store — used by tests to isolate it. */
 export function resetPolicyRuns() {
   state = { runs: [], dispatched: [] };

frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.retry.test.tsx

@@ -0,0 +1,106 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+import { renderHook, act } from "@testing-library/react";
+
+// The auto-run hook reaches into several contexts + the network; stub those so we can drive just
+// the queue-rejection retry path against the REAL run store.
+vi.mock("@app/constants/featureFlags", () => ({ POLICIES_ENABLED: true }));
+vi.mock("@app/contexts/FileContext", () => ({
+  useAllFiles: () => ({ fileStubs: [] }),
+  useFileManagement: () => ({ addFiles: vi.fn() }),
+  useFileContext: () => ({ consumeFiles: vi.fn() }),
+}));
+vi.mock("@app/hooks/usePolicies", () => ({
+  usePolicies: () => ({
+    policies: {
+      security: {
+        configured: true,
+        status: "active",
+        backendId: "backend-1",
+        runOn: "upload",
+      },
+    },
+  }),
+}));
+vi.mock("@app/services/policyApi", () => ({
+  runStoredPolicy: vi.fn(),
+  getPolicyRun: vi.fn(),
+  downloadPolicyOutput: vi.fn(),
+}));
+vi.mock("@app/services/fileStorage", () => ({
+  fileStorage: { getStirlingFile: vi.fn() },
+}));
+
+import { usePolicyAutoRun } from "@app/components/policies/usePolicyAutoRun";
+import {
+  recordRunStart,
+  getRun,
+  resetPolicyRuns,
+  usePolicyRuns,
+} from "@app/components/policies/policyRunStore";
+import { runStoredPolicy, getPolicyRun } from "@app/services/policyApi";
+import { fileStorage } from "@app/services/fileStorage";
+
+const getRunApi = vi.mocked(getPolicyRun);
+const runStored = vi.mocked(runStoredPolicy);
+const getFile = vi.mocked(fileStorage.getStirlingFile);
+
+const queueFullView = {
+  runId: "run-1",
+  status: "FAILED",
+  currentStep: 0,
+  stepCount: 2,
+  error: "Policy run could not be queued: Job queue full",
+  errorCode: "POLICY_QUEUE_FULL",
+  outputs: [],
+} as never;
+
+beforeEach(() => {
+  vi.useFakeTimers();
+  localStorage.clear();
+  resetPolicyRuns();
+  getRunApi.mockReset();
+  runStored.mockReset();
+  getFile.mockReset();
+});
+afterEach(() => vi.useRealTimers());
+
+describe("auto-run queue-rejection retry", () => {
+  it("relabels a queue-rejected run as retrying, then re-dispatches it in place", async () => {
+    // The polled run comes back queue-rejected; the retry resolves the file + fires a fresh run.
+    getRunApi.mockResolvedValue(queueFullView);
+    getFile.mockResolvedValue({ size: 1234 } as never);
+    runStored.mockResolvedValue("run-2");
+
+    recordRunStart({
+      runId: "run-1",
+      categoryId: "security",
+      fileId: "file-1",
+      fileName: "doc.pdf",
+      fileSize: 1234,
+      status: "RUNNING",
+      outputs: [],
+      error: null,
+      startedAt: 0,
+    });
+
+    renderHook(() => {
+      usePolicyAutoRun();
+      return usePolicyRuns();
+    });
+
+    // First poll (2s cadence) sees the rejection → relabel as a soft "retrying" row.
+    await act(async () => {
+      await vi.advanceTimersByTimeAsync(2000);
+    });
+    expect(getRun("run-1")?.retrying).toBe(true);
+    expect(runStored).not.toHaveBeenCalled();
+
+    // After the first backoff window (BASE 4s) the rejected record is dropped and a fresh run fires.
+    await act(async () => {
+      await vi.advanceTimersByTimeAsync(4000);
+    });
+    expect(runStored).toHaveBeenCalledWith("backend-1", [{ size: 1234 }]);
+    expect(getRun("run-1")).toBeUndefined();
+    expect(getRun("run-2")?.status).toBe("PENDING");
+  });
+});

frontend/editor/src/proprietary/components/policies/usePolicyAutoRun.test.ts

@@ -0,0 +1,160 @@
+import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
+
+// Mock the network + the run store so we can drive the poll loop deterministically.
+vi.mock("@app/services/policyApi", async (orig) => ({
+  ...(await orig<typeof import("@app/services/policyApi")>()),
+  getPolicyRun: vi.fn(),
+}));
+vi.mock("@app/components/policies/policyRunStore", async (orig) => ({
+  ...(await orig<typeof import("@app/components/policies/policyRunStore")>()),
+  updateRun: vi.fn(),
+}));
+
+import { poll } from "@app/components/policies/usePolicyAutoRun";
+import { getPolicyRun } from "@app/services/policyApi";
+import { updateRun } from "@app/components/policies/policyRunStore";
+
+const getRun = vi.mocked(getPolicyRun);
+const update = vi.mocked(updateRun);
+
+const POLL_MS = 2000;
+// Must match the poll loop's budget formula (stepCount × per-step timeout + grace).
+// With the 2-step `view()` below that's 2 × 300_000 + 30_000 = 630_000ms.
+const STEP_TIMEOUT_MS = 300_000;
+const POLL_GRACE_MS = 30_000;
+const STEP_COUNT = 2;
+const BUDGET_MS = STEP_COUNT * STEP_TIMEOUT_MS + POLL_GRACE_MS;
+const view = (status: string) =>
+  ({
+    runId: "r1",
+    status,
+    currentStep: 1,
+    stepCount: STEP_COUNT,
+    outputs: [],
+    error: null,
+  }) as never;
+
+beforeEach(() => {
+  vi.useFakeTimers();
+  getRun.mockReset();
+  update.mockReset();
+});
+afterEach(() => vi.useRealTimers());
+
+/** Run the poll loop for n cadence ticks, flushing the awaited fetch each tick. */
+async function tick(n: number) {
+  for (let i = 0; i < n; i++) await vi.advanceTimersByTimeAsync(POLL_MS);
+}
+
+describe("policy run poll loop", () => {
+  it("stops and marks FAILED after repeated run-not-found (404)", async () => {
+    getRun.mockRejectedValue({ code: "ERR_NOT_FOUND" });
+    const p = poll("r1");
+    await tick(3); // MAX_NOT_FOUND consecutive 404s
+    await p;
+    expect(update).toHaveBeenCalledWith(
+      "r1",
+      expect.objectContaining({ status: "FAILED" }),
+    );
+    // It gave up after the not-found streak, not after the full time budget.
+    expect(getRun.mock.calls.length).toBe(3);
+  });
+
+  it("also detects 404 via axios-style response.status", async () => {
+    getRun.mockRejectedValue({ response: { status: 404 } });
+    const p = poll("r1");
+    await tick(3);
+    await p;
+    expect(update).toHaveBeenCalledWith(
+      "r1",
+      expect.objectContaining({ status: "FAILED" }),
+    );
+  });
+
+  it("does NOT fail on a brief not-found blip that then recovers", async () => {
+    getRun
+      .mockRejectedValueOnce({ code: "ERR_NOT_FOUND" })
+      .mockResolvedValue(view("COMPLETED"));
+    const p = poll("r1");
+    await tick(2);
+    await p;
+    const statuses = update.mock.calls.map(
+      (c) => (c[1] as { status: string }).status,
+    );
+    expect(statuses).toContain("COMPLETED");
+    expect(statuses).not.toContain("FAILED");
+  });
+
+  it("transient non-404 errors don't count toward the not-found streak", async () => {
+    getRun
+      .mockRejectedValueOnce({ code: "ERR_NOT_FOUND" })
+      .mockRejectedValueOnce({ response: { status: 500 } })
+      .mockRejectedValueOnce({ code: "ERR_NOT_FOUND" })
+      .mockResolvedValue(view("COMPLETED"));
+    const p = poll("r1");
+    await tick(4);
+    await p;
+    const statuses = update.mock.calls.map(
+      (c) => (c[1] as { status: string }).status,
+    );
+    expect(statuses).toContain("COMPLETED");
+    expect(statuses).not.toContain("FAILED");
+  });
+
+  it("marks FAILED when the run never reaches a terminal state within the budget", async () => {
+    getRun.mockResolvedValue(view("RUNNING"));
+    const p = poll("r1");
+    // Advance past the full step-count-derived budget in one go.
+    await vi.advanceTimersByTimeAsync(BUDGET_MS + POLL_MS);
+    await p;
+    expect(update).toHaveBeenLastCalledWith(
+      "r1",
+      expect.objectContaining({ status: "FAILED" }),
+    );
+  });
+
+  it("keeps polling a long run for the whole step budget (no premature giveup)", async () => {
+    getRun.mockResolvedValue(view("RUNNING"));
+    const p = poll("r1");
+    // 200s in: a single step may run up to the per-step timeout, so the run is
+    // still legitimately in flight and must keep being polled, not failed.
+    await tick(100);
+    expect(update).not.toHaveBeenCalledWith(
+      "r1",
+      expect.objectContaining({ status: "FAILED" }),
+    );
+    // Let it run out so the dangling promise doesn't leak into other tests.
+    await vi.advanceTimersByTimeAsync(BUDGET_MS);
+    await p;
+  });
+
+  it("records pipeline progress (currentStep/stepCount) while running", async () => {
+    getRun
+      .mockResolvedValueOnce(view("RUNNING"))
+      .mockResolvedValue(view("COMPLETED"));
+    const p = poll("r1");
+    await tick(2);
+    await p;
+    expect(update).toHaveBeenCalledWith(
+      "r1",
+      expect.objectContaining({ currentStep: 1, stepCount: STEP_COUNT }),
+    );
+  });
+
+  it("finishes cleanly on a terminal status and fires onTerminal", async () => {
+    getRun.mockResolvedValue(view("COMPLETED"));
+    const onTerminal = vi.fn();
+    const p = poll("r1", onTerminal);
+    await tick(1);
+    await p;
+    expect(update).toHaveBeenCalledWith(
+      "r1",
+      expect.objectContaining({ status: "COMPLETED" }),
+    );
+    expect(onTerminal).toHaveBeenCalledTimes(1);
+    expect(update).not.toHaveBeenCalledWith(
+      "r1",
+      expect.objectContaining({ status: "FAILED" }),
+    );
+  });
+});