feat: secure autonomous outbound workflow

.env.example

@@ -6,13 +6,22 @@ OPENAI_EMBEDDING_MODEL=text-embedding-3-small
 AI_PROVIDER=kimi-code
 KIMI_CODE_API_KEY=
 KIMI_CODE_BASE_URL=https://api.kimi.com/coding/v1
-KIMI_RESEARCH_MODELS=kimi-for-coding,k3,kimi-for-coding-highspeed
-KIMI_SYNTHESIS_MODELS=kimi-for-coding,kimi-for-coding-highspeed,k3
+KIMI_RESEARCH_MODELS=k3,k3-256k
+KIMI_SYNTHESIS_MODELS=k3-256k,k3
 SEARXNG_URL=http://searxng:8080
 SEARXNG_SECRET=replace-with-a-random-internal-secret
 UNIPILE_DSN=https://api37.unipile.com:16796
 UNIPILE_API_KEY=
 UNIPILE_LINKEDIN_ACCOUNT_ID=
+UNIPILE_WHATSAPP_ACCOUNT_ID=
+UNIPILE_WEBHOOK_SECRET=
+UNIPILE_CHAT_SYNC_ENABLED=true
+CALENDAR_WEBHOOK_SIGNING_KEY=
+PUBLIC_WEBHOOK_BASE_URL=http://localhost:3001
+OUTBOUND_LINKEDIN_DAILY_LIMIT=20
+OUTBOUND_EMAIL_DAILY_LIMIT=50
+OUTBOUND_WHATSAPP_DAILY_LIMIT=30
+BOOKING_URL=
 CRAWLER_SERVICE_URL=http://127.0.0.1:8000
 CRAWLER_API_KEY=
 SEARCH_FALLBACK_ENABLED=true
@@ -35,6 +44,8 @@ BOOTSTRAP_WORKSPACE_SLUG=ignition-ai
 BOOTSTRAP_WORKSPACE_NAME=IgnitionAI
 PORT=3001
 WORKER_ID=research-worker-1
+DAILY_PROSPECTING_TIME=06:00
+DAILY_PROSPECTING_TIMEZONE=Europe/Paris
 JOB_LEASE_MS=60000
 JOB_BATCH_SIZE=1
 JOB_POLL_INTERVAL_MS=1000

.gitignore

@@ -6,6 +6,10 @@ node_modules/
 .next/
 dist/
 coverage/
+__pycache__/
+.pytest_cache/
+.venv/
+*.py[cod]
 *.tsbuildinfo
 *.log
 .env

apps/api/src/index.ts

@@ -9,6 +9,7 @@ import { createResearchDocumentHttpHandler } from "@outbound/interface/http/rese
 import { createCrmHttpHandler } from "@outbound/interface/http/crm-handler";
 import { createDiscoveryHttpHandler } from "@outbound/interface/http/discovery-handler";
 import { createSequenceHttpHandler } from "@outbound/interface/http/sequence-handler";
