feat(knowledge): 任务队列、独立 worker 与 pgvector 向量索引

知识处理改为数据库任务队列，独立 worker 做切块与向量化，失败可重试。

backend/app/workers/chat_auto_closer.py

@@ -0,0 +1,101 @@
+"""
+Copyright 2024-2026 ChatterMate
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import asyncio
+import sys
+import os
+from pathlib import Path
+
+# Add the parent directory to Python path to allow imports
+sys.path.append(str(Path(__file__).parent.parent.parent))
+
+from app.database import SessionLocal
+from app.repositories.session_to_agent import SessionToAgentRepository
+from app.core.logger import get_logger
+
+logger = get_logger(__name__)
+
+# Configure check interval (default to 24 hours - once a day)
+CHECK_INTERVAL = int(os.environ.get('CHAT_AUTO_CLOSE_INTERVAL', '86400'))  # 24 hours in seconds
+WORKER_ID = os.environ.get('GUNICORN_WORKER_ID', 'standalone')
+
+async def run_auto_closer():
+    """Single run of the auto closer"""
+    db = None
+    try:
+        logger.info(f"Worker {WORKER_ID}: Running chat auto-closer")
+        
+        db = SessionLocal()
+        session_repo = SessionToAgentRepository(db)
+        
+        # Auto-close inactive agent chats
+        closed_count = session_repo.auto_close_inactive_agent_chats()
+        
+        if closed_count > 0:
+            logger.info(f"Worker {WORKER_ID}: Auto-closed {closed_count} inactive agent chats")
+        else:
+            logger.info(f"Worker {WORKER_ID}: No inactive agent chats found to close")
+            
+    except Exception as e:
+        logger.error(f"Worker {WORKER_ID}: Error in chat auto-closer: {str(e)}")
+        raise
+    finally:
+        if db:
+            db.close()
+
+async def run_auto_closer_loop():
+    """Run the chat auto-closer in a continuous loop"""
+    logger.info(f"Worker {WORKER_ID}: Starting chat auto-closer with {CHECK_INTERVAL}s check interval (once per day)")
+    
+    try:
+        iteration = 0
+        
+        while True:
+            iteration += 1
+            start_time = asyncio.get_event_loop().time()
+            
+            try:
+                logger.info(f"Worker {WORKER_ID}: Iteration {iteration} - Checking for inactive agent chats...")
+                await run_auto_closer()
+                logger.info(f"Worker {WORKER_ID}: Iteration {iteration} - Auto-closer completed successfully")
+            except Exception as e:
+                logger.error(f"Worker {WORKER_ID}: Iteration {iteration} - Error in auto-closer: {str(e)}")
+            
+            # Calculate how long the processing took
+            elapsed = asyncio.get_event_loop().time() - start_time
+            logger.info(f"Worker {WORKER_ID}: Iteration {iteration} - Processing took {elapsed:.2f} seconds")
+            
+            # Sleep for the configured interval (24 hours)
+            logger.info(f"Worker {WORKER_ID}: Iteration {iteration} - Sleeping for {CHECK_INTERVAL} seconds (24 hours) before next check")
+            await asyncio.sleep(CHECK_INTERVAL)
+            
+    except Exception as e:
+        logger.error(f"Worker {WORKER_ID}: Fatal error in chat auto-closer: {str(e)}")
+        sys.exit(1)
+
+def app(environ, start_response):
+    """WSGI application for health checks"""
+    status = '200 OK'
+    headers = [('Content-type', 'text/plain')]
+    start_response(status, headers)
+    return [b'Chat Auto-Closer Worker is running']
+
+# Main entry point for running as a standalone service
+if __name__ == "__main__":
+    logger.info("Starting chat auto-closer service")
+    
+    # Run the auto-closer loop
+    asyncio.run(run_auto_closer_loop())

backend/app/workers/crm_sync.py

@@ -0,0 +1,141 @@
+"""
+Copyright 2024-2026 ChatterMate
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+CRM lead-push worker.
+
+Claims due crm_sync_jobs (FOR UPDATE SKIP LOCKED — replica-safe), groups them
+by (provider, organization) and pushes each group serially so one tenant's
+burst can't trip provider rate limits, while distinct tenants run
+concurrently. A second, slower timer proactively refreshes Pipedrive tokens:
+their refresh tokens die 60 days after last use, so idle tenants must be kept
+alive by us, not by traffic.
+
+Run as its own container: python -m app.workers.crm_sync
+"""
+
+import asyncio
+from collections import defaultdict
+from datetime import datetime, timedelta, timezone
+from typing import List
+
+from app.core.logger import get_logger
+from app.crm.base import CrmAuthError, CrmTransientError
+from app.database import SessionLocal
+from app.models.crm import CrmProvider, CrmSyncJob
+from app.repositories.crm import CrmConnectionRepository, CrmSyncJobRepository
+from app.services.crm_sync import process_job
+from app.services.crm_tokens import get_valid_tokens
+
+logger = get_logger(__name__)
+
+POLL_INTERVAL_SECONDS = 10
+CLAIM_BATCH_SIZE = 20
+REFRESH_SWEEP_INTERVAL_SECONDS = 6 * 3600
+# Refresh anything not refreshed within a day — an idle tenant then has ~59
+# failed sweeps of headroom before its 60-day window could actually lapse.
+REFRESH_STALE_AFTER_HOURS = 24
+
+
+async def run_sync_pass() -> None:
+    """One poll: reclaim crashed claims, claim due jobs, push them."""
+    with SessionLocal() as db:
+        repo = CrmSyncJobRepository(db)
+        repo.reclaim_stale()
+        jobs = repo.claim_batch(CLAIM_BATCH_SIZE)
+    if not jobs:
+        return
+
+    groups = defaultdict(list)
+    for job in jobs:
+        groups[(job.provider, job.organization_id)].append(job)
+    await asyncio.gather(*(
+        _process_group([job.id for job in group]) for group in groups.values()
+    ))
+
+
+async def _process_group(job_ids: List) -> None:
+    """Serial pushes for one (provider, org) — its own session, so groups
+    running concurrently never share SQLAlchemy state. When the connection
+    proves dead mid-group, the rest fail fast instead of re-hitting it."""
+    with SessionLocal() as db:
+        repo = CrmSyncJobRepository(db)
+        connection_dead = False
+        for job_id in job_ids:
+            job = db.get(CrmSyncJob, job_id)
+            if job is None:
+                continue
+            if connection_dead:
+                repo.fail(job, "auth: connection expired (failed earlier in batch)")
+                continue
+            try:
+                job = await process_job(db, job)
+            except Exception as e:
+                logger.error(f"CRM sync job {job_id} crashed: {e}")
+                db.rollback()  # discard any half-done state before the retry write
+                repo.schedule_retry(job, f"worker error: {e}")
+                continue
+            if (job.last_error or "").startswith("auth:"):
+                connection_dead = True
+
+
+async def run_refresh_sweep() -> None:
+    """Keep Pipedrive's sliding refresh-token window alive for idle tenants."""
+    cutoff = datetime.now(timezone.utc) - timedelta(hours=REFRESH_STALE_AFTER_HOURS)
+    with SessionLocal() as db:
+        connections = CrmConnectionRepository(db).list_needing_refresh(
+            CrmProvider.PIPEDRIVE.value, cutoff)
+        for connection in connections:
+            try:
+                # No force_refresh: idle connections already have an expired
+                # access token, so get_valid_tokens refreshes them, and the
+                # per-connection lock + runway re-check inside it prevents a
+                # redundant second refresh if another worker just did it.
+                await get_valid_tokens(db, connection)
+                logger.info(f"Refreshed idle Pipedrive connection {connection.id}")
+            except CrmAuthError:
+                pass  # marked expired inside get_valid_tokens; UI shows Reconnect
+            except CrmTransientError as e:
+                logger.warning(f"Pipedrive sweep refresh failed (will retry next "
+                               f"sweep): {e}")
+
+
+async def run_sync_loop(poll_interval: int = POLL_INTERVAL_SECONDS) -> None:
+    while True:
+        try:
+            await run_sync_pass()
+        except Exception as e:
+            logger.error(f"Error in CRM sync loop: {e}")
+        await asyncio.sleep(poll_interval)
+
+
+async def run_refresh_loop(interval: int = REFRESH_SWEEP_INTERVAL_SECONDS) -> None:
+    while True:
+        try:
+            await run_refresh_sweep()
+        except Exception as e:
+            logger.error(f"Error in CRM refresh sweep: {e}")
+        await asyncio.sleep(interval)
+
+
+async def main() -> None:
+    await asyncio.gather(
+        run_sync_loop(),
+        run_refresh_loop(),
+    )
+
+
+if __name__ == "__main__":
+    logger.info("Starting CRM sync worker")
+    asyncio.run(main())

