"""Project Knowledge Graph service.

第一阶段目标：
1. 构建统一实体/关系/证据写入能力
2. 支持结构化数据（project/task/signal/user）同步
3. 支持代码目录扫描快照
4. 提供图谱概览统计
"""

from __future__ import annotations

import hashlib
import json
import os
from datetime import datetime, timedelta, timezone
from pathlib import Path
from typing import Any, Optional

from sqlalchemy import and_, func, select
from sqlalchemy.ext.asyncio import AsyncSession

from ..models import (
    IngestEvent,
    IngestOffset,
    KGEvidence,
    KGEntity,
    KGRelation,
    KGSyncRun,
    Project,
    FeishuItem,
    Signal,
    SymbolDefinition,
    SymbolReference,
    Task,
    User,
)
from .context_cache import context_cache_service
from .symbol_index_service import SymbolIndexService


from ..utils.timezone import now_cn_naive
def _utc_now_naive() -> datetime:
    return now_cn_naive()


DDL_BLOCKED_PREFIXES = [
    "CREATE",
    "ALTER",
    "DROP",
    "TRUNCATE",
    "RENAME",
    "COMMENT",
    "GRANT",
    "REVOKE",
]
DDL_GUARDRAIL_POLICY_KEY = "policy:ddl_via_interface_only"
DDL_GUARDRAIL_POLICY_NAME = "DDL 只能通过受控接口/迁移流程"


