Fix race condition in subscribe() for fast-completing tasks

The watch command was failing in CI when tasks completed quickly because:
- subscribe() would sync() and find task already completed
- Progress data would be deleted on completion
- Watch would never see progress events

Solution: subscribe() now emits initial progress event along with initial
state event, ensuring subscribers capture progress state before deletion.

🤖 Generated with [Claude Code](https://claude.com/claude-code)

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

Author: desertaxle

src/docket/execution.py

@@ -757,6 +757,25 @@ async def subscribe(self) -> AsyncGenerator[dict[str, Any], None]:
 
         yield initial_state
 
+        # Also emit initial progress data if available
+        # This ensures subscribers see progress state even if task completes quickly
+        # and progress data is deleted before pub/sub connection is established
+        if self.progress.current is not None or self.progress.total != 100:
+            progress_event: dict[str, Any] = {
+                "type": "progress",
+                "key": self.key,
+                "current": self.progress.current
+                if self.progress.current is not None
+                else 0,
+                "total": self.progress.total,
+            }
+            if self.progress.message:
+                progress_event["message"] = self.progress.message
+            if self.progress.updated_at:
+                progress_event["updated_at"] = self.progress.updated_at.isoformat()
+
+            yield progress_event
+
         # Then subscribe to real-time updates
         state_channel = f"{self.docket.name}:state:{self.key}"
         progress_channel = f"{self.docket.name}:progress:{self.key}"

tests/cli/test_watch.py

@@ -122,17 +122,13 @@ async def task_with_progress(progress: ExecutionProgress = Progress()):
 
     worker_task = asyncio.create_task(worker.run_until_finished())
 
-    # Give worker time to start the task
-    await asyncio.sleep(0.05)
-
     result = await run_cli(
         "watch",
         "progress-task",
         "--url",
         docket.url,
         "--docket",
         docket.name,
-        timeout=3.0,
     )
 
     await worker_task
@@ -263,7 +259,7 @@ async def task_that_waits_then_progresses(progress: ExecutionProgress = Progress
     docket.register(task_that_waits_then_progresses)
 
     # Schedule task for slightly in future so watch can connect first
-    when = datetime.now(timezone.utc) + timedelta(milliseconds=200)
+    when = datetime.now(timezone.utc) + timedelta(seconds=1)
     await docket.add(task_that_waits_then_progresses, when=when, key="timing-test")()
 
     worker_task = asyncio.create_task(worker.run_until_finished())