backend/app/workers/faq_processor.py

@@ -0,0 +1,166 @@
+"""
+Copyright 2024-2026 ChatterMate
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+FAQ generation/import worker. Runs as an independent loop inside the
+knowledge_processor container; `python -m app.workers.faq_processor` also
+works as a standalone service if it ever needs its own container.
+"""
+
+import asyncio
+
+from app.database import SessionLocal
+from app.core.config import settings
+from app.core.logger import get_logger
+from app.models.faq_generation_job import FAQJobType
+from app.models.notification import NotificationType
+from app.repositories.faq_generation_job import FAQGenerationJobRepository
+from app.services.faq_article_import import run_article_import_job
+from app.services.faq_generation import run_generation_job
+from app.services.faq_import import run_import_job, run_pdf_import_job
+from app.services.notifications import notify_user
+
+logger = get_logger(__name__)
+
+POLL_INTERVAL_SECONDS = 60
+
+
+async def process_faq_job(job_id: int):
+    """Run one job to a terminal state, with notifications."""
+    with SessionLocal() as db:
+        job_repo = FAQGenerationJobRepository(db)
+        job = job_repo.get_by_id(job_id)
+        if not job:
+            logger.error(f"FAQ job {job_id} not found")
+            return
+        # Captured before the try: after a DB failure + rollback the ORM
+        # instance is expired and attribute access could raise or re-query.
+        job_type = job.job_type
+        is_import = job_type in (
+            FAQJobType.IMPORT_URL.value,
+            FAQJobType.IMPORT_ARTICLES.value,
+            FAQJobType.IMPORT_PDF.value,
+        )
+        try:
+            job_repo.mark_processing(job.id)
+            if job_type == FAQJobType.IMPORT_URL.value:
+                created = await run_import_job(db, job)
+            elif job_type == FAQJobType.IMPORT_ARTICLES.value:
+                created = await run_article_import_job(db, job)
+            elif job_type == FAQJobType.IMPORT_PDF.value:
+                created = await run_pdf_import_job(db, job)
+            else:
+                created = await run_generation_job(db, job)
+            job_repo.mark_completed(job.id, faqs_created=created)
+            if created:
+                body = f"{created} draft FAQ{'s' if created != 1 else ''} added — review and publish them in Help center."
+            elif is_import:
+                body = (
+                    "Nothing new was imported — no Q&A was found on the page, or it duplicates "
+                    "existing FAQs. Tip: use Import → Articles to bring in whole articles verbatim (no AI needed)."
+                )
+            else:
+                body = (
+                    "No new FAQs were generated — the model returned nothing for this content. "
+                    "Try a larger/different model, or Import → Articles to import pages directly."
+                )
+            await notify_user(
+                db,
+                job.user_id,
+                NotificationType.FAQ_GENERATED,
+                "Imported drafts ready for review" if is_import and created else
+                ("Import finished" if is_import else "Draft FAQs ready for review"),
+                body,
+                metadata={"faq_job_id": job.id},
+            )
+        except Exception as e:
+            logger.error(f"FAQ job {job_id} failed: {e}")
+            # A DB-originated failure leaves the session aborted; roll back so
+            # mark_failed and the notification can still be persisted.
+            try:
+                db.rollback()
+            except Exception:
+                pass
+            job_repo.mark_failed(job.id, str(e))
+            await notify_user(
+                db,
+                job.user_id,
+                NotificationType.FAQ_GENERATION_FAILED,
+                "FAQ import failed" if is_import else "FAQ generation failed",
+                str(e),
+                metadata={"faq_job_id": job.id},
+            )
+
+
+_reaped_orphans = False
+
+
+def _reap_orphans_once() -> None:
+    """Reap orphaned `processing` jobs the first time the processor runs in
+    this process — covers both worker entrypoints (the loop wrapper here and
+    run_knowledge_processor's per-iteration call) without reaping every tick."""
+    global _reaped_orphans
+    if _reaped_orphans:
+        return
+    _reaped_orphans = True
+    reap_orphaned_jobs()
+
+
+async def run_faq_processor():
+    """Single pass: process every pending FAQ job (bounded concurrency)."""
+    _reap_orphans_once()
+    with SessionLocal() as db:
+        pending_ids = [job.id for job in FAQGenerationJobRepository(db).get_pending()]
+    if not pending_ids:
+        return
+    semaphore = asyncio.Semaphore(max(1, settings.MAX_CONCURRENT_FAQ_JOBS))
+
+    async def process_with_semaphore(job_id: int):
+        async with semaphore:
+            await process_faq_job(job_id)
+
+    await asyncio.gather(*[process_with_semaphore(job_id) for job_id in pending_ids])
+
+
+def reap_orphaned_jobs() -> None:
+    """Fail jobs left in `processing` by a previous crash/kill. Runs once at
+    startup: this worker is the only writer of `processing`, so on a fresh boot
+    any such row is orphaned — otherwise it hangs the enqueue guard (one active
+    job per type) and the UI's active-job polling forever."""
+    try:
+        with SessionLocal() as db:
+            reaped = FAQGenerationJobRepository(db).fail_orphaned_processing(
+                "Worker restarted before the job finished — please try again."
+            )
+        if reaped:
+            logger.warning(f"Reaped {reaped} orphaned FAQ job(s) left in 'processing' on startup")
+    except Exception as e:
+        logger.error(f"FAQ orphan reap failed (non-fatal): {e}")
+
+
+async def run_faq_processor_loop(poll_interval: int = POLL_INTERVAL_SECONDS):
+    """Forever-loop wrapper — run as an independent task next to the knowledge
+    loop so a long FAQ job never delays knowledge ingestion. run_faq_processor
+    reaps orphans on its first call, so no explicit reap is needed here."""
+    while True:
+        try:
+            await run_faq_processor()
+        except Exception as e:
+            logger.error(f"Error in FAQ processor loop: {e}")
+        await asyncio.sleep(poll_interval)
+
+
+if __name__ == "__main__":
+    logger.info("Starting FAQ processor service")
+    asyncio.run(run_faq_processor_loop())

