diff --git a/src/mudlib/if_session.py b/src/mudlib/if_session.py
new file mode 100644
index 0000000..b9cb116
--- /dev/null
+++ b/src/mudlib/if_session.py
@@ -0,0 +1,105 @@
+"""Interactive fiction session management via dfrotz subprocess."""
+
+import asyncio
+from dataclasses import dataclass
+from pathlib import Path
+
+
+@dataclass
+class IFResponse:
+    """Response from IF session input handling."""
+
+    output: str
+    done: bool
+
+
+class IFSession:
+    """Manages an interactive fiction session via dfrotz subprocess."""
+
+    def __init__(self, player, story_path: str, game_name: str = ""):
+        self.player = player
+        self.story_path = story_path
+        self.game_name = game_name or Path(story_path).stem
+        self.process: asyncio.subprocess.Process | None = None
+
+    async def start(self) -> str:
+        """Spawn dfrotz and return intro text."""
+        self.process = await asyncio.create_subprocess_exec(
+            "dfrotz",
+            "-p",  # plain ASCII, no formatting
+            "-w",
+            "80",  # screen width
+            "-m",  # no MORE prompts
+            self.story_path,
+            stdin=asyncio.subprocess.PIPE,
+            stdout=asyncio.subprocess.PIPE,
+            stderr=asyncio.subprocess.PIPE,
+        )
+        # Read intro text until we see the prompt
+        intro = await self._read_response()
+        return intro
+
+    async def handle_input(self, text: str) -> IFResponse:
+        """Handle player input. Route to dfrotz or handle escape commands."""
+        # Check for escape commands (:: prefix)
+        if text == "::quit":
+            return IFResponse(output="", done=True)
+
+        if text == "::help":
+            help_text = """escape commands:
+::quit - exit the game
+::help - show this help"""
+            return IFResponse(output=help_text, done=False)
+
+        # Regular game input - send to dfrotz
+        if self.process and self.process.stdin:
+            stripped = text.strip()
+            self.process.stdin.write(f"{stripped}\n".encode())
+            await self.process.stdin.drain()
+
+            # Read response
+            output = await self._read_response()
+            return IFResponse(output=output, done=False)
+
+        # Process not running
+        return IFResponse(output="error: game not running", done=True)
+
+    async def stop(self):
+        """Terminate dfrotz process."""
+        if self.process and self.process.returncode is None:
+            self.process.terminate()
+            await self.process.wait()
+
+    async def _read_response(self) -> str:
+        """Read dfrotz output until the '>' prompt appears."""
+        if not self.process or not self.process.stdout:
+            return ""
+
+        output = []
+        try:
+            # Read byte by byte until we see "\n>"
+            while True:
+                byte = await asyncio.wait_for(self.process.stdout.read(1), timeout=5.0)
+                if not byte:
+                    break
+
+                char = byte.decode("latin-1")
+                output.append(char)
+
+                # Check if we've hit the prompt
+                # Prompt is "\n>" or just ">" at start
+                if len(output) >= 2:
+                    if output[-2] == "\n" and output[-1] == ">":
+                        # Strip the trailing "\n>"
+                        output = output[:-2]
+                        break
+                elif len(output) == 1 and output[0] == ">":
+                    # Prompt at very start
+                    output = []
+                    break
+        except TimeoutError:
+            # If we timeout, return what we got
+            pass
+
+        result = "".join(output)
+        return result.rstrip()
diff --git a/tests/test_if_session.py b/tests/test_if_session.py
new file mode 100644
index 0000000..a7f0db8
--- /dev/null
+++ b/tests/test_if_session.py
@@ -0,0 +1,292 @@
+"""Tests for IF session management."""
+
+from unittest.mock import AsyncMock, MagicMock, patch
+
+import pytest
+
+from mudlib.if_session import IFResponse, IFSession
+
+
+@pytest.mark.asyncio
+async def test_if_response_dataclass():
+    """IFResponse dataclass can be created."""
+    response = IFResponse(output="test output", done=False)
+    assert response.output == "test output"
+    assert response.done is False
+
+
+@pytest.mark.asyncio
+async def test_if_response_done():
+    """IFResponse can signal completion."""
+    response = IFResponse(output="", done=True)
+    assert response.done is True
+
+
+@pytest.mark.asyncio
+async def test_if_session_init():
+    """IFSession can be initialized."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5", "story")
+    assert session.player == player
+    assert session.story_path == "/path/to/story.z5"
+    assert session.game_name == "story"
+    assert session.process is None
+
+
+@pytest.mark.asyncio
+async def test_if_session_init_infers_game_name():
+    """IFSession infers game_name from story_path if not provided."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/zork.z5")
+    assert session.game_name == "zork"
+
+
+@pytest.mark.asyncio
+async def test_start_spawns_subprocess_and_returns_intro():
+    """start() spawns dfrotz subprocess and returns intro text."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+
+    # Mock the subprocess
+    mock_process = AsyncMock()
+    mock_process.stdout = AsyncMock()
+    mock_process.stdin = AsyncMock()
+
+    # Simulate dfrotz output: intro text followed by ">" prompt
+    intro_bytes = b"Welcome to the story!\nYou are in a room.\n>"
+
+    async def read_side_effect(n):
+        nonlocal intro_bytes
+        if intro_bytes:
+            byte = intro_bytes[:1]
+            intro_bytes = intro_bytes[1:]
+            return byte
+        return b""
+
+    mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
+
+    with patch("asyncio.create_subprocess_exec", return_value=mock_process):
+        intro = await session.start()
+
+    assert session.process == mock_process
+    assert "Welcome to the story!" in intro
+    assert "You are in a room." in intro
+    # The prompt should be stripped from the output
+    assert intro.strip().endswith("room.")
+
+
+@pytest.mark.asyncio
+async def test_handle_input_sends_to_dfrotz():
+    """handle_input() sends regular input to dfrotz and returns response."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+
+    # Mock process
+    mock_process = MagicMock()
+    mock_process.stdin = MagicMock()
+    mock_process.stdin.write = MagicMock()
+    mock_process.stdin.drain = AsyncMock()
+    mock_process.stdout = AsyncMock()
+    session.process = mock_process
+
+    # Simulate dfrotz response
+    response_bytes = b"You move north.\n>"
+
+    async def read_side_effect(n):
+        nonlocal response_bytes
+        if response_bytes:
+            byte = response_bytes[:1]
+            response_bytes = response_bytes[1:]
+            return byte
+        return b""
+
+    mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
+
+    response = await session.handle_input("north")
+
+    assert response.output == "You move north."
+    assert response.done is False
+    # Verify stdin.write was called
+    mock_process.stdin.write.assert_called()
+    mock_process.stdin.drain.assert_called()
+
+
+@pytest.mark.asyncio
+async def test_handle_input_quit_returns_done():
+    """handle_input('::quit') returns done=True without sending to dfrotz."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+
+    # Mock process
+    mock_process = AsyncMock()
+    mock_process.stdin = AsyncMock()
+    session.process = mock_process
+
+    response = await session.handle_input("::quit")
+
+    assert response.done is True
+    # Should NOT have written to dfrotz stdin
+    mock_process.stdin.write.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_handle_input_help_returns_help_text():
+    """handle_input('::help') returns help text listing escape commands."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+    session.process = AsyncMock()
+
+    response = await session.handle_input("::help")
+
+    assert response.done is False
+    assert "::quit" in response.output
+    assert "::help" in response.output
+
+
+@pytest.mark.asyncio
+async def test_stop_terminates_subprocess():
+    """stop() terminates the dfrotz process."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+
+    # Mock process
+    mock_process = MagicMock()
+    mock_process.returncode = None
+    mock_process.terminate = MagicMock()
+    mock_process.wait = AsyncMock()
+    session.process = mock_process
+
+    await session.stop()
+
+    mock_process.terminate.assert_called_once()
+    mock_process.wait.assert_called_once()
+
+
+@pytest.mark.asyncio
+async def test_stop_when_no_process():
+    """stop() does nothing if process is None."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+    session.process = None
+
+    # Should not raise
+    await session.stop()
+
+
+@pytest.mark.asyncio
+async def test_stop_when_already_terminated():
+    """stop() handles already-terminated process gracefully."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+
+    # Mock process that's already done
+    mock_process = AsyncMock()
+    mock_process.returncode = 0
+    session.process = mock_process
+
+    await session.stop()
+
+    # Should not call terminate on already-finished process
+    mock_process.terminate.assert_not_called()
+
+
+@pytest.mark.asyncio
+async def test_read_response_detects_prompt():
+    """_read_response() reads until '>' prompt appears."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+
+    # Mock process
+    mock_process = AsyncMock()
+    mock_process.stdout = AsyncMock()
+    session.process = mock_process
+
+    # Simulate multi-line output with prompt
+    output_bytes = b"Line 1\nLine 2\nLine 3\n>"
+
+    async def read_side_effect(n):
+        nonlocal output_bytes
+        if output_bytes:
+            byte = output_bytes[:1]
+            output_bytes = output_bytes[1:]
+            return byte
+        return b""
+
+    mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
+
+    result = await session._read_response()
+
+    assert "Line 1" in result
+    assert "Line 2" in result
+    assert "Line 3" in result
+    # Prompt should be stripped
+    assert not result.strip().endswith(">")
+
+
+@pytest.mark.asyncio
+async def test_handle_input_strips_whitespace():
+    """handle_input() strips input before sending to dfrotz."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+
+    # Mock process
+    mock_process = MagicMock()
+    mock_process.stdin = MagicMock()
+    mock_process.stdin.write = MagicMock()
+    mock_process.stdin.drain = AsyncMock()
+    mock_process.stdout = AsyncMock()
+    session.process = mock_process
+
+    # Simulate response
+    response_bytes = b"ok\n>"
+
+    async def read_side_effect(n):
+        nonlocal response_bytes
+        if response_bytes:
+            byte = response_bytes[:1]
+            response_bytes = response_bytes[1:]
+            return byte
+        return b""
+
+    mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
+
+    await session.handle_input("  look  ")
+
+    # Check that write was called with stripped input + newline
+    calls = mock_process.stdin.write.call_args_list
+    assert len(calls) == 1
+    written = calls[0][0][0]
+    assert written == b"look\n"
+
+
+@pytest.mark.asyncio
+async def test_handle_input_empty_string():
+    """handle_input() with empty string sends newline to dfrotz."""
+    player = MagicMock()
+    session = IFSession(player, "/path/to/story.z5")
+
+    # Mock process
+    mock_process = MagicMock()
+    mock_process.stdin = MagicMock()
+    mock_process.stdin.write = MagicMock()
+    mock_process.stdin.drain = AsyncMock()
+    mock_process.stdout = AsyncMock()
+    session.process = mock_process
+
+    # Simulate response
+    response_bytes = b"ok\n>"
+
+    async def read_side_effect(n):
+        nonlocal response_bytes
+        if response_bytes:
+            byte = response_bytes[:1]
+            response_bytes = response_bytes[1:]
+            return byte
+        return b""
+
+    mock_process.stdout.read = AsyncMock(side_effect=read_side_effect)
+
+    await session.handle_input("")
+
+    # Should still write a newline
+    mock_process.stdin.write.assert_called()