diff --git a/shell_memory/sessions.py b/shell_memory/sessions.py
new file mode 100644
index 0000000..00767e4
--- /dev/null
+++ b/shell_memory/sessions.py
@@ -0,0 +1,113 @@
+"""Session recording and replay functionality."""
+
+import time
+import subprocess
+from typing import List, Optional, Dict, Any
+from datetime import datetime
+
+from .models import Session
+from .database import Database
+
+
+class SessionRecorder:
+    """Records and replays terminal sessions."""
+
+    def __init__(self, db: Database):
+        self.db = db
+        self.current_session = None
+        self.recorded_commands = []
+
+    def start_session(self, name: Optional[str] = None) -> Session:
+        session_name = name or f"Session {datetime.now().strftime('%Y%m%d_%H%M%S')}"
+        self.current_session = Session(
+            name=session_name,
+            commands=[],
+            start_time=datetime.now(),
+        )
+        self.recorded_commands = []
+        return self.current_session
+
+    def record_command(self, command: str, output: str = "", success: bool = True):
+        if self.current_session is None:
+            return
+
+        cmd_entry = {
+            "command": command,
+            "output": output,
+            "success": success,
+            "timestamp": datetime.now().isoformat(),
+        }
+        self.recorded_commands.append(cmd_entry)
+        self.current_session.commands = self.recorded_commands
+
+    def stop_session(self) -> Optional[Session]:
+        if self.current_session is None:
+            return None
+
+        self.current_session.end_time = datetime.now()
+        session_id = self.db.add_session(self.current_session)
+        self.current_session.id = session_id
+
+        session = self.current_session
+        self.current_session = None
+        self.recorded_commands = []
+
+        return session
+
+    def get_sessions(self) -> List[Session]:
+        return self.db.get_sessions()
+
+    def get_session(self, session_id: int) -> Optional[Session]:
+        return self.db.get_session(session_id)
+
+    def replay_session(self, session_id: int, dry_run: bool = False, delay: float = 0.5) -> List[Dict[str, Any]]:
+        session = self.db.get_session(session_id)
+        if session is None:
+            return []
+
+        results = []
+        for cmd_entry in session.commands:
+            result = {
+                "command": cmd_entry["command"],
+                "executed": False,
+                "output": "",
+                "success": False,
+            }
+
+            if not dry_run:
+                try:
+                    proc = subprocess.run(
+                        cmd_entry["command"],
+                        shell=True,
+                        capture_output=True,
+                        text=True,
+                        timeout=30,
+                    )
+                    result["output"] = proc.stdout + proc.stderr
+                    result["success"] = proc.returncode == 0
+                    result["executed"] = True
+                except subprocess.TimeoutExpired:
+                    result["output"] = "Command timed out"
+                except Exception as e:
+                    result["output"] = str(e)
+
+            results.append(result)
+
+            if delay > 0 and not dry_run:
+                time.sleep(delay)
+
+        return results
+
+    def export_session(self, session_id: int) -> Optional[str]:
+        session = self.db.get_session(session_id)
+        if session is None:
+            return None
+
+        lines = ["#!/bin/bash", "", f"# Session: {session.name}", f"# Recorded: {session.start_time.isoformat()}", ""]
+
+        for cmd_entry in session.commands:
+            lines.append(f"# {cmd_entry['timestamp']}")
+            lines.append(cmd_entry["command"])
+            lines.append("")
+
+        return "\n".join(lines)
\ No newline at end of file