Agent System Overview

class ResearchOrchestrator:
    """Two-phase orchestrator: static agents + dynamic URL scrapers."""
    
    def __init__(self, settings: Settings, *, pool_size: int = 10):
        self._settings = settings
        self._exa = ExaEnrichmentClient(settings)
        
        # Create shared inbox pool for login-wall bypass
        self._inbox_pool: InboxPool | None = None
        if settings.agentmail_api_key:
            mail_client = AgentMailClient(api_key=settings.agentmail_api_key)
            self._inbox_pool = InboxPool(mail_client, pool_size=pool_size)
        
        # Initialize static agents with shared inbox pool
        self._static_agents: list[BaseBrowserAgent] = [
            LinkedInAgent(settings, inbox_pool=self._inbox_pool),
            TwitterAgent(settings, inbox_pool=self._inbox_pool),
            InstagramAgent(settings, inbox_pool=self._inbox_pool),
            GoogleAgent(settings, inbox_pool=self._inbox_pool),
            OsintAgent(settings, inbox_pool=self._inbox_pool),
            SocialAgent(settings, inbox_pool=self._inbox_pool),
        ]

class DeepResearcher:
    """Multi-phase deep research pipeline that streams results.
    
    Phase 0: Exa + SixtyFour enrich-lead in parallel (~3s)
    Phase 1: Platform + OSINT skills in parallel (~20-35s, up to 15 concurrent)
    Phase 2: Deep URL extraction + SixtyFour deep-search + dark web (~30-60s)
    Phase 3: Verification loop — retry failed skills with account creation (~30-90s)
    """

async def research_person(self, request: ResearchRequest) -> OrchestratorResult:
    # Launch Exa + all static agents in parallel
    exa_task = asyncio.create_task(self._exa.enrich_person(exa_req), name="exa")
    
    tasks: dict[str, asyncio.Task] = {"exa": exa_task}
    for agent in self._static_agents:
        task = asyncio.create_task(agent.run(request), name=agent.agent_name)
        tasks[agent.agent_name] = task
    
    # Wait for all with the overall timeout
    done, pending = await asyncio.wait(
        tasks.values(),
        timeout=timeout,
    )
    
    # Cancel any stragglers
    for task in pending:
        task.cancel()

async def research(
    self,
    request: ResearchRequest,
) -> AsyncGenerator[AgentResult, None]:
    """Stream research results as they complete across all phases."""
    
    # Phase 0: Yield Exa results IMMEDIATELY (~1s)
    exa_urls, exa_snippets = await self._exa_pass(person, company, seen_urls)
    if exa_urls:
        yield AgentResult(
            agent_name="exa_deep",
            status=AgentStatus.SUCCESS,
            snippets=exa_snippets,
            urls_found=exa_urls,
        )
    
    # Phase 1: Stream skill results as they complete
    async for result in self._phase1(person, company, exa_urls, ...):
        yield result

class BaseBrowserAgent(ABC):
    """Abstract base class for browser-based research agents.
    
    Handles timeout enforcement, error isolation, structured logging,
    and Browser Use cloud session management with persistent auth.
    """
    
    agent_name: str = "base"
    
    @abstractmethod
    async def _run_task(self, request: ResearchRequest) -> AgentResult:
        """Subclass-specific research logic."""
        ...
    
    async def run(self, request: ResearchRequest) -> AgentResult:
        """Execute the agent with timeout and error isolation."""
        timeout = request.timeout_seconds or DEFAULT_TIMEOUT_SECONDS
        try:
            result = await asyncio.wait_for(
                self._run_task(request),
                timeout=timeout,
            )
            return result
        except TimeoutError:
            return AgentResult(
                agent_name=self.agent_name,
                status=AgentStatus.TIMEOUT,
                error=f"Agent timed out after {timeout}s",
            )

if settings.agentmail_api_key:
    mail_client = AgentMailClient(api_key=settings.agentmail_api_key)
    self._inbox_pool = InboxPool(mail_client, pool_size=pool_size)
    logger.info("orchestrator: inbox pool created, pool_size={}", pool_size)

MAX_CONCURRENT_SESSIONS = 25

class DeepResearcher:
    def __init__(self, settings: Settings):
        self._semaphore = asyncio.Semaphore(MAX_CONCURRENT_SESSIONS)
    
    async def _run_skill_with_semaphore(self, skill_name: str, task_str: str):
        async with self._semaphore:
            return await self._cloud.run_skill(skill_name, task_str, ...)

COVERED_DOMAINS = frozenset({
    "linkedin.com", "www.linkedin.com",
    "twitter.com", "x.com", "www.x.com",
    "instagram.com", "www.instagram.com",
    "google.com", "www.google.com",
    "youtube.com", "www.youtube.com",
    "facebook.com", "www.facebook.com",
})

class AgentResult(BaseModel):
    """Result from a single research agent run."""
    
    agent_name: str
    status: AgentStatus = AgentStatus.PENDING
    profiles: list[SocialProfile] = Field(default_factory=list)
    snippets: list[str] = Field(default_factory=list)
    urls_found: list[str] = Field(default_factory=list)
    error: str | None = None
    confidence: float = Field(default=1.0, ge=0.0, le=1.0)
    duration_seconds: float = 0.0

class SocialProfile(BaseModel):
    """A social media profile discovered by an agent."""
    
    platform: str
    url: str
    username: str | None = None
    display_name: str | None = None
    bio: str | None = None
    followers: int | None = None
    following: int | None = None
    location: str | None = None
    verified: bool = False
    raw_data: dict | None = None