Coverage for src/lilbee/cli/tui/task

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, Collection

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"

42 ANALYZE = "analyze"

45class CancelOrigin(StrEnum):

46 """Who cancelled a task: the user, or the app on its way out."""

48 USER = "user"

49 EXIT = "exit"

52PERCENT = 100

53TERMINAL_STATUSES = (TaskStatus.DONE, TaskStatus.FAILED, TaskStatus.CANCELLED)

56# Every icon is one cell wide. An emoji here is double-width and shifts its row

57# out of the column the other statuses share.

58STATUS_ICONS: dict[TaskStatus, str] = {

59 TaskStatus.QUEUED: "⋯",

60 TaskStatus.ACTIVE: "▶",

61 TaskStatus.DONE: "✓",

62 TaskStatus.FAILED: "✗",

63 TaskStatus.CANCELLED: "⊘",

64}

67@dataclass

68class Task:

69 """A single unit of work in the queue."""

71 task_id: str

72 name: str

73 task_type: str

74 fn: Callable[[], None]

75 # Identity of the work; ``enqueue`` refuses a second pending task with the

76 # same key. None opts out.

77 dedupe_key: str | None = None

78 status: TaskStatus = TaskStatus.QUEUED

79 progress: float = 0.0

80 detail: str = ""

81 indeterminate: bool = False

82 # Monotonic timestamp at which the task transitioned to ACTIVE. None

83 # while QUEUED. Used by the Task Center row to render elapsed time.

84 started_at: float | None = None

85 # Monotonic timestamp at which the task reached a terminal state

86 # (DONE / FAILED / CANCELLED). None while still running. Used to

87 # freeze the elapsed-time display so it doesn't keep ticking during

88 # the 2-second post-finish flash.

89 completed_at: float | None = None

90 # Set with the CANCELLED status; None while the task is not cancelled.

91 cancel_origin: CancelOrigin | None = None

94class TaskQueue:

95 """Per-type concurrent task queue.

96 Thread-safe. Each task type (download, sync, crawl, etc.) has its own

97 independent FIFO queue. One task per type can be active simultaneously,

98 so a download does not block a sync.

100 Callers receive a *task_id* they can use to update progress, cancel, or

101 query status.

102 """

103

104 def __init__(

105 self,

106 *,

107 on_change: Callable[[], None] | None = None,

108 capacity: dict[str, int] | None = None,

109 ) -> None:

110 self._lock = threading.Lock()

111 self._tasks: dict[str, Task] = {}

112 self._queues: dict[str, list[str]] = {}

113 # Per-type set of currently-active task ids. A "type" here means

114 # sync/crawl/download/wiki; each has its own FIFO and own active slots.

115 self._active_ids: dict[str, set[str]] = {}

116 # Max concurrent active tasks per type. Defaults to 1 (single-active).

117 # Callers may override per type; types absent from the map cap at 1.

118 self._capacity: dict[str, int] = dict(capacity or {})

119 self._on_change: list[Callable[[], None]] = []

120 if on_change:

121 self._on_change.append(on_change)

122 self._history: list[Task] = []

123

124 def _capacity_for(self, task_type: str) -> int:

125 return self._capacity.get(task_type, 1)

126

127 def subscribe(self, callback: Callable[[], None]) -> None:

128 """Subscribe to task queue changes. Callback is called on any queue update."""

129 with self._lock:

130 if callback not in self._on_change:

131 self._on_change.append(callback)

132

133 def unsubscribe(self, callback: Callable[[], None]) -> None:

134 """Unsubscribe from task queue changes."""

135 with self._lock:

136 if callback in self._on_change:

137 self._on_change.remove(callback)

138

139 @property

140 def active_task(self) -> Task | None:

141 """Return any one active task. Prefer ``active_tasks`` for the full set."""

142 with self._lock:

143 for ids in self._active_ids.values():

144 for tid in ids:

145 task = self._tasks.get(tid)

146 if task:

147 return task

148 return None

149

150 @property

151 def active_tasks(self) -> list[Task]:

152 """Return all currently active tasks across all types."""

153 with self._lock:

154 tasks: list[Task] = []

155 for ids in self._active_ids.values():

156 for tid in ids:

157 task = self._tasks.get(tid)

158 if task:

159 tasks.append(task)

160 return tasks

161

162 @property

163 def queued_tasks(self) -> list[Task]:

164 with self._lock:

165 result: list[Task] = []

166 for tids in self._queues.values():

167 for tid in tids:

168 task = self._tasks.get(tid)

169 if task:

170 result.append(task)

171 return result

172

173 @property

174 def history(self) -> list[Task]:

175 with self._lock:

176 return list(self._history)

177

178 @property

179 def is_empty(self) -> bool:

180 with self._lock:

