diff --git a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java
index d84b6e5..b9d49c0 100644
--- a/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java
+++ b/control-plane/src/main/java/com/clouddb/controlplane/controller/DynamoController.java
@@ -192,6 +192,25 @@ private ResponseEntity<JsonNode> handlePutItem(JsonNode body) throws IOException
           .body(createError("Could not extract partition key from Item"));
     }
 
+    // ConditionExpression check: evaluate before write
+    JsonNode condExpr = body.get("ConditionExpression");
+    if (condExpr != null && !condExpr.isNull()) {
+      RpcReply preGet = tcpClient.getItem(partitionKeyValue, tableName, sortKeyValue);
+      JsonNode existingItem = null;
+      if (preGet.success && preGet.value != null && !preGet.value.isEmpty()) {
+        existingItem = objectMapper.readTree(preGet.value);
+      } else {
+        existingItem = objectMapper.createObjectNode();
+      }
+      JsonNode condAttribNames = body.get("ExpressionAttributeNames");
+      JsonNode condAttribValues = body.get("ExpressionAttributeValues");
+      JsonNode condAst = compileExpression(condExpr.asText(), condAttribNames, condAttribValues);
+      if (condAst != null && !evaluateCondition(condAst, existingItem)) {
+        return ResponseEntity.status(HttpStatus.BAD_REQUEST)
+            .body(createError("ConditionalCheckFailedException"));
+      }
+    }
+
     java.util.List<String> lsiKeys = new java.util.ArrayList<>();
     java.util.List<com.clouddb.controlplane.client.ItemArgs.GsiItem> gsiItems =
         new java.util.ArrayList<>();
@@ -234,10 +253,19 @@ private ResponseEntity<JsonNode> handlePutItem(JsonNode body) throws IOException
     }
 
     String value = objectMapper.writeValueAsString(itemNode);
+    String returnValues = body.has("ReturnValues") ? body.get("ReturnValues").asText() : "NONE";
     RpcReply reply =
         tcpClient.putItem(partitionKeyValue, tableName, sortKeyValue, value, lsiKeys, gsiItems);
 
     if (reply.success) {
+      if ("ALL_NEW".equals(returnValues)) {
+        RpcReply getReply = tcpClient.getItem(partitionKeyValue, tableName, sortKeyValue);
+        var resp = objectMapper.createObjectNode();
+        if (getReply.success && getReply.value != null && !getReply.value.isEmpty()) {
+          resp.set("Attributes", objectMapper.readTree(getReply.value));
+        }
+        return ResponseEntity.ok(resp);
+      }
       return ResponseEntity.ok(objectMapper.createObjectNode());
     } else {
       return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
@@ -325,6 +353,33 @@ private ResponseEntity<JsonNode> handleDeleteItem(JsonNode body) throws IOExcept
           .body(createError("Could not extract partition key from Key"));
     }
 
