feat(sessions): Implement Event Sourcing Session Summary v2

.claude/hooks/devstream/memory/post_tool_use.py

@@ -15,6 +15,7 @@
 import json
 import re
 import time
+import os
 from pathlib import Path
 from datetime import datetime
 from typing import Optional, Dict, Any, List
@@ -45,6 +46,15 @@
     PROTOCOL_SYNC_AVAILABLE = False
     _SYNC_IMPORT_ERROR = str(e)
 
+# Event Sourcing Session Log imports (Phase 3 Integration)
+try:
+    sys.path.insert(0, str(Path(__file__).parent.parent))
+    from sessions.session_event_log import get_session_log
+    SESSION_EVENT_LOG_AVAILABLE = True
+except ImportError as e:
+    SESSION_EVENT_LOG_AVAILABLE = False
+    _EVENT_LOG_IMPORT_ERROR = str(e)
+
 
 class PostToolUseHook:
     """
@@ -765,12 +775,81 @@ async def _get_current_session_id(self) -> Optional[str]:
             self.base.debug_log(f"Failed to get session ID: {e}")
             return None
 
+    async def _get_active_files(self, session_id: str) -> List[str]:
+        """
+        Get current active_files list from session.
+
+        Context7 Pattern: Read-only helper using aiosqlite async with.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            List of active file paths (empty list if session not found)
+        """
+        try:
+            import aiosqlite
+
+            async with aiosqlite.connect(self.db_path) as db:
+                async with db.execute(
+                    "SELECT active_files FROM work_sessions WHERE id = ?",
+                    (session_id,)
+                ) as cursor:
+                    row = await cursor.fetchone()
+
+                    if not row:
+                        self.base.debug_log(f"Session not found: {session_id[:8]}...")
+                        return []
+
+                    # Parse JSON (handle NULL case)
+                    return json.loads(row[0]) if row[0] else []
+
+        except Exception as e:
+            self.base.debug_log(f"Failed to get active files: {e}")
+            return []
+
+    async def _get_active_tasks(self, session_id: str) -> List[str]:
+        """
+        Get current active_tasks list from session.
+
+        Context7 Pattern: Read-only helper using aiosqlite async with.
+
+        Args:
+            session_id: Session identifier
+
+        Returns:
+            List of active task IDs/titles (empty list if session not found)
+        """
+        try:
+            import aiosqlite
+
+            async with aiosqlite.connect(self.db_path) as db:
+                async with db.execute(
+                    "SELECT active_tasks FROM work_sessions WHERE id = ?",
+                    (session_id,)
+                ) as cursor:
+                    row = await cursor.fetchone()
+
+                    if not row:
+                        self.base.debug_log(f"Session not found: {session_id[:8]}...")
+                        return []
+
+                    # Parse JSON (handle NULL case)
+                    return json.loads(row[0]) if row[0] else []
+
+        except Exception as e:
+            self.base.debug_log(f"Failed to get active tasks: {e}")
+            return []
+
     async def _add_active_file(self, session_id: str, file_path: str) -> bool:
         """
         Add file to session's active_files list (with deduplication).
 
         Memory Bank Pattern: Track files ACTIVELY modified during session.
 
+        DEPRECATED: Use update_session_tracking() with WorkSessionManager instead.
+        Kept for backward compatibility only.
+
         Args:
             session_id: Session identifier
             file_path: Path to file being modified
