Phase 3 Complete: Workflow, Tests, Documentation (#4)

* Complete Phase 3: Integration tests and documentation

- Added integration tests for error recovery and budget limits
- Created comprehensive workflow documentation (docs/WORKFLOW.md)
- Fixed SQLite checkpointer initialization
- Integration test suite in tests/integration/

* Fix async checkpointing with AsyncSqliteSaver

LANGSMITH_SETUP.md

@@ -1,51 +0,0 @@
-"""
-LangSmith Configuration and Setup Guide
-
-LangSmith provides observability for LangChain/LangGraph applications.
-It's critical for production debugging and performance optimization.
-"""
-
-# Setup Instructions:
-
-# 1. Sign up for LangSmith (free tier available):
-#    https://smith.langchain.com
-
-# 2. Get your API key from:
-#    https://smith.langchain.com/settings
-
-# 3. Add to .env file:
-#    LANGSMITH_API_KEY=ls_...
-#    LANGCHAIN_TRACING_V2=true
-#    LANGCHAIN_PROJECT=market-intelligence-prod
-#    LANGCHAIN_ENDPOINT=https://api.smith.langchain.com
-
-# 4. LangSmith will auto-trace all LangChain/LangGraph operations
-
-
-# What LangSmith Provides:
-
-# 1. Traces: Full execution tree
-#    - See which agent ran when
-#    - View all LLM calls and responses
-#    - Track token usage per call
-
-# 2. Debugging:
-#    - Why did the workflow fail?
-#    - Which prompt generated bad output?
-#    - What was the exact input that caused an error?
-
-# 3. Monitoring:
-#    - Latency per agent
-#    - Cost per run
-#    - Success/failure rates
-
-# 4. Optimization:
-#    - Compare different prompts
-#    - A/B test model choices
-#    - Identify bottlenecks
-
-
-# For Portfolio/Resume:
-# - Shows you understand production AI systems
-# - Demonstrates observability best practices
-# - Critical for enterprise deployments

docs/WORKFLOW.md

@@ -0,0 +1,231 @@
+# LangGraph Workflow Documentation
+
+## Overview
+
+The Market Intelligence workflow orchestrates three specialized agents using LangGraph's StateGraph to generate comprehensive market analysis reports.
+
+## Architecture
+
+```
+START → Research → Analysis → Writing → Human Review → END
+          ↓          ↓          ↓
+       Tavily    SWOT/Matrix  Report
+```
+
+### State Management
+
+The workflow maintains a shared state (`IntelligenceState`) that flows between agents:
+
+```python
+{
+    "company_name": str,
+    "industry": str | None,
+    "research_data": dict,      # From Research Agent
+    "swot": dict,                # From Analysis Agent  
+    "full_report": str,          # From Writer Agent
+    "total_cost": float,         # Cost tracking
+    "approved": bool,            # Human approval
+    # ... additional fields
+}
+```
+
+## Workflow Nodes
+
+### 1. Research Node
+- **Input**: Company name, industry
+- **Process**: Tavily search queries (company info, competitors, trends)
+- **Output**: Research data, competitors list, market trends
+- **Errors**: Network failures, API limits
+
+### 2. Analysis Node
+- **Input**: Research data
+- **Process**: LLM-powered SWOT, competitive positioning
+- **Output**: Structured analysis (SWOT, matrix, recommendations)
+- **Budget Check**: Enforces max cost before expensive analysis
+
+### 3. Writing Node
+- **Input**: Research + Analysis data
+- **Process**: Generate executive summary and full markdown report
+- **Output**: Professional business intelligence report
+
+### 4. Human Review Node
+- **Input**: Generated report
+- **Process**: Approval gate (currently auto-approves)
+- **Output**: Approval decision or revision request
+
+## Conditional Routing
+
+### Research → Analysis
+```python
+if errors or no_data:
+    END  # Stop workflow
+else:
+    CONTINUE to Analysis
+```
+
+### Human Review → END/Revision
+```python
+if approved:
+    END  # Complete
+elif max_revisions_reached:
+    END  # Give up
+else:
+    REVISE  # Loop back to Research
+```
+
+## Cost Management
+
+Budget is enforced at multiple points:
+- Before Analysis Node (most expensive)
+- After each LLM call via CostTracker
+- Workflow fails with BudgetExceededError if limit hit
+
+Default: $2.00 per run
+
+## Checkpointing
+
+SQLite checkpoints enable:
+- **Resume**: Continue after crashes
+- **Audit**: Full execution history
+- **Debug**: Inspect state at each step
+
+Checkpoint file: `./checkpoints.db`
+
+## Error Handling
+
+Errors accumulate in `state["errors"]` list:
+- Research failures → Workflow stops
+- Analysis errors → Logged, workflow may continue
+- Budget exceeded → Immediate stop
+
+## Usage Examples
+
+### Basic Usage
+
+```python
+from src.workflows.intelligence import MarketIntelligenceWorkflow
+
+workflow = MarketIntelligenceWorkflow()
+
+result = await workflow.run(
+    company_name="Tesla Model Y",
+    industry="Electric Vehicles"
+)
+
+print(result["full_report"])
+print(f"Cost: ${result['total_cost']:.2f}")
+```
+
+### Custom Budget
+
+```python
+workflow = MarketIntelligenceWorkflow(max_budget=5.0)
+
+result = await workflow.run(
+    company_name="Notion",
+    thread_id="notion-analysis-1"  # For checkpointing
+)
+```
+
+### Resume from Checkpoint
+
+```python
+# If workflow crashed, resume using same thread_id
+result = await workflow.run(
+    company_name="Notion",
+    thread_id="notion-analysis-1"  # Same ID resumes
+)
+```
+
+## Performance
+
+Typical execution:
+- **Time**: 3-5 minutes  
+- **Cost**: $0.00 (free Grok) to $1.50 (Claude 4.5)
+- **API Calls**: 6-8 LLM calls, 3 search queries
+- **Tokens**: 50K-100K total
+
+## Configuration
+
+Via `.env`:
+```bash
+DEFAULT_MODEL=x-ai/grok-4.1-fast:free  # Free tier
+MAX_COST_PER_RUN=2.0
+LANGCHAIN_TRACING_V2=true  # Enable LangSmith
+```
+
+## Observability
+
+With LangSmith enabled:
+- View full execution trace
+- Debug agent decisions
+- Optimize prompts
+- Track costs per call
+
+Dashboard: https://smith.langchain.com
+
+## Production Considerations
+
+1. **Checkpointing**: Essential for long-running workflows
+2. **Cost Limits**: Prevent runaway LLM costs
+3. **Error Recovery**: Graceful degradation
+4. **Human Review**: Required for high-stakes decisions
+5. **Observability**: Critical for debugging production issues
+
+## Testing
+
+```bash
+# Unit tests
+pytest tests/unit/test_workflow.py -v
+
+# Integration tests  
+pytest tests/integration/test_workflow_integration.py -v
+
+# End-to-end (uses real APIs)
+python scripts/test_workflow.py
+```
+
+## Extending
+
+### Add New Agent Node
+
+1. Create agent class in `src/agents/`
+2. Add node wrapper in workflow:
+   ```python
+   async def _my_agent_node(self, state):
+       result = await self.my_agent.run(state["research_data"])
+       return {"my_output": result}
+   ```
+3. Add to graph:
+   ```python
+   graph.add_node("my_agent", self._my_agent_node)
+   graph.add_edge("analysis", "my_agent")
+   ```
+
+### Modify Routing Logic
+
+Update conditional functions:
+```python
+def _should_use_special_analysis(self, state):
+    if state["company_name"].startswith("Enterprise"):
+        return "deep_analysis"
+    return "standard_analysis"
+```
+
+## Troubleshooting
+
+**Workflow stops early**:  
+- Check `result["errors"]` for failures  
+- Verify API keys in `.env`
+
+**Budget exceeded frequently**:  
+- Increase `max_budget` parameter  
+- Use cheaper models (grok-4.1-fast:free)
+
+**Slow performance**:  
+- Check LangSmith traces for bottlenecks  
+- Consider caching search results
+
+**Checkpoint errors**:  
+- Delete `checkpoints.db` to reset  
+- Check file permissions

src/workflows/intelligence.py

@@ -1,7 +1,7 @@
 """Main LangGraph workflow for market intelligence."""
 
 from langgraph.graph import StateGraph, END
-from langgraph.checkpoint.sqlite import SqliteSaver
+from langgraph.checkpoint.sqlite.aio import AsyncSqliteSaver
 
 from src.workflows.state import IntelligenceState
 from src.agents.researcher import ResearchAgent
@@ -79,9 +79,9 @@ class MarketIntelligenceWorkflow:
             {"approved": END, "revise": "research", "max_revisions": END},
         )
 