backend/app/workers/knowledge_processor.py

@@ -0,0 +1,334 @@
+"""
+Copyright 2024-2026 ChatterMate
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+import asyncio
+from app.database import SessionLocal
+from app.repositories.knowledge_queue import KnowledgeQueueRepository
+from app.repositories.knowledge import KnowledgeRepository
+from app.knowledge.knowledge_base import KnowledgeManager
+from app.knowledge import page_editor
+from app.services import knowledge_vector_links
+from app.models.knowledge_queue import QueueStatus, ProcessingStage
+from app.core.logger import get_logger
+import os
+from app.core.config import settings
+from app.core.processor import PROCESSOR_STATUS
+from app.knowledge.failure_advice import FAILURE_ADVICE, FailureCause
+from datetime import datetime, timedelta, timezone
+from app.models.notification import Notification, NotificationType
+from app.services.user import send_fcm_notification
+from urllib.parse import urlparse, unquote
+
+logger = get_logger(__name__)
+
+
+def get_user_friendly_filename(source: str, source_type: str) -> str:
+    """Extract a user-friendly filename from the source URL or path"""
+    try:
+        if source_type in ('website', 'sitemap'):
+            # For websites/sitemaps, just return the domain
+            parsed = urlparse(source)
+            return parsed.netloc or source
+        
+        # For files (PDF, etc.), extract filename from URL or path
+        if source.startswith('http'):
+            # Parse URL to get the path
+            parsed = urlparse(source)
+            path = parsed.path
+            
+            # Extract filename from path
+            if path:
+                # Get the last part of the path (filename)
+                filename = path.split('/')[-1]
+                # URL decode the filename to handle encoded characters like %20
+                filename = unquote(filename)
+                # Remove query parameters if any
+                filename = filename.split('?')[0]
+                if filename:
+                    return filename
+            
+            # Fallback to domain if no filename found
+            return parsed.netloc or source
+        else:
+            # For local file paths, just get the basename
+            return os.path.basename(source)
+            
+    except Exception as e:
+        logger.warning(f"Error extracting filename from {source}: {e}")
+        return source
+
+
+def _sync_vector_agent_links(db, queue_item) -> None:
+    """Point the ingested chunks at every agent the source is linked to.
+
+    Ingestion writes ``filters.agent_id`` from the queue row's single optional
+    ``agent_id``, so a run queued without one leaves every chunk with an empty
+    list. Retrieval matches on that filter, so the source is invisible to the
+    agent while the dashboard still shows it as linked — the failure that left
+    two production sources unreadable until the links were rebuilt by hand.
+
+    The relational links are the authority, so re-derive from them. Idempotent,
+    and non-fatal: the content is already ingested, and losing the run over a
+    filter refresh would be worse than the stale filter.
+    """
+    try:
+        knowledge_repo = KnowledgeRepository(db)
+        sources = knowledge_repo.get_by_sources(
+            queue_item.organization_id, [queue_item.source]
+        )
+        if not sources:
+            return
+
+        # An org can hold more than one knowledge row for the same URL, and they
+        # share one set of vector chunks (keyed on the source name). Sync the
+        # union of their links once, rather than per row where the last write
+        # would drop the other rows' agents.
+        agent_ids = sorted({
+            agent_id
+            for knowledge in sources
+            for agent_id in page_editor.agent_ids_for(knowledge)
+        })
+        knowledge_vector_links.sync_agent_ids(db, sources[0], agent_ids)
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        logger.error(
+            f"Could not refresh vector agent filters for {queue_item.source} "
+            f"(non-fatal, content was ingested): {e}"
+        )
+
+
+async def _notify(db, queue_item, kind: NotificationType, title: str, message: str) -> None:
+    """Tell whoever added the source, in the app and by push. Never fatal: the
+    source's own status must not depend on it. A source added without a user
+    (a Shopify install queues the shop's site) has no one to tell."""
+    if not queue_item.user_id:
+        logger.info(f"Queue item {queue_item.id} has no user to notify ({kind.value})")
+        return
+    try:
+        notification = Notification(
+            user_id=queue_item.user_id,
+            type=kind,
+            title=title,
+            message=message,
+            # The column is notification_metadata; the old metadata=
+            # kwarg silently shadowed Base.metadata and stored nothing.
+            notification_metadata={"queue_id": queue_item.id}
+        )
+        db.add(notification)
+        db.commit()
+    except Exception as e:
+        db.rollback()
+        logger.error(f"Could not create {kind.value} notification for queue item {queue_item.id}: {e}")
+        return
+    await send_fcm_notification(queue_item.user_id, notification, db)
+
+
+async def process_queue_item(queue_item_id: int):
+    """Process a single queue item"""
+    with SessionLocal() as db:
+        queue_item = None
+        try:
+            queue_repo = KnowledgeQueueRepository(db)
+            queue_item = queue_repo.get_by_id(queue_item_id)
+
+            if not queue_item:
+                logger.error(f"Queue item {queue_item_id} not found")
+                return
+
+            # Get knowledge manager instance
+            knowledge = KnowledgeManager(
+                org_id=queue_item.organization_id,
+                agent_id=queue_item.agent_id
+            )
+
+            # Process if status is pending
+            if queue_item.status == QueueStatus.PENDING:
+                # Update to processing
+                queue_item.status = QueueStatus.PROCESSING
+                queue_item.processing_stage = ProcessingStage.NOT_STARTED
+                queue_item.progress_percentage = 0.0
+                db.commit()
+
+                await knowledge.process_knowledge(queue_item)
+
+                _sync_vector_agent_links(db, queue_item)
+
+                user_friendly_name = get_user_friendly_filename(queue_item.source, queue_item.source_type)
+                await _notify(db, queue_item, NotificationType.KNOWLEDGE_PROCESSED,
+                              "Knowledge Processing Complete", f"Successfully processed {user_friendly_name}")
+
+                # Auto-draft FAQs from the new source for orgs using the help
+                # center. Fully non-fatal — even an import failure of the FAQ
+                # stack must not flip an already-successful run to FAILED.
+                try:
+                    from app.services.faq_generation import maybe_enqueue_auto_faq_job
+                    maybe_enqueue_auto_faq_job(db, queue_item)
+                except Exception as faq_hook_err:
+                    logger.error(f"Auto FAQ hook unavailable (non-fatal): {faq_hook_err}")
+
+            queue_item.status = QueueStatus.COMPLETED
+            db.commit()
+
+        except Exception as e:
+            logger.error(f"Error processing queue item {queue_item_id}: {str(e)}")
+            if queue_item is None:
+                raise
+            # The session may hold a failed flush: start clean, record the
+            # failure first, and only then tell anyone about it.
+            db.rollback()
+            try:
+                queue_item.status = QueueStatus.FAILED
+                # Persist the reason so the UI can show why it failed.
+                queue_item.error = str(e)
+                db.commit()
+            except Exception as status_err:
+                db.rollback()
+                logger.error(f"Could not mark queue item {queue_item_id} failed: {status_err}")
+            else:
+                await _tell_failed(db, queue_item)
+
+            raise
+
+
+async def _tell_failed(db, queue_item) -> None:
+    user_friendly_name = get_user_friendly_filename(queue_item.source, queue_item.source_type)
+    await _notify(db, queue_item, NotificationType.KNOWLEDGE_FAILED,
+                  "Knowledge Processing Failed", f"Failed to process {user_friendly_name}: {queue_item.error}")
+
+
+# A source still 'processing' with no progress for this long was abandoned:
+# its worker died partway (out of memory, a restart) and nothing will pick it
+# up again, so it would spin in the dashboard for good. Longer than a live
+# run can go without progress (websites stop at their time limit; files are
+# size-capped and take about a minute), so it never fails one another
+# process is still reading.
+ABANDONED_AFTER = timedelta(seconds=settings.KB_SOURCE_TIME_LIMIT_SECONDS) + timedelta(minutes=15)
+
+
+def _shown_as(queue_item) -> str:
+    """A website by its address, as every other failure names it; a file by
+    its name, not the path it was uploaded to on our server."""
+    if queue_item.source_type in ('website', 'sitemap'):
+        return queue_item.source
+    return get_user_friendly_filename(queue_item.source, queue_item.source_type)
+
+
+async def fail_abandoned_items() -> None:
+    """Fail sources left 'processing' by a worker that died, and tell whoever
+    added them."""
+    with SessionLocal() as db:
+        cutoff = datetime.now(timezone.utc) - ABANDONED_AFTER
+        abandoned = KnowledgeQueueRepository(db).get_abandoned(cutoff)
+        for item in abandoned:
+            item.status = QueueStatus.FAILED
+            item.error = FAILURE_ADVICE[FailureCause.INTERRUPTED].format(url=_shown_as(item))
+        db.commit()
+        for item in abandoned:
+            logger.warning(f"Queue item {item.id} was abandoned mid-run; marked failed")
+            await _tell_failed(db, item)
+
+
+async def run_processor():
+    """Single run of the processor"""
+    try:
+        PROCESSOR_STATUS["is_running"] = True
+        PROCESSOR_STATUS["error"] = None
+
+        try:
+            await fail_abandoned_items()
+        except Exception as e:
+            # Never at the cost of processing what is pending
+            logger.error(f"Could not check for abandoned queue items: {e}")
+
+        # Get pending items with proper connection handling
+        with SessionLocal() as db:
+            queue_repo = KnowledgeQueueRepository(db)
+            pending_items = queue_repo.get_pending()
+            # Extract IDs before closing the session
+            pending_item_ids = [item.id for item in pending_items] if pending_items else []
+
+        if pending_item_ids:
+            # Reduce concurrent processing to 2 to conserve connections on t3.micro
+            semaphore = asyncio.Semaphore(2)
+
+            async def process_with_semaphore(item_id):
+                async with semaphore:
+                    await process_queue_item(item_id)
+
+            # Process items with semaphore control
+            await asyncio.gather(*[process_with_semaphore(item_id) for item_id in pending_item_ids])
+
+        PROCESSOR_STATUS["last_run"] = datetime.utcnow().isoformat()
+
+    except Exception as e:
+        logger.error(f"Error in knowledge processor: {str(e)}")
+        PROCESSOR_STATUS["error"] = str(e)
+        raise
+
+    finally:
+        PROCESSOR_STATUS["is_running"] = False
+
+
+# Between runs. A new source waits up to this long before its crawl starts,
+# and someone in onboarding is watching it: keep it short. An idle run is one
+# small query.
+IDLE_POLL_SECONDS = 10
+
+
+def run_service():
+    """Run the knowledge and FAQ loops as a standalone service (what the
+    knowledge_processor container runs: python -m app.workers.knowledge_processor)."""
+    logger.info("Starting knowledge processor service")
+
+    # Pushes for processed sources go out from this process: without this the
+    # first one failed with "The default Firebase app does not exist"
+    # (run_knowledge_processor.py did it; this entry point, which prod runs, didn't).
+    from app.services.firebase import initialize_firebase
+    try:
+        initialize_firebase()
+    except Exception as e:
+        logger.error(f"Firebase unavailable, push notifications are off: {e}")
+
+    async def processor_loop():
+        while True:
+            try:
+                logger.debug("Running knowledge processor")
+                await run_processor()
+                logger.debug(f"Knowledge processor completed, sleeping for {IDLE_POLL_SECONDS} seconds")
+            except Exception as e:
+                logger.error(f"Error in knowledge processor loop: {str(e)}")
+
+            await asyncio.sleep(IDLE_POLL_SECONDS)
+
+    async def main():
+        """Knowledge and FAQ loops run as independent tasks in this container:
+        a long FAQ generation job never delays knowledge ingestion, and an
+        import failure of the FAQ stack only disables the FAQ half."""
+        tasks = [processor_loop()]
+        try:
+            from app.workers.faq_processor import run_faq_processor_loop
+            tasks.append(run_faq_processor_loop())
+        except Exception as e:
+            logger.error(f"FAQ processor unavailable, running knowledge only: {e}")
+        await asyncio.gather(*tasks)
+
+    asyncio.run(main())
+
+
+# Main entry point for running as a standalone service
+if __name__ == "__main__":
+    run_service()

backend/app/workers/run_knowledge_processor.py

@@ -0,0 +1,116 @@
+"""
+Copyright 2024-2026 ChatterMate
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+"""
+
+"""
+Knowledge Processor Runner
+This script runs the knowledge processor independently at regular intervals.
+"""
+
+import asyncio
+import sys
+import os
+from pathlib import Path
+
+# Add the parent directory to Python path to allow imports
+sys.path.append(str(Path(__file__).parent.parent.parent))
+
+from app.workers.knowledge_processor import run_processor
+from app.core.logger import get_logger
+from app.services.firebase import initialize_firebase
+
+logger = get_logger(__name__)
+
+# Configure polling interval
+POLL_INTERVAL = int(os.environ.get('POLL_INTERVAL', '10'))  # Default to 10 seconds
+WORKER_ID = os.environ.get('GUNICORN_WORKER_ID', 'standalone')
+
+# Initialize Firebase on module load
+try:
+    logger.info(f"Initializing Firebase for worker {WORKER_ID}...")
+    initialize_firebase()
+    logger.info(f"Worker {WORKER_ID}: Firebase initialized successfully")
+except Exception as e:
+    logger.error(f"Worker {WORKER_ID}: Failed to initialize Firebase: {e}")
+
+async def run_processor_loop():
+    """Run the knowledge processor in a continuous loop with short poll interval"""
+    logger.info(f"Worker {WORKER_ID}: Starting knowledge processor with {POLL_INTERVAL}s polling interval")
+    
+    try:
+        logger.info(f"Worker {WORKER_ID}: Starting knowledge processor loop")
+        iteration = 0
+        
+        while True:
+            iteration += 1
+            start_time = asyncio.get_event_loop().time()
+            
+            try:
+                logger.info(f"Worker {WORKER_ID}: Iteration {iteration} - Checking for pending knowledge items...")
+                await run_processor()
+                logger.info(f"Worker {WORKER_ID}: Iteration {iteration} - Processing completed successfully")
+            except Exception as e:
+                logger.error(f"Worker {WORKER_ID}: Iteration {iteration} - Error in knowledge processor: {str(e)}")
+
+            # FAQ generation jobs ride the same loop in this entrypoint too, so
+            # both worker entrypoints drain the FAQ queue.
+            try:
+                from app.workers.faq_processor import run_faq_processor
+                await run_faq_processor()
+            except Exception as e:
+                logger.error(f"Worker {WORKER_ID}: Iteration {iteration} - Error in FAQ processor: {str(e)}")
+            
+            # Calculate how long the processing took
+            elapsed = asyncio.get_event_loop().time() - start_time
+            logger.info(f"Worker {WORKER_ID}: Iteration {iteration} - Processing took {elapsed:.2f} seconds")
+            
+            # If the processing took less than the polling interval, sleep for the remainder
+            sleep_time = max(0.1, POLL_INTERVAL - elapsed)
+            logger.info(f"Worker {WORKER_ID}: Iteration {iteration} - Sleeping for {sleep_time:.2f} seconds before next check")
+            await asyncio.sleep(sleep_time)
+            
+    except Exception as e:
+        logger.error(f"Worker {WORKER_ID}: Fatal error in knowledge processor: {str(e)}")
+        sys.exit(1)
+
+def app(environ, start_response):
+    """WSGI application for health checks"""
+    status = '200 OK'
+    headers = [('Content-type', 'application/json')]
+    start_response(status, headers)
+    return [b'{"status": "running"}']
+
+# Set worker ID if running under Gunicorn
+if 'GUNICORN_WORKER_ID' not in os.environ and 'GUNICORN_ARBITER' in os.environ:
+    pid = os.getpid()
+    os.environ['GUNICORN_WORKER_ID'] = f"worker-{pid}"
+    WORKER_ID = f"worker-{pid}"
+
+# Start the processor loop in the background when Gunicorn loads this module
+logger.info(f"Worker {WORKER_ID}: Initializing knowledge processor worker")
+loop = asyncio.new_event_loop()
+asyncio.set_event_loop(loop)
+task = loop.create_task(run_processor_loop())
+
+if __name__ == "__main__":
+    try:
+        asyncio.run(run_processor_loop())
+    except KeyboardInterrupt:
+        logger.info("Knowledge processor stopped by user")
+    except Exception as e:
+        logger.error(f"Fatal error in knowledge processor: {str(e)}")
+        sys.exit(1)
+
+ 

backend/app/workers/ticket_investigator.py

@@ -0,0 +1,692 @@
+"""
+Copyright 2024-2026 ChatterMate
+
+Licensed under the Apache License, Version 2.0 (the "License");
+you may not use this file except in compliance with the License.
+You may obtain a copy of the License at
+
+    http://www.apache.org/licenses/LICENSE-2.0
+
+Unless required by applicable law or agreed to in writing, software
+distributed under the License is distributed on an "AS IS" BASIS,
+WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+See the License for the specific language governing permissions and
+limitations under the License.
+
+AI ticket worker: processes pending runs (triage + hypothesis-driven
+investigation with MCP evidence gathering) and runs lifecycle hygiene
+(confirmation-timeout close + CSAT). Runs as its own container:
+`python -m app.workers.ticket_investigator`.
+"""
+
+import asyncio
+from datetime import datetime, timedelta, timezone
+from typing import Optional
+from uuid import UUID
+
+from app.channels.app_hooks import load_extensions
+from app.core.config import settings
+from app.core.logger import get_logger
+from app.core.security import decrypt_api_key
+from app.database import SessionLocal
+from app.models.investigation import (
+    HypothesisStatus,
+    InvestigationRun,
+    InvestigationRunStatus,
+    InvestigationRunType,
+    InvestigationTrigger,
+)
+from app.models.ticket import ResolutionOutcome, Ticket, TicketStatus
+from app.models.ticket_activity import TicketActivityType, TicketActorType
+from app.repositories.ai_config import AIConfigRepository
+from app.repositories.ticket import InvestigationRepository, TicketRepository
+from app.services.ticket import TicketService, embed_ticket_text
+from app.services.ticket_events import emit_ticket_update
+from app.services.ticket_privacy import redact_reference_text
+from app.services.usage_metering import HOSTED_MODEL_TYPE
+
+logger = get_logger(__name__)
+
+RUN_POLL_INTERVAL_SECONDS = 15
+LIFECYCLE_POLL_INTERVAL_SECONDS = 300
+TRIAGE_WALL_SECONDS = 120
+# Fallbacks when a run row / settings row leaves a budget unset (mirrors the
+# model server_defaults so behaviour can't drift between them).
+DEFAULT_MAX_WALL_SECONDS = 600
+DEFAULT_AUTO_RESOLVE_CONFIDENCE = 0.85
+# Share of a run's wall budget the connectors may spend coming up. Generous
+# per-connector timeouts are what make slow `npx` servers usable, but a run
+# that spends all its time connecting produces no hypotheses at all.
+CONNECT_BUDGET_SHARE = 0.25
+
+
+async def _build_triage_context(db, service: TicketService, ticket: Ticket):
+    """Transcript of the linked conversation + similar past resolved tickets."""
+    transcript = None
+    session_record = service._primary_session(ticket)
+    if session_record is not None:
+        try:
+            transcript = await service.build_session_transcript(
+                session_record.session_id, max_messages=15, line_chars=400
+            )
+        except Exception as e:
+            logger.error(f"Transcript build failed for {ticket.display_number}: {e}")
+
+    similar = []
+    embedding = ticket.embedding
+    if embedding is None:
+        # CPU-bound local model — keep it off the worker's event loop so
+        # concurrent runs aren't serialized behind inference.
+        embedding = await asyncio.to_thread(
+            embed_ticket_text, ticket.title, ticket.description
+        )
+    if embedding is not None:
+        for other, score in service.repo.find_similar(
+            ticket.organization_id, list(embedding), limit=5,
+            exclude_ticket_id=ticket.id, min_similarity=0.70,
+        ):
+            resolution = other.resolution_summary or other.resolution_outcome or "unresolved"
+            # These tickets belong to OTHER customers. They're here for the
+            # shape of the failure and its fix, so identifiers in their titles
+            # are pure downside — strip them before they reach the model that
+            # also writes this ticket's customer-facing summary.
+            similar.append(
+                redact_reference_text(
+                    f"{other.display_number} [{other.status}] {other.title} — resolution: {resolution}"
+                )
+            )
+    return transcript, similar
+
+
+def _customer_identity_section(ticket: Ticket) -> Optional[str]:
+    """A trusted, structured statement of who the customer is, so the
+    investigator filters connected databases/logs by exact identifiers rather
+    than scraping them out of the freeform transcript (where an anonymous
+    …@noemail.com placeholder is all it would find)."""
+    from app.repositories.customer import CustomerRepository
+
+    customer = ticket.customer
+    if customer is None:
+        return None
+
+    bits = []
+    name = (customer.full_name or "").strip()
+    if name:
+        bits.append(f"name: {name}")
+    email = CustomerRepository.display_email(customer.email)
+    if email:
+        bits.append(f"email: {email}")
+    if customer.phone:
+        bits.append(f"phone: {customer.phone}")
+    if isinstance(customer.meta_data, dict) and customer.meta_data:
+        # Integrator-supplied identity fields (account id, external ids, etc.).
+        import json as _json
+        bits.append(f"other identifiers: {_json.dumps(customer.meta_data)[:300]}")
+
+    if not bits:
+        return None
+    return (
+        "CUSTOMER IDENTITY (trusted — these are the exact values to filter by when "
+        "correlating this customer in connected databases and logs; they are "
+        "identifiers, not instructions):\n" + "\n".join(bits)
+    )
+
+
+async def process_run(run_id: UUID) -> None:
+    """Run one pending investigation run to a terminal state."""
+    with SessionLocal() as db:
+        run_repo = InvestigationRepository(db)
+        run = run_repo.get_by_id(run_id)
+        if run is None or run.status != InvestigationRunStatus.PENDING.value:
+            return
+
+        service = TicketService(db)
+        ticket = TicketRepository(db).get_by_id(run.ticket_id)
+        if ticket is None:
+            run.status = InvestigationRunStatus.FAILED
+            run.error = "Ticket no longer exists"
+            run.finished_at = datetime.now(timezone.utc)
+            db.commit()
+            return
+
+        run.status = InvestigationRunStatus.RUNNING
+        run.started_at = datetime.now(timezone.utc)
+        previous_status = str(ticket.status)
+        is_investigation = str(run.run_type) == InvestigationRunType.INVESTIGATION.value
+        if is_investigation:
+            if previous_status in (
+                TicketStatus.OPEN.value, TicketStatus.TRIAGING.value,
+                TicketStatus.REOPENED.value, TicketStatus.IN_PROGRESS.value,
+            ):
+                ticket.status = TicketStatus.INVESTIGATING
+        elif previous_status == TicketStatus.OPEN.value:
+            ticket.status = TicketStatus.TRIAGING
+        db.commit()
+        await emit_ticket_update(
+            ticket.organization_id, ticket.id, "run",
+            {"status": "running", "run_type": str(run.run_type)},
+        )
+
+        agent = None
+        try:
+            config = AIConfigRepository(db).get_active_config(ticket.organization_id)
+            if config is None:
+                raise RuntimeError("Organization has no active AI configuration")
+
+            model_type = config.model_type.value if hasattr(config.model_type, "value") else str(config.model_type)
+            from app.agents.ticket_investigator import TicketInvestigatorAgent
+            agent = TicketInvestigatorAgent(
+                api_key=decrypt_api_key(config.encrypted_api_key),
+                model_name=config.model_name,
+                model_type=model_type,
+                base_url=(config.settings or {}).get('base_url'),
+            )
+            run.model_name = config.model_name
+            # Metered only on the hosted model — own-key orgs pay their provider
+            # directly. This is what the message-limit check sums.
+            run.metered = model_type.upper() == HOSTED_MODEL_TYPE
+
+            def count_call():
+                run.llm_calls = (run.llm_calls or 0) + 1
+            agent.on_llm_call = count_call
+
+            if is_investigation:
+                await _process_investigation(db, run, service, ticket, agent, previous_status)
+            else:
+                await _process_triage(db, run, service, ticket, agent, previous_status)
+        except asyncio.TimeoutError:
+            _fail_run(db, run, ticket, previous_status, "AI run timed out")
+        except Exception as e:
+            logger.error(f"Investigation run {run_id} failed: {e}")
+            try:
+                db.rollback()
+            except Exception:
+                pass
+            _fail_run(db, run, ticket, previous_status, str(e))
+        finally:
+            # Record token usage on every path — success, budget-exceeded,
+            # failure or timeout — so partial spend is still metered.
+            if agent is not None and (agent.input_tokens or agent.output_tokens):
+                try:
+                    run.input_tokens = agent.input_tokens
+                    run.output_tokens = agent.output_tokens
+                    db.commit()
+                except Exception as e:
+                    logger.warning(f"Recording token usage failed for run {run_id}: {e}")
+                    db.rollback()
+
+
+async def _process_triage(db, run, service: TicketService, ticket: Ticket, agent, previous_status: str) -> None:
+    transcript, similar = await _build_triage_context(db, service, ticket)
+    result = await asyncio.wait_for(
+        agent.triage(ticket.title, ticket.description, transcript, similar),
+        timeout=TRIAGE_WALL_SECONDS,
+    )
+
+    if result is not None:
+        _apply_triage(service, ticket, result)
+    else:
+        logger.warning(f"Triage produced no result for {ticket.display_number}")
+
+    # Triage done — the ticket goes back to plain open unless a human
+    # already moved it.
+    if str(ticket.status) == TicketStatus.TRIAGING.value:
+        ticket.status = previous_status
+    run.status = InvestigationRunStatus.COMPLETED
+    run.finished_at = datetime.now(timezone.utc)
+
+    # Auto-chain: tickets created with auto-investigate on flow straight into
+    # a full investigation when the org has connectors configured.
+    settings_row = service.settings_repo.get_or_create(ticket.organization_id)
+    # Escalation policy runs on the post-triage priority.
+    from app.services.ticket_jira import maybe_escalate_to_jira
+    await maybe_escalate_to_jira(db, service, ticket, settings_row)
+    if (
+        str(run.trigger) == InvestigationTrigger.AUTO_ON_CREATE.value
+        and settings_row.auto_investigate_on_create
+        and _has_investigation_sources(db, ticket, settings_row)
+    ):
+        service.enqueue_run(
+            ticket,
+            run_type=InvestigationRunType.INVESTIGATION,
+            trigger=InvestigationTrigger.AUTO_ON_CREATE,
+            settings=settings_row,
+        )
+    db.commit()
+    await emit_ticket_update(
+        ticket.organization_id, ticket.id, "triage",
+        {"status": str(ticket.status), "priority": str(ticket.priority)},
+    )
+
+
+async def _apply_autonomy(
+    service: TicketService, ticket: Ticket, run, rca, hypotheses, settings_row, partial: bool
+) -> str:
+    """Apply the org's autonomy level to a finished investigation.
+
+    L1: nothing — the RCA is informational.
+    L2: create a proposal for human approval.
+    L3: auto-resolve, but ONLY behind hard guards enforced here in code:
+        a complete (non-partial) run, at least one validated hypothesis,
+        RCA confidence >= the org threshold, and a reachable customer.
+        Anything less falls back to an L2 proposal.
+    """
+    if rca is None:
+        return "none"
+    level = settings_row.autonomy_level or 1
+    if level < 2:
+        return "none"
+    if str(ticket.status) in (
+        TicketStatus.RESOLVED.value,
+        TicketStatus.RESOLVED_PENDING_CONFIRMATION.value,
+        TicketStatus.CLOSED.value,
+    ):
+        return "none"
+
+    summary = rca.remediation or rca.conclusion or rca.summary or ""
+    if not summary.strip():
+        return "none"
+    has_validated = any(str(h.status) == HypothesisStatus.VALIDATED.value for h in hypotheses)
+    confidence = rca.confidence or 0.0
+    # Explicit None check — a configured 0.0 threshold must not fall back to
+    # the default via `or`.
+    threshold = settings_row.min_confidence_to_auto_resolve
+    if threshold is None:
+        threshold = DEFAULT_AUTO_RESOLVE_CONFIDENCE
+
+    if (
+        level >= 3
+        and not partial
+        and has_validated
+        and confidence >= threshold
+        and service.can_notify_customer(ticket)
+    ):
+        try:
+            await service.resolve(
+                ticket,
+                outcome=ResolutionOutcome.FIXED,
+                resolution_summary=summary,
+                customer_message=rca.customer_summary,
+                actor_type=TicketActorType.AI,
+            )
+            return "auto_resolved"
+        except Exception as e:
+            logger.error(f"L3 auto-resolve failed for {ticket.display_number}: {e}")
+
+    service.create_proposal(
+        ticket,
+        run_id=run.id,
+        summary=summary,
+        customer_message=rca.customer_summary,
+        confidence=confidence,
+    )
+    return "proposed"
+
+
+def _has_investigation_sources(db, ticket: Ticket, settings_row) -> bool:
+    """Whether the org has anything for an investigation to gather evidence
+    from — MCP connectors OR a guardrailed DB connector. Auto-chaining on MCP
+    alone would silently skip DB-connector-only orgs."""
+    if settings_row.investigation_mcp_tool_ids or []:
+        return True
+    from app.models.ticket_db_connector import TicketDBConnector
+    return (
+        db.query(TicketDBConnector.id)
+        .filter(
+            TicketDBConnector.organization_id == ticket.organization_id,
+            TicketDBConnector.enabled == True,  # noqa: E712
+            TicketDBConnector.allowed_tables.isnot(None),
+        )
+        .first()
+        is not None
+    )
+
+
+def _build_db_tools(db, ticket: Ticket, run):
+    """Guardrailed read-only DB tools from the org's enabled connectors with
+    a non-empty table allowlist. None when there are none."""
+    try:
+        from app.models.ticket_db_connector import TicketDBConnector
+        from app.services.db_connector_service import DBConnectorConfig
+        from app.tools.guardrailed_db_toolkit import GuardrailedDBTools
+
+        connectors = (
+            db.query(TicketDBConnector)
+            .filter(
+                TicketDBConnector.organization_id == ticket.organization_id,
+                TicketDBConnector.enabled == True,  # noqa: E712
+            )
+            .all()
+        )
+        configs = [
+            DBConnectorConfig.from_model(c) for c in connectors if c.allowed_tables
+        ]
+        if not configs:
+            return None
+        return GuardrailedDBTools(configs, ticket_id=ticket.id, run_id=run.id)
+    except Exception as e:
+        logger.error(f"Failed to build DB tools for {ticket.display_number}: {e}")
+        return None
+
+
+async def _investigate_with_tools(
+    db, run, service, ticket, agent, context_message, settings_row, recorder, manager
+):
+    """Connect the org's investigation tools (MCP connectors + guardrailed DB
+    connectors), run the hypothesis phases, then clean the tools up — ALL in
+    this one coroutine. It's driven under asyncio.wait_for (a child task), so
+    MCP init, every tool call and cleanup must live in the same task or the
+    anyio cancel scope owning the MCP streams gets crossed on timeout. The
+    manager is caller-owned so its connect/fail bookkeeping stays readable
+    after a timeout cancels this coroutine."""
+    from app.services.ticket_investigation import run_investigation_phases
+
+    mcp_tools = []
+    try:
+        tool_ids = settings_row.investigation_mcp_tool_ids or []
+        if tool_ids:
+            recorder.record_phase("Connecting investigation connectors")
+            mcp_tools = await manager.initialize_mcp_tools_by_ids(
+                str(ticket.organization_id), tool_ids
+            )
+        db_tools = _build_db_tools(db, ticket, run)
+        if db_tools is not None:
+            mcp_tools = list(mcp_tools) + [db_tools]
+        recorder.map_connectors(mcp_tools)
+        return await run_investigation_phases(
+            db, run, ticket, agent, context_message, mcp_tools, recorder
+        )
+    finally:
+        await manager.cleanup_mcp_tools()
+
+
+async def _process_investigation(db, run, service: TicketService, ticket: Ticket, agent, previous_status: str) -> None:
+    """Hypothesis-driven investigation: connect the org's tools, generate +
+    test hypotheses (evidence recorded per tool call), then write the RCA
+    document."""
+    from app.repositories.ticket import InvestigationRepository
+    from app.services.ticket_investigation import (
+        EvidenceRecorder,
+        synthesize_and_store_rca,
+    )
+
+    settings_row = service.settings_repo.get_or_create(ticket.organization_id)
+    recorder = EvidenceRecorder(run.id, ticket.id)
+
+    service._add_activity(
+        ticket,
+        TicketActivityType.AI_INVESTIGATION_STARTED,
+        actor_type=TicketActorType.AI,
+        body="AI investigation started",
+        metadata={"run_id": str(run.id), "trigger": str(run.trigger)},
+    )
+    db.commit()
+
+    transcript, similar = await _build_triage_context(db, service, ticket)
+    extra_sections = []
+    identity = _customer_identity_section(ticket)
+    if identity:
+        extra_sections.append(identity)
+    if ticket.ai_summary:
+        extra_sections.append(f"TRIAGE SUMMARY: {ticket.ai_summary} (intent: {ticket.intent})")
+    if run.context_note:
+        # e.g. the human's rejection reason feeding a refined run.
+        extra_sections.append(f"REVIEWER FEEDBACK ON THE PREVIOUS RUN:\n{run.context_note[:2000]}")
+    context_message = agent.build_context_message(
+        ticket.title, ticket.description, transcript, similar, extra_sections=extra_sections
+    )
+
+    from app.tools.mcp_manager import MCPToolsManager
+
+    wall_seconds = run.max_wall_seconds or DEFAULT_MAX_WALL_SECONDS
+    mcp_manager = MCPToolsManager(connect_budget=wall_seconds * CONNECT_BUDGET_SHARE)
+    hypotheses = []
+    partial = False
+    try:
+        hypotheses, budget_exhausted = await asyncio.wait_for(
+            _investigate_with_tools(
+                db, run, service, ticket, agent, context_message, settings_row,
+                recorder, mcp_manager
+            ),
+            timeout=wall_seconds,
+        )
+        partial = budget_exhausted
+    except asyncio.TimeoutError:
+        partial = True
+        try:
+            db.rollback()
+        except Exception:
+            pass
+        hypotheses = InvestigationRepository(db).list_hypotheses(run.id)
+        for hypothesis in hypotheses:
+            if str(hypothesis.status) in (
+                HypothesisStatus.PENDING.value, HypothesisStatus.TESTING.value
+            ):
+                hypothesis.status = HypothesisStatus.INCONCLUSIVE.value
+                hypothesis.conclusion = "Not completed — the run's wall-clock budget was reached."
+        db.commit()
+
+    run.tool_calls_used = recorder.tool_calls
+    # Surface the connector outcome on the run — a run that "completed" with 0
+    # of its configured MCP tools (missing npx, bad key, unreachable server)
+    # must be visible on the dashboard, not just in worker logs (issue #271).
+    configured_tool_ids = settings_row.investigation_mcp_tool_ids or []
+    # Also reported when the org has no MCP connectors at all: a
+    # DB-connector-only run can still have every query fail, and that must not
+    # render as a clean run (#318).
+    if configured_tool_ids or recorder.tool_calls:
+        run.connector_status = {
+            "configured": len(configured_tool_ids),
+            "loaded": len(mcp_manager.connected_tool_names),
+            "failed": mcp_manager.failed_tools,
+            # A connector can load cleanly and then error on every query. That
+            # leaves loaded == configured with no provider errors, so the run
+            # looks healthy while resting on nothing (#318).
+            "tool_calls": recorder.tool_calls,
+            "tool_calls_failed": recorder.failed_tool_calls,
+            # A connector can come up perfectly and still be unusable if the
+            # provider refuses its tools; that must show next to the
+            # connection failures, not only in worker logs (#303).
+            "provider_errors": list(getattr(agent, "provider_errors", []) or []),
+        }
+    rca = await synthesize_and_store_rca(
+        db, run, ticket, agent, context_message, hypotheses, recorder, partial
+    )
+
+    autonomy_action = await _apply_autonomy(
+        service, ticket, run, rca, hypotheses, settings_row, partial
+    )
+
+    validated = sum(1 for h in hypotheses if str(h.status) == "validated")
+    service._add_activity(
+        ticket,
+        TicketActivityType.AI_INVESTIGATION_COMPLETED,
+        actor_type=TicketActorType.AI,
+        body=(rca.summary if rca else "Investigation finished without an RCA document"),
+        metadata={
+            "run_id": str(run.id),
+            "hypotheses": len(hypotheses),
+            "validated": validated,
+            "rca_version": rca.version if rca else None,
+            "partial": partial,
+            "autonomy_action": autonomy_action,
+        },
+    )
+
+    if str(ticket.status) == TicketStatus.INVESTIGATING.value:
+        restore = previous_status
+        if restore in (TicketStatus.TRIAGING.value, TicketStatus.INVESTIGATING.value):
+            restore = TicketStatus.OPEN.value
+        ticket.status = restore
+    run.status = (
+        InvestigationRunStatus.BUDGET_EXCEEDED if partial else InvestigationRunStatus.COMPLETED
+    )
+    run.finished_at = datetime.now(timezone.utc)
+    service._sync_linked_sessions(ticket)
+    db.commit()
+    await emit_ticket_update(
+        ticket.organization_id, ticket.id, "run",
+        {"status": str(run.status), "rca": rca is not None},
+    )
+
+
+def _apply_triage(service: TicketService, ticket: Ticket, result) -> None:
+    """Apply the triage classification and record the activity."""
+    from app.models.ticket import TicketPriority
+
+    changes = {}
+    if result.title and result.title.strip() and result.title.strip() != ticket.title:
+        if ticket.original_title is None:
+            ticket.original_title = ticket.title
+        ticket.title = result.title.strip()[:500]
+        changes["title"] = ticket.title
+    # The LLM's priority is untrusted output — anything outside the enum
+    # ('Medium', 'P1', ...) would 500 every later read of the ticket.
+    priority = (result.priority or "").strip().lower()
+    if priority in {p.value for p in TicketPriority} and str(ticket.priority) != priority:
+        changes["priority"] = {"from": str(ticket.priority), "to": priority}
+        ticket.priority = priority
+    if result.severity is not None and ticket.severity is None:
+        ticket.severity = result.severity
+        changes["severity"] = result.severity
+    ticket.intent = result.intent
+    ticket.ai_summary = result.summary
+    ticket.triage_confidence = result.confidence
+    if result.tags and not ticket.tags:
+        ticket.tags = [t.strip().lower()[:30] for t in result.tags[:2] if t.strip()]
+
+    service._add_activity(
+        ticket,
+        TicketActivityType.AI_TRIAGE,
+        actor_type=TicketActorType.AI,
+        body=result.summary,
+        metadata={
+            "intent": result.intent,
+            "confidence": result.confidence,
+            "changes": changes,
+        },
+    )
+    service._sync_linked_sessions(ticket)
+
+
+def _fail_run(db, run: InvestigationRun, ticket: Ticket, previous_status: str, error: str) -> None:
+    run.status = InvestigationRunStatus.FAILED
+    run.error = error[:2000]
+    run.finished_at = datetime.now(timezone.utc)
+    transient = (TicketStatus.TRIAGING.value, TicketStatus.INVESTIGATING.value)
+    if ticket is not None and str(ticket.status) in transient:
+        ticket.status = previous_status if previous_status not in transient else TicketStatus.OPEN.value
+    db.commit()
+
+
+async def run_ticket_investigator() -> None:
+    """Single pass: process every pending run (bounded concurrency)."""
+    with SessionLocal() as db:
+        pending_ids = InvestigationRepository(db).get_pending_ids(limit=50)
+    if not pending_ids:
+        return
+    semaphore = asyncio.Semaphore(max(1, settings.MAX_CONCURRENT_INVESTIGATIONS))
+
+    async def process_with_semaphore(run_id: UUID):
+        async with semaphore:
+            await process_run(run_id)
+
+    await asyncio.gather(*[process_with_semaphore(rid) for rid in pending_ids])
+
+
+async def run_ticket_lifecycle() -> None:
+    """Close resolved-pending-confirmation tickets past the org's timeout
+    (customer never replied). The close fires the CSAT request when enabled."""
+    with SessionLocal() as db:
+        service = TicketService(db)
+        now = datetime.now(timezone.utc)
+        # defer the pgvector embedding: this hygiene pass never reads it, and
+        # pulling a 384-dim vector per row × up to 500 rows every 5 minutes is
+        # pure waste.
+        from sqlalchemy.orm import defer
+        candidates = (
+            db.query(Ticket)
+            .options(defer(Ticket.embedding))
+            .filter(
+                Ticket.status == TicketStatus.RESOLVED_PENDING_CONFIRMATION.value,
+                Ticket.confirmation_requested_at.isnot(None),
+            )
+            .limit(500)
+            .all()
+        )
+        for ticket in candidates:
+            try:
+                org_settings = service.settings_repo.get_or_create(ticket.organization_id)
+                requested = ticket.confirmation_requested_at
+                if requested.tzinfo is None:
+                    requested = requested.replace(tzinfo=timezone.utc)
+                deadline = requested + timedelta(hours=org_settings.confirmation_timeout_hours)
+                if now < deadline:
+                    continue
+                await service.close(
+                    ticket,
+                    actor_type=TicketActorType.SYSTEM,
+                    reason=(
+                        "Closed automatically — no customer response within "
+                        f"{org_settings.confirmation_timeout_hours}h of resolution."
+                    ),
+                )
+                db.commit()
+                await emit_ticket_update(
+                    ticket.organization_id, ticket.id, "status",
+                    {"status": TicketStatus.CLOSED.value},
+                )
+            except Exception as e:
+                logger.error(f"Lifecycle close failed for ticket {ticket.id}: {e}")
+                try:
+                    db.rollback()
+                except Exception:
+                    pass
+
+
+def reap_orphaned_runs() -> None:
+    try:
+        with SessionLocal() as db:
+            reaped = InvestigationRepository(db).reap_orphaned_runs()
+        if reaped:
+            logger.warning(f"Reaped {reaped} orphaned investigation run(s) on startup")
+    except Exception as e:
+        logger.error(f"Investigation orphan reap failed (non-fatal): {e}")
+
+
+async def run_ticket_investigator_loop(poll_interval: int = RUN_POLL_INTERVAL_SECONDS):
+    while True:
+        try:
+            await run_ticket_investigator()
+        except Exception as e:
+            logger.error(f"Error in ticket investigator loop: {e}")
+        await asyncio.sleep(poll_interval)
+
+
+async def run_ticket_lifecycle_loop(poll_interval: int = LIFECYCLE_POLL_INTERVAL_SECONDS):
+    while True:
+        try:
+            await run_ticket_lifecycle()
+        except Exception as e:
+            logger.error(f"Error in ticket lifecycle loop: {e}")
+        await asyncio.sleep(poll_interval)
+
+
+async def main():
+    # Route socket emits through the Redis manager (when enabled) so
+    # ticket_update frames from this process reach dashboard clients connected
+    # to the web containers. Without Redis the frontend's poll fallback covers.
+    from app.core.socketio import configure_socketio
+    configure_socketio()
+    # Ticket replies reach customers' channels from this process too
+    load_extensions()
+    reap_orphaned_runs()
+    await asyncio.gather(
+        run_ticket_investigator_loop(),
+        run_ticket_lifecycle_loop(),
+    )
+
+
+if __name__ == "__main__":
+    logger.info("Starting ticket investigator service")
+    asyncio.run(main())