Skip to content

Travel Assistant Roadmap

Achieving Next-Gen Personal Travel Assistant Capabilities

Created: December 2025 Status: Planning Target: Full Case Study Parity

Executive Summary

This roadmap defines how to implement the advanced travel assistant features from the case studies while maintaining clean architectural boundaries. The key insight is that not everything should be a mini-app — some features work better as Sage core capabilities, shared infrastructure, or proactive workflows.

Architecture Principles

  1. Sage Core = Conversational, memory-based, works everywhere without session state
  2. Mini-Apps = Structured, stateful, domain-specific workflows with rich UI
  3. Superpowers = Proactive background automation (triggers, schedules)
  4. Shared Infrastructure = Services used by multiple components

Feature Ownership Matrix

Feature Owner Rationale
"What's good for dinner?" Sage Core Quick Q&A, no workflow needed
"Plan my Tokyo trip" Trip Planner Structured, stateful, multi-turn
Venue management Trip Planner CRUD on trip-specific data
Map visualization Trip Planner Trip-specific UI artifact
"My train is cancelled!" Sage Core → Transport Solver Sage handles crisis, hands off to mini-app
Dietary allergy cards Trip Planner Destination-specific artifact
Real-time venue hours Shared Infrastructure Used by Sage queries AND trip planner
Proactive flight reminders Superpower Background automation, already exists
"How do I say X in Japanese?" Sage Core Quick translation, no session
User dietary preferences Supermemory (User Profile) Persistent across all contexts

Phase 1: Shared Infrastructure (Foundation)

Timeline: 2-3 weeks Priority: Critical — Unblocks all other phases

1.1 Location Services (app/services/location/)

New Files: 

app/services/location/
├── __init__.py
├── geocoder.py           # Address → lat/lng
├── places_client.py      # Google Places API wrapper
├── distance_calculator.py # Walking/driving times
└── types.py              # Shared types

Geocoder Service: 

class GeocoderService:
    """Convert addresses to coordinates with caching."""

    async def geocode(self, address: str, region_hint: str = None) -> Optional[Coordinates]:
        """
        Geocode an address with smart caching.

        Cache Strategy:
        - Redis cache: 24h TTL for geocoded results
        - Negative cache: 1h for "not found" to avoid re-queries
        """

    async def batch_geocode(self, addresses: List[str]) -> Dict[str, Coordinates]:
        """Geocode multiple addresses efficiently (max 50 per request)."""

Places Client: 

class PlacesClient:
    """Google Places API integration for venue enrichment."""

    async def get_place_details(self, place_id: str) -> PlaceDetails:
        """Get full place details including hours, ratings, photos."""

    async def search_nearby(
        self,
        location: Coordinates,
        query: str = None,
        category: str = None,
        radius_meters: int = 1000
    ) -> List[PlaceResult]:
        """Search for places near a location."""

    async def get_opening_hours(self, place_id: str) -> OpeningHours:
        """Check if a place is open and get today's hours."""

Distance Calculator: 

class DistanceCalculator:
    """Calculate distances and travel times between points."""

    async def get_walking_time(self, origin: Coordinates, dest: Coordinates) -> Duration:
        """Walking time between two points."""

    async def get_route(
        self,
        origin: Coordinates,
        destinations: List[Coordinates],
        mode: TravelMode = "walking"
    ) -> RouteResult:
        """Get optimized route through multiple points."""

Cost Estimate: Google Maps API ~$5-10/month for MVP usage

1.2 User Preferences System

Storage: Supermemory User Profiles (already integrated)

Schema Extension: 

# In memory storage, not database
USER_PREFERENCE_SCHEMA = {
    "dietary": {
        "restrictions": ["peanut_allergy", "vegetarian", "halal"],
        "preferences": ["spicy", "local_cuisine"],
        "avoid": ["shellfish"]
    },
    "travel_style": {
        "pace": "relaxed",  # relaxed, moderate, packed
        "budget": "mid_range",  # budget, mid_range, luxury
        "interests": ["photography", "food", "history"]
    },
    "communication": {
        "language_help": True,
        "proactive_suggestions": True
    }
}