class ProjectKnowledgeGraphService:
    def __init__(self, db: AsyncSession, *, tenant_id: int) -> None:
        self.db = db
        self.tenant_id = int(tenant_id)

    async def upsert_entity(
        self,
        *,
        entity_type: str,
        entity_key: str,
        display_name: str,
        payload: Optional[dict[str, Any]] = None,
        source_type: str = "system",
        source_ref: Optional[str] = None,
    ) -> KGEntity:
        row = (
            await self.db.execute(
                select(KGEntity).where(
                    and_(
                        KGEntity.tenant_id == self.tenant_id,
                        KGEntity.entity_type == str(entity_type).strip()[:40],
                        KGEntity.entity_key == str(entity_key).strip()[:255],
                    )
                )
            )
        ).scalar_one_or_none()

        if row is None:
            row = KGEntity(
                tenant_id=self.tenant_id,
                entity_type=str(entity_type).strip()[:40],
                entity_key=str(entity_key).strip()[:255],
                display_name=str(display_name or entity_key).strip()[:255],
                payload=payload or {},
                source_type=str(source_type or "system").strip()[:32],
                source_ref=(str(source_ref).strip()[:255] if source_ref else None),
                last_seen_at=_utc_now_naive(),
            )
            self.db.add(row)
            await self.db.flush()
            return row

        row.display_name = str(display_name or row.display_name).strip()[:255]
        row.payload = payload or {}
        row.source_type = str(source_type or row.source_type or "system").strip()[:32]
        row.source_ref = str(source_ref).strip()[:255] if source_ref else row.source_ref
        row.last_seen_at = _utc_now_naive()
        await self.db.flush()
        return row

    async def upsert_relation(
        self,
        *,
        relation_type: str,
        subject_entity_id: int,
        object_entity_id: int,
        weight: float = 1.0,
        payload: Optional[dict[str, Any]] = None,
        source_type: str = "system",
        source_ref: Optional[str] = None,
    ) -> KGRelation:
        row = (
            await self.db.execute(
                select(KGRelation).where(
                    and_(
                        KGRelation.tenant_id == self.tenant_id,
                        KGRelation.relation_type == str(relation_type).strip()[:40],
                        KGRelation.subject_entity_id == int(subject_entity_id),
                        KGRelation.object_entity_id == int(object_entity_id),
                    )
                )
            )
        ).scalar_one_or_none()

        if row is None:
            row = KGRelation(
                tenant_id=self.tenant_id,
                relation_type=str(relation_type).strip()[:40],
                subject_entity_id=int(subject_entity_id),
                object_entity_id=int(object_entity_id),
                weight=float(weight),
                payload=payload or {},
                source_type=str(source_type or "system").strip()[:32],
                source_ref=(str(source_ref).strip()[:255] if source_ref else None),
            )
            self.db.add(row)
            await self.db.flush()
            return row

        row.weight = float(weight)
        row.payload = payload or {}
        row.source_type = str(source_type or row.source_type or "system").strip()[:32]
        row.source_ref = str(source_ref).strip()[:255] if source_ref else row.source_ref
        await self.db.flush()
        return row

    async def add_evidence(
        self,
        *,
        evidence_type: str,
        content_excerpt: str,
        entity_id: Optional[int] = None,
        relation_id: Optional[int] = None,
        source_type: str = "system",
        source_ref: Optional[str] = None,
        metadata: Optional[dict[str, Any]] = None,
    ) -> KGEvidence:
        clean = str(content_excerpt or "").strip()
        digest = hashlib.md5(clean.encode("utf-8")).hexdigest()

        row = KGEvidence(
            tenant_id=self.tenant_id,
            entity_id=int(entity_id) if entity_id is not None else None,
            relation_id=int(relation_id) if relation_id is not None else None,
            evidence_type=str(evidence_type or "snapshot").strip()[:40],
            content_hash=digest,
            content_excerpt=clean[:4000] if clean else "(empty)",
            source_type=str(source_type or "system").strip()[:32],
            source_ref=(str(source_ref).strip()[:255] if source_ref else None),
            meta_payload=metadata or {},
            occurred_at=_utc_now_naive(),
        )
        self.db.add(row)
        await self.db.flush()
        return row

    async def enqueue_ingest_event(
        self,
        *,
        channel: str,
        event_key: str,
        payload: Optional[dict[str, Any]] = None,
        status: str = "pending",
    ) -> IngestEvent:
        row = (
            await self.db.execute(
                select(IngestEvent).where(
                    and_(
                        IngestEvent.tenant_id == self.tenant_id,
                        IngestEvent.channel == str(channel).strip()[:32],
                        IngestEvent.event_key == str(event_key).strip()[:128],
                    )
                )
            )
        ).scalar_one_or_none()
        if row is None:
            row = IngestEvent(
                tenant_id=self.tenant_id,
                channel=str(channel).strip()[:32],
                event_key=str(event_key).strip()[:128],
                status=str(status or "pending").strip()[:20],
                payload=payload or {},
            )
            self.db.add(row)
        else:
            row.status = str(status or row.status or "pending").strip()[:20]
            row.payload = payload or row.payload or {}
            row.error_message = None
        await self.db.flush()
        return row

    async def set_ingest_offset(
        self,
        *,
        channel: str,
        offset_key: str,
        offset_value: str,
        payload: Optional[dict[str, Any]] = None,
    ) -> IngestOffset:
        row = (
            await self.db.execute(
                select(IngestOffset).where(
                    and_(
                        IngestOffset.tenant_id == self.tenant_id,
                        IngestOffset.channel == str(channel).strip()[:32],
                        IngestOffset.offset_key == str(offset_key).strip()[:128],
                    )
                )
            )
        ).scalar_one_or_none()

        if row is None:
            row = IngestOffset(
                tenant_id=self.tenant_id,
                channel=str(channel).strip()[:32],
                offset_key=str(offset_key).strip()[:128],
                offset_value=str(offset_value or "").strip()[:255],
                payload=payload or {},
            )
            self.db.add(row)
        else:
            row.offset_value = str(offset_value or row.offset_value).strip()[:255]
            row.payload = payload or row.payload or {}
        await self.db.flush()
        return row

    async def get_ingest_offset(self, *, channel: str, offset_key: str) -> Optional[str]:
        row = (
            await self.db.execute(
                select(IngestOffset).where(
                    and_(
                        IngestOffset.tenant_id == self.tenant_id,
                        IngestOffset.channel == str(channel).strip()[:32],
                        IngestOffset.offset_key == str(offset_key).strip()[:128],
                    )
                )
            )
        ).scalar_one_or_none()
        if row is None:
            return None
        return str(row.offset_value or "").strip() or None

    @staticmethod
    def _parse_iso_dt(value: Optional[str]) -> Optional[datetime]:
        raw = str(value or "").strip()
        if not raw:
            return None
        try:
            dt = datetime.fromisoformat(raw)
            return dt.replace(tzinfo=None) if dt.tzinfo else dt
        except Exception:
            return None

    async def _open_sync_run(
        self,
        *,
        run_type: str,
        trigger_source: str,
        input_payload: Optional[dict[str, Any]] = None,
    ) -> KGSyncRun:
        now = _utc_now_naive()
        row = KGSyncRun(
            tenant_id=self.tenant_id,
            run_type=str(run_type or "incremental").strip()[:40],
            trigger_source=str(trigger_source or "system").strip()[:40],
            status="running",
            started_at=now,
            input_payload=input_payload or {},
            result_payload={},
        )
        self.db.add(row)
        await self.db.flush()
        await self.db.commit()
        await self.db.refresh(row)
        return row

    async def _close_sync_run(
        self,
        *,
        row: Optional[KGSyncRun],
        status: str,
        result_payload: Optional[dict[str, Any]] = None,
        error_message: Optional[str] = None,
        started_at_hint: Optional[datetime] = None,
    ) -> None:
        if row is None:
            return
        now = _utc_now_naive()
        started = started_at_hint or row.started_at or now
        row.status = str(status or "unknown").strip()[:20]
        row.ended_at = now
        row.duration_ms = max(int((now - started).total_seconds() * 1000), 0)
        row.result_payload = result_payload or {}
        row.error_message = (str(error_message).strip()[:2000] if error_message else None)
        await self.db.commit()

    async def list_sync_runs(
        self,
        *,
        limit: int = 50,
        run_type: Optional[str] = None,
        status: Optional[str] = None,
    ) -> dict[str, Any]:
        max_rows = max(1, min(int(limit), 200))
        stmt = select(KGSyncRun).where(KGSyncRun.tenant_id == self.tenant_id)
        if run_type:
            stmt = stmt.where(KGSyncRun.run_type == str(run_type).strip()[:40])
        if status:
            stmt = stmt.where(KGSyncRun.status == str(status).strip()[:20])
        rows = (
            await self.db.execute(stmt.order_by(KGSyncRun.started_at.desc(), KGSyncRun.id.desc()).limit(max_rows))
        ).scalars().all()

        items = []
        for row in rows:
            items.append(
                {
                    "id": int(row.id),
                    "run_type": str(row.run_type or ""),
                    "trigger_source": str(row.trigger_source or ""),
                    "status": str(row.status or ""),
                    "started_at": row.started_at.isoformat() if row.started_at else None,
                    "ended_at": row.ended_at.isoformat() if row.ended_at else None,
                    "duration_ms": int(row.duration_ms or 0),
                    "input_payload": row.input_payload or {},
                    "result_payload": row.result_payload or {},
                    "error_message": str(row.error_message or ""),
                }
            )
        return {
            "tenant_id": self.tenant_id,
            "count": len(items),
            "items": items,
        }

    async def get_sync_slo(self, *, lookback_hours: int = 24) -> dict[str, Any]:
        hours = max(1, min(int(lookback_hours), 24 * 30))
        now = _utc_now_naive()
        since = now - timedelta(hours=hours)

        runs = (
            await self.db.execute(
                select(KGSyncRun).where(
                    and_(
                        KGSyncRun.tenant_id == self.tenant_id,
                        KGSyncRun.started_at >= since,
                        KGSyncRun.run_type == "incremental",
                    )
                )
            )
        ).scalars().all()

        total = len(runs)
        success = sum(1 for x in runs if str(x.status or "") == "success")
        partial = sum(1 for x in runs if str(x.status or "") == "partial")
        failed = sum(1 for x in runs if str(x.status or "") == "failed")
        avg_duration_ms = (
            int(sum(int(x.duration_ms or 0) for x in runs) / total)
            if total > 0
            else 0
        )

        latest_success = (
            await self.db.execute(
                select(KGSyncRun)
                .where(
                    and_(
                        KGSyncRun.tenant_id == self.tenant_id,
                        KGSyncRun.run_type == "incremental",
                        KGSyncRun.status == "success",
                    )
                )
                .order_by(KGSyncRun.started_at.desc(), KGSyncRun.id.desc())
                .limit(1)
            )
        ).scalar_one_or_none()

        dead_letter = (
            await self.db.execute(
                select(func.count(IngestEvent.id)).where(
                    and_(
                        IngestEvent.tenant_id == self.tenant_id,
                        IngestEvent.status == "dead_letter",
                    )
                )
            )
        ).scalar_one()
        pending_events = (
            await self.db.execute(
                select(func.count(IngestEvent.id)).where(
                    and_(
                        IngestEvent.tenant_id == self.tenant_id,
                        IngestEvent.status.in_(["pending", "retry"]),
                    )
                )
            )
        ).scalar_one()

        async def _freshness_minutes(channel: str, offset_key: str) -> Optional[int]:
            raw = await self.get_ingest_offset(channel=channel, offset_key=offset_key)
            dt = self._parse_iso_dt(raw)
            if dt is None:
                return None
            return max(int((now - dt).total_seconds() // 60), 0)

        return {
            "tenant_id": self.tenant_id,
            "lookback_hours": hours,
            "window_started_at": since.isoformat(),
            "window_ended_at": now.isoformat(),
            "runs_total": total,
            "runs_success": success,
            "runs_partial": partial,
            "runs_failed": failed,
            "success_rate": round((success / total), 4) if total > 0 else 0.0,
            "avg_duration_ms": avg_duration_ms,
            "latest_success_at": latest_success.started_at.isoformat() if latest_success and latest_success.started_at else None,
            "dead_letter_events": int(dead_letter or 0),
            "pending_events": int(pending_events or 0),
            "freshness_minutes": {
                "db_structured": await _freshness_minutes("db_delta", "last_structured_sync_at"),
                "feishu_delta": await _freshness_minutes("feishu_delta", "last_feishu_item_updated_at"),
                "code_scan": await _freshness_minutes("code_scan", "last_code_scan_at"),
                "symbol_scan": await _freshness_minutes("symbol_scan", "last_symbol_scan_at"),
                "feature_catalog": await _freshness_minutes("feature_catalog", "last_feature_catalog_sync_at"),
            },
        }

    async def sync_structured_delta(self, *, limit: int = 1000) -> dict[str, Any]:
        max_rows = max(1, min(int(limit), 5000))
        last_raw = await self.get_ingest_offset(channel="db_delta", offset_key="last_structured_sync_at")
        last_dt = self._parse_iso_dt(last_raw) or (datetime(1970, 1, 1))
        counters = {
            "projects": 0,
            "tasks": 0,
            "signals": 0,
            "users": 0,
            "relations": 0,
            "evidences": 0,
            "from_offset": last_dt.isoformat(),
        }
        newest_seen = last_dt

        project_rows = (
            await self.db.execute(
                select(Project)
                .where(
                    and_(
                        Project.tenant_id == self.tenant_id,
                        Project.updated_at >= last_dt,
                    )
                )
                .order_by(Project.updated_at.asc(), Project.id.asc())
                .limit(max_rows)
            )
        ).scalars().all()
        project_map: dict[int, int] = {}
        for p in project_rows:
            entity = await self.upsert_entity(
                entity_type="project",
                entity_key=f"project:{int(p.id)}",
                display_name=str(p.title or p.name or f"Project#{int(p.id)}"),
                payload={
                    "project_id": int(p.id),
                    "status": str(p.status or "unknown"),
                    "owner_id": int(p.owner_id) if p.owner_id is not None else None,
                },
                source_type="db",
                source_ref=f"projects:{int(p.id)}",
            )
            project_map[int(p.id)] = int(entity.id)
            counters["projects"] += 1
            if p.updated_at and p.updated_at > newest_seen:
                newest_seen = p.updated_at

        user_rows = (
            await self.db.execute(
                select(User)
                .where(
                    and_(
                        User.tenant_id == self.tenant_id,
                        User.updated_at >= last_dt,
                    )
                )
                .order_by(User.updated_at.asc(), User.id.asc())
                .limit(max_rows)
            )
        ).scalars().all()
        user_map: dict[int, int] = {}
        for u in user_rows:
            entity = await self.upsert_entity(
                entity_type="user",
                entity_key=f"user:{int(u.id)}",
                display_name=str(u.name or f"User#{int(u.id)}"),
                payload={
                    "user_id": int(u.id),
                    "role": str(u.role or "unknown"),
                    "is_active": bool(u.is_active),
                },
                source_type="db",
                source_ref=f"users:{int(u.id)}",
            )
            user_map[int(u.id)] = int(entity.id)
            counters["users"] += 1
            if u.updated_at and u.updated_at > newest_seen:
                newest_seen = u.updated_at

        task_rows = (
            await self.db.execute(
                select(Task)
                .where(
                    and_(
                        Task.tenant_id == self.tenant_id,
                        Task.updated_at >= last_dt,
                    )
                )
                .order_by(Task.updated_at.asc(), Task.id.asc())
                .limit(max_rows)
            )
        ).scalars().all()
        for t in task_rows:
            task_entity = await self.upsert_entity(
                entity_type="task",
                entity_key=f"task:{int(t.id)}",
                display_name=str(t.title or f"Task#{int(t.id)}"),
                payload={
                    "task_id": int(t.id),
                    "project_id": int(t.project_id) if t.project_id is not None else None,
                    "assignee_id": int(t.assignee_id) if t.assignee_id is not None else None,
                    "status": str(t.status or "unknown"),
                },
                source_type="db",
                source_ref=f"tasks:{int(t.id)}",
            )
            counters["tasks"] += 1
            if t.updated_at and t.updated_at > newest_seen:
                newest_seen = t.updated_at

            if t.project_id is not None:
                target_pid = int(t.project_id)
                if target_pid not in project_map:
                    p = (
                        await self.db.execute(
                            select(Project).where(
                                and_(
                                    Project.tenant_id == self.tenant_id,
                                    Project.id == target_pid,
                                )
                            )
                        )
                    ).scalar_one_or_none()
                    if p is not None:
                        p_entity = await self.upsert_entity(
                            entity_type="project",
                            entity_key=f"project:{int(p.id)}",
                            display_name=str(p.title or p.name or f"Project#{int(p.id)}"),
                            payload={
                                "project_id": int(p.id),
                                "status": str(p.status or "unknown"),
                                "owner_id": int(p.owner_id) if p.owner_id is not None else None,
                            },
                            source_type="db",
                            source_ref=f"projects:{int(p.id)}",
                        )
                        project_map[int(p.id)] = int(p_entity.id)
                if target_pid in project_map:
                    rel = await self.upsert_relation(
                        relation_type="belongs_to",
                        subject_entity_id=int(task_entity.id),
                        object_entity_id=int(project_map[target_pid]),
                        source_type="db",
                        source_ref=f"tasks:{int(t.id)}",
                    )
                    counters["relations"] += 1
                    await self.add_evidence(
                        evidence_type="db_delta",
                        content_excerpt=f"task#{int(t.id)} belongs_to project#{target_pid}",
                        relation_id=int(rel.id),
                        source_type="db",
                        source_ref=f"tasks:{int(t.id)}",
                        metadata={"table": "tasks"},
                    )
                    counters["evidences"] += 1

            if t.assignee_id is not None:
                uid = int(t.assignee_id)
                if uid not in user_map:
                    u = (
                        await self.db.execute(
                            select(User).where(and_(User.tenant_id == self.tenant_id, User.id == uid))
                        )
                    ).scalar_one_or_none()
                    if u is not None:
                        u_entity = await self.upsert_entity(
                            entity_type="user",
                            entity_key=f"user:{uid}",
                            display_name=str(u.name or f"User#{uid}"),
                            payload={
                                "user_id": uid,
                                "role": str(u.role or "unknown"),
                                "is_active": bool(u.is_active),
                            },
                            source_type="db",
                            source_ref=f"users:{uid}",
                        )
                        user_map[uid] = int(u_entity.id)
                if uid in user_map:
                    rel = await self.upsert_relation(
                        relation_type="assigned_to",
                        subject_entity_id=int(task_entity.id),
                        object_entity_id=int(user_map[uid]),
                        source_type="db",
                        source_ref=f"tasks:{int(t.id)}",
                    )
                    counters["relations"] += 1
                    await self.add_evidence(
                        evidence_type="db_delta",
                        content_excerpt=f"task#{int(t.id)} assigned_to user#{uid}",
                        relation_id=int(rel.id),
                        source_type="db",
                        source_ref=f"tasks:{int(t.id)}",
                        metadata={"table": "tasks"},
                    )
                    counters["evidences"] += 1

        signal_rows = (
            await self.db.execute(
                select(Signal)
                .where(
                    and_(
                        Signal.tenant_id == self.tenant_id,
                        Signal.updated_at >= last_dt,
                    )
                )
                .order_by(Signal.updated_at.asc(), Signal.id.asc())
                .limit(max_rows)
            )
        ).scalars().all()
        for s in signal_rows:
            signal_entity = await self.upsert_entity(
                entity_type="signal",
                entity_key=f"signal:{int(s.id)}",
                display_name=str(s.title or s.content or f"Signal#{int(s.id)}")[:255],
                payload={
                    "signal_id": int(s.id),
                    "priority": str(s.priority or ""),
                    "status": str(s.status or ""),
                    "matched_project_id": (
                        int(s.matched_project_id) if getattr(s, "matched_project_id", None) is not None else None
                    ),
                },
                source_type="db",
                source_ref=f"signals:{int(s.id)}",
            )
            counters["signals"] += 1
            if s.updated_at and s.updated_at > newest_seen:
                newest_seen = s.updated_at

            pid = int(s.matched_project_id) if getattr(s, "matched_project_id", None) is not None else None
            if pid is not None:
                if pid not in project_map:
                    p = (
                        await self.db.execute(
                            select(Project).where(and_(Project.tenant_id == self.tenant_id, Project.id == pid))
                        )
                    ).scalar_one_or_none()
                    if p is not None:
                        p_entity = await self.upsert_entity(
                            entity_type="project",
                            entity_key=f"project:{pid}",
                            display_name=str(p.title or p.name or f"Project#{pid}"),
                            payload={
                                "project_id": pid,
                                "status": str(p.status or "unknown"),
                                "owner_id": int(p.owner_id) if p.owner_id is not None else None,
                            },
                            source_type="db",
                            source_ref=f"projects:{pid}",
                        )
                        project_map[pid] = int(p_entity.id)
                if pid in project_map:
                    rel = await self.upsert_relation(
                        relation_type="matches_project",
                        subject_entity_id=int(signal_entity.id),
                        object_entity_id=int(project_map[pid]),
                        source_type="db",
                        source_ref=f"signals:{int(s.id)}",
                    )
                    counters["relations"] += 1
                    await self.add_evidence(
                        evidence_type="db_delta",
                        content_excerpt=f"signal#{int(s.id)} matches project#{pid}",
                        relation_id=int(rel.id),
                        source_type="db",
                        source_ref=f"signals:{int(s.id)}",
                        metadata={"table": "signals"},
                    )
                    counters["evidences"] += 1

        await self.enqueue_ingest_event(
            channel="db_delta",
            event_key=f"db_delta:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
            payload=counters,
            status="done",
        )
        await self.set_ingest_offset(
            channel="db_delta",
            offset_key="last_structured_sync_at",
            offset_value=newest_seen.isoformat(),
            payload={"counters": counters},
        )
        await self.db.commit()
        counters["to_offset"] = newest_seen.isoformat()
        return counters

    async def enqueue_feishu_delta_events(self, *, limit: int = 500) -> dict[str, Any]:
        max_rows = max(1, min(int(limit), 2000))
        last_raw = await self.get_ingest_offset(channel="feishu_delta", offset_key="last_feishu_item_updated_at")
        last_dt = self._parse_iso_dt(last_raw) or (datetime(1970, 1, 1))

        rows = (
            await self.db.execute(
                select(FeishuItem)
                .where(
                    and_(
                        FeishuItem.tenant_id == self.tenant_id,
                        FeishuItem.updated_at >= last_dt,
                    )
                )
                .order_by(FeishuItem.updated_at.asc(), FeishuItem.id.asc())
                .limit(max_rows)
            )
        ).scalars().all()
        enqueued = 0
        newest_seen = last_dt
        for item in rows:
            event_key = str(item.feishu_message_id or f"feishu_item:{int(item.id)}")
            await self.enqueue_ingest_event(
                channel="feishu_delta",
                event_key=event_key[:128],
                payload={
                    "feishu_item_id": int(item.id),
                    "matched_project_id": int(item.matched_project_id) if item.matched_project_id is not None else None,
                    "level": str(item.level or ""),
                    "route_status": str(item.route_status or ""),
                },
                status="pending",
            )
            enqueued += 1
            if item.updated_at and item.updated_at > newest_seen:
                newest_seen = item.updated_at

        await self.set_ingest_offset(
            channel="feishu_delta",
            offset_key="last_feishu_item_updated_at",
            offset_value=newest_seen.isoformat(),
            payload={"batch_size": len(rows)},
        )
        await self.db.commit()
        return {
            "enqueued": enqueued,
            "from_offset": last_dt.isoformat(),
            "to_offset": newest_seen.isoformat(),
            "candidate_count": len(rows),
        }

    async def process_pending_events(self, *, max_batch: int = 200, max_retry: int = 3) -> dict[str, Any]:
        batch = max(1, min(int(max_batch), 1000))
        retries = max(0, min(int(max_retry), 10))

        rows = (
            await self.db.execute(
                select(IngestEvent)
                .where(
                    and_(
                        IngestEvent.tenant_id == self.tenant_id,
                        IngestEvent.status.in_(["pending", "retry"]),
                    )
                )
                .order_by(IngestEvent.created_at.asc(), IngestEvent.id.asc())
                .limit(batch)
            )
        ).scalars().all()

        stats = {"processed": 0, "done": 0, "retry": 0, "dead_letter": 0}
        for ev in rows:
            stats["processed"] += 1
            try:
                if str(ev.channel) == "feishu_delta":
                    await self._process_feishu_event(ev)
                ev.status = "done"
                ev.error_message = None
                ev.processed_at = _utc_now_naive()
                stats["done"] += 1
            except Exception as exc:
                ev.retry_count = int(ev.retry_count or 0) + 1
                ev.error_message = str(exc)[:1000]
                if ev.retry_count > retries:
                    ev.status = "dead_letter"
                    ev.processed_at = _utc_now_naive()
                    stats["dead_letter"] += 1
                else:
                    ev.status = "retry"
                    stats["retry"] += 1

        await self.db.commit()
        return stats

    async def _process_feishu_event(self, event: IngestEvent) -> None:
        payload = event.payload if isinstance(event.payload, dict) else {}
        item_id = payload.get("feishu_item_id")
        if item_id is None:
            return
        item = (
            await self.db.execute(
                select(FeishuItem).where(
                    and_(
                        FeishuItem.tenant_id == self.tenant_id,
                        FeishuItem.id == int(item_id),
                    )
                )
            )
        ).scalar_one_or_none()
        if item is None:
            return

        signal_entity = await self.upsert_entity(
            entity_type="signal",
            entity_key=f"feishu_item:{int(item.id)}",
            display_name=str(item.raw_message or f"FeishuItem#{int(item.id)}")[:255],
            payload={
                "feishu_item_id": int(item.id),
                "level": str(item.level or ""),
                "route_status": str(item.route_status or ""),
                "matched_project_id": int(item.matched_project_id) if item.matched_project_id is not None else None,
            },
            source_type="feishu",
            source_ref=str(item.feishu_message_id or f"item:{int(item.id)}")[:255],
        )
        await self.add_evidence(
            evidence_type="feishu_delta",
            content_excerpt=str(item.raw_message or "")[:1000] or f"feishu_item#{int(item.id)}",
            entity_id=int(signal_entity.id),
            source_type="feishu",
            source_ref=str(item.feishu_message_id or f"item:{int(item.id)}")[:255],
            metadata={"feishu_item_id": int(item.id)},
        )

        if item.matched_project_id is not None:
            project_entity = await self.upsert_entity(
                entity_type="project",
                entity_key=f"project:{int(item.matched_project_id)}",
                display_name=f"Project#{int(item.matched_project_id)}",
                payload={"project_id": int(item.matched_project_id)},
                source_type="db",
                source_ref=f"projects:{int(item.matched_project_id)}",
            )
            rel = await self.upsert_relation(
                relation_type="matches_project",
                subject_entity_id=int(signal_entity.id),
                object_entity_id=int(project_entity.id),
                source_type="feishu",
                source_ref=str(item.feishu_message_id or f"item:{int(item.id)}")[:255],
            )
            await self.add_evidence(
                evidence_type="feishu_delta",
                content_excerpt=f"feishu_item#{int(item.id)} matches project#{int(item.matched_project_id)}",
                relation_id=int(rel.id),
                source_type="feishu",
                source_ref=str(item.feishu_message_id or f"item:{int(item.id)}")[:255],
                metadata={"feishu_item_id": int(item.id)},
            )

    async def sync_codebase_if_due(
        self,
        *,
        base_dir: Optional[str] = None,
        interval_hours: int = 6,
    ) -> dict[str, Any]:
        hours = max(1, int(interval_hours))
        offset_raw = await self.get_ingest_offset(channel="code_scan", offset_key="last_code_scan_at")
        last_scan_at = self._parse_iso_dt(offset_raw)
        now = _utc_now_naive()
        if last_scan_at is not None and now - last_scan_at < timedelta(hours=hours):
            return {
                "executed": False,
                "reason": "not_due",
                "last_scan_at": last_scan_at.isoformat(),
                "next_due_at": (last_scan_at + timedelta(hours=hours)).isoformat(),
            }
        payload = await self.sync_codebase_snapshot(base_dir=base_dir)
        return {
            "executed": True,
            "reason": "due",
            "payload": payload,
        }

    async def sync_symbol_index_if_due(
        self,
        *,
        base_dir: Optional[str] = None,
        interval_hours: int = 6,
        max_files: int = 500,
    ) -> dict[str, Any]:
        hours = max(1, int(interval_hours))
        offset_raw = await self.get_ingest_offset(channel="symbol_scan", offset_key="last_symbol_scan_at")
        last_scan_at = self._parse_iso_dt(offset_raw)
        now = _utc_now_naive()
        if last_scan_at is not None and now - last_scan_at < timedelta(hours=hours):
            return {
                "executed": False,
                "reason": "not_due",
                "last_scan_at": last_scan_at.isoformat(),
                "next_due_at": (last_scan_at + timedelta(hours=hours)).isoformat(),
            }

        symbol_service = SymbolIndexService(self.db, tenant_id=self.tenant_id)
        payload = await symbol_service.sync_codebase_symbols(
            base_dir=base_dir,
            max_files=max_files,
            force_reindex=False,
        )
        symbol_graph = await self._sync_symbol_entities(limit=800)
        payload["kg_symbol_sync"] = symbol_graph
        await self.enqueue_ingest_event(
            channel="symbol_scan",
            event_key=f"symbol_scan:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
            payload=payload,
            status="done",
        )
        await self.set_ingest_offset(
            channel="symbol_scan",
            offset_key="last_symbol_scan_at",
            offset_value=_utc_now_naive().isoformat(),
            payload=payload,
        )
        await self.db.commit()
        return {
            "executed": True,
            "reason": "due",
            "payload": payload,
        }

    async def _sync_symbol_entities(self, *, limit: int = 800) -> dict[str, Any]:
        max_rows = max(10, min(int(limit), 3000))
        defs = (
            await self.db.execute(
                select(SymbolDefinition)
                .where(SymbolDefinition.tenant_id == self.tenant_id)
                .order_by(SymbolDefinition.updated_at.desc(), SymbolDefinition.id.desc())
                .limit(max_rows)
            )
        ).scalars().all()
        if not defs:
            return {"symbols": 0, "modules": 0, "relations": 0, "evidences": 0}

        counters = {"symbols": 0, "modules": 0, "relations": 0, "evidences": 0}
        module_entity_map: dict[str, int] = {}
        symbol_entity_map: dict[int, int] = {}
        name_entity_map: dict[str, int] = {}
        qn_entity_map: dict[str, int] = {}

        for row in defs:
            qn = str(row.qualified_name or row.symbol_name or f"symbol#{int(row.id)}")
            symbol_entity = await self.upsert_entity(
                entity_type="symbol",
                entity_key=f"symbol:{int(row.id)}",
                display_name=qn[:255],
                payload={
                    "symbol_id": int(row.id),
                    "symbol_name": str(row.symbol_name or ""),
                    "qualified_name": str(row.qualified_name or ""),
                    "symbol_type": str(row.symbol_type or ""),
                    "language": str(row.language or ""),
                    "file_path": str(row.file_path or ""),
                    "start_line": int(row.start_line) if row.start_line is not None else None,
                    "end_line": int(row.end_line) if row.end_line is not None else None,
                },
                source_type="symbol_index",
                source_ref=str(row.file_path or "")[:255],
            )
            counters["symbols"] += 1
            symbol_entity_map[int(row.id)] = int(symbol_entity.id)
            if str(row.symbol_name or "").strip():
                name_entity_map[str(row.symbol_name).strip()] = int(symbol_entity.id)
            if str(row.qualified_name or "").strip():
                qn_entity_map[str(row.qualified_name).strip()] = int(symbol_entity.id)

            file_key = str(row.file_path or "").strip()
            if not file_key:
                continue
            module_entity_id = module_entity_map.get(file_key)
            if module_entity_id is None:
                module_entity = await self.upsert_entity(
                    entity_type="module",
                    entity_key=f"module:{file_key}",
                    display_name=file_key[:255],
                    payload={"file_path": file_key},
                    source_type="symbol_index",
                    source_ref=file_key[:255],
                )
                module_entity_id = int(module_entity.id)
                module_entity_map[file_key] = module_entity_id
                counters["modules"] += 1
            rel = await self.upsert_relation(
                relation_type="defined_in",
                subject_entity_id=int(symbol_entity.id),
                object_entity_id=int(module_entity_id),
                source_type="symbol_index",
                source_ref=file_key[:255],
                payload={
                    "symbol_id": int(row.id),
                    "symbol_type": str(row.symbol_type or ""),
                    "language": str(row.language or ""),
                },
            )
            counters["relations"] += 1
            await self.add_evidence(
                evidence_type="symbol_index",
                content_excerpt=(
                    f"{str(row.symbol_name or '')} defined_in {file_key}:"
                    f"{int(row.start_line) if row.start_line is not None else 0}"
                ),
                relation_id=int(rel.id),
                source_type="symbol_index",
                source_ref=file_key[:255],
                metadata={"symbol_id": int(row.id), "file_path": file_key},
            )
            counters["evidences"] += 1

        refs = (
            await self.db.execute(
                select(SymbolReference)
                .where(SymbolReference.tenant_id == self.tenant_id)
                .order_by(SymbolReference.updated_at.desc(), SymbolReference.id.desc())
                .limit(max_rows * 3)
            )
        ).scalars().all()
        for row in refs:
            source_entity_id = None
            if row.source_symbol_id is not None:
                source_entity_id = symbol_entity_map.get(int(row.source_symbol_id))
            if source_entity_id is None:
                continue
            target_entity_id = None
            target_qn = str(row.target_qualified_name or "").strip()
            target_name = str(row.target_symbol_name or "").strip()
            if target_qn:
                target_entity_id = qn_entity_map.get(target_qn)
            if target_entity_id is None and target_name:
                target_entity_id = name_entity_map.get(target_name)
            if target_entity_id is None:
                continue
            rel_type = str(row.relation_type or "references").strip()
            rel = await self.upsert_relation(
                relation_type=rel_type[:40],
                subject_entity_id=int(source_entity_id),
                object_entity_id=int(target_entity_id),
                source_type="symbol_index",
                source_ref=str(row.file_path or "")[:255],
                payload={
                    "symbol_reference_id": int(row.id),
                    "target_symbol_name": target_name,
                    "target_qualified_name": target_qn,
                },
            )
            counters["relations"] += 1
            await self.add_evidence(
                evidence_type="symbol_reference",
                content_excerpt=f"{int(source_entity_id)} {rel_type} {target_name or target_qn}",
                relation_id=int(rel.id),
                source_type="symbol_index",
                source_ref=str(row.file_path or "")[:255],
                metadata={
                    "symbol_reference_id": int(row.id),
                    "target_symbol_name": target_name,
                    "target_qualified_name": target_qn,
                },
            )
            counters["evidences"] += 1
        return counters

    @staticmethod
    def _routing_rule_candidate_paths() -> list[Path]:
        candidates: list[Path] = []
        configured = str(os.getenv("OPENCLAW_ROUTING_RULES_PATH", "") or "").strip()
        if configured:
            candidates.append(Path(configured).expanduser())

        backend_root = Path(__file__).resolve().parents[2]
        project_root = Path(__file__).resolve().parents[3]
        candidates.extend(
            [
                project_root / ".openclaw" / "pmflow-routing-rules.json",
                backend_root / ".openclaw" / "pmflow-routing-rules.json",
            ]
        )

        deduped: list[Path] = []
        seen: set[str] = set()
        for path in candidates:
            key = str(path)
            if key in seen:
                continue
            seen.add(key)
            deduped.append(path)
        return deduped

    @classmethod
    def _resolve_guardrail_target_models(cls) -> dict[str, dict[str, str]]:
        specs: dict[str, dict[str, str]] = {
            "low": {
                "tier": "low",
                "provider": str(os.getenv("LLM_ROUTING_LOW_PROVIDER", "glm") or "glm").strip(),
                "model": str(os.getenv("LLM_ROUTING_LOW_MODEL", "GLM-4.7") or "GLM-4.7").strip(),
                "api_env": str(os.getenv("LLM_ROUTING_LOW_API_ENV", "GLM_API_KEY") or "GLM_API_KEY").strip(),
                "source": "env_default",
                "source_path": "",
                "rule_name": "",
            },
            "high": {
                "tier": "high",
                "provider": str(os.getenv("LLM_ROUTING_HIGH_PROVIDER", "openai") or "openai").strip(),
                "model": str(os.getenv("LLM_ROUTING_HIGH_MODEL", "MiniMax-M2.5") or "MiniMax-M2.5").strip(),
                "api_env": str(os.getenv("LLM_ROUTING_HIGH_API_ENV", "OPENAI_API_KEY") or "OPENAI_API_KEY").strip(),
                "source": "env_default",
                "source_path": "",
                "rule_name": "",
            },
        }

        for path in cls._routing_rule_candidate_paths():
            try:
                content = path.read_text(encoding="utf-8")
                payload = json.loads(content)
            except Exception:
                continue
            rows = payload.get("rules") if isinstance(payload, dict) and isinstance(payload.get("rules"), list) else []
            if not rows:
                continue
            enabled_rows = [row for row in rows if isinstance(row, dict) and bool(row.get("enabled", True))]
            enabled_rows.sort(key=lambda item: int(item.get("priority") or 0), reverse=True)
            resolved_any = False
            for tier in ("low", "high"):
                for row in enabled_rows:
                    if str(row.get("tier") or "").strip().lower() != tier:
                        continue
                    model = str(row.get("model") or "").strip()
                    if not model:
                        continue
                    specs[tier] = {
                        "tier": tier,
                        "provider": str(row.get("provider") or specs[tier]["provider"]).strip(),
                        "model": model,
                        "api_env": str(row.get("api_env") or specs[tier]["api_env"]).strip(),
                        "source": "routing_rule",
                        "source_path": str(path),
                        "rule_name": str(row.get("name") or "").strip(),
                    }
                    resolved_any = True
                    break
            if resolved_any and all(str(specs[tier].get("source") or "") == "routing_rule" for tier in ("low", "high")):
                break
        return specs

    async def _sync_model_guardrail_entities(self) -> dict[str, Any]:
        counters = {"models": 0, "policies": 0, "interfaces": 0, "relations": 0, "evidences": 0}
        model_specs = self._resolve_guardrail_target_models()

        policy_entity = await self.upsert_entity(
            entity_type="governance_policy",
            entity_key=DDL_GUARDRAIL_POLICY_KEY,
            display_name=DDL_GUARDRAIL_POLICY_NAME,
            payload={
                "policy_code": "ddl_via_interface_only",
                "enforcement_level": "hard_block",
                "blocked_sql_prefixes": DDL_BLOCKED_PREFIXES,
                "required_execution_path": "api_or_migration_interface",
                "enforced_by": [
                    "backend/app/database.py",
                    "backend/app/services/llm.py",
                    "backend/app/config/ai_prompts.py",
                ],
            },
            source_type="governance",
            source_ref="backend/app/database.py",
        )
        counters["policies"] += 1
        await self.add_evidence(
            evidence_type="policy_guardrail",
            content_excerpt="Runtime DDL is blocked; model planning must route DDL via interface/migration only.",
            entity_id=int(policy_entity.id),
            source_type="governance",
            source_ref="backend/app/database.py",
            metadata={"blocked_sql_prefixes": DDL_BLOCKED_PREFIXES},
        )
        counters["evidences"] += 1

        interface_specs = [
            {
                "entity_key": "interface:api:/ai-steward/command/dispatch",
                "display_name": "POST /ai-steward/command/dispatch",
                "payload": {"path": "/ai-steward/command/dispatch", "kind": "api"},
                "source_ref": "backend/app/api/ai_steward.py",
            },
            {
                "entity_key": "interface:migration:alembic",
                "display_name": "Alembic Migration Interface",
                "payload": {"path": "alembic", "kind": "migration"},
                "source_ref": "backend/alembic/versions",
            },
        ]
        interface_ids: list[int] = []
        for item in interface_specs:
            interface_entity = await self.upsert_entity(
                entity_type="interface",
                entity_key=str(item["entity_key"]),
                display_name=str(item["display_name"]),
                payload=dict(item["payload"]),
                source_type="governance",
                source_ref=str(item["source_ref"]),
            )
            counters["interfaces"] += 1
            interface_ids.append(int(interface_entity.id))
            rel = await self.upsert_relation(
                relation_type="policy_enforced_by_interface",
                subject_entity_id=int(policy_entity.id),
                object_entity_id=int(interface_entity.id),
                source_type="governance",
                source_ref=str(item["source_ref"]),
            )
            counters["relations"] += 1
            await self.add_evidence(
                evidence_type="policy_guardrail",
                content_excerpt=f"{DDL_GUARDRAIL_POLICY_NAME} -> {item['display_name']}",
                relation_id=int(rel.id),
                source_type="governance",
                source_ref=str(item["source_ref"]),
                metadata=dict(item["payload"]),
            )
            counters["evidences"] += 1

        for tier in ("low", "high"):
            spec = model_specs.get(tier) if isinstance(model_specs.get(tier), dict) else {}
            model_name = str(spec.get("model") or "").strip() or ("GLM-4.7" if tier == "low" else "MiniMax-M2.5")
            provider = str(spec.get("provider") or "").strip()
            api_env = str(spec.get("api_env") or "").strip()
            source = str(spec.get("source") or "env_default").strip()
            source_ref = str(spec.get("source_path") or spec.get("rule_name") or "llm_routing_env").strip()[:255]
            entity = await self.upsert_entity(
                entity_type="ai_model",
                entity_key=f"ai_model:{tier}:{model_name.lower()[:160]}",
                display_name=model_name[:255],
                payload={
                    "tier": tier,
                    "provider": provider,
                    "model": model_name,
                    "api_env": api_env,
                    "source": source,
                    "routing_rule": str(spec.get("rule_name") or "").strip(),
                },
                source_type="model_routing",
                source_ref=source_ref or None,
            )
            counters["models"] += 1
            rel = await self.upsert_relation(
                relation_type="model_must_follow_policy",
                subject_entity_id=int(entity.id),
                object_entity_id=int(policy_entity.id),
                source_type="governance",
                source_ref=source_ref or None,
                payload={"tier": tier, "policy_code": "ddl_via_interface_only"},
            )
            counters["relations"] += 1
            await self.add_evidence(
                evidence_type="policy_guardrail",
                content_excerpt=f"{tier} model {model_name} must follow {DDL_GUARDRAIL_POLICY_KEY}",
                relation_id=int(rel.id),
                source_type="governance",
                source_ref=source_ref or None,
                metadata={"tier": tier, "provider": provider, "model": model_name},
            )
            counters["evidences"] += 1

            for interface_id in interface_ids:
                rel = await self.upsert_relation(
                    relation_type="model_executes_via_interface",
                    subject_entity_id=int(entity.id),
                    object_entity_id=int(interface_id),
                    source_type="governance",
                    source_ref=source_ref or None,
                    payload={"tier": tier},
                )
                counters["relations"] += 1
                await self.add_evidence(
                    evidence_type="policy_guardrail",
                    content_excerpt=f"{tier} model {model_name} executes via controlled interface",
                    relation_id=int(rel.id),
                    source_type="governance",
                    source_ref=source_ref or None,
                    metadata={"tier": tier},
                )
                counters["evidences"] += 1

        return counters

    async def sync_model_guardrails_if_due(
        self,
        *,
        interval_hours: int = 6,
    ) -> dict[str, Any]:
        hours = max(1, int(interval_hours))
        offset_raw = await self.get_ingest_offset(channel="model_guardrail", offset_key="last_model_guardrail_sync_at")
        last_sync_at = self._parse_iso_dt(offset_raw)
        now = _utc_now_naive()
        if last_sync_at is not None and now - last_sync_at < timedelta(hours=hours):
            return {
                "executed": False,
                "reason": "not_due",
                "last_sync_at": last_sync_at.isoformat(),
                "next_due_at": (last_sync_at + timedelta(hours=hours)).isoformat(),
            }

        payload = await self._sync_model_guardrail_entities()
        await self.enqueue_ingest_event(
            channel="model_guardrail",
            event_key=f"model_guardrail:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
            payload=payload,
            status="done",
        )
        await self.set_ingest_offset(
            channel="model_guardrail",
            offset_key="last_model_guardrail_sync_at",
            offset_value=_utc_now_naive().isoformat(),
            payload=payload,
        )
        await self.db.commit()
        return {
            "executed": True,
            "reason": "due",
            "payload": payload,
        }

    async def sync_model_guardrails_now(self) -> dict[str, Any]:
        payload = await self._sync_model_guardrail_entities()
        await self.enqueue_ingest_event(
            channel="model_guardrail",
            event_key=f"model_guardrail:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
            payload=payload,
            status="done",
        )
        await self.set_ingest_offset(
            channel="model_guardrail",
            offset_key="last_model_guardrail_sync_at",
            offset_value=_utc_now_naive().isoformat(),
            payload=payload,
        )
        await self.db.commit()
        return payload

    async def sync_feature_catalog_if_due(
        self,
        *,
        interval_hours: int = 6,
    ) -> dict[str, Any]:
        hours = max(1, int(interval_hours))
        offset_raw = await self.get_ingest_offset(channel="feature_catalog", offset_key="last_feature_catalog_sync_at")
        last_sync_at = self._parse_iso_dt(offset_raw)
        now = _utc_now_naive()
        if last_sync_at is not None and now - last_sync_at < timedelta(hours=hours):
            return {
                "executed": False,
                "reason": "not_due",
                "last_sync_at": last_sync_at.isoformat(),
                "next_due_at": (last_sync_at + timedelta(hours=hours)).isoformat(),
            }

        payload = await self._sync_feature_catalog_entities(limit=300)
        await self.enqueue_ingest_event(
            channel="feature_catalog",
            event_key=f"feature_catalog:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
            payload=payload,
            status="done",
        )
        await self.set_ingest_offset(
            channel="feature_catalog",
            offset_key="last_feature_catalog_sync_at",
            offset_value=_utc_now_naive().isoformat(),
            payload=payload,
        )
        await self.db.commit()
        return {
            "executed": True,
            "reason": "due",
            "payload": payload,
        }

    async def sync_feature_catalog_now(self) -> dict[str, Any]:
        payload = await self._sync_feature_catalog_entities(limit=300)
        await self.enqueue_ingest_event(
            channel="feature_catalog",
            event_key=f"feature_catalog:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
            payload=payload,
            status="done",
        )
        await self.set_ingest_offset(
            channel="feature_catalog",
            offset_key="last_feature_catalog_sync_at",
            offset_value=_utc_now_naive().isoformat(),
            payload=payload,
        )
        await self.db.commit()
        return payload

    async def _sync_feature_catalog_entities(self, *, limit: int = 300) -> dict[str, Any]:
        snapshot = await context_cache_service.get_system_snapshot(
            db=self.db,
            tenant_id=self.tenant_id,
        )
        platform = snapshot.get("platform_profile") if isinstance(snapshot.get("platform_profile"), dict) else {}
        catalog = platform.get("feature_catalog") if isinstance(platform.get("feature_catalog"), list) else []
        rows = [row for row in catalog if isinstance(row, dict)][: max(1, min(int(limit), 1000))]
        if not rows:
            return {"features": 0, "routes": 0, "pages": 0, "apis": 0, "relations": 0, "evidences": 0}

        counters = {"features": 0, "routes": 0, "pages": 0, "apis": 0, "relations": 0, "evidences": 0}
        route_entity_map: dict[str, int] = {}
        page_entity_map: dict[str, int] = {}
        api_entity_map: dict[str, int] = {}

        for row in rows:
            name = str(row.get("name") or "").strip() or "unknown_feature"
            route = str(row.get("route") or "").strip()
            summary = str(row.get("summary") or "").strip()
            page_file = str(row.get("page_file") or "").strip()
            page_component = str(row.get("page_component") or "").strip()
            apis = row.get("apis") if isinstance(row.get("apis"), list) else []
            aliases = row.get("aliases") if isinstance(row.get("aliases"), list) else []

            feature_slug = (route or name).strip().lower()[:200]
            feature_entity = await self.upsert_entity(
                entity_type="feature",
                entity_key=f"feature:{feature_slug}",
                display_name=name[:255],
                payload={
                    "name": name,
                    "route": route,
                    "summary": summary,
                    "page_file": page_file,
                    "page_component": page_component,
                    "apis": [str(x) for x in apis[:20] if str(x).strip()],
                    "aliases": [str(x) for x in aliases[:20] if str(x).strip()],
                },
                source_type="feature_catalog",
                source_ref=(route or page_file or name)[:255],
            )
            counters["features"] += 1
            await self.add_evidence(
                evidence_type="feature_catalog",
                content_excerpt=f"{name} | route={route or 'n/a'} | page={page_file or 'n/a'}",
                entity_id=int(feature_entity.id),
                source_type="feature_catalog",
                source_ref=(route or page_file or name)[:255],
                metadata={"feature_name": name, "route": route, "page_file": page_file},
            )
            counters["evidences"] += 1

            route_entity_id: Optional[int] = None
            if route:
                if route not in route_entity_map:
                    route_entity = await self.upsert_entity(
                        entity_type="route",
                        entity_key=f"route:{route[:200]}",
                        display_name=route[:255],
                        payload={"route": route},
                        source_type="feature_catalog",
                        source_ref=route[:255],
                    )
                    route_entity_map[route] = int(route_entity.id)
                    counters["routes"] += 1
                route_entity_id = route_entity_map[route]
                rel = await self.upsert_relation(
                    relation_type="feature_has_route",
                    subject_entity_id=int(feature_entity.id),
                    object_entity_id=int(route_entity_id),
                    source_type="feature_catalog",
                    source_ref=route[:255],
                )
                counters["relations"] += 1
                await self.add_evidence(
                    evidence_type="feature_catalog",
                    content_excerpt=f"{name} has route {route}",
                    relation_id=int(rel.id),
                    source_type="feature_catalog",
                    source_ref=route[:255],
                    metadata={"feature_name": name, "route": route},
                )
                counters["evidences"] += 1

            page_entity_id: Optional[int] = None
            if page_file:
                if page_file not in page_entity_map:
                    page_entity = await self.upsert_entity(
                        entity_type="page",
                        entity_key=f"page:{page_file[:220]}",
                        display_name=(page_component or page_file)[:255],
                        payload={"page_file": page_file, "page_component": page_component},
                        source_type="feature_catalog",
                        source_ref=page_file[:255],
                    )
                    page_entity_map[page_file] = int(page_entity.id)
                    counters["pages"] += 1
                page_entity_id = page_entity_map[page_file]
                rel = await self.upsert_relation(
                    relation_type="feature_rendered_by",
                    subject_entity_id=int(feature_entity.id),
                    object_entity_id=int(page_entity_id),
                    source_type="feature_catalog",
                    source_ref=page_file[:255],
                )
                counters["relations"] += 1
                await self.add_evidence(
                    evidence_type="feature_catalog",
                    content_excerpt=f"{name} rendered_by {page_file}",
                    relation_id=int(rel.id),
                    source_type="feature_catalog",
                    source_ref=page_file[:255],
                    metadata={"feature_name": name, "page_file": page_file},
                )
                counters["evidences"] += 1

            if route_entity_id is not None and page_entity_id is not None:
                rel = await self.upsert_relation(
                    relation_type="route_to_page",
                    subject_entity_id=int(route_entity_id),
                    object_entity_id=int(page_entity_id),
                    source_type="feature_catalog",
                    source_ref=(route or page_file)[:255],
                )
                counters["relations"] += 1
                await self.add_evidence(
                    evidence_type="feature_catalog",
                    content_excerpt=f"{route} maps_to {page_file}",
                    relation_id=int(rel.id),
                    source_type="feature_catalog",
                    source_ref=(route or page_file)[:255],
                    metadata={"route": route, "page_file": page_file},
                )
                counters["evidences"] += 1

            for raw_api in apis[:20]:
                api_path = str(raw_api or "").strip()
                if not api_path:
                    continue
                if api_path not in api_entity_map:
                    api_entity = await self.upsert_entity(
                        entity_type="api_endpoint",
                        entity_key=f"api:{api_path[:220]}",
                        display_name=api_path[:255],
                        payload={"path": api_path},
                        source_type="feature_catalog",
                        source_ref=api_path[:255],
                    )
                    api_entity_map[api_path] = int(api_entity.id)
                    counters["apis"] += 1
                rel = await self.upsert_relation(
                    relation_type="feature_calls_api",
                    subject_entity_id=int(feature_entity.id),
                    object_entity_id=int(api_entity_map[api_path]),
                    source_type="feature_catalog",
                    source_ref=api_path[:255],
                )
                counters["relations"] += 1
                await self.add_evidence(
                    evidence_type="feature_catalog",
                    content_excerpt=f"{name} calls {api_path}",
                    relation_id=int(rel.id),
                    source_type="feature_catalog",
                    source_ref=api_path[:255],
                    metadata={"feature_name": name, "api_path": api_path},
                )
                counters["evidences"] += 1

        return counters

    async def run_incremental_cycle(
        self,
        *,
        delta_limit: int = 1000,
        event_batch: int = 200,
        code_scan_hours: int = 6,
        feature_scan_hours: int = 6,
        code_scan_base_dir: Optional[str] = None,
        run_source: str = "system",
    ) -> dict[str, Any]:
        started = _utc_now_naive()
        sync_run = await self._open_sync_run(
            run_type="incremental",
            trigger_source=run_source,
            input_payload={
                "delta_limit": int(delta_limit),
                "event_batch": int(event_batch),
                "code_scan_hours": int(code_scan_hours),
                "feature_scan_hours": int(feature_scan_hours),
                "code_scan_base_dir": str(code_scan_base_dir or ""),
            },
        )
        run_id = int(sync_run.id)
        run_started_at = sync_run.started_at or started
        stage_results: dict[str, Any] = {}
        stage_errors: dict[str, str] = {}
        success_count = 0

        try:
            try:
                stage_results["structured_delta"] = await self.sync_structured_delta(limit=delta_limit)
                success_count += 1
            except Exception as exc:
                await self.db.rollback()
                stage_errors["structured_delta"] = str(exc)[:500]
                stage_results["structured_delta"] = {"error": stage_errors["structured_delta"]}

            try:
                stage_results["feishu_enqueue"] = await self.enqueue_feishu_delta_events(limit=min(delta_limit, 1000))
                success_count += 1
            except Exception as exc:
                await self.db.rollback()
                stage_errors["feishu_enqueue"] = str(exc)[:500]
                stage_results["feishu_enqueue"] = {"error": stage_errors["feishu_enqueue"]}

            try:
                stage_results["event_processing"] = await self.process_pending_events(max_batch=event_batch)
                success_count += 1
            except Exception as exc:
                await self.db.rollback()
                stage_errors["event_processing"] = str(exc)[:500]
                stage_results["event_processing"] = {"error": stage_errors["event_processing"]}

            try:
                stage_results["code_scan"] = await self.sync_codebase_if_due(
                    base_dir=code_scan_base_dir,
                    interval_hours=code_scan_hours,
                )
                success_count += 1
            except Exception as exc:
                await self.db.rollback()
                stage_errors["code_scan"] = str(exc)[:500]
                stage_results["code_scan"] = {"error": stage_errors["code_scan"]}

            try:
                stage_results["symbol_index"] = await self.sync_symbol_index_if_due(
                    base_dir=code_scan_base_dir,
                    interval_hours=code_scan_hours,
                )
                success_count += 1
            except Exception as exc:
                await self.db.rollback()
                stage_errors["symbol_index"] = str(exc)[:500]
                stage_results["symbol_index"] = {"error": stage_errors["symbol_index"]}

            try:
                stage_results["feature_catalog"] = await self.sync_feature_catalog_if_due(
                    interval_hours=feature_scan_hours,
                )
                success_count += 1
            except Exception as exc:
                await self.db.rollback()
                stage_errors["feature_catalog"] = str(exc)[:500]
                stage_results["feature_catalog"] = {"error": stage_errors["feature_catalog"]}

            try:
                stage_results["model_guardrails"] = await self.sync_model_guardrails_if_due(
                    interval_hours=feature_scan_hours,
                )
                success_count += 1
            except Exception as exc:
                await self.db.rollback()
                stage_errors["model_guardrails"] = str(exc)[:500]
                stage_results["model_guardrails"] = {"error": stage_errors["model_guardrails"]}

            ended = _utc_now_naive()
            if success_count == 7:
                final_status = "success"
            elif success_count == 0:
                final_status = "failed"
            else:
                final_status = "partial"

            payload = {
                "tenant_id": self.tenant_id,
                "started_at": started.isoformat(),
                "ended_at": ended.isoformat(),
                "run_id": run_id,
                "run_status": final_status,
                "success_stage_count": success_count,
                "failed_stage_count": len(stage_errors),
                "stage_errors": stage_errors,
                **stage_results,
            }
            await self._close_sync_run(
                row=sync_run,
                status=final_status,
                result_payload=payload,
                error_message=("; ".join(f"{k}:{v}" for k, v in stage_errors.items()) if stage_errors else None),
                started_at_hint=run_started_at,
            )
            return payload
        except Exception as exc:
            await self.db.rollback()
            ended = _utc_now_naive()
            payload = {
                "tenant_id": self.tenant_id,
                "started_at": started.isoformat(),
                "ended_at": ended.isoformat(),
                "run_id": run_id,
                "run_status": "failed",
                "success_stage_count": success_count,
                "failed_stage_count": max(1, len(stage_errors)),
                "stage_errors": stage_errors,
                **stage_results,
            }
            await self._close_sync_run(
                row=sync_run,
                status="failed",
                result_payload=payload,
                error_message=str(exc)[:1000],
                started_at_hint=run_started_at,
            )
            return payload

    async def sync_structured_entities(self, *, limit: int = 1000) -> dict[str, Any]:
        max_rows = max(1, min(int(limit), 5000))
        counters = {
            "projects": 0,
            "tasks": 0,
            "signals": 0,
            "users": 0,
            "relations": 0,
            "evidences": 0,
        }

        project_map: dict[int, int] = {}
        projects = (
            await self.db.execute(
                select(Project)
                .where(Project.tenant_id == self.tenant_id)
                .order_by(Project.updated_at.desc(), Project.id.desc())
                .limit(max_rows)
            )
        ).scalars().all()
        for p in projects:
            entity = await self.upsert_entity(
                entity_type="project",
                entity_key=f"project:{int(p.id)}",
                display_name=str(p.title or p.name or f"Project#{int(p.id)}"),
                payload={
                    "project_id": int(p.id),
                    "status": str(p.status or "unknown"),
                    "owner_id": int(p.owner_id) if p.owner_id is not None else None,
                },
                source_type="db",
                source_ref=f"projects:{int(p.id)}",
            )
            project_map[int(p.id)] = int(entity.id)
            counters["projects"] += 1

        user_map: dict[int, int] = {}
        users = (
            await self.db.execute(
                select(User)
                .where(User.tenant_id == self.tenant_id)
                .order_by(User.updated_at.desc(), User.id.desc())
                .limit(max_rows)
            )
        ).scalars().all()
        for u in users:
            entity = await self.upsert_entity(
                entity_type="user",
                entity_key=f"user:{int(u.id)}",
                display_name=str(u.name or f"User#{int(u.id)}"),
                payload={
                    "user_id": int(u.id),
                    "role": str(u.role or "unknown"),
                    "is_active": bool(u.is_active),
                },
                source_type="db",
                source_ref=f"users:{int(u.id)}",
            )
            user_map[int(u.id)] = int(entity.id)
            counters["users"] += 1

        tasks = (
            await self.db.execute(
                select(Task)
                .where(Task.tenant_id == self.tenant_id)
                .order_by(Task.updated_at.desc(), Task.id.desc())
                .limit(max_rows)
            )
        ).scalars().all()
        for t in tasks:
            task_entity = await self.upsert_entity(
                entity_type="task",
                entity_key=f"task:{int(t.id)}",
                display_name=str(t.title or f"Task#{int(t.id)}"),
                payload={
                    "task_id": int(t.id),
                    "project_id": int(t.project_id) if t.project_id is not None else None,
                    "assignee_id": int(t.assignee_id) if t.assignee_id is not None else None,
                    "status": str(t.status or "unknown"),
                },
                source_type="db",
                source_ref=f"tasks:{int(t.id)}",
            )
            counters["tasks"] += 1

            if t.project_id is not None and int(t.project_id) in project_map:
                rel = await self.upsert_relation(
                    relation_type="belongs_to",
                    subject_entity_id=int(task_entity.id),
                    object_entity_id=int(project_map[int(t.project_id)]),
                    source_type="db",
                    source_ref=f"tasks:{int(t.id)}",
                )
                counters["relations"] += 1
                await self.add_evidence(
                    evidence_type="db_row",
                    content_excerpt=f"task#{int(t.id)} belongs_to project#{int(t.project_id)}",
                    relation_id=int(rel.id),
                    source_type="db",
                    source_ref=f"tasks:{int(t.id)}",
                    metadata={"table": "tasks"},
                )
                counters["evidences"] += 1

            if t.assignee_id is not None and int(t.assignee_id) in user_map:
                rel = await self.upsert_relation(
                    relation_type="assigned_to",
                    subject_entity_id=int(task_entity.id),
                    object_entity_id=int(user_map[int(t.assignee_id)]),
                    source_type="db",
                    source_ref=f"tasks:{int(t.id)}",
                )
                counters["relations"] += 1
                await self.add_evidence(
                    evidence_type="db_row",
                    content_excerpt=f"task#{int(t.id)} assigned_to user#{int(t.assignee_id)}",
                    relation_id=int(rel.id),
                    source_type="db",
                    source_ref=f"tasks:{int(t.id)}",
                    metadata={"table": "tasks"},
                )
                counters["evidences"] += 1

        signals = (
            await self.db.execute(
                select(Signal)
                .where(Signal.tenant_id == self.tenant_id)
                .order_by(Signal.updated_at.desc(), Signal.id.desc())
                .limit(max_rows)
            )
        ).scalars().all()
        for s in signals:
            signal_entity = await self.upsert_entity(
                entity_type="signal",
                entity_key=f"signal:{int(s.id)}",
                display_name=str(s.title or s.content or f"Signal#{int(s.id)}")[:255],
                payload={
                    "signal_id": int(s.id),
                    "priority": str(s.priority or ""),
                    "status": str(s.status or ""),
                    "matched_project_id": (
                        int(s.matched_project_id) if getattr(s, "matched_project_id", None) is not None else None
                    ),
                },
                source_type="db",
                source_ref=f"signals:{int(s.id)}",
            )
            counters["signals"] += 1

            target_pid = int(s.matched_project_id) if getattr(s, "matched_project_id", None) is not None else None
            if target_pid is not None and target_pid in project_map:
                rel = await self.upsert_relation(
                    relation_type="matches_project",
                    subject_entity_id=int(signal_entity.id),
                    object_entity_id=int(project_map[target_pid]),
                    source_type="db",
                    source_ref=f"signals:{int(s.id)}",
                )
                counters["relations"] += 1
                await self.add_evidence(
                    evidence_type="db_row",
                    content_excerpt=f"signal#{int(s.id)} matches project#{target_pid}",
                    relation_id=int(rel.id),
                    source_type="db",
                    source_ref=f"signals:{int(s.id)}",
                    metadata={"table": "signals"},
                )
                counters["evidences"] += 1

        counters["model_guardrails"] = await self._sync_model_guardrail_entities()

        await self.enqueue_ingest_event(
            channel="db_delta",
            event_key=f"structured_sync:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
            payload={"limit": max_rows, "counters": counters},
            status="done",
        )
        await self.set_ingest_offset(
            channel="db_delta",
            offset_key="last_structured_sync_at",
            offset_value=_utc_now_naive().isoformat(),
            payload={"counters": counters},
        )

        await self.db.commit()
        return counters

    async def sync_codebase_snapshot(self, *, base_dir: Optional[str] = None) -> dict[str, Any]:
        root = Path(base_dir or "/app/backend").resolve()
        if not root.exists() or not root.is_dir():
            raise ValueError(f"codebase path not found: {root}")

        py_files = [p for p in root.rglob("*.py") if p.is_file()]
        route_markers = ("@router.get(", "@router.post(", "@router.put(", "@router.patch(", "@router.delete(")
        route_count = 0
        service_file_count = 0
        model_file_count = 0

        for py_file in py_files:
            rel = str(py_file.relative_to(root))
            if rel.startswith("app/services/"):
                service_file_count += 1
            if rel.startswith("app/models/"):
                model_file_count += 1
            try:
                text = py_file.read_text(encoding="utf-8", errors="ignore")
            except Exception:
                continue
            for marker in route_markers:
                route_count += text.count(marker)

        payload = {
            "root": str(root),
            "python_files": len(py_files),
            "service_files": service_file_count,
            "model_files": model_file_count,
            "api_route_markers": route_count,
            "scanned_at": _utc_now_naive().isoformat(),
        }
        entity = await self.upsert_entity(
            entity_type="service",
            entity_key="service:backend_codebase",
            display_name="Backend Codebase Snapshot",
            payload=payload,
            source_type="code_scan",
            source_ref="filesystem",
        )
        await self.add_evidence(
            evidence_type="code_scan",
            content_excerpt=f"python_files={len(py_files)}, routes={route_count}",
            entity_id=int(entity.id),
            source_type="code_scan",
            source_ref=str(root),
            metadata=payload,
        )
        await self.enqueue_ingest_event(
            channel="code_scan",
            event_key=f"code_scan:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}",
            payload=payload,
            status="done",
        )
        await self.set_ingest_offset(
            channel="code_scan",
            offset_key="last_code_scan_at",
            offset_value=_utc_now_naive().isoformat(),
            payload=payload,
        )
        await self.db.commit()
        return payload

    async def get_overview(self) -> dict[str, Any]:
        entity_total = (
            await self.db.execute(select(func.count(KGEntity.id)).where(KGEntity.tenant_id == self.tenant_id))
        ).scalar_one()
        relation_total = (
            await self.db.execute(select(func.count(KGRelation.id)).where(KGRelation.tenant_id == self.tenant_id))
        ).scalar_one()
        evidence_total = (
            await self.db.execute(select(func.count(KGEvidence.id)).where(KGEvidence.tenant_id == self.tenant_id))
        ).scalar_one()
        pending_events = (
            await self.db.execute(
                select(func.count(IngestEvent.id)).where(
                    and_(
                        IngestEvent.tenant_id == self.tenant_id,
                        IngestEvent.status.in_(["pending", "retry"]),
                    )
                )
            )
        ).scalar_one()
        latest_run = (
            await self.db.execute(
                select(KGSyncRun)
                .where(KGSyncRun.tenant_id == self.tenant_id)
                .order_by(KGSyncRun.started_at.desc(), KGSyncRun.id.desc())
                .limit(1)
            )
        ).scalar_one_or_none()

        entity_groups = (
            await self.db.execute(
                select(KGEntity.entity_type, func.count(KGEntity.id))
                .where(KGEntity.tenant_id == self.tenant_id)
                .group_by(KGEntity.entity_type)
            )
        ).all()
        relation_groups = (
            await self.db.execute(
                select(KGRelation.relation_type, func.count(KGRelation.id))
                .where(KGRelation.tenant_id == self.tenant_id)
                .group_by(KGRelation.relation_type)
            )
        ).all()

        return {
            "tenant_id": self.tenant_id,
            "entity_total": int(entity_total or 0),
            "relation_total": int(relation_total or 0),
            "evidence_total": int(evidence_total or 0),
            "pending_events": int(pending_events or 0),
            "entities_by_type": {str(k): int(v or 0) for k, v in entity_groups},
            "relations_by_type": {str(k): int(v or 0) for k, v in relation_groups},
            "latest_sync_run": (
                {
                    "id": int(latest_run.id),
                    "run_type": str(latest_run.run_type or ""),
                    "trigger_source": str(latest_run.trigger_source or ""),
                    "status": str(latest_run.status or ""),
                    "started_at": latest_run.started_at.isoformat() if latest_run.started_at else None,
                    "ended_at": latest_run.ended_at.isoformat() if latest_run.ended_at else None,
                    "duration_ms": int(latest_run.duration_ms or 0),
                }
                if latest_run is not None
                else None
            ),
            "snapshot_at": _utc_now_naive().isoformat(),
        }