DM20 Protocol

Instructions

Implementation Reference

• src/dm20_protocol/claudmaster/tools/session_tools.py:1076-1200 (handler)

  The 'end_session' tool (MCP tool) implementation, which ends or pauses a Claudmaster AI DM session and persists its state. Note: The tool name is 'end_session', and it corresponds to the permission 'end_claudmaster_session' registered in 'permissions.py'.

  async def end_session(
      session_id: str,
      mode: str = "pause",
      summary_notes: Optional[str] = None,
      campaign_path: Optional[str] = None,
  ) -> dict:
      """
      End or pause a Claudmaster session, saving all state.
  
      This MCP tool cleanly terminates or pauses a Claudmaster session.
      In "pause" mode, all state is persisted to disk so the session can
      be resumed later. In "end" mode, state is saved as a final snapshot
      and the session is terminated.
  
      Args:
          session_id: The session ID to end or pause
          mode: "pause" (save for later resumption) or "end" (final termination)
          summary_notes: Optional DM notes to save with the session snapshot
          campaign_path: Optional path for disk persistence.
              If provided, session state is written to disk under this path.
  
      Returns:
          Dictionary with the following keys:
              - status: "paused" or "ended" on success, "error" on failure
              - session_id: The session ID that was ended
              - session_summary: Brief summary of the session
              - save_path: Where state was persisted (if campaign_path provided)
              - stats: Session statistics (duration, turn count, etc.)
              - error_message: Error description if status is "error"
  
      Examples:
          Pause a session for later:
          >>> result = await end_session(
          ...     session_id="abc123",
          ...     mode="pause",
          ...     summary_notes="Party just entered the dungeon"
          ... )
  
          End a session permanently:
          >>> result = await end_session(session_id="abc123", mode="end")
  
          Pause with disk persistence:
          >>> result = await end_session(
          ...     session_id="abc123",
          ...     mode="pause",
          ...     campaign_path="/data/campaigns/my-campaign"
          ... )
      """
      try:
          # Validate mode
          if mode not in ("pause", "end"):
              return {
                  "status": "error",
                  "session_id": session_id,
                  "error_message": f"*The DM tilts their head, puzzled*\n\nI'm not familiar with the '{mode}' command. When ending a session, please choose either 'pause' (to resume later) or 'end' (to conclude permanently).",
              }
  
          # Check session exists
          if session_id not in _session_manager._active_sessions:
              return {
                  "status": "error",
                  "session_id": session_id,
                  "error_message": _error_formatter.format_session_not_found(session_id),
              }
  
          # Get session info before ending
          orchestrator, session = _session_manager._active_sessions[session_id]
          turn_count = session.turn_count
          started_at = session.started_at
          campaign_id = session.campaign_id
  
          # Calculate duration
          duration_minutes = int((datetime.now() - started_at).total_seconds() / 60)
  
          # Build stats
          stats = {
              "turn_count": turn_count,
              "duration_minutes": duration_minutes,
              "started_at": started_at.isoformat(),
              "ended_at": datetime.now().isoformat(),
          }
  
          # Save to memory first (for resume capability)
          _session_manager.save_session(session_id)
  
          # Persist to disk if campaign_path provided
          save_path = None
          if campaign_path:
              serializer = SessionSerializer(Path(campaign_path))
              saved_data = _session_manager._saved_sessions.get(session_id, {})
              save_dir = serializer.save_session(
                  session_data=saved_data,
                  mode=mode,
                  summary_notes=summary_notes,
              )
              save_path = str(save_dir)
  
          # End the session (removes from active)
          result_status = "paused" if mode == "pause" else "ended"
          _session_manager.end_session(session_id)
  
          # Build summary
          session_summary = (
              f"Session {session_id} for campaign '{campaign_id}': "
              f"{turn_count} turns over {duration_minutes} minutes."
          )
  
          logger.info(f"MCP end_session: {result_status} session {session_id}")
  
          return {
              "status": result_status,
              "session_id": session_id,
              "session_summary": session_summary,
              "save_path": save_path,
              "stats": stats,
          }
  
      except Exception as e:
          logger.error(f"Error in end_session for {session_id}: {e}", exc_info=True)
          return {
              "status": "error",
              "session_id": session_id,
              "error_message": _error_formatter.format_error(e),
          }