PMFlow/backend/app/services/signal_capture.py

"""
Proactive Signal Capture Service - 主动式信号捕获服务

核心功能：
1. 飞书群聊监控 - 自动捕获群内讨论的产品/技术相关内容
2. 浏览器标签监控 - 检测用户访问的技术文档、竞品页面
3. GitHub 仓库监控 - 跟踪上游仓库更新、竞品开源项目
4. HackerNews 自动抓取 - 定时抓取技术前沿信息

每个信号都会自动：
- 评估战略重要性
- 提取关键技术栈
- 生成任务建议
- 推送给相关角色
"""
import asyncio
import inspect
import json
import re
from uuid import uuid4
from datetime import datetime, timedelta, timezone
from typing import Any, Dict, List, Optional
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.future import select
from sqlalchemy import and_, text

from ..core.logging import get_logger
from ..database import AsyncSessionLocal
from ..models import (
    Signal, Project, ProjectDomain, User,
    ProjectSignalLink, Task, TruthNode,
    SkillTag, UserProject, UserWorkload, UserSkillBinding,
)
from ..config.ai_prompts import (
    PROMPT_FEISHU_PROJECT_SIGNAL_CLASSIFICATION,
    PROMPT_FEISHU_TASK_ORCHESTRATION,
)
from .llm import LLMService
from .notification_service import NotificationService
from . import workload_service
from .adaptive_rag_service import AdaptiveRAGService
from .context_cache import context_cache_service
from .signal_source_orchestration_service import SignalSourceOrchestrationService
from .ai_pmo_kernel_service import AIPMOKernelService
from ..utils.project_members import ensure_project_default_members
from ..utils.sequence import ensure_project_id_sequence, ensure_signal_id_sequence


EMOJI_UNICODE_RE = re.compile(
    r"[\U0001F300-\U0001F6FF\U0001F700-\U0001F77F\U0001F780-\U0001F7FF"
    r"\U0001F800-\U0001F8FF\U0001F900-\U0001F9FF\U0001FA00-\U0001FAFF"
    r"\U00002600-\U000026FF\U00002700-\U000027BF]+"
)
ONLY_PUNCT_OR_SPACE_RE = re.compile(r"^[\s\W_]+$", re.UNICODE)
MENTION_TAG_RE = re.compile(r"</?at[^>]*>", re.IGNORECASE)
CONTROL_CHAR_RE = re.compile(r"[\x00-\x08\x0B\x0C\x0E-\x1F\x7F]+")
NON_REQUIREMENT_EXACT = {
    "收到",
    "收到啦",
    "好的",
    "ok",
    "okk",
    "okay",
    "yes",
    "no",
    "1",
    "2",
    "3",
    "thanks",
    "thx",
    "赞",
    "辛苦",
    "你是谁",
    "你是誰",
    "你是谁？",
    "你是誰？",
    "说话",
    "在吗",
    "在不在",
    "今天什么日期",
    "今天几号",
    "今天星期几",
    "现在几点",
    "当前日期",
}
NON_REQUIREMENT_CONTAINS = [
    "你是谁",
    "什么身份",
    "身份确认",
    "今天什么日期",
    "当前日期",
    "日期疑问",
    "询问日期",
    "现在几点",
    "几点了",
    "今天周几",
    "今天星期几",
    "说话",
    "在吗",
    "在不在",
]
REQUIREMENT_HINTS = [
    "需求",
    "优化",
    "改一下",
    "调整",
    "修复",
    "bug",
    "故障",
    "报错",
    "项目",
    "页面",
    "接口",
    "功能",
    "排版",
    "样式",
    "任务",
    "立项",
    "上线",
    "配置",
]

logger = get_logger(__name__)
AI_THINK_STAGE_ZH = {
    "capture_start": "开始处理飞书消息",
    "capture_skip_empty": "消息为空，忽略",
    "capture_skip_noise": "识别为噪音消息，忽略",
    "capture_done_ai_pmo": "AI PMO kernel 已完成决策闭环",
    "capture_kernel_failed": "AI PMO kernel 执行异常",
    "capture_kernel_failed_rollback_failed": "kernel 异常后回滚失败",
    "classification_done": "语义分类完成",
    "classification_parse_failed": "语义分类结果解析失败",
    "capture_skip_not_requirement": "非需求消息，忽略",
    "capture_skip_non_requirement": "判定为非需求，忽略",
    "signal_created": "信号已创建",
    "project_link_done": "项目关联完成",
    "project_link_failed": "项目关联失败",
    "task_orchestrate_failed": "任务编排失败",
    "capture_done": "飞书消息处理完成",
    "pipeline_post_orchestrate_failed": "分流后置任务编排失败",
    "pipeline_post_update_signal_failed": "分流后置信号结果回写失败",
    "pipeline_post_source_onboard_failed": "分流后置信号源接入编排失败",
    "task_plan_llm_done": "任务编排 LLM 返回成功",
    "task_plan_llm_failed": "任务编排 LLM 调用失败",
    "task_created": "任务已创建",
    "charter_auto_created": "自动创建宪章节点",
    "hn_story_captured": "HackerNews 信号已捕获",
    "hn_monitor_error": "HackerNews 监控异常",
    "new_project_created_from_signal": "已根据信号自动创建项目",
    "draft_new_project_brief_failed": "新项目摘要生成失败，已走兜底",
}


def _build_feishu_rag_conversation_key(
    *,
    source_chat_id: Optional[str],
    source_sender_open_id: Optional[str],
    sender: Optional[str],
) -> str:
    chat = str(source_chat_id or "").strip()
    if not chat:
        return ""
    actor = str(source_sender_open_id or "").strip()
    if not actor:
        actor = str(sender or "").strip().lower()
        actor = re.sub(r"\s+", "", actor)
    actor = actor[:96] if actor else "unknown"
    return f"feishu:{chat}:{actor}"


