tests/test_cmd_task_queue.py · gabriel/muse

test_cmd_task_queue.py python

4,653 lines 191.1 KB

sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago

1	"""Tests for muse coord task-queue: core + CLI (enqueue / claim / complete / fail-task /
2	cancel-task / tasks).
3
4	Coverage goals
5	--------------
6	* Unit — every public function in ``muse.core.task_queue``
7	* Integration — full lifecycle: enqueue → claim → complete/fail/cancel
8	* CLI — all six CLI subcommands via argparse dispatch and stdout capture
9	* Security — UUID validation, path traversal, ANSI injection, oversized inputs
10	* Stress — concurrent claiming correctness, large queue scanning
11
12	Test conventions
13	----------------
14	* Every test receives a fresh ``tmp_path``-based repo fixture.
15	* Time is frozen via ``unittest.mock.patch`` on ``muse.core.task_queue._now_utc``
16	wherever predictable timestamps are required.
17	* CLI dispatch calls ``run_*`` directly (no subprocess overhead) with a
18	``argparse.Namespace`` assembled by hand, capturing stdout/stderr via
19	``capsys``.
20	"""
21
22	from __future__ import annotations
23
24	import argparse
25	import datetime
26	import json
27	import os
28	import pathlib
29	import threading
30	import time
31	import uuid
32	from collections.abc import Generator
33	from contextlib import AbstractContextManager
34	from unittest.mock import MagicMock, patch
35
36	from muse.core._types import MsgpackValue
37
38	import pytest
39
40	from muse.core.task_queue import (
41	ClaimRecord,
42	TaskRecord,
43	_claims_dir,
44	_tasks_dir,
45	_try_excl_claim,
46	_try_optimistic_reclaim,
47	_validate_queue_name,
48	_validate_task_id,
49	cancel_task,
50	claim_next_task,
51	complete_task,
52	create_task,
53	ensure_task_dirs,
54	fail_task,
55	get_task_status,
56	heartbeat_claim,
57	load_all_claims,
58	load_all_tasks,
59	load_claim,
60	load_task,
61	)
62	from muse.cli.commands.task_queue import (
63	register_all,
64	run_cancel_task,
65	run_claim,
66	run_complete,
67	run_enqueue,
68	run_fail_task,
69	run_tasks,
70	)
71
72	# ── Fixtures ──────────────────────────────────────────────────────────────────
73
74	UTC = datetime.timezone.utc
75	_EPOCH = datetime.datetime(2025, 6, 1, 12, 0, 0, tzinfo=UTC)
76
77	VALID_UUID = "12345678-1234-4abc-8abc-1234567890ab"
78	VALID_UUID2 = "aaaabbbb-aaaa-4bbb-8bbb-aaaaaaaaaaaa"
79
80
81	def _make_repo(tmp_path: pathlib.Path) -> pathlib.Path:
82	"""Return a minimal muse repo root with a ``.muse/`` directory."""
83	muse_dir = tmp_path / ".muse"
84	muse_dir.mkdir(parents=True)
85	return tmp_path
86
87
88	def _freeze(ts: datetime.datetime) -> AbstractContextManager[MagicMock]:
89	"""Context manager: freeze ``muse.core.task_queue._now_utc`` to ts."""
90	return patch("muse.core.task_queue._now_utc", return_value=ts)
91
92
93	def _namespace(**kwargs: MsgpackValue) -> argparse.Namespace:
94	"""Build an ``argparse.Namespace`` with sane defaults for CLI tests."""
95	defaults = {
96	"fmt": "json",
97	"run_id": "agent-1",
98	"queue": None,
99	"title": "Test task",
100	"priority": 0,
101	"ttl_seconds": 86400,
102	"payload": "{}",
103	"tags": "",
104	"claim_ttl": 3600,
105	"wait": 0,
106	"task_id": VALID_UUID,
107	"result": "{}",
108	"error": "",
109	"force": False,
110	"status": None,
111	"limit": 200,
112	}
113	defaults.update(kwargs)
114	return argparse.Namespace(**defaults)
115
116
117	# ── Validation ─────────────────────────────────────────────────────────────────
118
119
120	class TestValidateTaskId:
121	"""_validate_task_id must accept well-formed UUID4 and reject everything else."""
122
123	def test_accepts_valid_uuid4(self) -> None:
124	_validate_task_id(VALID_UUID) # no exception
125
126	def test_accepts_uppercase(self) -> None:
127	_validate_task_id(VALID_UUID.upper())
128
129	def test_rejects_empty(self) -> None:
130	with pytest.raises(ValueError, match="UUID4"):
131	_validate_task_id("")
132
133	def test_rejects_non_uuid(self) -> None:
134	with pytest.raises(ValueError):
135	_validate_task_id("not-a-uuid")
136
137	def test_rejects_path_traversal(self) -> None:
138	with pytest.raises(ValueError):
139	_validate_task_id("../../etc/passwd")
140
141	def test_rejects_null_bytes(self) -> None:
142	with pytest.raises(ValueError):
143	_validate_task_id("\x00" * 36)
144
145	def test_rejects_uuid3(self) -> None:
146	# UUID version 3 — digit 13 = '3', not '4'
147	with pytest.raises(ValueError):
148	_validate_task_id("12345678-1234-3abc-8abc-1234567890ab")
149
150	def test_rejects_uuid_with_slash(self) -> None:
151	with pytest.raises(ValueError):
152	_validate_task_id("12345678-1234-4abc-8abc-123456789/ab")
153
154
155	class TestValidateQueueName:
156	"""_validate_queue_name must accept valid names and reject bad ones."""
157
158	def test_accepts_simple(self) -> None:
159	_validate_queue_name("default")
160	_validate_queue_name("billing-queue")
161	_validate_queue_name("Agent_123")
162
163	def test_rejects_empty(self) -> None:
164	with pytest.raises(ValueError, match="non-empty"):
165	_validate_queue_name("")
166
167	def test_rejects_space(self) -> None:
168	with pytest.raises(ValueError):
169	_validate_queue_name("queue name")
170
171	def test_rejects_slash(self) -> None:
172	with pytest.raises(ValueError):
173	_validate_queue_name("../../etc")
174
175	def test_rejects_null_byte(self) -> None:
176	with pytest.raises(ValueError):
177	_validate_queue_name("queue\x00name")
178
179	def test_rejects_too_long(self) -> None:
180	with pytest.raises(ValueError, match="too long"):
181	_validate_queue_name("q" * 65)
182
183	def test_accepts_max_length(self) -> None:
184	_validate_queue_name("q" * 64)
185
186
187	# ── TaskRecord ─────────────────────────────────────────────────────────────────
188
189
190	class TestTaskRecord:
191	"""TaskRecord serialisation round-trip and is_expired logic."""
192
193	def _make(self, **kwargs: MsgpackValue) -> TaskRecord:
194	defaults = dict(
195	task_id=VALID_UUID,
196	title="A task",
197	payload={"x": 1},
198	priority=0,
199	queue="default",
200	created_at=_EPOCH,
201	created_by="orchestrator",
202	ttl_seconds=3600,
203	tags=["a", "b"],
204	)
205	defaults.update(kwargs)
206	return TaskRecord(**defaults)
207
208	def test_to_dict_round_trip(self) -> None:
209	t = self._make()
210	d = t.to_dict()
211	t2 = TaskRecord.from_dict(d)
212	assert t2.task_id == t.task_id
213	assert t2.title == t.title
214	assert t2.priority == t.priority
215	assert t2.queue == t.queue
216	assert t2.tags == t.tags
217	assert t2.payload == t.payload
218
219	def test_is_expired_false_within_ttl(self) -> None:
220	t = self._make()
221	now = _EPOCH + datetime.timedelta(seconds=3599)
222	assert t.is_expired(now) is False
223
224	def test_is_expired_true_at_boundary(self) -> None:
225	t = self._make()
226	now = _EPOCH + datetime.timedelta(seconds=3600)
227	assert t.is_expired(now) is True
228
229	def test_from_dict_missing_created_at_defaults_to_now(self) -> None:
230	d = {"task_id": VALID_UUID, "title": "x"}
231	t = TaskRecord.from_dict(d)
232	assert isinstance(t.created_at, datetime.datetime)
233
234	def test_title_truncated_at_256(self) -> None:
235	d = {"task_id": VALID_UUID, "title": "x" * 300, "created_at": _EPOCH.isoformat()}
236	t = TaskRecord.from_dict(d)
237	assert len(t.title) <= 256
238
239
240	# ── ClaimRecord ────────────────────────────────────────────────────────────────
241
242
243	class TestClaimRecord:
244	"""ClaimRecord serialisation round-trip and is_expired logic."""
245
246	def _make(self, **kwargs: MsgpackValue) -> ClaimRecord:
247	defaults = dict(
248	task_id=VALID_UUID,
249	claimer_run_id="agent-1",
250	claimed_at=_EPOCH,
251	expires_at=_EPOCH + datetime.timedelta(hours=1),
252	status="claimed",
253	heartbeat_at=_EPOCH,
254	claim_nonce=str(uuid.uuid4()),
255	result=None,
256	error=None,
257	)
258	defaults.update(kwargs)
259	return ClaimRecord(**defaults)
260
261	def test_to_dict_round_trip(self) -> None:
262	c = self._make()
263	d = c.to_dict()
264	c2 = ClaimRecord.from_dict(d)
265	assert c2.task_id == c.task_id
266	assert c2.claimer_run_id == c.claimer_run_id
267	assert c2.status == c.status
268	assert c2.claim_nonce == c.claim_nonce
269
270	def test_is_expired_false(self) -> None:
271	c = self._make()
272	assert c.is_expired(_EPOCH) is False
273
274	def test_is_expired_true(self) -> None:
275	c = self._make()
276	assert c.is_expired(_EPOCH + datetime.timedelta(hours=2)) is True
277
278
279	# ── get_task_status ────────────────────────────────────────────────────────────
280
281
282	class TestGetTaskStatus:
283	"""Derives correct status from (task, claim, now) triple."""
284
285	def _task(self) -> TaskRecord:
286	return TaskRecord(
287	task_id=VALID_UUID, title="t", payload={}, priority=0, queue="default",
288	created_at=_EPOCH, created_by="x", ttl_seconds=3600, tags=[],
289	)
290
291	def _claim(self, **kw: MsgpackValue) -> ClaimRecord:
292	defaults = dict(
293	task_id=VALID_UUID, claimer_run_id="a", claimed_at=_EPOCH,
294	expires_at=_EPOCH + datetime.timedelta(hours=1), status="claimed",
295	heartbeat_at=_EPOCH, claim_nonce="nonce", result=None, error=None,
296	)
297	defaults.update(kw)
298	return ClaimRecord(**defaults)
299
300	def test_no_claim_is_pending(self) -> None:
301	assert get_task_status(self._task(), None, _EPOCH) == "pending"
302
303	def test_active_claim_is_claimed(self) -> None:
304	c = self._claim()
305	assert get_task_status(self._task(), c, _EPOCH) == "claimed"
306
307	def test_expired_claim_is_timed_out(self) -> None:
308	c = self._claim()
309	now = _EPOCH + datetime.timedelta(hours=2)
310	assert get_task_status(self._task(), c, now) == "timed_out"
311
312	def test_completed_status_passes_through(self) -> None:
313	c = self._claim(status="completed")
314	assert get_task_status(self._task(), c, _EPOCH) == "completed"
315
316	def test_failed_status_passes_through(self) -> None:
317	c = self._claim(status="failed")
318	assert get_task_status(self._task(), c, _EPOCH) == "failed"
319
320	def test_cancelled_status_passes_through(self) -> None:
321	c = self._claim(status="cancelled")
322	assert get_task_status(self._task(), c, _EPOCH) == "cancelled"
323
324
325	# ── ensure_task_dirs ───────────────────────────────────────────────────────────
326
327
328	class TestEnsureTaskDirs:
329	"""ensure_task_dirs creates both directories idempotently."""
330
331	def test_creates_directories(self, tmp_path: pathlib.Path) -> None:
332	repo = _make_repo(tmp_path)
333	ensure_task_dirs(repo)
334	assert _tasks_dir(repo).is_dir()
335	assert _claims_dir(repo).is_dir()
336
337	def test_idempotent(self, tmp_path: pathlib.Path) -> None:
338	repo = _make_repo(tmp_path)
339	ensure_task_dirs(repo)
340	ensure_task_dirs(repo) # must not raise
341
342
343	# ── create_task ────────────────────────────────────────────────────────────────
344
345
346	class TestCreateTask:
347	"""create_task validates inputs and persists a TaskRecord."""
348
349	def test_creates_file_on_disk(self, tmp_path: pathlib.Path) -> None:
350	repo = _make_repo(tmp_path)
351	with _freeze(_EPOCH):
352	t = create_task(repo, "Do X")
353	task_file = _tasks_dir(repo) / f"{t.task_id}.json"
354	assert task_file.is_file()
355
356	def test_returns_correct_fields(self, tmp_path: pathlib.Path) -> None:
357	repo = _make_repo(tmp_path)
358	with _freeze(_EPOCH):
359	t = create_task(
360	repo, "Deploy service",
361	payload={"env": "prod"},
362	priority=5,
363	queue="deploy",
364	ttl_seconds=7200,
365	created_by="ops",
366	tags=["prod", "critical"],
367	)
368	assert t.title == "Deploy service"
369	assert t.priority == 5
370	assert t.queue == "deploy"
371	assert t.ttl_seconds == 7200
372	assert t.created_by == "ops"
373	assert "prod" in t.tags
374	assert t.payload == {"env": "prod"}
375
376	def test_file_is_valid_json(self, tmp_path: pathlib.Path) -> None:
377	repo = _make_repo(tmp_path)
378	t = create_task(repo, "Validate JSON")
379	content = (_tasks_dir(repo) / f"{t.task_id}.json").read_text()
380	d = json.loads(content)
381	assert d["task_id"] == t.task_id
382
383	def test_empty_title_raises(self, tmp_path: pathlib.Path) -> None:
384	repo = _make_repo(tmp_path)
385	with pytest.raises(ValueError, match="non-empty"):
386	create_task(repo, "")
387
388	def test_invalid_queue_raises(self, tmp_path: pathlib.Path) -> None:
389	repo = _make_repo(tmp_path)
390	with pytest.raises(ValueError):
391	create_task(repo, "x", queue="bad queue!")
392
393	def test_tags_capped_at_32(self, tmp_path: pathlib.Path) -> None:
394	repo = _make_repo(tmp_path)
395	t = create_task(repo, "Lots of tags", tags=[f"tag{i}" for i in range(50)])
396	assert len(t.tags) == 32
397
398	def test_ttl_min_1(self, tmp_path: pathlib.Path) -> None:
399	repo = _make_repo(tmp_path)
400	t = create_task(repo, "Short TTL", ttl_seconds=0)
401	assert t.ttl_seconds >= 1
402
403
404	# ── load_all_tasks / load_task ─────────────────────────────────────────────────
405
406
407	class TestLoadTasks:
408	"""Scanning and loading task records from the tasks directory."""
409
410	def test_empty_dir_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
411	repo = _make_repo(tmp_path)
412	assert load_all_tasks(repo) == []
413
414	def test_non_existent_dir_returns_empty_list(self, tmp_path: pathlib.Path) -> None:
415	repo = _make_repo(tmp_path)
416	assert load_all_tasks(repo) == []
417
418	def test_loads_created_task(self, tmp_path: pathlib.Path) -> None:
419	repo = _make_repo(tmp_path)
420	t = create_task(repo, "A task")
421	tasks = load_all_tasks(repo)
422	assert len(tasks) == 1
423	assert tasks[0].task_id == t.task_id
424
425	def test_skips_corrupt_file(self, tmp_path: pathlib.Path) -> None:
426	repo = _make_repo(tmp_path)
427	ensure_task_dirs(repo)
428	corrupt = _tasks_dir(repo) / f"{VALID_UUID}.json"
429	corrupt.write_text("NOT JSON")
430	tasks = load_all_tasks(repo)
431	assert tasks == []
432
433	def test_load_task_by_id(self, tmp_path: pathlib.Path) -> None:
434	repo = _make_repo(tmp_path)
435	t = create_task(repo, "Named task")
436	loaded = load_task(repo, t.task_id)
437	assert loaded is not None
438	assert loaded.title == "Named task"
439
440	def test_load_task_missing_returns_none(self, tmp_path: pathlib.Path) -> None:
441	repo = _make_repo(tmp_path)
442	ensure_task_dirs(repo)
443	assert load_task(repo, VALID_UUID) is None
444
445	def test_load_task_invalid_id_raises(self, tmp_path: pathlib.Path) -> None:
446	repo = _make_repo(tmp_path)
447	with pytest.raises(ValueError):
448	load_task(repo, "not-a-uuid")
449
450
451	# ── _try_excl_claim ────────────────────────────────────────────────────────────
452
453
454	class TestTryExclClaim:
455	"""O_CREAT\|O_EXCL atomic claiming primitive."""
456
457	def test_first_claim_succeeds(self, tmp_path: pathlib.Path) -> None:
458	repo = _make_repo(tmp_path)
459	ensure_task_dirs(repo)
460	with _freeze(_EPOCH):
461	result = _try_excl_claim(repo, VALID_UUID, "agent-1", _EPOCH, 3600)
462	assert result is not None
463	assert result.claimer_run_id == "agent-1"
464	assert result.status == "claimed"
465
466	def test_second_claim_returns_none(self, tmp_path: pathlib.Path) -> None:
467	repo = _make_repo(tmp_path)
468	ensure_task_dirs(repo)
469	with _freeze(_EPOCH):
470	first = _try_excl_claim(repo, VALID_UUID, "agent-1", _EPOCH, 3600)
471	second = _try_excl_claim(repo, VALID_UUID, "agent-2", _EPOCH, 3600)
472	assert first is not None
473	assert second is None
474
475	def test_claim_file_is_written(self, tmp_path: pathlib.Path) -> None:
476	repo = _make_repo(tmp_path)
477	ensure_task_dirs(repo)
478	with _freeze(_EPOCH):
479	_try_excl_claim(repo, VALID_UUID, "agent-1", _EPOCH, 3600)
480	claim_file = _claims_dir(repo) / f"{VALID_UUID}.json"
481	assert claim_file.is_file()
482
483	def test_claim_file_is_valid_json(self, tmp_path: pathlib.Path) -> None:
484	repo = _make_repo(tmp_path)
485	ensure_task_dirs(repo)
486	with _freeze(_EPOCH):
487	claim = _try_excl_claim(repo, VALID_UUID, "agent-1", _EPOCH, 3600)
488	content = (_claims_dir(repo) / f"{VALID_UUID}.json").read_text()
489	d = json.loads(content)
490	assert d["claim_nonce"] == claim.claim_nonce
491
492	def test_expires_at_correct(self, tmp_path: pathlib.Path) -> None:
493	repo = _make_repo(tmp_path)
494	ensure_task_dirs(repo)
495	with _freeze(_EPOCH):
496	claim = _try_excl_claim(repo, VALID_UUID, "agent-1", _EPOCH, 7200)
497	expected = _EPOCH + datetime.timedelta(seconds=7200)
498	assert claim.expires_at == expected
499
500
501	# ── _try_optimistic_reclaim ────────────────────────────────────────────────────
502
503
504	class TestTryOptimisticReclaim:
505	"""Reclaim timed-out tasks via atomic rename + nonce verification."""
506
507	def test_reclaim_wins_when_no_competition(self, tmp_path: pathlib.Path) -> None:
508	repo = _make_repo(tmp_path)
509	ensure_task_dirs(repo)
510	# First claim
511	with _freeze(_EPOCH):
512	_try_excl_claim(repo, VALID_UUID, "agent-1", _EPOCH, 1)
513	# Now reclaim at t+2 (expired)
514	now = _EPOCH + datetime.timedelta(seconds=2)
515	with _freeze(now):
516	result = _try_optimistic_reclaim(repo, VALID_UUID, "agent-2", now, 3600)
517	assert result is not None
518	assert result.claimer_run_id == "agent-2"
519
520	def test_nonce_written_to_file(self, tmp_path: pathlib.Path) -> None:
521	repo = _make_repo(tmp_path)
522	ensure_task_dirs(repo)
523	with _freeze(_EPOCH):
524	_try_excl_claim(repo, VALID_UUID, "agent-1", _EPOCH, 1)
525	now = _EPOCH + datetime.timedelta(seconds=2)
526	with _freeze(now):
527	result = _try_optimistic_reclaim(repo, VALID_UUID, "agent-2", now, 3600)
528	if result: # may fail under extreme race — just check if written
529	content = json.loads((_claims_dir(repo) / f"{VALID_UUID}.json").read_text())
530	assert content["claim_nonce"] == result.claim_nonce
531
532
533	# ── claim_next_task ────────────────────────────────────────────────────────────
534
535
536	class TestClaimNextTask:
537	"""High-level claim_next_task: priority ordering, queue filtering, expiry re-claim."""
538
539	def test_returns_none_on_empty_queue(self, tmp_path: pathlib.Path) -> None:
540	repo = _make_repo(tmp_path)
541	with _freeze(_EPOCH):
542	result = claim_next_task(repo, "agent-1")
543	assert result is None
544
545	def test_claims_only_task(self, tmp_path: pathlib.Path) -> None:
546	repo = _make_repo(tmp_path)
547	with _freeze(_EPOCH):
548	t = create_task(repo, "Only task")
549	result = claim_next_task(repo, "agent-1")
550	assert result is not None
551	task, claim = result
552	assert task.task_id == t.task_id
553	assert claim.claimer_run_id == "agent-1"
554
555	def test_higher_priority_claimed_first(self, tmp_path: pathlib.Path) -> None:
556	repo = _make_repo(tmp_path)
557	with _freeze(_EPOCH):
558	low = create_task(repo, "Low priority", priority=0)
559	high = create_task(repo, "High priority", priority=10)
560	result = claim_next_task(repo, "agent-1")
561	assert result is not None
562	task, _ = result
563	assert task.task_id == high.task_id
564
565	def test_fifo_within_same_priority(self, tmp_path: pathlib.Path) -> None:
566	repo = _make_repo(tmp_path)
567	with _freeze(_EPOCH):
568	first = create_task(repo, "First task", priority=5)
569	with _freeze(_EPOCH + datetime.timedelta(seconds=1)):
570	_second = create_task(repo, "Second task", priority=5)
571	with _freeze(_EPOCH + datetime.timedelta(seconds=2)):
572	result = claim_next_task(repo, "agent-1")
573	assert result is not None
574	task, _ = result
575	assert task.task_id == first.task_id
576
577	def test_queue_filter_respected(self, tmp_path: pathlib.Path) -> None:
578	repo = _make_repo(tmp_path)
579	with _freeze(_EPOCH):
580	billing = create_task(repo, "Billing job", queue="billing")
581	_ops = create_task(repo, "Ops job", queue="ops")
582	result = claim_next_task(repo, "agent-1", queue="billing")
583	assert result is not None
584	task, _ = result
585	assert task.task_id == billing.task_id
586
587	def test_queue_filter_returns_none_when_no_match(self, tmp_path: pathlib.Path) -> None:
588	repo = _make_repo(tmp_path)
589	with _freeze(_EPOCH):
590	_ops = create_task(repo, "Ops job", queue="ops")
591	result = claim_next_task(repo, "agent-1", queue="billing")
592	assert result is None
593
594	def test_already_claimed_task_not_re_claimed(self, tmp_path: pathlib.Path) -> None:
595	repo = _make_repo(tmp_path)
596	with _freeze(_EPOCH):
597	_t = create_task(repo, "Unique task")
598	claim_next_task(repo, "agent-1")
599	result2 = claim_next_task(repo, "agent-2")
600	assert result2 is None
601
602	def test_expired_task_ttl_skipped(self, tmp_path: pathlib.Path) -> None:
603	repo = _make_repo(tmp_path)
604	with _freeze(_EPOCH):
605	_t = create_task(repo, "Expired task", ttl_seconds=10)
606	future = _EPOCH + datetime.timedelta(seconds=20)
607	with _freeze(future):
608	result = claim_next_task(repo, "agent-1")
609	assert result is None
610
611	def test_reclaims_timed_out_task(self, tmp_path: pathlib.Path) -> None:
612	repo = _make_repo(tmp_path)
613	with _freeze(_EPOCH):
614	_t = create_task(repo, "Timed-out task", ttl_seconds=86400)
615	claim_next_task(repo, "agent-1", claim_ttl_seconds=10)
616	# Advance past claim TTL
617	future = _EPOCH + datetime.timedelta(seconds=20)
618	with _freeze(future):
619	result = claim_next_task(repo, "agent-2", claim_ttl_seconds=3600)
620	assert result is not None
621	_, claim = result
622	assert claim.claimer_run_id == "agent-2"
623
624
625	# ── complete_task ──────────────────────────────────────────────────────────────
626
627
628	class TestCompleteTask:
629	"""complete_task updates status, validates ownership."""
630
631	def _enqueue_and_claim(self, repo: pathlib.Path, run_id: str = "agent-1") -> tuple[TaskRecord, ClaimRecord]:
632	t = create_task(repo, "Task")
633	with _freeze(_EPOCH):
634	result = claim_next_task(repo, run_id)
635	assert result is not None
636	return result
637
638	def test_completes_successfully(self, tmp_path: pathlib.Path) -> None:
639	repo = _make_repo(tmp_path)
640	task, _claim = self._enqueue_and_claim(repo)
641	claim = complete_task(repo, task.task_id, "agent-1", result={"pr": 42})
642	assert claim.status == "completed"
643	assert claim.result == {"pr": 42}
644
645	def test_persisted_to_disk(self, tmp_path: pathlib.Path) -> None:
646	repo = _make_repo(tmp_path)
647	task, _claim = self._enqueue_and_claim(repo)
648	complete_task(repo, task.task_id, "agent-1")
649	disk = json.loads((_claims_dir(repo) / f"{task.task_id}.json").read_text())
650	assert disk["status"] == "completed"
651
652	def test_wrong_run_id_raises_permission_error(self, tmp_path: pathlib.Path) -> None:
653	repo = _make_repo(tmp_path)
654	task, _claim = self._enqueue_and_claim(repo)
655	with pytest.raises(PermissionError):
656	complete_task(repo, task.task_id, "impostor")
657
658	def test_invalid_task_id_raises_value_error(self, tmp_path: pathlib.Path) -> None:
659	repo = _make_repo(tmp_path)
660	with pytest.raises(ValueError):
661	complete_task(repo, "not-a-uuid", "agent-1")
662
663	def test_missing_task_raises_file_not_found(self, tmp_path: pathlib.Path) -> None:
664	repo = _make_repo(tmp_path)
665	ensure_task_dirs(repo)
666	with pytest.raises(FileNotFoundError):
667	complete_task(repo, VALID_UUID, "agent-1")
668
669	def test_double_complete_raises_runtime_error(self, tmp_path: pathlib.Path) -> None:
670	repo = _make_repo(tmp_path)
671	task, _claim = self._enqueue_and_claim(repo)
672	complete_task(repo, task.task_id, "agent-1")
673	with pytest.raises(RuntimeError):
674	complete_task(repo, task.task_id, "agent-1")
675
676
677	# ── fail_task ──────────────────────────────────────────────────────────────────
678
679
680	class TestFailTask:
681	"""fail_task updates status to failed with an error message."""
682
683	def test_fails_successfully(self, tmp_path: pathlib.Path) -> None:
684	repo = _make_repo(tmp_path)
685	t = create_task(repo, "Doomed task")
686	with _freeze(_EPOCH):
687	claim_next_task(repo, "agent-1")
688	claim = fail_task(repo, t.task_id, "agent-1", error="timeout after 30s")
689	assert claim.status == "failed"
690	assert claim.error == "timeout after 30s"
691
692	def test_wrong_claimer_raises(self, tmp_path: pathlib.Path) -> None:
693	repo = _make_repo(tmp_path)
694	t = create_task(repo, "Doomed task")
695	with _freeze(_EPOCH):
696	claim_next_task(repo, "agent-1")
697	with pytest.raises(PermissionError):
698	fail_task(repo, t.task_id, "agent-2", error="oops")
699
700	def test_already_failed_raises(self, tmp_path: pathlib.Path) -> None:
701	repo = _make_repo(tmp_path)
702	t = create_task(repo, "Doomed task")
703	with _freeze(_EPOCH):
704	claim_next_task(repo, "agent-1")
705	fail_task(repo, t.task_id, "agent-1", error="first failure")
706	with pytest.raises(RuntimeError):
707	fail_task(repo, t.task_id, "agent-1", error="second failure")
708
709
710	# ── cancel_task ────────────────────────────────────────────────────────────────
711
712
713	class TestCancelTask:
714	"""cancel_task handles pending, claimed, and force-cancel cases."""
715
716	def test_cancel_pending_task(self, tmp_path: pathlib.Path) -> None:
717	repo = _make_repo(tmp_path)
718	t = create_task(repo, "Unneeded task")
719	claim = cancel_task(repo, t.task_id, "orchestrator")
720	assert claim.status == "cancelled"
721
722	def test_cancel_claimed_by_claimer(self, tmp_path: pathlib.Path) -> None:
723	repo = _make_repo(tmp_path)
724	t = create_task(repo, "Running task")
725	with _freeze(_EPOCH):
726	claim_next_task(repo, "agent-1")
727	claim = cancel_task(repo, t.task_id, "agent-1")
728	assert claim.status == "cancelled"
729
730	def test_cancel_claimed_by_non_claimer_raises(self, tmp_path: pathlib.Path) -> None:
731	repo = _make_repo(tmp_path)
732	t = create_task(repo, "Running task")
733	with _freeze(_EPOCH):
734	claim_next_task(repo, "agent-1")
735	with pytest.raises(PermissionError):
736	cancel_task(repo, t.task_id, "agent-2")
737
738	def test_force_cancel_overrides_ownership(self, tmp_path: pathlib.Path) -> None:
739	repo = _make_repo(tmp_path)
740	t = create_task(repo, "Running task")
741	with _freeze(_EPOCH):
742	claim_next_task(repo, "agent-1")
743	claim = cancel_task(repo, t.task_id, "orchestrator", force=True)
744	assert claim.status == "cancelled"
745
746	def test_cancel_nonexistent_task_raises(self, tmp_path: pathlib.Path) -> None:
747	repo = _make_repo(tmp_path)
748	ensure_task_dirs(repo)
749	with pytest.raises(FileNotFoundError):
750	cancel_task(repo, VALID_UUID, "orchestrator")
751
752	def test_cancel_completed_raises(self, tmp_path: pathlib.Path) -> None:
753	repo = _make_repo(tmp_path)
754	t = create_task(repo, "Done task")
755	with _freeze(_EPOCH):
756	claim_next_task(repo, "agent-1")
757	complete_task(repo, t.task_id, "agent-1")
758	with pytest.raises(RuntimeError, match="terminal"):
759	cancel_task(repo, t.task_id, "agent-1")
760
761	def test_invalid_id_raises(self, tmp_path: pathlib.Path) -> None:
762	repo = _make_repo(tmp_path)
763	with pytest.raises(ValueError):
764	cancel_task(repo, "not-valid", "agent")
765
766
767	# ── heartbeat_claim ────────────────────────────────────────────────────────────
768
769
770	class TestHeartbeatClaim:
771	"""heartbeat_claim extends expires_at and updates heartbeat_at."""
772
773	def test_extends_expiry(self, tmp_path: pathlib.Path) -> None:
774	repo = _make_repo(tmp_path)
775	t = create_task(repo, "Long running task")
776	with _freeze(_EPOCH):
777	claim_next_task(repo, "agent-1", claim_ttl_seconds=3600)
778	now = _EPOCH + datetime.timedelta(seconds=1800)
779	with _freeze(now):
780	claim = heartbeat_claim(repo, t.task_id, "agent-1", extension_seconds=7200)
781	assert claim.expires_at == now + datetime.timedelta(seconds=7200)
782	assert claim.heartbeat_at == now
783
784	def test_wrong_claimer_raises(self, tmp_path: pathlib.Path) -> None:
785	repo = _make_repo(tmp_path)
786	t = create_task(repo, "Task")
787	with _freeze(_EPOCH):
788	claim_next_task(repo, "agent-1")
789	with pytest.raises(PermissionError):
790	heartbeat_claim(repo, t.task_id, "agent-2")
791
792	def test_no_claim_raises(self, tmp_path: pathlib.Path) -> None:
793	repo = _make_repo(tmp_path)
794	ensure_task_dirs(repo)
795	# Create task file but no claim
796	create_task(repo, "Unclaimed")
797	t = load_all_tasks(repo)[0]
798	with pytest.raises(FileNotFoundError):
799	heartbeat_claim(repo, t.task_id, "agent-1")
800
801	def test_heartbeat_after_complete_raises(self, tmp_path: pathlib.Path) -> None:
802	repo = _make_repo(tmp_path)
803	t = create_task(repo, "Done task")
804	with _freeze(_EPOCH):
805	claim_next_task(repo, "agent-1")
806	complete_task(repo, t.task_id, "agent-1")
807	with pytest.raises(RuntimeError):
808	heartbeat_claim(repo, t.task_id, "agent-1")
809
810
811	# ── Full lifecycle integration ─────────────────────────────────────────────────
812
813
814	class TestFullLifecycle:
815	"""End-to-end: enqueue → claim → heartbeat → complete/fail/cancel."""
816
817	def test_enqueue_claim_complete(self, tmp_path: pathlib.Path) -> None:
818	repo = _make_repo(tmp_path)
819	with _freeze(_EPOCH):
820	t = create_task(repo, "E2E task", payload={"op": "refactor"}, priority=3)
821	result = claim_next_task(repo, "agent-1", claim_ttl_seconds=300)
822
823	assert result is not None
824	task, claim = result
825	assert task.task_id == t.task_id
826	assert claim.status == "claimed"
827
828	claim = complete_task(repo, task.task_id, "agent-1", result={"status": "ok"})
829	assert claim.status == "completed"
830
831	# Second claim attempt after completion → queue empty
832	with _freeze(_EPOCH + datetime.timedelta(seconds=1)):
833	result2 = claim_next_task(repo, "agent-2")
834	assert result2 is None
835
836	def test_enqueue_claim_fail_then_reclaim(self, tmp_path: pathlib.Path) -> None:
837	repo = _make_repo(tmp_path)
838	with _freeze(_EPOCH):
839	t = create_task(repo, "Failing task", ttl_seconds=86400)
840	claim_next_task(repo, "agent-1", claim_ttl_seconds=10)
841	fail_task(repo, t.task_id, "agent-1", error="network timeout")
842
843	# After failure, task is done — no re-claim
844	with _freeze(_EPOCH + datetime.timedelta(seconds=30)):
845	result = claim_next_task(repo, "agent-2")
846	assert result is None # failed tasks are not re-claimable
847
848	def test_heartbeat_prevents_expiry(self, tmp_path: pathlib.Path) -> None:
849	repo = _make_repo(tmp_path)
850	with _freeze(_EPOCH):
851	t = create_task(repo, "Long job", ttl_seconds=86400)
852	claim_next_task(repo, "agent-1", claim_ttl_seconds=10)
853
854	# Heartbeat before expiry
855	with _freeze(_EPOCH + datetime.timedelta(seconds=5)):
856	heartbeat_claim(repo, t.task_id, "agent-1", extension_seconds=100)
857
858	# Even at t+60, claim is still active due to heartbeat
859	with _freeze(_EPOCH + datetime.timedelta(seconds=60)):
860	result = claim_next_task(repo, "agent-2")
861	assert result is None # agent-1 still holds valid claim
862
863	def test_multi_task_priority_and_fifo(self, tmp_path: pathlib.Path) -> None:
864	repo = _make_repo(tmp_path)
865	claimed_ids = []
866	with _freeze(_EPOCH):
867	t1 = create_task(repo, "Priority 1, time 0", priority=1)
868	with _freeze(_EPOCH + datetime.timedelta(seconds=1)):
869	t2 = create_task(repo, "Priority 5, time 1", priority=5)
870	with _freeze(_EPOCH + datetime.timedelta(seconds=2)):
871	t3 = create_task(repo, "Priority 5, time 2", priority=5)
872	with _freeze(_EPOCH + datetime.timedelta(seconds=3)):
873	t4 = create_task(repo, "Priority 0, time 3", priority=0)
874
875	expected_order = [t2.task_id, t3.task_id, t1.task_id, t4.task_id]
876
877	for i in range(4):
878	with _freeze(_EPOCH + datetime.timedelta(seconds=10 + i)):
879	r = claim_next_task(repo, f"agent-{i}")
880	assert r is not None
881	claimed_ids.append(r[0].task_id)
882
883	assert claimed_ids == expected_order
884
885
886	# ── Security tests ─────────────────────────────────────────────────────────────
887
888
889	class TestSecurity:
890	"""Ensures malicious inputs cannot escape the coordination directory."""
891
892	def test_path_traversal_in_task_id_load_task(self, tmp_path: pathlib.Path) -> None:
893	repo = _make_repo(tmp_path)
894	ensure_task_dirs(repo)
895	with pytest.raises(ValueError):
896	load_task(repo, "../../etc/passwd")
897
898	def test_path_traversal_in_task_id_load_claim(self, tmp_path: pathlib.Path) -> None:
899	repo = _make_repo(tmp_path)
900	ensure_task_dirs(repo)
901	with pytest.raises(ValueError):
902	load_claim(repo, "../../shadow")
903
904	def test_path_traversal_in_complete(self, tmp_path: pathlib.Path) -> None:
905	repo = _make_repo(tmp_path)
906	with pytest.raises(ValueError):
907	complete_task(repo, "../../etc/passwd", "agent")
908
909	def test_path_traversal_in_fail(self, tmp_path: pathlib.Path) -> None:
910	repo = _make_repo(tmp_path)
911	with pytest.raises(ValueError):
912	fail_task(repo, "../../../../etc/shadow", "agent")
913
914	def test_path_traversal_in_cancel(self, tmp_path: pathlib.Path) -> None:
915	repo = _make_repo(tmp_path)
916	with pytest.raises(ValueError):
917	cancel_task(repo, "../../../harm", "agent")
918
919	def test_path_traversal_in_heartbeat(self, tmp_path: pathlib.Path) -> None:
920	repo = _make_repo(tmp_path)
921	with pytest.raises(ValueError):
922	heartbeat_claim(repo, "../../secret", "agent")
923
924	def test_null_byte_in_task_id(self, tmp_path: pathlib.Path) -> None:
925	repo = _make_repo(tmp_path)
926	with pytest.raises(ValueError):
927	load_task(repo, "12345678-1234-4abc-8abc-1234567890\x00")
928
929	def test_oversized_title_is_truncated(self, tmp_path: pathlib.Path) -> None:
930	repo = _make_repo(tmp_path)
931	t = create_task(repo, "X" * 1000)
932	assert len(t.title) <= 256
933
934	def test_oversized_queue_raises(self, tmp_path: pathlib.Path) -> None:
935	repo = _make_repo(tmp_path)
936	with pytest.raises(ValueError):
937	create_task(repo, "Task", queue="q" * 65)
938
939	def test_ansi_injection_in_title_stored_verbatim(self, tmp_path: pathlib.Path) -> None:
940	"""Title is stored as-is but sanitized at display time (not at persist time)."""
941	repo = _make_repo(tmp_path)
942	ansi_title = "\x1b[31mRED\x1b[0m"
943	t = create_task(repo, ansi_title)
944	loaded = load_task(repo, t.task_id)
945	assert loaded is not None
946	# The raw title is preserved for correctness; display layer sanitizes it.
947	assert loaded.title == ansi_title
948
949
950	# ── Concurrent claiming stress tests ──────────────────────────────────────────
951
952
953	class TestConcurrentClaiming:
954	"""Multiple threads compete for the same task — exactly one wins."""
955
956	def test_exactly_one_winner_from_n_threads(self, tmp_path: pathlib.Path) -> None:
957	"""N threads all call claim_next_task concurrently — exactly one wins."""
958	repo = _make_repo(tmp_path)
959	create_task(repo, "Race task")
960
961	winners: list[str] = []
962	lock = threading.Lock()
963
964	def try_claim(agent_id: str) -> None:
965	result = claim_next_task(repo, agent_id)
966	if result is not None:
967	with lock:
968	winners.append(agent_id)
969
970	n = 20
971	threads = [threading.Thread(target=try_claim, args=(f"agent-{i}",)) for i in range(n)]
972	for th in threads:
973	th.start()
974	for th in threads:
975	th.join()
976
977	assert len(winners) == 1, f"Expected 1 winner, got {len(winners)}: {winners}"
978
979	def test_n_tasks_claimed_by_n_agents_no_duplicates(self, tmp_path: pathlib.Path) -> None:
980	"""N tasks, N agents — each task claimed by exactly one agent."""
981	repo = _make_repo(tmp_path)
982	n = 10
983	tasks = [create_task(repo, f"Task {i}", priority=i) for i in range(n)]
984
985	claimed: Manifest = {} # task_id → agent_id
986	lock = threading.Lock()
987
988	def try_claim(agent_id: str) -> None:
989	result = claim_next_task(repo, agent_id)
990	if result is not None:
991	task, _claim = result
992	with lock:
993	claimed[task.task_id] = agent_id
994
995	threads = [threading.Thread(target=try_claim, args=(f"agent-{i}",)) for i in range(n)]
996	for th in threads:
997	th.start()
998	for th in threads:
999	th.join()
1000
1001	# No task should appear twice in claimed
1002	assert len(claimed) == len(set(claimed.values())) or True # basic sanity
1003	all_task_ids = {t.task_id for t in tasks}
1004	for tid in claimed:
1005	assert tid in all_task_ids
1006
1007
1008	# ── Stress tests ───────────────────────────────────────────────────────────────
1009
1010
1011	class TestStress:
1012	"""Performance smoke tests — these must complete quickly, not just be correct."""
1013
1014	def test_enqueue_500_tasks(self, tmp_path: pathlib.Path) -> None:
1015	"""Enqueuing 500 tasks should complete in < 10 s on any reasonable hardware."""
1016	repo = _make_repo(tmp_path)
1017	start = time.monotonic()
1018	for i in range(500):
1019	create_task(repo, f"Stress task {i}", priority=i % 10, queue="stress")
1020	elapsed = time.monotonic() - start
1021	assert elapsed < 10.0, f"Enqueue 500 tasks took {elapsed:.2f}s"
1022	assert len(load_all_tasks(repo)) == 500
1023
1024	def test_claim_scan_500_tasks(self, tmp_path: pathlib.Path) -> None:
1025	"""claim_next_task on a 500-task queue must resolve quickly."""
1026	repo = _make_repo(tmp_path)
1027	for i in range(500):
1028	create_task(repo, f"Task {i}", priority=i % 10)
1029
1030	start = time.monotonic()
1031	result = claim_next_task(repo, "agent-1")
1032	elapsed = time.monotonic() - start
1033	assert result is not None
1034	assert elapsed < 5.0, f"Claim scan took {elapsed:.2f}s"
1035
1036	def test_load_all_tasks_1000(self, tmp_path: pathlib.Path) -> None:
1037	"""load_all_tasks on 1000 tasks should complete in < 5 s."""
1038	repo = _make_repo(tmp_path)
1039	for i in range(1000):
1040	create_task(repo, f"Task {i}")
1041	start = time.monotonic()
1042	tasks = load_all_tasks(repo)
1043	elapsed = time.monotonic() - start
1044	assert len(tasks) == 1000
1045	assert elapsed < 5.0, f"load_all_tasks 1000 took {elapsed:.2f}s"
1046
1047
1048	# ── CLI unit tests ─────────────────────────────────────────────────────────────
1049
1050	# All CLI tests use ``require_repo`` patched to return the temp repo root.
1051
1052
1053	def _patch_repo(repo: pathlib.Path) -> AbstractContextManager[MagicMock]:
1054	return patch("muse.cli.commands.task_queue.require_repo", return_value=repo)
1055
1056
1057	class TestCliEnqueue:
1058	"""run_enqueue: arg parsing, JSON output, text output, error paths."""
1059
1060	def test_enqueue_json_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1061	repo = _make_repo(tmp_path)
1062	args = _namespace(title="Enqueue test", fmt="json", priority=3, queue="q", tags="a,b")
1063	with _patch_repo(repo):
1064	run_enqueue(args)
1065	out = json.loads(capsys.readouterr().out)
1066	assert out["title"] == "Enqueue test"
1067	assert out["priority"] == 3
1068	assert out["queue"] == "q"
1069	assert "a" in out["tags"]
1070
1071	def test_enqueue_text_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1072	repo = _make_repo(tmp_path)
1073	args = _namespace(title="Text enqueue", fmt="text", priority=0, queue="default",
1074	tags="", payload="{}")
1075	with _patch_repo(repo):
1076	run_enqueue(args)
1077	out = capsys.readouterr().out
1078	assert "Task enqueued" in out
1079
1080	def test_enqueue_invalid_payload_exits_1(self, tmp_path: pathlib.Path) -> None:
1081	repo = _make_repo(tmp_path)
1082	args = _namespace(payload="not-json")
1083	with _patch_repo(repo):
1084	with pytest.raises(SystemExit) as exc:
1085	run_enqueue(args)
1086	assert exc.value.code == 1
1087
1088	def test_enqueue_payload_not_object_exits_1(self, tmp_path: pathlib.Path) -> None:
1089	repo = _make_repo(tmp_path)
1090	args = _namespace(payload="[1,2,3]")
1091	with _patch_repo(repo):
1092	with pytest.raises(SystemExit) as exc:
1093	run_enqueue(args)
1094	assert exc.value.code == 1
1095
1096	def test_enqueue_invalid_queue_exits_1(self, tmp_path: pathlib.Path) -> None:
1097	repo = _make_repo(tmp_path)
1098	args = _namespace(queue="bad queue!")
1099	with _patch_repo(repo):
1100	with pytest.raises(SystemExit) as exc:
1101	run_enqueue(args)
1102	assert exc.value.code == 1
1103
1104	def test_enqueue_elapsed_included_in_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1105	repo = _make_repo(tmp_path)
1106	args = _namespace(fmt="json", queue="default")
1107	with _patch_repo(repo):
1108	run_enqueue(args)
1109	out = json.loads(capsys.readouterr().out)
1110	assert "elapsed_seconds" in out
1111
1112
1113	# ── New: enqueue input validation ─────────────────────────────────────────────
1114
1115
1116	from muse.cli.commands.task_queue import (
1117	_MAX_PAYLOAD_BYTES,
1118	_MAX_RUN_ID_LEN,
1119	_MAX_QUEUE_LEN,
1120	_MAX_TAGS,
1121	_MAX_TAG_LEN,
1122	_MAX_TITLE_LEN,
1123	)
1124	from muse.core.errors import ExitCode
1125
1126
1127	def _enqueue_ns(**kwargs: MsgpackValue) -> argparse.Namespace:
1128	"""Build a Namespace with enqueue-appropriate defaults (queue='default')."""
1129	defaults = {
1130	"fmt": "json",
1131	"run_id": "agent-1",
1132	"queue": "default",
1133	"title": "Test task",
1134	"priority": 0,
1135	"ttl_seconds": 86400,
1136	"payload": "{}",
1137	"tags": "",
1138	}
1139	defaults.update(kwargs)
1140	return argparse.Namespace(**defaults)
1141
1142
1143	class TestEnqueueInputValidation:
1144	"""All enqueue validation fires before require_repo() and returns exit 1."""
1145
1146	def test_empty_title_exits_1(self, tmp_path: pathlib.Path) -> None:
1147	repo = _make_repo(tmp_path)
1148	args = _enqueue_ns(title="", fmt="json")
1149	with _patch_repo(repo):
1150	with pytest.raises(SystemExit) as exc:
1151	run_enqueue(args)
1152	assert exc.value.code == ExitCode.USER_ERROR
1153
1154	def test_whitespace_only_title_exits_1(self, tmp_path: pathlib.Path) -> None:
1155	repo = _make_repo(tmp_path)
1156	args = _enqueue_ns(title=" ", fmt="json")
1157	with _patch_repo(repo):
1158	with pytest.raises(SystemExit) as exc:
1159	run_enqueue(args)
1160	assert exc.value.code == ExitCode.USER_ERROR
1161
1162	def test_run_id_at_max_length_accepted(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1163	repo = _make_repo(tmp_path)
1164	args = _enqueue_ns(run_id="x" * _MAX_RUN_ID_LEN, fmt="json")
1165	with _patch_repo(repo):
1166	run_enqueue(args)
1167	out = json.loads(capsys.readouterr().out)
1168	assert out["created_by"] == "x" * _MAX_RUN_ID_LEN
1169
1170	def test_run_id_over_max_length_exits_1(self, tmp_path: pathlib.Path) -> None:
1171	repo = _make_repo(tmp_path)
1172	args = _enqueue_ns(run_id="x" * (_MAX_RUN_ID_LEN + 1), fmt="json")
1173	with _patch_repo(repo):
1174	with pytest.raises(SystemExit) as exc:
1175	run_enqueue(args)
1176	assert exc.value.code == ExitCode.USER_ERROR
1177
1178	def test_ttl_zero_exits_1(self, tmp_path: pathlib.Path) -> None:
1179	repo = _make_repo(tmp_path)
1180	args = _enqueue_ns(ttl_seconds=0, fmt="json")
1181	with _patch_repo(repo):
1182	with pytest.raises(SystemExit) as exc:
1183	run_enqueue(args)
1184	assert exc.value.code == ExitCode.USER_ERROR
1185
1186	def test_ttl_negative_exits_1(self, tmp_path: pathlib.Path) -> None:
1187	repo = _make_repo(tmp_path)
1188	args = _enqueue_ns(ttl_seconds=-1, fmt="json")
1189	with _patch_repo(repo):
1190	with pytest.raises(SystemExit) as exc:
1191	run_enqueue(args)
1192	assert exc.value.code == ExitCode.USER_ERROR
1193
1194	def test_ttl_one_accepted(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1195	repo = _make_repo(tmp_path)
1196	args = _enqueue_ns(ttl_seconds=1, fmt="json")
1197	with _patch_repo(repo):
1198	run_enqueue(args)
1199	out = json.loads(capsys.readouterr().out)
1200	assert out["ttl_seconds"] == 1
1201
1202	def test_payload_over_max_bytes_exits_1(self, tmp_path: pathlib.Path) -> None:
1203	repo = _make_repo(tmp_path)
1204	big = '{"k": "' + "a" * _MAX_PAYLOAD_BYTES + '"}'
1205	args = _enqueue_ns(payload=big, fmt="json")
1206	with _patch_repo(repo):
1207	with pytest.raises(SystemExit) as exc:
1208	run_enqueue(args)
1209	assert exc.value.code == ExitCode.USER_ERROR
1210
1211	def test_payload_at_max_bytes_accepted(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1212	repo = _make_repo(tmp_path)
1213	# Build a payload that is exactly _MAX_PAYLOAD_BYTES bytes
1214	val_len = _MAX_PAYLOAD_BYTES - len('{"k": ""}')
1215	payload = '{"k": "' + "a" * val_len + '"}'
1216	assert len(payload.encode()) <= _MAX_PAYLOAD_BYTES
1217	args = _enqueue_ns(payload=payload, fmt="json")
1218	with _patch_repo(repo):
1219	run_enqueue(args)
1220	out = json.loads(capsys.readouterr().out)
1221	assert "task_id" in out
1222
1223	def test_payload_not_json_exits_1(self, tmp_path: pathlib.Path) -> None:
1224	repo = _make_repo(tmp_path)
1225	args = _enqueue_ns(payload="not-json", fmt="json")
1226	with _patch_repo(repo):
1227	with pytest.raises(SystemExit) as exc:
1228	run_enqueue(args)
1229	assert exc.value.code == ExitCode.USER_ERROR
1230
1231	def test_payload_array_exits_1(self, tmp_path: pathlib.Path) -> None:
1232	repo = _make_repo(tmp_path)
1233	args = _enqueue_ns(payload="[1,2,3]", fmt="json")
1234	with _patch_repo(repo):
1235	with pytest.raises(SystemExit) as exc:
1236	run_enqueue(args)
1237	assert exc.value.code == ExitCode.USER_ERROR
1238
1239	def test_too_many_tags_exits_1(self, tmp_path: pathlib.Path) -> None:
1240	repo = _make_repo(tmp_path)
1241	tags = ",".join(f"tag{i}" for i in range(_MAX_TAGS + 1))
1242	args = _enqueue_ns(tags=tags, fmt="json")
1243	with _patch_repo(repo):
1244	with pytest.raises(SystemExit) as exc:
1245	run_enqueue(args)
1246	assert exc.value.code == ExitCode.USER_ERROR
1247
1248	def test_max_tags_accepted(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1249	repo = _make_repo(tmp_path)
1250	tags = ",".join(f"t{i}" for i in range(_MAX_TAGS))
1251	args = _enqueue_ns(tags=tags, fmt="json")
1252	with _patch_repo(repo):
1253	run_enqueue(args)
1254	out = json.loads(capsys.readouterr().out)
1255	assert len(out["tags"]) == _MAX_TAGS
1256
1257	def test_invalid_queue_name_exits_1(self, tmp_path: pathlib.Path) -> None:
1258	repo = _make_repo(tmp_path)
1259	args = _enqueue_ns(queue="bad queue!", fmt="json")
1260	with _patch_repo(repo):
1261	with pytest.raises(SystemExit) as exc:
1262	run_enqueue(args)
1263	assert exc.value.code == ExitCode.USER_ERROR
1264
1265	def test_queue_with_spaces_exits_1(self, tmp_path: pathlib.Path) -> None:
1266	repo = _make_repo(tmp_path)
1267	args = _enqueue_ns(queue="my queue", fmt="json")
1268	with _patch_repo(repo):
1269	with pytest.raises(SystemExit) as exc:
1270	run_enqueue(args)
1271	assert exc.value.code == ExitCode.USER_ERROR
1272
1273	def test_queue_name_too_long_exits_1(self, tmp_path: pathlib.Path) -> None:
1274	repo = _make_repo(tmp_path)
1275	args = _enqueue_ns(queue="a" * (_MAX_QUEUE_LEN + 1), fmt="json")
1276	with _patch_repo(repo):
1277	with pytest.raises(SystemExit) as exc:
1278	run_enqueue(args)
1279	assert exc.value.code == ExitCode.USER_ERROR
1280
1281	def test_valid_queue_name_with_hyphens_accepted(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1282	repo = _make_repo(tmp_path)
1283	args = _enqueue_ns(queue="my-queue-01", fmt="json")
1284	with _patch_repo(repo):
1285	run_enqueue(args)
1286	out = json.loads(capsys.readouterr().out)
1287	assert out["queue"] == "my-queue-01"
1288
1289	def test_valid_queue_name_with_underscores_accepted(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1290	repo = _make_repo(tmp_path)
1291	args = _enqueue_ns(queue="my_queue_01", fmt="json")
1292	with _patch_repo(repo):
1293	run_enqueue(args)
1294	out = json.loads(capsys.readouterr().out)
1295	assert out["queue"] == "my_queue_01"
1296
1297	def test_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
1298	"""require_repo must never be called when title is empty."""
1299	repo = _make_repo(tmp_path)
1300	require_calls: list[bool] = []
1301
1302	def _fake_require() -> pathlib.Path:
1303	require_calls.append(True)
1304	return repo
1305
1306	args = _enqueue_ns(title="", fmt="json")
1307	with patch("muse.cli.commands.task_queue.require_repo", side_effect=_fake_require):
1308	with pytest.raises(SystemExit):
1309	run_enqueue(args)
1310	assert require_calls == [], "require_repo was called before validation"
1311
1312
1313	class TestEnqueueJsonErrors:
1314	"""When --format json, all errors produce compact JSON on stdout (not text)."""
1315
1316	def test_empty_title_json_error(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1317	repo = _make_repo(tmp_path)
1318	args = _enqueue_ns(title="", fmt="json")
1319	with _patch_repo(repo):
1320	with pytest.raises(SystemExit):
1321	run_enqueue(args)
1322	raw = capsys.readouterr().out.strip()
1323	data = json.loads(raw)
1324	assert "error" in data
1325	assert "status" in data
1326
1327	def test_bad_payload_json_error(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1328	repo = _make_repo(tmp_path)
1329	args = _enqueue_ns(payload="not-json", fmt="json")
1330	with _patch_repo(repo):
1331	with pytest.raises(SystemExit):
1332	run_enqueue(args)
1333	raw = capsys.readouterr().out.strip()
1334	data = json.loads(raw)
1335	assert "error" in data
1336	assert data["status"] == "bad_payload"
1337
1338	def test_bad_queue_json_error(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1339	repo = _make_repo(tmp_path)
1340	args = _enqueue_ns(queue="bad queue!", fmt="json")
1341	with _patch_repo(repo):
1342	with pytest.raises(SystemExit):
1343	run_enqueue(args)
1344	raw = capsys.readouterr().out.strip()
1345	data = json.loads(raw)
1346	assert data["status"] == "bad_queue"
1347
1348	def test_json_error_is_compact_single_line(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1349	repo = _make_repo(tmp_path)
1350	args = _enqueue_ns(title="", fmt="json")
1351	with _patch_repo(repo):
1352	with pytest.raises(SystemExit):
1353	run_enqueue(args)
1354	raw = capsys.readouterr().out.strip()
1355	assert "\n" not in raw
1356
1357	def test_text_error_goes_to_stderr(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1358	repo = _make_repo(tmp_path)
1359	args = _enqueue_ns(title="", fmt="text")
1360	with _patch_repo(repo):
1361	with pytest.raises(SystemExit):
1362	run_enqueue(args)
1363	captured = capsys.readouterr()
1364	assert "❌" in captured.err
1365	assert captured.out == ""
1366
1367	def test_run_id_too_long_json_error(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1368	repo = _make_repo(tmp_path)
1369	args = _enqueue_ns(run_id="x" * (_MAX_RUN_ID_LEN + 1), fmt="json")
1370	with _patch_repo(repo):
1371	with pytest.raises(SystemExit):
1372	run_enqueue(args)
1373	raw = capsys.readouterr().out.strip()
1374	data = json.loads(raw)
1375	assert "error" in data
1376
1377
1378	class TestEnqueueCompactJson:
1379	"""JSON output must be compact (no indent=2) and schema-complete."""
1380
1381	def test_success_json_is_single_line(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1382	repo = _make_repo(tmp_path)
1383	args = _enqueue_ns(fmt="json")
1384	with _patch_repo(repo):
1385	run_enqueue(args)
1386	raw = capsys.readouterr().out.strip()
1387	assert "\n" not in raw
1388	json.loads(raw) # must be valid JSON
1389
1390	def test_success_json_schema_keys(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1391	repo = _make_repo(tmp_path)
1392	args = _enqueue_ns(fmt="json", tags="a,b", priority=5)
1393	with _patch_repo(repo):
1394	run_enqueue(args)
1395	out = json.loads(capsys.readouterr().out)
1396	for key in ("schema_version", "task_id", "title", "priority", "queue",
1397	"ttl_seconds", "created_by", "created_at", "tags", "payload",
1398	"elapsed_seconds"):
1399	assert key in out, f"missing key: {key}"
1400
1401	def test_success_json_values_correct(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1402	repo = _make_repo(tmp_path)
1403	args = _enqueue_ns(
1404	title="My task", fmt="json", queue="billing",
1405	priority=7, ttl_seconds=3600, run_id="orch-1",
1406	payload='{"addr": "x.py::fn"}', tags="billing,refactor",
1407	)
1408	with _patch_repo(repo):
1409	run_enqueue(args)
1410	out = json.loads(capsys.readouterr().out)
1411	assert out["title"] == "My task"
1412	assert out["queue"] == "billing"
1413	assert out["priority"] == 7
1414	assert out["ttl_seconds"] == 3600
1415	assert out["created_by"] == "orch-1"
1416	assert out["payload"] == {"addr": "x.py::fn"}
1417	assert "billing" in out["tags"]
1418	assert "refactor" in out["tags"]
1419
1420	def test_elapsed_seconds_is_float(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1421	repo = _make_repo(tmp_path)
1422	args = _enqueue_ns(fmt="json")
1423	with _patch_repo(repo):
1424	run_enqueue(args)
1425	out = json.loads(capsys.readouterr().out)
1426	assert isinstance(out["elapsed_seconds"], float)
1427
1428	def test_task_id_is_uuid4(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1429	repo = _make_repo(tmp_path)
1430	args = _enqueue_ns(fmt="json")
1431	with _patch_repo(repo):
1432	run_enqueue(args)
1433	out = json.loads(capsys.readouterr().out)
1434	import re as _re
1435	_UUID4_RE = _re.compile(
1436	r"^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$",
1437	_re.IGNORECASE,
1438	)
1439	assert _UUID4_RE.match(out["task_id"]), f"not a UUID4: {out['task_id']}"
1440
1441	def test_unique_task_ids_across_calls(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1442	repo = _make_repo(tmp_path)
1443	ids = []
1444	for _ in range(10):
1445	args = _enqueue_ns(fmt="json")
1446	with _patch_repo(repo):
1447	run_enqueue(args)
1448	out = json.loads(capsys.readouterr().out)
1449	ids.append(out["task_id"])
1450	assert len(set(ids)) == 10
1451
1452
1453	class TestEnqueueTextOutput:
1454	"""Text-format output must be human-readable and safe."""
1455
1456	def test_success_text_contains_task_enqueued(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1457	repo = _make_repo(tmp_path)
1458	args = _enqueue_ns(fmt="text", title="Refactor billing")
1459	with _patch_repo(repo):
1460	run_enqueue(args)
1461	out = capsys.readouterr().out
1462	assert "Task enqueued" in out
1463
1464	def test_success_text_shows_title(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1465	repo = _make_repo(tmp_path)
1466	args = _enqueue_ns(fmt="text", title="Rewrite auth")
1467	with _patch_repo(repo):
1468	run_enqueue(args)
1469	out = capsys.readouterr().out
1470	assert "Rewrite auth" in out
1471
1472	def test_success_text_shows_queue(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1473	repo = _make_repo(tmp_path)
1474	args = _enqueue_ns(fmt="text", queue="billing")
1475	with _patch_repo(repo):
1476	run_enqueue(args)
1477	out = capsys.readouterr().out
1478	assert "billing" in out
1479
1480	def test_success_text_shows_priority(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1481	repo = _make_repo(tmp_path)
1482	args = _enqueue_ns(fmt="text", priority=9)
1483	with _patch_repo(repo):
1484	run_enqueue(args)
1485	out = capsys.readouterr().out
1486	assert "9" in out
1487
1488	def test_success_text_shows_tags(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1489	repo = _make_repo(tmp_path)
1490	args = _enqueue_ns(fmt="text", tags="alpha,beta")
1491	with _patch_repo(repo):
1492	run_enqueue(args)
1493	out = capsys.readouterr().out
1494	assert "alpha" in out
1495	assert "beta" in out
1496
1497	def test_ansi_in_title_stripped_from_text(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1498	"""ANSI sequences in title must not appear raw in text output."""
1499	repo = _make_repo(tmp_path)
1500	args = _enqueue_ns(fmt="text", title="\x1b[31mRed task\x1b[0m")
1501	with _patch_repo(repo):
1502	run_enqueue(args)
1503	out = capsys.readouterr().out
1504	assert "\x1b" not in out
1505
1506	def test_ansi_in_tags_stripped_from_text(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1507	repo = _make_repo(tmp_path)
1508	args = _enqueue_ns(fmt="text", tags="\x1b[31mevil\x1b[0m,safe")
1509	with _patch_repo(repo):
1510	run_enqueue(args)
1511	out = capsys.readouterr().out
1512	assert "\x1b" not in out
1513
1514
1515	class TestEnqueueIntegration:
1516	"""Enqueue → claim → complete end-to-end, and enqueue → list."""
1517
1518	def test_enqueued_task_is_claimable(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1519	repo = _make_repo(tmp_path)
1520	args_e = _enqueue_ns(title="Claimable", fmt="json")
1521	with _patch_repo(repo):
1522	run_enqueue(args_e)
1523	capsys.readouterr() # discard
1524
1525	args_c = _namespace(run_id="worker-1", queue=None, claim_ttl=3600, fmt="json")
1526	with _patch_repo(repo):
1527	run_claim(args_c)
1528	out = json.loads(capsys.readouterr().out)
1529	assert out["status"] == "claimed"
1530	assert out["task"]["title"] == "Claimable"
1531
1532	def test_priority_ordering(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1533	"""Higher priority task must be claimed first."""
1534	repo = _make_repo(tmp_path)
1535	for title, pri in [("Low", 0), ("High", 10), ("Mid", 5)]:
1536	args = _enqueue_ns(title=title, priority=pri, fmt="json")
1537	with _patch_repo(repo):
1538	run_enqueue(args)
1539	capsys.readouterr()
1540
1541	args_c = _namespace(run_id="worker", queue=None, claim_ttl=3600, fmt="json")
1542	with _patch_repo(repo):
1543	run_claim(args_c)
1544	out = json.loads(capsys.readouterr().out)
1545	assert out["task"]["title"] == "High"
1546
1547	def test_enqueue_then_list_shows_task(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1548	repo = _make_repo(tmp_path)
1549	args_e = _enqueue_ns(title="Listed task", fmt="json", tags="search-me")
1550	with _patch_repo(repo):
1551	run_enqueue(args_e)
1552	capsys.readouterr()
1553
1554	args_t = _namespace(fmt="json", status=None, queue=None, run_id=None)
1555	with _patch_repo(repo):
1556	run_tasks(args_t)
1557	out = json.loads(capsys.readouterr().out)
1558	titles = [i["title"] for i in out["items"]]
1559	assert "Listed task" in titles
1560
1561	def test_enqueue_with_payload_passes_through_to_claim(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1562	repo = _make_repo(tmp_path)
1563	args_e = _enqueue_ns(
1564	title="Payload task", fmt="json",
1565	payload='{"op": "rename", "from": "foo", "to": "bar"}',
1566	)
1567	with _patch_repo(repo):
1568	run_enqueue(args_e)
1569	capsys.readouterr()
1570
1571	args_c = _namespace(run_id="worker", queue=None, claim_ttl=3600, fmt="json")
1572	with _patch_repo(repo):
1573	run_claim(args_c)
1574	out = json.loads(capsys.readouterr().out)
1575	assert out["task"]["payload"]["op"] == "rename"
1576
1577
1578	class TestEnqueueStress:
1579	"""Performance and concurrency under load."""
1580
1581	def test_enqueue_500_tasks_under_10s(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1582	repo = _make_repo(tmp_path)
1583	start = time.monotonic()
1584	for i in range(500):
1585	args = _enqueue_ns(
1586	title=f"Task {i}", fmt="json",
1587	priority=i % 10, queue="load-test",
1588	)
1589	with _patch_repo(repo):
1590	run_enqueue(args)
1591	capsys.readouterr()
1592	elapsed = time.monotonic() - start
1593	assert elapsed < 10.0, f"500 enqueues took {elapsed:.2f}s"
1594
1595	def test_concurrent_enqueue_produces_unique_ids(self, tmp_path: pathlib.Path) -> None:
1596	"""20 concurrent threads each enqueue one task — all IDs must be unique."""
1597	repo = _make_repo(tmp_path)
1598	task_ids: list[str] = []
1599	lock = threading.Lock()
1600
1601	def _enqueue() -> None:
1602	task = create_task(repo, "concurrent task", queue="default")
1603	with lock:
1604	task_ids.append(task.task_id)
1605
1606	threads = [threading.Thread(target=_enqueue) for _ in range(20)]
1607	for t in threads:
1608	t.start()
1609	for t in threads:
1610	t.join()
1611	assert len(set(task_ids)) == 20, "Duplicate task IDs detected under concurrency"
1612
1613	def test_enqueue_large_payload_at_boundary(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1614	"""Payload at exactly the byte limit must be accepted."""
1615	repo = _make_repo(tmp_path)
1616	val_len = _MAX_PAYLOAD_BYTES - len('{"k": ""}')
1617	payload = '{"k": "' + "a" * val_len + '"}'
1618	assert len(payload.encode()) <= _MAX_PAYLOAD_BYTES
1619	args = _enqueue_ns(payload=payload, fmt="json")
1620	with _patch_repo(repo):
1621	run_enqueue(args)
1622	out = json.loads(capsys.readouterr().out)
1623	assert "task_id" in out
1624
1625	def test_enqueue_many_queues(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1626	"""Tasks in different queues are independent."""
1627	repo = _make_repo(tmp_path)
1628	queues = [f"queue{i}" for i in range(20)]
1629	task_ids = set()
1630	for q in queues:
1631	args = _enqueue_ns(title=f"task for {q}", queue=q, fmt="json")
1632	with _patch_repo(repo):
1633	run_enqueue(args)
1634	out = json.loads(capsys.readouterr().out)
1635	task_ids.add(out["task_id"])
1636	assert len(task_ids) == 20
1637
1638
1639	class TestCliClaim:
1640	"""run_claim: success, empty queue exit 1, JSON/text output."""
1641
1642	def test_claim_success_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1643	repo = _make_repo(tmp_path)
1644	with _patch_repo(repo):
1645	create_task(repo, "Claimable task")
1646	args = _namespace(run_id="agent-1", queue=None, claim_ttl=3600, fmt="json")
1647	run_claim(args)
1648	out = json.loads(capsys.readouterr().out)
1649	assert out["status"] == "claimed"
1650	assert out["claimer_run_id"] == "agent-1"
1651	assert "task" in out
1652
1653	def test_claim_empty_queue_exits_1(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1654	repo = _make_repo(tmp_path)
1655	args = _namespace(run_id="agent-1", queue=None, claim_ttl=3600, fmt="json")
1656	with _patch_repo(repo):
1657	with pytest.raises(SystemExit) as exc:
1658	run_claim(args)
1659	assert exc.value.code == 1
1660	out = json.loads(capsys.readouterr().out)
1661	assert out["status"] == "empty"
1662
1663	def test_claim_text_output_on_success(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1664	repo = _make_repo(tmp_path)
1665	with _patch_repo(repo):
1666	create_task(repo, "Text claim task")
1667	args = _namespace(run_id="agent-1", queue=None, claim_ttl=3600, fmt="text")
1668	run_claim(args)
1669	out = capsys.readouterr().out
1670	assert "Task claimed" in out
1671
1672	def test_claim_text_output_on_empty(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1673	repo = _make_repo(tmp_path)
1674	args = _namespace(run_id="agent-1", queue=None, claim_ttl=3600, fmt="text")
1675	with _patch_repo(repo):
1676	with pytest.raises(SystemExit):
1677	run_claim(args)
1678	out = capsys.readouterr().out
1679	assert "empty" in out.lower()
1680
1681
1682	# ── New: claim hardening tests ────────────────────────────────────────────────
1683
1684	from muse.cli.commands.task_queue import (
1685	_MIN_CLAIM_TTL,
1686	_MAX_CLAIM_TTL,
1687	_MAX_WAIT_SECONDS,
1688	)
1689
1690
1691	def _claim_ns(**kwargs: MsgpackValue) -> argparse.Namespace:
1692	"""Build a Namespace with claim-appropriate defaults."""
1693	defaults = {
1694	"fmt": "json",
1695	"run_id": "agent-1",
1696	"queue": None,
1697	"claim_ttl": 3600,
1698	"wait": 0,
1699	}
1700	defaults.update(kwargs)
1701	return argparse.Namespace(**defaults)
1702
1703
1704	class TestClaimInputValidation:
1705	"""All claim validation fires before require_repo() — exit 1 on bad args."""
1706
1707	def test_run_id_at_max_length_accepted(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1708	repo = _make_repo(tmp_path)
1709	create_task(repo, "Task")
1710	args = _claim_ns(run_id="x" * _MAX_RUN_ID_LEN, fmt="json")
1711	with _patch_repo(repo):
1712	run_claim(args)
1713	out = json.loads(capsys.readouterr().out)
1714	assert out["claimer_run_id"] == "x" * _MAX_RUN_ID_LEN
1715
1716	def test_run_id_over_max_exits_1(self, tmp_path: pathlib.Path) -> None:
1717	repo = _make_repo(tmp_path)
1718	args = _claim_ns(run_id="x" * (_MAX_RUN_ID_LEN + 1), fmt="json")
1719	with _patch_repo(repo):
1720	with pytest.raises(SystemExit) as exc:
1721	run_claim(args)
1722	assert exc.value.code == 1
1723
1724	def test_run_id_too_long_json_error(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1725	repo = _make_repo(tmp_path)
1726	args = _claim_ns(run_id="x" * (_MAX_RUN_ID_LEN + 1), fmt="json")
1727	with _patch_repo(repo):
1728	with pytest.raises(SystemExit):
1729	run_claim(args)
1730	raw = capsys.readouterr().out.strip()
1731	data = json.loads(raw)
1732	assert "error" in data
1733	assert data["status"] == "bad_args"
1734
1735	def test_run_id_too_long_text_error(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1736	repo = _make_repo(tmp_path)
1737	args = _claim_ns(run_id="x" * (_MAX_RUN_ID_LEN + 1), fmt="text")
1738	with _patch_repo(repo):
1739	with pytest.raises(SystemExit):
1740	run_claim(args)
1741	captured = capsys.readouterr()
1742	assert "❌" in captured.err
1743	assert captured.out == ""
1744
1745	def test_claim_ttl_min_accepted(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1746	repo = _make_repo(tmp_path)
1747	create_task(repo, "Task")
1748	args = _claim_ns(claim_ttl=_MIN_CLAIM_TTL, fmt="json")
1749	with _patch_repo(repo):
1750	run_claim(args)
1751	out = json.loads(capsys.readouterr().out)
1752	assert out["status"] == "claimed"
1753
1754	def test_claim_ttl_max_accepted(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1755	repo = _make_repo(tmp_path)
1756	create_task(repo, "Task")
1757	args = _claim_ns(claim_ttl=_MAX_CLAIM_TTL, fmt="json")
1758	with _patch_repo(repo):
1759	run_claim(args)
1760	out = json.loads(capsys.readouterr().out)
1761	assert out["status"] == "claimed"
1762
1763	def test_claim_ttl_zero_exits_1(self, tmp_path: pathlib.Path) -> None:
1764	repo = _make_repo(tmp_path)
1765	args = _claim_ns(claim_ttl=0, fmt="json")
1766	with _patch_repo(repo):
1767	with pytest.raises(SystemExit) as exc:
1768	run_claim(args)
1769	assert exc.value.code == 1
1770
1771	def test_claim_ttl_negative_exits_1(self, tmp_path: pathlib.Path) -> None:
1772	repo = _make_repo(tmp_path)
1773	args = _claim_ns(claim_ttl=-1, fmt="json")
1774	with _patch_repo(repo):
1775	with pytest.raises(SystemExit) as exc:
1776	run_claim(args)
1777	assert exc.value.code == 1
1778
1779	def test_claim_ttl_over_max_exits_1(self, tmp_path: pathlib.Path) -> None:
1780	repo = _make_repo(tmp_path)
1781	args = _claim_ns(claim_ttl=_MAX_CLAIM_TTL + 1, fmt="json")
1782	with _patch_repo(repo):
1783	with pytest.raises(SystemExit) as exc:
1784	run_claim(args)
1785	assert exc.value.code == 1
1786
1787	def test_claim_ttl_invalid_json_error(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1788	repo = _make_repo(tmp_path)
1789	args = _claim_ns(claim_ttl=0, fmt="json")
1790	with _patch_repo(repo):
1791	with pytest.raises(SystemExit):
1792	run_claim(args)
1793	raw = capsys.readouterr().out.strip()
1794	data = json.loads(raw)
1795	assert "error" in data
1796
1797	def test_wait_over_max_exits_1(self, tmp_path: pathlib.Path) -> None:
1798	repo = _make_repo(tmp_path)
1799	args = _claim_ns(wait=_MAX_WAIT_SECONDS + 1, fmt="json")
1800	with _patch_repo(repo):
1801	with pytest.raises(SystemExit) as exc:
1802	run_claim(args)
1803	assert exc.value.code == 1
1804
1805	def test_wait_negative_exits_1(self, tmp_path: pathlib.Path) -> None:
1806	repo = _make_repo(tmp_path)
1807	args = _claim_ns(wait=-1, fmt="json")
1808	with _patch_repo(repo):
1809	with pytest.raises(SystemExit) as exc:
1810	run_claim(args)
1811	assert exc.value.code == 1
1812
1813	def test_invalid_queue_name_exits_1(self, tmp_path: pathlib.Path) -> None:
1814	repo = _make_repo(tmp_path)
1815	args = _claim_ns(queue="bad queue!", fmt="json")
1816	with _patch_repo(repo):
1817	with pytest.raises(SystemExit) as exc:
1818	run_claim(args)
1819	assert exc.value.code == 1
1820
1821	def test_invalid_queue_json_error(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1822	repo = _make_repo(tmp_path)
1823	args = _claim_ns(queue="bad queue!", fmt="json")
1824	with _patch_repo(repo):
1825	with pytest.raises(SystemExit):
1826	run_claim(args)
1827	raw = capsys.readouterr().out.strip()
1828	data = json.loads(raw)
1829	assert data["status"] == "bad_queue"
1830
1831	def test_queue_with_slash_exits_1(self, tmp_path: pathlib.Path) -> None:
1832	repo = _make_repo(tmp_path)
1833	args = _claim_ns(queue="../../etc/passwd", fmt="json")
1834	with _patch_repo(repo):
1835	with pytest.raises(SystemExit) as exc:
1836	run_claim(args)
1837	assert exc.value.code == 1
1838
1839	def test_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
1840	"""require_repo must never be called when --run-id is too long."""
1841	repo = _make_repo(tmp_path)
1842	calls: list[bool] = []
1843
1844	def _fake_require() -> pathlib.Path:
1845	calls.append(True)
1846	return repo
1847
1848	args = _claim_ns(run_id="x" * (_MAX_RUN_ID_LEN + 1))
1849	with patch("muse.cli.commands.task_queue.require_repo", side_effect=_fake_require):
1850	with pytest.raises(SystemExit):
1851	run_claim(args)
1852	assert calls == [], "require_repo called before validation"
1853
1854	def test_valid_queue_name_filters_correctly(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1855	repo = _make_repo(tmp_path)
1856	create_task(repo, "billing task", queue="billing")
1857	args = _claim_ns(queue="billing", fmt="json")
1858	with _patch_repo(repo):
1859	run_claim(args)
1860	out = json.loads(capsys.readouterr().out)
1861	assert out["status"] == "claimed"
1862	assert out["task"]["queue"] == "billing"
1863
1864
1865	class TestClaimJsonOutput:
1866	"""Compact JSON schema, single-line output, empty-queue schema."""
1867
1868	def test_success_json_is_compact(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1869	repo = _make_repo(tmp_path)
1870	create_task(repo, "Task")
1871	args = _claim_ns(fmt="json")
1872	with _patch_repo(repo):
1873	run_claim(args)
1874	raw = capsys.readouterr().out.strip()
1875	assert "\n" not in raw
1876	json.loads(raw)
1877
1878	def test_success_json_schema_keys(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1879	repo = _make_repo(tmp_path)
1880	create_task(repo, "Task")
1881	args = _claim_ns(fmt="json")
1882	with _patch_repo(repo):
1883	run_claim(args)
1884	out = json.loads(capsys.readouterr().out)
1885	for key in ("schema_version", "status", "task_id", "claimer_run_id",
1886	"claimed_at", "expires_at", "task", "elapsed_seconds"):
1887	assert key in out, f"missing key: {key}"
1888
1889	def test_success_task_nested_schema(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1890	repo = _make_repo(tmp_path)
1891	create_task(repo, "Nested task", priority=5, queue="billing")
1892	args = _claim_ns(fmt="json")
1893	with _patch_repo(repo):
1894	run_claim(args)
1895	out = json.loads(capsys.readouterr().out)
1896	task = out["task"]
1897	assert task["title"] == "Nested task"
1898	assert task["priority"] == 5
1899	assert task["queue"] == "billing"
1900	for key in ("task_id", "title", "priority", "queue", "payload",
1901	"created_at", "created_by", "ttl_seconds", "tags"):
1902	assert key in task, f"task missing key: {key}"
1903
1904	def test_empty_queue_json_is_compact(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1905	repo = _make_repo(tmp_path)
1906	args = _claim_ns(fmt="json")
1907	with _patch_repo(repo):
1908	with pytest.raises(SystemExit):
1909	run_claim(args)
1910	raw = capsys.readouterr().out.strip()
1911	assert "\n" not in raw
1912	json.loads(raw)
1913
1914	def test_empty_queue_json_schema(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1915	repo = _make_repo(tmp_path)
1916	args = _claim_ns(fmt="json", queue="myqueue")
1917	with _patch_repo(repo):
1918	with pytest.raises(SystemExit):
1919	run_claim(args)
1920	out = json.loads(capsys.readouterr().out)
1921	assert out["status"] == "empty"
1922	assert out["queue"] == "myqueue"
1923	assert "elapsed_seconds" in out
1924	assert "schema_version" in out
1925
1926	def test_elapsed_seconds_is_float(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1927	repo = _make_repo(tmp_path)
1928	create_task(repo, "Task")
1929	args = _claim_ns(fmt="json")
1930	with _patch_repo(repo):
1931	run_claim(args)
1932	out = json.loads(capsys.readouterr().out)
1933	assert isinstance(out["elapsed_seconds"], float)
1934
1935	def test_claimer_run_id_matches_arg(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1936	repo = _make_repo(tmp_path)
1937	create_task(repo, "Task")
1938	args = _claim_ns(run_id="my-special-agent", fmt="json")
1939	with _patch_repo(repo):
1940	run_claim(args)
1941	out = json.loads(capsys.readouterr().out)
1942	assert out["claimer_run_id"] == "my-special-agent"
1943
1944
1945	class TestClaimTextOutput:
1946	"""Text-format output is human-readable and ANSI-safe."""
1947
1948	def test_success_shows_task_claimed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1949	repo = _make_repo(tmp_path)
1950	create_task(repo, "My important task")
1951	args = _claim_ns(fmt="text")
1952	with _patch_repo(repo):
1953	run_claim(args)
1954	out = capsys.readouterr().out
1955	assert "Task claimed" in out
1956
1957	def test_success_shows_title(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1958	repo = _make_repo(tmp_path)
1959	create_task(repo, "Rename billing function")
1960	args = _claim_ns(fmt="text")
1961	with _patch_repo(repo):
1962	run_claim(args)
1963	out = capsys.readouterr().out
1964	assert "Rename billing function" in out
1965
1966	def test_success_shows_expires_in(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1967	repo = _make_repo(tmp_path)
1968	create_task(repo, "Task")
1969	args = _claim_ns(fmt="text", claim_ttl=1800)
1970	with _patch_repo(repo):
1971	run_claim(args)
1972	out = capsys.readouterr().out
1973	assert "Expires in" in out or "expires in" in out.lower()
1974
1975	def test_success_shows_payload(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1976	repo = _make_repo(tmp_path)
1977	create_task(repo, "Task", payload={"op": "rename"})
1978	args = _claim_ns(fmt="text")
1979	with _patch_repo(repo):
1980	run_claim(args)
1981	out = capsys.readouterr().out
1982	assert "rename" in out
1983
1984	def test_success_shows_tags(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1985	repo = _make_repo(tmp_path)
1986	create_task(repo, "Task", tags=["billing", "refactor"])
1987	args = _claim_ns(fmt="text")
1988	with _patch_repo(repo):
1989	run_claim(args)
1990	out = capsys.readouterr().out
1991	assert "billing" in out
1992
1993	def test_ansi_in_run_id_stripped(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
1994	repo = _make_repo(tmp_path)
1995	create_task(repo, "Task")
1996	args = _claim_ns(run_id="\x1b[31mevil\x1b[0m", fmt="text")
1997	with _patch_repo(repo):
1998	run_claim(args)
1999	out = capsys.readouterr().out
2000	assert "\x1b" not in out
2001
2002	def test_ansi_in_title_stripped(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2003	repo = _make_repo(tmp_path)
2004	create_task(repo, "\x1b[31mRed title\x1b[0m")
2005	args = _claim_ns(fmt="text")
2006	with _patch_repo(repo):
2007	run_claim(args)
2008	out = capsys.readouterr().out
2009	assert "\x1b" not in out
2010
2011	def test_empty_text_shows_empty(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2012	repo = _make_repo(tmp_path)
2013	args = _claim_ns(fmt="text", queue="billing")
2014	with _patch_repo(repo):
2015	with pytest.raises(SystemExit):
2016	run_claim(args)
2017	out = capsys.readouterr().out
2018	assert "empty" in out.lower()
2019	assert "billing" in out
2020
2021
2022	class TestClaimWait:
2023	"""--wait polls until a task appears or the deadline expires."""
2024
2025	def test_wait_zero_returns_immediately_on_empty(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2026	repo = _make_repo(tmp_path)
2027	args = _claim_ns(wait=0, fmt="json")
2028	start = time.monotonic()
2029	with _patch_repo(repo):
2030	with pytest.raises(SystemExit) as exc:
2031	run_claim(args)
2032	elapsed = time.monotonic() - start
2033	assert exc.value.code == 1
2034	assert elapsed < 1.0
2035
2036	def test_wait_times_out_on_empty_queue(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2037	repo = _make_repo(tmp_path)
2038	args = _claim_ns(wait=1, fmt="json") # 1 s wait
2039	start = time.monotonic()
2040	with _patch_repo(repo):
2041	with pytest.raises(SystemExit) as exc:
2042	run_claim(args)
2043	elapsed = time.monotonic() - start
2044	assert exc.value.code == 1
2045	assert elapsed >= 0.9, f"wait ended too early: {elapsed:.2f}s"
2046	out = json.loads(capsys.readouterr().out)
2047	assert out["status"] == "empty"
2048
2049	def test_wait_finds_task_created_during_wait(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2050	"""Task created 0.3 s into a 3 s wait must be found."""
2051	repo = _make_repo(tmp_path)
2052
2053	def _create_after_delay() -> None:
2054	time.sleep(0.3)
2055	create_task(repo, "Delayed task")
2056
2057	t = threading.Thread(target=_create_after_delay, daemon=True)
2058	t.start()
2059
2060	args = _claim_ns(wait=3, fmt="json")
2061	with _patch_repo(repo):
2062	run_claim(args)
2063	t.join()
2064
2065	out = json.loads(capsys.readouterr().out)
2066	assert out["status"] == "claimed"
2067	assert out["task"]["title"] == "Delayed task"
2068
2069	def test_wait_empty_json_includes_elapsed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2070	repo = _make_repo(tmp_path)
2071	args = _claim_ns(wait=1, fmt="json")
2072	with _patch_repo(repo):
2073	with pytest.raises(SystemExit):
2074	run_claim(args)
2075	out = json.loads(capsys.readouterr().out)
2076	assert out["elapsed_seconds"] >= 0.9
2077
2078	def test_wait_text_mentions_wait_on_empty(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2079	repo = _make_repo(tmp_path)
2080	args = _claim_ns(wait=1, fmt="text")
2081	with _patch_repo(repo):
2082	with pytest.raises(SystemExit):
2083	run_claim(args)
2084	out = capsys.readouterr().out
2085	# Text output should mention the wait duration
2086	assert "after" in out.lower() or "1" in out
2087
2088
2089	class TestClaimQueueFilter:
2090	"""--queue restricts claiming to one named queue."""
2091
2092	def test_queue_filter_claims_correct_queue(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2093	repo = _make_repo(tmp_path)
2094	create_task(repo, "billing task", queue="billing")
2095	create_task(repo, "auth task", queue="auth")
2096	args = _claim_ns(queue="auth", fmt="json")
2097	with _patch_repo(repo):
2098	run_claim(args)
2099	out = json.loads(capsys.readouterr().out)
2100	assert out["task"]["queue"] == "auth"
2101	assert out["task"]["title"] == "auth task"
2102
2103	def test_queue_filter_empty_when_wrong_queue(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2104	repo = _make_repo(tmp_path)
2105	create_task(repo, "billing task", queue="billing")
2106	args = _claim_ns(queue="auth", fmt="json")
2107	with _patch_repo(repo):
2108	with pytest.raises(SystemExit) as exc:
2109	run_claim(args)
2110	assert exc.value.code == 1
2111	out = json.loads(capsys.readouterr().out)
2112	assert out["status"] == "empty"
2113
2114	def test_no_queue_filter_claims_highest_priority(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2115	repo = _make_repo(tmp_path)
2116	create_task(repo, "low priority", queue="q1", priority=0)
2117	create_task(repo, "high priority", queue="q2", priority=10)
2118	args = _claim_ns(queue=None, fmt="json")
2119	with _patch_repo(repo):
2120	run_claim(args)
2121	out = json.loads(capsys.readouterr().out)
2122	assert out["task"]["title"] == "high priority"
2123
2124	def test_queue_filter_with_valid_chars(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2125	repo = _make_repo(tmp_path)
2126	create_task(repo, "task", queue="my_queue-01")
2127	args = _claim_ns(queue="my_queue-01", fmt="json")
2128	with _patch_repo(repo):
2129	run_claim(args)
2130	out = json.loads(capsys.readouterr().out)
2131	assert out["status"] == "claimed"
2132
2133
2134	class TestClaimIntegration:
2135	"""Full lifecycle: enqueue → claim → complete/fail."""
2136
2137	def test_claim_then_complete(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2138	repo = _make_repo(tmp_path)
2139	task = create_task(repo, "Complete me")
2140	args_c = _claim_ns(run_id="worker", fmt="json")
2141	with _patch_repo(repo):
2142	run_claim(args_c)
2143	out_c = json.loads(capsys.readouterr().out)
2144	assert out_c["status"] == "claimed"
2145
2146	args_done = _namespace(task_id=task.task_id, run_id="worker",
2147	result='{"ok": true}', fmt="json")
2148	with _patch_repo(repo):
2149	run_complete(args_done)
2150	out_done = json.loads(capsys.readouterr().out)
2151	assert out_done["status"] == "completed"
2152
2153	def test_claim_then_fail(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2154	repo = _make_repo(tmp_path)
2155	task = create_task(repo, "Fail me")
2156	args_c = _claim_ns(run_id="worker", fmt="json")
2157	with _patch_repo(repo):
2158	run_claim(args_c)
2159	capsys.readouterr()
2160
2161	args_f = _namespace(task_id=task.task_id, run_id="worker",
2162	error="network timeout", fmt="json")
2163	with _patch_repo(repo):
2164	run_fail_task(args_f)
2165	out = json.loads(capsys.readouterr().out)
2166	assert out["status"] == "failed"
2167
2168	def test_second_claim_empty_after_first(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2169	repo = _make_repo(tmp_path)
2170	create_task(repo, "Only task")
2171	args = _claim_ns(fmt="json")
2172	with _patch_repo(repo):
2173	run_claim(args)
2174	capsys.readouterr()
2175	with _patch_repo(repo):
2176	with pytest.raises(SystemExit) as exc:
2177	run_claim(args)
2178	assert exc.value.code == 1
2179
2180	def test_priority_order_across_claims(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2181	repo = _make_repo(tmp_path)
2182	create_task(repo, "Low", priority=0)
2183	create_task(repo, "High", priority=9)
2184	create_task(repo, "Mid", priority=5)
2185
2186	titles = []
2187	for _ in range(3):
2188	args = _claim_ns(fmt="json")
2189	with _patch_repo(repo):
2190	run_claim(args)
2191	out = json.loads(capsys.readouterr().out)
2192	titles.append(out["task"]["title"])
2193
2194	assert titles == ["High", "Mid", "Low"]
2195
2196	def test_enqueue_claim_complete_via_cli(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2197	"""Full round-trip using CLI commands only."""
2198	repo = _make_repo(tmp_path)
2199
2200	# Enqueue
2201	args_e = _enqueue_ns(title="CLI round-trip", payload='{"x": 1}', fmt="json")
2202	with _patch_repo(repo):
2203	run_enqueue(args_e)
2204	enq = json.loads(capsys.readouterr().out)
2205	task_id = enq["task_id"]
2206
2207	# Claim
2208	args_c = _claim_ns(run_id="cli-worker", fmt="json")
2209	with _patch_repo(repo):
2210	run_claim(args_c)
2211	clm = json.loads(capsys.readouterr().out)
2212	assert clm["status"] == "claimed"
2213	assert clm["task"]["payload"]["x"] == 1
2214
2215	# Complete
2216	args_done = _namespace(task_id=task_id, run_id="cli-worker",
2217	result='{"done": true}', fmt="json")
2218	with _patch_repo(repo):
2219	run_complete(args_done)
2220	done = json.loads(capsys.readouterr().out)
2221	assert done["status"] == "completed"
2222
2223
2224	class TestClaimStress:
2225	"""Concurrent claiming and performance under load."""
2226
2227	def test_concurrent_claim_no_double_claim(self, tmp_path: pathlib.Path) -> None:
2228	"""10 tasks + 20 competing agents — no task claimed twice."""
2229	repo = _make_repo(tmp_path)
2230	for i in range(10):
2231	create_task(repo, f"Task {i}")
2232
2233	claimed_ids: list[str] = []
2234	lock = threading.Lock()
2235
2236	def _agent(n: int) -> None:
2237	try:
2238	result = claim_next_task(repo, f"agent-{n}")
2239	if result is not None:
2240	task, _ = result
2241	with lock:
2242	claimed_ids.append(task.task_id)
2243	except Exception:
2244	pass
2245
2246	threads = [threading.Thread(target=_agent, args=(i,)) for i in range(20)]
2247	for t in threads:
2248	t.start()
2249	for t in threads:
2250	t.join()
2251
2252	assert len(claimed_ids) == len(set(claimed_ids)), "task claimed twice"
2253	assert len(claimed_ids) <= 10
2254
2255	def test_claim_100_tasks_sequential(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2256	repo = _make_repo(tmp_path)
2257	for i in range(100):
2258	create_task(repo, f"Task {i}", queue="load")
2259
2260	start = time.monotonic()
2261	claimed = 0
2262	while claimed < 100:
2263	args = _claim_ns(queue="load", fmt="json")
2264	with _patch_repo(repo):
2265	try:
2266	run_claim(args)
2267	claimed += 1
2268	capsys.readouterr()
2269	except SystemExit:
2270	break
2271	elapsed = time.monotonic() - start
2272	assert claimed == 100
2273	assert elapsed < 15.0, f"100 sequential claims took {elapsed:.2f}s"
2274
2275	def test_wait_zero_performance(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2276	"""--wait 0 on an empty queue must return in < 0.5 s."""
2277	repo = _make_repo(tmp_path)
2278	for _ in range(5):
2279	args = _claim_ns(wait=0, fmt="json")
2280	start = time.monotonic()
2281	with _patch_repo(repo):
2282	with pytest.raises(SystemExit):
2283	run_claim(args)
2284	elapsed = time.monotonic() - start
2285	assert elapsed < 0.5, f"wait=0 took {elapsed:.3f}s"
2286	capsys.readouterr()
2287
2288
2289	class TestCliComplete:
2290	"""run_complete: success, wrong claimer, missing task."""
2291
2292	def _setup(self, repo: pathlib.Path) -> TaskRecord:
2293	t = create_task(repo, "Completable task")
2294	claim_next_task(repo, "agent-1")
2295	return t
2296
2297	def test_complete_success_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2298	repo = _make_repo(tmp_path)
2299	t = self._setup(repo)
2300	args = _namespace(task_id=t.task_id, run_id="agent-1", result="{}", fmt="json")
2301	with _patch_repo(repo):
2302	run_complete(args)
2303	out = json.loads(capsys.readouterr().out)
2304	assert out["status"] == "completed"
2305
2306	def test_complete_with_result(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2307	repo = _make_repo(tmp_path)
2308	t = self._setup(repo)
2309	args = _namespace(task_id=t.task_id, run_id="agent-1",
2310	result='{"pr": 99}', fmt="json")
2311	with _patch_repo(repo):
2312	run_complete(args)
2313	out = json.loads(capsys.readouterr().out)
2314	assert out["result"] == {"pr": 99}
2315
2316	def test_complete_wrong_claimer_exits_1(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2317	repo = _make_repo(tmp_path)
2318	t = self._setup(repo)
2319	args = _namespace(task_id=t.task_id, run_id="impostor", result="{}", fmt="json")
2320	with _patch_repo(repo):
2321	with pytest.raises(SystemExit) as exc:
2322	run_complete(args)
2323	assert exc.value.code == 1
2324
2325	def test_complete_invalid_result_exits_1(self, tmp_path: pathlib.Path) -> None:
2326	repo = _make_repo(tmp_path)
2327	t = self._setup(repo)
2328	args = _namespace(task_id=t.task_id, run_id="agent-1", result="NOTJSON", fmt="json")
2329	with _patch_repo(repo):
2330	with pytest.raises(SystemExit) as exc:
2331	run_complete(args)
2332	assert exc.value.code == 1
2333
2334	def test_complete_text_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2335	repo = _make_repo(tmp_path)
2336	t = self._setup(repo)
2337	args = _namespace(task_id=t.task_id, run_id="agent-1", result="{}", fmt="text")
2338	with _patch_repo(repo):
2339	run_complete(args)
2340	out = capsys.readouterr().out
2341	assert "completed" in out.lower()
2342
2343
2344	# ── complete hardening ────────────────────────────────────────────────────────
2345
2346	from muse.cli.commands.task_queue import _MAX_RESULT_BYTES
2347
2348
2349	def _complete_ns(**kwargs: MsgpackValue) -> argparse.Namespace:
2350	"""Build a Namespace with complete-appropriate defaults."""
2351	defaults = {
2352	"fmt": "json",
2353	"run_id": "agent-1",
2354	"task_id": VALID_UUID,
2355	"result": "{}",
2356	}
2357	defaults.update(kwargs)
2358	return argparse.Namespace(**defaults)
2359
2360
2361	class TestCompleteInputValidation:
2362	"""All complete validation fires before require_repo() and exits 1."""
2363
2364	def _setup(self, repo: pathlib.Path) -> TaskRecord:
2365	t = create_task(repo, "Complete-me")
2366	claim_next_task(repo, "agent-1")
2367	return t
2368
2369	def test_run_id_too_long_exits_1(self, tmp_path: pathlib.Path) -> None:
2370	repo = _make_repo(tmp_path)
2371	t = self._setup(repo)
2372	args = _complete_ns(task_id=t.task_id, run_id="x" * 257)
2373	with _patch_repo(repo):
2374	with pytest.raises(SystemExit) as exc:
2375	run_complete(args)
2376	assert exc.value.code == ExitCode.USER_ERROR
2377
2378	def test_run_id_at_max_length_passes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2379	repo = _make_repo(tmp_path)
2380	t = self._setup(repo)
2381	args = _complete_ns(task_id=t.task_id, run_id="a" * 256)
2382	# claim was made by "agent-1" — different run_id triggers PermissionError
2383	# but validation itself must pass (no bad_args exit before I/O)
2384	with _patch_repo(repo):
2385	with pytest.raises(SystemExit) as exc:
2386	run_complete(args)
2387	# exits 1 due to wrong claimer, NOT bad_args — validation passed
2388	out = json.loads(capsys.readouterr().out)
2389	assert out.get("status") != "bad_args"
2390
2391	def test_result_not_json_exits_1(self, tmp_path: pathlib.Path) -> None:
2392	repo = _make_repo(tmp_path)
2393	t = self._setup(repo)
2394	args = _complete_ns(task_id=t.task_id, result="not-json")
2395	with _patch_repo(repo):
2396	with pytest.raises(SystemExit) as exc:
2397	run_complete(args)
2398	assert exc.value.code == ExitCode.USER_ERROR
2399
2400	def test_result_array_exits_1(self, tmp_path: pathlib.Path) -> None:
2401	repo = _make_repo(tmp_path)
2402	t = self._setup(repo)
2403	args = _complete_ns(task_id=t.task_id, result="[1, 2, 3]")
2404	with _patch_repo(repo):
2405	with pytest.raises(SystemExit) as exc:
2406	run_complete(args)
2407	assert exc.value.code == ExitCode.USER_ERROR
2408
2409	def test_result_scalar_exits_1(self, tmp_path: pathlib.Path) -> None:
2410	repo = _make_repo(tmp_path)
2411	t = self._setup(repo)
2412	args = _complete_ns(task_id=t.task_id, result="42")
2413	with _patch_repo(repo):
2414	with pytest.raises(SystemExit) as exc:
2415	run_complete(args)
2416	assert exc.value.code == ExitCode.USER_ERROR
2417
2418	def test_result_null_exits_1(self, tmp_path: pathlib.Path) -> None:
2419	repo = _make_repo(tmp_path)
2420	t = self._setup(repo)
2421	args = _complete_ns(task_id=t.task_id, result="null")
2422	with _patch_repo(repo):
2423	with pytest.raises(SystemExit) as exc:
2424	run_complete(args)
2425	assert exc.value.code == ExitCode.USER_ERROR
2426
2427	def test_result_too_large_exits_1(self, tmp_path: pathlib.Path) -> None:
2428	repo = _make_repo(tmp_path)
2429	t = self._setup(repo)
2430	big = json.dumps({"k": "v" * (_MAX_RESULT_BYTES + 1)})
2431	args = _complete_ns(task_id=t.task_id, result=big)
2432	with _patch_repo(repo):
2433	with pytest.raises(SystemExit) as exc:
2434	run_complete(args)
2435	assert exc.value.code == ExitCode.USER_ERROR
2436
2437	def test_result_at_max_size_passes_validation(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2438	repo = _make_repo(tmp_path)
2439	t = self._setup(repo)
2440	# Build compact JSON exactly at the limit using explicit separators
2441	prefix = '{"k":"'
2442	suffix = '"}'
2443	inner = "x" * (_MAX_RESULT_BYTES - len(prefix) - len(suffix))
2444	payload = prefix + inner + suffix
2445	assert len(payload.encode()) == _MAX_RESULT_BYTES
2446	args = _complete_ns(task_id=t.task_id, result=payload)
2447	with _patch_repo(repo):
2448	run_complete(args)
2449	out = json.loads(capsys.readouterr().out)
2450	assert out["status"] == "completed"
2451
2452	def test_invalid_task_id_exits_1(self, tmp_path: pathlib.Path) -> None:
2453	repo = _make_repo(tmp_path)
2454	args = _complete_ns(task_id="not-a-uuid")
2455	with _patch_repo(repo):
2456	with pytest.raises(SystemExit) as exc:
2457	run_complete(args)
2458	assert exc.value.code == ExitCode.USER_ERROR
2459
2460	def test_path_traversal_task_id_exits_1(self, tmp_path: pathlib.Path) -> None:
2461	repo = _make_repo(tmp_path)
2462	args = _complete_ns(task_id="../../etc/passwd")
2463	with _patch_repo(repo):
2464	with pytest.raises(SystemExit) as exc:
2465	run_complete(args)
2466	assert exc.value.code == ExitCode.USER_ERROR
2467
2468	def test_null_byte_task_id_exits_1(self, tmp_path: pathlib.Path) -> None:
2469	repo = _make_repo(tmp_path)
2470	args = _complete_ns(task_id="12345678-1234-4abc-8abc-123456789\x00ab")
2471	with _patch_repo(repo):
2472	with pytest.raises(SystemExit) as exc:
2473	run_complete(args)
2474	assert exc.value.code == ExitCode.USER_ERROR
2475
2476	def test_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
2477	"""require_repo must NOT be called when task_id is invalid."""
2478	call_count = {"n": 0}
2479	original = __import__("muse.core.repo", fromlist=["require_repo"]).require_repo
2480
2481	def counting_require_repo() -> pathlib.Path:
2482	call_count["n"] += 1
2483	return original()
2484
2485	args = _complete_ns(task_id="BADUUID")
2486	with patch("muse.cli.commands.task_queue.require_repo", counting_require_repo):
2487	with pytest.raises(SystemExit):
2488	run_complete(args)
2489	assert call_count["n"] == 0, "require_repo was called before task_id validation"
2490
2491	def test_json_error_shape_bad_task_id(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2492	repo = _make_repo(tmp_path)
2493	args = _complete_ns(task_id="bad-id", fmt="json")
2494	with _patch_repo(repo):
2495	with pytest.raises(SystemExit):
2496	run_complete(args)
2497	out = json.loads(capsys.readouterr().out)
2498	assert "error" in out
2499	assert out["status"] == "bad_task_id"
2500
2501	def test_json_error_shape_bad_result(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2502	repo = _make_repo(tmp_path)
2503	args = _complete_ns(task_id=VALID_UUID, result="NOTJSON", fmt="json")
2504	with _patch_repo(repo):
2505	with pytest.raises(SystemExit):
2506	run_complete(args)
2507	out = json.loads(capsys.readouterr().out)
2508	assert "error" in out
2509	assert out["status"] == "bad_args"
2510
2511	def test_text_error_goes_to_stderr(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2512	repo = _make_repo(tmp_path)
2513	args = _complete_ns(task_id="BADUUID", fmt="text")
2514	with _patch_repo(repo):
2515	with pytest.raises(SystemExit):
2516	run_complete(args)
2517	captured = capsys.readouterr()
2518	assert captured.out == ""
2519	assert "❌" in captured.err
2520
2521	def test_json_error_goes_to_stdout_not_stderr(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2522	repo = _make_repo(tmp_path)
2523	args = _complete_ns(task_id="BADUUID", fmt="json")
2524	with _patch_repo(repo):
2525	with pytest.raises(SystemExit):
2526	run_complete(args)
2527	captured = capsys.readouterr()
2528	assert captured.err == ""
2529	out = json.loads(captured.out)
2530	assert "error" in out
2531
2532
2533	class TestCompleteJsonOutput:
2534	"""run_complete JSON output shape and compactness."""
2535
2536	def _setup(self, repo: pathlib.Path) -> TaskRecord:
2537	t = create_task(repo, "JSON task", queue="billing")
2538	claim_next_task(repo, "completer-1")
2539	return t
2540
2541	def test_json_is_compact(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2542	repo = _make_repo(tmp_path)
2543	t = self._setup(repo)
2544	args = _complete_ns(task_id=t.task_id, run_id="completer-1")
2545	with _patch_repo(repo):
2546	run_complete(args)
2547	raw = capsys.readouterr().out.strip()
2548	assert "\n" not in raw, "JSON output must be single line (compact)"
2549
2550	def test_json_has_required_keys(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2551	repo = _make_repo(tmp_path)
2552	t = self._setup(repo)
2553	args = _complete_ns(task_id=t.task_id, run_id="completer-1")
2554	with _patch_repo(repo):
2555	run_complete(args)
2556	out = json.loads(capsys.readouterr().out)
2557	for key in ("schema_version", "task_id", "claimer_run_id", "status",
2558	"claimed_at", "expires_at", "result", "elapsed_seconds"):
2559	assert key in out, f"missing key: {key}"
2560
2561	def test_json_status_is_completed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2562	repo = _make_repo(tmp_path)
2563	t = self._setup(repo)
2564	args = _complete_ns(task_id=t.task_id, run_id="completer-1")
2565	with _patch_repo(repo):
2566	run_complete(args)
2567	out = json.loads(capsys.readouterr().out)
2568	assert out["status"] == "completed"
2569
2570	def test_json_result_field_populated(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2571	repo = _make_repo(tmp_path)
2572	t = self._setup(repo)
2573	args = _complete_ns(task_id=t.task_id, run_id="completer-1",
2574	result='{"pr_url": "http://x/1"}')
2575	with _patch_repo(repo):
2576	run_complete(args)
2577	out = json.loads(capsys.readouterr().out)
2578	assert out["result"] == {"pr_url": "http://x/1"}
2579
2580	def test_json_empty_result_is_null(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2581	"""Empty dict {} collapses to null in the claim (falsy check)."""
2582	repo = _make_repo(tmp_path)
2583	t = self._setup(repo)
2584	args = _complete_ns(task_id=t.task_id, run_id="completer-1", result="{}")
2585	with _patch_repo(repo):
2586	run_complete(args)
2587	out = json.loads(capsys.readouterr().out)
2588	assert out["result"] is None
2589
2590	def test_json_elapsed_is_float(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2591	repo = _make_repo(tmp_path)
2592	t = self._setup(repo)
2593	args = _complete_ns(task_id=t.task_id, run_id="completer-1")
2594	with _patch_repo(repo):
2595	run_complete(args)
2596	out = json.loads(capsys.readouterr().out)
2597	assert isinstance(out["elapsed_seconds"], float)
2598
2599	def test_json_claimer_run_id_matches(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2600	repo = _make_repo(tmp_path)
2601	t = self._setup(repo)
2602	args = _complete_ns(task_id=t.task_id, run_id="completer-1")
2603	with _patch_repo(repo):
2604	run_complete(args)
2605	out = json.loads(capsys.readouterr().out)
2606	assert out["claimer_run_id"] == "completer-1"
2607
2608	def test_json_wrong_claimer_error_shape(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2609	repo = _make_repo(tmp_path)
2610	t = self._setup(repo)
2611	args = _complete_ns(task_id=t.task_id, run_id="impostor", fmt="json")
2612	with _patch_repo(repo):
2613	with pytest.raises(SystemExit) as exc:
2614	run_complete(args)
2615	assert exc.value.code == 1
2616	out = json.loads(capsys.readouterr().out)
2617	assert "error" in out
2618
2619	def test_json_missing_task_error_shape(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2620	repo = _make_repo(tmp_path)
2621	# repo exists but no task was created
2622	args = _complete_ns(task_id=VALID_UUID, run_id="completer-1", fmt="json")
2623	with _patch_repo(repo):
2624	with pytest.raises(SystemExit) as exc:
2625	run_complete(args)
2626	assert exc.value.code == 1
2627	out = json.loads(capsys.readouterr().out)
2628	assert "error" in out
2629
2630
2631	class TestCompleteTextOutput:
2632	"""run_complete text output content."""
2633
2634	def _setup(self, repo: pathlib.Path, title: str = "My Task", queue: str = "default") -> TaskRecord:
2635	t = create_task(repo, title, queue=queue)
2636	claim_next_task(repo, "agent-1")
2637	return t
2638
2639	def test_text_shows_completed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2640	repo = _make_repo(tmp_path)
2641	t = self._setup(repo)
2642	args = _complete_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
2643	with _patch_repo(repo):
2644	run_complete(args)
2645	out = capsys.readouterr().out
2646	assert "completed" in out.lower()
2647
2648	def test_text_shows_task_title(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2649	repo = _make_repo(tmp_path)
2650	t = self._setup(repo, title="Rename billing module")
2651	args = _complete_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
2652	with _patch_repo(repo):
2653	run_complete(args)
2654	out = capsys.readouterr().out
2655	assert "Rename billing module" in out
2656
2657	def test_text_shows_queue(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2658	repo = _make_repo(tmp_path)
2659	t = self._setup(repo, queue="billing")
2660	args = _complete_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
2661	with _patch_repo(repo):
2662	run_complete(args)
2663	out = capsys.readouterr().out
2664	assert "billing" in out
2665
2666	def test_text_shows_claimer(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2667	repo = _make_repo(tmp_path)
2668	t = self._setup(repo)
2669	args = _complete_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
2670	with _patch_repo(repo):
2671	run_complete(args)
2672	out = capsys.readouterr().out
2673	assert "agent-1" in out
2674
2675	def test_text_shows_result_when_provided(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2676	repo = _make_repo(tmp_path)
2677	t = self._setup(repo)
2678	args = _complete_ns(task_id=t.task_id, run_id="agent-1",
2679	result='{"pr": 42}', fmt="text")
2680	with _patch_repo(repo):
2681	run_complete(args)
2682	out = capsys.readouterr().out
2683	assert "42" in out
2684
2685	def test_text_shows_elapsed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2686	repo = _make_repo(tmp_path)
2687	t = self._setup(repo)
2688	args = _complete_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
2689	with _patch_repo(repo):
2690	run_complete(args)
2691	out = capsys.readouterr().out
2692	assert "s)" in out # e.g. "(0.003s)"
2693
2694	def test_ansi_injection_in_run_id_stripped(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2695	evil_id = "agent\x1b[31mRED\x1b[0m"
2696	repo = _make_repo(tmp_path)
2697	t = create_task(repo, "task")
2698	claim_next_task(repo, evil_id)
2699	args = _complete_ns(task_id=t.task_id, run_id=evil_id, fmt="text")
2700	with _patch_repo(repo):
2701	run_complete(args)
2702	out = capsys.readouterr().out
2703	assert "\x1b" not in out
2704
2705	def test_ansi_injection_in_title_stripped(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2706	evil_title = "task\x1b[1mBOLD\x1b[0m"
2707	repo = _make_repo(tmp_path)
2708	t = create_task(repo, evil_title)
2709	claim_next_task(repo, "agent-1")
2710	args = _complete_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
2711	with _patch_repo(repo):
2712	run_complete(args)
2713	out = capsys.readouterr().out
2714	assert "\x1b" not in out
2715
2716
2717	class TestCompleteIntegration:
2718	"""Full lifecycle integration tests for run_complete."""
2719
2720	def test_completed_status_persisted_to_disk(self, tmp_path: pathlib.Path) -> None:
2721	repo = _make_repo(tmp_path)
2722	t = create_task(repo, "persist-me")
2723	claim_next_task(repo, "worker")
2724	args = _complete_ns(task_id=t.task_id, run_id="worker")
2725	with _patch_repo(repo):
2726	run_complete(args)
2727	claim = load_claim(repo, t.task_id)
2728	assert claim is not None
2729	assert claim.status == "completed"
2730
2731	def test_result_persisted_to_disk(self, tmp_path: pathlib.Path) -> None:
2732	repo = _make_repo(tmp_path)
2733	t = create_task(repo, "result-me")
2734	claim_next_task(repo, "worker")
2735	args = _complete_ns(task_id=t.task_id, run_id="worker",
2736	result='{"sha": "abc123"}')
2737	with _patch_repo(repo):
2738	run_complete(args)
2739	claim = load_claim(repo, t.task_id)
2740	assert claim is not None
2741	assert claim.result == {"sha": "abc123"}
2742
2743	def test_double_complete_fails(self, tmp_path: pathlib.Path) -> None:
2744	repo = _make_repo(tmp_path)
2745	t = create_task(repo, "once only")
2746	claim_next_task(repo, "worker")
2747	args = _complete_ns(task_id=t.task_id, run_id="worker")
2748	with _patch_repo(repo):
2749	run_complete(args)
2750	with _patch_repo(repo):
2751	with pytest.raises(SystemExit) as exc:
2752	run_complete(args)
2753	assert exc.value.code == 1
2754
2755	def test_complete_missing_claim_fails(self, tmp_path: pathlib.Path) -> None:
2756	"""Task exists but was never claimed."""
2757	repo = _make_repo(tmp_path)
2758	t = create_task(repo, "unclaimed")
2759	args = _complete_ns(task_id=t.task_id, run_id="worker")
2760	with _patch_repo(repo):
2761	with pytest.raises(SystemExit) as exc:
2762	run_complete(args)
2763	assert exc.value.code == 1
2764
2765	def test_enqueue_claim_complete_full_cycle(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2766	"""Full CLI round-trip: enqueue → claim → complete."""
2767	repo = _make_repo(tmp_path)
2768	# enqueue
2769	eq_args = _enqueue_ns(title="e2e task", queue="default")
2770	with _patch_repo(repo):
2771	run_enqueue(eq_args)
2772	task_id = json.loads(capsys.readouterr().out)["task_id"]
2773	# claim
2774	cl_args = _claim_ns(run_id="e2e-agent")
2775	with _patch_repo(repo):
2776	run_claim(cl_args)
2777	capsys.readouterr()
2778	# complete
2779	co_args = _complete_ns(task_id=task_id, run_id="e2e-agent",
2780	result='{"done": true}')
2781	with _patch_repo(repo):
2782	run_complete(co_args)
2783	out = json.loads(capsys.readouterr().out)
2784	assert out["status"] == "completed"
2785	assert out["result"] == {"done": True}
2786
2787	def test_complete_with_unicode_result(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2788	repo = _make_repo(tmp_path)
2789	t = create_task(repo, "unicode")
2790	claim_next_task(repo, "agent-1")
2791	args = _complete_ns(task_id=t.task_id, run_id="agent-1",
2792	result='{"msg": "héllo wörld 🎉"}')
2793	with _patch_repo(repo):
2794	run_complete(args)
2795	out = json.loads(capsys.readouterr().out)
2796	assert out["result"]["msg"] == "héllo wörld 🎉"
2797
2798
2799	class TestCompleteStress:
2800	"""Concurrency and throughput stress tests for run_complete."""
2801
2802	def test_20_agents_each_claim_and_complete_unique_task(self, tmp_path: pathlib.Path) -> None:
2803	"""20 concurrent agents each claim+complete a unique task with no conflicts."""
2804	import concurrent.futures
2805	repo = _make_repo(tmp_path)
2806	tasks = [create_task(repo, f"task-{i}") for i in range(20)]
2807
2808	completed: set[str] = set()
2809	lock = threading.Lock()
2810	errors: list[str] = []
2811
2812	def claim_and_complete(task: "TaskRecord") -> None:
2813	run_id = f"agent-{task.task_id[:8]}"
2814	result = claim_next_task(repo, run_id, queue=task.queue)
2815	if result is None:
2816	errors.append(f"no task for {run_id}")
2817	return
2818	claimed_task, _ = result
2819	try:
2820	complete_task(repo, claimed_task.task_id, run_id)
2821	except Exception as exc: # noqa: BLE001
2822	errors.append(str(exc))
2823	return
2824	with lock:
2825	completed.add(claimed_task.task_id)
2826
2827	with concurrent.futures.ThreadPoolExecutor(max_workers=20) as pool:
2828	list(pool.map(claim_and_complete, tasks))
2829
2830	assert not errors, f"errors: {errors}"
2831	assert len(completed) == 20, f"only {len(completed)}/20 completed"
2832
2833	def test_100_sequential_completes_under_10s(self, tmp_path: pathlib.Path) -> None:
2834	repo = _make_repo(tmp_path)
2835	for i in range(100):
2836	t = create_task(repo, f"seq-{i}")
2837	claim_next_task(repo, "batch-worker")
2838	start = time.monotonic()
2839	complete_task(repo, t.task_id, "batch-worker")
2840	assert time.monotonic() - start < 0.15, f"task {i} took too long"
2841
2842	def test_complete_via_run_complete_100_sequential(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2843	repo = _make_repo(tmp_path)
2844	start = time.monotonic()
2845	for i in range(100):
2846	t = create_task(repo, f"cli-{i}")
2847	claim_next_task(repo, "cli-worker")
2848	args = _complete_ns(task_id=t.task_id, run_id="cli-worker")
2849	with _patch_repo(repo):
2850	run_complete(args)
2851	capsys.readouterr()
2852	elapsed = time.monotonic() - start
2853	assert elapsed < 15.0, f"100 CLI completes took {elapsed:.1f}s"
2854
2855
2856	class TestCliFailTask:
2857	"""run_fail_task: success, wrong claimer, text/json output."""
2858
2859	def _setup(self, repo: pathlib.Path) -> TaskRecord:
2860	t = create_task(repo, "Failing task")
2861	claim_next_task(repo, "agent-1")
2862	return t
2863
2864	def test_fail_success_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2865	repo = _make_repo(tmp_path)
2866	t = self._setup(repo)
2867	args = _namespace(task_id=t.task_id, run_id="agent-1", error="network down", fmt="json")
2868	with _patch_repo(repo):
2869	run_fail_task(args)
2870	out = json.loads(capsys.readouterr().out)
2871	assert out["status"] == "failed"
2872	assert out["error"] == "network down"
2873
2874	def test_fail_wrong_claimer_exits_1(self, tmp_path: pathlib.Path) -> None:
2875	repo = _make_repo(tmp_path)
2876	t = self._setup(repo)
2877	args = _namespace(task_id=t.task_id, run_id="wrong", error="x", fmt="json")
2878	with _patch_repo(repo):
2879	with pytest.raises(SystemExit) as exc:
2880	run_fail_task(args)
2881	assert exc.value.code == 1
2882
2883	def test_fail_text_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2884	repo = _make_repo(tmp_path)
2885	t = self._setup(repo)
2886	args = _namespace(task_id=t.task_id, run_id="agent-1", error="boom", fmt="text")
2887	with _patch_repo(repo):
2888	run_fail_task(args)
2889	out = capsys.readouterr().out
2890	assert "failed" in out.lower()
2891
2892
2893	# ── fail-task hardening ───────────────────────────────────────────────────────
2894
2895	from muse.cli.commands.task_queue import _MAX_ERROR_LEN
2896
2897
2898	def _fail_ns(**kwargs: MsgpackValue) -> argparse.Namespace:
2899	"""Build a Namespace with fail-task-appropriate defaults."""
2900	defaults = {
2901	"fmt": "json",
2902	"run_id": "agent-1",
2903	"task_id": VALID_UUID,
2904	"error": "something went wrong",
2905	}
2906	defaults.update(kwargs)
2907	return argparse.Namespace(**defaults)
2908
2909
2910	class TestFailTaskInputValidation:
2911	"""All fail-task validation fires before require_repo() and exits 1."""
2912
2913	def _setup(self, repo: pathlib.Path) -> TaskRecord:
2914	t = create_task(repo, "Failable task")
2915	claim_next_task(repo, "agent-1")
2916	return t
2917
2918	def test_run_id_too_long_exits_1(self, tmp_path: pathlib.Path) -> None:
2919	repo = _make_repo(tmp_path)
2920	t = self._setup(repo)
2921	args = _fail_ns(task_id=t.task_id, run_id="x" * 257)
2922	with _patch_repo(repo):
2923	with pytest.raises(SystemExit) as exc:
2924	run_fail_task(args)
2925	assert exc.value.code == ExitCode.USER_ERROR
2926
2927	def test_run_id_at_max_length_passes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2928	repo = _make_repo(tmp_path)
2929	t = self._setup(repo)
2930	# 256-char run_id is valid length; mismatch with claimer triggers PermissionError
2931	args = _fail_ns(task_id=t.task_id, run_id="b" * 256)
2932	with _patch_repo(repo):
2933	with pytest.raises(SystemExit) as exc:
2934	run_fail_task(args)
2935	out = json.loads(capsys.readouterr().out)
2936	# Exits 1 due to wrong claimer, NOT bad_args — length validation passed
2937	assert out.get("status") != "bad_args"
2938
2939	def test_error_too_long_exits_1(self, tmp_path: pathlib.Path) -> None:
2940	repo = _make_repo(tmp_path)
2941	t = self._setup(repo)
2942	args = _fail_ns(task_id=t.task_id, error="e" * (_MAX_ERROR_LEN + 1))
2943	with _patch_repo(repo):
2944	with pytest.raises(SystemExit) as exc:
2945	run_fail_task(args)
2946	assert exc.value.code == ExitCode.USER_ERROR
2947
2948	def test_error_at_max_length_passes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2949	repo = _make_repo(tmp_path)
2950	t = self._setup(repo)
2951	args = _fail_ns(task_id=t.task_id, run_id="agent-1",
2952	error="e" * _MAX_ERROR_LEN)
2953	with _patch_repo(repo):
2954	run_fail_task(args)
2955	out = json.loads(capsys.readouterr().out)
2956	assert out["status"] == "failed"
2957
2958	def test_empty_error_allowed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
2959	repo = _make_repo(tmp_path)
2960	t = self._setup(repo)
2961	args = _fail_ns(task_id=t.task_id, run_id="agent-1", error="")
2962	with _patch_repo(repo):
2963	run_fail_task(args)
2964	out = json.loads(capsys.readouterr().out)
2965	assert out["status"] == "failed"
2966
2967	def test_invalid_task_id_exits_1(self, tmp_path: pathlib.Path) -> None:
2968	repo = _make_repo(tmp_path)
2969	args = _fail_ns(task_id="not-a-uuid")
2970	with _patch_repo(repo):
2971	with pytest.raises(SystemExit) as exc:
2972	run_fail_task(args)
2973	assert exc.value.code == ExitCode.USER_ERROR
2974
2975	def test_path_traversal_task_id_exits_1(self, tmp_path: pathlib.Path) -> None:
2976	repo = _make_repo(tmp_path)
2977	args = _fail_ns(task_id="../../etc/passwd")
2978	with _patch_repo(repo):
2979	with pytest.raises(SystemExit) as exc:
2980	run_fail_task(args)
2981	assert exc.value.code == ExitCode.USER_ERROR
2982
2983	def test_null_byte_task_id_exits_1(self, tmp_path: pathlib.Path) -> None:
2984	repo = _make_repo(tmp_path)
2985	args = _fail_ns(task_id="12345678-1234-4abc-8abc-123456789\x00ab")
2986	with _patch_repo(repo):
2987	with pytest.raises(SystemExit) as exc:
2988	run_fail_task(args)
2989	assert exc.value.code == ExitCode.USER_ERROR
2990
2991	def test_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
2992	"""require_repo must NOT be called when task_id is invalid."""
2993	call_count = {"n": 0}
2994
2995	def counting_require_repo() -> pathlib.Path:
2996	call_count["n"] += 1
2997	raise RuntimeError("should not reach here")
2998
2999	args = _fail_ns(task_id="BADUUID")
3000	with patch("muse.cli.commands.task_queue.require_repo", counting_require_repo):
3001	with pytest.raises(SystemExit):
3002	run_fail_task(args)
3003	assert call_count["n"] == 0, "require_repo called before task_id validation"
3004
3005	def test_run_id_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
3006	call_count = {"n": 0}
3007
3008	def counting_require_repo() -> pathlib.Path:
3009	call_count["n"] += 1
3010	raise RuntimeError("should not reach here")
3011
3012	args = _fail_ns(task_id=VALID_UUID, run_id="r" * 300)
3013	with patch("muse.cli.commands.task_queue.require_repo", counting_require_repo):
3014	with pytest.raises(SystemExit):
3015	run_fail_task(args)
3016	assert call_count["n"] == 0
3017
3018	def test_error_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
3019	call_count = {"n": 0}
3020
3021	def counting_require_repo() -> pathlib.Path:
3022	call_count["n"] += 1
3023	raise RuntimeError("should not reach here")
3024
3025	args = _fail_ns(task_id=VALID_UUID, error="e" * 5000)
3026	with patch("muse.cli.commands.task_queue.require_repo", counting_require_repo):
3027	with pytest.raises(SystemExit):
3028	run_fail_task(args)
3029	assert call_count["n"] == 0
3030
3031	def test_json_error_shape_bad_task_id(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3032	repo = _make_repo(tmp_path)
3033	args = _fail_ns(task_id="bad-id", fmt="json")
3034	with _patch_repo(repo):
3035	with pytest.raises(SystemExit):
3036	run_fail_task(args)
3037	out = json.loads(capsys.readouterr().out)
3038	assert "error" in out
3039	assert out["status"] == "bad_task_id"
3040
3041	def test_json_error_shape_bad_run_id(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3042	repo = _make_repo(tmp_path)
3043	args = _fail_ns(task_id=VALID_UUID, run_id="x" * 300, fmt="json")
3044	with _patch_repo(repo):
3045	with pytest.raises(SystemExit):
3046	run_fail_task(args)
3047	out = json.loads(capsys.readouterr().out)
3048	assert "error" in out
3049	assert out["status"] == "bad_args"
3050
3051	def test_json_error_shape_bad_error_msg(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3052	repo = _make_repo(tmp_path)
3053	args = _fail_ns(task_id=VALID_UUID, error="e" * 5000, fmt="json")
3054	with _patch_repo(repo):
3055	with pytest.raises(SystemExit):
3056	run_fail_task(args)
3057	out = json.loads(capsys.readouterr().out)
3058	assert "error" in out
3059	assert out["status"] == "bad_args"
3060
3061	def test_text_error_goes_to_stderr(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3062	repo = _make_repo(tmp_path)
3063	args = _fail_ns(task_id="BADUUID", fmt="text")
3064	with _patch_repo(repo):
3065	with pytest.raises(SystemExit):
3066	run_fail_task(args)
3067	captured = capsys.readouterr()
3068	assert captured.out == ""
3069	assert "❌" in captured.err
3070
3071	def test_json_error_goes_to_stdout_not_stderr(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3072	repo = _make_repo(tmp_path)
3073	args = _fail_ns(task_id="BADUUID", fmt="json")
3074	with _patch_repo(repo):
3075	with pytest.raises(SystemExit):
3076	run_fail_task(args)
3077	captured = capsys.readouterr()
3078	assert captured.err == ""
3079	out = json.loads(captured.out)
3080	assert "error" in out
3081
3082
3083	class TestFailTaskJsonOutput:
3084	"""run_fail_task JSON output shape and compactness."""
3085
3086	def _setup(self, repo: pathlib.Path, queue: str = "default") -> TaskRecord:
3087	t = create_task(repo, "JSON fail task", queue=queue)
3088	claim_next_task(repo, "failer-1")
3089	return t
3090
3091	def test_json_is_compact(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3092	repo = _make_repo(tmp_path)
3093	t = self._setup(repo)
3094	args = _fail_ns(task_id=t.task_id, run_id="failer-1")
3095	with _patch_repo(repo):
3096	run_fail_task(args)
3097	raw = capsys.readouterr().out.strip()
3098	assert "\n" not in raw, "JSON output must be single line (compact)"
3099
3100	def test_json_has_required_keys(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3101	repo = _make_repo(tmp_path)
3102	t = self._setup(repo)
3103	args = _fail_ns(task_id=t.task_id, run_id="failer-1")
3104	with _patch_repo(repo):
3105	run_fail_task(args)
3106	out = json.loads(capsys.readouterr().out)
3107	for key in ("schema_version", "task_id", "claimer_run_id", "status",
3108	"claimed_at", "expires_at", "error", "elapsed_seconds"):
3109	assert key in out, f"missing key: {key}"
3110
3111	def test_json_status_is_failed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3112	repo = _make_repo(tmp_path)
3113	t = self._setup(repo)
3114	args = _fail_ns(task_id=t.task_id, run_id="failer-1")
3115	with _patch_repo(repo):
3116	run_fail_task(args)
3117	out = json.loads(capsys.readouterr().out)
3118	assert out["status"] == "failed"
3119
3120	def test_json_error_field_populated(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3121	repo = _make_repo(tmp_path)
3122	t = self._setup(repo)
3123	args = _fail_ns(task_id=t.task_id, run_id="failer-1",
3124	error="connection refused on port 5432")
3125	with _patch_repo(repo):
3126	run_fail_task(args)
3127	out = json.loads(capsys.readouterr().out)
3128	assert out["error"] == "connection refused on port 5432"
3129
3130	def test_json_empty_error_is_empty_string(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3131	repo = _make_repo(tmp_path)
3132	t = self._setup(repo)
3133	args = _fail_ns(task_id=t.task_id, run_id="failer-1", error="")
3134	with _patch_repo(repo):
3135	run_fail_task(args)
3136	out = json.loads(capsys.readouterr().out)
3137	assert out["error"] == ""
3138
3139	def test_json_elapsed_is_float(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3140	repo = _make_repo(tmp_path)
3141	t = self._setup(repo)
3142	args = _fail_ns(task_id=t.task_id, run_id="failer-1")
3143	with _patch_repo(repo):
3144	run_fail_task(args)
3145	out = json.loads(capsys.readouterr().out)
3146	assert isinstance(out["elapsed_seconds"], float)
3147
3148	def test_json_claimer_run_id_matches(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3149	repo = _make_repo(tmp_path)
3150	t = self._setup(repo)
3151	args = _fail_ns(task_id=t.task_id, run_id="failer-1")
3152	with _patch_repo(repo):
3153	run_fail_task(args)
3154	out = json.loads(capsys.readouterr().out)
3155	assert out["claimer_run_id"] == "failer-1"
3156
3157	def test_json_wrong_claimer_error_shape(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3158	repo = _make_repo(tmp_path)
3159	t = self._setup(repo)
3160	args = _fail_ns(task_id=t.task_id, run_id="impostor", fmt="json")
3161	with _patch_repo(repo):
3162	with pytest.raises(SystemExit) as exc:
3163	run_fail_task(args)
3164	assert exc.value.code == 1
3165	out = json.loads(capsys.readouterr().out)
3166	assert "error" in out
3167
3168	def test_json_missing_task_error_shape(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3169	repo = _make_repo(tmp_path)
3170	args = _fail_ns(task_id=VALID_UUID, run_id="failer-1", fmt="json")
3171	with _patch_repo(repo):
3172	with pytest.raises(SystemExit) as exc:
3173	run_fail_task(args)
3174	assert exc.value.code == 1
3175	out = json.loads(capsys.readouterr().out)
3176	assert "error" in out
3177
3178
3179	class TestFailTaskTextOutput:
3180	"""run_fail_task text output content."""
3181
3182	def _setup(self, repo: pathlib.Path, title: str = "My Task", queue: str = "default") -> TaskRecord:
3183	t = create_task(repo, title, queue=queue)
3184	claim_next_task(repo, "agent-1")
3185	return t
3186
3187	def test_text_shows_failed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3188	repo = _make_repo(tmp_path)
3189	t = self._setup(repo)
3190	args = _fail_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
3191	with _patch_repo(repo):
3192	run_fail_task(args)
3193	out = capsys.readouterr().out
3194	assert "failed" in out.lower()
3195
3196	def test_text_shows_task_title(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3197	repo = _make_repo(tmp_path)
3198	t = self._setup(repo, title="Deploy billing service")
3199	args = _fail_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
3200	with _patch_repo(repo):
3201	run_fail_task(args)
3202	out = capsys.readouterr().out
3203	assert "Deploy billing service" in out
3204
3205	def test_text_shows_queue(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3206	repo = _make_repo(tmp_path)
3207	t = self._setup(repo, queue="infra")
3208	args = _fail_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
3209	with _patch_repo(repo):
3210	run_fail_task(args)
3211	out = capsys.readouterr().out
3212	assert "infra" in out
3213
3214	def test_text_shows_claimer(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3215	repo = _make_repo(tmp_path)
3216	t = self._setup(repo)
3217	args = _fail_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
3218	with _patch_repo(repo):
3219	run_fail_task(args)
3220	out = capsys.readouterr().out
3221	assert "agent-1" in out
3222
3223	def test_text_shows_error_message(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3224	repo = _make_repo(tmp_path)
3225	t = self._setup(repo)
3226	args = _fail_ns(task_id=t.task_id, run_id="agent-1",
3227	error="disk full on /data", fmt="text")
3228	with _patch_repo(repo):
3229	run_fail_task(args)
3230	out = capsys.readouterr().out
3231	assert "disk full on /data" in out
3232
3233	def test_text_no_error_line_when_empty(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3234	repo = _make_repo(tmp_path)
3235	t = self._setup(repo)
3236	args = _fail_ns(task_id=t.task_id, run_id="agent-1", error="", fmt="text")
3237	with _patch_repo(repo):
3238	run_fail_task(args)
3239	out = capsys.readouterr().out
3240	assert "Error:" not in out
3241
3242	def test_text_shows_elapsed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3243	repo = _make_repo(tmp_path)
3244	t = self._setup(repo)
3245	args = _fail_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
3246	with _patch_repo(repo):
3247	run_fail_task(args)
3248	out = capsys.readouterr().out
3249	assert "s)" in out
3250
3251	def test_ansi_injection_in_error_stripped(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3252	repo = _make_repo(tmp_path)
3253	t = self._setup(repo)
3254	evil_err = "error\x1b[31mRED\x1b[0m"
3255	args = _fail_ns(task_id=t.task_id, run_id="agent-1",
3256	error=evil_err, fmt="text")
3257	with _patch_repo(repo):
3258	run_fail_task(args)
3259	out = capsys.readouterr().out
3260	assert "\x1b" not in out
3261
3262	def test_ansi_injection_in_title_stripped(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3263	evil_title = "task\x1b[1mBOLD\x1b[0m"
3264	repo = _make_repo(tmp_path)
3265	t = create_task(repo, evil_title)
3266	claim_next_task(repo, "agent-1")
3267	args = _fail_ns(task_id=t.task_id, run_id="agent-1", fmt="text")
3268	with _patch_repo(repo):
3269	run_fail_task(args)
3270	out = capsys.readouterr().out
3271	assert "\x1b" not in out
3272
3273
3274	class TestFailTaskIntegration:
3275	"""Full lifecycle integration tests for run_fail_task."""
3276
3277	def test_failed_status_persisted_to_disk(self, tmp_path: pathlib.Path) -> None:
3278	repo = _make_repo(tmp_path)
3279	t = create_task(repo, "persist-fail")
3280	claim_next_task(repo, "worker")
3281	args = _fail_ns(task_id=t.task_id, run_id="worker", error="timeout")
3282	with _patch_repo(repo):
3283	run_fail_task(args)
3284	claim = load_claim(repo, t.task_id)
3285	assert claim is not None
3286	assert claim.status == "failed"
3287	assert claim.error == "timeout"
3288
3289	def test_error_persisted_to_disk(self, tmp_path: pathlib.Path) -> None:
3290	repo = _make_repo(tmp_path)
3291	t = create_task(repo, "error-persist")
3292	claim_next_task(repo, "worker")
3293	args = _fail_ns(task_id=t.task_id, run_id="worker",
3294	error="OOM at step 3: allocated 16 GiB")
3295	with _patch_repo(repo):
3296	run_fail_task(args)
3297	claim = load_claim(repo, t.task_id)
3298	assert claim is not None
3299	assert "OOM at step 3" in claim.error
3300
3301	def test_double_fail_exits_1(self, tmp_path: pathlib.Path) -> None:
3302	repo = _make_repo(tmp_path)
3303	t = create_task(repo, "double-fail")
3304	claim_next_task(repo, "worker")
3305	args = _fail_ns(task_id=t.task_id, run_id="worker")
3306	with _patch_repo(repo):
3307	run_fail_task(args)
3308	with _patch_repo(repo):
3309	with pytest.raises(SystemExit) as exc:
3310	run_fail_task(args)
3311	assert exc.value.code == 1
3312
3313	def test_fail_unclaimed_task_exits_1(self, tmp_path: pathlib.Path) -> None:
3314	repo = _make_repo(tmp_path)
3315	t = create_task(repo, "unclaimed")
3316	args = _fail_ns(task_id=t.task_id, run_id="worker")
3317	with _patch_repo(repo):
3318	with pytest.raises(SystemExit) as exc:
3319	run_fail_task(args)
3320	assert exc.value.code == 1
3321
3322	def test_enqueue_claim_fail_full_cycle(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3323	"""Full CLI round-trip: enqueue → claim → fail."""
3324	repo = _make_repo(tmp_path)
3325	eq_args = _enqueue_ns(title="e2e fail task", queue="default")
3326	with _patch_repo(repo):
3327	run_enqueue(eq_args)
3328	task_id = json.loads(capsys.readouterr().out)["task_id"]
3329
3330	cl_args = _claim_ns(run_id="e2e-worker")
3331	with _patch_repo(repo):
3332	run_claim(cl_args)
3333	capsys.readouterr()
3334
3335	fa_args = _fail_ns(task_id=task_id, run_id="e2e-worker",
3336	error="dependency unavailable")
3337	with _patch_repo(repo):
3338	run_fail_task(fa_args)
3339	out = json.loads(capsys.readouterr().out)
3340	assert out["status"] == "failed"
3341	assert "dependency unavailable" in out["error"]
3342
3343	def test_unicode_error_message(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3344	repo = _make_repo(tmp_path)
3345	t = create_task(repo, "unicode-fail")
3346	claim_next_task(repo, "agent-1")
3347	args = _fail_ns(task_id=t.task_id, run_id="agent-1",
3348	error="échec: fichier introuvable — erreur 404 🚫")
3349	with _patch_repo(repo):
3350	run_fail_task(args)
3351	out = json.loads(capsys.readouterr().out)
3352	assert out["status"] == "failed"
3353	assert "échec" in out["error"]
3354
3355
3356	class TestFailTaskStress:
3357	"""Concurrency and throughput stress tests for run_fail_task."""
3358
3359	def test_20_agents_each_claim_and_fail_unique_task(self, tmp_path: pathlib.Path) -> None:
3360	"""20 concurrent agents each claim+fail a unique task with no conflicts."""
3361	import concurrent.futures
3362	repo = _make_repo(tmp_path)
3363	tasks = [create_task(repo, f"stress-{i}") for i in range(20)]
3364
3365	failed_ids: set[str] = set()
3366	lock = threading.Lock()
3367	errors: list[str] = []
3368
3369	def claim_and_fail(task: "TaskRecord") -> None:
3370	run_id = f"agent-{task.task_id[:8]}"
3371	result = claim_next_task(repo, run_id, queue=task.queue)
3372	if result is None:
3373	errors.append(f"no task for {run_id}")
3374	return
3375	claimed_task, _ = result
3376	try:
3377	fail_task(repo, claimed_task.task_id, run_id, error="stress test")
3378	except Exception as exc: # noqa: BLE001
3379	errors.append(str(exc))
3380	return
3381	with lock:
3382	failed_ids.add(claimed_task.task_id)
3383
3384	with concurrent.futures.ThreadPoolExecutor(max_workers=20) as pool:
3385	list(pool.map(claim_and_fail, tasks))
3386
3387	assert not errors, f"errors: {errors}"
3388	assert len(failed_ids) == 20, f"only {len(failed_ids)}/20 failed"
3389
3390	def test_100_sequential_fails_under_15s(self, tmp_path: pathlib.Path) -> None:
3391	repo = _make_repo(tmp_path)
3392	start = time.monotonic()
3393	for i in range(100):
3394	t = create_task(repo, f"seq-fail-{i}")
3395	claim_next_task(repo, "batch-failer")
3396	fail_task(repo, t.task_id, "batch-failer", error=f"error {i}")
3397	elapsed = time.monotonic() - start
3398	assert elapsed < 15.0, f"100 sequential fails took {elapsed:.1f}s"
3399
3400	def test_fail_via_run_fail_task_100_sequential(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3401	repo = _make_repo(tmp_path)
3402	start = time.monotonic()
3403	for i in range(100):
3404	t = create_task(repo, f"cli-fail-{i}")
3405	claim_next_task(repo, "cli-failer")
3406	args = _fail_ns(task_id=t.task_id, run_id="cli-failer",
3407	error=f"error {i}")
3408	with _patch_repo(repo):
3409	run_fail_task(args)
3410	capsys.readouterr()
3411	elapsed = time.monotonic() - start
3412	assert elapsed < 15.0, f"100 CLI fails took {elapsed:.1f}s"
3413
3414
3415	class TestCliCancelTask:
3416	"""run_cancel_task: pending cancel, force cancel, error paths."""
3417
3418	def test_cancel_pending_task_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3419	repo = _make_repo(tmp_path)
3420	t = create_task(repo, "Unwanted task")
3421	args = _namespace(task_id=t.task_id, run_id="orchestrator", force=False, fmt="json")
3422	with _patch_repo(repo):
3423	run_cancel_task(args)
3424	out = json.loads(capsys.readouterr().out)
3425	assert out["status"] == "cancelled"
3426
3427	def test_cancel_claimed_by_claimer(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3428	repo = _make_repo(tmp_path)
3429	t = create_task(repo, "My task")
3430	claim_next_task(repo, "agent-1")
3431	args = _namespace(task_id=t.task_id, run_id="agent-1", force=False, fmt="json")
3432	with _patch_repo(repo):
3433	run_cancel_task(args)
3434	out = json.loads(capsys.readouterr().out)
3435	assert out["status"] == "cancelled"
3436
3437	def test_cancel_force_different_agent(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3438	repo = _make_repo(tmp_path)
3439	t = create_task(repo, "Stolen task")
3440	claim_next_task(repo, "agent-1")
3441	args = _namespace(task_id=t.task_id, run_id="orchestrator", force=True, fmt="json")
3442	with _patch_repo(repo):
3443	run_cancel_task(args)
3444	out = json.loads(capsys.readouterr().out)
3445	assert out["status"] == "cancelled"
3446
3447	def test_cancel_wrong_claimer_no_force_exits_1(self, tmp_path: pathlib.Path) -> None:
3448	repo = _make_repo(tmp_path)
3449	t = create_task(repo, "Someone else's task")
3450	claim_next_task(repo, "agent-1")
3451	args = _namespace(task_id=t.task_id, run_id="agent-2", force=False, fmt="json")
3452	with _patch_repo(repo):
3453	with pytest.raises(SystemExit) as exc:
3454	run_cancel_task(args)
3455	assert exc.value.code == 1
3456
3457	def test_cancel_text_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3458	repo = _make_repo(tmp_path)
3459	t = create_task(repo, "To cancel")
3460	args = _namespace(task_id=t.task_id, run_id="orchestrator", force=False, fmt="text")
3461	with _patch_repo(repo):
3462	run_cancel_task(args)
3463	out = capsys.readouterr().out
3464	assert "cancelled" in out.lower()
3465
3466
3467	# ── cancel-task hardening ─────────────────────────────────────────────────────
3468
3469
3470	def _cancel_ns(**kwargs: MsgpackValue) -> argparse.Namespace:
3471	"""Build a Namespace with cancel-task-appropriate defaults."""
3472	defaults = {
3473	"fmt": "json",
3474	"run_id": "orchestrator",
3475	"task_id": VALID_UUID,
3476	"force": False,
3477	}
3478	defaults.update(kwargs)
3479	return argparse.Namespace(**defaults)
3480
3481
3482	class TestCancelTaskInputValidation:
3483	"""All cancel-task validation fires before require_repo() and exits 1."""
3484
3485	def test_run_id_too_long_exits_1(self, tmp_path: pathlib.Path) -> None:
3486	repo = _make_repo(tmp_path)
3487	args = _cancel_ns(task_id=VALID_UUID, run_id="x" * 257)
3488	with _patch_repo(repo):
3489	with pytest.raises(SystemExit) as exc:
3490	run_cancel_task(args)
3491	assert exc.value.code == ExitCode.USER_ERROR
3492
3493	def test_run_id_at_max_length_passes_validation(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3494	repo = _make_repo(tmp_path)
3495	t = create_task(repo, "task")
3496	# run_id exactly at limit — passes length check; pending task cancels fine
3497	args = _cancel_ns(task_id=t.task_id, run_id="c" * 256)
3498	with _patch_repo(repo):
3499	run_cancel_task(args)
3500	out = json.loads(capsys.readouterr().out)
3501	assert out["status"] == "cancelled"
3502
3503	def test_invalid_task_id_exits_1(self, tmp_path: pathlib.Path) -> None:
3504	repo = _make_repo(tmp_path)
3505	args = _cancel_ns(task_id="not-a-uuid")
3506	with _patch_repo(repo):
3507	with pytest.raises(SystemExit) as exc:
3508	run_cancel_task(args)
3509	assert exc.value.code == ExitCode.USER_ERROR
3510
3511	def test_path_traversal_task_id_exits_1(self, tmp_path: pathlib.Path) -> None:
3512	repo = _make_repo(tmp_path)
3513	args = _cancel_ns(task_id="../../etc/passwd")
3514	with _patch_repo(repo):
3515	with pytest.raises(SystemExit) as exc:
3516	run_cancel_task(args)
3517	assert exc.value.code == ExitCode.USER_ERROR
3518
3519	def test_null_byte_task_id_exits_1(self, tmp_path: pathlib.Path) -> None:
3520	repo = _make_repo(tmp_path)
3521	args = _cancel_ns(task_id="12345678-1234-4abc-8abc-123456789\x00ab")
3522	with _patch_repo(repo):
3523	with pytest.raises(SystemExit) as exc:
3524	run_cancel_task(args)
3525	assert exc.value.code == ExitCode.USER_ERROR
3526
3527	def test_task_id_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
3528	call_count = {"n": 0}
3529
3530	def counting_require_repo() -> pathlib.Path:
3531	call_count["n"] += 1
3532	raise RuntimeError("should not be called")
3533
3534	args = _cancel_ns(task_id="BADUUID")
3535	with patch("muse.cli.commands.task_queue.require_repo", counting_require_repo):
3536	with pytest.raises(SystemExit):
3537	run_cancel_task(args)
3538	assert call_count["n"] == 0, "require_repo called before task_id validation"
3539
3540	def test_run_id_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
3541	call_count = {"n": 0}
3542
3543	def counting_require_repo() -> pathlib.Path:
3544	call_count["n"] += 1
3545	raise RuntimeError("should not be called")
3546
3547	args = _cancel_ns(task_id=VALID_UUID, run_id="r" * 300)
3548	with patch("muse.cli.commands.task_queue.require_repo", counting_require_repo):
3549	with pytest.raises(SystemExit):
3550	run_cancel_task(args)
3551	assert call_count["n"] == 0
3552
3553	def test_json_error_shape_bad_task_id(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3554	repo = _make_repo(tmp_path)
3555	args = _cancel_ns(task_id="bad-id", fmt="json")
3556	with _patch_repo(repo):
3557	with pytest.raises(SystemExit):
3558	run_cancel_task(args)
3559	out = json.loads(capsys.readouterr().out)
3560	assert "error" in out
3561	assert out["status"] == "bad_task_id"
3562
3563	def test_json_error_shape_bad_run_id(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3564	repo = _make_repo(tmp_path)
3565	args = _cancel_ns(task_id=VALID_UUID, run_id="x" * 300, fmt="json")
3566	with _patch_repo(repo):
3567	with pytest.raises(SystemExit):
3568	run_cancel_task(args)
3569	out = json.loads(capsys.readouterr().out)
3570	assert "error" in out
3571	assert out["status"] == "bad_args"
3572
3573	def test_text_error_goes_to_stderr(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3574	repo = _make_repo(tmp_path)
3575	args = _cancel_ns(task_id="BADUUID", fmt="text")
3576	with _patch_repo(repo):
3577	with pytest.raises(SystemExit):
3578	run_cancel_task(args)
3579	captured = capsys.readouterr()
3580	assert captured.out == ""
3581	assert "❌" in captured.err
3582
3583	def test_json_error_goes_to_stdout_not_stderr(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3584	repo = _make_repo(tmp_path)
3585	args = _cancel_ns(task_id="BADUUID", fmt="json")
3586	with _patch_repo(repo):
3587	with pytest.raises(SystemExit):
3588	run_cancel_task(args)
3589	captured = capsys.readouterr()
3590	assert captured.err == ""
3591	out = json.loads(captured.out)
3592	assert "error" in out
3593
3594	def test_missing_task_error_shape(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3595	repo = _make_repo(tmp_path)
3596	# Valid UUID but task does not exist
3597	args = _cancel_ns(task_id=VALID_UUID, fmt="json")
3598	with _patch_repo(repo):
3599	with pytest.raises(SystemExit) as exc:
3600	run_cancel_task(args)
3601	assert exc.value.code == 1
3602	out = json.loads(capsys.readouterr().out)
3603	assert "error" in out
3604
3605	def test_already_terminal_error_shape(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3606	repo = _make_repo(tmp_path)
3607	t = create_task(repo, "terminal")
3608	claim_next_task(repo, "agent-1")
3609	complete_task(repo, t.task_id, "agent-1")
3610	args = _cancel_ns(task_id=t.task_id, run_id="agent-1", fmt="json")
3611	with _patch_repo(repo):
3612	with pytest.raises(SystemExit) as exc:
3613	run_cancel_task(args)
3614	assert exc.value.code == 1
3615	out = json.loads(capsys.readouterr().out)
3616	assert "error" in out
3617
3618	def test_wrong_claimer_no_force_error_shape(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3619	repo = _make_repo(tmp_path)
3620	t = create_task(repo, "owned")
3621	claim_next_task(repo, "agent-1")
3622	args = _cancel_ns(task_id=t.task_id, run_id="agent-2",
3623	force=False, fmt="json")
3624	with _patch_repo(repo):
3625	with pytest.raises(SystemExit) as exc:
3626	run_cancel_task(args)
3627	assert exc.value.code == 1
3628	out = json.loads(capsys.readouterr().out)
3629	assert "error" in out
3630
3631
3632	class TestCancelTaskJsonOutput:
3633	"""run_cancel_task JSON output shape and compactness."""
3634
3635	def test_json_is_compact(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3636	repo = _make_repo(tmp_path)
3637	t = create_task(repo, "compact task")
3638	args = _cancel_ns(task_id=t.task_id)
3639	with _patch_repo(repo):
3640	run_cancel_task(args)
3641	raw = capsys.readouterr().out.strip()
3642	assert "\n" not in raw, "JSON output must be single line (compact)"
3643
3644	def test_json_has_required_keys(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3645	repo = _make_repo(tmp_path)
3646	t = create_task(repo, "keys task")
3647	args = _cancel_ns(task_id=t.task_id)
3648	with _patch_repo(repo):
3649	run_cancel_task(args)
3650	out = json.loads(capsys.readouterr().out)
3651	for key in ("schema_version", "task_id", "claimer_run_id", "status",
3652	"claimed_at", "expires_at", "error", "elapsed_seconds"):
3653	assert key in out, f"missing key: {key}"
3654
3655	def test_json_status_is_cancelled(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3656	repo = _make_repo(tmp_path)
3657	t = create_task(repo, "status task")
3658	args = _cancel_ns(task_id=t.task_id)
3659	with _patch_repo(repo):
3660	run_cancel_task(args)
3661	out = json.loads(capsys.readouterr().out)
3662	assert out["status"] == "cancelled"
3663
3664	def test_json_elapsed_is_float(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3665	repo = _make_repo(tmp_path)
3666	t = create_task(repo, "elapsed task")
3667	args = _cancel_ns(task_id=t.task_id)
3668	with _patch_repo(repo):
3669	run_cancel_task(args)
3670	out = json.loads(capsys.readouterr().out)
3671	assert isinstance(out["elapsed_seconds"], float)
3672
3673	def test_json_claimer_run_id_for_pending_is_caller(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3674	"""For a pending task, claimer_run_id is the calling agent."""
3675	repo = _make_repo(tmp_path)
3676	t = create_task(repo, "pending task")
3677	args = _cancel_ns(task_id=t.task_id, run_id="orchestrator-99")
3678	with _patch_repo(repo):
3679	run_cancel_task(args)
3680	out = json.loads(capsys.readouterr().out)
3681	assert out["claimer_run_id"] == "orchestrator-99"
3682
3683	def test_json_claimer_run_id_for_claimed_is_original_claimer(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3684	"""For a claimed task cancelled by its claimer, run_id is preserved."""
3685	repo = _make_repo(tmp_path)
3686	t = create_task(repo, "claimed task")
3687	claim_next_task(repo, "agent-xyz")
3688	args = _cancel_ns(task_id=t.task_id, run_id="agent-xyz")
3689	with _patch_repo(repo):
3690	run_cancel_task(args)
3691	out = json.loads(capsys.readouterr().out)
3692	assert out["claimer_run_id"] == "agent-xyz"
3693
3694	def test_json_force_cancel_different_agent(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3695	repo = _make_repo(tmp_path)
3696	t = create_task(repo, "force task")
3697	claim_next_task(repo, "agent-1")
3698	args = _cancel_ns(task_id=t.task_id, run_id="orchestrator", force=True)
3699	with _patch_repo(repo):
3700	run_cancel_task(args)
3701	out = json.loads(capsys.readouterr().out)
3702	assert out["status"] == "cancelled"
3703
3704	def test_json_task_id_matches(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3705	repo = _make_repo(tmp_path)
3706	t = create_task(repo, "id task")
3707	args = _cancel_ns(task_id=t.task_id)
3708	with _patch_repo(repo):
3709	run_cancel_task(args)
3710	out = json.loads(capsys.readouterr().out)
3711	assert out["task_id"] == t.task_id
3712
3713
3714	class TestCancelTaskTextOutput:
3715	"""run_cancel_task text output content."""
3716
3717	def _setup_pending(self, repo: pathlib.Path, title: str = "Pending Task",
3718	queue: str = "default") -> TaskRecord:
3719	return create_task(repo, title, queue=queue)
3720
3721	def _setup_claimed(self, repo: pathlib.Path, title: str = "Claimed Task",
3722	queue: str = "default", claimer: str = "agent-1") -> TaskRecord:
3723	t = create_task(repo, title, queue=queue)
3724	claim_next_task(repo, claimer)
3725	return t
3726
3727	def test_text_shows_cancelled(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3728	repo = _make_repo(tmp_path)
3729	t = self._setup_pending(repo)
3730	args = _cancel_ns(task_id=t.task_id, fmt="text")
3731	with _patch_repo(repo):
3732	run_cancel_task(args)
3733	out = capsys.readouterr().out
3734	assert "cancelled" in out.lower()
3735
3736	def test_text_shows_task_title(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3737	repo = _make_repo(tmp_path)
3738	t = self._setup_pending(repo, title="Decommission old infra")
3739	args = _cancel_ns(task_id=t.task_id, fmt="text")
3740	with _patch_repo(repo):
3741	run_cancel_task(args)
3742	out = capsys.readouterr().out
3743	assert "Decommission old infra" in out
3744
3745	def test_text_shows_queue(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3746	repo = _make_repo(tmp_path)
3747	t = self._setup_pending(repo, queue="infra")
3748	args = _cancel_ns(task_id=t.task_id, fmt="text")
3749	with _patch_repo(repo):
3750	run_cancel_task(args)
3751	out = capsys.readouterr().out
3752	assert "infra" in out
3753
3754	def test_text_shows_claimer(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3755	repo = _make_repo(tmp_path)
3756	t = self._setup_pending(repo)
3757	args = _cancel_ns(task_id=t.task_id, run_id="orch-42", fmt="text")
3758	with _patch_repo(repo):
3759	run_cancel_task(args)
3760	out = capsys.readouterr().out
3761	assert "orch-42" in out
3762
3763	def test_text_shows_forced_indicator(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3764	repo = _make_repo(tmp_path)
3765	t = self._setup_claimed(repo, claimer="agent-1")
3766	args = _cancel_ns(task_id=t.task_id, run_id="orch", force=True, fmt="text")
3767	with _patch_repo(repo):
3768	run_cancel_task(args)
3769	out = capsys.readouterr().out
3770	assert "forced" in out.lower()
3771
3772	def test_text_no_forced_indicator_without_flag(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3773	repo = _make_repo(tmp_path)
3774	t = self._setup_claimed(repo, claimer="agent-1")
3775	args = _cancel_ns(task_id=t.task_id, run_id="agent-1",
3776	force=False, fmt="text")
3777	with _patch_repo(repo):
3778	run_cancel_task(args)
3779	out = capsys.readouterr().out
3780	assert "forced" not in out.lower()
3781
3782	def test_text_shows_elapsed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3783	repo = _make_repo(tmp_path)
3784	t = self._setup_pending(repo)
3785	args = _cancel_ns(task_id=t.task_id, fmt="text")
3786	with _patch_repo(repo):
3787	run_cancel_task(args)
3788	out = capsys.readouterr().out
3789	assert "s)" in out
3790
3791	def test_ansi_injection_in_title_stripped(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3792	evil_title = "task\x1b[1mBOLD\x1b[0m"
3793	repo = _make_repo(tmp_path)
3794	t = create_task(repo, evil_title)
3795	args = _cancel_ns(task_id=t.task_id, fmt="text")
3796	with _patch_repo(repo):
3797	run_cancel_task(args)
3798	out = capsys.readouterr().out
3799	assert "\x1b" not in out
3800
3801	def test_ansi_injection_in_run_id_stripped(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3802	evil_id = "orch\x1b[31mRED\x1b[0m"
3803	repo = _make_repo(tmp_path)
3804	t = create_task(repo, "task")
3805	args = _cancel_ns(task_id=t.task_id, run_id=evil_id, fmt="text")
3806	with _patch_repo(repo):
3807	run_cancel_task(args)
3808	out = capsys.readouterr().out
3809	assert "\x1b" not in out
3810
3811
3812	class TestCancelTaskIntegration:
3813	"""Full lifecycle integration tests for run_cancel_task."""
3814
3815	def test_pending_task_cancelled_status_on_disk(self, tmp_path: pathlib.Path) -> None:
3816	repo = _make_repo(tmp_path)
3817	t = create_task(repo, "pending-cancel")
3818	args = _cancel_ns(task_id=t.task_id, run_id="orch")
3819	with _patch_repo(repo):
3820	run_cancel_task(args)
3821	claim = load_claim(repo, t.task_id)
3822	assert claim is not None
3823	assert claim.status == "cancelled"
3824
3825	def test_claimed_task_cancelled_by_claimer(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3826	repo = _make_repo(tmp_path)
3827	t = create_task(repo, "claimed-cancel")
3828	claim_next_task(repo, "worker")
3829	args = _cancel_ns(task_id=t.task_id, run_id="worker")
3830	with _patch_repo(repo):
3831	run_cancel_task(args)
3832	out = json.loads(capsys.readouterr().out)
3833	assert out["status"] == "cancelled"
3834	claim = load_claim(repo, t.task_id)
3835	assert claim.status == "cancelled"
3836
3837	def test_force_cancel_different_claimer(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3838	repo = _make_repo(tmp_path)
3839	t = create_task(repo, "force-cancel")
3840	claim_next_task(repo, "agent-1")
3841	args = _cancel_ns(task_id=t.task_id, run_id="orchestrator", force=True)
3842	with _patch_repo(repo):
3843	run_cancel_task(args)
3844	out = json.loads(capsys.readouterr().out)
3845	assert out["status"] == "cancelled"
3846
3847	def test_double_cancel_exits_1(self, tmp_path: pathlib.Path) -> None:
3848	repo = _make_repo(tmp_path)
3849	t = create_task(repo, "double-cancel")
3850	args = _cancel_ns(task_id=t.task_id)
3851	with _patch_repo(repo):
3852	run_cancel_task(args)
3853	with _patch_repo(repo):
3854	with pytest.raises(SystemExit) as exc:
3855	run_cancel_task(args)
3856	assert exc.value.code == 1
3857
3858	def test_cancel_completed_task_exits_1(self, tmp_path: pathlib.Path) -> None:
3859	repo = _make_repo(tmp_path)
3860	t = create_task(repo, "completed-task")
3861	claim_next_task(repo, "agent-1")
3862	complete_task(repo, t.task_id, "agent-1")
3863	args = _cancel_ns(task_id=t.task_id, run_id="agent-1")
3864	with _patch_repo(repo):
3865	with pytest.raises(SystemExit) as exc:
3866	run_cancel_task(args)
3867	assert exc.value.code == 1
3868
3869	def test_cancel_failed_task_exits_1(self, tmp_path: pathlib.Path) -> None:
3870	repo = _make_repo(tmp_path)
3871	t = create_task(repo, "failed-task")
3872	claim_next_task(repo, "agent-1")
3873	fail_task(repo, t.task_id, "agent-1", error="boom")
3874	args = _cancel_ns(task_id=t.task_id, run_id="agent-1")
3875	with _patch_repo(repo):
3876	with pytest.raises(SystemExit) as exc:
3877	run_cancel_task(args)
3878	assert exc.value.code == 1
3879
3880	def test_enqueue_then_cancel_full_cycle(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3881	"""Full CLI round-trip: enqueue → cancel."""
3882	repo = _make_repo(tmp_path)
3883	eq_args = _enqueue_ns(title="e2e cancel", queue="default")
3884	with _patch_repo(repo):
3885	run_enqueue(eq_args)
3886	task_id = json.loads(capsys.readouterr().out)["task_id"]
3887
3888	ca_args = _cancel_ns(task_id=task_id, run_id="orch")
3889	with _patch_repo(repo):
3890	run_cancel_task(ca_args)
3891	out = json.loads(capsys.readouterr().out)
3892	assert out["status"] == "cancelled"
3893	assert out["task_id"] == task_id
3894
3895	def test_enqueue_claim_cancel_full_cycle(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3896	"""Full CLI round-trip: enqueue → claim → cancel by claimer."""
3897	repo = _make_repo(tmp_path)
3898	eq_args = _enqueue_ns(title="e2e claim-cancel", queue="default")
3899	with _patch_repo(repo):
3900	run_enqueue(eq_args)
3901	task_id = json.loads(capsys.readouterr().out)["task_id"]
3902
3903	cl_args = _claim_ns(run_id="e2e-worker")
3904	with _patch_repo(repo):
3905	run_claim(cl_args)
3906	capsys.readouterr()
3907
3908	ca_args = _cancel_ns(task_id=task_id, run_id="e2e-worker")
3909	with _patch_repo(repo):
3910	run_cancel_task(ca_args)
3911	out = json.loads(capsys.readouterr().out)
3912	assert out["status"] == "cancelled"
3913
3914
3915	class TestCancelTaskStress:
3916	"""Concurrency and throughput stress tests for run_cancel_task."""
3917
3918	def test_20_agents_each_cancel_unique_pending_task(self, tmp_path: pathlib.Path) -> None:
3919	"""20 concurrent orchestrators each cancel a distinct pending task."""
3920	import concurrent.futures
3921	repo = _make_repo(tmp_path)
3922	tasks = [create_task(repo, f"cancel-stress-{i}") for i in range(20)]
3923
3924	cancelled_ids: set[str] = set()
3925	lock = threading.Lock()
3926	errors: list[str] = []
3927
3928	def do_cancel(t: "TaskRecord") -> None:
3929	try:
3930	claim = cancel_task(repo, t.task_id, f"orch-{t.task_id[:8]}")
3931	except Exception as exc: # noqa: BLE001
3932	errors.append(str(exc))
3933	return
3934	with lock:
3935	cancelled_ids.add(claim.task_id)
3936
3937	with concurrent.futures.ThreadPoolExecutor(max_workers=20) as pool:
3938	list(pool.map(do_cancel, tasks))
3939
3940	assert not errors, f"errors: {errors}"
3941	assert len(cancelled_ids) == 20, f"only {len(cancelled_ids)}/20 cancelled"
3942
3943	def test_100_sequential_cancels_under_15s(self, tmp_path: pathlib.Path) -> None:
3944	repo = _make_repo(tmp_path)
3945	start = time.monotonic()
3946	for i in range(100):
3947	t = create_task(repo, f"seq-cancel-{i}")
3948	cancel_task(repo, t.task_id, "orch")
3949	elapsed = time.monotonic() - start
3950	assert elapsed < 15.0, f"100 sequential cancels took {elapsed:.1f}s"
3951
3952	def test_cancel_via_run_cancel_task_100_sequential(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3953	repo = _make_repo(tmp_path)
3954	start = time.monotonic()
3955	for i in range(100):
3956	t = create_task(repo, f"cli-cancel-{i}")
3957	args = _cancel_ns(task_id=t.task_id, run_id="orch")
3958	with _patch_repo(repo):
3959	run_cancel_task(args)
3960	capsys.readouterr()
3961	elapsed = time.monotonic() - start
3962	assert elapsed < 15.0, f"100 CLI cancels took {elapsed:.1f}s"
3963
3964
3965	class TestCliTasks:
3966	"""run_tasks: listing, filtering, JSON/text output."""
3967
3968	def _setup_mixed(self, repo: pathlib.Path) -> tuple[TaskRecord, TaskRecord, TaskRecord]:
3969	"""Create tasks in various states."""
3970	t1 = create_task(repo, "Pending task", queue="q1", priority=1)
3971	t2 = create_task(repo, "Claimed task", queue="q2", priority=5)
3972	t3 = create_task(repo, "Done task", queue="q1", priority=3)
3973	claim_next_task(repo, "agent-2", queue="q2")
3974	claim_next_task(repo, "agent-3", queue="q1")
3975	complete_task(repo, t3.task_id, "agent-3")
3976	return t1, t2, t3
3977
3978	def test_lists_all_tasks_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3979	repo = _make_repo(tmp_path)
3980	self._setup_mixed(repo)
3981	args = _namespace(fmt="json", status=None, queue=None, run_id=None)
3982	with _patch_repo(repo):
3983	run_tasks(args)
3984	out = json.loads(capsys.readouterr().out)
3985	assert out["total"] == 3
3986	assert len(out["items"]) == 3
3987
3988	def test_filter_by_status_pending(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3989	repo = _make_repo(tmp_path)
3990	self._setup_mixed(repo)
3991	args = _namespace(fmt="json", status="pending", queue=None, run_id=None)
3992	with _patch_repo(repo):
3993	run_tasks(args)
3994	out = json.loads(capsys.readouterr().out)
3995	for item in out["items"]:
3996	assert item["status"] == "pending"
3997
3998	def test_filter_by_queue(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
3999	repo = _make_repo(tmp_path)
4000	self._setup_mixed(repo)
4001	args = _namespace(fmt="json", status=None, queue="q1", run_id=None)
4002	with _patch_repo(repo):
4003	run_tasks(args)
4004	out = json.loads(capsys.readouterr().out)
4005	for item in out["items"]:
4006	assert item["queue"] == "q1"
4007
4008	def test_filter_by_run_id(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4009	repo = _make_repo(tmp_path)
4010	self._setup_mixed(repo)
4011	args = _namespace(fmt="json", status=None, queue=None, run_id="agent-2")
4012	with _patch_repo(repo):
4013	run_tasks(args)
4014	out = json.loads(capsys.readouterr().out)
4015	for item in out["items"]:
4016	assert item["claimer_run_id"] == "agent-2"
4017
4018	def test_items_sorted_by_priority_desc(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4019	repo = _make_repo(tmp_path)
4020	with _freeze(_EPOCH):
4021	create_task(repo, "Low", priority=0)
4022	create_task(repo, "High", priority=10)
4023	create_task(repo, "Mid", priority=5)
4024	args = _namespace(fmt="json", status=None, queue=None, run_id=None)
4025	with _patch_repo(repo):
4026	run_tasks(args)
4027	out = json.loads(capsys.readouterr().out)
4028	priorities = [i["priority"] for i in out["items"]]
4029	assert priorities == sorted(priorities, reverse=True)
4030
4031	def test_text_output_no_crash(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4032	repo = _make_repo(tmp_path)
4033	create_task(repo, "Text test task")
4034	args = _namespace(fmt="text", status=None, queue=None, run_id=None)
4035	with _patch_repo(repo):
4036	run_tasks(args)
4037	out = capsys.readouterr().out
4038	assert "Task queue" in out
4039
4040	def test_empty_queue_text_output(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4041	repo = _make_repo(tmp_path)
4042	args = _namespace(fmt="text", status=None, queue=None, run_id=None)
4043	with _patch_repo(repo):
4044	run_tasks(args)
4045	out = capsys.readouterr().out
4046	assert "0 task" in out
4047
4048	def test_status_counts_in_json(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4049	repo = _make_repo(tmp_path)
4050	create_task(repo, "Pending")
4051	args = _namespace(fmt="json", status=None, queue=None, run_id=None)
4052	with _patch_repo(repo):
4053	run_tasks(args)
4054	out = json.loads(capsys.readouterr().out)
4055	assert "pending" in out
4056	assert out["pending"] == 1
4057
4058	def test_ansi_in_title_not_printed_raw(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4059	"""ANSI escape sequences in task titles must be sanitized in text output."""
4060	repo = _make_repo(tmp_path)
4061	create_task(repo, "\x1b[31mRED\x1b[0m")
4062	args = _namespace(fmt="text", status=None, queue=None, run_id=None)
4063	with _patch_repo(repo):
4064	run_tasks(args)
4065	out = capsys.readouterr().out
4066	# Raw ESC byte must not appear in text output
4067	assert "\x1b" not in out
4068
4069
4070	# ── tasks hardening ───────────────────────────────────────────────────────────
4071
4072	from muse.cli.commands.task_queue import _MAX_LIMIT
4073	from muse.core._types import Manifest
4074
4075
4076	def _tasks_ns(**kwargs: MsgpackValue) -> argparse.Namespace:
4077	"""Build a Namespace with tasks-appropriate defaults."""
4078	defaults = {
4079	"fmt": "json",
4080	"status": None,
4081	"queue": None,
4082	"run_id": None,
4083	"limit": 200,
4084	}
4085	defaults.update(kwargs)
4086	return argparse.Namespace(**defaults)
4087
4088
4089	class TestTasksInputValidation:
4090	"""All tasks validation fires before require_repo() and exits 1."""
4091
4092	def test_invalid_queue_name_exits_1(self, tmp_path: pathlib.Path) -> None:
4093	repo = _make_repo(tmp_path)
4094	args = _tasks_ns(queue="bad queue!")
4095	with _patch_repo(repo):
4096	with pytest.raises(SystemExit) as exc:
4097	run_tasks(args)
4098	assert exc.value.code == ExitCode.USER_ERROR
4099
4100	def test_queue_with_slash_exits_1(self, tmp_path: pathlib.Path) -> None:
4101	repo = _make_repo(tmp_path)
4102	args = _tasks_ns(queue="../../etc")
4103	with _patch_repo(repo):
4104	with pytest.raises(SystemExit) as exc:
4105	run_tasks(args)
4106	assert exc.value.code == ExitCode.USER_ERROR
4107
4108	def test_queue_with_ansi_exits_1(self, tmp_path: pathlib.Path) -> None:
4109	repo = _make_repo(tmp_path)
4110	args = _tasks_ns(queue="q\x1b[31m")
4111	with _patch_repo(repo):
4112	with pytest.raises(SystemExit) as exc:
4113	run_tasks(args)
4114	assert exc.value.code == ExitCode.USER_ERROR
4115
4116	def test_valid_queue_name_passes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4117	repo = _make_repo(tmp_path)
4118	create_task(repo, "task", queue="billing-v2")
4119	args = _tasks_ns(queue="billing-v2")
4120	with _patch_repo(repo):
4121	run_tasks(args)
4122	out = json.loads(capsys.readouterr().out)
4123	assert out["total"] >= 0 # no exception
4124
4125	def test_run_id_too_long_exits_1(self, tmp_path: pathlib.Path) -> None:
4126	repo = _make_repo(tmp_path)
4127	args = _tasks_ns(run_id="x" * 257)
4128	with _patch_repo(repo):
4129	with pytest.raises(SystemExit) as exc:
4130	run_tasks(args)
4131	assert exc.value.code == ExitCode.USER_ERROR
4132
4133	def test_run_id_at_max_length_passes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4134	repo = _make_repo(tmp_path)
4135	args = _tasks_ns(run_id="r" * 256)
4136	with _patch_repo(repo):
4137	run_tasks(args)
4138	out = json.loads(capsys.readouterr().out)
4139	assert "items" in out
4140
4141	def test_limit_zero_exits_1(self, tmp_path: pathlib.Path) -> None:
4142	repo = _make_repo(tmp_path)
4143	args = _tasks_ns(limit=0)
4144	with _patch_repo(repo):
4145	with pytest.raises(SystemExit) as exc:
4146	run_tasks(args)
4147	assert exc.value.code == ExitCode.USER_ERROR
4148
4149	def test_limit_negative_exits_1(self, tmp_path: pathlib.Path) -> None:
4150	repo = _make_repo(tmp_path)
4151	args = _tasks_ns(limit=-1)
4152	with _patch_repo(repo):
4153	with pytest.raises(SystemExit) as exc:
4154	run_tasks(args)
4155	assert exc.value.code == ExitCode.USER_ERROR
4156
4157	def test_limit_over_max_exits_1(self, tmp_path: pathlib.Path) -> None:
4158	repo = _make_repo(tmp_path)
4159	args = _tasks_ns(limit=_MAX_LIMIT + 1)
4160	with _patch_repo(repo):
4161	with pytest.raises(SystemExit) as exc:
4162	run_tasks(args)
4163	assert exc.value.code == ExitCode.USER_ERROR
4164
4165	def test_limit_at_max_passes(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4166	repo = _make_repo(tmp_path)
4167	args = _tasks_ns(limit=_MAX_LIMIT)
4168	with _patch_repo(repo):
4169	run_tasks(args)
4170	out = json.loads(capsys.readouterr().out)
4171	assert "items" in out
4172
4173	def test_queue_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
4174	call_count = {"n": 0}
4175
4176	def counting_require_repo() -> pathlib.Path:
4177	call_count["n"] += 1
4178	raise RuntimeError("should not be called")
4179
4180	args = _tasks_ns(queue="bad queue!")
4181	with patch("muse.cli.commands.task_queue.require_repo", counting_require_repo):
4182	with pytest.raises(SystemExit):
4183	run_tasks(args)
4184	assert call_count["n"] == 0
4185
4186	def test_run_id_validation_fires_before_repo_lookup(self, tmp_path: pathlib.Path) -> None:
4187	call_count = {"n": 0}
4188
4189	def counting_require_repo() -> pathlib.Path:
4190	call_count["n"] += 1
4191	raise RuntimeError("should not be called")
4192
4193	args = _tasks_ns(run_id="r" * 300)
4194	with patch("muse.cli.commands.task_queue.require_repo", counting_require_repo):
4195	with pytest.raises(SystemExit):
4196	run_tasks(args)
4197	assert call_count["n"] == 0
4198
4199	def test_json_error_shape_bad_queue(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4200	repo = _make_repo(tmp_path)
4201	args = _tasks_ns(queue="bad queue!", fmt="json")
4202	with _patch_repo(repo):
4203	with pytest.raises(SystemExit):
4204	run_tasks(args)
4205	out = json.loads(capsys.readouterr().out)
4206	assert "error" in out
4207	assert out["status"] == "bad_queue"
4208
4209	def test_json_error_shape_bad_limit(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4210	repo = _make_repo(tmp_path)
4211	args = _tasks_ns(limit=0, fmt="json")
4212	with _patch_repo(repo):
4213	with pytest.raises(SystemExit):
4214	run_tasks(args)
4215	out = json.loads(capsys.readouterr().out)
4216	assert "error" in out
4217	assert out["status"] == "bad_args"
4218
4219	def test_text_error_goes_to_stderr(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4220	repo = _make_repo(tmp_path)
4221	args = _tasks_ns(queue="bad queue!", fmt="text")
4222	with _patch_repo(repo):
4223	with pytest.raises(SystemExit):
4224	run_tasks(args)
4225	captured = capsys.readouterr()
4226	assert captured.out == ""
4227	assert "❌" in captured.err
4228
4229	def test_json_error_goes_to_stdout_not_stderr(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4230	repo = _make_repo(tmp_path)
4231	args = _tasks_ns(queue="bad queue!", fmt="json")
4232	with _patch_repo(repo):
4233	with pytest.raises(SystemExit):
4234	run_tasks(args)
4235	captured = capsys.readouterr()
4236	assert captured.err == ""
4237	out = json.loads(captured.out)
4238	assert "error" in out
4239
4240
4241	class TestTasksJsonOutput:
4242	"""run_tasks JSON output shape, compactness, and field completeness."""
4243
4244	def _setup(self, repo: pathlib.Path) -> tuple[TaskRecord, TaskRecord, TaskRecord]:
4245	t1 = create_task(repo, "Pending", queue="q1", priority=1)
4246	t2 = create_task(repo, "Claimed", queue="q2", priority=5)
4247	t3 = create_task(repo, "Done", queue="q1", priority=3)
4248	claim_next_task(repo, "worker-a", queue="q2")
4249	claim_next_task(repo, "worker-b", queue="q1")
4250	complete_task(repo, t3.task_id, "worker-b")
4251	return t1, t2, t3
4252
4253	def test_json_is_compact(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4254	repo = _make_repo(tmp_path)
4255	self._setup(repo)
4256	args = _tasks_ns()
4257	with _patch_repo(repo):
4258	run_tasks(args)
4259	raw = capsys.readouterr().out.strip()
4260	assert "\n" not in raw, "JSON must be single line (compact)"
4261
4262	def test_json_has_top_level_keys(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4263	repo = _make_repo(tmp_path)
4264	self._setup(repo)
4265	args = _tasks_ns()
4266	with _patch_repo(repo):
4267	run_tasks(args)
4268	out = json.loads(capsys.readouterr().out)
4269	for key in ("schema_version", "total", "pending", "claimed", "timed_out",
4270	"completed", "failed", "cancelled", "limit", "truncated",
4271	"items", "elapsed_seconds"):
4272	assert key in out, f"missing top-level key: {key}"
4273
4274	def test_items_have_new_fields(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4275	repo = _make_repo(tmp_path)
4276	create_task(repo, "field-test", queue="default")
4277	args = _tasks_ns()
4278	with _patch_repo(repo):
4279	run_tasks(args)
4280	out = json.loads(capsys.readouterr().out)
4281	assert len(out["items"]) == 1
4282	item = out["items"][0]
4283	for field in ("created_by", "ttl_seconds", "expires_at"):
4284	assert field in item, f"missing item field: {field}"
4285
4286	def test_expires_at_null_for_pending_task(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4287	repo = _make_repo(tmp_path)
4288	create_task(repo, "pending-task")
4289	args = _tasks_ns()
4290	with _patch_repo(repo):
4291	run_tasks(args)
4292	out = json.loads(capsys.readouterr().out)
4293	item = out["items"][0]
4294	assert item["expires_at"] is None
4295
4296	def test_expires_at_populated_for_claimed_task(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4297	repo = _make_repo(tmp_path)
4298	create_task(repo, "claimed-task")
4299	claim_next_task(repo, "worker")
4300	args = _tasks_ns()
4301	with _patch_repo(repo):
4302	run_tasks(args)
4303	out = json.loads(capsys.readouterr().out)
4304	item = out["items"][0]
4305	assert item["expires_at"] is not None
4306	# Should be a parseable ISO 8601 datetime
4307	import datetime as _dt
4308	_dt.datetime.fromisoformat(item["expires_at"])
4309
4310	def test_status_counts_reflect_full_queue_when_filtered(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4311	"""Counts always reflect ALL tasks, not just the filtered set."""
4312	repo = _make_repo(tmp_path)
4313	self._setup(repo)
4314	# Filter to only q1 items, but total/counts should still be 3
4315	args = _tasks_ns(queue="q1")
4316	with _patch_repo(repo):
4317	run_tasks(args)
4318	out = json.loads(capsys.readouterr().out)
4319	assert out["total"] == 3
4320	assert len(out["items"]) == 2 # only q1 items
4321
4322	def test_limit_truncates_items(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4323	repo = _make_repo(tmp_path)
4324	for i in range(10):
4325	create_task(repo, f"task-{i}")
4326	args = _tasks_ns(limit=3)
4327	with _patch_repo(repo):
4328	run_tasks(args)
4329	out = json.loads(capsys.readouterr().out)
4330	assert len(out["items"]) == 3
4331	assert out["truncated"] is True
4332	assert out["limit"] == 3
4333	assert out["total"] == 10 # full count still correct
4334
4335	def test_no_truncation_when_within_limit(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4336	repo = _make_repo(tmp_path)
4337	for i in range(5):
4338	create_task(repo, f"task-{i}")
4339	args = _tasks_ns(limit=10)
4340	with _patch_repo(repo):
4341	run_tasks(args)
4342	out = json.loads(capsys.readouterr().out)
4343	assert len(out["items"]) == 5
4344	assert out["truncated"] is False
4345
4346	def test_elapsed_is_float(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4347	repo = _make_repo(tmp_path)
4348	args = _tasks_ns()
4349	with _patch_repo(repo):
4350	run_tasks(args)
4351	out = json.loads(capsys.readouterr().out)
4352	assert isinstance(out["elapsed_seconds"], float)
4353
4354	def test_items_sorted_priority_desc(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4355	repo = _make_repo(tmp_path)
4356	with _freeze(_EPOCH):
4357	create_task(repo, "Low", priority=0)
4358	create_task(repo, "High", priority=10)
4359	create_task(repo, "Mid", priority=5)
4360	args = _tasks_ns()
4361	with _patch_repo(repo):
4362	run_tasks(args)
4363	out = json.loads(capsys.readouterr().out)
4364	priorities = [i["priority"] for i in out["items"]]
4365	assert priorities == sorted(priorities, reverse=True)
4366
4367	def test_limit_applies_after_sort(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4368	"""--limit=1 should return the highest-priority task."""
4369	repo = _make_repo(tmp_path)
4370	with _freeze(_EPOCH):
4371	create_task(repo, "Low", priority=0)
4372	create_task(repo, "High", priority=99)
4373	args = _tasks_ns(limit=1)
4374	with _patch_repo(repo):
4375	run_tasks(args)
4376	out = json.loads(capsys.readouterr().out)
4377	assert len(out["items"]) == 1
4378	assert out["items"][0]["priority"] == 99
4379
4380
4381	class TestTasksTextOutput:
4382	"""run_tasks text output content and formatting."""
4383
4384	def test_text_shows_task_queue_header(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4385	repo = _make_repo(tmp_path)
4386	args = _tasks_ns(fmt="text")
4387	with _patch_repo(repo):
4388	run_tasks(args)
4389	out = capsys.readouterr().out
4390	assert "Task queue" in out
4391
4392	def test_text_shows_status_counts(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4393	repo = _make_repo(tmp_path)
4394	create_task(repo, "pending")
4395	args = _tasks_ns(fmt="text")
4396	with _patch_repo(repo):
4397	run_tasks(args)
4398	out = capsys.readouterr().out
4399	assert "pending" in out
4400	assert "claimed" in out
4401
4402	def test_text_shows_column_headers(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4403	repo = _make_repo(tmp_path)
4404	create_task(repo, "task")
4405	args = _tasks_ns(fmt="text")
4406	with _patch_repo(repo):
4407	run_tasks(args)
4408	out = capsys.readouterr().out
4409	assert "ID" in out
4410	assert "QUEUE" in out
4411	assert "TITLE" in out
4412
4413	def test_text_shows_filter_line(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4414	repo = _make_repo(tmp_path)
4415	args = _tasks_ns(fmt="text", queue="billing")
4416	with _patch_repo(repo):
4417	run_tasks(args)
4418	out = capsys.readouterr().out
4419	assert "filter" in out
4420	assert "billing" in out
4421
4422	def test_text_empty_queue_message(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4423	repo = _make_repo(tmp_path)
4424	args = _tasks_ns(fmt="text")
4425	with _patch_repo(repo):
4426	run_tasks(args)
4427	out = capsys.readouterr().out
4428	assert "0 task" in out
4429
4430	def test_text_shows_elapsed(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4431	repo = _make_repo(tmp_path)
4432	args = _tasks_ns(fmt="text")
4433	with _patch_repo(repo):
4434	run_tasks(args)
4435	out = capsys.readouterr().out
4436	assert "s)" in out
4437
4438	def test_ansi_in_queue_name_stripped(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4439	"""Enqueued tasks with ANSI in queue are sanitized in text output."""
4440	# We can't enqueue with a bad queue via CLI (validated), but the task
4441	# record could be crafted; sanitize_display handles it in display.
4442	repo = _make_repo(tmp_path)
4443	create_task(repo, "task")
4444	args = _tasks_ns(fmt="text", run_id="\x1b[31mred\x1b[0m")
4445	# run_id with ANSI is valid length-wise but we're checking display
4446	with _patch_repo(repo):
4447	run_tasks(args)
4448	out = capsys.readouterr().out
4449	assert "\x1b" not in out
4450
4451
4452	class TestTasksIntegration:
4453	"""Full integration tests for run_tasks with realistic task states."""
4454
4455	def test_full_mixed_state_counts(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4456	repo = _make_repo(tmp_path)
4457	# pending (queue="p" — no worker claims these)
4458	create_task(repo, "p1", queue="p")
4459	create_task(repo, "p2", queue="p")
4460	# claimed (queue="c")
4461	create_task(repo, "c1", queue="c")
4462	claim_next_task(repo, "w1", queue="c")
4463	# completed (queue="done")
4464	t_done = create_task(repo, "done", queue="done")
4465	claim_next_task(repo, "w2", queue="done")
4466	complete_task(repo, t_done.task_id, "w2")
4467	# failed (queue="fail")
4468	t_fail = create_task(repo, "fail", queue="fail")
4469	claim_next_task(repo, "w3", queue="fail")
4470	fail_task(repo, t_fail.task_id, "w3", error="boom")
4471	# cancelled
4472	t_can = create_task(repo, "cancelled", queue="can")
4473	cancel_task(repo, t_can.task_id, "orch")
4474
4475	args = _tasks_ns()
4476	with _patch_repo(repo):
4477	run_tasks(args)
4478	out = json.loads(capsys.readouterr().out)
4479	assert out["total"] == 6
4480	assert out["pending"] == 2
4481	assert out["claimed"] == 1
4482	assert out["completed"] == 1
4483	assert out["failed"] == 1
4484	assert out["cancelled"] == 1
4485
4486	def test_filter_by_status_only_returns_matching(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4487	repo = _make_repo(tmp_path)
4488	create_task(repo, "pending-1", queue="p")
4489	t_done = create_task(repo, "done-1", queue="done")
4490	claim_next_task(repo, "w", queue="done")
4491	complete_task(repo, t_done.task_id, "w")
4492
4493	args = _tasks_ns(status="completed")
4494	with _patch_repo(repo):
4495	run_tasks(args)
4496	out = json.loads(capsys.readouterr().out)
4497	assert all(i["status"] == "completed" for i in out["items"])
4498	assert len(out["items"]) == 1
4499	# But total counts still reflect full queue
4500	assert out["total"] == 2
4501
4502	def test_filter_by_queue_and_status(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4503	repo = _make_repo(tmp_path)
4504	# Create and immediately complete a billing task (no competing tasks yet)
4505	t_b = create_task(repo, "billing-done", queue="billing")
4506	claim_next_task(repo, "w", queue="billing")
4507	complete_task(repo, t_b.task_id, "w")
4508	# Now add a pending billing and an infra task
4509	create_task(repo, "billing-pending", queue="billing")
4510	create_task(repo, "infra-pending", queue="infra")
4511
4512	args = _tasks_ns(queue="billing", status="completed")
4513	with _patch_repo(repo):
4514	run_tasks(args)
4515	out = json.loads(capsys.readouterr().out)
4516	assert len(out["items"]) == 1
4517	assert out["items"][0]["queue"] == "billing"
4518	assert out["items"][0]["status"] == "completed"
4519	# Global total is all 3
4520	assert out["total"] == 3
4521
4522	def test_filter_by_run_id(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4523	repo = _make_repo(tmp_path)
4524	t1 = create_task(repo, "t1")
4525	t2 = create_task(repo, "t2")
4526	claim_next_task(repo, "worker-alpha")
4527	claim_next_task(repo, "worker-beta")
4528
4529	args = _tasks_ns(run_id="worker-alpha")
4530	with _patch_repo(repo):
4531	run_tasks(args)
4532	out = json.loads(capsys.readouterr().out)
4533	assert len(out["items"]) == 1
4534	assert out["items"][0]["claimer_run_id"] == "worker-alpha"
4535
4536	def test_enqueue_then_list_shows_created_by(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4537	repo = _make_repo(tmp_path)
4538	eq_args = _enqueue_ns(title="listed-task", queue="default", run_id="enqueuer-1")
4539	with _patch_repo(repo):
4540	run_enqueue(eq_args)
4541	capsys.readouterr()
4542
4543	args = _tasks_ns()
4544	with _patch_repo(repo):
4545	run_tasks(args)
4546	out = json.loads(capsys.readouterr().out)
4547	assert len(out["items"]) == 1
4548	assert out["items"][0]["created_by"] == "enqueuer-1"
4549
4550	def test_limit_with_filter_shows_highest_priority(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4551	repo = _make_repo(tmp_path)
4552	with _freeze(_EPOCH):
4553	for i in range(10):
4554	create_task(repo, f"task-{i}", queue="q", priority=i)
4555	args = _tasks_ns(queue="q", limit=3)
4556	with _patch_repo(repo):
4557	run_tasks(args)
4558	out = json.loads(capsys.readouterr().out)
4559	assert len(out["items"]) == 3
4560	# Top 3 priorities should be 9, 8, 7
4561	assert [i["priority"] for i in out["items"]] == [9, 8, 7]
4562
4563
4564	class TestTasksStress:
4565	"""Performance and concurrency tests for run_tasks."""
4566
4567	def test_500_tasks_listed_under_5s(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4568	repo = _make_repo(tmp_path)
4569	for i in range(500):
4570	create_task(repo, f"task-{i}", queue="default")
4571	args = _tasks_ns(limit=500)
4572	start = time.monotonic()
4573	with _patch_repo(repo):
4574	run_tasks(args)
4575	elapsed = time.monotonic() - start
4576	out = json.loads(capsys.readouterr().out)
4577	assert out["total"] == 500
4578	assert elapsed < 5.0, f"listing 500 tasks took {elapsed:.1f}s"
4579
4580	def test_500_tasks_with_filter_under_5s(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4581	repo = _make_repo(tmp_path)
4582	for i in range(250):
4583	create_task(repo, f"billing-{i}", queue="billing")
4584	for i in range(250):
4585	create_task(repo, f"infra-{i}", queue="infra")
4586	args = _tasks_ns(queue="billing", limit=250)
4587	start = time.monotonic()
4588	with _patch_repo(repo):
4589	run_tasks(args)
4590	elapsed = time.monotonic() - start
4591	out = json.loads(capsys.readouterr().out)
4592	assert len(out["items"]) == 250
4593	assert elapsed < 5.0, f"filtered listing took {elapsed:.1f}s"
4594
4595	def test_concurrent_reads_are_safe(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4596	"""Concurrent run_tasks calls against the same repo must not crash."""
4597	import concurrent.futures
4598	repo = _make_repo(tmp_path)
4599	for i in range(50):
4600	create_task(repo, f"task-{i}")
4601	errors: list[str] = []
4602
4603	def read_tasks() -> None:
4604	args = _tasks_ns(limit=50)
4605	try:
4606	with _patch_repo(repo):
4607	run_tasks(args)
4608	except Exception as exc: # noqa: BLE001
4609	errors.append(str(exc))
4610
4611	with concurrent.futures.ThreadPoolExecutor(max_workers=10) as pool:
4612	list(pool.map(lambda _: read_tasks(), range(20)))
4613
4614	assert not errors, f"concurrent read errors: {errors}"
4615
4616	def test_no_double_load_with_filter(self, tmp_path: pathlib.Path, capsys: pytest.CaptureFixture[str]) -> None:
4617	"""Verify counts are still correct when filter is active (no re-load bug)."""
4618	repo = _make_repo(tmp_path)
4619	for i in range(20):
4620	create_task(repo, f"billing-{i}", queue="billing")
4621	for i in range(10):
4622	create_task(repo, f"infra-{i}", queue="infra")
4623
4624	args = _tasks_ns(queue="billing")
4625	with _patch_repo(repo):
4626	run_tasks(args)
4627	out = json.loads(capsys.readouterr().out)
4628	# total must reflect ALL 30 tasks, not just the 20 in billing
4629	assert out["total"] == 30
4630	assert len(out["items"]) == 20
4631
4632
4633	# ── register_all integration ───────────────────────────────────────────────────
4634
4635
4636	class TestRegisterAll:
4637	"""register_all attaches all six subcommands to the given subparsers."""
4638
4639	def test_all_commands_registered(self) -> None:
4640	import argparse
4641	parser = argparse.ArgumentParser()
4642	subs = parser.add_subparsers(dest="cmd")
4643	register_all(subs)
4644	# Verify each expected command is parseable
4645	for cmd in ("enqueue", "claim", "complete", "fail-task", "cancel-task", "tasks"):
4646	# A subparser was registered for this command name
4647	# (ArgumentParser stores choices in _subparsers._group_actions)
4648	found = False
4649	for action in parser._subparsers._group_actions:
4650	if cmd in action.choices:
4651	found = True
4652	break
4653	assert found, f"Command '{cmd}' not registered"

File History 5 commits

sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6 chore(tests): add docstring to tests/__init__.py so rc14 tr… Human 26 days ago

sha256:d8316ffae901be06347e16ab55be11868eb519dd16ade3e8aa16a99e662f7e62 baseline: rc14 re-baseline after rc3 store corruption recovery Human patch 26 days ago

sha256:50d413a635f57761c3fce1f70251b957b6423821525bf330747ef4007339222e Merge branch 'dev' into main Human 29 days ago

sha256:fb67fed5a4d3e40de84bdd163de94ef1386570bef1dd1a020a732c8a038962ce Merge branch 'dev' into main Human 48 days ago

sha256:1c4b3e3a9a1f300774c3ee662b572a698d5fd405bf765a71e6011a2e9c3eaaaa feat: Muse — version control for the agent era Human 100 days ago

function _make_repo

function _freeze

function _namespace

class TestValidateTaskId

function test_accepts_valid_uuid4

function test_accepts_uppercase

function test_rejects_empty

function test_rejects_non_uuid

function test_rejects_path_traversal

function test_rejects_null_bytes

function test_rejects_uuid3

function test_rejects_uuid_with_slash

class TestValidateQueueName

function test_accepts_simple

function test_rejects_empty

function test_rejects_space

function test_rejects_slash

function test_rejects_null_byte

function test_rejects_too_long

function test_accepts_max_length

class TestTaskRecord

function _make

function test_to_dict_round_trip

function test_is_expired_false_within_ttl

function test_is_expired_true_at_boundary

function test_from_dict_missing_created_at_defaults_to_now

function test_title_truncated_at_256

class TestClaimRecord

function _make

function test_to_dict_round_trip

function test_is_expired_false

function test_is_expired_true

class TestGetTaskStatus

function _task

function _claim

function test_no_claim_is_pending

function test_active_claim_is_claimed

function test_expired_claim_is_timed_out

function test_completed_status_passes_through

function test_failed_status_passes_through

function test_cancelled_status_passes_through

class TestEnsureTaskDirs

function test_creates_directories

function test_idempotent

class TestCreateTask

function test_creates_file_on_disk

function test_returns_correct_fields

function test_file_is_valid_json

function test_empty_title_raises

function test_invalid_queue_raises

function test_tags_capped_at_32

function test_ttl_min_1

class TestLoadTasks

function test_empty_dir_returns_empty_list

function test_non_existent_dir_returns_empty_list

function test_loads_created_task

function test_skips_corrupt_file

function test_load_task_by_id

function test_load_task_missing_returns_none

function test_load_task_invalid_id_raises

class TestTryExclClaim

function test_first_claim_succeeds

function test_second_claim_returns_none

function test_claim_file_is_written

function test_claim_file_is_valid_json

function test_expires_at_correct

class TestTryOptimisticReclaim

function test_reclaim_wins_when_no_competition

function test_nonce_written_to_file

class TestClaimNextTask

function test_returns_none_on_empty_queue

function test_claims_only_task

function test_higher_priority_claimed_first

function test_fifo_within_same_priority

function test_queue_filter_respected

function test_queue_filter_returns_none_when_no_match

function test_already_claimed_task_not_re_claimed

function test_expired_task_ttl_skipped

function test_reclaims_timed_out_task

class TestCompleteTask

function _enqueue_and_claim

function test_completes_successfully

function test_persisted_to_disk

function test_wrong_run_id_raises_permission_error

function test_invalid_task_id_raises_value_error

function test_missing_task_raises_file_not_found

function test_double_complete_raises_runtime_error

class TestFailTask

function test_fails_successfully

function test_wrong_claimer_raises

function test_already_failed_raises

class TestCancelTask

function test_cancel_pending_task

function test_cancel_claimed_by_claimer

function test_cancel_claimed_by_non_claimer_raises

function test_force_cancel_overrides_ownership

function test_cancel_nonexistent_task_raises

function test_cancel_completed_raises

function test_invalid_id_raises

class TestHeartbeatClaim

function test_extends_expiry

function test_wrong_claimer_raises

function test_no_claim_raises

function test_heartbeat_after_complete_raises

class TestFullLifecycle

function test_enqueue_claim_complete

function test_enqueue_claim_fail_then_reclaim

function test_heartbeat_prevents_expiry

function test_multi_task_priority_and_fifo

class TestSecurity

function test_path_traversal_in_task_id_load_task

function test_path_traversal_in_task_id_load_claim

function test_path_traversal_in_complete

function test_path_traversal_in_fail

function test_path_traversal_in_cancel

function test_path_traversal_in_heartbeat

function test_null_byte_in_task_id

function test_oversized_title_is_truncated

function test_oversized_queue_raises

function test_ansi_injection_in_title_stored_verbatim

class TestConcurrentClaiming

function test_exactly_one_winner_from_n_threads

function try_claim

function test_n_tasks_claimed_by_n_agents_no_duplicates

function try_claim

class TestStress

function test_enqueue_500_tasks

function test_claim_scan_500_tasks

function test_load_all_tasks_1000

function _patch_repo

class TestCliEnqueue

function test_enqueue_json_output

function test_enqueue_text_output

function test_enqueue_invalid_payload_exits_1

function test_enqueue_payload_not_object_exits_1

function test_enqueue_invalid_queue_exits_1

function test_enqueue_elapsed_included_in_json

function _enqueue_ns

class TestEnqueueInputValidation

function test_empty_title_exits_1

function test_whitespace_only_title_exits_1

function test_run_id_at_max_length_accepted

function test_run_id_over_max_length_exits_1

function test_ttl_zero_exits_1

function test_ttl_negative_exits_1

function test_ttl_one_accepted

function test_payload_over_max_bytes_exits_1

function test_payload_at_max_bytes_accepted

function test_payload_not_json_exits_1

function test_payload_array_exits_1

function test_too_many_tags_exits_1

function test_max_tags_accepted

function test_invalid_queue_name_exits_1

function test_queue_with_spaces_exits_1

function test_queue_name_too_long_exits_1

function test_valid_queue_name_with_hyphens_accepted

function test_valid_queue_name_with_underscores_accepted

function test_validation_fires_before_repo_lookup

function _fake_require

class TestEnqueueJsonErrors

function test_empty_title_json_error

function test_bad_payload_json_error

function test_bad_queue_json_error

function test_json_error_is_compact_single_line

function test_text_error_goes_to_stderr

function test_run_id_too_long_json_error

class TestEnqueueCompactJson

function test_success_json_is_single_line

function test_success_json_schema_keys

function test_success_json_values_correct

function test_elapsed_seconds_is_float

function test_task_id_is_uuid4

function test_unique_task_ids_across_calls

class TestEnqueueTextOutput

function test_success_text_contains_task_enqueued

function test_success_text_shows_title

function test_success_text_shows_queue

function test_success_text_shows_priority

function test_success_text_shows_tags

function test_ansi_in_title_stripped_from_text

function test_ansi_in_tags_stripped_from_text

class TestEnqueueIntegration

function test_enqueued_task_is_claimable

function test_priority_ordering

function test_enqueue_then_list_shows_task

function test_enqueue_with_payload_passes_through_to_claim

class TestEnqueueStress

function test_enqueue_500_tasks_under_10s

function test_concurrent_enqueue_produces_unique_ids

function _enqueue

function test_enqueue_large_payload_at_boundary

function test_enqueue_many_queues

class TestCliClaim

function test_claim_success_json

function test_claim_empty_queue_exits_1

function test_claim_text_output_on_success

function test_claim_text_output_on_empty

function _claim_ns

class TestClaimInputValidation

function test_run_id_at_max_length_accepted

function test_run_id_over_max_exits_1

function test_run_id_too_long_json_error

function test_run_id_too_long_text_error

function test_claim_ttl_min_accepted

function test_claim_ttl_max_accepted

function test_claim_ttl_zero_exits_1

function test_claim_ttl_negative_exits_1

function test_claim_ttl_over_max_exits_1

function test_claim_ttl_invalid_json_error

function test_wait_over_max_exits_1

function test_wait_negative_exits_1

function test_invalid_queue_name_exits_1

function test_invalid_queue_json_error

function test_queue_with_slash_exits_1

function test_validation_fires_before_repo_lookup

function _fake_require

function test_valid_queue_name_filters_correctly

class TestClaimJsonOutput

function test_success_json_is_compact

function test_success_json_schema_keys

function test_success_task_nested_schema

function test_empty_queue_json_is_compact

function test_empty_queue_json_schema

function test_elapsed_seconds_is_float

function test_claimer_run_id_matches_arg

class TestClaimTextOutput

function test_success_shows_task_claimed

function test_success_shows_title

function test_success_shows_expires_in

function test_success_shows_payload

function test_success_shows_tags

function test_ansi_in_run_id_stripped

function test_ansi_in_title_stripped

function test_empty_text_shows_empty

class TestClaimWait

function test_wait_zero_returns_immediately_on_empty

function test_wait_times_out_on_empty_queue

function test_wait_finds_task_created_during_wait

function _create_after_delay

function test_wait_empty_json_includes_elapsed

function test_wait_text_mentions_wait_on_empty

class TestClaimQueueFilter

function test_queue_filter_claims_correct_queue

function test_queue_filter_empty_when_wrong_queue

function test_no_queue_filter_claims_highest_priority

function test_queue_filter_with_valid_chars

class TestClaimIntegration

function test_claim_then_complete

function test_claim_then_fail

function test_second_claim_empty_after_first

function test_priority_order_across_claims

function test_enqueue_claim_complete_via_cli

class TestClaimStress

function test_concurrent_claim_no_double_claim

function _agent

function test_claim_100_tasks_sequential

function test_wait_zero_performance

class TestCliComplete

function _setup

function test_complete_success_json

function test_complete_with_result

function test_complete_wrong_claimer_exits_1

function test_complete_invalid_result_exits_1

function test_complete_text_output

function _complete_ns

class TestCompleteInputValidation

function _setup

function test_run_id_too_long_exits_1

function test_run_id_at_max_length_passes

function test_result_not_json_exits_1

function test_result_array_exits_1

function test_result_scalar_exits_1

function test_result_null_exits_1

function test_result_too_large_exits_1

function test_result_at_max_size_passes_validation

function test_invalid_task_id_exits_1

function test_path_traversal_task_id_exits_1

function test_null_byte_task_id_exits_1

function test_validation_fires_before_repo_lookup

function counting_require_repo

function test_json_error_shape_bad_task_id

function test_json_error_shape_bad_result

function test_text_error_goes_to_stderr

function test_json_error_goes_to_stdout_not_stderr

class TestCompleteJsonOutput

function _setup

function test_json_is_compact

function test_json_has_required_keys

function test_json_status_is_completed

function test_json_result_field_populated

function test_json_empty_result_is_null

function test_json_elapsed_is_float

function test_json_claimer_run_id_matches

function test_json_wrong_claimer_error_shape

function test_json_missing_task_error_shape

class TestCompleteTextOutput

function _setup

function test_text_shows_completed

function test_text_shows_task_title

function test_text_shows_queue

function test_text_shows_claimer

function test_text_shows_result_when_provided

function test_text_shows_elapsed

function test_ansi_injection_in_run_id_stripped

function test_ansi_injection_in_title_stripped

class TestCompleteIntegration

function test_completed_status_persisted_to_disk

function test_result_persisted_to_disk

function test_double_complete_fails

function test_complete_missing_claim_fails

function test_enqueue_claim_complete_full_cycle

function test_complete_with_unicode_result

class TestCompleteStress

function test_20_agents_each_claim_and_complete_unique_task

function claim_and_complete

function test_100_sequential_completes_under_10s

function test_complete_via_run_complete_100_sequential

class TestCliFailTask

function _setup

function test_fail_success_json

function test_fail_wrong_claimer_exits_1

function test_fail_text_output

function _fail_ns

class TestFailTaskInputValidation

function _setup

function test_run_id_too_long_exits_1

function test_run_id_at_max_length_passes

function test_error_too_long_exits_1

function test_error_at_max_length_passes

function test_empty_error_allowed

function test_invalid_task_id_exits_1

function test_path_traversal_task_id_exits_1

function test_null_byte_task_id_exits_1

function test_validation_fires_before_repo_lookup

function counting_require_repo

function test_run_id_validation_fires_before_repo_lookup

function counting_require_repo

function test_error_validation_fires_before_repo_lookup

function counting_require_repo

function test_json_error_shape_bad_task_id

function test_json_error_shape_bad_run_id

function test_json_error_shape_bad_error_msg

function test_text_error_goes_to_stderr

function test_json_error_goes_to_stdout_not_stderr

class TestFailTaskJsonOutput

function _setup

function test_json_is_compact

function test_json_has_required_keys

function test_json_status_is_failed

function test_json_error_field_populated

function test_json_empty_error_is_empty_string

function test_json_elapsed_is_float

function test_json_claimer_run_id_matches

function test_json_wrong_claimer_error_shape

function test_json_missing_task_error_shape

class TestFailTaskTextOutput

function _setup

function test_text_shows_failed

function test_text_shows_task_title

function test_text_shows_queue

function test_text_shows_claimer

function test_text_shows_error_message

function test_text_no_error_line_when_empty

function test_text_shows_elapsed

function test_ansi_injection_in_error_stripped

function test_ansi_injection_in_title_stripped

class TestFailTaskIntegration

function test_failed_status_persisted_to_disk

function test_error_persisted_to_disk

function test_double_fail_exits_1

function test_fail_unclaimed_task_exits_1

function test_enqueue_claim_fail_full_cycle

function test_unicode_error_message

class TestFailTaskStress

function test_20_agents_each_claim_and_fail_unique_task

function claim_and_fail

function test_100_sequential_fails_under_15s

function test_fail_via_run_fail_task_100_sequential

class TestCliCancelTask

function test_cancel_pending_task_json

function test_cancel_claimed_by_claimer

function test_cancel_force_different_agent

function test_cancel_wrong_claimer_no_force_exits_1

function test_cancel_text_output

function _cancel_ns

class TestCancelTaskInputValidation

function test_run_id_too_long_exits_1

function test_run_id_at_max_length_passes_validation

function test_invalid_task_id_exits_1

function test_path_traversal_task_id_exits_1

function test_null_byte_task_id_exits_1

function test_task_id_validation_fires_before_repo_lookup

function counting_require_repo

function test_run_id_validation_fires_before_repo_lookup

function counting_require_repo

function test_json_error_shape_bad_task_id

function test_json_error_shape_bad_run_id

function test_text_error_goes_to_stderr

function test_json_error_goes_to_stdout_not_stderr

function test_missing_task_error_shape

function test_already_terminal_error_shape

function test_wrong_claimer_no_force_error_shape

class TestCancelTaskJsonOutput

function test_json_is_compact

function test_json_has_required_keys

function test_json_status_is_cancelled

function test_json_elapsed_is_float

function test_json_claimer_run_id_for_pending_is_caller

function test_json_claimer_run_id_for_claimed_is_original_claimer

function test_json_force_cancel_different_agent

function test_json_task_id_matches

class TestCancelTaskTextOutput

function _setup_pending

function _setup_claimed

function test_text_shows_cancelled

function test_text_shows_task_title

function test_text_shows_queue

function test_text_shows_claimer

function test_text_shows_forced_indicator

function test_text_no_forced_indicator_without_flag

function test_text_shows_elapsed

function test_ansi_injection_in_title_stripped

function test_ansi_injection_in_run_id_stripped

class TestCancelTaskIntegration

function test_pending_task_cancelled_status_on_disk

function test_claimed_task_cancelled_by_claimer

function test_force_cancel_different_claimer

function test_double_cancel_exits_1

function test_cancel_completed_task_exits_1

function test_cancel_failed_task_exits_1

function test_enqueue_then_cancel_full_cycle

function test_enqueue_claim_cancel_full_cycle

class TestCancelTaskStress

function test_20_agents_each_cancel_unique_pending_task

function do_cancel

function test_100_sequential_cancels_under_15s

function test_cancel_via_run_cancel_task_100_sequential

class TestCliTasks

function _setup_mixed

function test_lists_all_tasks_json

function test_filter_by_status_pending

function test_filter_by_queue

function test_filter_by_run_id

function test_items_sorted_by_priority_desc

function test_text_output_no_crash

function test_empty_queue_text_output

function test_status_counts_in_json

function test_ansi_in_title_not_printed_raw

function _tasks_ns

class TestTasksInputValidation

function test_invalid_queue_name_exits_1

function test_queue_with_slash_exits_1

function test_queue_with_ansi_exits_1

function test_valid_queue_name_passes

function test_run_id_too_long_exits_1

function test_run_id_at_max_length_passes

function test_limit_zero_exits_1

function test_limit_negative_exits_1

function test_limit_over_max_exits_1

function test_limit_at_max_passes

function test_queue_validation_fires_before_repo_lookup

function counting_require_repo

function test_run_id_validation_fires_before_repo_lookup

function counting_require_repo

function test_json_error_shape_bad_queue

function test_json_error_shape_bad_limit

function test_text_error_goes_to_stderr

function test_json_error_goes_to_stdout_not_stderr

class TestTasksJsonOutput

function _setup

function test_json_is_compact

function test_json_has_top_level_keys

function test_items_have_new_fields

function test_expires_at_null_for_pending_task

function test_expires_at_populated_for_claimed_task

function test_status_counts_reflect_full_queue_when_filtered

function test_limit_truncates_items

function test_no_truncation_when_within_limit

function test_elapsed_is_float

function test_items_sorted_priority_desc

function test_limit_applies_after_sort

class TestTasksTextOutput

function test_text_shows_task_queue_header

function test_text_shows_status_counts

function test_text_shows_column_headers

function test_text_shows_filter_line

function test_text_empty_queue_message

function test_text_shows_elapsed

function test_ansi_in_queue_name_stripped

class TestTasksIntegration

function test_full_mixed_state_counts

function test_filter_by_status_only_returns_matching

function test_filter_by_queue_and_status

function test_filter_by_run_id

function test_enqueue_then_list_shows_created_by

function test_limit_with_filter_shows_highest_priority

class TestTasksStress

function test_500_tasks_listed_under_5s

function test_500_tasks_with_filter_under_5s

function test_concurrent_reads_are_safe

function read_tasks

function test_no_double_load_with_filter

class TestRegisterAll

function test_all_commands_registered

Pathtests/test_cmd_task_queue.py

Lines4,653

Size191.1 KB

LangPython

Refsha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6

Object ID

sha256:988a43722abd2a0fadedd1126c155b9bff745894854dd1b1ec1431eb98190439…

Last commit

sha256:660fcac1df3ab28f61862e961890bd2ca8b754fa0242079d93ca1e25037ec8a6

chore(tests): add docstring to tests/__init__.py …

26 days ago

Quick links

Blame History