diff --git a/.dockerignore b/.dockerignore
index c0d5dc8b..0ddf444a 100644
--- a/.dockerignore
+++ b/.dockerignore
@@ -2,7 +2,6 @@
 target/
 bench/
 docs/
-tests/
 clients/
 *.md
 !LICENSE*
diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml
index 2a0245d7..68e33ed7 100644
--- a/.github/workflows/ci.yml
+++ b/.github/workflows/ci.yml
@@ -37,6 +37,8 @@ jobs:
       - uses: actions/checkout@v4
       - uses: dtolnay/rust-toolchain@stable
       - uses: Swatinem/rust-cache@v2
+      - name: build
+        run: cargo build --workspace
       - name: test
         run: cargo test --workspace
 
diff --git a/Cargo.lock b/Cargo.lock
index a137c901..a73da7e8 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -567,6 +567,16 @@ dependencies = [
  "uuid",
 ]
 
+[[package]]
+name = "ember-integration-tests"
+version = "0.4.0"
+dependencies = [
+ "bytes",
+ "ember-protocol",
+ "tempfile",
+ "tokio",
+]
+
 [[package]]
 name = "ember-persistence"
 version = "0.4.0"
diff --git a/Cargo.toml b/Cargo.toml
index 13571346..5617e750 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -6,6 +6,7 @@ members = [
     "crates/ember-persistence",
     "crates/ember-cluster",
     "crates/ember-cli",
+    "tests/integration",
 ]
 resolver = "2"
 
diff --git a/crates/ember-cli/src/repl.rs b/crates/ember-cli/src/repl.rs
index aac30f93..897fcf37 100644
--- a/crates/ember-cli/src/repl.rs
+++ b/crates/ember-cli/src/repl.rs
@@ -16,7 +16,9 @@ use rustyline::hint::Hinter;
 use rustyline::validate::Validator;
 use rustyline::{CompletionType, Config, Context, Editor, Helper};
 
-use crate::commands::{command_names, commands_by_group, find_command, has_subcommands, subcommands};
+use crate::commands::{
+    command_names, commands_by_group, find_command, has_subcommands, subcommands,
+};
 use crate::connection::{Connection, ConnectionError};
 use crate::format::format_response;
 