+import { createCampaignHttpHandler } from "@outbound/interface/http/campaign-handler";
 import {
   ProviderUnavailableError,
   UnipileProspectSource,
@@ -19,6 +20,19 @@ import { WorkspaceAiSettingsApplication } from "@outbound/application/workspaces
 import { PostgresWorkspaceAiSettingsRepository } from "@outbound/infrastructure/workspaces/postgres-workspace-ai-settings-repository";
 import { createWorkspaceAiSettingsHttpHandler } from "@outbound/interface/http/workspace-ai-settings-handler";
 import { resolveResearchModelPolicyFromEnvironment } from "@outbound/infrastructure/ai/langchain-research-agent-executor";
+import { CrawlerClient } from "@outbound/infrastructure/ai/crawler-client";
+import { CrawlerProspectEnricher } from "@outbound/infrastructure/crm/crawler-prospect-enricher";
+import { UnipileWebhookIngestor } from "@outbound/infrastructure/campaigns/unipile-webhook-ingestor";
+import { createUnipileWebhookHttpHandler } from "@outbound/interface/http/unipile-webhook-handler";
+import { PostgresCalendarIntegration } from "@outbound/infrastructure/calendar/postgres-calendar-integration";
+import { createCalendarConnectionHttpHandler } from "@outbound/interface/http/calendar-connection-handler";
+import { createCalendarWebhookHttpHandler } from "@outbound/interface/http/calendar-webhook-handler";
+import { PostgresOpportunityRepository } from "@outbound/infrastructure/pipeline/postgres-opportunity-repository";
+import { createOpportunityHttpHandler } from "@outbound/interface/http/opportunity-handler";
+import { LangChainConversationDraftImprover } from "@outbound/infrastructure/campaigns/langchain-conversation-draft-improver";
+import { PostgresUnipileChannelConnections } from "@outbound/infrastructure/channels/postgres-unipile-channel-connections";
+import { createChannelConnectionHttpHandler } from "@outbound/interface/http/channel-connection-handler";
+import { PostgresChannelCapabilityReassessment } from "@outbound/infrastructure/campaigns/channel-capability-reassessment";
 
 const databaseUrl = requiredEnvironment("DATABASE_URL");
 const database = createDatabase(databaseUrl);
@@ -74,10 +88,27 @@ const crm = createCrmHttpHandler({
 });
 const unipileDsn = process.env.UNIPILE_DSN ?? "";
 const unipileApiKey = process.env.UNIPILE_API_KEY ?? "";
+const unipileChannelConnections = unipileDsn && unipileApiKey
+  ? new PostgresUnipileChannelConnections(database.db, { dsn: unipileDsn, apiKey: unipileApiKey })
+  : null;
+const channelConnection = createChannelConnectionHttpHandler({
+  connections: unipileChannelConnections,
+  contextResolver: auth.contextResolver,
+  reassessment: new PostgresChannelCapabilityReassessment(database.db),
+});
+const discoveryCrawler =
+  process.env.CRAWLER_SERVICE_URL && process.env.CRAWLER_API_KEY
+    ? new CrawlerClient({
+        baseUrl: process.env.CRAWLER_SERVICE_URL,
+        apiKey: process.env.CRAWLER_API_KEY,
+        maxConcurrentPageReads: 2,
+      })
+    : null;
 const discovery = createDiscoveryHttpHandler({
   database: database.db,
   contextResolver: auth.contextResolver,
-  prospectSource: () => {
+  jobQueue: queue,
+  prospectSource: (workspaceId) => {
     if (!unipileDsn || !unipileApiKey) {
       return {
         async searchPeople() {
@@ -94,24 +125,67 @@ const discovery = createDiscoveryHttpHandler({
       ...(process.env.UNIPILE_LINKEDIN_ACCOUNT_ID
         ? { accountId: process.env.UNIPILE_LINKEDIN_ACCOUNT_ID }
         : {}),
+      ...(process.env.UNIPILE_WHATSAPP_ACCOUNT_ID
+        ? { whatsappAccountId: process.env.UNIPILE_WHATSAPP_ACCOUNT_ID }
+        : {}),
+      ...(unipileChannelConnections
+        ? { resolveWhatsappAccountId: () => unipileChannelConnections.selectedAccountId(workspaceId, "whatsapp") }
+        : {}),
     });
   },
+  prospectEnricher: () =>
+    discoveryCrawler ? new CrawlerProspectEnricher(discoveryCrawler) : null,
 });
 const sequenceHandler = createSequenceHttpHandler({
   database: database.db,
   contextResolver: auth.contextResolver,
 });
+const campaignHandler = createCampaignHttpHandler({
+  database: database.db,
+  contextResolver: auth.contextResolver,
+  jobQueue: queue,
+  draftImprover: new LangChainConversationDraftImprover(
+    database.db,
+    process.env,
+    workspaceAiSettingsRepository,
+  ),
+});
+const unipileWebhook = createUnipileWebhookHttpHandler({
+  ingestor: new UnipileWebhookIngestor(database.db),
+  secret: process.env.UNIPILE_WEBHOOK_SECRET ?? "",
+});
+const calendarSigningKey = process.env.CALENDAR_WEBHOOK_SIGNING_KEY
+  ?? requiredSecretEnvironment("BETTER_AUTH_SECRET");
+const calendarIntegration = new PostgresCalendarIntegration(database.db, calendarSigningKey);
+const calendarConnection = createCalendarConnectionHttpHandler({
+  integration: calendarIntegration,
+  contextResolver: auth.contextResolver,
+  publicWebhookBaseUrl: process.env.PUBLIC_WEBHOOK_BASE_URL ?? requiredEnvironment("BETTER_AUTH_URL"),
+});
+const calendarWebhook = createCalendarWebhookHttpHandler({
+  integration: calendarIntegration,
+  signingKey: calendarSigningKey,
+});
+const opportunityHandler = createOpportunityHttpHandler({
+  repository: new PostgresOpportunityRepository(database.db),
+  contextResolver: auth.contextResolver,
+});
 const port = positiveIntegerEnvironment("PORT", 3000);
 const server = Bun.serve({
   port,
   maxRequestBodySize: 1_048_576,
   async fetch(request) {
     const pathname = new URL(request.url).pathname;
     if (pathname.startsWith("/api/auth/")) return auth.handle(request);
+    if (pathname === "/api/v1/webhooks/unipile") return unipileWebhook(request);
+    if (pathname.startsWith("/api/v1/webhooks/calendar/")) return calendarWebhook(request);
+    if (pathname === "/api/v1/calendar-connection") return calendarConnection(request);
+    if (pathname.startsWith("/api/v1/channel-connections/")) return channelConnection(request);
+    if (pathname.startsWith("/api/v1/opportunities")) return opportunityHandler(request);
     if (pathname === "/api/v1/workspaces") return workspace(request);
     if (pathname === "/api/v1/workspace-ai-settings") return workspaceAiSettings(request);
     if (pathname.startsWith("/api/v1/research-documents")) return documents(request);
-    if (pathname.startsWith("/api/v1/companies") || pathname.startsWith("/api/v1/contacts")) {
+    if (pathname.startsWith("/api/v1/companies") || pathname.startsWith("/api/v1/contacts") || pathname.startsWith("/api/v1/prospects")) {
       return crm(request);
     }
     if (pathname.startsWith("/api/v1/icp-versions") || pathname.startsWith("/api/v1/discovery-runs")) {
@@ -120,6 +194,14 @@ const server = Bun.serve({
     if (pathname.startsWith("/api/v1/sequences")) {
       return sequenceHandler(request);
     }
+    if (
+      pathname.startsWith("/api/v1/campaigns") ||
+      pathname.startsWith("/api/v1/prospecting-plans") ||
+      pathname.startsWith("/api/v1/channel-assessments")
+      || pathname.startsWith("/api/v1/conversations")
+    ) {
+      return campaignHandler(request);
+    }
     if (pathname === "/health/live") return Response.json({ status: "ok" });
     if (pathname === "/health/ready") {
       try {

apps/crawler/src/crawler_service/api/routes.py

@@ -296,6 +296,15 @@ async def crawl_selected_pages(request: CrawlPagesRequest):
     Returns a job ID for tracking progress.
     """
     urls = [str(u) for u in request.urls]
+    if request.idempotencyKey:
+        existing = job_manager.get_job_by_idempotency_key(request.idempotencyKey)
+        if existing:
+            return CrawlPagesStartResponse(
+                success=True,
+                id=existing.id,
+                urlCount=len(urls),
+                message="Existing idempotent crawl job returned",
+            )
 
     # Acquire a concurrency slot BEFORE creating anything — same rule as
     # /crawl: no slot, no job, and release_slot() is only called for slots
@@ -323,6 +332,7 @@ async def crawl_selected_pages(request: CrawlPagesRequest):
         include_images=request.includeImages,
         exclude_patterns=[],
         include_patterns=[],
+        idempotency_key=request.idempotencyKey,
     )
 
     # Start the job

apps/crawler/src/crawler_service/api/schemas.py

@@ -70,6 +70,7 @@ class CrawlRequest(BaseModel):
         description="URL patterns to include (regex)",
     )
     correlationId: str | None = Field(default=None, max_length=200)
+    idempotencyKey: str | None = Field(default=None, min_length=8, max_length=500)
 
     _validate_patterns = field_validator(
         "excludePatterns",
@@ -197,6 +198,7 @@ class CrawlPagesRequest(BaseModel):
         description="Extract image URLs from pages",
     )
     correlationId: str | None = Field(default=None, max_length=200)
+    idempotencyKey: str | None = Field(default=None, min_length=8, max_length=500)
 
 
 class CrawlPagesStartResponse(BaseModel):

apps/crawler/src/crawler_service/core/job_manager.py

@@ -67,6 +67,7 @@ class JobManager:
 
     def __init__(self):
         self._jobs: dict[str, CrawlJob] = {}
+        self._idempotency_keys: dict[str, str] = {}
         self._semaphore = asyncio.Semaphore(settings.max_concurrent_crawls)
         self._lock = asyncio.Lock()
 
@@ -86,8 +87,14 @@ def create_job(
         include_images: bool = True,
         exclude_patterns: list[str] | None = None,
         include_patterns: list[str] | None = None,
+        idempotency_key: str | None = None,
     ) -> CrawlJob:
         """Create a new crawl job."""
+        if idempotency_key:
+            existing_id = self._idempotency_keys.get(idempotency_key)
+            existing = self._jobs.get(existing_id) if existing_id else None
+            if existing:
+                return existing
         job_id = str(uuid.uuid4())
         job = CrawlJob(
             id=job_id,
@@ -101,8 +108,15 @@ def create_job(
             event_queue=asyncio.Queue(),
         )
         self._jobs[job_id] = job
+        if idempotency_key:
+            self._idempotency_keys[idempotency_key] = job_id
         return job
 
+    def get_job_by_idempotency_key(self, key: str) -> CrawlJob | None:
+        """Return the current in-memory job for an idempotent request."""
+        job_id = self._idempotency_keys.get(key)
+        return self._jobs.get(job_id) if job_id else None
+
     def get_job(self, job_id: str) -> CrawlJob | None:
         """Get a job by ID."""
         return self._jobs.get(job_id)

apps/crawler/src/crawler_service/core/request_safety.py

@@ -32,21 +32,81 @@ def collected_at() -> str:
 async def install_safe_request_interceptor(page, **_kwargs):
     """Abort every browser request whose target is not publicly routable."""
 
+    context = getattr(page, "context", None)
+    new_cdp_session = getattr(context, "new_cdp_session", None)
+    if callable(new_cdp_session):
+        try:
+            client = await new_cdp_session(page)
+
+            async def guard_cdp(event):
+                request_id = event["requestId"]
+                target = event["request"]["url"]
+                scheme = urlparse(target).scheme
+                if scheme in ("data", "blob", "about") or await is_url_allowed_async(target):
+                    await client.send(
+                        "Fetch.continueRequest",
+                        {"requestId": request_id},
+                    )
+                else:
+                    await client.send(
+                        "Fetch.failRequest",
+                        {
+                            "requestId": request_id,
+                            "errorReason": "BlockedByClient",
+                        },
+                    )
+
+            client.on("Fetch.requestPaused", guard_cdp)
+            await client.send(
+                "Fetch.enable",
+                {
+                    "patterns": [
+                        {"urlPattern": "*", "requestStage": "Request"},
+                    ]
+                },
+            )
+            # Keep the CDP session alive for the page lifetime.
+            setattr(page, "_ignition_safe_cdp_session", client)
+            return page
+        except Exception:
+            # Non-Chromium adapters and test doubles use Playwright routing.
+            pass
+
     async def guard(route, request):
         target = request.url
         scheme = urlparse(target).scheme
         if scheme in ("data", "blob", "about"):
-            await route.continue_()
+            await _continue_safely(route)
             return
         if await is_url_allowed_async(target):
-            await route.continue_()
+            await _continue_safely(route)
         else:
             await route.abort("blockedbyclient")
 
-    await page.route("**/*", guard)
+    # Browser-context routing also sees redirected requests created by a page
+    # route fulfillment. Page-level routing alone can miss that transition.
+    if context is not None and hasattr(context, "route"):
+        await context.route("**/*", guard)
+    else:
+        await page.route("**/*", guard)
     return page
 
 
+async def _continue_safely(route):
+    """Continue through any other route handlers before reaching the network.
+
+    Playwright's ``continue_`` bypasses older handlers. ``fallback`` preserves
+    the interception chain and is therefore required when another adapter
+    fulfills a public response that redirects or embeds a private target.
+    """
+
+    fallback = getattr(route, "fallback", None)
+    if fallback is not None:
+        await fallback()
+    else:
+        await route.continue_()
+
+
 def configure_safe_crawler(crawler) -> None:
     crawler.crawler_strategy.set_hook(
         "on_page_context_created",

apps/crawler/tests/test_api.py

@@ -73,3 +73,36 @@ async def fake_execute_crawl(job):
     # Cancel the job to clean up
     job_id = data["id"]
     client.delete(f"/crawl/{job_id}")
+
+
+@pytest.mark.asyncio
+async def test_selective_crawl_reuses_idempotency_key(client: TestClient, monkeypatch):
+    """The Bun orchestrator can safely replay a lost selective crawl request."""
+
+    async def fake_execute_selective_crawl(job, urls):
+        await asyncio.sleep(0.05)
+
+    monkeypatch.setattr(
+        "crawler_service.api.routes.execute_selective_crawl",
+        fake_execute_selective_crawl,
+    )
+    payload = {
+        "urls": ["https://example.com/a"],
+        "includeImages": False,
+        "idempotencyKey": "run-stage-page-example-a",
+    }
+    first = client.post("/crawl/pages", json=payload)
+    second = client.post("/crawl/pages", json=payload)
+    different = client.post(
+        "/crawl/pages",
+        json={**payload, "idempotencyKey": "run-stage-page-example-b"},
+    )
+
+    assert first.status_code == 200
+    assert second.status_code == 200
+    assert different.status_code == 200
+    assert first.json()["id"] == second.json()["id"]
+    assert different.json()["id"] != first.json()["id"]
+
+    client.delete(f"/crawl/{first.json()['id']}")
+    client.delete(f"/crawl/{different.json()['id']}")