Coverage for src/lilbee/cli/tui/task_queue.py: 100%
220 statements
« prev ^ index » next coverage.py v7.15.2, created at 2026-08-14 11:46 +0000
1"""Per-type concurrent task queue for background operations (downloads, syncs, crawls).
3Each task type (download, sync, crawl) gets its own independent queue, so a long
4download does not block a sync from starting. Within a type, tasks run sequentially.
5"""
7from __future__ import annotations
9import logging
10import threading
11import time
12import uuid
13from collections.abc import Callable
14from dataclasses import dataclass
15from enum import StrEnum
17log = logging.getLogger(__name__)
20class TaskStatus(StrEnum):
21 """Lifecycle states for a queued task."""
23 QUEUED = "queued"
24 ACTIVE = "active"
25 DONE = "done"
26 FAILED = "failed"
27 CANCELLED = "cancelled"
30class TaskType(StrEnum):
31 """Canonical task types. Replaces raw string literals at call sites."""
33 DOWNLOAD = "download"
34 SYNC = "sync"
35 CRAWL = "crawl"
36 WIKI = "wiki"
37 ADD = "add"
38 REMOVE = "remove"
39 SETUP = "setup"
40 IMPORT = "import"
41 EXPORT = "export"
44TERMINAL_STATUSES = (TaskStatus.DONE, TaskStatus.FAILED, TaskStatus.CANCELLED)
47# Every icon is one cell wide. An emoji here is double-width and shifts its row
48# out of the column the other statuses share.
49STATUS_ICONS: dict[TaskStatus, str] = {
50 TaskStatus.QUEUED: "⋯",
51 TaskStatus.ACTIVE: "▶",
52 TaskStatus.DONE: "✓",
53 TaskStatus.FAILED: "✗",
54 TaskStatus.CANCELLED: "⊘",
55}
58@dataclass
59class Task:
60 """A single unit of work in the queue."""
62 task_id: str
63 name: str
64 task_type: str
65 fn: Callable[[], None]
66 # Identity of the work; ``enqueue`` refuses a second pending task with the
67 # same key. None opts out.
68 dedupe_key: str | None = None
69 status: TaskStatus = TaskStatus.QUEUED
70 progress: float = 0.0
71 detail: str = ""
72 indeterminate: bool = False
73 # Monotonic timestamp at which the task transitioned to ACTIVE. None
74 # while QUEUED. Used by the Task Center row to render elapsed time.
75 started_at: float | None = None
76 # Monotonic timestamp at which the task reached a terminal state
77 # (DONE / FAILED / CANCELLED). None while still running. Used to
78 # freeze the elapsed-time display so it doesn't keep ticking during
79 # the 2-second post-finish flash.
80 completed_at: float | None = None
83class TaskQueue:
84 """Per-type concurrent task queue.
85 Thread-safe. Each task type (download, sync, crawl, etc.) has its own
86 independent FIFO queue. One task per type can be active simultaneously,
87 so a download does not block a sync.
89 Callers receive a *task_id* they can use to update progress, cancel, or
90 query status.
91 """
93 def __init__(
94 self,
95 *,
96 on_change: Callable[[], None] | None = None,
97 capacity: dict[str, int] | None = None,
98 ) -> None:
99 self._lock = threading.Lock()
100 self._tasks: dict[str, Task] = {}
101 self._queues: dict[str, list[str]] = {}
102 # Per-type set of currently-active task ids. A "type" here means
103 # sync/crawl/download/wiki; each has its own FIFO and own active slots.
104 self._active_ids: dict[str, set[str]] = {}
105 # Max concurrent active tasks per type. Defaults to 1 (single-active).
106 # Callers may override per type; types absent from the map cap at 1.
107 self._capacity: dict[str, int] = dict(capacity or {})
108 self._on_change: list[Callable[[], None]] = []
109 if on_change:
110 self._on_change.append(on_change)
111 self._history: list[Task] = []
113 def _capacity_for(self, task_type: str) -> int:
114 return self._capacity.get(task_type, 1)
116 def subscribe(self, callback: Callable[[], None]) -> None:
117 """Subscribe to task queue changes. Callback is called on any queue update."""
118 with self._lock:
119 if callback not in self._on_change:
120 self._on_change.append(callback)
122 def unsubscribe(self, callback: Callable[[], None]) -> None:
123 """Unsubscribe from task queue changes."""
124 with self._lock:
125 if callback in self._on_change:
126 self._on_change.remove(callback)
128 @property
129 def active_task(self) -> Task | None:
130 """Return any one active task. Prefer ``active_tasks`` for the full set."""
131 with self._lock:
132 for ids in self._active_ids.values():
133 for tid in ids:
134 task = self._tasks.get(tid)
135 if task:
136 return task
137 return None
139 @property
140 def active_tasks(self) -> list[Task]:
141 """Return all currently active tasks across all types."""
142 with self._lock:
143 tasks: list[Task] = []
144 for ids in self._active_ids.values():
145 for tid in ids:
146 task = self._tasks.get(tid)
147 if task:
148 tasks.append(task)
149 return tasks
151 @property
152 def queued_tasks(self) -> list[Task]:
153 with self._lock:
154 result: list[Task] = []
155 for tids in self._queues.values():
156 for tid in tids:
157 task = self._tasks.get(tid)
158 if task:
159 result.append(task)
160 return result
162 @property
163 def history(self) -> list[Task]:
164 with self._lock:
165 return list(self._history)
167 @property
168 def is_empty(self) -> bool:
169 with self._lock:
170 has_active = any(ids for ids in self._active_ids.values())
171 has_queued = any(len(q) > 0 for q in self._queues.values())
172 return not has_active and not has_queued
174 def get_task(self, task_id: str) -> Task | None:
175 """Look up a task by ID. Returns None if not found."""
176 with self._lock:
177 return self._tasks.get(task_id)
179 def find_pending(self, task_type: str, dedupe_key: str) -> Task | None:
180 """Return the queued-or-active task for this key, if there is one."""
181 with self._lock:
182 return self._find_pending_locked(task_type, dedupe_key)
184 def _find_pending_locked(self, task_type: str, dedupe_key: str) -> Task | None:
185 for task in self._tasks.values():
186 if (
187 task.task_type == task_type
188 and task.dedupe_key == dedupe_key
189 and task.status not in TERMINAL_STATUSES
190 ):
191 return task
192 return None
194 def enqueue(
195 self,
196 fn: Callable[[], None],
197 name: str,
198 task_type: str,
199 *,
200 indeterminate: bool = False,
201 dedupe_key: str | None = None,
202 ) -> str:
203 """Add a task to the per-type queue. Returns a task_id.
205 With a *dedupe_key* already queued or active, returns that task's id and
206 adds nothing: asking twice for one download is a double keypress, not a
207 request for two copies. The check lives here rather than at the call
208 sites so a new caller cannot reintroduce the duplicate by forgetting it.
209 """
210 task_id = uuid.uuid4().hex[:8]
211 task = Task(
212 task_id=task_id,
213 name=name,
214 task_type=task_type,
215 fn=fn,
216 indeterminate=indeterminate,
217 dedupe_key=dedupe_key,
218 )
219 with self._lock:
220 if dedupe_key is not None:
221 existing = self._find_pending_locked(task_type, dedupe_key)
222 if existing is not None:
223 return existing.task_id
224 self._tasks[task_id] = task
225 self._queues.setdefault(task_type, []).append(task_id)
226 self._notify()
227 return task_id
229 def update_task(
230 self,
231 task_id: str,
232 progress: float,
233 detail: str = "",
234 *,
235 indeterminate: bool | None = None,
236 ) -> None:
237 """Update progress and detail text for a task.
238 When *indeterminate* is True the task's progress bar renders as a
239 pulsing indeterminate bar instead of a percentage. When explicitly
240 False it returns to determinate mode. ``None`` leaves the flag as-is
241 so incremental progress updates don't clobber the caller's intent.
242 """
243 with self._lock:
244 task = self._tasks.get(task_id)
245 if task:
246 task.progress = progress
247 task.detail = detail
248 if indeterminate is not None:
249 task.indeterminate = indeterminate
250 self._notify()
252 def complete_task(self, task_id: str) -> None:
253 """Mark a task as done and append it to history.
255 A row that already reached a terminal state is left alone, so a
256 worker that finishes after the user cancelled it does not flip the
257 cancelled row to DONE and append it to history twice.
259 The task record stays in ``_tasks`` so callers can still look it
260 up by id. Bulk removal happens in ``clear_history`` or targeted
261 removal via ``remove_task``.
262 """
263 with self._lock:
264 task = self._tasks.get(task_id)
265 if task and task.status not in TERMINAL_STATUSES:
266 task.status = TaskStatus.DONE
267 task.progress = 100
268 task.indeterminate = False
269 task.completed_at = time.monotonic()
270 self._history.append(task)
271 self._remove_from_active_locked(task_id, task.task_type)
272 self._remove_from_queue_locked(task_id, task.task_type)
273 self._notify()
275 def fail_task(self, task_id: str, detail: str = "") -> None:
276 """Mark a task as failed and append it to history.
278 Terminal rows are immutable here too: a cancelled task whose worker
279 then raises stays cancelled.
281 The task record stays in ``_tasks`` so callers can still inspect
282 its detail. Bulk removal happens in ``clear_history`` or
283 targeted removal via ``remove_task``.
284 """
285 with self._lock:
286 task = self._tasks.get(task_id)
287 if task and task.status not in TERMINAL_STATUSES:
288 task.status = TaskStatus.FAILED
289 task.detail = detail
290 task.completed_at = time.monotonic()
291 self._history.append(task)
292 self._remove_from_active_locked(task_id, task.task_type)
293 self._remove_from_queue_locked(task_id, task.task_type)
294 self._notify()
296 def cancel(self, task_id: str) -> bool:
297 """Cancel a queued or active task. Returns True if cancelled.
299 Marks the row only. Stopping a running download needs
300 ``TaskBarController.cancel_task``, which also aborts the transfer.
302 Terminal rows (DONE / FAILED / CANCELLED) are immutable: a cancel
303 call against an already-finished task is a no-op and returns
304 False so callers that key off the return value (e.g. UI actions)
305 don't treat it as success.
307 Appends the cancelled task to ``_history`` so the Task Center
308 renders it as a lingering ``cancelled`` row, matching the
309 post-flash contract for DONE and FAILED.
310 """
311 with self._lock:
312 task = self._tasks.get(task_id)
313 if not task:
314 return False
315 if task.status in TERMINAL_STATUSES:
316 return False
317 task.status = TaskStatus.CANCELLED
318 task.completed_at = time.monotonic()
319 self._history.append(task)
320 self._remove_from_active_locked(task_id, task.task_type)
321 self._remove_from_queue_locked(task_id, task.task_type)
322 self._notify()
323 return True
325 def advance(self, task_type: str | None = None) -> Task | None:
326 """Pop the next queued task of this type and mark it active.
327 If *task_type* is given, only advance that type's queue.
328 If omitted, advance any type that still has a free slot.
329 Respects the per-type capacity: returns None once all slots are full.
330 """
331 advanced: Task | None = None
332 with self._lock:
333 types = [task_type] if task_type else list(self._queues.keys())
334 for tt in types:
335 active = self._active_ids.setdefault(tt, set())
336 if len(active) >= self._capacity_for(tt):
337 continue
338 queue = self._queues.get(tt, [])
339 if not queue:
340 continue
341 tid = queue.pop(0)
342 task = self._tasks.get(tid)
343 if task:
344 task.status = TaskStatus.ACTIVE
345 task.started_at = time.monotonic()
346 active.add(tid)
347 advanced = task
348 break
349 if advanced is not None:
350 self._notify()
351 return advanced
353 def remove_task(self, task_id: str) -> None:
354 """Remove a task from both live tracking and history.
356 Most callers shouldn't need this in normal flow. Completed rows
357 linger in the Task Center on purpose so users can review recent
358 work, and ``clear_history()`` handles bulk pruning. This stays
359 for tests and administrative paths that need to drop a specific
360 id.
361 """
362 with self._lock:
363 task = self._tasks.pop(task_id, None)
364 if task:
365 self._remove_from_active_locked(task_id, task.task_type)
366 self._remove_from_queue_locked(task_id, task.task_type)
367 self._history = [t for t in self._history if t.task_id != task_id]
368 self._notify()
370 def clear_history(self) -> int:
371 """Drop all DONE/FAILED/CANCELLED entries from history.
373 Returns the number of rows cleared. The Task Center binds this
374 to ``C`` so the user can tidy up once they're done inspecting.
375 """
376 with self._lock:
377 cleared = len(self._history)
378 self._history = []
379 # Also drop the backing task records so memory is actually freed.
380 self._tasks = {
381 tid: t
382 for tid, t in self._tasks.items()
383 if t.status in (TaskStatus.QUEUED, TaskStatus.ACTIVE)
384 }
385 if cleared:
386 self._notify()
387 return cleared
389 def _remove_from_active_locked(self, task_id: str, task_type: str) -> None:
390 """Remove a task from active tracking. Caller must hold _lock."""
391 active = self._active_ids.get(task_type)
392 if active is not None:
393 active.discard(task_id)
395 def _remove_from_queue_locked(self, task_id: str, task_type: str) -> None:
396 """Remove a task from its type queue. Caller must hold _lock."""
397 queue = self._queues.get(task_type)
398 if queue:
399 self._queues[task_type] = [tid for tid in queue if tid != task_id]
401 def _notify(self) -> None:
402 # Snapshot under the lock so subscribe/unsubscribe from another thread
403 # (or from inside a callback) cannot mutate the list mid-iteration.
404 # Callbacks run outside the lock so synchronous subscribers that
405 # re-enter the queue (e.g. TaskBar refreshing from active_tasks)
406 # do not deadlock on the non-reentrant lock.
407 with self._lock:
408 callbacks = list(self._on_change)
409 for callback in callbacks:
410 callback()