Memory Integration: 

# Add to SupermemoryService
async def get_user_preferences(self, user_id: str) -> UserPreferences:
    """Retrieve user preferences from User Profile."""

async def update_user_preference(
    self,
    user_id: str,
    category: str,
    key: str,
    value: Any
) -> None:
    """Update a specific preference."""

Preference Learning: - When user says "I'm allergic to peanuts", Sage stores to preferences - When user says "I love spicy food", Sage stores to preferences - Preferences are available to ALL contexts (Sage, trip planner, etc.)

1.3 Translation Service (app/services/translation/)

Integration: Google Translate API or OpenAI

class TranslationService:
    """Translation with travel-specific context."""

    async def translate(
        self,
        text: str,
        target_language: str,
        context: str = None  # "restaurant", "emergency", "navigation"
    ) -> Translation:
        """Translate text with context-aware phrasing."""

    async def generate_allergy_card(
        self,
        allergies: List[str],
        destination_country: str
    ) -> AllergyCard:
        """Generate a translated allergy card for a destination."""

Phase 2: Sage Core Upgrades

Timeline: 2 weeks Priority: High — Enables conversational travel features

2.1 Travel-Aware Context Loading

Current Flow: 

User message → Stage 1 classify → Stage 2 generate

Enhanced Flow: 

User message
    → Stage 1 classify (detect travel context)
    → Load travel context if relevant:
        - Active trip from memory
        - User location (if shared)
        - User preferences
    → Stage 2 generate with travel awareness

Implementation in two_stage_handler.py: 

async def _enrich_travel_context(
    self,
    user_id: str,
    intent: ClassificationResult
) -> Dict[str, Any]:
    """Enrich context with travel-relevant data."""

    travel_context = {}

    # Check for active trips in memory
    trips = await self.memory_service.search(
        user_id=user_id,
        query="current trip destination",
        limit=1
    )
    if trips:
        travel_context["active_trip"] = trips[0]

    # Load dietary preferences if food-related
    if intent.category in ["food", "restaurant", "dining"]:
        prefs = await self.memory_service.get_user_preferences(user_id)
        travel_context["dietary"] = prefs.dietary

    return travel_context

2.2 Crisis Detection & Handoff

Sage detects crisis → Responds empathetically → Offers mini-app handoff

Crisis Keywords: 

CRISIS_PATTERNS = [
    r"(cancelled|canceled|struck|strike|delayed|stuck|stranded)",
    r"(missed|lost) (my|the) (flight|train|bus|connection)",
    r"(emergency|urgent|help).*(transport|travel|flight)",
    r"can't get (to|there|home)",
]

Response Pattern: 

User: "My train to Bordeaux got cancelled!"

Sage (immediate, empathetic):
"oh no!! don't panic — let me check alternatives rn"
"checking flights, cars, and buses that can get you there"

[2 seconds later]

Sage (with options):
"ok found some options:
• Air France 2pm from Orly (€180, arrives 3pm)
• Hertz car rental at the station (€90, ~6h drive)

want me to open a comparison dashboard so you can book?"

Handoff to Transport Solver: 

# If user says yes, create transport_solver session
if user_confirms_handoff:
    session = await session_manager.create_session(
        chat_guid=chat_guid,
        user_id=user_id,
        mini_app_id="transport_solver",
        initial_data={
            "origin": "Paris Montparnasse",
            "destination": "Bordeaux",
            "deadline": "8 PM today",
            "crisis_mode": True
        }
    )

2.3 Quick Travel Queries (No Mini-App)

Sage should handle these WITHOUT creating a session:

Query Response Source
"What's the weather in Tokyo?" Weather API (existing)
"Is TeamLabs open now?" Places Client → opening hours
"How do I say 'no peanuts' in Japanese?" Translation Service
"What's a good dinner spot near Shibuya?" Parallel/Perplexity + preferences
"How far is it from Shibuya to Ginza?" Distance Calculator

Integration Point: Add to SmartRouter: 

# New data sources
DataSource.PLACES = "places"
DataSource.TRANSLATION = "translation"
DataSource.DISTANCE = "distance"

# Query patterns
PLACES_PATTERNS = ["is .* open", "hours for", "open now"]
TRANSLATION_PATTERNS = ["how do i say", "translate", "in japanese"]
DISTANCE_PATTERNS = ["how far", "distance to", "walk to"]

Phase 3: Trip Planner Enhancements

Timeline: 3-4 weeks Priority: High — Core case study features

3.1 Venue Geocoding Pipeline

When venues are added: 

async def _add_venue(self, trip: TripState, ...) -> MiniAppResponse:
    venue = Venue(...)

    # Async geocode in background
    asyncio.create_task(self._geocode_venue(venue, trip.destination))

    # Return immediately
    return MiniAppResponse(...)

async def _geocode_venue(self, venue: Venue, destination: str):
    """Background task to geocode venue."""
    geocoder = get_geocoder_service()

    # Try specific address first
    if venue.address:
        coords = await geocoder.geocode(f"{venue.address}, {destination}")
    else:
        # Fallback to name + district
        coords = await geocoder.geocode(
            f"{venue.name}, {venue.district}, {destination}"
        )

    if coords:
        venue.latitude = coords.lat
        venue.longitude = coords.lng
        # Update session state
        await self._update_venue_coords(venue.id, coords)

Schema Update: 

@dataclass
class Venue:
    # ... existing fields ...

    # New: Geocoding
    latitude: Optional[float] = None
    longitude: Optional[float] = None
    place_id: Optional[str] = None  # Google Place ID for enrichment

    # New: Enrichment data
    opening_hours: Optional[Dict[str, str]] = None  # {"Monday": "9am-10pm", ...}
    price_level: Optional[int] = None  # 1-4
    google_rating: Optional[float] = None
    google_reviews_count: Optional[int] = None
    photos: List[str] = field(default_factory=list)  # Photo URLs

    # New: Dietary
    dietary_options: List[str] = field(default_factory=list)  # vegetarian, vegan, etc.

3.2 Real-Time Venue Enrichment

On-Demand Enrichment: 

async def _enrich_venue(self, venue: Venue) -> Venue:
    """Fetch real-time data for a venue."""
    places = get_places_client()

    if venue.place_id:
        details = await places.get_place_details(venue.place_id)
    else:
        # Search for place
        results = await places.search_nearby(
            location=Coordinates(venue.latitude, venue.longitude),
            query=venue.name,
            radius_meters=200
        )
        if results:
            venue.place_id = results[0].place_id
            details = await places.get_place_details(venue.place_id)

    # Update venue with enriched data
    venue.opening_hours = details.opening_hours
    venue.price_level = details.price_level
    venue.google_rating = details.rating
    venue.dietary_options = self._extract_dietary(details)

    return venue

"Tell me about X" Handler: 

if message_lower.startswith(("tell me about", "what about", "info on")):
    venue_name = extract_venue_reference(message)
    venue = self._find_venue_by_name(trip, venue_name)

    if venue:
        enriched = await self._enrich_venue(venue)
        return self._format_venue_details(enriched)

3.3 Dietary Safety Features

Allergy Card Generation: 

async def _generate_allergy_card(
    self,
    context: MiniAppContext,
    trip: TripState
) -> MiniAppResponse:
    """Generate a translated allergy card."""

    # Get user's allergies from preferences
    prefs = await self.memory_service.get_user_preferences(context.user_id)
    allergies = prefs.dietary.get("restrictions", [])

    if not allergies:
        return MiniAppResponse(
            message="i don't have any allergies on file for you. what should i add?"
        )

    # Generate card using translation service
    translation = get_translation_service()
    card = await translation.generate_allergy_card(
        allergies=allergies,
        destination_country=self._get_country(trip.destination)
    )

    # Return UI config with card
    return MiniAppResponse(
        message=f"here's your allergy card for {trip.destination}",
        ui_update=True,
        web_ui_view="allergy_card"
    )