class ProactiveSignalCaptureService:
    """主动式信号捕获服务"""

    def __init__(self, db: AsyncSession):
        self.db = db
        self.llm = LLMService()

    async def capture_from_feishu(
        self,
        group_chat_id: str,
        message_content: str,
        sender: str,
        tenant_id: int = 1,
        proposer_open_id: Optional[str] = None,
        captured_at: Optional[datetime] = None,
        mentioned_bot_names: Optional[List[str]] = None,
        feishu_message_id: Optional[str] = None,
        chat_type: Optional[str] = None,
    ) -> Dict[str, Any]:
        """
        从飞书群聊捕获信号

        触发条件：
        - 提到产品、竞品、技术关键词
        - 讨论架构、技术选型
        - 分享技术文章
        """
        trace_id = f"feishu-{uuid4().hex[:12]}"
        captured_dt = captured_at or datetime.now(timezone.utc)
        if captured_dt.tzinfo is None:
            captured_dt = captured_dt.replace(tzinfo=timezone.utc)

        proposer_name = self._normalize_single_line_text(sender or "未知用户", max_len=80) or "未知用户"
        proposer_open_id = self._normalize_single_line_text(proposer_open_id or "", max_len=128) or None
        self._emit_ai_think(
            "capture_start",
            trace_id=trace_id,
            tenant_id=tenant_id,
            group_chat_id=group_chat_id,
            sender=proposer_name,
        )

        cleaned_message = self._sanitize_feishu_message(message_content)
        if not cleaned_message:
            self._emit_ai_think("capture_skip_empty", trace_id=trace_id)
            return {"status": "ignored", "reason": "empty_or_emoji_only"}
        if self._is_non_requirement_message(cleaned_message) and not self._is_question_like(cleaned_message):
            self._emit_ai_think(
                "capture_skip_noise",
                trace_id=trace_id,
                message_preview=cleaned_message[:120],
            )
            return {"status": "ignored", "reason": "non_requirement_noise"}

        kernel = AIPMOKernelService(self.db, tenant_id=tenant_id)
        try:
            pipeline_result = await kernel.process_feishu_message(
                group_chat_id=group_chat_id,
                message_content=cleaned_message,
                sender=proposer_name,
                source_message_id=feishu_message_id,
                source_sender_open_id=proposer_open_id,
                captured_at=captured_dt,
                chat_type=chat_type,
            )
            if not bool(pipeline_result.get("handled")):
                self._emit_ai_think(
                    "capture_kernel_failed",
                    trace_id=trace_id,
                    reason=pipeline_result.get("reason"),
                    error=pipeline_result.get("error"),
                )
                return {
                    "status": "error",
                    "reason": str(pipeline_result.get("reason") or "kernel_not_handled"),
                    "classification": "error",
                    "project_scope": "error",
                    "linked_project_ids": [],
                    "signal_id": None,
                    "created_project_id": None,
                    "auto_tasks": [],
                    "ai_trace_id": trace_id,
                    "direct_reply_required": True,
                    "direct_reply_text": "系统处理中出现异常，本次未执行入库。请稍后重试，或补充更明确的目标/对象/动作。",
                }
            self._emit_ai_think(
                "capture_done_ai_pmo",
                trace_id=trace_id,
                classification=pipeline_result.get("classification"),
                linked_project_ids=pipeline_result.get("linked_project_ids", []),
                work_item_id=pipeline_result.get("work_item_id"),
            )
            pipeline_result = await self._apply_pipeline_post_actions(
                trace_id=trace_id,
                tenant_id=tenant_id,
                cleaned_message=cleaned_message,
                proposer_name=proposer_name,
                captured_at=captured_dt,
                pipeline_result=pipeline_result,
            )
            try:
                await self._sync_feishu_recent_cache(
                    tenant_id=tenant_id,
                    group_chat_id=group_chat_id,
                    sender=proposer_name,
                    sender_open_id=proposer_open_id,
                    message=cleaned_message,
                    pipeline_result=pipeline_result,
                )
            except Exception as cache_exc:
                self._emit_ai_think(
                    "capture_recent_cache_sync_failed",
                    trace_id=trace_id,
                    error=str(cache_exc),
                )
            try:
                asyncio.create_task(
                    AdaptiveRAGService.ingest_feishu_message_async(
                        tenant_id=tenant_id,
                        content=cleaned_message,
                        sender=proposer_name,
                        chat_id=group_chat_id,
                        message_id=feishu_message_id,
                        captured_at=captured_dt,
                        project_id=(
                            int(pipeline_result["linked_project_ids"][0])
                            if isinstance(pipeline_result.get("linked_project_ids"), list)
                            and pipeline_result.get("linked_project_ids")
                            else None
                        ),
                    )
                )
            except Exception:
                pass
            return pipeline_result
        except Exception as exc:
            self._emit_ai_think("capture_kernel_failed", trace_id=trace_id, error=str(exc))
            try:
                await self.db.rollback()
            except Exception as rollback_exc:
                self._emit_ai_think(
                    "capture_kernel_failed_rollback_failed",
                    trace_id=trace_id,
                    error=str(rollback_exc),
                )
            return {
                "status": "error",
                "reason": "kernel_failed",
                "classification": "error",
                "project_scope": "error",
                "linked_project_ids": [],
                "signal_id": None,
                "created_project_id": None,
                "auto_tasks": [],
                "ai_trace_id": trace_id,
                "direct_reply_required": True,
                "direct_reply_text": "系统处理中出现异常，本次未执行入库。请稍后重试，或补充更明确的目标/对象/动作。",
            }

        projects_context = await self._load_active_projects_context(
            tenant_id=tenant_id,
        )
        classification = await self._analyze_chat_message(
            message=cleaned_message,
            projects_context=projects_context,
            tenant_id=tenant_id,
        )
        self._emit_ai_think(
            "classification_done",
            trace_id=trace_id,
            classification=classification.get("classification"),
            confidence=classification.get("confidence"),
            matched_project_ids=classification.get("matched_project_ids", []),
            reason=classification.get("reason"),
        )
        if not classification.get("is_requirement", False):
            self._emit_ai_think(
                "capture_skip_not_requirement",
                trace_id=trace_id,
                reason=classification.get("reason", "non_requirement"),
            )
            return {"status": "ignored", "reason": classification.get("reason", "non_requirement")}

        project_scope = str(classification.get("classification") or "pending_classification")
        if project_scope == "non_requirement":
            self._emit_ai_think(
                "capture_skip_non_requirement",
                trace_id=trace_id,
                reason=classification.get("reason", "non_requirement"),
            )
            return {"status": "ignored", "reason": classification.get("reason", "non_requirement")}

        # 创建信号
        await ensure_signal_id_sequence(self.db)
        strategic_payload = {
            "captured_at": captured_dt.isoformat(),
            "source_context": {
                "group_id": group_chat_id,
                "sender": proposer_name,
                "sender_open_id": proposer_open_id,
                "mentioned_bot_names": [
                    self._normalize_single_line_text(n, max_len=80)
                    for n in (mentioned_bot_names or [])
                    if self._normalize_single_line_text(n, max_len=80)
                ],
            },
            "original_message": (message_content or "")[:500],
            "cleaned_message": cleaned_message[:500],
            "ai_analysis": {
                "requirement_project_scope": project_scope,
                "reason": classification.get("reason", ""),
                "confidence": classification.get("confidence", 0.0),
                "matched_project_ids": classification.get("matched_project_ids", []),
                "matched_project_titles": classification.get("matched_project_titles", []),
                "summary": classification.get("summary", ""),
            },
        }

        signal = Signal(
            source="Feishu",
            content=f"[{proposer_name}]: {self._normalize_single_line_text(cleaned_message, max_len=200)}",
            impact_score=float(classification.get("impact_score") or 0.7),
            priority=(classification.get("priority") or "medium"),
            processed=False,
            strategic_analysis=json.dumps(strategic_payload, ensure_ascii=False),
            tasks_generated=json.dumps([], ensure_ascii=False),
            tenant_id=tenant_id
        )
        self.db.add(signal)
        await self.db.commit()
        await self.db.refresh(signal)
        self._emit_ai_think("signal_created", trace_id=trace_id, signal_id=signal.id)

        # 自动关联到相关项目（失败不阻断后续通知）
        linked_project_ids: List[int] = []
        try:
            if project_scope == "new_project":
                new_project_id = await self._auto_create_project_from_new_signal(
                    signal_id=int(signal.id),
                    message=cleaned_message,
                    sender=proposer_name,
                    tenant_id=tenant_id,
                    requested_at=captured_dt,
                )
                linked_project_ids = [new_project_id] if new_project_id else []
            else:
                linked_project_ids = await self._auto_link_to_projects(
                    signal.id,
                    classification,
                    tenant_id,
                    message_content=cleaned_message,
                )
            self._emit_ai_think(
                "project_link_done",
                trace_id=trace_id,
                signal_id=signal.id,
                linked_project_ids=linked_project_ids,
            )
        except Exception as exc:
            self._emit_ai_think("project_link_failed", trace_id=trace_id, error=str(exc))

        auto_tasks: List[Dict[str, Any]] = []
        ai_execution: Dict[str, Any] = {
            "trace_id": trace_id,
            "status": "skipped",
            "reason": "no_linked_project",
            "project_id": None,
            "summary": "",
            "tasks_created": 0,
        }
        if linked_project_ids:
            primary_project_id = int(linked_project_ids[0])
            try:
                auto_tasks, ai_execution = await self._auto_orchestrate_project_tasks(
                    trace_id=trace_id,
                    tenant_id=tenant_id,
                    project_id=primary_project_id,
                    signal_id=int(signal.id),
                    cleaned_message=cleaned_message,
                    classification=classification,
                    proposer_name=proposer_name,
                    captured_at=captured_dt,
                )
            except Exception as exc:
                self._emit_ai_think(
                    "task_orchestrate_failed",
                    trace_id=trace_id,
                    project_id=primary_project_id,
                    error=str(exc),
                )
                ai_execution = {
                    "trace_id": trace_id,
                    "status": "failed",
                    "reason": str(exc),
                    "project_id": primary_project_id,
                    "summary": "",
                    "tasks_created": 0,
                }

        strategic_payload["ai_execution"] = ai_execution
        signal.tasks_generated = json.dumps(auto_tasks, ensure_ascii=False)
        signal.strategic_analysis = json.dumps(strategic_payload, ensure_ascii=False)
        signal.processed = bool(linked_project_ids or auto_tasks)
        await self.db.commit()

        # 站内通知：右上角消息提醒
        target_url = "/ai-steward"
        if linked_project_ids:
            target_url = f"/projects/{linked_project_ids[0]}/board"
        try:
            notification_service = NotificationService(self.db)
            await notification_service.notify_signal_created(
                tenant_id=tenant_id,
                signal_id=int(signal.id),
                signal_content=cleaned_message,
                source="feishu",
                target_url=target_url,
                proposer=proposer_name,
                occurred_at=captured_dt,
            )
        except Exception as exc:
            self._emit_ai_think("notify_failed", trace_id=trace_id, error=str(exc))

        self._emit_ai_think(
            "capture_done",
            trace_id=trace_id,
            signal_id=signal.id,
            linked_project_ids=linked_project_ids,
            auto_tasks=len(auto_tasks),
        )
        return {
            "signal_id": str(signal.id),
            "status": "captured",
            "impact_score": signal.impact_score,
            "auto_tasks": auto_tasks,
            "project_scope": project_scope,
            "classification": project_scope,
            "linked_project_ids": linked_project_ids,
            "created_project_id": linked_project_ids[0] if project_scope == "new_project" and linked_project_ids else None,
            "proposer": proposer_name,
            "ai_trace_id": trace_id,
        }

    async def _sync_feishu_recent_cache(
        self,
        *,
        tenant_id: int,
        group_chat_id: str,
        sender: str,
        sender_open_id: Optional[str],
        message: str,
        pipeline_result: Dict[str, Any],
    ) -> None:
        if not isinstance(pipeline_result, dict):
            return

        conversation_key = _build_feishu_rag_conversation_key(
            source_chat_id=group_chat_id,
            source_sender_open_id=sender_open_id,
            sender=sender,
        )
        external_session_id = AdaptiveRAGService._external_session_id_from_key(conversation_key)
        if external_session_id is None:
            return

        answer_text = str(pipeline_result.get("direct_reply_text") or "").strip()
        if not answer_text:
            answer_text = self._build_recent_cache_ack_text(pipeline_result)
        if not answer_text:
            return

        recent_messages = await context_cache_service.get_session_messages(
            tenant_id=int(tenant_id),
            session_id=int(external_session_id),
        )
        recent_turns = context_cache_service.messages_to_turns(recent_messages)
        if recent_turns:
            latest_turn = recent_turns[-1]
            latest_question = str(latest_turn.get("question") or "").strip()
            latest_answer = str(latest_turn.get("answer") or "").strip()
            if latest_question == str(message or "").strip() and latest_answer == answer_text:
                return

        await context_cache_service.append_session_messages(
            tenant_id=int(tenant_id),
            session_id=int(external_session_id),
            question=str(message or "")[:3000],
            answer=answer_text[:4000],
        )

    @staticmethod
    def _build_recent_cache_ack_text(pipeline_result: Dict[str, Any]) -> str:
        status = str(pipeline_result.get("status") or "").strip().lower()
        if status == "captured":
            return "已收到并记录该需求，已进入信号处理队列。"
        if status == "ignored":
            return "已收到该消息，本次未进入需求处理队列。"
        if status == "error":
            return str(pipeline_result.get("direct_reply_text") or "系统处理中出现异常，请稍后重试。").strip()
        return "已收到该消息。"

    async def _apply_pipeline_post_actions(
        self,
        *,
        trace_id: str,
        tenant_id: int,
        cleaned_message: str,
        proposer_name: str,
        captured_at: datetime,
        pipeline_result: Dict[str, Any],
    ) -> Dict[str, Any]:
        if not isinstance(pipeline_result, dict):
            return pipeline_result
        if str(pipeline_result.get("status") or "").lower() != "captured":
            return pipeline_result

        existing_tasks = pipeline_result.get("auto_tasks")
        if isinstance(existing_tasks, list) and existing_tasks:
            return pipeline_result

        linked_project_ids = pipeline_result.get("linked_project_ids")
        if not isinstance(linked_project_ids, list) or not linked_project_ids:
            return pipeline_result

        signal_id = self._to_int(pipeline_result.get("signal_id"))
        project_id = self._to_int(linked_project_ids[0])
        if signal_id is None or project_id is None:
            return pipeline_result

        signal = (
            await self.db.execute(
                select(Signal).where(
                    and_(
                        Signal.id == signal_id,
                        Signal.tenant_id == tenant_id,
                    )
                )
            )
        ).scalar_one_or_none()
        if signal is None:
            return pipeline_result

        if self._is_question_like(cleaned_message):
            return pipeline_result

        source_orchestration_result: Optional[Dict[str, Any]] = None
        if SignalSourceOrchestrationService.looks_like_signal_source_request(cleaned_message):
            try:
                source_orchestration_result = await SignalSourceOrchestrationService(self.db).orchestrate_from_signal(
                    tenant_id=tenant_id,
                    project_id=project_id,
                    signal_id=signal_id,
                    initiating_text=cleaned_message,
                    requested_via="feishu",
                    operator_user_id=None,
                    dry_run=False,
                )
            except Exception as exc:
                self._emit_ai_think(
                    "pipeline_post_source_onboard_failed",
                    trace_id=trace_id,
                    signal_id=signal_id,
                    project_id=project_id,
                    error=str(exc),
                )
            else:
                orchestration_status = str(source_orchestration_result.get("status") or "").lower()
                if orchestration_status in {"completed", "partial", "planned", "pending_manual"}:
                    auto_tasks = self._build_openclaw_auto_tasks(
                        source_orchestration_result.get("openclaw"),
                    )
                    try:
                        strategic_payload: Dict[str, Any] = {}
                        raw_strategic = signal.strategic_analysis
                        if isinstance(raw_strategic, dict):
                            strategic_payload = dict(raw_strategic)
                        elif isinstance(raw_strategic, str) and raw_strategic.strip():
                            parsed = json.loads(raw_strategic)
                            if isinstance(parsed, dict):
                                strategic_payload = parsed
                        strategic_payload["signal_source_orchestration"] = source_orchestration_result
                        strategic_payload["ai_execution"] = source_orchestration_result
                        signal.strategic_analysis = json.dumps(strategic_payload, ensure_ascii=False)
                        signal.tasks_generated = json.dumps(auto_tasks, ensure_ascii=False)
                        signal.processed = orchestration_status in {"completed", "partial"}
                        await self.db.commit()
                    except Exception as exc:
                        self._emit_ai_think(
                            "pipeline_post_update_signal_failed",
                            trace_id=trace_id,
                            signal_id=signal_id,
                            project_id=project_id,
                            error=str(exc),
                        )
                    pipeline_result["auto_tasks"] = auto_tasks
                    pipeline_result["ai_execution"] = source_orchestration_result
                    pipeline_result["signal_source_orchestration"] = source_orchestration_result
                    return pipeline_result

        signal_type = str(signal.signal_type or "").strip()
        recommended_action = str(signal.recommended_action or "").strip()
        should_orchestrate = (
            recommended_action in {"create_charter", "trigger_alert"}
            or signal_type == "新需求"
        )
        if not should_orchestrate:
            return pipeline_result

        classification_payload: Dict[str, Any] = {
            "classification": str(pipeline_result.get("classification") or "signal"),
            "impact_score": float(signal.impact_score or 0.0),
            "priority": str(signal.priority or "medium"),
            "confidence": float(signal.confidence or 0.0),
            "reason": str(signal.summary or "")[:300],
            "matched_project_ids": [project_id],
            "matched_project_titles": [],
        }

        try:
            auto_tasks, ai_execution = await self._auto_orchestrate_project_tasks(
                trace_id=f"{trace_id}-pipeline",
                tenant_id=tenant_id,
                project_id=project_id,
                signal_id=signal_id,
                cleaned_message=cleaned_message,
                classification=classification_payload,
                proposer_name=proposer_name,
                captured_at=captured_at,
            )
        except Exception as exc:
            self._emit_ai_think(
                "pipeline_post_orchestrate_failed",
                trace_id=trace_id,
                signal_id=signal_id,
                project_id=project_id,
                error=str(exc),
            )
            return pipeline_result

        try:
            strategic_payload: Dict[str, Any] = {}
            raw_strategic = signal.strategic_analysis
            if isinstance(raw_strategic, dict):
                strategic_payload = dict(raw_strategic)
            elif isinstance(raw_strategic, str) and raw_strategic.strip():
                parsed = json.loads(raw_strategic)
                if isinstance(parsed, dict):
                    strategic_payload = parsed
            strategic_payload["ai_execution"] = ai_execution
            signal.strategic_analysis = json.dumps(strategic_payload, ensure_ascii=False)
            signal.tasks_generated = json.dumps(auto_tasks, ensure_ascii=False)
            signal.processed = bool(signal.processed or auto_tasks)
            await self.db.commit()
        except Exception as exc:
            self._emit_ai_think(
                "pipeline_post_update_signal_failed",
                trace_id=trace_id,
                signal_id=signal_id,
                project_id=project_id,
                error=str(exc),
            )

        pipeline_result["auto_tasks"] = auto_tasks
        pipeline_result["ai_execution"] = ai_execution
        return pipeline_result

    @staticmethod
    def _build_openclaw_auto_tasks(openclaw_payload: Any) -> List[Dict[str, Any]]:
        if not isinstance(openclaw_payload, dict):
            return []
        result = openclaw_payload.get("result") if isinstance(openclaw_payload.get("result"), dict) else openclaw_payload
        execution = result.get("execution") if isinstance(result, dict) and isinstance(result.get("execution"), dict) else {}
        rows = execution.get("results") if isinstance(execution.get("results"), list) else []
        items: List[Dict[str, Any]] = []
        for row in rows:
            if not isinstance(row, dict):
                continue
            operation_code = str(row.get("operation_code") or "").strip()
            payload = row.get("payload") if isinstance(row.get("payload"), dict) else {}
            status = str(row.get("status") or "").strip()
            if operation_code not in {"task.create", "task.assign", "task.prioritize", "task.status_transition"}:
                continue
            if status not in {"applied", "planned", "skipped"}:
                continue
            items.append(
                {
                    "operation_code": operation_code,
                    "status": status,
                    "entity_id": row.get("entity_id"),
                    "task_id": payload.get("task_id"),
                    "title": payload.get("title"),
                    "summary": str(row.get("message") or "").strip(),
                }
            )
        return items

    async def capture_from_browser(
        self,
        user_id: int,
        url: str,
        title: str,
        content_preview: str,
        tenant_id: int = 1
    ) -> Dict[str, Any]:
        """
        从浏览器标签页捕获信号

        触发条件：
        - 访问技术文档、API 文档
        - 浏览竞品网站、产品页面
        - 查看 GitHub issues/PRs
        - 阅读技术博客
        """
        # AI 分析页面价值
        analysis = await self._analyze_webpage(
            url,
            title,
            content_preview,
            tenant_id=tenant_id,
        )

        if not analysis['is_valuable']:
            return {'status': 'ignored', 'reason': analysis['reason']}

        # 创建信号
        await ensure_signal_id_sequence(self.db)

        signal = Signal(
            source='Browser',
            content=f"[浏览] {title}: {content_preview[:200]}",
            impact_score=analysis['impact_score'],
            priority=analysis['priority'],
            processed=True,
            strategic_analysis=json.dumps({
                'url': url,
                'captured_at': datetime.now(timezone.utc).isoformat(),
                'user_id': user_id,
                'ai_analysis': analysis.get('analysis', {})
            }),
            tasks_generated=json.dumps([]),
            tenant_id=tenant_id
        )
        self.db.add(signal)
        await self.db.commit()
        await self.db.refresh(signal)

        # 自动关联到相关项目
        await self._auto_link_to_projects(signal.id, analysis, tenant_id)

        return {
            'signal_id': str(signal.id),
            'status': 'captured',
            'impact_score': signal.impact_score
        }

    async def capture_from_github(
        self,
        repo_url: str,
        event_type: str,
        event_data: Dict[str, Any],
        tenant_id: int = 1
    ) -> Dict[str, Any]:
        """
        从 GitHub 事件捕获信号

        监控事件：
        - 竞品仓库更新
        - 上游依赖更新
        - 技术趋势项目发布
        """
        # AI 分析 GitHub 事件
        analysis = await self._analyze_github_event(
            repo_url,
            event_type,
            event_data,
            tenant_id=tenant_id,
        )

        if not analysis['is_valuable']:
            return {'status': 'ignored', 'reason': analysis['reason']}

        # 创建信号
        await ensure_signal_id_sequence(self.db)

        signal = Signal(
            source='GitHub',
            content=f"[{event_type}] {repo_url}: {event_data.get('title', event_data.get('message', ''))[:200]}",
            impact_score=analysis['impact_score'],
            priority=analysis['priority'],
            processed=True,
            strategic_analysis=json.dumps({
                'repo_url': repo_url,
                'event_type': event_type,
                'event_data': event_data,
                'captured_at': datetime.now(timezone.utc).isoformat(),
                'ai_analysis': analysis.get('analysis', {})
            }),
            tasks_generated=json.dumps([]),
            tenant_id=tenant_id
        )
        self.db.add(signal)
        await self.db.commit()
        await self.db.refresh(signal)

        return {
            'signal_id': str(signal.id),
            'status': 'captured',
            'impact_score': signal.impact_score
        }

    async def start_hackernews_monitor(
        self,
        interval_minutes: int = 30,
        tenant_id: int = 1
    ):
        """
        启动 HackerNews 定时监控

        定期抓取 HackerNews 首页，分析技术趋势
        """
        while True:
            try:
                # 抓取 HackerNews 首页
                from .fetcher import fetch_hn_stories
                # 兼容三种实现：
                # 1) async def -> 直接 await
                # 2) sync def -> to_thread 防阻塞
                # 3) sync def 返回 coroutine -> 对返回值再 await
                if inspect.iscoroutinefunction(fetch_hn_stories):
                    stories = await fetch_hn_stories(limit=30)
                else:
                    result = await asyncio.to_thread(fetch_hn_stories, 30)
                    stories = await result if inspect.isawaitable(result) else result

                for story in stories[:10]:  # 只分析前10条
                    # AI 分析每条故事
                    analysis = await self._analyze_hackernews_story(
                        story,
                        tenant_id=tenant_id,
                    )

                    if analysis.get('is_valuable'):
                        # 创建信号
                        await ensure_signal_id_sequence(self.db)

                        signal = Signal(
                            source='HackerNews',
                            content=f"HN: {story.get('title', '')[:200]}",
                            impact_score=analysis['impact_score'],
                            priority=analysis['priority'],
                            processed=True,
                            strategic_analysis=json.dumps({
                                'hn_url': story.get('url'),
                                'hn_points': story.get('points'),
                                'captured_at': datetime.now(timezone.utc).isoformat(),
                                'ai_analysis': analysis.get('analysis', {})
                            }),
                            tasks_generated=json.dumps(analysis.get('suggested_tasks', [])),
                            tenant_id=tenant_id
                        )
                        self.db.add(signal)
                        await self.db.commit()

                        self._emit_ai_think(
                            "hn_story_captured",
                            title=str(story.get("title") or "")[:50],
                        )

                # 等待下次抓取
                await asyncio.sleep(interval_minutes * 60)

            except Exception as e:
                self._emit_ai_think("hn_monitor_error", error=str(e))
                await asyncio.sleep(interval_minutes * 60)

    async def _load_active_projects_context(self, tenant_id: int) -> List[Dict[str, Any]]:
        result = await self.db.execute(
            select(Project).where(
                Project.status == "active",
                Project.tenant_id == tenant_id,
            )
        )
        projects = result.scalars().all()
        rows: List[Dict[str, Any]] = []
        for project in projects:
            rows.append(
                {
                    "id": int(project.id),
                    "title": (project.title or project.name or "").strip(),
                    "description": (project.description or "").strip()[:300],
                    "status": project.status,
                }
            )
        return rows

    def _emit_ai_think(self, stage: str, **payload: Any) -> None:
        safe_payload: Dict[str, Any] = {}
        for key, value in payload.items():
            safe_payload[key] = self._safe_log_value(value)
        logger.info("signal_capture_ai_think", stage=stage, **safe_payload)
        try:
            stage_zh = AI_THINK_STAGE_ZH.get(stage, stage)
            now_text = self._format_cn_time(datetime.now(timezone.utc))
            print(
                f"[{now_text}][AI思考][SignalCapture] {stage_zh} "
                + json.dumps({"stage": stage, **safe_payload}, ensure_ascii=False)
            )
        except Exception:
            pass

    @staticmethod
    def _safe_log_value(value: Any, max_len: int = 600) -> Any:
        if isinstance(value, (str, int, float, bool)) or value is None:
            text = str(value) if value is not None else None
            if text is None:
                return None
            return text if len(text) <= max_len else f"{text[:max_len]}...(truncated)"
        try:
            text = json.dumps(value, ensure_ascii=False, default=str)
        except Exception:
            text = str(value)
        return text if len(text) <= max_len else f"{text[:max_len]}...(truncated)"

    async def _auto_orchestrate_project_tasks(
        self,
        *,
        trace_id: str,
        tenant_id: int,
        project_id: int,
        signal_id: int,
        cleaned_message: str,
        classification: Dict[str, Any],
        proposer_name: str,
        captured_at: datetime,
    ) -> tuple[List[Dict[str, Any]], Dict[str, Any]]:
        project = (
            await self.db.execute(
                select(Project).where(
                    and_(
                        Project.id == project_id,
                        Project.tenant_id == tenant_id,
                    )
                )
            )
        ).scalar_one_or_none()
        if not project:
            return [], {
                "trace_id": trace_id,
                "status": "skipped",
                "reason": "project_not_found",
                "project_id": project_id,
                "summary": "",
                "tasks_created": 0,
            }

        charter_id = await self._ensure_project_charter_node(
            tenant_id=tenant_id,
            project_id=project_id,
            proposer_name=proposer_name,
            cleaned_message=cleaned_message,
            captured_at=captured_at,
        )

        team_members = await self._load_project_team_context(
            tenant_id=tenant_id,
            project_id=project_id,
            owner_id=int(project.owner_id) if project.owner_id else None,
        )
        if not team_members:
            return [], {
                "trace_id": trace_id,
                "status": "skipped",
                "reason": "no_active_members",
                "project_id": project_id,
                "summary": "",
                "tasks_created": 0,
            }

        project_payload = {
            "id": int(project.id),
            "title": (project.title or project.name or "").strip(),
            "description": (project.description or "").strip()[:600],
            "assigned_role": (project.assigned_role or "").strip(),
        }
        signal_payload = {
            "signal_id": signal_id,
            "message": cleaned_message[:1200],
            "classification": classification,
            "proposer": proposer_name,
            "captured_at": captured_at.isoformat(),
        }
        prompt = PROMPT_FEISHU_TASK_ORCHESTRATION.format(
            project_json=json.dumps(project_payload, ensure_ascii=False),
            signal_json=json.dumps(signal_payload, ensure_ascii=False),
            members_json=json.dumps(team_members, ensure_ascii=False),
        )

        llm_status = "success"
        llm_error = ""
        parsed: Dict[str, Any] = {}
        try:
            llm_raw = await self.llm.generate(
                prompt,
                db=self.db,
                tenant_id=tenant_id,
                function_name="feishu_task_orchestration",
            )
            parsed = self._parse_llm_json_response(llm_raw)
            self._emit_ai_think(
                "task_plan_llm_done",
                trace_id=trace_id,
                project_id=project_id,
                llm_summary=parsed.get("overall_summary", ""),
            )
        except Exception as exc:
            llm_status = "failed"
            llm_error = str(exc)
            self._emit_ai_think(
                "task_plan_llm_failed",
                trace_id=trace_id,
                project_id=project_id,
                error=llm_error,
            )

        tasks_payload = parsed.get("tasks") if isinstance(parsed.get("tasks"), list) else []
        if not tasks_payload:
            tasks_payload = self._build_task_orchestration_fallback(
                cleaned_message=cleaned_message,
                classification=classification,
            )

        created_items: List[Dict[str, Any]] = []
        touched_user_ids: set[int] = set()
        created_count = 0
        for idx, raw_task in enumerate(tasks_payload[:8]):
            if not isinstance(raw_task, dict):
                continue
            title = self._normalize_single_line_text(
                str(raw_task.get("title") or raw_task.get("task_name") or ""),
                max_len=64,
            )
            if not title:
                continue
            description = self._normalize_multiline_text(
                str(raw_task.get("description") or cleaned_message),
                max_len=800,
            )
            priority = self._normalize_priority(raw_task.get("priority"))
            estimated_hours = self._normalize_estimated_hours(raw_task.get("estimated_hours"))
            required_skill_names = self._normalize_skill_name_list(raw_task.get("required_skill_names"))
            required_skill_tag_ids = await self._resolve_skill_tag_ids(
                tenant_id=tenant_id,
                skill_names=required_skill_names,
            )
            ai_recommended_assignee_id = self._to_int(raw_task.get("recommended_assignee_id"))
            assignee_id, decision = self._pick_task_assignee(
                team_members=team_members,
                required_skill_tag_ids=required_skill_tag_ids,
                ai_recommended_assignee_id=ai_recommended_assignee_id,
            )

            task = Task(
                tenant_id=tenant_id,
                truth_node_id=charter_id,
                project_id=project_id,
                assignee_id=assignee_id,
                title=title,
                description=description,
                required_skill_tag_ids=required_skill_tag_ids,
                status="todo",
                priority=priority,
                estimated_hours=estimated_hours,
                ai_recommended_assignees={
                    "trace_id": trace_id,
                    "ai_recommended_assignee_id": ai_recommended_assignee_id,
                    "final_assignee_id": assignee_id,
                    "assignee_reason": self._normalize_single_line_text(
                        str(raw_task.get("assignee_reason") or ""),
                        max_len=300,
                    ),
                    "required_skill_names": required_skill_names,
                    "decision": decision,
                    "confidence": float(raw_task.get("confidence") or 0.0),
                },
            )
            self.db.add(task)
            await self.db.flush()
            created_count += 1

            assignee_name = ""
            if assignee_id:
                touched_user_ids.add(int(assignee_id))
                match_member = next((m for m in team_members if int(m.get("user_id", 0)) == int(assignee_id)), None)
                if isinstance(match_member, dict):
                    assignee_name = str(match_member.get("name") or "")

            item = {
                "task_id": int(task.id),
                "task_name": title,
                "priority": priority,
                "estimated_hours": float(task.estimated_hours or 0),
                "required_skill_names": required_skill_names,
                "required_skill_tag_ids": required_skill_tag_ids,
                "assignee_id": int(assignee_id) if assignee_id else None,
                "assignee_name": assignee_name,
                "reason": decision.get("picked_reason", ""),
            }
            created_items.append(item)
            self._emit_ai_think(
                "task_created",
                trace_id=trace_id,
                project_id=project_id,
                rank=idx + 1,
                task_id=task.id,
                task_name=title,
                assignee_id=item["assignee_id"],
                decision=item["reason"],
            )

        await self.db.commit()
        for uid in sorted(touched_user_ids):
            await workload_service.refresh_user_assigned_hours_from_tasks(
                self.db,
                tenant_id=tenant_id,
                user_id=uid,
            )

        summary = self._normalize_single_line_text(
            str(parsed.get("overall_summary") or ""),
            max_len=200,
        )
        if not summary:
            summary = "AI 自动拆解并分配飞书需求任务"

        execution_meta = {
            "trace_id": trace_id,
            "status": "success" if created_count > 0 else "skipped",
            "reason": "ok" if created_count > 0 else "no_valid_tasks",
            "project_id": project_id,
            "charter_id": charter_id,
            "summary": summary,
            "tasks_created": created_count,
            "llm_status": llm_status,
            "llm_error": llm_error[:300] if llm_error else "",
        }
        return created_items, execution_meta

    async def _ensure_project_charter_node(
        self,
        *,
        tenant_id: int,
        project_id: int,
        proposer_name: str,
        cleaned_message: str,
        captured_at: datetime,
    ) -> int:
        existing = (
            await self.db.execute(
                select(TruthNode).where(
                    and_(
                        TruthNode.tenant_id == tenant_id,
                        TruthNode.project_id == project_id,
                        TruthNode.node_type == "charter",
                    )
                ).order_by(TruthNode.updated_at.desc(), TruthNode.id.desc())
            )
        ).scalars().first()
        if existing:
            return int(existing.id)

        project = (
            await self.db.execute(
                select(Project).where(
                    and_(
                        Project.id == project_id,
                        Project.tenant_id == tenant_id,
                    )
                )
            )
        ).scalar_one_or_none()
        if not project:
            raise ValueError("project_not_found_for_charter")

        creator_id = await self._pick_truth_node_creator(
            tenant_id=tenant_id,
            project_id=project_id,
            owner_id=int(project.owner_id) if project.owner_id else None,
        )
        node = TruthNode(
            tenant_id=tenant_id,
            node_type="charter",
            project_id=project_id,
            created_by=creator_id,
            title=self._normalize_single_line_text(
                f"{project.title or project.name or '项目'} 自动宪章",
                max_len=120,
            ),
            content={
                "rationale": "由飞书信号自动触发的初版执行宪章",
                "source": "feishu_auto_orchestration",
                "proposer": proposer_name,
                "captured_at": captured_at.isoformat(),
                "seed_message": cleaned_message[:500],
                "success_criteria": ["建立可执行任务链", "明确负责人", "控制团队负载"],
                "constraints": ["后续可由管理员更新细化"],
            },
            version=1,
            status="active",
        )
        self.db.add(node)
        await self.db.flush()
        self._emit_ai_think(
            "charter_auto_created",
            project_id=project_id,
            charter_id=node.id,
            created_by=creator_id,
        )
        return int(node.id)

    async def _pick_truth_node_creator(
        self,
        *,
        tenant_id: int,
        project_id: int,
        owner_id: Optional[int],
    ) -> int:
        if owner_id:
            owner = (
                await self.db.execute(
                    select(User.id).where(
                        and_(
                            User.tenant_id == tenant_id,
                            User.id == owner_id,
                            User.status == "active",
                        )
                    ).limit(1)
                )
            ).scalars().first()
            if owner:
                return int(owner)

        project_member = (
            await self.db.execute(
                select(UserProject.user_id)
                .join(
                    User,
                    and_(
                        User.id == UserProject.user_id,
                        User.tenant_id == tenant_id,
                        User.status == "active",
                        User.is_active.is_(True),
                    ),
                )
                .where(
                    and_(
                        UserProject.tenant_id == tenant_id,
                        UserProject.project_id == project_id,
                    )
                )
                .order_by(UserProject.user_id.asc())
                .limit(1)
            )
        ).scalars().first()
        if project_member:
            return int(project_member)

        admin_user = (
            await self.db.execute(
                select(User.id).where(
                    and_(
                        User.tenant_id == tenant_id,
                        User.status == "active",
                        User.is_active.is_(True),
                        User.role.in_(["BOSS", "ADMIN"]),
                    )
                ).order_by(User.id.asc()).limit(1)
            )
        ).scalars().first()
        if admin_user:
            return int(admin_user)

        any_active = (
            await self.db.execute(
                select(User.id).where(
                    and_(
                        User.tenant_id == tenant_id,
                        User.status == "active",
                        User.is_active.is_(True),
                    )
                ).order_by(User.id.asc()).limit(1)
            )
        ).scalars().first()
        if any_active:
            return int(any_active)
        raise ValueError("no_active_user_for_charter_creator")

    async def _load_project_team_context(
        self,
        *,
        tenant_id: int,
        project_id: int,
        owner_id: Optional[int],
    ) -> List[Dict[str, Any]]:
        member_ids_result = await self.db.execute(
            select(UserProject.user_id).where(
                and_(
                    UserProject.tenant_id == tenant_id,
                    UserProject.project_id == project_id,
                )
            )
        )
        member_ids = {int(uid) for uid in member_ids_result.scalars().all()}
        if owner_id:
            member_ids.add(int(owner_id))
        if not member_ids:
            manager_result = await self.db.execute(
                select(User.id).where(
                    and_(
                        User.tenant_id == tenant_id,
                        User.status == "active",
                        User.is_active.is_(True),
                        User.role.in_(["BOSS", "ADMIN"]),
                    )
                ).limit(5)
            )
            member_ids = {int(uid) for uid in manager_result.scalars().all()}
        if not member_ids:
            return []

        users = (
            await self.db.execute(
                select(User).where(
                    and_(
                        User.tenant_id == tenant_id,
                        User.id.in_(sorted(member_ids)),
                        User.status == "active",
                        User.is_active.is_(True),
                    )
                )
            )
        ).scalars().all()
        if not users:
            return []

        user_ids = [int(u.id) for u in users]
        workload_rows = (
            await self.db.execute(
                select(UserWorkload).where(
                    and_(
                        UserWorkload.tenant_id == tenant_id,
                        UserWorkload.user_id.in_(user_ids),
                    )
                )
            )
        ).scalars().all()
        workload_map = {int(item.user_id): item for item in workload_rows}

        skill_rows = (
            await self.db.execute(
                select(
                    UserSkillBinding.user_id,
                    SkillTag.id,
                    SkillTag.name,
                    SkillTag.slug,
                    UserSkillBinding.proficiency,
                )
                .join(SkillTag, SkillTag.id == UserSkillBinding.skill_tag_id)
                .where(
                    and_(
                        UserSkillBinding.tenant_id == tenant_id,
                        UserSkillBinding.user_id.in_(user_ids),
                        SkillTag.tenant_id == tenant_id,
                    )
                )
            )
        ).all()

        skill_map: Dict[int, List[Dict[str, Any]]] = {}
        for uid, sid, name, slug, proficiency in skill_rows:
            skill_map.setdefault(int(uid), []).append(
                {
                    "skill_id": int(sid),
                    "name": str(name or "").strip(),
                    "slug": str(slug or "").strip(),
                    "proficiency": int(proficiency or 0),
                }
            )

        members: List[Dict[str, Any]] = []
        for user in users:
            uid = int(user.id)
            workload_row = workload_map.get(uid)
            member_skills = skill_map.get(uid, [])
            members.append(
                {
                    "user_id": uid,
                    "name": str(user.name or user.username or uid),
                    "role": str(user.role or ""),
                    "workload_ratio": round(
                        float(
                            workload_row.workload_ratio
                            if workload_row is not None
                            else float(user.workload or 0.0)
                        ),
                        2,
                    ),
                    "current_assigned_hours": round(
                        float(workload_row.current_assigned_hours or 0.0) if workload_row else 0.0,
                        2,
                    ),
                    "max_available_hours_per_week": round(
                        float(workload_row.max_available_hours_per_week or 40.0) if workload_row else 40.0,
                        2,
                    ),
                    "skill_tag_ids": [int(item["skill_id"]) for item in member_skills],
                    "skill_names": [str(item["name"]) for item in member_skills],
                    "skills": member_skills,
                }
            )
        members.sort(key=lambda x: (float(x.get("workload_ratio") or 0.0), int(x.get("user_id") or 0)))
        return members

    async def _resolve_skill_tag_ids(
        self,
        *,
        tenant_id: int,
        skill_names: List[str],
    ) -> List[int]:
        if not skill_names:
            return []
        result = await self.db.execute(
            select(SkillTag).where(SkillTag.tenant_id == tenant_id)
        )
        tags = result.scalars().all()
        if not tags:
            return []

        exact: Dict[str, int] = {}
        fuzzy: List[tuple[str, int]] = []
        for tag in tags:
            name = str(tag.name or "").strip().lower()
            slug = str(tag.slug or "").strip().lower()
            if name:
                exact[name] = int(tag.id)
                fuzzy.append((name, int(tag.id)))
            if slug:
                exact[slug] = int(tag.id)
                fuzzy.append((slug, int(tag.id)))

        resolved_ids: List[int] = []
        for raw in skill_names:
            key = str(raw or "").strip().lower()
            if not key:
                continue
            if key in exact:
                resolved_ids.append(exact[key])
                continue
            matched = None
            for candidate, sid in fuzzy:
                if key in candidate or candidate in key:
                    matched = sid
                    break
            if matched:
                resolved_ids.append(int(matched))
        return sorted(set(resolved_ids))

    def _pick_task_assignee(
        self,
        *,
        team_members: List[Dict[str, Any]],
        required_skill_tag_ids: List[int],
        ai_recommended_assignee_id: Optional[int],
    ) -> tuple[Optional[int], Dict[str, Any]]:
        if not team_members:
            return None, {"picked_reason": "no_team_members", "candidate_scores": []}

        candidates: List[Dict[str, Any]] = []
        for member in team_members:
            uid = self._to_int(member.get("user_id"))
            if not uid:
                continue
            skill_ids = {int(v) for v in (member.get("skill_tag_ids") or []) if self._to_int(v)}
            workload_ratio = float(member.get("workload_ratio") or 0.0)
            if required_skill_tag_ids:
                matched = sum(1 for sid in required_skill_tag_ids if sid in skill_ids)
                skill_score = matched / len(required_skill_tag_ids)
            else:
                skill_score = 0.4
            workload_score = max(0.0, 140.0 - workload_ratio) / 140.0
            ai_bonus = 0.15 if ai_recommended_assignee_id and uid == int(ai_recommended_assignee_id) else 0.0
            overload_penalty = 0.2 if workload_ratio >= 150 else 0.0
            total_score = (skill_score * 0.7) + (workload_score * 0.3) + ai_bonus - overload_penalty
            candidates.append(
                {
                    "user_id": uid,
                    "name": member.get("name"),
                    "skill_score": round(skill_score, 4),
                    "workload_ratio": round(workload_ratio, 2),
                    "total_score": round(total_score, 4),
                }
            )

        candidates.sort(key=lambda x: (-x["total_score"], x["workload_ratio"], x["user_id"]))
        if not candidates:
            return None, {"picked_reason": "no_valid_candidate", "candidate_scores": []}

        top = candidates[0]
        if float(top.get("workload_ratio") or 0.0) > 160 and float(top.get("skill_score") or 0.0) < 0.5:
            return None, {
                "picked_reason": "all_candidates_overloaded",
                "candidate_scores": candidates[:3],
            }
        picked_uid = int(top["user_id"])
        return picked_uid, {
            "picked_reason": (
                f"score={top['total_score']},skill={top['skill_score']},"
                f"workload={top['workload_ratio']}"
            ),
            "candidate_scores": candidates[:3],
        }

    def _build_task_orchestration_fallback(
        self,
        *,
        cleaned_message: str,
        classification: Dict[str, Any],
    ) -> List[Dict[str, Any]]:
        summary = self._normalize_single_line_text(
            str(classification.get("summary") or cleaned_message),
            max_len=120,
        )
        return [
            {
                "title": "需求澄清与范围确认",
                "description": f"基于飞书信号补全需求边界、验收口径与优先级。上下文：{summary}",
                "priority": "high",
                "estimated_hours": 4,
                "required_skill_names": ["产品管理", "项目管理"],
                "recommended_assignee_id": None,
                "assignee_reason": "fallback_without_llm",
                "confidence": 0.0,
            },
            {
                "title": "核心方案落地",
                "description": f"围绕需求信号推进可交付实现，拆分后端/前端/AI 子任务并排期。上下文：{summary}",
                "priority": "medium",
                "estimated_hours": 12,
                "required_skill_names": ["FastAPI", "React", "LangChain"],
                "recommended_assignee_id": None,
                "assignee_reason": "fallback_without_llm",
                "confidence": 0.0,
            },
        ]

    @staticmethod
    def _normalize_priority(value: Any) -> str:
        priority = str(value or "medium").strip().lower()
        if priority in {"p0", "critical", "high"}:
            return "high"
        if priority in {"p2", "low"}:
            return "low"
        return "medium"

    @staticmethod
    def _normalize_estimated_hours(value: Any) -> Optional[float]:
        try:
            hours = float(value)
        except Exception:
            return 8.0
        if hours <= 0:
            return 1.0
        return round(min(hours, 80.0), 2)

    def _normalize_skill_name_list(self, value: Any) -> List[str]:
        if isinstance(value, list):
            names = [self._normalize_single_line_text(str(item or ""), max_len=60) for item in value]
            return [name for name in names if name]
        if isinstance(value, str):
            split = re.split(r"[,，/|]", value)
            names = [self._normalize_single_line_text(part, max_len=60) for part in split]
            return [name for name in names if name]
        return []

    @staticmethod
    def _to_int(value: Any) -> Optional[int]:
        try:
            if value is None or value == "":
                return None
            return int(value)
        except Exception:
            return None

    @staticmethod
    def _parse_llm_json_response(raw_response: str) -> Dict[str, Any]:
        content = (raw_response or "").strip()
        if not content:
            raise ValueError("empty llm response")
        if content.startswith("```"):
            parts = content.split("```")
            if len(parts) >= 2:
                content = parts[1].strip()
            if content.startswith("json"):
                content = content[4:].strip()
        try:
            parsed = json.loads(content.strip())
            if isinstance(parsed, dict):
                return parsed
        except Exception:
            pass

        # 兼容模型返回“解释文本 + JSON”场景：抽取首个 JSON 对象再解析
        obj_text = ProactiveSignalCaptureService._extract_first_json_object(content)
        if not obj_text:
            raise ValueError("empty llm response")
        parsed = json.loads(obj_text)
        if isinstance(parsed, dict):
            return parsed
        raise ValueError("llm json is not object")

    @staticmethod
    def _extract_first_json_object(text: str) -> Optional[str]:
        raw = str(text or "")
        start = raw.find("{")
        if start < 0:
            return None
        depth = 0
        in_str = False
        escaped = False
        for i in range(start, len(raw)):
            ch = raw[i]
            if in_str:
                if escaped:
                    escaped = False
                elif ch == "\\":
                    escaped = True
                elif ch == '"':
                    in_str = False
                continue
            if ch == '"':
                in_str = True
                continue
            if ch == "{":
                depth += 1
            elif ch == "}":
                depth -= 1
                if depth == 0:
                    return raw[start : i + 1]
        return None

    async def _analyze_chat_message(
        self,
        *,
        message: str,
        projects_context: List[Dict[str, Any]],
        tenant_id: int,
    ) -> Dict[str, Any]:
        """AI 语义分类：existing/new/pending/non_requirement。"""
        prompt = PROMPT_FEISHU_PROJECT_SIGNAL_CLASSIFICATION.format(
            projects_json=json.dumps(projects_context, ensure_ascii=False),
            message=message[:800],
        )

        try:
            response = await self.llm.generate(
                prompt,
                db=self.db,
                tenant_id=tenant_id,
                function_name="feishu_project_signal_classification",
            )
            try:
                parsed = self._parse_llm_json_response(response)
            except Exception as parse_exc:
                self._emit_ai_think(
                    "classification_parse_failed",
                    error=str(parse_exc),
                    response_preview=(response or "")[:500],
                )
                raise
            classification = str(parsed.get("classification") or "pending_classification").lower()
            if classification not in {
                "existing_project",
                "new_project",
                "pending_classification",
                "non_requirement",
            }:
                classification = "pending_classification"

            matched_project_ids: List[int] = []
            for item in parsed.get("matched_project_ids", []) or []:
                try:
                    matched_project_ids.append(int(item))
                except Exception:
                    continue

            matched_titles = [str(x) for x in (parsed.get("matched_project_titles") or []) if str(x).strip()]
            if classification == "existing_project" and not matched_project_ids:
                classification = "pending_classification"
            # 对“明确新建项目意图”进行升级，避免模型过度保守地给 pending。
            if classification == "pending_classification" and self._has_explicit_new_project_intent(message):
                classification = "new_project"
            raw_is_requirement = parsed.get("is_requirement")
            if isinstance(raw_is_requirement, bool):
                is_requirement = raw_is_requirement
            elif isinstance(raw_is_requirement, str):
                is_requirement = raw_is_requirement.strip().lower() in {"1", "true", "yes", "y"}
            else:
                is_requirement = classification != "non_requirement"
            if classification in {"existing_project", "new_project", "pending_classification"}:
                is_requirement = True
            if classification == "non_requirement":
                is_requirement = False
            return {
                "is_requirement": is_requirement,
                "reason": str(parsed.get("reason", "") or "AI 判定"),
                "confidence": float(parsed.get("confidence") or 0.0),
                "classification": classification,
                "matched_project_ids": matched_project_ids,
                "matched_project_titles": matched_titles,
                "impact_score": float(parsed.get("impact_score") or 0.7),
                "priority": str(parsed.get("priority", "medium") or "medium"),
                "summary": str(parsed.get("summary") or "")[:200],
            }
        except Exception as exc:
            self._emit_ai_think("classification_failed", error=str(exc), message_preview=message[:120])
            fallback_new_project = self._has_explicit_new_project_intent(message)
            return {
                "is_requirement": bool(fallback_new_project),
                "reason": "AI 分析失败，启用保守策略",
                "confidence": 0.0,
                "classification": "new_project" if fallback_new_project else "non_requirement",
                "matched_project_ids": [],
                "matched_project_titles": [],
                "impact_score": 0.6,
                "priority": "medium",
                "summary": "",
            }

    @staticmethod
    def _sanitize_feishu_message(message: str) -> str:
        text = ProactiveSignalCaptureService._normalize_single_line_text(message or "", max_len=2000)
        if not text:
            return ""

        text = MENTION_TAG_RE.sub(" ", text)
        text = text.replace("@所有人", " ").replace("@everyone", " ")
        text = re.sub(r"@[_a-zA-Z0-9\-]+", " ", text)
        text = EMOJI_UNICODE_RE.sub(" ", text)
        text = re.sub(r"\s+", " ", text).strip()

        if not text or ONLY_PUNCT_OR_SPACE_RE.match(text):
            return ""
        return text

    @staticmethod
    def _normalize_single_line_text(text: str, *, max_len: int = 500) -> str:
        content = str(text or "")
        if not content:
            return ""
        content = content.replace("\ufffd", "")
        content = CONTROL_CHAR_RE.sub(" ", content)
        content = re.sub(r"\s+", " ", content).strip()
        if max_len > 0:
            content = content[:max_len]
        return content

    @staticmethod
    def _normalize_multiline_text(text: str, *, max_len: int = 2000) -> str:
        content = str(text or "")
        if not content:
            return ""
        content = content.replace("\ufffd", "")
        content = content.replace("\r\n", "\n").replace("\r", "\n")
        content = CONTROL_CHAR_RE.sub(" ", content)
        content = re.sub(r"[ \t]+", " ", content)
        content = re.sub(r"\n{3,}", "\n\n", content).strip()
        if max_len > 0:
            content = content[:max_len]
        return content

    @staticmethod
    def _format_cn_time(dt: datetime) -> str:
        safe_dt = dt
        if safe_dt.tzinfo is None:
            safe_dt = safe_dt.replace(tzinfo=timezone.utc)
        cn_tz = timezone(timedelta(hours=8))
        return safe_dt.astimezone(cn_tz).strftime("%Y-%m-%d %H:%M:%S")

    @staticmethod
    def _is_non_requirement_message(message: str) -> bool:
        content = (message or "").strip().lower()
        if not content:
            return True
        normalized = re.sub(r"\s+", "", content)
        if content in NON_REQUIREMENT_EXACT:
            return True
        if normalized in NON_REQUIREMENT_EXACT:
            return True
        if any(k in normalized for k in NON_REQUIREMENT_CONTAINS) and not any(
            hint in normalized for hint in REQUIREMENT_HINTS
        ):
            return True
        if len(content) <= 2 and content.isdigit():
            return True
        non_requirement_keywords = [
            "收到",
            "辛苦了",
            "赞",
            "哈哈",
            "good",
            "nice",
            "ok",
        ]
        if len(content) <= 8 and any(k in content for k in non_requirement_keywords):
            return True
        if ONLY_PUNCT_OR_SPACE_RE.match(content):
            return True
        return False

    @staticmethod
    def _has_explicit_new_project_intent(message: str) -> bool:
        content = (message or "").strip().lower()
        if not content:
            return False
        explicit_phrases = [
            "新建项目",
            "新增项目",
            "新开项目",
            "独立立项",
            "重新立项",
            "单独立项",
            "开个新项目",
            "做个新项目",
            "从0到1",
        ]
        if any(p in content for p in explicit_phrases):
            return True
        # 兼容常见中文表达：“新增个需求，关于 xxx 的项目”
        if ("新增" in content or "新增个需求" in content) and "项目" in content:
            return True
        return False

    @staticmethod
    def _is_question_like(message: str) -> bool:
        text = str(message or "").strip().lower()
        if not text:
            return False
        if "?" in text or "？" in text:
            return True
        hints = ("吗", "怎么", "如何", "为什么", "谁", "多少", "是否", "有没有", "几号", "几点", "周几")
        return any(k in text for k in hints)

    def _rule_based_chat_signal(self, message: str) -> Optional[Dict[str, Any]]:
        _ = message
        return None

    async def _analyze_webpage(self, url: str, title: str, content: str, tenant_id: int) -> Dict[str, Any]:
        """AI 分析网页价值"""
        prompt = f"""分析以下网页内容是否有价值：

标题：{title}
URL: {url}
内容摘要：{content[:500]}

请判断：
1. 是否是技术文档、竞品信息、行业趋势？
2. 是否值得团队关注？
3. 影响程度（0-1）？

以 JSON 格式返回：
{{
    "is_valuable": boolean,
    "reason": "原因",
    "impact_score": 0.8,
    "priority": "high/medium/low",
    "analysis": {{}},
    "suggested_tasks": [{{"task_name": "任务", "type": "backend", "estimated_hours": 8}}]
}}

如果不是有价值内容，is_valuable 设为 false。"""

        try:
            response = await self.llm.generate(
                prompt,
                db=self.db,
                tenant_id=tenant_id,
            )
            response = response.strip()
            if response.startswith('```'):
                response = response.split('```')[1]
                if response.startswith('json'):
                    response = response[4:]
            return json.loads(response.strip())
        except:
            return {'is_valuable': False, 'reason': 'AI 分析失败'}

    async def _analyze_github_event(self, repo_url: str, event_type: str, event_data: Dict, tenant_id: int) -> Dict[str, Any]:
        """AI 分析 GitHub 事件"""
        description = f"{event_type}: {event_data.get('title', event_data.get('message', ''))}"

        prompt = f"""分析以下 GitHub 事件是否有价值：

仓库：{repo_url}
事件：{description}

请判断：
1. 是否是技术栈更新、安全修复、重要功能？
2. 是否影响我们的项目？
3. 影响程度（0-1）？

以 JSON 格式返回：
{{
    "is_valuable": boolean,
    "reason": "原因",
    "impact_score": 0.8,
    "priority": "high/medium/low",
    "analysis": {{}},
    "suggested_tasks": [{{"task_name": "任务", "type": "backend", "estimated_hours": 8}}]
}}

如果不是有价值内容，is_valuable 设为 false。"""

        try:
            response = await self.llm.generate(
                prompt,
                db=self.db,
                tenant_id=tenant_id,
            )
            response = response.strip()
            if response.startswith('```'):
                response = response.split('```')[1]
                if response.startswith('json'):
                    response = response[4:]
            return json.loads(response.strip())
        except:
            return {'is_valuable': False, 'reason': 'AI 分析失败'}

    async def _analyze_hackernews_story(self, story: Dict, tenant_id: int) -> Dict[str, Any]:
        """AI 分析 HackerNews 故事"""
        prompt = f"""分析以下 HackerNews 故事是否是重要技术趋势：

标题：{story.get('title', '')}
链接：{story.get('url', '')}
得分：{story.get('points', 0)}

请判断：
1. 是否是行业趋势、技术突破？
2. 是否值得我们关注？
3. 影响程度（0-1）？

以 JSON 格式返回：
{{
    "is_strategic": boolean,
    "impact_score": 0.8,
    "priority": "high/medium/low",
    "analysis": {{}},
    "suggested_tasks": [{{"task_name": "任务", "type": "backend", "estimated_hours": 8}}]
}}

如果不是战略性内容，is_strategic 设为 false。"""

        try:
            response = await self.llm.generate(
                prompt,
                db=self.db,
                tenant_id=tenant_id,
            )
            response = response.strip()
            if response.startswith('```'):
                response = response.split('```')[1]
                if response.startswith('json'):
                    response = response[4:]
            result = json.loads(response.strip())
            # 重命名字段以匹配其他方法
            return {
                'is_valuable': result.get('is_strategic', False),
                'reason': result.get('reason', ''),
                'impact_score': result.get('impact_score', 0.5),
                'priority': result.get('priority', 'low'),
                'analysis': result.get('analysis', {}),
                'suggested_tasks': result.get('suggested_tasks', [])
            }
        except:
            return {'is_valuable': False, 'reason': 'AI 分析失败'}

    async def _auto_link_to_projects(
        self,
        signal_id: int,
        analysis: Dict[str, Any],
        tenant_id: int,
        message_content: Optional[str] = None,
    ) -> List[int]:
        """自动关联信号到相关项目"""
        _ = message_content
        classification = str(analysis.get("classification") or "").lower()
        if classification in {"new_project", "pending_classification", "non_requirement"}:
            return []

        if classification == "existing_project":
            matched_ids: List[int] = []
            for pid in analysis.get("matched_project_ids", []) or []:
                try:
                    matched_ids.append(int(pid))
                except Exception:
                    continue
            matched_ids = sorted(set(matched_ids))
            if not matched_ids:
                return []

            result = await self.db.execute(
                select(Project.id).where(
                    and_(
                        Project.id.in_(matched_ids),
                        Project.status == "active",
                        Project.tenant_id == tenant_id,
                    )
                )
            )
            valid_ids = {int(v) for v in result.scalars().all()}
            if not valid_ids:
                return []
            confidence = float(analysis.get("confidence") or 0.0)
            relevance = max(0.5, min(0.98, 0.55 + confidence * 0.4))
            link_type = "ai_existing_project_signal"
        else:
            result = await self.db.execute(
                select(Project.id).where(
                    and_(Project.status == "active", Project.tenant_id == tenant_id)
                )
            )
            valid_ids = {int(v) for v in result.scalars().all()}
            if not valid_ids:
                return []
            relevance = 0.7
            link_type = "auto_captured"

        linked_project_ids: List[int] = []

        for project_id in sorted(valid_ids):
            existing = await self.db.execute(
                text("SELECT id FROM project_signal_links WHERE project_id = :project_id AND signal_id = :signal_id"),
                {"project_id": project_id, "signal_id": signal_id},
            )
            if existing.first() is None:
                from ..models import ProjectSignalLink
                link = ProjectSignalLink(
                    project_id=project_id,
                    signal_id=signal_id,
                    link_type=link_type,
                    relevance_score=relevance,
                    tenant_id=tenant_id,
                )
                self.db.add(link)
                linked_project_ids.append(project_id)

        await self.db.commit()
        return linked_project_ids

    async def _auto_create_project_from_new_signal(
        self,
        *,
        signal_id: int,
        message: str,
        sender: str,
        tenant_id: int,
        requested_at: Optional[datetime] = None,
    ) -> Optional[int]:
        """当分类为 new_project 时，自动创建项目并关联当前信号。"""
        requested_dt = requested_at or datetime.now(timezone.utc)
        if requested_dt.tzinfo is None:
            requested_dt = requested_dt.replace(tzinfo=timezone.utc)
        project_brief = await self._draft_new_project_brief(
            message=message,
            tenant_id=tenant_id,
            sender=sender,
            requested_at=requested_dt,
        )
        title = self._normalize_single_line_text(project_brief.get("title") or "", max_len=120)
        if not title:
            title = self._normalize_single_line_text(message or "飞书新需求立项", max_len=40) or "飞书新需求立项"
        raw_desc = self._normalize_multiline_text(project_brief.get("description") or message or "", max_len=1200)
        requested_at_text = self._format_cn_time(requested_dt)
        description = self._normalize_multiline_text(
            f"需求提出人：{sender}\n"
            f"提出时间：{requested_at_text}\n"
            f"需求来源：飞书群聊\n\n"
            f"需求说明：{raw_desc or self._normalize_multiline_text(message, max_len=500)}",
            max_len=1800,
        )
        assigned_role = str(project_brief.get("primary_role") or "ai_engineer").strip()[:50] or "ai_engineer"

        owner_id: Optional[int] = None
        owner_result = await self.db.execute(
            select(User.id).where(
                and_(
                    User.tenant_id == tenant_id,
                    User.status == "active",
                    User.role.in_(["BOSS", "ADMIN"]),
                )
            ).order_by(User.id.asc()).limit(1)
        )
        owner_id = owner_result.scalar_one_or_none()

        await ensure_project_id_sequence(self.db)
        project = Project(
            name=title[:80] or "新项目",
            title=title,
            description=description,
            owner_id=owner_id,
            status="active",
            assigned_role=assigned_role,
            signal_count=1,
            tenant_id=tenant_id,
        )
        self.db.add(project)
        await self.db.flush()
        await ensure_project_default_members(
            self.db,
            tenant_id=tenant_id,
            project_id=int(project.id),
            owner_id=int(project.owner_id) if project.owner_id is not None else None,
            assigned_role=project.assigned_role,
        )

        existing = await self.db.execute(
            text("SELECT id FROM project_signal_links WHERE project_id = :project_id AND signal_id = :signal_id"),
            {"project_id": int(project.id), "signal_id": int(signal_id)},
        )
        if existing.first() is None:
            link = ProjectSignalLink(
                project_id=int(project.id),
                signal_id=int(signal_id),
                link_type="ai_new_project_signal",
                relevance_score=0.98,
                tenant_id=tenant_id,
            )
            self.db.add(link)

        await self.db.execute(
            text("UPDATE signals SET processed = :processed WHERE id = :signal_id"),
            {"processed": True, "signal_id": int(signal_id)},
        )
        await self.db.commit()

        self._emit_ai_think(
            "new_project_created_from_signal",
            signal_id=signal_id,
            project_id=int(project.id),
            sender=sender,
        )
        return int(project.id)

    async def _draft_new_project_brief(
        self,
        *,
        message: str,
        tenant_id: int,
        sender: str,
        requested_at: datetime,
    ) -> Dict[str, Any]:
        requested_at_text = self._format_cn_time(requested_at)
        prompt = f"""你是 PMFlow 的立项助手。基于下面需求消息，提炼新项目基础信息。

需求提出人：{sender}
提出时间：{requested_at_text}

需求消息：
{message[:1000]}

输出 JSON：
{{
  "title": "项目标题（15-30字）",
  "description": "项目简介（80-200字）",
  "primary_role": "ai_engineer|backend|frontend|devops"
}}

只输出 JSON，不要 markdown，不要解释，不要乱码。"""
        try:
            raw = await self.llm.generate(
                prompt,
                db=self.db,
                tenant_id=tenant_id,
                function_name="feishu_new_project_brief",
            )
            parsed = self._parse_llm_json_response(raw)
            role = str(parsed.get("primary_role") or "ai_engineer").strip().lower()
            if role not in {"ai_engineer", "backend", "frontend", "devops"}:
                role = "ai_engineer"
            return {
                "title": self._normalize_single_line_text(str(parsed.get("title") or ""), max_len=120),
                "description": self._normalize_multiline_text(str(parsed.get("description") or ""), max_len=1200),
                "primary_role": role,
            }
        except Exception as exc:
            self._emit_ai_think("draft_new_project_brief_failed", error=str(exc))
            return {
                "title": self._normalize_single_line_text(message or "飞书新需求立项", max_len=40),
                "description": self._normalize_multiline_text(message, max_len=300),
                "primary_role": "ai_engineer",
            }


# 全局服务实例
_signal_capture_service: Optional[ProactiveSignalCaptureService] = None


def get_signal_capture_service() -> ProactiveSignalCaptureService:
    """获取信号捕获服务实例（需要在 async 上下文中使用）"""
    return ProactiveSignalCaptureService(None)  # Placeholder