181 has_active = any(ids for ids in self._active_ids.values())

182 has_queued = any(len(q) > 0 for q in self._queues.values())

183 return not has_active and not has_queued

184

185 def get_task(self, task_id: str) -> Task | None:

186 """Look up a task by ID. Returns None if not found."""

187 with self._lock:

188 return self._tasks.get(task_id)

189

190 def find_pending(self, task_type: str, dedupe_key: str) -> Task | None:

191 """Return the queued-or-active task for this key, if there is one."""

192 with self._lock:

193 return self._find_pending_locked(task_type, dedupe_key)

194

195 def _find_pending_locked(self, task_type: str, dedupe_key: str) -> Task | None:

196 for task in self._tasks.values():

197 if (

198 task.task_type == task_type

199 and task.dedupe_key == dedupe_key

200 and task.status not in TERMINAL_STATUSES

201 ):

202 return task

203 return None

204

205 def enqueue(

206 self,

207 fn: Callable[[], None],

208 name: str,

209 task_type: str,

210 *,

211 indeterminate: bool = False,

212 dedupe_key: str | None = None,

213 ) -> str:

214 """Add a task to the per-type queue. Returns a task_id.

215

216 With a *dedupe_key* already queued or active, returns that task's id and

217 adds nothing: asking twice for one download is a double keypress, not a

218 request for two copies. The check lives here rather than at the call

219 sites so a new caller cannot reintroduce the duplicate by forgetting it.

220 """

221 task_id = uuid.uuid4().hex[:8]

222 task = Task(

223 task_id=task_id,

224 name=name,

225 task_type=task_type,

226 fn=fn,

227 indeterminate=indeterminate,

228 dedupe_key=dedupe_key,

229 )

230 with self._lock:

231 if dedupe_key is not None:

232 existing = self._find_pending_locked(task_type, dedupe_key)

233 if existing is not None:

234 return existing.task_id

235 self._tasks[task_id] = task

236 self._queues.setdefault(task_type, []).append(task_id)

237 self._notify()

238 return task_id

239

240 def update_task(

241 self,

242 task_id: str,

243 progress: float,

244 detail: str = "",

245 *,

246 indeterminate: bool | None = None,

247 ) -> None:

248 """Update progress and detail text for a task.

249 When *indeterminate* is True the task's progress bar renders as a

250 pulsing indeterminate bar instead of a percentage. When explicitly

251 False it returns to determinate mode. ``None`` leaves the flag as-is

252 so incremental progress updates don't clobber the caller's intent.

253 """

254 with self._lock:

255 task = self._tasks.get(task_id)

256 if task:

257 task.progress = progress

258 task.detail = detail

259 if indeterminate is not None:

260 task.indeterminate = indeterminate

261 self._notify()

262

263 def complete_task(self, task_id: str) -> None:

264 """Mark a task as done and append it to history.

265

266 A row that already reached a terminal state is left alone, so a

267 worker that finishes after the user cancelled it does not flip the

268 cancelled row to DONE and append it to history twice.

269

270 The task record stays in ``_tasks`` so callers can still look it

271 up by id. Bulk removal happens in ``clear_history`` or targeted

272 removal via ``remove_task``.

273 """

274 with self._lock:

275 task = self._tasks.get(task_id)

276 if task and task.status not in TERMINAL_STATUSES:

277 task.status = TaskStatus.DONE

278 task.progress = PERCENT

279 task.indeterminate = False

280 task.completed_at = time.monotonic()

281 self._history.append(task)

282 self._remove_from_active_locked(task_id, task.task_type)

283 self._remove_from_queue_locked(task_id, task.task_type)

284 self._notify()

285

286 def fail_task(self, task_id: str, detail: str = "") -> None:

287 """Mark a task as failed and append it to history.

288

289 Terminal rows are immutable here too: a cancelled task whose worker

290 then raises stays cancelled.

291

292 The task record stays in ``_tasks`` so callers can still inspect

293 its detail. Bulk removal happens in ``clear_history`` or

294 targeted removal via ``remove_task``.

295 """

296 with self._lock:

297 task = self._tasks.get(task_id)

298 if task and task.status not in TERMINAL_STATUSES:

299 task.status = TaskStatus.FAILED

300 task.detail = detail

301 task.completed_at = time.monotonic()

302 self._history.append(task)

303 self._remove_from_active_locked(task_id, task.task_type)

304 self._remove_from_queue_locked(task_id, task.task_type)

305 self._notify()

306

307 def cancel(self, task_id: str, origin: CancelOrigin = CancelOrigin.USER) -> bool:

308 """Cancel a queued or active task. Returns True if cancelled.

309

310 Marks the row only. Stopping a running download needs

311 ``TaskBarController.cancel_task``, which also aborts the transfer.

312

313 Terminal rows (DONE / FAILED / CANCELLED) are immutable: a cancel

314 call against an already-finished task is a no-op and returns

315 False so callers that key off the return value (e.g. UI actions)

316 don't treat it as success.

317

318 Appends the cancelled task to ``_history`` so the Task Center

319 renders it as a lingering ``cancelled`` row, matching the

320 post-flash contract for DONE and FAILED.

321 """

322 with self._lock:

323 task = self._tasks.get(task_id)

324 if not task:

325 return False

326 if task.status in TERMINAL_STATUSES:

327 return False

328 task.status = TaskStatus.CANCELLED

329 task.cancel_origin = origin

330 task.completed_at = time.monotonic()

331 self._history.append(task)

332 self._remove_from_active_locked(task_id, task.task_type)

333 self._remove_from_queue_locked(task_id, task.task_type)

334 self._notify()

335 return True

336

337 def set_cancel_reason(self, task_id: str, detail: str) -> None:

338 """Show *detail* as the reason on a cancelled row; other rows are left alone."""

339 with self._lock:

340 task = self._tasks.get(task_id)

341 if task and task.status is TaskStatus.CANCELLED:

342 task.detail = detail

343 self._notify()

344

345 def advance(self, task_type: str | None = None) -> Task | None:

346 """Pop the next queued task of this type and mark it active.

347 If *task_type* is given, only advance that type's queue.

348 If omitted, advance any type that still has a free slot.

349 Respects the per-type capacity: returns None once all slots are full.

350 """

351 advanced: Task | None = None

352 with self._lock:

353 types = [task_type] if task_type else list(self._queues.keys())

354 for tt in types:

355 active = self._active_ids.setdefault(tt, set())

356 if len(active) >= self._capacity_for(tt):

357 continue

358 queue = self._queues.get(tt, [])

359 if not queue:

360 continue

361 tid = queue.pop(0)

362 task = self._tasks.get(tid)

363 if task:

364 task.status = TaskStatus.ACTIVE

365 task.started_at = time.monotonic()

366 active.add(tid)

367 advanced = task

368 break

369 if advanced is not None:

370 self._notify()

371 return advanced

372

373 def remove_task(self, task_id: str) -> None:

374 """Remove a task from both live tracking and history.

375

376 Most callers shouldn't need this in normal flow. Completed rows

377 linger in the Task Center on purpose so users can review recent

378 work, and ``clear_history()`` handles bulk pruning. This stays

379 for tests and administrative paths that need to drop a specific

380 id.

381 """

382 with self._lock:

383 task = self._tasks.pop(task_id, None)

384 if task:

385 self._remove_from_active_locked(task_id, task.task_type)

386 self._remove_from_queue_locked(task_id, task.task_type)

387 self._history = [t for t in self._history if t.task_id != task_id]

388 self._notify()

389

390 def clear_history(self, keep: Collection[str] = ()) -> int:

391 """Drop all DONE/FAILED/CANCELLED entries from history, except the ids in ``keep``.

392

393 Returns the number of rows cleared. The Task Center binds this

394 to ``C`` so the user can tidy up once they're done inspecting.

395 """

396 with self._lock:

397 kept = [t for t in self._history if t.task_id in keep]

398 cleared = len(self._history) - len(kept)

399 self._history = kept

400 # Also drop the backing task records so memory is actually freed.

401 self._tasks = {

402 tid: t

403 for tid, t in self._tasks.items()

404 if t.status in (TaskStatus.QUEUED, TaskStatus.ACTIVE) or tid in keep

405 }

406 if cleared:

407 self._notify()

408 return cleared

409

410 def _remove_from_active_locked(self, task_id: str, task_type: str) -> None:

411 """Remove a task from active tracking. Caller must hold _lock."""

412 active = self._active_ids.get(task_type)

413 if active is not None:

414 active.discard(task_id)

415

416 def _remove_from_queue_locked(self, task_id: str, task_type: str) -> None:

417 """Remove a task from its type queue. Caller must hold _lock."""

418 queue = self._queues.get(task_type)

419 if queue:

420 self._queues[task_type] = [tid for tid in queue if tid != task_id]

421

422 def _notify(self) -> None:

423 # Snapshot under the lock so subscribe/unsubscribe from another thread

424 # (or from inside a callback) cannot mutate the list mid-iteration.

425 # Callbacks run outside the lock so synchronous subscribers that

426 # re-enter the queue (e.g. TaskBar refreshing from active_tasks)

427 # do not deadlock on the non-reentrant lock.

428 with self._lock:

429 callbacks = list(self._on_change)

430 for callback in callbacks:

431 callback()

Coverage for src/lilbee/cli/tui/task_queue.py: 100%

234 statements