Venue Filtering by Dietary: 

async def _query_venues(self, trip: TripState, query: str, context: MiniAppContext):
    # Get user preferences
    prefs = await self.memory_service.get_user_preferences(context.user_id)
    dietary_restrictions = prefs.dietary.get("restrictions", [])

    venues = trip.venues

    # Apply dietary filter if user has restrictions
    if dietary_restrictions and "safe" in query.lower():
        venues = [
            v for v in venues
            if self._is_dietary_compatible(v, dietary_restrictions)
        ]

3.4 Smart Suggestion Engine

Gap Detection: 

async def _detect_gaps_and_suggest(
    self,
    trip: TripState,
    day_schedule: List[ScheduledVenue]
) -> List[Suggestion]:
    """Detect gaps in schedule and suggest fills."""

    suggestions = []

    for i in range(len(day_schedule) - 1):
        current = day_schedule[i]
        next_venue = day_schedule[i + 1]

        # Check for time gap
        gap_hours = (next_venue.time - current.end_time).total_seconds() / 3600

        if gap_hours >= 1.5:  # 1.5+ hour gap
            # Find nearby venues that fit
            nearby = await self._find_nearby_unvisited(
                trip,
                location=current.venue.coordinates,
                category_hint=self._suggest_category_for_time(next_venue.time),
                user_preferences=await self._get_user_prefs(context.user_id)
            )

            if nearby:
                suggestions.append(Suggestion(
                    venue=nearby[0],
                    reason=f"fits between {current.venue.name} and {next_venue.venue.name}",
                    walking_time=await self._walking_time(current.venue, nearby[0])
                ))

    return suggestions

User Preference Matching: 

def _score_venue_for_user(
    self,
    venue: Venue,
    user_prefs: UserPreferences
) -> float:
    """Score how well a venue matches user preferences."""

    score = 0.0

    # Match interests
    for interest in user_prefs.travel_style.interests:
        if interest in venue.vibe or interest in venue.category:
            score += 0.3

    # Match budget
    if venue.price_level and user_prefs.travel_style.budget:
        budget_map = {"budget": 1, "mid_range": 2, "luxury": 4}
        if venue.price_level <= budget_map.get(user_prefs.budget, 3):
            score += 0.2

    # Dietary compatibility
    if self._is_dietary_compatible(venue, user_prefs.dietary.restrictions):
        score += 0.3

    return score

3.5 Interactive Map View

UI Config for Map: 

def _build_map_view(self, trip: TripState, day: str = None) -> Dict[str, Any]:
    """Generate map view configuration."""

    # Filter venues by day if specified
    venues = trip.venues if not day else [
        v for v in trip.venues if v.scheduled_day == day
    ]

    # Build markers
    markers = []
    for i, venue in enumerate(venues):
        if venue.latitude and venue.longitude:
            markers.append({
                "id": venue.id,
                "label": f"{i+1}. {venue.name}",
                "lat": venue.latitude,
                "lng": venue.longitude,
                "category": venue.category,
                "status": venue.status,
                "color": self._status_color(venue.status),
                "icon": self._category_icon(venue.category)
            })

    # Calculate route if day view
    route = None
    if day and len(markers) >= 2:
        route = await self._get_route([m for m in markers])

    return {
        "type": "map_view",
        "props": {
            "center": self._calculate_center(markers),
            "zoom": self._calculate_zoom(markers),
            "markers": markers,
            "route": route,
            "show_walking_times": True
        }
    }

3.6 Export Features

PDF Export: 

