diff --git a/pytest.ini b/pytest.ini
new file mode 100644
index 0000000..cda2c8e
--- /dev/null
+++ b/pytest.ini
@@ -0,0 +1,9 @@
+[pytest]
+markers =
+    e2e: End-to-end tests that require running Modal service
+    slow: Tests that take >30 seconds to complete
+    unit: Fast unit tests
+testpaths = tests
+python_files = test_*.py
+python_classes = Test*
+python_functions = test_*
diff --git a/requirements-test.txt b/requirements-test.txt
new file mode 100644
index 0000000..01006d9
--- /dev/null
+++ b/requirements-test.txt
@@ -0,0 +1,5 @@
+-r requirements.txt
+pytest>=7.4.0
+pytest-asyncio>=0.21.0
+pytest-timeout>=2.1.0
+requests>=2.31.0
diff --git a/tests/README.md b/tests/README.md
new file mode 100644
index 0000000..7dda792
--- /dev/null
+++ b/tests/README.md
@@ -0,0 +1,212 @@
+# Letta Schedules Test Suite
+
+Comprehensive test suite with unit tests and end-to-end integration tests.
+
+## Test Structure
+
+```
+tests/
+├── conftest.py              # Pytest fixtures and configuration
+├── test_crypto_utils.py     # Unit tests for encryption/hashing
+├── test_scheduler.py        # Unit tests for schedule logic
+├── test_api_e2e.py          # E2E API tests (requires running service)
+└── README.md                # This file
+```
+
+## Setup
+
+Install test dependencies:
+
+```bash
+pip install -r requirements-test.txt
+```
+
+## Running Tests
+
+### Unit Tests Only (Fast)
+
+Unit tests run without requiring a running Modal service:
+
+```bash
+# Run all unit tests
+pytest -m "not e2e"
+
+# Run specific unit test file
+pytest tests/test_crypto_utils.py
+pytest tests/test_scheduler.py
+
+# Run with verbose output
+pytest -v -m "not e2e"
+```
+
+### E2E Tests (Requires Running Service)
+
+E2E tests require a running Modal service and valid Letta credentials:
+
+**1. Start the service:**
+```bash
+export LETTA_SCHEDULES_DEV_MODE=true
+modal serve app.py
+```
+
+**2. In another terminal, set environment variables:**
+```bash
+export LETTA_API_KEY="sk-..."                              # Required: Valid Letta API key
+export LETTA_AGENT_ID="agent-xxx"                          # Required: Valid agent ID
+export LETTA_SCHEDULES_URL="https://your-modal-url"        # Optional: defaults to dev URL
+export LETTA_API_KEY_2="sk-different-key"                  # Optional: for multi-user tests
+```
+
+**3. Run E2E tests:**
+```bash
+# Run all E2E tests (excluding slow tests)
+pytest -m "e2e and not slow"
+
+# Run all E2E tests including slow ones
+pytest -m "e2e"
+
+# Run specific E2E test class
+pytest tests/test_api_e2e.py::TestAuthentication
+pytest tests/test_api_e2e.py::TestRecurringScheduleCRUD
+
+# Run execution tests (these take 60-90 seconds each)
+pytest -m "slow" -v
+```
+
+### Run All Tests
+
+```bash
+# Run everything (unit + e2e, excluding slow tests)
+pytest -m "not slow"
+
+# Run absolutely everything
+pytest
+```
+
+## Test Markers
+
+Tests are organized with pytest markers:
+
+- `@pytest.mark.unit` - Fast unit tests (no external dependencies)
+- `@pytest.mark.e2e` - End-to-end tests (requires running service)
+- `@pytest.mark.slow` - Tests that wait for cron execution (60-90s each)
+
+## Test Categories
+
+### 1. Unit Tests
+
+**test_crypto_utils.py:**
+- API key hashing (deterministic, correct length)
+- Dev mode detection
+- Encryption/decryption roundtrip
+- Dev mode plaintext storage
+- Production mode encryption
+
+**test_scheduler.py:**
+- Cron schedule due checking
+- One-time schedule due checking
+- Timezone handling
+- Edge cases (just ran, long ago, exact time)
+
+### 2. E2E API Tests
+
+**test_api_e2e.py:**
+
+**Authentication:**
+- Invalid API key → 401
+- Missing auth header → 403
+- API keys never in responses
+
+**Recurring Schedules CRUD:**
+- Create, list, get, delete operations
+- Authorization checks
+- 404 on non-existent schedules
+
+**One-Time Schedules CRUD:**
+- Create, list, get, delete operations
+- Timezone support
+- Authorization checks
+
+**Results:**
+- List execution results
+- Get specific result
+- Results persist after schedule deletion
+
+**Execution (Slow Tests):**
+- Past schedules execute immediately (<90s)
+- One-time schedules deleted after execution
+- No duplicate executions
+- Results created with run_id
+
+## Environment Variables
+
+**Required for E2E tests:**
+- `LETTA_API_KEY` - Valid Letta API key
+- `LETTA_AGENT_ID` - Valid Letta agent ID
+
+**Optional:**
+- `LETTA_SCHEDULES_URL` - Override API base URL (default: dev URL)
+- `LETTA_API_KEY_2` - Second API key for multi-user isolation tests
+- `LETTA_SCHEDULES_DEV_MODE` - Enable dev mode (default: true for tests)
+
+## CI/CD Integration
+
+For continuous integration, run only fast tests:
+
+```bash
+# Run unit tests + fast E2E tests
+pytest -m "not slow" --timeout=30
+
+# Generate coverage report
+pytest --cov=. --cov-report=html -m "not slow"
+```
+
+For full integration testing (slower):
+
+```bash
+# Run everything including execution tests
+pytest -v --timeout=120
+```
+
+## Troubleshooting
+
+**Tests fail with "LETTA_API_KEY not set":**
+- E2E tests require valid Letta credentials
+- Set environment variables or run only unit tests: `pytest -m "not e2e"`
+
+**Tests timeout:**
+- Execution tests wait up to 90 seconds for cron to run
+- Ensure Modal service is running: `modal serve app.py`
+- Check service logs: `modal app logs letta-schedules --follow`
+
+**"Service not reachable":**
+- Verify Modal service is running
+- Check `LETTA_SCHEDULES_URL` points to correct endpoint
+- Ensure service accepts connections: `curl <service-url>/schedules/recurring`
+
+**Dev mode warnings:**
+- Tests automatically enable dev mode via `LETTA_SCHEDULES_DEV_MODE=true`
+- No encryption key needed for local tests
+- Files stored in plaintext for easy inspection
+
+## Example Test Run
+
+```bash
+# Terminal 1: Start service
+export LETTA_SCHEDULES_DEV_MODE=true
+modal serve app.py
+
+# Terminal 2: Run tests
+export LETTA_API_KEY="sk-..."
+export LETTA_AGENT_ID="agent-..."
+pytest -v -m "e2e and not slow"
+```
+
+Expected output:
+```
+tests/test_api_e2e.py::TestAuthentication::test_invalid_api_key_returns_401 PASSED
+tests/test_api_e2e.py::TestRecurringScheduleCRUD::test_create_recurring_schedule PASSED
+tests/test_api_e2e.py::TestOneTimeScheduleCRUD::test_create_onetime_schedule PASSED
+...
+========== 15 passed in 12.34s ==========
+```
diff --git a/tests/conftest.py b/tests/conftest.py
new file mode 100644
index 0000000..b5904d0
--- /dev/null
+++ b/tests/conftest.py
@@ -0,0 +1,107 @@
+import pytest
+import os
+import sys
+import tempfile
+import shutil
+from pathlib import Path
+from datetime import datetime, timezone, timedelta
+from cryptography.fernet import Fernet
+
+# Add parent directory to path for imports
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+
+os.environ["LETTA_SCHEDULES_DEV_MODE"] = "true"
+
+
+@pytest.fixture
+def temp_volume_path():
+    """Create a temporary directory to simulate Modal volume."""
+    temp_dir = tempfile.mkdtemp()
+    yield temp_dir
+    shutil.rmtree(temp_dir)
+
+
+@pytest.fixture
+def mock_api_key():
+    """A mock API key for testing."""
+    return "test-api-key-12345"
+
+
+@pytest.fixture
+def mock_api_key_2():
+    """A second mock API key for multi-user testing."""
+    return "test-api-key-67890"
+
+
+@pytest.fixture
+def mock_agent_id():
+    """A mock agent ID for testing."""
+    return "agent-test-123"
+
+
+@pytest.fixture
+def encryption_key():
+    """Generate a test encryption key."""
+    return Fernet.generate_key()
+
+
+@pytest.fixture
+def test_recurring_schedule_data(mock_api_key, mock_agent_id):
+    """Sample recurring schedule data."""
+    return {
+        "agent_id": mock_agent_id,
+        "api_key": mock_api_key,
+        "cron": "*/5 * * * *",
+        "message": "Test recurring message",
+        "role": "user"
+    }
+
+
+@pytest.fixture
+def test_onetime_schedule_data(mock_api_key, mock_agent_id):
+    """Sample one-time schedule data."""
+    future_time = datetime.now(timezone.utc) + timedelta(minutes=5)
+    return {
+        "agent_id": mock_agent_id,
+        "api_key": mock_api_key,
+        "execute_at": future_time.isoformat(),
+        "message": "Test one-time message",
+        "role": "user"
+    }
+
+
+@pytest.fixture
+def past_onetime_schedule_data(mock_api_key, mock_agent_id):
+    """One-time schedule scheduled in the past (should execute immediately)."""
+    past_time = datetime.now(timezone.utc) - timedelta(minutes=5)
+    return {
+        "agent_id": mock_agent_id,
+        "api_key": mock_api_key,
+        "execute_at": past_time.isoformat(),
+        "message": "Test past message",
+        "role": "user"
+    }
+
+
+@pytest.fixture
+def api_base_url():
+    """Base URL for API testing. Override with LETTA_SCHEDULES_URL env var."""
+    return os.getenv("LETTA_SCHEDULES_URL", "https://letta--letta-schedules-api-dev.modal.run")
+
+
+@pytest.fixture
+def valid_letta_api_key():
+    """Real Letta API key for E2E tests. Must be set via env var."""
+    api_key = os.getenv("LETTA_API_KEY")
+    if not api_key:
+        pytest.skip("LETTA_API_KEY not set - skipping E2E test")
+    return api_key
+
+
+@pytest.fixture
+def valid_letta_agent_id():
+    """Real Letta agent ID for E2E tests."""
+    agent_id = os.getenv("LETTA_AGENT_ID")
+    if not agent_id:
+        pytest.skip("LETTA_AGENT_ID not set - skipping E2E test")
+    return agent_id
diff --git a/tests/test_api_e2e.py b/tests/test_api_e2e.py
new file mode 100644
index 0000000..8b74802
--- /dev/null
+++ b/tests/test_api_e2e.py
@@ -0,0 +1,390 @@
+import pytest
+import requests
+import time
+import os
+from datetime import datetime, timezone, timedelta
+
+
+@pytest.mark.e2e
+class TestAuthentication:
+    def test_invalid_api_key_returns_401(self, api_base_url):
+        """Invalid API key should return 401 Unauthorized."""
+        headers = {"Authorization": "Bearer invalid-fake-key-123"}
+        response = requests.get(f"{api_base_url}/schedules/recurring", headers=headers)
+        assert response.status_code == 401
+    
+    def test_no_auth_header_returns_403(self, api_base_url):
+        """Missing auth header should return 403."""
+        response = requests.get(f"{api_base_url}/schedules/recurring")
+        assert response.status_code == 403
+    
+    def test_api_key_not_in_response(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """API key should never be returned in responses."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        
+        # Create schedule
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "cron": "*/5 * * * *",
+            "message": "Test message",
+            "role": "user"
+        }
+        response = requests.post(f"{api_base_url}/schedules/recurring", json=payload, headers=headers)
+        
+        # API key should not be in response
+        data = response.json()
+        assert "api_key" not in data
+        
+        # Clean up
+        schedule_id = data["id"]
+        requests.delete(f"{api_base_url}/schedules/recurring/{schedule_id}", headers=headers)
+
+
+@pytest.mark.e2e
+class TestRecurringScheduleCRUD:
+    def test_create_recurring_schedule(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """Should successfully create a recurring schedule."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "cron": "0 9 * * *",
+            "message": "Daily morning message",
+            "role": "user"
+        }
+        
+        response = requests.post(f"{api_base_url}/schedules/recurring", json=payload, headers=headers)
+        assert response.status_code == 201
+        
+        data = response.json()
+        assert "id" in data
+        assert data["cron"] == "0 9 * * *"
+        assert data["message"] == "Daily morning message"
+        
+        # Clean up
+        requests.delete(f"{api_base_url}/schedules/recurring/{data['id']}", headers=headers)
+    
+    def test_list_recurring_schedules(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """List should only return schedules for authenticated user."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        
+        # Create a schedule
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "cron": "*/10 * * * *",
+            "message": "Test",
+            "role": "user"
+        }
+        create_response = requests.post(f"{api_base_url}/schedules/recurring", json=payload, headers=headers)
+        schedule_id = create_response.json()["id"]
+        
+        # List schedules
+        response = requests.get(f"{api_base_url}/schedules/recurring", headers=headers)
+        assert response.status_code == 200
+        
+        schedules = response.json()
+        assert isinstance(schedules, list)
+        
+        # Find our schedule
+        our_schedule = next((s for s in schedules if s["id"] == schedule_id), None)
+        assert our_schedule is not None
+        assert our_schedule["cron"] == "*/10 * * * *"
+        
+        # Clean up
+        requests.delete(f"{api_base_url}/schedules/recurring/{schedule_id}", headers=headers)
+    
+    def test_get_recurring_schedule(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """Get should return specific schedule."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        
+        # Create schedule
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "cron": "0 12 * * *",
+            "message": "Noon message",
+            "role": "user"
+        }
+        create_response = requests.post(f"{api_base_url}/schedules/recurring", json=payload, headers=headers)
+        schedule_id = create_response.json()["id"]
+        
+        # Get schedule
+        response = requests.get(f"{api_base_url}/schedules/recurring/{schedule_id}", headers=headers)
+        assert response.status_code == 200
+        
+        data = response.json()
+        assert data["id"] == schedule_id
+        assert data["cron"] == "0 12 * * *"
+        
+        # Clean up
+        requests.delete(f"{api_base_url}/schedules/recurring/{schedule_id}", headers=headers)
+    
+    def test_delete_recurring_schedule(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """Delete should remove schedule."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        
+        # Create schedule
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "cron": "0 0 * * *",
+            "message": "To be deleted",
+            "role": "user"
+        }
+        create_response = requests.post(f"{api_base_url}/schedules/recurring", json=payload, headers=headers)
+        schedule_id = create_response.json()["id"]
+        
+        # Delete schedule
+        delete_response = requests.delete(f"{api_base_url}/schedules/recurring/{schedule_id}", headers=headers)
+        assert delete_response.status_code == 200
+        
+        # Verify it's gone
+        get_response = requests.get(f"{api_base_url}/schedules/recurring/{schedule_id}", headers=headers)
+        assert get_response.status_code == 404
+    
+    def test_delete_nonexistent_returns_404(self, api_base_url, valid_letta_api_key):
+        """Deleting non-existent schedule should return 404."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        response = requests.delete(f"{api_base_url}/schedules/recurring/fake-uuid-123", headers=headers)
+        assert response.status_code == 404
+
+
+@pytest.mark.e2e
+class TestOneTimeScheduleCRUD:
+    def test_create_onetime_schedule(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """Should successfully create a one-time schedule."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        future_time = datetime.now(timezone.utc) + timedelta(hours=1)
+        
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "execute_at": future_time.isoformat(),
+            "message": "Future message",
+            "role": "user"
+        }
+        
+        response = requests.post(f"{api_base_url}/schedules/one-time", json=payload, headers=headers)
+        assert response.status_code == 201
+        
+        data = response.json()
+        assert "id" in data
+        assert data["execute_at"] == future_time.isoformat()
+        
+        # Clean up
+        requests.delete(f"{api_base_url}/schedules/one-time/{data['id']}", headers=headers)
+    
+    def test_list_onetime_schedules(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """List should only return user's schedules."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        future_time = datetime.now(timezone.utc) + timedelta(hours=2)
+        
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "execute_at": future_time.isoformat(),
+            "message": "Test",
+            "role": "user"
+        }
+        create_response = requests.post(f"{api_base_url}/schedules/one-time", json=payload, headers=headers)
+        schedule_id = create_response.json()["id"]
+        
+        # List schedules
+        response = requests.get(f"{api_base_url}/schedules/one-time", headers=headers)
+        assert response.status_code == 200
+        
+        schedules = response.json()
+        our_schedule = next((s for s in schedules if s["id"] == schedule_id), None)
+        assert our_schedule is not None
+        
+        # Clean up
+        requests.delete(f"{api_base_url}/schedules/one-time/{schedule_id}", headers=headers)
+    
+    def test_delete_onetime_schedule(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """Delete should remove one-time schedule."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        future_time = datetime.now(timezone.utc) + timedelta(hours=3)
+        
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "execute_at": future_time.isoformat(),
+            "message": "To be deleted",
+            "role": "user"
+        }
+        create_response = requests.post(f"{api_base_url}/schedules/one-time", json=payload, headers=headers)
+        schedule_id = create_response.json()["id"]
+        
+        # Delete
+        delete_response = requests.delete(f"{api_base_url}/schedules/one-time/{schedule_id}", headers=headers)
+        assert delete_response.status_code == 200
+        
+        # Verify gone
+        get_response = requests.get(f"{api_base_url}/schedules/one-time/{schedule_id}", headers=headers)
+        assert get_response.status_code == 404
+
+
+@pytest.mark.e2e
+class TestResults:
+    def test_list_results(self, api_base_url, valid_letta_api_key):
+        """Should list execution results."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        response = requests.get(f"{api_base_url}/results", headers=headers)
+        assert response.status_code == 200
+        assert isinstance(response.json(), list)
+    
+    def test_get_result_nonexistent_returns_404(self, api_base_url, valid_letta_api_key):
+        """Getting non-existent result should return 404."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        response = requests.get(f"{api_base_url}/results/fake-uuid-123", headers=headers)
+        assert response.status_code == 404
+
+
+@pytest.mark.e2e
+@pytest.mark.slow
+class TestExecution:
+    def test_past_onetime_executes_immediately(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """One-time schedule in the past should execute within 1 minute."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        past_time = datetime.now(timezone.utc) - timedelta(seconds=30)
+        
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "execute_at": past_time.isoformat(),
+            "message": "Should execute immediately",
+            "role": "user"
+        }
+        
+        create_response = requests.post(f"{api_base_url}/schedules/one-time", json=payload, headers=headers)
+        schedule_id = create_response.json()["id"]
+        
+        # Wait up to 90 seconds for execution
+        max_wait = 90
+        for i in range(max_wait):
+            time.sleep(1)
+            
+            # Check if result exists
+            result_response = requests.get(f"{api_base_url}/results/{schedule_id}", headers=headers)
+            if result_response.status_code == 200:
+                result = result_response.json()
+                assert "run_id" in result
+                assert result["schedule_type"] == "one-time"
+                print(f"✓ Executed in {i+1} seconds, run_id: {result['run_id']}")
+                return
+        
+        pytest.fail(f"Schedule did not execute within {max_wait} seconds")
+    
+    def test_onetime_schedule_deleted_after_execution(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """One-time schedule should be deleted from filesystem after execution."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        past_time = datetime.now(timezone.utc) - timedelta(seconds=30)
+        
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "execute_at": past_time.isoformat(),
+            "message": "Should be deleted after execution",
+            "role": "user"
+        }
+        
+        create_response = requests.post(f"{api_base_url}/schedules/one-time", json=payload, headers=headers)
+        schedule_id = create_response.json()["id"]
+        
+        # Wait for execution
+        for _ in range(90):
+            time.sleep(1)
+            result_response = requests.get(f"{api_base_url}/results/{schedule_id}", headers=headers)
+            if result_response.status_code == 200:
+                break
+        
+        # Schedule should be deleted
+        schedule_response = requests.get(f"{api_base_url}/schedules/one-time/{schedule_id}", headers=headers)
+        assert schedule_response.status_code == 404
+    
+    def test_result_persists_after_schedule_deletion(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """Execution result should persist even after schedule is deleted."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        past_time = datetime.now(timezone.utc) - timedelta(seconds=30)
+        
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "execute_at": past_time.isoformat(),
+            "message": "Test result persistence",
+            "role": "user"
+        }
+        
+        create_response = requests.post(f"{api_base_url}/schedules/one-time", json=payload, headers=headers)
+        schedule_id = create_response.json()["id"]
+        
+        # Wait for execution
+        for _ in range(90):
+            time.sleep(1)
+            result_response = requests.get(f"{api_base_url}/results/{schedule_id}", headers=headers)
+            if result_response.status_code == 200:
+                result = result_response.json()
+                break
+        
+        # Result should still exist
+        final_result = requests.get(f"{api_base_url}/results/{schedule_id}", headers=headers)
+        assert final_result.status_code == 200
+        assert "run_id" in final_result.json()
+    
+    def test_no_duplicate_execution(self, api_base_url, valid_letta_api_key, valid_letta_agent_id):
+        """One-time schedule should execute exactly once, even if multiple executors spawn."""
+        headers = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        past_time = datetime.now(timezone.utc) - timedelta(seconds=30)
+        
+        payload = {
+            "agent_id": valid_letta_agent_id,
+            "api_key": valid_letta_api_key,
+            "execute_at": past_time.isoformat(),
+            "message": "Should only execute once",
+            "role": "user"
+        }
+        
+        create_response = requests.post(f"{api_base_url}/schedules/one-time", json=payload, headers=headers)
+        schedule_id = create_response.json()["id"]
+        
+        # Wait for execution
+        time.sleep(90)
+        
+        # Check result
+        result_response = requests.get(f"{api_base_url}/results/{schedule_id}", headers=headers)
+        assert result_response.status_code == 200
+        
+        # There should be exactly one result file, indicating one execution
+        # (We can't directly verify this without filesystem access, but we can check result exists)
+        result = result_response.json()
+        assert result["schedule_id"] == schedule_id
+        assert "run_id" in result
+
+
+@pytest.mark.e2e
+class TestAuthorization:
+    def test_user_isolation_list(self, api_base_url, valid_letta_api_key):
+        """Users should only see their own schedules in list."""
+        # This test requires a second valid API key
+        # Skip if not available
+        second_key = os.getenv("LETTA_API_KEY_2")
+        if not second_key:
+            pytest.skip("LETTA_API_KEY_2 not set for multi-user testing")
+        
+        headers1 = {"Authorization": f"Bearer {valid_letta_api_key}"}
+        headers2 = {"Authorization": f"Bearer {second_key}"}
+        
+        # User 1's list should not contain User 2's schedules
+        response1 = requests.get(f"{api_base_url}/schedules/recurring", headers=headers1)
+        response2 = requests.get(f"{api_base_url}/schedules/recurring", headers=headers2)
+        
+        schedules1 = response1.json()
+        schedules2 = response2.json()
+        
+        # Lists should be independent (no overlap)
+        ids1 = {s["id"] for s in schedules1}
+        ids2 = {s["id"] for s in schedules2}
+        
+        assert ids1.isdisjoint(ids2), "Users should not see each other's schedules"
diff --git a/tests/test_crypto_utils.py b/tests/test_crypto_utils.py
new file mode 100644
index 0000000..690df6e
--- /dev/null
+++ b/tests/test_crypto_utils.py
@@ -0,0 +1,133 @@
+import pytest
+import os
+from cryptography.fernet import Fernet
+from crypto_utils import (
+    get_api_key_hash,
+    is_dev_mode,
+    get_encryption_key,
+    encrypt_json,
+    decrypt_json,
+)
+
+
+class TestApiKeyHash:
+    def test_hash_is_deterministic(self):
+        """Same API key should produce same hash."""
+        api_key = "test-key-123"
+        hash1 = get_api_key_hash(api_key)
+        hash2 = get_api_key_hash(api_key)
+        assert hash1 == hash2
+    
+    def test_hash_is_16_chars(self):
+        """Hash should be 16 characters long."""
+        api_key = "test-key-123"
+        hash_val = get_api_key_hash(api_key)
+        assert len(hash_val) == 16
+    
+    def test_different_keys_different_hashes(self):
+        """Different API keys should produce different hashes."""
+        hash1 = get_api_key_hash("key1")
+        hash2 = get_api_key_hash("key2")
+        assert hash1 != hash2
+
+
+class TestDevMode:
+    def test_dev_mode_true(self):
+        """Dev mode should be enabled when env var is 'true'."""
+        os.environ["LETTA_SCHEDULES_DEV_MODE"] = "true"
+        assert is_dev_mode() is True
+    
+    def test_dev_mode_1(self):
+        """Dev mode should be enabled when env var is '1'."""
+        os.environ["LETTA_SCHEDULES_DEV_MODE"] = "1"
+        assert is_dev_mode() is True
+    
+    def test_dev_mode_yes(self):
+        """Dev mode should be enabled when env var is 'yes'."""
+        os.environ["LETTA_SCHEDULES_DEV_MODE"] = "yes"
+        assert is_dev_mode() is True
+    
+    def test_dev_mode_false(self):
+        """Dev mode should be disabled when env var is 'false'."""
+        os.environ["LETTA_SCHEDULES_DEV_MODE"] = "false"
+        assert is_dev_mode() is False
+    
+    def test_dev_mode_unset(self):
+        """Dev mode should be disabled when env var is not set."""
+        if "LETTA_SCHEDULES_DEV_MODE" in os.environ:
+            del os.environ["LETTA_SCHEDULES_DEV_MODE"]
+        assert is_dev_mode() is False
+
+
+class TestEncryption:
+    def test_encrypt_decrypt_roundtrip(self, encryption_key):
+        """Data should survive encryption/decryption roundtrip."""
+        original_data = {
+            "id": "test-123",
+            "message": "Hello world",
+            "nested": {"key": "value"}
+        }
+        
+        encrypted = encrypt_json(original_data, encryption_key)
+        decrypted = decrypt_json(encrypted, encryption_key)
+        
+        assert decrypted == original_data
+    
+    def test_encrypted_data_is_bytes(self, encryption_key):
+        """Encrypted data should be bytes."""
+        data = {"test": "data"}
+        encrypted = encrypt_json(data, encryption_key)
+        assert isinstance(encrypted, bytes)
+    
+    def test_dev_mode_plaintext(self):
+        """In dev mode, data should be plaintext JSON."""
+        os.environ["LETTA_SCHEDULES_DEV_MODE"] = "true"
+        
+        data = {"test": "data", "number": 123}
+        key = b"ignored-in-dev-mode"
+        
+        encrypted = encrypt_json(data, key)
+        assert isinstance(encrypted, bytes)
+        
+        # Should be valid JSON
+        import json
+        parsed = json.loads(encrypted)
+        assert parsed == data
+    
+    def test_dev_mode_decrypt(self):
+        """In dev mode, decrypt should parse plaintext JSON."""
+        os.environ["LETTA_SCHEDULES_DEV_MODE"] = "true"
+        
+        data = {"test": "data"}
+        key = b"ignored"
+        
+        encrypted = encrypt_json(data, key)
+        decrypted = decrypt_json(encrypted, key)
+        
+        assert decrypted == data
+    
+    def test_production_mode_encrypted(self):
+        """In production mode, data should be encrypted (not plaintext)."""
+        os.environ["LETTA_SCHEDULES_DEV_MODE"] = "false"
+        
+        data = {"test": "secret"}
+        key = Fernet.generate_key()
+        
+        encrypted = encrypt_json(data, key)
+        
+        # Should NOT be valid JSON
+        import json
+        with pytest.raises(json.JSONDecodeError):
+            json.loads(encrypted)
+    
+    def test_wrong_key_fails(self, encryption_key):
+        """Decrypting with wrong key should fail."""
+        os.environ["LETTA_SCHEDULES_DEV_MODE"] = "false"
+        
+        data = {"test": "data"}
+        encrypted = encrypt_json(data, encryption_key)
+        
+        wrong_key = Fernet.generate_key()
+        
+        with pytest.raises(Exception):
+            decrypt_json(encrypted, wrong_key)
diff --git a/tests/test_scheduler.py b/tests/test_scheduler.py
new file mode 100644
index 0000000..8e03a6e
--- /dev/null
+++ b/tests/test_scheduler.py
@@ -0,0 +1,123 @@
+import pytest
+from datetime import datetime, timezone, timedelta
+from scheduler import is_recurring_schedule_due, is_onetime_schedule_due
+
+
+class TestRecurringScheduleDue:
+    def test_new_schedule_is_due(self):
+        """A newly created schedule with no last_run should be due."""
+        schedule = {
+            "cron": "* * * * *",
+            "created_at": (datetime.now(timezone.utc) - timedelta(minutes=2)).isoformat(),
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        assert is_recurring_schedule_due(schedule, current_time) is True
+    
+    def test_just_ran_not_due(self):
+        """A schedule that just ran should not be due yet."""
+        schedule = {
+            "cron": "*/5 * * * *",
+            "created_at": (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat(),
+            "last_run": datetime.now(timezone.utc).isoformat(),
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        assert is_recurring_schedule_due(schedule, current_time) is False
+    
+    def test_five_minutes_ago_is_due(self):
+        """A schedule with last_run 5+ minutes ago should be due (cron every 5 min)."""
+        schedule = {
+            "cron": "*/5 * * * *",
+            "created_at": (datetime.now(timezone.utc) - timedelta(hours=1)).isoformat(),
+            "last_run": (datetime.now(timezone.utc) - timedelta(minutes=6)).isoformat(),
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        assert is_recurring_schedule_due(schedule, current_time) is True
+    
+    def test_daily_cron_not_due(self):
+        """Daily schedule that ran today should not be due."""
+        schedule = {
+            "cron": "0 9 * * *",  # Every day at 9am
+            "created_at": (datetime.now(timezone.utc) - timedelta(days=2)).isoformat(),
+            "last_run": datetime.now(timezone.utc).isoformat(),
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        assert is_recurring_schedule_due(schedule, current_time) is False
+    
+    def test_timezone_aware_comparison(self):
+        """Timezone-aware datetimes should be handled correctly."""
+        schedule = {
+            "cron": "*/5 * * * *",
+            "created_at": datetime.now(timezone.utc).isoformat(),
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        # Should not raise timezone comparison errors
+        is_recurring_schedule_due(schedule, current_time)
+
+
+class TestOneTimeScheduleDue:
+    def test_past_schedule_is_due(self):
+        """Schedule in the past should be due for execution."""
+        past_time = datetime.now(timezone.utc) - timedelta(minutes=5)
+        schedule = {
+            "execute_at": past_time.isoformat(),
+            "executed": False,
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        assert is_onetime_schedule_due(schedule, current_time) is True
+    
+    def test_future_schedule_not_due(self):
+        """Schedule in the future should not be due."""
+        future_time = datetime.now(timezone.utc) + timedelta(minutes=5)
+        schedule = {
+            "execute_at": future_time.isoformat(),
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        assert is_onetime_schedule_due(schedule, current_time) is False
+    
+    def test_executed_schedule_not_due(self):
+        """Already executed schedule should not be due (legacy check)."""
+        past_time = datetime.now(timezone.utc) - timedelta(minutes=5)
+        schedule = {
+            "execute_at": past_time.isoformat(),
+            "executed": True,
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        assert is_onetime_schedule_due(schedule, current_time) is False
+    
+    def test_timezone_est(self):
+        """Schedule with EST timezone should be handled correctly."""
+        # 2 hours ago EST
+        past_time = datetime.now(timezone.utc) - timedelta(hours=2)
+        schedule = {
+            "execute_at": past_time.isoformat(),
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        assert is_onetime_schedule_due(schedule, current_time) is True
+    
+    def test_timezone_naive_defaults_utc(self):
+        """Timezone-naive timestamps should be treated as UTC."""
+        past_time = datetime.now(timezone.utc) - timedelta(minutes=5)
+        schedule = {
+            "execute_at": past_time.replace(tzinfo=None).isoformat(),
+        }
+        current_time = datetime.now(timezone.utc)
+        
+        assert is_onetime_schedule_due(schedule, current_time) is True
+    
+    def test_exact_time_is_due(self):
+        """Schedule at exactly current time should be due."""
+        current_time = datetime.now(timezone.utc)
+        schedule = {
+            "execute_at": current_time.isoformat(),
+        }
+        
+        assert is_onetime_schedule_due(schedule, current_time) is True