Log errors from failing Perpetual tasks instead of swallowing them

When a Perpetual task fails, `on_complete()` returns `True` which tells
the Worker "I handled it" — skipping the normal error log. The reschedule
works fine, but the exception is completely invisible. Now we log at ERROR
with the full traceback before the reschedule info line.

Also fixes the logger names in the private dependency modules
(`_perpetual`, `_concurrency`, `_retry`) to use `docket.dependencies`
instead of leaking internal names like `docket.dependencies._perpetual`
into log output.

Closes #360

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>

Author: chrisguidry

src/docket/dependencies/_concurrency.py

@@ -16,7 +16,7 @@
     current_worker,
 )
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger("docket.dependencies")
 
 if TYPE_CHECKING:  # pragma: no cover
     from redis.asyncio import Redis

src/docket/dependencies/_perpetual.py

@@ -20,7 +20,7 @@
 
 from ..instrumentation import TASKS_PERPETUATED, TASKS_SUPERSEDED
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger("docket.dependencies")
 
 
 class Perpetual(CompletionHandler["Perpetual"]):
@@ -131,6 +131,15 @@ async def on_complete(self, execution: Execution, outcome: TaskOutcome) -> bool:
         )
 
         TASKS_PERPETUATED.add(1, {**worker.labels(), **execution.general_labels()})
+
+        if outcome.exception:
+            logger.error(
+                "↩ [%s] %s",
+                format_duration(outcome.duration.total_seconds()),
+                execution.call_repr(),
+                exc_info=outcome.exception,
+            )
+
         logger.info(
             "↫ [%s] %s",
             format_duration(outcome.duration.total_seconds()),

src/docket/dependencies/_retry.py

@@ -19,7 +19,7 @@
 
 from ..instrumentation import TASKS_RETRIED
 
-logger = logging.getLogger(__name__)
+logger = logging.getLogger("docket.dependencies")
 
 
 class ForcedRetry(Exception):

tests/fundamentals/test_perpetual.py

@@ -1,8 +1,11 @@
 """Tests for Perpetual dependency (automatically rescheduling tasks)."""
 
+import logging
 from datetime import datetime, timedelta, timezone
 
-from docket import Docket, ExecutionState, Perpetual, Worker
+import pytest
+
+from docket import Depends, Docket, ExecutionState, Perpetual, Worker
 
 
 async def test_perpetual_tasks(docket: Docket, worker: Worker):
@@ -106,6 +109,52 @@ async def perpetual_task(
     assert calls == 3
 
 
+async def test_perpetual_tasks_log_errors(
+    docket: Docket, worker: Worker, caplog: pytest.LogCaptureFixture
+):
+    """Failing Perpetual tasks log the exception instead of swallowing it."""
+
+    async def perpetual_task(
+        perpetual: Perpetual = Perpetual(every=timedelta(milliseconds=50)),
+    ):
+        raise ValueError("something broke")
+
+    execution = await docket.add(perpetual_task)()
+
+    with caplog.at_level(logging.ERROR):
+        await worker.run_at_most({execution.key: 1})
+
+    error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
+    assert any(
+        "something broke" in r.getMessage()
+        or (r.exc_info and "something broke" in str(r.exc_info[1]))
+        for r in error_records
+    )
+
+
+async def test_perpetual_tasks_log_dependency_errors(
+    docket: Docket, worker: Worker, caplog: pytest.LogCaptureFixture
+):
+    """Failing dependencies in Perpetual tasks log the error instead of swallowing it."""
+
+    async def broken_dep() -> str:
+        raise RuntimeError("dependency failed")
+
+    async def perpetual_task(
+        value: str = Depends(broken_dep),
+        perpetual: Perpetual = Perpetual(every=timedelta(milliseconds=50)),
+    ):
+        pass
+
+    execution = await docket.add(perpetual_task)()
+
+    with caplog.at_level(logging.ERROR):
+        await worker.run_at_most({execution.key: 1})
+
+    error_records = [r for r in caplog.records if r.levelno == logging.ERROR]
+    assert any("dependency failed" in (r.exc_text or "") for r in error_records)
+
+
 async def test_perpetual_tasks_can_be_automatically_scheduled(
     docket: Docket, worker: Worker
 ):