async def _export_pdf(self, trip: TripState) -> bytes:
    """Generate a PDF itinerary."""
    # Use a PDF library like WeasyPrint or ReportLab
    # Include: venues by day, map thumbnails, allergy card

Google Maps List Export: 

async def _export_to_google_maps(self, trip: TripState, user_id: str):
    """Export venues to a Google Maps saved list."""
    # Requires OAuth - use google_client.py
    # Creates a new list named "Trip to {destination}"

Phase 4: Transport Solver Mini-App (NEW)

Timeline: 3 weeks Priority: Medium — Addresses Case 3 (Crisis Averted)

4.1 Overview

A new mini-app specifically for transport crisis management. Unlike trip planner (planning-focused), this is action-focused for immediate decisions.

Trigger Patterns: 

TRANSPORT_CRISIS_TRIGGERS = [
    r"(cancelled|canceled|delayed|struck) (flight|train|bus)",
    r"stuck at (airport|station)",
    r"missed my (connection|flight|train)",
    r"need to get to .* by",
    r"alternative (routes?|transport|ways?) to"
]

4.2 Handler Architecture

# app/miniapps/handlers/transport_solver.py

class TransportSolverHandler(BaseMiniAppHandler):
    app_id = "transport_solver"
    app_name = "Transport Solver"
    url_path = "transport"

    # State machine
    STATES = ["crisis_intake", "options_generated", "booking"]

State: crisis_intake 

@dataclass
class TransportCrisis:
    origin: str
    origin_coords: Optional[Coordinates]
    destination: str
    destination_coords: Optional[Coordinates]
    deadline: Optional[datetime]
    constraints: List[str]  # "no flying", "budget under $200", etc.
    crisis_type: str  # "cancellation", "missed_connection", "delay"

State: options_generated 

@dataclass
class TransportOption:
    mode: str  # "flight", "train", "car", "bus"
    provider: str  # "Air France", "Hertz", "FlixBus"
    departure_time: datetime
    arrival_time: datetime
    price: float
    currency: str
    booking_url: str
    pros: List[str]
    cons: List[str]
    reliability_score: float  # Based on current conditions

4.3 Data Sources

Flight Search: - Primary: Amadeus Self-Service API (free tier) - Fallback: Perplexity search

Train/Bus Search: - Rome2Rio API or Perplexity search

Car Rental: - Kayak affiliate links or Perplexity search

Real-Time Conditions: - FlightAware for flight status - Google Maps for traffic

4.4 UI Config

def get_ui_config(self, context) -> Dict[str, Any]:
    return {
        "app_id": "transport_solver",
        "theme": "emergency",  # Red accents, urgent styling
        "header": {
            "title": f"Get to {self.crisis.destination}",
            "subtitle": f"Deadline: {self.crisis.deadline}",
            "countdown": True  # Shows time remaining
        },
        "components": [
            {
                "type": "stats_grid",
                "props": {
                    "stats": [
                        {"label": "Fastest", "value": "2h 15m", "icon": "plane"},
                        {"label": "Cheapest", "value": "€90", "icon": "wallet"}
                    ]
                }
            },
            {
                "type": "card_list",
                "id": "options",
                "props": {
                    "cards": [
                        self._format_option(opt) for opt in self.options
                    ]
                }
            }
        ]
    }

Phase 5: Proactive Travel Intelligence

Timeline: 2 weeks Priority: Medium — Enhances existing superpowers

5.1 Enhance Travel Support Workflow

Current: app/superpowers/catalog/proactive/travel_support.py

Enhancements: - Add venue opening hours check before suggesting - Add weather-aware suggestions ("it's raining, maybe skip the outdoor market") - Add restaurant reservation reminders

5.2 New Workflow: Destination Prep

# app/superpowers/catalog/proactive/destination_prep.py

