# User Experience
UX patterns for streaming, error handling, and traceability.
## Real-time Streaming
### Agent Progress Display
The UI shows agent activity in real-time using a Claude Code-inspired format:
```
● 🎯 **Triage Agent**
├── Geocoding Address("Broadway and 42nd")
│ ⎿ ✓ Manhattan, NYC
● 🔍 **Research Agent**
├── Looking Up City Records...
│ ⎿ RD-MN-0042, 3 complaints
├── Checking Nearby Reports...
│ ⎿ 2 reports (1 open)
├── Getting Weather...
│ ⎿ 45°F, Clear
● 📋 **Report Agent**
├── Getting Department Info...
│ ⎿ DOT (24-48h response)
├── Generating PDF Report...
│ ⎿ 📄 FMN-20251129 (12.5KB)
`8.3s`
```
**Implementation** (`app.py:222-277`):
```python
def build_inline_progress(agent_states: dict, elapsed: float,
thinking: str = "", tool_history: list = None) -> str:
"""Build multi-agent style progress display."""
lines = []
for tool in tool_history:
agent = tool.get("agent")
name = tool.get("display_name")
result = tool.get("result_summary", "")
# Agent header
if agent != current_agent:
lines.append(f"● {icon} **{agent_name}**")
# Tool call line
lines.append(f" ├── {name}...")
# Result line
if result:
lines.append(f" │ ⎿ {result}")
return "\n".join(lines)
```
### Event-Driven Updates
Events flow from agents to UI via generator:
```python
# In controller.py
for event_type, data in orchestrator.process(message, image_analysis):
if event_type == "tool_call":
# Update UI immediately
yield build_outputs()
elif event_type == "agent_done":
# Show completion
yield build_outputs()
elif event_type == "complete":
# Final response
yield build_outputs()
```
### Result Summarization
Tool results are summarized for display (`app.py:68-198`):
| Tool | Raw Output | Display Summary |
|------|------------|-----------------|
| `validate_address` | `{"is_valid_nyc": true, "borough": "Manhattan", ...}` | `✓ Manhattan` |
| `weather_get_current` | `{"temp_f": 45, "conditions": "Clear", ...}` | `45°F, Clear` |
| `get_nearby_reports` | `{"total_reports": 2, "open_reports": 1, ...}` | `2 reports (1 open)` |
| `pdf_generate_report` | `{"report_id": "FMN-...", "size_kb": 12.5, ...}` | `📄 FMN-... (12.5KB)` |
## Reasoning Trace
### Visible Thinking
The Reasoning Trace panel shows the agent's thought process:
```
**🧠 Autonomous Reasoning Trace**
[0.0s] 💭 Received infrastructure report request
[0.2s] 💭 Delegating full control to Controller agent
[1.5s] 💭 Controller delegated to Triage Agent
[2.3s] 🔧 Agent calling: Geocoding Address
[3.1s] 👁️ Tool result: {"is_valid_nyc": true...
[3.2s] ✅ Triage Agent completed (1.7s) [confidence: 0.85]
[3.3s] 💭 Controller delegated to Research Agent
...
[8.0s] ✅ Presenting Controller's response to user [confidence: 0.92]
---
**Final Quality Score:** 8.7/10
```
**Implementation** (`agents/reasoning.py`):
```python
@dataclass
class ReasoningTrace:
thoughts: List[Dict[str, Any]] = field(default_factory=list)
def think(self, thought: str, category: str = "reasoning") -> None:
"""Record a thought."""
self.thoughts.append({
"type": "thought",
"category": category,
"content": thought,
"timestamp": time.time() - self.start_time,
})
def to_display(self) -> str:
"""Format for UI display."""
lines = []
for t in self.thoughts:
icon = {"thought": "💭", "decision": "✅", "observation": "👁️"}
lines.append(f"[{t['timestamp']:.1f}s] {icon} {t['content']}")
return "\n".join(lines)
```
### Self-Evaluation Display
Quality scores shown after completion:
```python
# In controller.py
quality_score = self._evaluate_response_quality(response, agent_states, trace)
# In app.py
if quality_score is not None:
current_reasoning += f"\n\n**Final Quality Score:** {quality_score * 10:.1f}/10"
```
## Error Handling
### Friendly Error Messages
Technical errors are mapped to user-friendly messages:
| Error Type | User Sees |
|------------|-----------|
| Rate limit | "You're sending requests too quickly. Please wait 30 seconds." |
| API timeout | "We're having trouble connecting. Please try again." |
| Invalid input | "Please provide a valid NYC address." |
| MCP server down | "The infrastructure tools server is temporarily unavailable." |
**Implementation** (`security/error_handler.py`):
```python
class ErrorHandler:
def handle(self, error: Exception, context: str = "") -> FriendlyError:
if isinstance(error, RateLimitExceeded):
return FriendlyError(
message="You're sending requests too quickly.",
suggestion=f"Please wait {error.wait_seconds} seconds.",
is_recoverable=True,
)
```
### Graceful Degradation
If services are unavailable:
```python
# MCP server check
mcp_available, mcp_error = self._check_mcp_server()
if not mcp_available:
yield ("needs_info", {
"message": "⚠️ **Service Temporarily Unavailable**\n\n"
"The NYC infrastructure tools server is offline.\n\n"
"Please wait a moment and try again."
})
return
```
### Input Validation Feedback
Clear feedback for invalid input:
```python
if not validation_result.is_valid:
error_msg = validation_result.get_friendly_message()
# "Please provide a description of the infrastructure issue."
# "Your message is too long. Please limit to 5000 characters."
```
## Multi-turn Conversation
### Context Awareness
The Controller sees full conversation history:
```python
# In prompts.py
CONVERSATION HISTORY (you have full context):
User: pothole on my street
Agent: I'd be happy to help! To file a report, I need the street address...
User: Broadway and 42nd St
```
### Follow-up Questions
Controller asks for missing information naturally:
```
User: "There's a pothole"
Agent: "I'd be happy to help you report that pothole! To file an accurate
report with NYC DOT, I need a bit more information:
**What's the street address?** (e.g., "Broadway and 42nd St" or
"123 Main Street, Brooklyn")
Once you provide the location, I can look up the city records and
generate an official report."
```
## Map Integration
### Real-time Location Updates
Map updates as soon as location is validated:
```python
elif event_type == "location_update":
lat = data.get("lat")
lon = data.get("lon")
if lat and lon:
current_map = create_issue_map(
lat=lat,
lon=lon,
borough=data.get("borough"),
address=data.get("address")
)
yield build_outputs()
```
### Folium Map Display
Interactive map with issue marker:
```python
# In ui/mapping.py
def create_issue_map(lat, lon, borough, address):
m = folium.Map(location=[lat, lon], zoom_start=15)
folium.Marker(
[lat, lon],
popup=f"{address}
{borough}",
icon=folium.Icon(color="red", icon="exclamation-sign")
).add_to(m)
folium.Circle(
[lat, lon],
radius=50,
color="red",
fill=True,
).add_to(m)
return m._repr_html_()
```
## Agent Pipeline Display
### Status Cards
Each agent has a status card showing:
```
🔍 **Research Agent** ✅ 3.2s
*Classifies issue & gathers location data*
- ✅ Looking Up City Records (1.2s)
- ✅ Checking Nearby Reports (0.8s)
- ✅ Getting Weather (1.1s)
```
### Completion Summary
Final response includes execution summary:
```
**🤖 Agent Execution**
● 🎯 **Triage Agent**
└─ Geocoding Address → ✓ Broadway & 42nd, Manhattan
● 🔍 **Research Agent**
└─ Looking Up City Records → RD-MN-0042, 3 complaints
└─ Checking Nearby Reports → 2 reports (1 open)
└─ Getting Weather → 45°F, Clear
● 📋 **Report Agent**
└─ Getting Department Info → DOT (24-48h response)
└─ Generating PDF Report → 📄 FMN-20251129
*Completed in 8.3s*
---
## Infrastructure Report
**Report ID:** FMN-20251129-001
**Issue Type:** Pothole
**Location:** Broadway & 42nd St, Manhattan
...
```
## Accessibility
### Clear Visual Hierarchy
- Agent names in **bold**
- Tool calls with tree-style indentation
- Results with checkmarks/icons
- Timing in subtle `code` format
### Status Icons
| Icon | Meaning |
|------|---------|
| ⌛ | Pending |
| ⏳ | Running |
| ✅ | Completed |
| ❌ | Error |
| 📄 | PDF generated |
| ✉️ | Email sent |