+    // ConditionExpression check: evaluate before tombstone write
+    JsonNode condExpr = body.get("ConditionExpression");
+    if (condExpr != null && !condExpr.isNull()) {
+      RpcReply preGet = tcpClient.getItem(partitionKeyValue, tableName, sortKeyValue);
+      JsonNode existingItem = null;
+      if (preGet.success && preGet.value != null && !preGet.value.isEmpty()) {
+        existingItem = objectMapper.readTree(preGet.value);
+      } else {
+        existingItem = objectMapper.createObjectNode();
+      }
+      JsonNode condAttribNames = body.get("ExpressionAttributeNames");
+      JsonNode condAttribValues = body.get("ExpressionAttributeValues");
+      JsonNode condAst = compileExpression(condExpr.asText(), condAttribNames, condAttribValues);
+      if (condAst != null && !evaluateCondition(condAst, existingItem)) {
+        return ResponseEntity.status(HttpStatus.BAD_REQUEST)
+            .body(createError("ConditionalCheckFailedException"));
+      }
+    }
+
+    // GET existing item BEFORE tombstone (for ReturnValues)
+    RpcReply preReply = tcpClient.getItem(partitionKeyValue, tableName, sortKeyValue);
+    JsonNode preItem = null;
+    if (preReply.success && preReply.value != null && !preReply.value.isEmpty()) {
+      preItem = objectMapper.readTree(preReply.value);
+    }
+
+    String returnValues = body.has("ReturnValues") ? body.get("ReturnValues").asText() : "NONE";
     RpcReply reply =
         tcpClient.putItem(
             partitionKeyValue,
@@ -335,6 +390,11 @@ private ResponseEntity<JsonNode> handleDeleteItem(JsonNode body) throws IOExcept
             new java.util.ArrayList<>());
 
     if (reply.success) {
+      if ("ALL_OLD".equals(returnValues) && preItem != null) {
+        var resp = objectMapper.createObjectNode();
+        resp.set("Attributes", preItem);
+        return ResponseEntity.ok(resp);
+      }
       return ResponseEntity.ok(objectMapper.createObjectNode());
     } else {
       return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
@@ -715,12 +775,38 @@ private ResponseEntity<JsonNode> handleBatchWriteItem(JsonNode body) throws IOEx
 
     var responseNode = objectMapper.createObjectNode();
     if (reply.success) {
-      // BatchWriteItem doesn't return items, just unprocessed if any
       responseNode.put("UnprocessedKeys", objectMapper.createObjectNode());
       responseNode.put("ConsumedCapacity", objectMapper.createObjectNode());
     } else {
-      return ResponseEntity.status(HttpStatus.INTERNAL_SERVER_ERROR)
-          .body(createError("BatchWriteItem failed: " + reply.errorMsg));
+      // Build UnprocessedKeys from failed items in conflicts
+      var unprocessed = responseNode.putObject("UnprocessedKeys");
+      if (reply.conflicts != null) {
+        for (String conflictJson : reply.conflicts) {
+          JsonNode failedItem = objectMapper.readTree(conflictJson);
+          String tname = failedItem.get("table").asText();
+          var tableUnproc = unprocessed.has(tname)
+              ? (com.fasterxml.jackson.databind.node.ObjectNode) unprocessed.get(tname)
+              : unprocessed.putObject(tname);
+          var writeReqs = tableUnproc.putArray("WriteRequest");
+          var writeReq = writeReqs.addObject();
+          if (failedItem.has("value") && !failedItem.get("value").asText().isEmpty()) {
+            var putReq = writeReq.putObject("PutRequest");
+            var itemNode = putReq.putObject("Item");
+            itemNode.put(failedItem.get("partitionKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("partitionKey").asText()));
+            if (failedItem.has("sortKey") && !failedItem.get("sortKey").asText().isEmpty()) {
+              itemNode.put(failedItem.get("sortKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("sortKey").asText()));
+            }
+          } else {
+            var delReq = writeReq.putObject("DeleteRequest");
+            var keyNode = delReq.putObject("Key");
+            keyNode.put(failedItem.get("partitionKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("partitionKey").asText()));
+            if (failedItem.has("sortKey") && !failedItem.get("sortKey").asText().isEmpty()) {
+              keyNode.put(failedItem.get("sortKey").asText(), objectMapper.createObjectNode().put("S", failedItem.get("sortKey").asText()));
+            }
+          }
+        }
+      }
+      responseNode.put("ConsumedCapacity", objectMapper.createObjectNode());
     }
     return ResponseEntity.ok(responseNode);
   }
@@ -1276,6 +1362,125 @@ else if (c == ',' && parenDepth == 0) {
     return fallback;
   }
 
+  private boolean evaluateCondition(JsonNode ast, JsonNode item) {
+    if (ast == null || item == null) return true;
+    String type = ast.has("type") ? ast.get("type").asText() : "";
+
+    switch (type) {
+      case "and": {
+        JsonNode left = ast.get("left");
+        JsonNode right = ast.get("right");
+        if (left == null || right == null) return false;
+        return evaluateCondition(left, item) && evaluateCondition(right, item);
+      }
+      case "or": {
+        JsonNode left = ast.get("left");
+        JsonNode right = ast.get("right");
+        if (left == null && right == null) return false;
+        if (left == null) return evaluateCondition(right, item);
+        if (right == null) return evaluateCondition(left, item);
+        return evaluateCondition(left, item) || evaluateCondition(right, item);
+      }
+      case "not": {
+        JsonNode exprNode = ast.get("expr");
+        if (exprNode == null) return false;
+        return !evaluateCondition(exprNode, item);
+      }
+      case "function": {
+        String name = ast.has("name") ? ast.get("name").asText() : "";
+        String path = ast.has("path") ? ast.get("path").asText() : "";
+        if ("attribute_exists".equals(name)) {
+          return hasPath(item, path);
+        }
+        if ("attribute_not_exists".equals(name)) {
+          return !hasPath(item, path);
+        }
+        if ("begins_with".equals(name) || "contains".equals(name)) {
+          JsonNode valNode = ast.get("value");
+          String itemVal = extractScalarValue(item, path);
+          String compareVal = extractScalarValue(valNode, null);
+          if ("begins_with".equals(name)) {
+            return itemVal.startsWith(compareVal);
+          }
+          if ("contains".equals(name)) {
+            return itemVal.contains(compareVal);
+          }
+        }
+        return true;
+      }
+      case "comparison": {
+        String op = ast.has("op") ? ast.get("op").asText() : "=";
+        String path = ast.has("path") ? ast.get("path").asText() : "";
+        JsonNode valNode = ast.get("value");
+        String itemVal = extractScalarValue(item, path);
+        String compareVal = extractScalarValue(valNode, null);
+        switch (op) {
+          case "=": return itemVal.equals(compareVal);
+          case "<>": return !itemVal.equals(compareVal);
+          case ">": return itemVal.compareTo(compareVal) > 0;
+          case "<": return itemVal.compareTo(compareVal) < 0;
+          case ">=": return itemVal.compareTo(compareVal) >= 0;
+          case "<=": return itemVal.compareTo(compareVal) <= 0;
+          default: return false;
+        }
+      }
+      case "between": {
+        String path = ast.has("path") ? ast.get("path").asText() : "";
+        JsonNode v1Node = ast.get("value1");
+        JsonNode v2Node = ast.get("value2");
+        String itemVal = extractScalarValue(item, path);
+        String val1 = extractScalarValue(v1Node, null);
+        String val2 = extractScalarValue(v2Node, null);
+        return itemVal.compareTo(val1) >= 0 && itemVal.compareTo(val2) <= 0;
+      }
+      case "in": {
+        String path = ast.has("path") ? ast.get("path").asText() : "";
+        JsonNode valuesNode = ast.get("values");
+        String itemVal = extractScalarValue(item, path);
+        if (valuesNode != null && valuesNode.isArray()) {
+          for (JsonNode v : valuesNode) {
+            if (itemVal.equals(extractScalarValue(v, null))) {
+              return true;
+            }
+          }
+        }
+        return false;
+      }
+      default:
+        return true;
+    }
+  }
+
+  private String extractScalarValue(JsonNode node, String path) {
+    if (node == null) return "";
+    if (path != null && path.contains(".")) {
+      int dot = path.indexOf('.');
+      String first = path.substring(0, dot);
+      String rest = path.substring(dot + 1);
+      if (!node.has(first)) return "";
+      return extractScalarValue(node.get(first), rest);
+    }
+    if (path != null && node.has(path)) {
+      node = node.get(path);
+    }
+    if (node.has("S")) return node.get("S").asText();
+    if (node.has("N")) return node.get("N").asText();
+    if (node.has("BOOL")) return node.get("BOOL").asText();
+    if (node.isTextual()) return node.asText();
+    if (node.isNumber()) return node.asText();
+    return "";
+  }
+
+  private boolean hasPath(JsonNode node, String path) {
+    if (path == null || path.isEmpty()) return false;
+    if (!path.contains(".")) return node.has(path);
+    int dot = path.indexOf('.');
+    String first = path.substring(0, dot);
+    String rest = path.substring(dot + 1);
+    if (!node.has(first)) return false;
+    return hasPath(node.get(first), rest);
+  }
+
   private int findTopLevelOperator(String expr, String op) {
     int balance = 0;
     int opLen = op.length();
diff --git a/storage-engine/src/main.zig b/storage-engine/src/main.zig
index 52304e7..b2a7aab 100644
--- a/storage-engine/src/main.zig
+++ b/storage-engine/src/main.zig
@@ -740,13 +740,32 @@ fn rpcHandler(
 
             for (list.items.items) |item| {
                 // Empty value means delete (tombstone), non-empty means put
+                var ok = false;
                 if (item.value.len > 0) {
-                    _ = s_ctx.engine.put(item.partition_key, item.value, &.{}, item.table, item.sort_key);
+                    ok = s_ctx.engine.put(item.partition_key, item.value, &.{}, item.table, item.sort_key);
                 } else {
-                    _ = s_ctx.engine.delete(item.partition_key, &.{}, item.table, item.sort_key);
+                    ok = s_ctx.engine.delete(item.partition_key, &.{}, item.table, item.sort_key);
+                }
+                if (!ok) {
+                    // Serialize failed item as JSON for UnprocessedKeys
+                    const FailedItemJson = struct {
+                        table: []const u8,
+                        partitionKey: []const u8,
+                        sortKey: []const u8,
+                        value: []const u8,
+                    };
+                    const failed_item = FailedItemJson{
+                        .table = item.table,
+                        .partitionKey = item.partition_key,
+                        .sortKey = item.sort_key,
+                        .value = item.value,
+                    };
+                    var failed_json = ArrayList(u8).init(s_ctx.allocator);
+                    failed_json.print("{}", .{std.json.fmt(failed_item, .{})}) catch continue;
+                    reply_conflicts.append(failed_json.toOwnedSlice() catch &.{}) catch {};
                 }
             }
-            reply_success = true;
+            reply_success = (reply_conflicts.items.len == 0);
         }
     } else if (rpc_type == 15) { // TransactGetItems
         var count: usize = 0;