destination_prep_workflow = Workflow(
    id="proactive_destination_prep",
    name="Destination Prep",
    description="Prepare user for upcoming trip",
    trigger=WorkflowTrigger(
        type="schedule_trigger",
        config={"cron": "0 9 * * *"}  # Daily at 9 AM
    ),
    nodes=[
        # Check for trips starting in next 7 days
        # Generate packing suggestions
        # Check visa requirements
        # Generate allergy cards
        # Weather forecast summary
    ]
)

Phase 6: Frontend Implementation

Timeline: 4 weeks (parallel with backend) Owner: Frontend Team

6.1 Map Component (web/components/MapView.tsx)

  • Integrate Mapbox or Google Maps
  • Marker clustering for zoomed-out views
  • Route polylines with walking times
  • Tap-to-navigate functionality

6.2 Allergy Card Component

  • Printable layout
  • Large, clear text
  • Multiple languages on one card
  • QR code linking to detailed info

6.3 Transport Solver Dashboard

  • Countdown timer
  • Side-by-side comparison
  • One-tap booking links
  • Real-time price refresh

Implementation Priority Matrix

Phase Impact Effort Priority Dependencies
1.1 Location Services High Medium P0 None
1.2 User Preferences High Low P0 Supermemory
2.1 Travel Context Medium Low P1 Phase 1
2.2 Crisis Detection High Medium P1 None
3.1 Venue Geocoding High Medium P1 Phase 1.1
3.2 Venue Enrichment Medium Medium P2 Phase 1.1, 3.1
3.3 Dietary Features High Medium P1 Phase 1.2, 1.3
3.4 Smart Suggestions Medium High P2 Phase 3.1, 3.2
3.5 Map View High Medium P1 Phase 3.1, Frontend
4.x Transport Solver Medium High P3 Phase 1.1, 2.2
5.x Proactive Workflows Low Medium P3 Phase 1-3

API Cost Estimates

Service Cost Model Monthly Estimate
Google Geocoding $5/1000 requests $10-20
Google Places $17/1000 requests $20-40
Google Directions $5/1000 requests $5-10
Google Translate $20/1M characters $5-10
Amadeus Flights Free tier (2000/mo) $0
Parallel Search ~$0.01/query $5-10
Total $45-90/month

Success Metrics

Case 2 (Living Itinerary) Parity

  • Bulk venue import from text ✅ (exists)
  • Venues clustered by district on map
  • Smart gap-filling suggestions
  • Nearby venue recommendations with preferences

Case 3 (Crisis Averted) Parity

  • Crisis detection in natural language
  • Multi-modal transport comparison
  • Real-time pricing
  • One-tap booking

Case 4 (Hyper-Local Foodie) Parity

  • Dietary restrictions stored in profile
  • Venue filtering by dietary compatibility
  • Translated allergy cards
  • Real-time "is it open?" checks

Appendix A: File Structure

app/
├── services/
│   ├── location/
│   │   ├── __init__.py
│   │   ├── geocoder.py
│   │   ├── places_client.py
│   │   ├── distance_calculator.py
│   │   └── types.py
│   └── translation/
│       ├── __init__.py
│       └── translation_service.py
├── miniapps/
│   └── handlers/
│       ├── trip_planner.py          # Enhanced
│       └── transport_solver.py      # NEW
└── superpowers/
    └── catalog/
        └── proactive/
            ├── travel_support.py    # Enhanced
            └── destination_prep.py  # NEW

Appendix B: Quick Reference

When to Use Sage Core

  • One-shot questions ("Is X open?")
  • Quick translations
  • Weather checks
  • Simple recommendations
  • Crisis detection and empathy
  • Memory storage of preferences

When to Use Trip Planner

  • Multi-venue planning
  • Itinerary building
  • Visit tracking
  • Map visualization
  • Group trip coordination
  • Export and sharing

When to Use Transport Solver

  • Travel disruption
  • Multi-modal comparison
  • Time-critical decisions
  • Booking assistance

When to Use Superpowers

  • Proactive reminders
  • Background monitoring
  • Automated preparation
  • Scheduled notifications