@@ -889,7 +968,10 @@ async def update_session_tracking(
         tool_input: Dict[str, Any]
     ) -> None:
         """
-        Update work_sessions with active files and tasks (Memory Bank pattern).
+        Update work_sessions with active files and tasks via WorkSessionManager.
+
+        Context7 Pattern: Delegates to WorkSessionManager.update_session_progress()
+        instead of direct database writes for proper abstraction layer.
 
         Called after memory storage to track active work in current session.
         Non-blocking - failures logged but don't affect hook execution.
@@ -899,7 +981,7 @@ async def update_session_tracking(
             tool_input: Tool input parameters
 
         Note:
-            Tracks:
+            Tracks via WorkSessionManager:
             - Write/Edit/MultiEdit → active_files
             - TodoWrite → active_tasks (from in_progress todos)
             - MCP devstream_update_task → active_tasks
@@ -911,22 +993,67 @@ async def update_session_tracking(
                 self.base.debug_log("No active session - skip tracking")
                 return
 
+            # Initialize WorkSessionManager for proper session updates
+            import sys
+            from pathlib import Path
+            sys.path.insert(0, str(Path(__file__).parent.parent / 'sessions'))
+            from work_session_manager import WorkSessionManager
+
+            session_manager = WorkSessionManager()
+
             # Track active files (Write/Edit/MultiEdit)
             if tool_name in ["Write", "Edit", "MultiEdit"]:
                 file_path = tool_input.get("file_path")
                 if file_path:
-                    await self._add_active_file(session_id, file_path)
+                    # Get current active_files
+                    current_files = await self._get_active_files(session_id)
+
+                    # Add new file if not already tracked
+                    if file_path not in current_files:
+                        current_files.append(file_path)
+
+                        # DISABLED: WorkSessionManager.update_session_progress() doesn't accept active_files
+                        # Event Sourcing captures this via capture_session_event() instead
+                        # await session_manager.update_session_progress(
+                        #     session_id=session_id,
+                        #     active_files=current_files
+                        # )
+
+                        self.base.debug_log(
+                            f"Updated active_files via WorkSessionManager: {file_path} "
+                            f"(total: {len(current_files)})"
+                        )
 
             # Track active tasks (TodoWrite)
             elif tool_name == "TodoWrite":
                 todos = tool_input.get("todos", [])
+
+                # Get current active_tasks
+                current_tasks = await self._get_active_tasks(session_id)
+
+                tasks_updated = False
                 for todo in todos:
                     # Track in_progress todos (actively being worked on)
                     if todo.get("status") == "in_progress":
                         task_content = todo.get("content", "")
-                        # Use content as task_id (or extract ID if available)
-                        if task_content:
-                            await self._add_active_task(session_id, task_content)
+
+                        # Add if not already tracked
+                        if task_content and task_content not in current_tasks:
+                            current_tasks.append(task_content)
+                            tasks_updated = True
+
+                # DISABLED: WorkSessionManager.update_session_progress() doesn't accept active_tasks
+                # Event Sourcing captures this via capture_session_event() instead
+                # if tasks_updated:
+                #     await session_manager.update_session_progress(
+                #         session_id=session_id,
+                #         active_tasks=current_tasks
+                #     )
+
+                    self.base.debug_log(
+                        f"Updated active_tasks via WorkSessionManager: "
+                        f"{len(current_tasks)} tasks"
+                    )
 
             # Track MCP task operations (devstream_update_task, devstream_create_task)
             # Note: These are called via MCP, not directly as tool_name
@@ -979,6 +1106,104 @@ def log_capture_audit(
         # with open(audit_file, "a") as f:
         #     f.write(json.dumps(audit_entry) + "\n")
 
+    async def capture_session_event(
+        self,
+        tool_name: str,
+        tool_input: Dict[str, Any],
+        tool_response: Dict[str, Any]
+    ) -> None:
+        """
+        Capture session events for Event Sourcing session summary.
+
+        Phase 3 Integration: Capture events in append-only log for session_end_v2.py.
+        Non-blocking - failures logged but don't affect hook execution.
+
+        Args:
+            tool_name: Name of the tool executed
+            tool_input: Tool input parameters
+            tool_response: Tool execution response
+        """
+        self.base.debug_log(f"🎯 capture_session_event called: tool={tool_name}, SESSION_EVENT_LOG_AVAILABLE={SESSION_EVENT_LOG_AVAILABLE}")
+
+        if not SESSION_EVENT_LOG_AVAILABLE:
+            # Event log not available - skip silently
+            self.base.debug_log("❌ SESSION_EVENT_LOG_AVAILABLE=False, skipping event capture")
+            return
+
+        try:
+            # Get session ID from environment or tool input
+            session_id = os.environ.get("CLAUDE_SESSION_ID")
+            if not session_id:
+                # Try to extract from tool input if available
+                session_id = tool_input.get("session_id", "sess-unknown")
+
+            self.base.debug_log(f"🎯 Event capture: session_id={session_id}")
+
+            # Get session event log
+            event_log = await get_session_log(session_id)
+            self.base.debug_log(f"🎯 Event log retrieved: {event_log.session_id}, events={len(event_log.events)}")
+
+            # Capture events based on tool type
+            if tool_name in ["Write", "Edit", "MultiEdit"]:
+                # File modification events
+                file_path = tool_input.get("file_path", "")
+                content = tool_input.get("content", "") or tool_input.get("new_string", "")
+
+                if file_path and content:
+                    self.base.debug_log(f"🎯 Recording file_modified event: {file_path}")
+                    await event_log.record_event("file_modified", {
+                        "path": str(file_path),
+                        "tool": tool_name,
+                        "size_bytes": len(content),
+                        "session_id": session_id
+                    })
+                    self.base.debug_log(f"✅ file_modified event recorded, total events: {len(event_log.events)}")
+
+            elif tool_name == "TodoWrite":
+                # Task events - check for task completion
+                todos = tool_input.get("todos", [])
+
+                for todo in todos:
+                    todo_content = todo.get("content", "")
+                    todo_status = todo.get("status", "")
+
+                    if todo_content:
+                        if todo_status == "completed":
+                            await event_log.record_event("task_completed", {
+                                "task_id": f"todo-{hash(todo_content) % 10000}",
+                                "title": todo_content[:100],  # Limit title length
+                                "session_id": session_id
+                            })
+                        elif todo_status == "in_progress":
+                            await event_log.record_event("task_started", {
+                                "task_id": f"todo-{hash(todo_content) % 10000}",
+                                "title": todo_content[:100],
+                                "session_id": session_id
+                            })
+
+            elif tool_name == "Bash":
+                # Error events for failed commands
+                if not tool_response.get("success", True):
+                    command = tool_input.get("command", "")
+                    error_output = tool_response.get("error", "") or tool_response.get("output", "")
+
+                    if command:
+                        await event_log.record_event("error", {
+                            "error_type": "bash_command",
+                            "message": f"Command failed: {command[:100]}",
+                            "command": command[:200],
+                            "output": error_output[:200] if error_output else "",
+                            "session_id": session_id
+                        })
+
+            # TODO: Add more event types as needed
+            # - Decision events (could be extracted from comments)
+            # - Learning events (could be extracted from documentation)
+
+        except Exception as e:
+            # Non-blocking - log but don't fail the hook
+            self.base.debug_log(f"Event capture failed (non-blocking): {e}")
+
     async def process(self, context: PostToolUseContext) -> None:
         """
         Main hook processing logic - Enhanced multi-tool capture with Protocol State Sync (FASE 2).
@@ -1052,6 +1277,10 @@ async def process(self, context: PostToolUseContext) -> None:
 
         self.base.debug_log(f"Processing {tool_name}")
 
+        # Phase 3: Capture session events (Event Sourcing)
+        # Non-blocking - capture events before any other processing
+        await self.capture_session_event(tool_name, tool_input, tool_response)
+
         # Define critical tools that trigger checkpoints
         critical_tools = ["Write", "Edit", "MultiEdit", "Bash", "TodoWrite"]
         is_critical_tool = tool_name in critical_tools