diff --git a/src/mudlib/server.py b/src/mudlib/server.py
index 864af57..f66540e 100644
--- a/src/mudlib/server.py
+++ b/src/mudlib/server.py
@@ -20,6 +20,15 @@ from mudlib.combat.engine import process_combat
 from mudlib.content import load_commands
 from mudlib.effects import clear_expired
 from mudlib.player import Player, players
+from mudlib.store import (
+    account_exists,
+    authenticate,
+    create_account,
+    init_db,
+    load_player_data,
+    save_player,
+    update_last_login,
+)
 from mudlib.world.terrain import World
 
 log = logging.getLogger(__name__)
@@ -27,6 +36,7 @@ log = logging.getLogger(__name__)
 PORT = 6789
 TICK_RATE = 10  # ticks per second
 TICK_INTERVAL = 1.0 / TICK_RATE
+AUTOSAVE_INTERVAL = 60.0  # seconds between auto-saves
 
 # Module-level world instance, generated once at startup
 _world: World | None = None
@@ -43,10 +53,23 @@ def load_world_config(world_name: str = "earth") -> dict:
 async def game_loop() -> None:
     """Run periodic game tasks at TICK_RATE ticks per second."""
     log.info("game loop started (%d ticks/sec)", TICK_RATE)
+    last_save_time = time.monotonic()
+
     while True:
         t0 = asyncio.get_event_loop().time()
         clear_expired()
         process_combat()
+
+        # Periodic auto-save (every 60 seconds)
+        current_time = time.monotonic()
+        if current_time - last_save_time >= AUTOSAVE_INTERVAL:
+            player_count = len(players)
+            if player_count > 0:
+                log.debug("auto-saving %d players", player_count)
+                for player in list(players.values()):
+                    save_player(player)
+                last_save_time = current_time
+
         elapsed = asyncio.get_event_loop().time() - t0
         sleep_time = TICK_INTERVAL - elapsed
         if sleep_time > 0:
@@ -85,6 +108,83 @@ def find_passable_start(world: World, start_x: int, start_y: int) -> tuple[int,
     return start_x, start_y
 
 
+async def handle_login(
+    name: str,
+    read_func,
+    write_func,
+) -> dict:
+    """Handle login or registration for a player.
+
+    Args:
+        name: Player name
+        read_func: Async function to read input
+        write_func: Async function to write output
+
+    Returns:
+        Dict with 'success' (bool) and 'player_data' (dict or None)
+    """
+    if account_exists(name):
+        # Existing account - authenticate
+        max_attempts = 3
+        for attempt in range(max_attempts):
+            await write_func("Password: ")
+            password = await read_func()
+
+            if password is None or not password.strip():
+                return {"success": False, "player_data": None}
+
+            if authenticate(name, password.strip()):
+                # Success - load player data
+                player_data = load_player_data(name)
+                return {"success": True, "player_data": player_data}
+
+            remaining = max_attempts - attempt - 1
+            if remaining > 0:
+                msg = f"Incorrect password. {remaining} attempts remaining.\r\n"
+                await write_func(msg)
+            else:
+                await write_func("Too many failed attempts.\r\n")
+
+        return {"success": False, "player_data": None}
+
+    # New account - registration
+    await write_func(f"Account '{name}' does not exist. Create new account? (y/n) ")
+    response = await read_func()
+
+    if response is None or response.strip().lower() != "y":
+        return {"success": False, "player_data": None}
+
+    # Get password with confirmation
+    while True:
+        await write_func("Choose a password: ")
+        password1 = await read_func()
+
+        if password1 is None or not password1.strip():
+            return {"success": False, "player_data": None}
+
+        await write_func("Confirm password: ")
+        password2 = await read_func()
+
+        if password2 is None or not password2.strip():
+            return {"success": False, "player_data": None}
+
+        if password1.strip() == password2.strip():
+            # Passwords match - create account
+            if create_account(name, password1.strip()):
+                await write_func("Account created successfully!\r\n")
+                # Return default data for new account
+                player_data = load_player_data(name)
+                return {"success": True, "player_data": player_data}
+            await write_func("Failed to create account.\r\n")
+            return {"success": False, "player_data": None}
+
+        # Passwords don't match - retry or cancel
+        await write_func("Passwords do not match. Try again? (y/n) ")
+        retry = await read_func()
+        if retry is None or retry.strip().lower() != "y":
+            return {"success": False, "player_data": None}
+
+
 async def shell(
     reader: telnetlib3.TelnetReader | telnetlib3.TelnetReaderUnicode,
     writer: telnetlib3.TelnetWriter | telnetlib3.TelnetWriterUnicode,
@@ -108,23 +208,72 @@ async def shell(
 
     player_name = name_input.strip()
 
-    # Find a passable starting position (start at world center)
-    center_x = _world.width // 2
-    center_y = _world.height // 2
-    start_x, start_y = find_passable_start(_world, center_x, center_y)
+    # Handle login/registration
+    async def read_input():
+        result = await readline2(_reader, _writer)
+        return result
 
-    # Create player
+    async def write_output(msg: str):
+        _writer.write(msg)
+        await _writer.drain()
+
+    login_result = await handle_login(player_name, read_input, write_output)
+
+    if not login_result["success"]:
+        _writer.write("Login failed. Disconnecting.\r\n")
+        await _writer.drain()
+        _writer.close()
+        return
+
+    # Update last login timestamp
+    update_last_login(player_name)
+
+    # Load player data from database or use defaults for new player
+    player_data = login_result["player_data"]
+    if player_data is None:
+        # New player - find a passable starting position
+        center_x = _world.width // 2
+        center_y = _world.height // 2
+        start_x, start_y = find_passable_start(_world, center_x, center_y)
+        player_data = {
+            "x": start_x,
+            "y": start_y,
+            "pl": 100.0,
+            "stamina": 100.0,
+            "max_stamina": 100.0,
+            "flying": False,
+        }
+    else:
+        # Existing player - verify spawn position is still passable
+        if not _world.is_passable(player_data["x"], player_data["y"]):
+            # Saved position is no longer passable, find a new one
+            start_x, start_y = find_passable_start(
+                _world, player_data["x"], player_data["y"]
+            )
+            player_data["x"] = start_x
+            player_data["y"] = start_y
+
+    # Create player instance
     player = Player(
         name=player_name,
-        x=start_x,
-        y=start_y,
+        x=player_data["x"],
+        y=player_data["y"],
+        pl=player_data["pl"],
+        stamina=player_data["stamina"],
+        max_stamina=player_data["max_stamina"],
+        flying=player_data["flying"],
         writer=_writer,
         reader=_reader,
     )
 
     # Register player
     players[player_name] = player
-    log.info("%s connected at (%d, %d)", player_name, start_x, start_y)
+    log.info(
+        "%s connected at (%d, %d)",
+        player_name,
+        player_data["x"],
+        player_data["y"],
+    )
 
     _writer.write(f"\r\nWelcome, {player_name}!\r\n")
     _writer.write("Type 'help' for commands or 'quit' to disconnect.\r\n\r\n")
@@ -134,37 +283,45 @@ async def shell(
     await mudlib.commands.look.cmd_look(player, "")
 
     # Command loop
-    while not _writer.is_closing():
-        _writer.write("mud> ")
-        await _writer.drain()
+    try:
+        while not _writer.is_closing():
+            _writer.write("mud> ")
+            await _writer.drain()
 
-        inp = await readline2(_reader, _writer)
-        if inp is None:
-            break
+            inp = await readline2(_reader, _writer)
+            if inp is None:
+                break
 
-        command = inp.strip()
-        if not command:
-            continue
+            command = inp.strip()
+            if not command:
+                continue
 
-        # Dispatch command
-        await mudlib.commands.dispatch(player, command)
+            # Dispatch command
+            await mudlib.commands.dispatch(player, command)
 
-        # Check if writer was closed by quit command
-        if _writer.is_closing():
-            break
+            # Check if writer was closed by quit command
+            if _writer.is_closing():
+                break
+    finally:
+        # Save player state on disconnect (if not already saved by quit command)
+        if player_name in players:
+            save_player(player)
+            del players[player_name]
+            log.info("%s disconnected", player_name)
 
-    # Clean up: remove from registry if still present
-    if player_name in players:
-        del players[player_name]
-        log.info("%s disconnected", player_name)
-
-    _writer.close()
+        _writer.close()
 
 
 async def run_server() -> None:
     """Start the MUD telnet server."""
     global _world
 
+    # Initialize database
+    data_dir = pathlib.Path(__file__).resolve().parents[2] / "data"
+    db_path = data_dir / "mud.db"
+    log.info("initializing database at %s", db_path)
+    init_db(db_path)
+
     # Generate world once at startup (cached to build/ after first run)
     cache_dir = pathlib.Path(__file__).resolve().parents[2] / "build"
     config = load_world_config()
diff --git a/tests/test_login_flow.py b/tests/test_login_flow.py
new file mode 100644
index 0000000..9766e0a
--- /dev/null
+++ b/tests/test_login_flow.py
@@ -0,0 +1,164 @@
+"""Tests for login and registration flow."""
+
+import os
+import tempfile
+
+import pytest
+
+from mudlib.server import handle_login
+from mudlib.store import account_exists, authenticate, init_db
+
+
+@pytest.fixture
+def temp_db():
+    """Create a temporary database for testing."""
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f:
+        db_path = f.name
+
+    init_db(db_path)
+
+    yield db_path
+
+    # Cleanup
+    os.unlink(db_path)
+
+
+@pytest.mark.asyncio
+async def test_login_existing_account_correct_password(temp_db):
+    """Login succeeds with correct password."""
+    from mudlib.store import create_account
+
+    create_account("TestUser", "correct_password")
+
+    # Mock I/O
+    inputs = ["correct_password"]
+    outputs = []
+
+    async def mock_write(msg: str):
+        outputs.append(msg)
+
+    async def mock_read():
+        if inputs:
+            return inputs.pop(0)
+        return None
+
+    result = await handle_login("TestUser", mock_read, mock_write)
+
+    assert result["success"] is True
+    assert result["player_data"] is not None
+    assert result["player_data"]["x"] == 0
+    assert result["player_data"]["y"] == 0
+
+
+@pytest.mark.asyncio
+async def test_login_existing_account_wrong_password(temp_db):
+    """Login fails with wrong password after max attempts."""
+    from mudlib.store import create_account
+
+    create_account("TestUser", "correct_password")
+
+    # Mock I/O - provide wrong password 3 times
+    inputs = ["wrong1", "wrong2", "wrong3"]
+    outputs = []
+
+    async def mock_write(msg: str):
+        outputs.append(msg)
+
+    async def mock_read():
+        if inputs:
+            return inputs.pop(0)
+        return None
+
+    result = await handle_login("TestUser", mock_read, mock_write)
+
+    assert result["success"] is False
+    assert "Too many failed attempts" in "".join(outputs)
+
+
+@pytest.mark.asyncio
+async def test_login_existing_account_retry_success(temp_db):
+    """Login succeeds after retrying wrong password."""
+    from mudlib.store import create_account
+
+    create_account("TestUser", "correct_password")
+
+    # Mock I/O - wrong password first, then correct
+    inputs = ["wrong_password", "correct_password"]
+    outputs = []
+
+    async def mock_write(msg: str):
+        outputs.append(msg)
+
+    async def mock_read():
+        if inputs:
+            return inputs.pop(0)
+        return None
+
+    result = await handle_login("TestUser", mock_read, mock_write)
+
+    assert result["success"] is True
+    assert result["player_data"] is not None
+
+
+@pytest.mark.asyncio
+async def test_registration_new_account(temp_db):
+    """Registration creates a new account."""
+    # Mock I/O - answer 'y' to create account, then provide password twice
+    inputs = ["y", "newpassword", "newpassword"]
+    outputs = []
+
+    async def mock_write(msg: str):
+        outputs.append(msg)
+
+    async def mock_read():
+        if inputs:
+            return inputs.pop(0)
+        return None
+
+    result = await handle_login("NewUser", mock_read, mock_write)
+
+    assert result["success"] is True
+    assert account_exists("NewUser")
+    assert authenticate("NewUser", "newpassword")
+
+
+@pytest.mark.asyncio
+async def test_registration_password_mismatch(temp_db):
+    """Registration fails if passwords don't match."""
+    # Mock I/O - answer 'y', then mismatched passwords, then cancel
+    inputs = ["y", "password1", "password2", "n"]
+    outputs = []
+
+    async def mock_write(msg: str):
+        outputs.append(msg)
+
+    async def mock_read():
+        if inputs:
+            return inputs.pop(0)
+        return None
+
+    result = await handle_login("NewUser", mock_read, mock_write)
+
+    assert result["success"] is False
+    assert not account_exists("NewUser")
+
+
+@pytest.mark.asyncio
+async def test_registration_declined(temp_db):
+    """User declines to create account."""
+    # Mock I/O - answer 'n' to create account
+    inputs = ["n"]
+    outputs = []
+
+    async def mock_write(msg: str):
+        outputs.append(msg)
+
+    async def mock_read():
+        if inputs:
+            return inputs.pop(0)
+        return None
+
+    result = await handle_login("NewUser", mock_read, mock_write)
+
+    assert result["success"] is False
+    assert not account_exists("NewUser")
diff --git a/tests/test_server.py b/tests/test_server.py
index d086d0f..851f769 100644
--- a/tests/test_server.py
+++ b/tests/test_server.py
@@ -2,14 +2,31 @@
 
 import asyncio
 import contextlib
+import os
+import tempfile
 from unittest.mock import AsyncMock, MagicMock, patch
 
 import pytest
 
 from mudlib import server
+from mudlib.store import init_db
 from mudlib.world.terrain import World
 
 
+@pytest.fixture
+def temp_db():
+    """Create a temporary database for testing."""
+    with tempfile.NamedTemporaryFile(delete=False, suffix=".db") as f:
+        db_path = f.name
+
+    init_db(db_path)
+
+    yield db_path
+
+    # Cleanup
+    os.unlink(db_path)
+
+
 def test_port_constant():
     assert server.PORT == 6789
     assert isinstance(server.PORT, int)
@@ -34,8 +51,9 @@ def test_find_passable_start():
 
 
 @pytest.mark.asyncio
-async def test_shell_greets_and_accepts_commands():
-    server._world = World(seed=42, width=100, height=100)
+async def test_shell_greets_and_accepts_commands(temp_db):
+    world = World(seed=42, width=100, height=100)
+    server._world = world
 
     reader = AsyncMock()
     writer = MagicMock()
@@ -43,15 +61,32 @@ async def test_shell_greets_and_accepts_commands():
     writer.drain = AsyncMock()
     writer.close = MagicMock()
 
-    readline = "mudlib.server.readline2"
-    with patch(readline, new_callable=AsyncMock) as mock_readline:
-        mock_readline.side_effect = ["TestPlayer", "look", "quit"]
-        await server.shell(reader, writer)
+    # Need to mock the look command's world reference too
+    import mudlib.commands.look
 
-    calls = [str(call) for call in writer.write.call_args_list]
-    assert any("Welcome" in call for call in calls)
-    assert any("TestPlayer" in call for call in calls)
-    writer.close.assert_called()
+    original_world = mudlib.commands.look.world
+    mudlib.commands.look.world = world
+
+    try:
+        readline = "mudlib.server.readline2"
+        with patch(readline, new_callable=AsyncMock) as mock_readline:
+            # Simulate: name, create account (y), password, confirm password, look, quit
+            mock_readline.side_effect = [
+                "TestPlayer",
+                "y",
+                "password",
+                "password",
+                "look",
+                "quit",
+            ]
+            await server.shell(reader, writer)
+
+        calls = [str(call) for call in writer.write.call_args_list]
+        assert any("Welcome" in call for call in calls)
+        assert any("TestPlayer" in call for call in calls)
+        writer.close.assert_called()
+    finally:
+        mudlib.commands.look.world = original_world
 
 
 @pytest.mark.asyncio
@@ -73,8 +108,9 @@ async def test_shell_handles_eof():
 
 
 @pytest.mark.asyncio
-async def test_shell_handles_quit():
-    server._world = World(seed=42, width=100, height=100)
+async def test_shell_handles_quit(temp_db):
+    world = World(seed=42, width=100, height=100)
+    server._world = world
 
     reader = AsyncMock()
     writer = MagicMock()
@@ -82,14 +118,30 @@ async def test_shell_handles_quit():
     writer.drain = AsyncMock()
     writer.close = MagicMock()
 
-    readline = "mudlib.server.readline2"
-    with patch(readline, new_callable=AsyncMock) as mock_readline:
-        mock_readline.side_effect = ["TestPlayer", "quit"]
-        await server.shell(reader, writer)
+    # Need to mock the look command's world reference too
+    import mudlib.commands.look
 
-    calls = [str(call) for call in writer.write.call_args_list]
-    assert any("Goodbye" in call for call in calls)
-    writer.close.assert_called()
+    original_world = mudlib.commands.look.world
+    mudlib.commands.look.world = world
+
+    try:
+        readline = "mudlib.server.readline2"
+        with patch(readline, new_callable=AsyncMock) as mock_readline:
+            # Simulate: name, create account (y), password, confirm password, quit
+            mock_readline.side_effect = [
+                "TestPlayer",
+                "y",
+                "password",
+                "password",
+                "quit",
+            ]
+            await server.shell(reader, writer)
+
+        calls = [str(call) for call in writer.write.call_args_list]
+        assert any("Goodbye" in call for call in calls)
+        writer.close.assert_called()
+    finally:
+        mudlib.commands.look.world = original_world
 
 
 def test_load_world_config():