Implement issue #100: Session persistence and recovery after restart

mark_interrupted_sessions() on startup, graceful shutdown marks active
sessions. resume_interrupted_session() replays last user message.
POST /api/sessions/{id}/resume-interrupted endpoint. Status transitions
in send_message (executing -> waiting_input/failed). 16 new tests.

Author: alexeygrigorev

backend/codehive/api/app.py

@@ -1,6 +1,7 @@
 """FastAPI application factory."""
 
 import contextlib
+import logging
 from collections.abc import AsyncGenerator
 
 from fastapi import Depends, FastAPI
@@ -12,6 +13,7 @@
 from codehive.api.errors import register_error_handling
 from codehive.logging import configure_logging
 from codehive.core.first_run import print_credentials, seed_first_run
+from codehive.core.session import mark_interrupted_sessions
 from codehive.db.session import async_session_factory
 from codehive.api.routes.approvals import approvals_router
 from codehive.api.routes.auth import auth_router
@@ -39,20 +41,34 @@
 from codehive.api.routes.transcript import transcript_router
 from codehive.api.ws import router as ws_router
 
+logger = logging.getLogger(__name__)
+
 
 def create_app() -> FastAPI:
     """Create and configure the FastAPI application."""
 
     @contextlib.asynccontextmanager
     async def lifespan(app: FastAPI) -> AsyncGenerator[None, None]:
-        """Run first-run setup on startup."""
+        """Run first-run setup and session recovery on startup; mark sessions on shutdown."""
         session_maker = async_session_factory()
         async with session_maker() as db:
             credentials = await seed_first_run(db)
             if credentials is not None:
                 print_credentials(credentials)
+
+        # Startup recovery: mark any sessions stuck in 'executing' as 'interrupted'
+        async with session_maker() as db:
+            count = await mark_interrupted_sessions(db)
+            logger.info("Startup recovery: %d session(s) marked as interrupted", count)
+
         yield
 
+        # Graceful shutdown: mark executing sessions as interrupted
+        async with session_maker() as db:
+            count = await mark_interrupted_sessions(db)
+            if count:
+                logger.info("Shutdown: marked %d executing session(s) as interrupted", count)
+
     settings = Settings()
     configure_logging(settings)
 

backend/codehive/api/routes/sessions.py

@@ -20,6 +20,7 @@
 from codehive.core.session import (
     InvalidStatusTransitionError,
     IssueNotFoundError,
+    NoUserMessageError,
     ProjectNotFoundError,
     SessionHasDependentsError,
     SessionNotFoundError,
@@ -29,6 +30,7 @@
     list_child_sessions,
     list_sessions,
     pause_session,
+    resume_interrupted_session,
     resume_session,
     update_session,
 )