-        # Compile with SQLite checkpointing for production persistence
-        checkpointer = SqliteSaver.from_conn_string(self.checkpoint_path)
-        return graph.compile(checkpointer=checkpointer)
+        # Compile with async SQLite checkpointing
+        with AsyncSqliteSaver.from_conn_string(self.checkpoint_path) as checkpointer:
+            return graph.compile(checkpointer=checkpointer)
 
     async def _research_node(self, state: IntelligenceState) -> dict:
         """Research agent node."""

tests/integration/test_workflow_integration.py

@@ -0,0 +1,127 @@
+"""Integration tests for workflow error handling and cost limits."""
+
+import pytest
+from unittest.mock import AsyncMock, patch
+
+from src.workflows.intelligence import MarketIntelligenceWorkflow
+from src.utils.cost_tracker import BudgetExceededError
+
+
+@pytest.mark.asyncio
+class TestWorkflowErrorRecovery:
+    """Test workflow error handling and recovery."""
+
+    async def test_research_error_ends_workflow(self):
+        """Test workflow ends gracefully when research fails."""
+        workflow = MarketIntelligenceWorkflow()
+
+        # Mock research to fail
+        async def mock_research_error(state):
+            return {
+                "errors": ["Research API failed"],
+                "current_agent": "research",
+            }
+
+        workflow._research_node = mock_research_error
+
+        result = await workflow.run(company_name="Test Co", thread_id="test-error-1")
+
+        assert len(result["errors"]) > 0
+        assert result["current_agent"] == "research"
+
+    async def test_budget_exceeded_stops_workflow(self):
+        """Test workflow stops when budget is exceeded."""
+        workflow = MarketIntelligenceWorkflow(max_budget=0.001)
+
+        # Mock research to succeed with some cost
+        async def mock_research_success(state):
+            workflow.cost_tracker.track_usage("openai/gpt-5-mini", 10000, 5000)
+            return {
+                "current_agent": "research",
+                "research_data": {"some": "data"},
+                "competitors": [],
+                "market_trends": {},
+                "raw_sources": [],
+                "iteration": 1,
+            }
+
+        workflow._research_node = mock_research_success
+
+        result = await workflow.run(company_name="Test Co", thread_id="test-budget-1")
+
+        # Should have errors about budget
+        assert len(result.get("errors", [])) > 0 or result["total_cost"] < 0.001
+
+
+@pytest.mark.asyncio
+class TestWorkflowIntegration:
+    """Integration tests for full workflow."""
+
+    async def test_workflow_with_mocked_agents(self):
+        """Test complete workflow with mocked agent responses."""
+        workflow = MarketIntelligenceWorkflow()
+
+        # Mock all agents
+        async def mock_research(state):
+            return {
+                "current_agent": "research",
+                "research_data": {"company": "Test Co"},
+                "competitors": [{"name": "Competitor A"}],
+                "market_trends": {"trend": "growing"},
+                "raw_sources": [{"url": "test.com"}],
+                "iteration": state.get("iteration", 0) + 1,
+            }
+
+        async def mock_analysis(state):
+            return {
+                "current_agent": "analysis",
+                "swot": {"strengths": ["good"]},
+                "competitive_matrix": {},
+                "positioning": {},
+                "strategic_recommendations": {},
+            }
+
+        async def mock_writing(state):
+            return {
+                "current_agent": "writing",
+                "executive_summary": "Test summary",
+                "full_report": "# Test Report",
+                "report_metadata": {},
+                "total_cost": 0.0,
+                "total_tokens": 0,
+            }
+
+        workflow._research_node = mock_research
+        workflow._analysis_node = mock_analysis
+        workflow._writing_node = mock_writing
+
+        result = await workflow.run(
+            company_name="Test Co", thread_id="test-integration-1"
+        )
+
+        assert result["approved"] is True
+        assert "Test summary" in result["executive_summary"]
+        assert result["total_cost"] == 0.0
+
+
+class TestWorkflowCheckpointing:
+    """Test checkpoint persistence and recovery."""
+
+    def test_checkpoint_file_created(self):
+        """Test that checkpoint database is created."""
+        import os
+
+        checkpoint_path = "./test_checkpoint.db"
+
+        # Clean up first
+        if os.path.exists(checkpoint_path):
+            os.remove(checkpoint_path)
+
+        workflow = MarketIntelligenceWorkflow(checkpoint_path=checkpoint_path)
+
+        # Checkpoint file should be created when workflow is compiled
+        assert workflow.workflow is not None
+
+        # Clean up
+        if os.path.exists(checkpoint_path):
+            os.remove(checkpoint_path)