docs/ux.md

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):

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:

# 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):

@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:

# 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):

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:

# 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:

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:

# 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:

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:

# 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}<br><b>{borough}</b>",
        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