@@ -160,6 +162,24 @@ async def resume_session_endpoint(
     return SessionRead.model_validate(session)
 
 
+@sessions_router.post("/{session_id}/resume-interrupted", response_model=SessionRead)
+async def resume_interrupted_endpoint(
+    session_id: uuid.UUID,
+    db: AsyncSession = Depends(get_db),
+) -> SessionRead:
+    """Resume an interrupted session by replaying the last user message."""
+    await _get_session_or_404(db, session_id)
+    try:
+        session, _last_message = await resume_interrupted_session(db, session_id)
+    except SessionNotFoundError:
+        raise HTTPException(status_code=404, detail="Session not found")
+    except InvalidStatusTransitionError as exc:
+        raise HTTPException(status_code=409, detail=str(exc))
+    except NoUserMessageError as exc:
+        raise HTTPException(status_code=409, detail=str(exc))
+    return SessionRead.model_validate(session)
+
+
 @sessions_router.post("/{session_id}/switch-mode", response_model=SessionRead)
 async def switch_mode_endpoint(
     session_id: uuid.UUID,
@@ -355,15 +375,34 @@ async def send_message_endpoint(
     if session is None:
         raise HTTPException(status_code=404, detail="Session not found")
 
+    # Mark session as executing
+    try:
+        await update_session(db, session_id, status="executing")
+    except SessionNotFoundError:
+        raise HTTPException(status_code=404, detail="Session not found")
+
     try:
         engine = await _build_engine(session.config, engine_type=session.engine)
 
         events: list[dict[str, Any]] = []
         async for event in engine.send_message(session_id, body.content, db=db):
             events.append(event)
 
+        # Engine finished a turn -- mark as waiting_input
+        await update_session(db, session_id, status="waiting_input")
+
         return events
     except HTTPException:
+        # On HTTP errors, mark as failed
+        try:
+            await update_session(db, session_id, status="failed")
+        except Exception:
+            pass
         raise
     except Exception as exc:
+        # On unexpected errors, mark as failed
+        try:
+            await update_session(db, session_id, status="failed")
+        except Exception:
+            pass
         raise HTTPException(status_code=500, detail=str(exc))

backend/codehive/core/session.py

@@ -1,15 +1,18 @@
 """Session business logic (DB queries, state machine)."""
 
+import logging
 import uuid
 from datetime import datetime, timezone
 
-from sqlalchemy import select
+from sqlalchemy import select, update
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from codehive.api.schemas.session import QueueEmptyAction
-from codehive.db.models import Issue, Project
+from codehive.db.models import Issue, Message, Project
 from codehive.db.models import Session as SessionModel
 
+logger = logging.getLogger(__name__)
+
 
 class ProjectNotFoundError(Exception):
     """Raised when a project_id does not exist."""
@@ -31,6 +34,10 @@ class InvalidStatusTransitionError(Exception):
     """Raised when a status transition is not allowed."""
 
 
+class NoUserMessageError(Exception):
+    """Raised when an interrupted session has no user messages to replay."""
+
+
 def _validate_queue_empty_action(config: dict | None) -> None:
     """Raise ValueError if config contains an invalid queue_empty_action."""
     if config is None:
@@ -237,3 +244,61 @@ async def resume_session(
     await db.commit()
     await db.refresh(session)
     return session
+
+
+async def mark_interrupted_sessions(db: AsyncSession) -> int:
+    """Bulk-update all sessions with status ``executing`` to ``interrupted``.
+
+    Returns the number of sessions that were updated.
+    """
+    result = await db.execute(
+        update(SessionModel).where(SessionModel.status == "executing").values(status="interrupted")
+    )
+    await db.commit()
+    count = result.rowcount  # type: ignore[union-attr]
+    if count:
+        logger.info("Marked %d executing session(s) as interrupted", count)
+    return count
+
+
+async def resume_interrupted_session(
+    db: AsyncSession,
+    session_id: uuid.UUID,
+) -> tuple[SessionModel, str]:
+    """Resume an interrupted session.
+
+    Validates that the session status is ``interrupted``, fetches the last
+    user message, and transitions the status to ``executing``.
+
+    Returns a tuple of (session, last_user_message_content).
+
+    Raises:
+        SessionNotFoundError: if the session does not exist
+        InvalidStatusTransitionError: if the session is not in ``interrupted`` status
+        NoUserMessageError: if there are no user messages in the session
+    """
+    session = await db.get(SessionModel, session_id)
+    if session is None:
+        raise SessionNotFoundError(f"Session {session_id} not found")
+
+    if session.status != "interrupted":
+        raise InvalidStatusTransitionError(
+            f"Cannot resume-interrupted session in '{session.status}' status. "
+            f"Resume-interrupted is only allowed from: interrupted"
+        )
+
+    # Fetch the last user message
+    result = await db.execute(
+        select(Message)
+        .where(Message.session_id == session_id, Message.role == "user")
+        .order_by(Message.created_at.desc())
+        .limit(1)
+    )
+    last_msg = result.scalars().first()
+    if last_msg is None:
+        raise NoUserMessageError(f"Session {session_id} has no user messages to replay")
+
+    session.status = "executing"
+    await db.commit()
+    await db.refresh(session)
+    return session, last_msg.content