fix(cdp_queue): auto-purge stale queue tickets older than 2x acquire timeout

bin/cdp_queue.py

@@ -89,9 +89,10 @@ def _ensure_dirs(node):
 
 
 def _read_tickets(node):
-    """Return sorted list of (priority, timestamp, ticket_name)."""
+    """Return sorted list of (priority, timestamp, ticket_name), purging stale tickets."""
     tdir = _tickets_dir(node)
     out = []
+    now = time.time()
     try:
         for name in os.listdir(tdir):
             if not name.endswith(".ticket"):
@@ -100,7 +101,15 @@ def _read_tickets(node):
                 # ticket name: "<prio>-<timestamp>-<uuid>.ticket"
                 parts = name[:-7].split("-")
                 prio_s, ts_s = parts[0], parts[1]
-                out.append((int(prio_s), float(ts_s), name))
+                ts = float(ts_s)
+                # Purge stale ticket if process crashed or timed out ungracefully
+                if now - ts > (ACQUIRE_TIMEOUT * 2):
+                    try:
+                        os.unlink(os.path.join(tdir, name))
+                    except Exception:
+                        pass
+                    continue
+                out.append((int(prio_s), ts, name))
             except (ValueError, IndexError):
                 continue
     except FileNotFoundError:

job-sidechats.json

@@ -17,5 +17,10 @@
     "thread_uuid": "da55b05c-19c6-4bdb-8f8b-e507d7430ca2",
     "agent": "opm",
     "created_at": "2026-10-04T16:39:34.721269+00:00"
+  },
+  "pipe-9735f2": {
+    "thread_uuid": "0077e918-9ac8-40ba-b0ad-65f9f78037c4",
+    "agent": "opm",
+    "created_at": "2026-10-04T16:43:03.587214+00:00"
   }
 }