fix(phase-v): improve event publishing, logging, and WebSocket service reliability

- Add logging configuration and startup diagnostics to backend
- Improve event publisher graceful handling when Dapr is unavailable
- Enhance MCP server with better tool handling and error recovery
- Fix WebSocket service task update handler for real-time sync
- Update DashboardClient and ThemedChatWidget with UI improvements

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

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

Author: DanielHashmi

CLAUDE.md

@@ -278,6 +278,47 @@ Complete list of specialized agents available:
 
 ## Critical Rules
 
+### Spec-Kit-Plus Compliance & Implementation Change Documentation
+
+Given strict adherence to the spec-kit-plus methodology, the following rules govern how implementation changes are documented:
+
+**1. Change Documentation Workflow**
+When directly requested to modify code or implementation details:
+
+   - **Prompt for Documentation First**: Before making changes, ask the user to document:
+     - The specific deviation from spec.md, plan.md, tasks.md, or other design documents
+     - The rationale/justification for the change
+     - Impact on other components, user stories, or downstream tasks
+     - Whether this is a temporary workaround or permanent fix
+
+   - **Update Tasks After Approval**: Once changes are approved and implemented:
+     - Mark the original task as `[X]` complete
+     - Add a new task documenting what was actually done vs. what was specified
+     - Note any errors, issues, or workarounds encountered
+     - Reference the divergence in task comments (e.g., "// Deviation from T005: Changed X due to Y")
+
+**2. Progressive Specification Alignment**
+The goal is to progressively align specifications with actual implementation:
+
+   - **Minor Deviations**: Update task comments and continue
+   - **Moderate Changes**: Create an ADR using `/sp.adr <title>` to document the architectural decision
+   - **Major Scope Changes**: Update the relevant spec.md, plan.md, or tasks.md sections with user approval
+
+**3. Preserving Core Requirements**
+   - **NEVER modify** statements that represent core requirements from:
+     - `.specify/memory/constitution.md` (core principles)
+     - spec.md (user stories, success criteria, acceptance scenarios)
+     - plan.md (architecture decisions, technical constraints)
+   - Only modify specification-related files with explicit user approval
+   - Document all changes, even approved ones, for traceability
+
+**4. Why This Matters**
+Without systematic documentation:
+   - Specifications become outdated and inaccurate
+   - New team members cannot understand what was actually implemented
+   - Future changes may inadvertently break working implementations
+   - The gap between "what we planned" and "what we built" grows unmanageably
+
 ### Stateless Architecture
 - ALL state persisted to database
 - Store user message BEFORE agent runs

backend/main.py

@@ -1,5 +1,6 @@
 """FastAPI application entry point for LifeStepsAI backend."""
 import asyncio
+import logging
 import os
 from contextlib import asynccontextmanager
 from pathlib import Path
@@ -21,6 +22,13 @@
 
 load_dotenv()
 
+# Configure logging
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
 # CORS settings - support multiple origins from CORS_ORIGINS env var
 FRONTEND_URL = os.getenv("FRONTEND_URL", "http://localhost:3000")
 CORS_ORIGINS_ENV = os.getenv("CORS_ORIGINS", "")
@@ -43,6 +51,13 @@ async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
     # Startup: Create database tables
     create_db_and_tables()
 
+    # Log configuration for event publishing
+    dapr_http_port = os.getenv("DAPR_HTTP_PORT", "3500")
+    websocket_url = os.getenv("WEBSOCKET_SERVICE_URL", "http://localhost:8004")
+    logger.info(f"Backend starting...")
+    logger.info(f"  DAPR_HTTP_PORT: {dapr_http_port}")
+    logger.info(f"  WEBSOCKET_SERVICE_URL: {websocket_url}")
+
     # Start notification polling in background
     notification_task = asyncio.create_task(notification_polling_loop())
 

backend/src/chatbot/mcp_agent.py

@@ -182,6 +182,7 @@ def __init__(self, provider: str | None = None, model: str | None = None):
                     # Explicitly pass critical env vars to subprocess
                     "DATABASE_URL": os.getenv("DATABASE_URL", ""),
                     "OPENAI_API_KEY": os.getenv("OPENAI_API_KEY", ""),
