diff --git a/intelligence/deepdive/tests/test_e2e.py b/intelligence/deepdive/tests/test_e2e.py
new file mode 100644
index 0000000..1915dc6
--- /dev/null
+++ b/intelligence/deepdive/tests/test_e2e.py
@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""End-to-end pipeline test (dry-run)"""
+
+import asyncio
+import pytest
+import yaml
+from datetime import datetime, timedelta
+from pathlib import Path
+import sys
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from pipeline import DeepDivePipeline
+
+
+class TestEndToEnd:
+    """End-to-end pipeline tests."""
+    
+    @pytest.fixture
+    def test_config(self):
+        """Minimal test configuration."""
+        return {
+            'sources': [
+                {
+                    'name': 'arxiv_cs_ai',
+                    'url': 'http://export.arxiv.org/rss/cs.AI',
+                    'max_items': 5
+                }
+            ],
+            'relevance': {
+                'model': 'all-MiniLM-L6-v2',
+                'top_n': 3,
+                'min_score': 0.3
+            },
+            'synthesis': {
+                'llm_endpoint': 'http://localhost:11435/v1'
+            },
+            'audio': {
+                'enabled': False
+            },
+            'delivery': {
+                # Empty = no live delivery
+            }
+        }
+    
+    @pytest.mark.asyncio
+    async def test_full_pipeline_dry_run(self, test_config):
+        """Test full pipeline execution (no LLM, no delivery)."""
+        pipeline = DeepDivePipeline(test_config)
+        
+        since = datetime.utcnow() - timedelta(hours=48)
+        result = await pipeline.run(since=since, dry_run=True)
+        
+        # Should complete successfully
+        assert result['status'] in ['success', 'empty']
+        
+        if result['status'] == 'success':
+            assert 'items_aggregated' in result
+            assert 'items_ranked' in result
+            assert 'briefing_path' in result
+            
+            # Verify briefing file was created
+            if result.get('briefing_path'):
+                briefing_path = Path(result['briefing_path'])
+                assert briefing_path.exists(), "Briefing file should exist"
+                
+                # Verify it's valid JSON
+                import json
+                with open(briefing_path) as f:
+                    briefing = json.load(f)
+                    assert 'headline' in briefing
+                    assert 'briefing' in briefing
+    
+    def test_pipeline_initialization(self, test_config):
+        """Test pipeline components initialize correctly."""
+        pipeline = DeepDivePipeline(test_config)
+        
+        assert pipeline.aggregator is not None
+        assert pipeline.scorer is not None
+        assert pipeline.synthesizer is not None
+        assert pipeline.telegram is None  # No token configured
+
+
+if __name__ == "__main__":
+    pytest.main([__file__, "-v"])