@@ -433,9 +435,7 @@ impl Highlighter for EmberHelper {
         let rest = &trimmed[first_end..];
 
         let is_known = find_command(first).is_some()
-            || LOCAL_COMMANDS
-                .iter()
-                .any(|c| c.eq_ignore_ascii_case(first));
+            || LOCAL_COMMANDS.iter().any(|c| c.eq_ignore_ascii_case(first));
 
         let highlighted_cmd = if is_known {
             format!("\x1b[1;36m{first}\x1b[0m") // bold cyan
@@ -466,7 +466,12 @@ impl Highlighter for EmberHelper {
         Cow::Owned(format!("\x1b[2m{hint}\x1b[0m")) // dim
     }
 
-    fn highlight_char(&self, _line: &str, _pos: usize, _kind: rustyline::highlight::CmdKind) -> bool {
+    fn highlight_char(
+        &self,
+        _line: &str,
+        _pos: usize,
+        _kind: rustyline::highlight::CmdKind,
+    ) -> bool {
         true // re-highlight on every keystroke
     }
 }
@@ -656,7 +661,11 @@ mod tests {
     #[test]
     fn hint_shows_args_for_known_command() {
         let h = EmberHelper;
-        let hint = h.hint("SET ", 4, &Context::new(&rustyline::history::DefaultHistory::new()));
+        let hint = h.hint(
+            "SET ",
+            4,
+            &Context::new(&rustyline::history::DefaultHistory::new()),
+        );
         assert!(hint.is_some());
         let hint = hint.unwrap();
         assert!(hint.contains("key"));
@@ -665,7 +674,11 @@ mod tests {
     #[test]
     fn hint_shows_subcommands_for_cluster() {
         let h = EmberHelper;
-        let hint = h.hint("CLUSTER ", 8, &Context::new(&rustyline::history::DefaultHistory::new()));
+        let hint = h.hint(
+            "CLUSTER ",
+            8,
+            &Context::new(&rustyline::history::DefaultHistory::new()),
+        );
         assert!(hint.is_some());
         let hint = hint.unwrap();
         assert!(hint.contains("INFO"));
@@ -674,14 +687,22 @@ mod tests {
     #[test]
     fn hint_none_for_unknown_command() {
         let h = EmberHelper;
-        let hint = h.hint("FOOBAR ", 7, &Context::new(&rustyline::history::DefaultHistory::new()));
+        let hint = h.hint(
+            "FOOBAR ",
+            7,
+            &Context::new(&rustyline::history::DefaultHistory::new()),
+        );
         assert!(hint.is_none());
     }
 
     #[test]
     fn hint_none_when_cursor_not_at_end() {
         let h = EmberHelper;
-        let hint = h.hint("SET key", 3, &Context::new(&rustyline::history::DefaultHistory::new()));
+        let hint = h.hint(
+            "SET key",
+            3,
+            &Context::new(&rustyline::history::DefaultHistory::new()),
+        );
         assert!(hint.is_none());
     }
 }
diff --git a/crates/ember-core/src/shard.rs b/crates/ember-core/src/shard.rs
index edc8698c..78c8dca8 100644
--- a/crates/ember-core/src/shard.rs
+++ b/crates/ember-core/src/shard.rs
@@ -248,9 +248,14 @@ pub enum ShardRequest {
         pattern: Option<String>,
     },
     /// Counts keys in this shard that hash to the given cluster slot.
-    CountKeysInSlot { slot: u16 },
+    CountKeysInSlot {
+        slot: u16,
+    },
     /// Returns up to `count` keys that hash to the given cluster slot.
-    GetKeysInSlot { slot: u16, count: usize },
+    GetKeysInSlot {
+        slot: u16,
+        count: usize,
+    },
 }
 
 /// The shard's response to a request.
diff --git a/crates/ember-server/src/cluster.rs b/crates/ember-server/src/cluster.rs
index e91a4e08..2ee97a43 100644
--- a/crates/ember-server/src/cluster.rs
+++ b/crates/ember-server/src/cluster.rs
@@ -298,9 +298,7 @@ impl ClusterCoordinator {
         {
             let state = self.state.read().await;
             if !state.owns_slot(slot) {
-                return Frame::Error(format!(
-                    "ERR I'm not the owner of hash slot {slot}"
-                ));
+                return Frame::Error(format!("ERR I'm not the owner of hash slot {slot}"));
             }
         }
 
@@ -644,9 +642,7 @@ mod tests {
             state.add_node(node);
         }
 
-        let resp = coord
-            .cluster_setslot_node(100, &target.0.to_string())
-            .await;
+        let resp = coord.cluster_setslot_node(100, &target.0.to_string()).await;
         assert!(matches!(resp, Frame::Simple(_)));
 
         // verify the slot is now owned by the target
@@ -672,9 +668,7 @@ mod tests {
         }
 
         // complete with NODE — should clean up migration state
-        let resp = coord
-            .cluster_setslot_node(0, &target.0.to_string())
-            .await;
+        let resp = coord.cluster_setslot_node(0, &target.0.to_string()).await;
         assert!(matches!(resp, Frame::Simple(_)));
 
         // migration should be cleaned up
diff --git a/crates/ember-server/src/connection.rs b/crates/ember-server/src/connection.rs
index 73b666ab..efe2edcc 100644
--- a/crates/ember-server/src/connection.rs
+++ b/crates/ember-server/src/connection.rs
@@ -1554,13 +1554,9 @@ async fn execute(
             }
         }
 
-        Command::ClusterReplicate { .. } => {
-            Frame::Error("ERR REPLICATE not yet supported".into())
-        }
+        Command::ClusterReplicate { .. } => Frame::Error("ERR REPLICATE not yet supported".into()),
 
-        Command::ClusterFailover { .. } => {
-            Frame::Error("ERR FAILOVER not yet supported".into())
-        }
+        Command::ClusterFailover { .. } => Frame::Error("ERR FAILOVER not yet supported".into()),
 
         Command::Migrate { .. } => Frame::Error("ERR not yet implemented".into()),
 
diff --git a/tests/integration/Cargo.toml b/tests/integration/Cargo.toml
new file mode 100644
index 00000000..5e8b7e36
--- /dev/null
+++ b/tests/integration/Cargo.toml
@@ -0,0 +1,17 @@
+[package]
+name = "ember-integration-tests"
+version.workspace = true
+edition.workspace = true
+publish = false
+autobins = false
+
+[[test]]
+name = "integration"
+path = "src/main.rs"
+harness = true
+
+[dependencies]
+tokio = { workspace = true, features = ["rt-multi-thread", "macros", "net", "io-util", "time"] }
+bytes = { workspace = true }
+ember-protocol = { workspace = true }
+tempfile = "3"
diff --git a/tests/integration/src/auth.rs b/tests/integration/src/auth.rs
new file mode 100644
index 00000000..a1efe314
--- /dev/null
+++ b/tests/integration/src/auth.rs
@@ -0,0 +1,40 @@
+//! Integration tests for authentication.
+
+use crate::helpers::{ServerOptions, TestServer};
+
+#[tokio::test]
+async fn auth_required() {
+    let server = TestServer::start_with(ServerOptions {
+        requirepass: Some("secret123".into()),
+        ..Default::default()
+    });
+    let mut c = server.connect().await;
+
+    // commands should be rejected before AUTH
+    let msg = c.err(&["SET", "key", "val"]).await;
+    assert!(msg.contains("NOAUTH"));
+
+    // wrong password
+    let msg = c.err(&["AUTH", "wrongpass"]).await;
+    assert!(msg.contains("WRONGPASS"));
+
+    // correct password
+    c.ok(&["AUTH", "secret123"]).await;
+
+    // commands work after auth
+    c.ok(&["SET", "key", "val"]).await;
+    assert_eq!(c.get_bulk(&["GET", "key"]).await, Some("val".into()));
+}
+
+#[tokio::test]
+async fn ping_allowed_without_auth() {
+    let server = TestServer::start_with(ServerOptions {
+        requirepass: Some("pass".into()),
+        ..Default::default()
+    });
+    let mut c = server.connect().await;
+
+    // PING should work even without auth
+    let resp = c.cmd(&["PING"]).await;
+    assert!(matches!(resp, ember_protocol::Frame::Simple(ref s) if s == "PONG"));
+}
diff --git a/tests/integration/src/basic_operations.rs b/tests/integration/src/basic_operations.rs
new file mode 100644
index 00000000..1e488ae4
--- /dev/null
+++ b/tests/integration/src/basic_operations.rs
@@ -0,0 +1,322 @@
+//! Integration tests for basic string/key operations.
+
+use ember_protocol::Frame;
+
+use crate::helpers::TestServer;
+
+#[tokio::test]
+async fn ping_pong() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    let resp = c.cmd(&["PING"]).await;
+    assert!(matches!(resp, Frame::Simple(ref s) if s == "PONG"));
+}
+
+#[tokio::test]
+async fn ping_with_message() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    let resp = c.get_bulk(&["PING", "hello"]).await;
+    assert_eq!(resp, Some("hello".into()));
+}
+
+#[tokio::test]
+async fn echo() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    let resp = c.get_bulk(&["ECHO", "test"]).await;
+    assert_eq!(resp, Some("test".into()));
+}
+
+#[tokio::test]
+async fn set_get_roundtrip() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "foo", "bar"]).await;
+    let val = c.get_bulk(&["GET", "foo"]).await;
+    assert_eq!(val, Some("bar".into()));
+}
+
+#[tokio::test]
+async fn get_missing_key() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    let resp = c.cmd(&["GET", "nonexistent"]).await;
+    assert!(matches!(resp, Frame::Null));
+}
+
+#[tokio::test]
+async fn set_with_nx() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "key", "first", "NX"]).await;
+    // second SET NX should return null (key already exists)
+    let resp = c.cmd(&["SET", "key", "second", "NX"]).await;
+    assert!(matches!(resp, Frame::Null));
+    // original value preserved
+    assert_eq!(c.get_bulk(&["GET", "key"]).await, Some("first".into()));
+}
+
+#[tokio::test]
+async fn set_with_xx() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    // XX on missing key → null
+    let resp = c.cmd(&["SET", "key", "val", "XX"]).await;
+    assert!(matches!(resp, Frame::Null));
+
+    c.ok(&["SET", "key", "val"]).await;
+    // XX on existing key → OK
+    c.ok(&["SET", "key", "updated", "XX"]).await;
+    assert_eq!(c.get_bulk(&["GET", "key"]).await, Some("updated".into()));
+}
+
+#[tokio::test]
+async fn set_with_ex() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "ttl", "val", "EX", "10"]).await;
+    let ttl = c.get_int(&["TTL", "ttl"]).await;
+    assert!(ttl > 0 && ttl <= 10);
+}
+
+#[tokio::test]
+async fn set_with_px() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "ttl", "val", "PX", "10000"]).await;
+    let pttl = c.get_int(&["PTTL", "ttl"]).await;
+    assert!(pttl > 0 && pttl <= 10000);
+}
+
+#[tokio::test]
+async fn del_existing_and_missing() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "a", "1"]).await;
+    c.ok(&["SET", "b", "2"]).await;
+
+    let count = c.get_int(&["DEL", "a", "b", "c"]).await;
+    assert_eq!(count, 2);
+}
+
+#[tokio::test]
+async fn exists() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "here", "yes"]).await;
+    assert_eq!(c.get_int(&["EXISTS", "here"]).await, 1);
+    assert_eq!(c.get_int(&["EXISTS", "gone"]).await, 0);
+}
+
+#[tokio::test]
+async fn unlink() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "key", "val"]).await;
+    let count = c.get_int(&["UNLINK", "key"]).await;
+    assert_eq!(count, 1);
+    assert!(matches!(c.cmd(&["GET", "key"]).await, Frame::Null));
+}
+
+#[tokio::test]
+async fn incr_decr() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    assert_eq!(c.get_int(&["INCR", "counter"]).await, 1);
+    assert_eq!(c.get_int(&["INCR", "counter"]).await, 2);
+    assert_eq!(c.get_int(&["DECR", "counter"]).await, 1);
+}
+
+#[tokio::test]
+async fn incrby_decrby() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "n", "10"]).await;
+    assert_eq!(c.get_int(&["INCRBY", "n", "5"]).await, 15);
+    assert_eq!(c.get_int(&["DECRBY", "n", "3"]).await, 12);
+}
+
+#[tokio::test]
+async fn expire_ttl_persist() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "key", "val"]).await;
+    assert_eq!(c.get_int(&["TTL", "key"]).await, -1); // no expiry
+
+    c.get_int(&["EXPIRE", "key", "100"]).await;
+    let ttl = c.get_int(&["TTL", "key"]).await;
+    assert!(ttl > 0 && ttl <= 100);
+
+    c.get_int(&["PERSIST", "key"]).await;
+    assert_eq!(c.get_int(&["TTL", "key"]).await, -1);
+}
+
+#[tokio::test]
+async fn mget_mset() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["MSET", "a", "1", "b", "2", "c", "3"]).await;
+
+    let resp = c.cmd(&["MGET", "a", "b", "c", "missing"]).await;
+    match resp {
+        Frame::Array(frames) => {
+            assert_eq!(frames.len(), 4);
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"1"[..]));
+            assert!(matches!(&frames[1], Frame::Bulk(b) if b == &b"2"[..]));
+            assert!(matches!(&frames[2], Frame::Bulk(b) if b == &b"3"[..]));
+            assert!(matches!(&frames[3], Frame::Null));
+        }
+        other => panic!("expected Array, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn type_command() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "s", "val"]).await;
+    let resp = c.cmd(&["TYPE", "s"]).await;
+    assert!(matches!(resp, Frame::Simple(ref t) if t == "string"));
+
+    c.cmd(&["LPUSH", "l", "a"]).await;
+    let resp = c.cmd(&["TYPE", "l"]).await;
+    assert!(matches!(resp, Frame::Simple(ref t) if t == "list"));
+
+    let resp = c.cmd(&["TYPE", "missing"]).await;
+    assert!(matches!(resp, Frame::Simple(ref t) if t == "none"));
+}
+
+#[tokio::test]
+async fn rename() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "old", "value"]).await;
+    c.ok(&["RENAME", "old", "new"]).await;
+    assert!(matches!(c.cmd(&["GET", "old"]).await, Frame::Null));
+    assert_eq!(c.get_bulk(&["GET", "new"]).await, Some("value".into()));
+}
+
+#[tokio::test]
+async fn strlen_append() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "key", "hello"]).await;
+    assert_eq!(c.get_int(&["STRLEN", "key"]).await, 5);
+    assert_eq!(c.get_int(&["APPEND", "key", " world"]).await, 11);
+    assert_eq!(
+        c.get_bulk(&["GET", "key"]).await,
+        Some("hello world".into())
+    );
+}
+
+#[tokio::test]
+async fn dbsize() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    assert_eq!(c.get_int(&["DBSIZE"]).await, 0);
+    c.ok(&["SET", "a", "1"]).await;
+    c.ok(&["SET", "b", "2"]).await;
+    assert_eq!(c.get_int(&["DBSIZE"]).await, 2);
+}
+
+#[tokio::test]
+async fn flushdb() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "a", "1"]).await;
+    c.ok(&["SET", "b", "2"]).await;
+    c.ok(&["FLUSHDB"]).await;
+    assert_eq!(c.get_int(&["DBSIZE"]).await, 0);
+}
+
+#[tokio::test]
+async fn keys_pattern() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.ok(&["SET", "user:1", "a"]).await;
+    c.ok(&["SET", "user:2", "b"]).await;
+    c.ok(&["SET", "item:1", "c"]).await;
+
+    let resp = c.cmd(&["KEYS", "user:*"]).await;
+    match resp {
+        Frame::Array(frames) => {
+            assert_eq!(frames.len(), 2);
+        }
+        other => panic!("expected Array, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn scan_basic() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    for i in 0..5 {
+        c.ok(&["SET", &format!("key:{i}"), "v"]).await;
+    }
+
+    // scan with a large count to get everything in one pass
+    let resp = c.cmd(&["SCAN", "0", "COUNT", "100"]).await;
+    match resp {
+        Frame::Array(frames) => {
+            assert_eq!(frames.len(), 2);
+            // cursor should be "0" (complete)
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"0"[..]));
+            // should have 5 keys
+            if let Frame::Array(ref keys) = frames[1] {
+                assert_eq!(keys.len(), 5);
+            } else {
+                panic!("expected array of keys");
+            }
+        }
+        other => panic!("expected Array, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn info_returns_sections() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    let resp = c.cmd(&["INFO"]).await;
+    match resp {
+        Frame::Bulk(data) => {
+            let text = String::from_utf8_lossy(&data);
+            assert!(text.contains("# Server"));
+            assert!(text.contains("ember_version"));
+        }
+        other => panic!("expected Bulk, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn unknown_command() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    let msg = c.err(&["NOTACOMMAND"]).await;
+    assert!(msg.contains("unknown command"));
+}
diff --git a/tests/integration/src/data_types.rs b/tests/integration/src/data_types.rs
new file mode 100644
index 00000000..c56041b0
--- /dev/null
+++ b/tests/integration/src/data_types.rs
@@ -0,0 +1,246 @@
+//! Integration tests for lists, hashes, sets, and sorted sets.
+
+use ember_protocol::Frame;
+
+use crate::helpers::TestServer;
+
+// --- lists ---
+
+#[tokio::test]
+async fn list_push_pop() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    assert_eq!(c.get_int(&["LPUSH", "list", "a"]).await, 1);
+    assert_eq!(c.get_int(&["RPUSH", "list", "b"]).await, 2);
+    assert_eq!(c.get_int(&["LPUSH", "list", "c"]).await, 3);
+
+    // order is: c, a, b
+    assert_eq!(c.get_bulk(&["LPOP", "list"]).await, Some("c".into()));
+    assert_eq!(c.get_bulk(&["RPOP", "list"]).await, Some("b".into()));
+    assert_eq!(c.get_bulk(&["LPOP", "list"]).await, Some("a".into()));
+
+    // empty list
+    let resp = c.cmd(&["LPOP", "list"]).await;
+    assert!(matches!(resp, Frame::Null));
+}
+
+#[tokio::test]
+async fn list_lrange() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.cmd(&["RPUSH", "list", "a", "b", "c", "d"]).await;
+
+    let resp = c.cmd(&["LRANGE", "list", "0", "-1"]).await;
+    match resp {
+        Frame::Array(frames) => {
+            assert_eq!(frames.len(), 4);
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"a"[..]));
+            assert!(matches!(&frames[3], Frame::Bulk(b) if b == &b"d"[..]));
+        }
+        other => panic!("expected Array, got {other:?}"),
+    }
+
+    // partial range
+    let resp = c.cmd(&["LRANGE", "list", "1", "2"]).await;
+    match resp {
+        Frame::Array(frames) => {
+            assert_eq!(frames.len(), 2);
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"b"[..]));
+            assert!(matches!(&frames[1], Frame::Bulk(b) if b == &b"c"[..]));
+        }
+        other => panic!("expected Array, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn list_llen() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    assert_eq!(c.get_int(&["LLEN", "list"]).await, 0);
+    c.cmd(&["RPUSH", "list", "a", "b", "c"]).await;
+    assert_eq!(c.get_int(&["LLEN", "list"]).await, 3);
+}
+
+// --- hashes ---
+
+#[tokio::test]
+async fn hash_set_get() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.get_int(&["HSET", "h", "name", "ember", "version", "0.4"])
+        .await;
+    assert_eq!(
+        c.get_bulk(&["HGET", "h", "name"]).await,
+        Some("ember".into())
+    );
+    assert_eq!(
+        c.get_bulk(&["HGET", "h", "version"]).await,
+        Some("0.4".into())
+    );
+
+    // missing field
+    let resp = c.cmd(&["HGET", "h", "missing"]).await;
+    assert!(matches!(resp, Frame::Null));
+}
+
+#[tokio::test]
+async fn hash_getall() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.get_int(&["HSET", "h", "a", "1", "b", "2"]).await;
+    let resp = c.cmd(&["HGETALL", "h"]).await;
+    match resp {
+        Frame::Array(frames) => {
+            // 2 fields × 2 (field + value) = 4 elements
+            assert_eq!(frames.len(), 4);
+        }
+        other => panic!("expected Array, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn hash_del_exists_len() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.get_int(&["HSET", "h", "a", "1", "b", "2", "c", "3"])
+        .await;
+    assert_eq!(c.get_int(&["HLEN", "h"]).await, 3);
+    assert_eq!(c.get_int(&["HEXISTS", "h", "a"]).await, 1);
+
+    assert_eq!(c.get_int(&["HDEL", "h", "a"]).await, 1);
+    assert_eq!(c.get_int(&["HEXISTS", "h", "a"]).await, 0);
+    assert_eq!(c.get_int(&["HLEN", "h"]).await, 2);
+}
+
+#[tokio::test]
+async fn hash_incrby() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.get_int(&["HSET", "h", "counter", "10"]).await;
+    assert_eq!(c.get_int(&["HINCRBY", "h", "counter", "5"]).await, 15);
+    assert_eq!(c.get_int(&["HINCRBY", "h", "counter", "-3"]).await, 12);
+
+    // new field
+    assert_eq!(c.get_int(&["HINCRBY", "h", "new", "1"]).await, 1);
+}
+
+#[tokio::test]
+async fn hash_keys_vals_hmget() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.get_int(&["HSET", "h", "x", "10", "y", "20"]).await;
+
+    let resp = c.cmd(&["HKEYS", "h"]).await;
+    match resp {
+        Frame::Array(keys) => assert_eq!(keys.len(), 2),
+        other => panic!("expected Array, got {other:?}"),
+    }
+
+    let resp = c.cmd(&["HVALS", "h"]).await;
+    match resp {
+        Frame::Array(vals) => assert_eq!(vals.len(), 2),
+        other => panic!("expected Array, got {other:?}"),
+    }
+
+    let resp = c.cmd(&["HMGET", "h", "x", "y", "z"]).await;
+    match resp {
+        Frame::Array(frames) => {
+            assert_eq!(frames.len(), 3);
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"10"[..]));
+            assert!(matches!(&frames[1], Frame::Bulk(b) if b == &b"20"[..]));
+            assert!(matches!(&frames[2], Frame::Null));
+        }
+        other => panic!("expected Array, got {other:?}"),
+    }
+}
+
+// --- sets ---
+
+#[tokio::test]
+async fn set_add_rem_members() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    assert_eq!(c.get_int(&["SADD", "s", "a", "b", "c"]).await, 3);
+    // duplicate
+    assert_eq!(c.get_int(&["SADD", "s", "a"]).await, 0);
+    assert_eq!(c.get_int(&["SCARD", "s"]).await, 3);
+
+    assert_eq!(c.get_int(&["SISMEMBER", "s", "a"]).await, 1);
+    assert_eq!(c.get_int(&["SISMEMBER", "s", "z"]).await, 0);
+
+    assert_eq!(c.get_int(&["SREM", "s", "a", "z"]).await, 1);
+    assert_eq!(c.get_int(&["SCARD", "s"]).await, 2);
+}
+
+#[tokio::test]
+async fn set_smembers() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.get_int(&["SADD", "s", "x", "y"]).await;
+    let resp = c.cmd(&["SMEMBERS", "s"]).await;
+    match resp {
+        Frame::Array(members) => assert_eq!(members.len(), 2),
+        other => panic!("expected Array, got {other:?}"),
+    }
+}
+
+// --- sorted sets ---
+
+#[tokio::test]
+async fn zset_add_score_rank() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    assert_eq!(
+        c.get_int(&["ZADD", "z", "1.0", "a", "2.0", "b", "3.0", "c"])
+            .await,
+        3
+    );
+    assert_eq!(c.get_int(&["ZCARD", "z"]).await, 3);
+
+    let score = c.get_bulk(&["ZSCORE", "z", "b"]).await;
+    assert_eq!(score, Some("2".into()));
+
+    let rank = c.get_int(&["ZRANK", "z", "a"]).await;
+    assert_eq!(rank, 0); // lowest score = rank 0
+}
+
+#[tokio::test]
+async fn zset_range() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.get_int(&["ZADD", "z", "1", "a", "2", "b", "3", "c"])
+        .await;
+
+    let resp = c.cmd(&["ZRANGE", "z", "0", "-1"]).await;
+    match resp {
+        Frame::Array(frames) => {
+            assert_eq!(frames.len(), 3);
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"a"[..]));
+            assert!(matches!(&frames[1], Frame::Bulk(b) if b == &b"b"[..]));
+            assert!(matches!(&frames[2], Frame::Bulk(b) if b == &b"c"[..]));
+        }
+        other => panic!("expected Array, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn zset_rem() {
+    let server = TestServer::start();
+    let mut c = server.connect().await;
+
+    c.get_int(&["ZADD", "z", "1", "a", "2", "b"]).await;
+    assert_eq!(c.get_int(&["ZREM", "z", "a", "missing"]).await, 1);
+    assert_eq!(c.get_int(&["ZCARD", "z"]).await, 1);
+}
diff --git a/tests/integration/src/helpers.rs b/tests/integration/src/helpers.rs
new file mode 100644
index 00000000..ca3cab55
--- /dev/null
+++ b/tests/integration/src/helpers.rs
@@ -0,0 +1,251 @@
+//! Test helpers for spawning an ember-server and sending commands.
+
+use std::net::TcpListener;
+use std::path::PathBuf;
+use std::process::{Child, Command};
+use std::time::Duration;
+
+use bytes::{Bytes, BytesMut};
+use ember_protocol::{parse_frame, Frame};
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+
+/// An ember-server subprocess managed by the test harness.
+pub struct TestServer {
+    child: Child,
+    pub port: u16,
+    _data_dir: Option<tempfile::TempDir>,
+}
+
+/// Options for starting a test server.
+#[derive(Default)]
+pub struct ServerOptions {
+    pub requirepass: Option<String>,
+    pub appendonly: bool,
+    /// Owned temp directory (cleaned up when the server drops).
+    pub data_dir: Option<tempfile::TempDir>,
+    /// Use an existing path without taking ownership.
+    /// If both `data_dir` and `data_dir_path` are set, `data_dir_path` wins.
+    pub data_dir_path: Option<PathBuf>,
+}
+
+impl TestServer {
+    /// Starts a new ember-server on a random port.
+    ///
+    /// Blocks until the server is accepting connections (up to 5 seconds).
+    pub fn start() -> Self {
+        Self::start_with(ServerOptions::default())
+    }
+
+    /// Starts a new ember-server with custom options.
+    pub fn start_with(opts: ServerOptions) -> Self {
+        let port = find_free_port();
+
+        let binary = server_binary();
+
+        let mut cmd = Command::new(&binary);
+        cmd.arg("--port").arg(port.to_string());
+        cmd.arg("--host").arg("127.0.0.1");
+        cmd.arg("--shards").arg("2");
+        // suppress tracing output in tests
+        cmd.env("RUST_LOG", "error");
+
+        if let Some(ref pass) = opts.requirepass {
+            cmd.arg("--requirepass").arg(pass);
+        }
+
+        let data_dir = if opts.appendonly {
+            cmd.arg("--appendonly");
+            cmd.arg("--appendfsync").arg("always");
+
+            if let Some(ref path) = opts.data_dir_path {
+                cmd.arg("--data-dir").arg(path);
+                None // caller manages the directory lifetime
+            } else {
+                let dir = opts
+                    .data_dir
+                    .unwrap_or_else(|| tempfile::tempdir().unwrap());
+                cmd.arg("--data-dir").arg(dir.path());
+                Some(dir)
+            }
+        } else {
+            None
+        };
+
+        let child = cmd
+            .stdout(std::process::Stdio::null())
+            .stderr(std::process::Stdio::null())
+            .spawn()
+            .unwrap_or_else(|e| {
+                panic!("failed to spawn ember-server at {}: {e}", binary.display())
+            });
+
+        // wait for the server to be ready
+        let deadline = std::time::Instant::now() + Duration::from_secs(5);
+        loop {
+            if std::time::Instant::now() > deadline {
+                panic!("ember-server failed to start within 5 seconds on port {port}");
+            }
+            if std::net::TcpStream::connect(format!("127.0.0.1:{port}")).is_ok() {
+                break;
+            }
+            std::thread::sleep(Duration::from_millis(50));
+        }
+
+        Self {
+            child,
+            port,
+            _data_dir: data_dir,
+        }
+    }
+
+    /// Connects a test client to this server.
+    pub async fn connect(&self) -> TestClient {
+        TestClient::connect(self.port).await
+    }
+}
+
+impl Drop for TestServer {
+    fn drop(&mut self) {
+        let _ = self.child.kill();
+        let _ = self.child.wait();
+    }
+}
+
+/// A minimal RESP3 client for integration testing.
+pub struct TestClient {
+    stream: TcpStream,
+    buf: BytesMut,
+}
+
+impl TestClient {
+    async fn connect(port: u16) -> Self {
+        let stream = TcpStream::connect(format!("127.0.0.1:{port}"))
+            .await
+            .unwrap_or_else(|e| panic!("failed to connect to 127.0.0.1:{port}: {e}"));
+        Self {
+            stream,
+            buf: BytesMut::with_capacity(4096),
+        }
+    }
+
+    /// Sends a command and returns the parsed response frame.
+    pub async fn cmd(&mut self, args: &[&str]) -> Frame {
+        // build RESP3 array
+        let parts: Vec<Frame> = args
+            .iter()
+            .map(|a| Frame::Bulk(Bytes::copy_from_slice(a.as_bytes())))
+            .collect();
+        let frame = Frame::Array(parts);
+
+        let mut out = BytesMut::new();
+        frame.serialize(&mut out);
+        self.stream.write_all(&out).await.unwrap();
+
+        // read response
+        loop {
+            match parse_frame(&self.buf) {
+                Ok(Some((frame, consumed))) => {
+                    let _ = self.buf.split_to(consumed);
+                    return frame;
+                }
+                Ok(None) => {
+                    let n = self.stream.read_buf(&mut self.buf).await.unwrap();
+                    if n == 0 {
+                        panic!("server closed connection while waiting for response");
+                    }
+                }
+                Err(e) => panic!("protocol error: {e}"),
+            }
+        }
+    }
+
+    /// Sends a command and extracts the bulk string value.
+    pub async fn get_bulk(&mut self, args: &[&str]) -> Option<String> {
+        match self.cmd(args).await {
+            Frame::Bulk(data) => Some(String::from_utf8_lossy(&data).to_string()),
+            Frame::Null => None,
+            other => panic!("expected Bulk or Null, got {other:?}"),
+        }
+    }
+
+    /// Sends a command and extracts the integer value.
+    pub async fn get_int(&mut self, args: &[&str]) -> i64 {
+        match self.cmd(args).await {
+            Frame::Integer(n) => n,
+            other => panic!("expected Integer, got {other:?}"),
+        }
+    }
+
+    /// Sends a command and expects a Simple "OK" response.
+    pub async fn ok(&mut self, args: &[&str]) {
+        match self.cmd(args).await {
+            Frame::Simple(s) if s == "OK" => {}
+            other => panic!("expected OK, got {other:?}"),
+        }
+    }
+
+    /// Sends a command and expects an error response. Returns the error message.
+    pub async fn err(&mut self, args: &[&str]) -> String {
+        match self.cmd(args).await {
+            Frame::Error(msg) => msg,
+            other => panic!("expected Error, got {other:?}"),
+        }
+    }
+
+    /// Reads the next frame from the connection without sending a command.
+    /// Useful for pub/sub where the server pushes messages asynchronously.
+    pub async fn read_frame(&mut self) -> Frame {
+        loop {
+            match parse_frame(&self.buf) {
+                Ok(Some((frame, consumed))) => {
+                    let _ = self.buf.split_to(consumed);
+                    return frame;
+                }
+                Ok(None) => {
+                    let n = self.stream.read_buf(&mut self.buf).await.unwrap();
+                    if n == 0 {
+                        panic!("server closed connection while waiting for frame");
+                    }
+                }
+                Err(e) => panic!("protocol error: {e}"),
+            }
+        }
+    }
+}
+
+/// Finds a free TCP port by binding to port 0.
+fn find_free_port() -> u16 {
+    let listener = TcpListener::bind("127.0.0.1:0").unwrap();
+    listener.local_addr().unwrap().port()
+}
+
+/// Locates the ember-server binary in the cargo target directory.
+fn server_binary() -> PathBuf {
+    // cargo sets OUT_DIR for build scripts, but for integration tests
+    // we can find the binary relative to the test binary itself
+    let mut path = std::env::current_exe().unwrap();
+    // test binary is in target/debug/deps/ — go up to target/debug/
+    path.pop();
+    if path.ends_with("deps") {
+        path.pop();
+    }
+    path.push("ember-server");
+    if !path.exists() {
+        // try release
+        let mut release = std::env::current_exe().unwrap();
+        release.pop();
+        if release.ends_with("deps") {
+            release.pop();
+        }
+        release.push("ember-server");
+        if release.exists() {
+            return release;
+        }
+        panic!(
+            "ember-server binary not found. run `cargo build` first.\nlooked at: {}",
+            path.display()
+        );
+    }
+    path
+}
diff --git a/tests/integration/src/main.rs b/tests/integration/src/main.rs
new file mode 100644
index 00000000..e9af83ca
--- /dev/null
+++ b/tests/integration/src/main.rs
@@ -0,0 +1,7 @@
+mod helpers;
+
+mod auth;
+mod basic_operations;
+mod data_types;
+mod persistence;
+mod pubsub;
diff --git a/tests/integration/src/persistence.rs b/tests/integration/src/persistence.rs
new file mode 100644
index 00000000..18830591
--- /dev/null
+++ b/tests/integration/src/persistence.rs
@@ -0,0 +1,83 @@
+//! Integration tests for snapshot and AOF persistence/recovery.
+
+use std::time::Duration;
+
+use ember_protocol::Frame;
+
+use crate::helpers::{ServerOptions, TestServer};
+
+#[tokio::test]
+async fn bgsave_and_snapshot_recovery() {
+    let data_dir = tempfile::tempdir().unwrap();
+    let path = data_dir.path().to_path_buf();
+
+    // start server, write keys, trigger snapshot
+    {
+        let server = TestServer::start_with(ServerOptions {
+            appendonly: true,
+            data_dir_path: Some(path.clone()),
+            ..Default::default()
+        });
+        let mut c = server.connect().await;
+
+        c.ok(&["SET", "snap:a", "alpha"]).await;
+        c.ok(&["SET", "snap:b", "beta"]).await;
+
+        let resp = c.cmd(&["BGSAVE"]).await;
+        assert!(matches!(resp, Frame::Simple(ref s) if s.contains("saving started")));
+
+        // give the snapshot time to flush to disk
+        tokio::time::sleep(Duration::from_millis(500)).await;
+    }
+    // server killed on drop, but data_dir still alive
+
+    // restart with same data directory
+    let server = TestServer::start_with(ServerOptions {
+        appendonly: true,
+        data_dir_path: Some(path),
+        ..Default::default()
+    });
+    let mut c = server.connect().await;
+
+    assert_eq!(c.get_bulk(&["GET", "snap:a"]).await, Some("alpha".into()));
+    assert_eq!(c.get_bulk(&["GET", "snap:b"]).await, Some("beta".into()));
+
+    // keep data_dir alive until assertions complete
+    drop(data_dir);
+}
+
+#[tokio::test]
+async fn aof_recovery() {
+    let data_dir = tempfile::tempdir().unwrap();
+    let path = data_dir.path().to_path_buf();
+
+    // start server with AOF (fsync=always), write keys
+    {
+        let server = TestServer::start_with(ServerOptions {
+            appendonly: true,
+            data_dir_path: Some(path.clone()),
+            ..Default::default()
+        });
+        let mut c = server.connect().await;
+
+        c.ok(&["SET", "aof:x", "100"]).await;
+        c.ok(&["SET", "aof:y", "200"]).await;
+        c.get_int(&["INCR", "aof:x"]).await;
+
+        // small sleep to ensure fsync completes
+        tokio::time::sleep(Duration::from_millis(200)).await;
+    }
+
+    // restart — AOF replay should restore state
+    let server = TestServer::start_with(ServerOptions {
+        appendonly: true,
+        data_dir_path: Some(path),
+        ..Default::default()
+    });
+    let mut c = server.connect().await;
+
+    assert_eq!(c.get_bulk(&["GET", "aof:x"]).await, Some("101".into()));
+    assert_eq!(c.get_bulk(&["GET", "aof:y"]).await, Some("200".into()));
+
+    drop(data_dir);
+}
diff --git a/tests/integration/src/pubsub.rs b/tests/integration/src/pubsub.rs
new file mode 100644
index 00000000..c833989f
--- /dev/null
+++ b/tests/integration/src/pubsub.rs
@@ -0,0 +1,91 @@
+//! Integration tests for pub/sub.
+
+use ember_protocol::Frame;
+
+use crate::helpers::TestServer;
+
+#[tokio::test]
+async fn subscribe_and_receive_message() {
+    let server = TestServer::start();
+    let mut sub = server.connect().await;
+    let mut publisher = server.connect().await;
+
+    // subscribe — confirmation frame
+    let resp = sub.cmd(&["SUBSCRIBE", "events"]).await;
+    match resp {
+        Frame::Array(ref frames) => {
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"subscribe"[..]));
+            assert!(matches!(&frames[1], Frame::Bulk(b) if b == &b"events"[..]));
+            assert!(matches!(&frames[2], Frame::Integer(1)));
+        }
+        other => panic!("expected subscribe confirmation, got {other:?}"),
+    }
+
+    // publish a message from another connection
+    let count = publisher.get_int(&["PUBLISH", "events", "hello"]).await;
+    assert_eq!(count, 1);
+
+    // subscriber receives the message (pushed frame, no command needed)
+    let msg = sub.read_frame().await;
+    match msg {
+        Frame::Array(ref frames) => {
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"message"[..]));
+            assert!(matches!(&frames[1], Frame::Bulk(b) if b == &b"events"[..]));
+            assert!(matches!(&frames[2], Frame::Bulk(b) if b == &b"hello"[..]));
+        }
+        other => panic!("expected message frame, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn psubscribe_pattern_match() {
+    let server = TestServer::start();
+    let mut sub = server.connect().await;
+    let mut publisher = server.connect().await;
+
+    // pattern subscribe
+    let resp = sub.cmd(&["PSUBSCRIBE", "user:*"]).await;
+    match resp {
+        Frame::Array(ref frames) => {
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"psubscribe"[..]));
+            assert!(matches!(&frames[1], Frame::Bulk(b) if b == &b"user:*"[..]));
+            assert!(matches!(&frames[2], Frame::Integer(1)));
+        }
+        other => panic!("expected psubscribe confirmation, got {other:?}"),
+    }
+
+    // publish to a matching channel
+    let count = publisher.get_int(&["PUBLISH", "user:login", "alice"]).await;
+    assert_eq!(count, 1);
+
+    // subscriber receives pmessage
+    let msg = sub.read_frame().await;
+    match msg {
+        Frame::Array(ref frames) => {
+            assert_eq!(frames.len(), 4);
+            assert!(matches!(&frames[0], Frame::Bulk(b) if b == &b"pmessage"[..]));
+            assert!(matches!(&frames[1], Frame::Bulk(b) if b == &b"user:*"[..]));
+            assert!(matches!(&frames[2], Frame::Bulk(b) if b == &b"user:login"[..]));
+            assert!(matches!(&frames[3], Frame::Bulk(b) if b == &b"alice"[..]));
+        }
+        other => panic!("expected pmessage frame, got {other:?}"),
+    }
+}
+
+#[tokio::test]
+async fn publish_returns_subscriber_count() {
+    let server = TestServer::start();
+    let mut sub = server.connect().await;
+    let mut publisher = server.connect().await;
+
+    // no subscribers yet
+    let count = publisher.get_int(&["PUBLISH", "chan", "msg"]).await;
+    assert_eq!(count, 0);
+
+    // subscribe
+    sub.cmd(&["SUBSCRIBE", "chan"]).await;
+
+    // now there's 1 subscriber
+    let count = publisher.get_int(&["PUBLISH", "chan", "msg"]).await;
+    assert_eq!(count, 1);
+}