+                    "WEBSOCKET_SERVICE_URL": os.getenv("WEBSOCKET_SERVICE_URL", "http://localhost:8004"),
                 },
             },
             client_session_timeout_seconds=30.0,

backend/src/mcp_server/server.py

@@ -52,6 +52,56 @@ def get_db_session():
     return Session(engine)
 
 
+def publish_event_sync(event_type: str, task: any, user_id: str, changes: list = None, task_before: dict = None):
+    """Publish event synchronously for MCP tools.
+
+    Used in sync MCP tools - runs the async event publishing in a thread
+    with its own event loop. This works even when called from within
+    an async context (like MCP tools called by OpenAI Agents SDK).
+
+    Args:
+        event_type: Event type (created, updated, completed, deleted)
+        task: Task SQLModel instance
+        user_id: User who performed the action
+        changes: List of field changes (for update events)
+        task_before: Task state before changes (for update events)
+    """
+    from src.services.event_publisher import publish_task_event
+    import threading
+
+    try:
+        # Create a new thread with its own event loop to run async publishing
+        # This works even when called from within an async context
+        result = [None]
+        exception = [None]
+
+        def run_in_thread():
+            try:
+                loop = asyncio.new_event_loop()
+                asyncio.set_event_loop(loop)
+                result[0] = loop.run_until_complete(
+                    publish_task_event(event_type, task, user_id, changes, task_before)
+                )
+                loop.close()
+            except Exception as e:
+                exception[0] = e
+
+        thread = threading.Thread(target=run_in_thread, daemon=True)
+        thread.start()
+        thread.join(timeout=10)  # Wait up to 10 seconds
+
+        if exception[0]:
+            raise exception[0]
+
+        if result[0]:
+            logger.debug(f"Event published synchronously: task.{event_type}")
+        else:
+            logger.warning(f"Event publishing returned False: task.{event_type}")
+    except Exception as e:
+        # Log error but don't fail the tool
+        logger.error(f"Failed to publish event task.{event_type}: {e}", exc_info=True)
+
+
 def fire_and_forget_event(coro):
     """Run an async coroutine in the background (fire-and-forget).
 
@@ -138,9 +188,8 @@ def add_task(
         session.commit()
         session.refresh(task)
 
-        # Phase V: Publish task.created event (fire-and-forget)
-        from src.services.event_publisher import publish_task_event
-        fire_and_forget_event(publish_task_event("created", task, user_id))
+        # Phase V: Publish task.created event synchronously (before returning)
+        publish_event_sync("created", task, user_id)
 
         # Calculate urgency for display
         urgency = calculate_urgency(task.due_date, task.timezone) if task.due_date else None
@@ -252,15 +301,10 @@ def complete_task(
         session.refresh(updated_task)
 
         # Phase V: Publish event based on completion state change
-        from src.services.event_publisher import publish_task_event, task_to_dict
         if updated_task.completed and not was_completed:
-            fire_and_forget_event(publish_task_event("completed", updated_task, user_id))
+            publish_event_sync("completed", updated_task, user_id)
         elif not updated_task.completed and was_completed:
-            fire_and_forget_event(publish_task_event(
-                "updated", updated_task, user_id,
-                changes=["completed"],
-                task_before=task_to_dict(task)
-            ))
+            publish_event_sync("updated", updated_task, user_id, changes=["completed"])
 
         return {
             "task_id": updated_task.id,
@@ -307,9 +351,8 @@ def delete_task(
         task_service.delete_task(task_id, user_id)
         session.commit()
 
-        # Phase V: Publish task.deleted event with task snapshot
-        from src.services.event_publisher import publish_task_event
-        fire_and_forget_event(publish_task_event("deleted", task_snapshot, user_id))
+        # Phase V: Publish task.deleted event with task snapshot (synchronous)
+        publish_event_sync("deleted", task_snapshot, user_id)
 
         return {
             "task_id": task_id,
@@ -404,12 +447,8 @@ def update_task(
         session.commit()
         session.refresh(updated_task)
 
-        # Phase V: Publish task.updated event with before/after state
-        fire_and_forget_event(publish_task_event(
-            "updated", updated_task, user_id,
-            changes=changes,
-            task_before=task_before_dict
-        ))
+        # Phase V: Publish task.updated event with before/after state (synchronous)
+        publish_event_sync("updated", updated_task, user_id, changes, task_before_dict)
 
         # Calculate urgency for display
         urgency = calculate_urgency(updated_task.due_date, updated_task.timezone) if updated_task.due_date else None

backend/src/services/event_publisher.py

@@ -180,64 +180,71 @@ async def publish_task_event(
         # Create CloudEvents envelope
         cloud_event = create_cloud_event(event_type, event_data)
 
-        # Publish to both task-events (audit, recurring) and task-updates (websocket)
+        # Track success across all publish attempts
+        success = False
+
         async with httpx.AsyncClient(timeout=5.0) as client:
-            # Publish to task-events topic via Dapr
-            response = await client.post(
-                f"{DAPR_PUBLISH_URL}/{TOPIC_TASK_EVENTS}",
-                json=cloud_event,
-                headers={
-                    "Content-Type": "application/cloudevents+json",
-                },
-            )
+            # Try to publish to Dapr (if running in Kubernetes)
+            try:
+                # Publish to task-events topic via Dapr
+                response = await client.post(
+                    f"{DAPR_PUBLISH_URL}/{TOPIC_TASK_EVENTS}",
+                    json=cloud_event,
+                    headers={
+                        "Content-Type": "application/cloudevents+json",
+                    },
+                )
 
-            if response.status_code not in (200, 204):
-                logger.warning(
-                    f"Failed to publish to {TOPIC_TASK_EVENTS}: "
-                    f"status={response.status_code}, body={response.text}"
+                if response.status_code not in (200, 204):
+                    logger.warning(
+                        f"Failed to publish to {TOPIC_TASK_EVENTS}: "
+                        f"status={response.status_code}, body={response.text}"
+                    )
+                else:
+                    success = True
+
+                # Publish to task-updates topic via Dapr (for real-time sync)
+                response_updates = await client.post(
+                    f"{DAPR_PUBLISH_URL}/{TOPIC_TASK_UPDATES}",
+                    json=cloud_event,
+                    headers={
+                        "Content-Type": "application/cloudevents+json",
+                    },
                 )
 
-            # Publish to task-updates topic via Dapr (for real-time sync)
-            response_updates = await client.post(
-                f"{DAPR_PUBLISH_URL}/{TOPIC_TASK_UPDATES}",
-                json=cloud_event,
-                headers={
-                    "Content-Type": "application/cloudevents+json",
-                },
-            )
+                if response_updates.status_code not in (200, 204):
+                    logger.warning(
+                        f"Failed to publish to {TOPIC_TASK_UPDATES}: "
+                        f"status={response_updates.status_code}, body={response_updates.text}"
+                    )
+                else:
+                    success = True
 
-            if response_updates.status_code not in (200, 204):
-                logger.warning(
-                    f"Failed to publish to {TOPIC_TASK_UPDATES}: "
-                    f"status={response_updates.status_code}, body={response_updates.text}"
-                )
+                logger.debug(f"Published to Dapr pub/sub: task.{event_type}")
+
+            except httpx.ConnectError:
+                # Dapr sidecar not running (local dev without Kubernetes)
+                logger.debug(f"Dapr sidecar not available (expected in local dev)")
 
-            # Also publish directly to WebSocket service for local dev (no Dapr)
+            # ALWAYS try direct WebSocket service publish (for local dev without Dapr)
             try:
                 ws_response = await client.post(
                     f"{WEBSOCKET_SERVICE_URL}/api/events/task-updates",
                     json=cloud_event,
                     timeout=3.0,
                 )
                 if ws_response.status_code == 200:
-                    logger.debug("Published event directly to WebSocket service")
+                    logger.info(f"Published task.{event_type} to WebSocket service: task_id={task.id}, user_id={user_id}")
+                    success = True
+                else:
+                    logger.warning(f"WebSocket service returned {ws_response.status_code}: {ws_response.text}")
             except httpx.ConnectError:
-                # WebSocket service not running - skip
-                pass
+                # WebSocket service not running
+                logger.warning(f"WebSocket service not available at {WEBSOCKET_SERVICE_URL}")
             except Exception as ws_err:
-                logger.debug(f"Failed to publish to WebSocket service: {ws_err}")
+                logger.error(f"Failed to publish to WebSocket service: {ws_err}")
 
-        logger.info(
-            f"Published task.{event_type} event: task_id={task.id}, user_id={user_id}"
-        )
-        return True
-
-    except httpx.ConnectError:
-        # Dapr sidecar not running (local dev without Kubernetes)
-        logger.debug(
-            f"Dapr sidecar not available, skipping event publish: task.{event_type}"
-        )
-        return False
+        return success
 
     except Exception as e:
         # Log error but don't fail the API call

frontend/app/dashboard/DashboardClient.tsx

@@ -102,23 +102,59 @@ export default function DashboardClient({ session: initialSession }: DashboardCl
   const { updateName, updateImage } = useProfileUpdate();
   const { isSyncing, pendingCount, lastError } = useSyncQueue();
 
-  // Phase V: WebSocket for real-time task updates
+  // Phase V: WebSocket for real-time task updates with optimistic updates
   const { connectionState, isConnected } = useWebSocket({
-    onTaskCreated: useCallback(() => {
-      // Revalidate tasks when a new task is created from another tab/device
-      revalidateTasks();
+    onTaskCreated: useCallback((taskData: Record<string, unknown>) => {
+      // Optimistically add the new task to local state
+      revalidateTasks(
+        (currentTasks) => {
+          if (!currentTasks) return currentTasks;
+          const newTask = taskData as Task;
+          // Check if task already exists (avoid duplicates)
+          if (currentTasks.some(t => t.id === newTask.id)) {
+            return currentTasks;
+          }
+          return [newTask, ...currentTasks];
+        },
+        { revalidate: false } // Don't refetch from server
+      );
     }, [revalidateTasks]),
-    onTaskUpdated: useCallback(() => {
-      // Revalidate tasks when a task is updated from another tab/device
-      revalidateTasks();
+    onTaskUpdated: useCallback((taskData: Record<string, unknown>) => {
+      // Optimistically update the task in local state
+      revalidateTasks(
+        (currentTasks) => {
+          if (!currentTasks) return currentTasks;
+          const updatedTask = taskData as Task;
+          return currentTasks.map(t =>
+            t.id === updatedTask.id ? updatedTask : t
+          );
+        },
+        { revalidate: false } // Don't refetch from server
+      );
     }, [revalidateTasks]),
-    onTaskCompleted: useCallback(() => {
-      // Revalidate tasks when a task is completed from another tab/device
-      revalidateTasks();
+    onTaskCompleted: useCallback((taskData: Record<string, unknown>) => {
+      // Optimistically update completion status in local state
+      revalidateTasks(
+        (currentTasks) => {
+          if (!currentTasks) return currentTasks;
+          const completedTask = taskData as Task;
+          return currentTasks.map(t =>
+            t.id === completedTask.id ? { ...t, completed: completedTask.completed } : t
+          );
+        },
+        { revalidate: false } // Don't refetch from server
+      );
     }, [revalidateTasks]),
-    onTaskDeleted: useCallback(() => {
-      // Revalidate tasks when a task is deleted from another tab/device
-      revalidateTasks();
+    onTaskDeleted: useCallback((taskData: Record<string, unknown>) => {
+      // Optimistically remove the task from local state
+      revalidateTasks(
+        (currentTasks) => {
+          if (!currentTasks) return currentTasks;
+          const deletedTask = taskData as Task;
+          return currentTasks.filter(t => t.id !== deletedTask.id);
+        },
+        { revalidate: false } // Don't refetch from server
+      );
     }, [revalidateTasks]),
   });