"""AI PMO kernel ingress flow mixin.""" from __future__ import annotations from . import ai_pmo_kernel_common as _common for _name, _value in vars(_common).items(): if not _name.startswith("__"): globals()[_name] = _value class KernelIngressFlowMixin: async def process_feishu_message( self, *, group_chat_id: str, message_content: str, sender: str, source_message_id: Optional[str], source_sender_open_id: Optional[str] = None, captured_at: Optional[datetime] = None, chat_type: Optional[str] = None, ) -> dict[str, Any]: return await self.process_ingress( source_system="feishu", source_message_id=source_message_id, message_text=message_content, sender=sender, channel_ref=group_chat_id, source_sender_open_id=source_sender_open_id, captured_at=captured_at, chat_type=chat_type, ) async def process_ingress( self, *, source_system: str, source_message_id: Optional[str], message_text: str, sender: str, channel_ref: Optional[str] = None, source_sender_open_id: Optional[str] = None, captured_at: Optional[datetime] = None, chat_type: Optional[str] = None, project_hint_id: Optional[int] = None, ) -> dict[str, Any]: started = time.perf_counter() perf_work_item_id: Optional[int] = None perf_intent_name = "" def _log_perf(stage_name: str, stage_started_at: float) -> None: elapsed_seconds = time.perf_counter() - stage_started_at perf_line = ( f"[PERF] {stage_name} took {elapsed_seconds:.2f}s " f"work_item={perf_work_item_id if perf_work_item_id is not None else 'N/A'} " f"intent={perf_intent_name or 'N/A'}" ) logger.info(perf_line) print(perf_line, flush=True) clean_text = self._normalize_text(message_text) if not clean_text: return {"handled": True, "status": "ignored", "classification": "empty", "direct_reply_required": False} transport_mode_reply = self._build_transport_mode_reply(clean_text) if transport_mode_reply: return { "handled": True, "status": "answered", "classification": "question", "linked_project_ids": [int(project_hint_id)] if project_hint_id is not None else [], "signal_id": None, "created_project_id": None, "auto_tasks": [], "ai_trace_id": None, "results": [], "intent": INTENT_QUERY, "direct_reply_required": True, "direct_reply_text": transport_mode_reply, "reply_template_key": "transport_mode_ack", "work_item_id": None, "decision_level": None, "review_summary": None, } synthetic_message_id = self._build_source_message_id( source_system=source_system, source_message_id=source_message_id, channel_ref=channel_ref, sender=sender, message_text=clean_text, ) closure_payload = await self._build_message_closure_payload( source_system=source_system, channel_ref=channel_ref, sender=sender, source_sender_open_id=source_sender_open_id, message_text=clean_text, source_message_id=synthetic_message_id, captured_at=captured_at, project_id=project_hint_id, ) ingest_event, created = await self._claim_ingress_event( source_system=source_system, source_message_id=synthetic_message_id, payload={ "sender": sender, "source_sender_open_id": source_sender_open_id, "channel_ref": channel_ref, "chat_type": chat_type, "captured_at": captured_at.isoformat() if isinstance(captured_at, datetime) else None, "message_text": clean_text, "closure_dialogue": closure_payload.get("closure_dialogue"), "closure_summary": closure_payload.get("closure_summary"), "closure_modules": closure_payload.get("closure_modules"), "closure_turns": closure_payload.get("closure_turns"), }, ) if not created: return { "handled": True, "status": "ignored", "reason": "duplicate_filtered", "classification": "duplicate", "direct_reply_required": False, "source_message_id": synthetic_message_id, } operator = await self._resolve_operator_user(sender=sender, sender_open_id=source_sender_open_id) work_item = await self._create_work_item( ingest_event=ingest_event, source_system=source_system, source_message_id=synthetic_message_id, owner_user_id=_safe_int(getattr(operator, "id", None)), ) perf_work_item_id = int(work_item.id) session = await self._get_or_create_session(tenant_id=self.tenant_id, sender=sender) try: stage_started_at = time.perf_counter() project, world_model = await self.context_assembler.assemble_world_model( sender=sender, message_text=clean_text, project_hint_id=project_hint_id, session=session, ) _log_perf("_build_world_model", stage_started_at) self._schedule_portfolio_snapshot_persist(world_model=world_model) work_item.state = STATE_MODELED stage_started_at = time.perf_counter() intent, intent_context_source = await self._resolve_intent( message_text=clean_text, session=session, world_model=world_model, ) intent = self._enrich_signal_source_intent_from_closure( intent=intent, message_text=clean_text, closure_payload=closure_payload, ) if intent.name in WRITE_INTENTS and IntentResolver._looks_like_negated_write_statement(text=clean_text): metadata = dict(intent.metadata or {}) metadata["guard_reason"] = "negated_write_statement" metadata["guarded_from_intent"] = intent.name intent = CanonicalIntent( INTENT_QUERY, max(float(intent.confidence or 0.0), 0.9), intent.project_id, "", clean_text[:300], metadata, ) workflow_route = self.route_resolver.from_canonical( canonical_intent=intent, message_text=clean_text, ) route_metadata = dict(intent.metadata or {}) route_metadata.update( { "workflow_route": workflow_route.key, "workflow_family": workflow_route.family, "workflow_action": workflow_route.action, "requested_count": workflow_route.requested_count, "requires_confirmation": workflow_route.requires_confirmation, "world_model_capacity": world_model.get("capacity") if isinstance(world_model.get("capacity"), list) else [], "world_model_skill_graph": world_model.get("skill_graph") if isinstance(world_model.get("skill_graph"), list) else [], } ) intent.metadata = route_metadata workflow_route.metadata = route_metadata perf_intent_name = workflow_route.key _log_perf("_resolve_intent", stage_started_at) if intent_context_source: world_model["context_source"] = intent_context_source work_item.context_snapshot = world_model resolved_project_for_session = _safe_int(workflow_route.project_id) or _safe_int(world_model.get("resolved_project_id")) fallback_intent = self.intent_resolver.resolve( text=clean_text, project_id=_safe_int(world_model.get("resolved_project_id")), ) if intent.name == INTENT_IGNORE: work_item.state = STATE_IGNORED work_item.intent = INTENT_IGNORE ingest_event.status = "done" ingest_event.processed_at = _utc_now_naive() await self._write_processing_log( source_message_id=synthetic_message_id, work_item=work_item, message_text=clean_text, route="ai_pmo.ignore", operation_code=None, operation_result={"reason": intent.metadata.get("reason")}, processing_status="success", duration_ms=int((time.perf_counter() - started) * 1000), ) await self._update_session( tenant_id=self.tenant_id, sender=sender, current_project_id=resolved_project_for_session, last_intent=intent.name, last_message_text=clean_text[:200], ) await self.db.commit() return { "handled": True, "status": "ignored", "classification": "ignored", "direct_reply_required": False, "work_item_id": int(work_item.id), "ai_trace_id": f"ai-pmo-{int(work_item.id)}", } work_item.intent = workflow_route.key work_item.project_id = workflow_route.project_id stage_started_at = time.perf_counter() decision_payload = self.decision_engine.decide( intent=intent, world_model=world_model, message_text=clean_text, ) _log_perf("DecisionEngine.decide", stage_started_at) correction_action = str((intent.metadata or {}).get("correction_action") or "none").strip() or "none" requires_manual_review = bool((intent.metadata or {}).get("requires_manual_review")) manual_review_code = str((intent.metadata or {}).get("manual_review_code") or "manual_review_required").strip() manual_review_reason = str((intent.metadata or {}).get("manual_review_reason") or "").strip() llm_function = str((intent.metadata or {}).get("llm_function") or INTENT_LLM_FUNCTION_NAME).strip() route_tier = str((intent.metadata or {}).get("route_tier") or "rule").strip() route_reasons = (intent.metadata or {}).get("route_reasons") gate = kernel_gate_execution( confidence=float(intent.confidence or 0.0), risk_level=self._risk_level_for_intent(intent.name), correction_action=correction_action, force_manual=requires_manual_review, force_manual_code=manual_review_code, force_manual_reason=manual_review_reason, ) decision_trace = build_decision_trace( intent=intent.name, confidence=float(intent.confidence or 0.0), route_tier=route_tier, route_function=llm_function, gate_action=gate.action, gate_code=gate.gate_code, correction_action=correction_action, correction_keywords=(intent.metadata or {}).get("correction_keywords") if isinstance(intent.metadata, dict) else [], complexity_reasons=route_reasons if isinstance(route_reasons, list) else [], ) work_item.state = STATE_DECIDED work_item.decision_level = str((decision_payload["strategic"] or {}).get("decision_level") or "portfolio") decision_row = await self._create_decision_record( work_item=work_item, world_model=world_model, decision_payload=decision_payload, ) await self._create_plan_bundle( work_item_id=int(work_item.id), plan_type="strategic", summary="Strategic Brain 产出的项目群/项目策略。", payload=decision_payload["strategic_plan"], ) await self._create_plan_bundle( work_item_id=int(work_item.id), plan_type="execution", summary="Execution Plan typed action bundle。", payload=decision_payload["execution_plan"], ) policy = decision_payload["policy"] if self._should_bypass_policy_gate( intent=intent, workflow_route=workflow_route, ): policy["auto_execute"] = True policy["policy_gate_hit"] = False policy["policy_gate_code"] = None policy["approval_requirement"] = { "required": False, "reason": "workflow_policy_bypass", "message": "工作流恢复路径按快照执行，不触发仿真门禁拦截。", } if not gate.auto_execute: policy["auto_execute"] = False policy["policy_gate_hit"] = True policy["policy_gate_code"] = gate.gate_code policy["approval_requirement"] = { "required": True, "reason": gate.gate_code, "message": gate.reason, } if not bool(policy.get("auto_execute")): work_item.state = STATE_WAITING_MANUAL work_item.action_bundle = decision_payload["execution_plan"] work_item.execution_result = { "status": "waiting_manual", "reason": ((policy.get("approval_requirement") or {}).get("message") or "").strip(), "workflow_route": workflow_route.key, "workflow_action": workflow_route.action, "decision_trace": decision_trace, } manual_issues = ["waiting_manual"] policy_gate_code = str(policy.get("policy_gate_code") or "").strip() if policy_gate_code: manual_issues.append(policy_gate_code) work_item.review_summary = await self.review_service.review_work_item( work_item=work_item, intent_name=intent.name, execution_result=work_item.execution_result, world_model=world_model, verification={"verified": False, "issues": manual_issues}, simulation=decision_payload["simulation"], ) render = self.reply_renderer.render( work_state=STATE_WAITING_MANUAL, intent=intent, decision_record={ "approval_requirement": policy.get("approval_requirement"), }, execution_result={}, ) work_item.reply_template_key = str(render["template_key"]) await self._project_waiting_manual( work_item=work_item, message_text=clean_text, render=render, intent=intent, ) await self._maybe_emit_risk_cadence_snapshot( project_id=_safe_int(work_item.project_id), review_summary=work_item.review_summary, ) direct_reply_required = await self._claim_reply_receipt( work_item_id=int(work_item.id), source_system=source_system, source_message_id=synthetic_message_id, receipt_phase=str(render["receipt_phase"]), template_key=str(render["template_key"]), payload=render["payload"], ) ingest_event.status = "done" ingest_event.processed_at = _utc_now_naive() await self._write_processing_log( source_message_id=synthetic_message_id, work_item=work_item, message_text=clean_text, route="ai_pmo.waiting_manual", operation_code=None, operation_result=work_item.execution_result, processing_status="success", duration_ms=int((time.perf_counter() - started) * 1000), ) await self._update_session( tenant_id=self.tenant_id, sender=sender, current_project_id=_safe_int(work_item.project_id), last_intent=intent.name, last_message_text=clean_text[:200], ) await self.db.commit() stage_started_at = time.perf_counter() capture_result = self._build_capture_result( work_item=work_item, intent=intent, execution_result=work_item.execution_result or {}, render=render, direct_reply_required=direct_reply_required, status="captured", classification="command", ) _log_perf("_build_capture_result", stage_started_at) return capture_result work_item.state = STATE_PLANNED work_item.action_bundle = decision_payload["execution_plan"] work_item.state = STATE_EXECUTING if intent.name in {INTENT_FOLLOWUP_CONFIRM, INTENT_FOLLOWUP_CONTINUE}: stage_started_at = time.perf_counter() execution_result, verification = await self._execute_followup_intent( work_item=work_item, intent=intent, message_text=clean_text, sender=sender, operator_user_id=_safe_int(getattr(operator, "id", None)), source_system=source_system, source_message_id=synthetic_message_id, channel_ref=channel_ref, source_sender_open_id=source_sender_open_id, world_model=world_model, workflow_route=workflow_route, ) _log_perf("TypedExecutor.execute", stage_started_at) else: stage_started_at = time.perf_counter() handler = self.workflow_registry.get_handler(route=workflow_route) execution_result = await handler.start( route=workflow_route, intent=intent, message_text=clean_text, sender=sender, operator_user_id=_safe_int(getattr(operator, "id", None)), source_system=source_system, source_message_id=synthetic_message_id, channel_ref=channel_ref, ) _log_perf("TypedExecutor.execute", stage_started_at) verification = await self._verify_execution(intent=intent, execution_result=execution_result, world_model=world_model) if isinstance(execution_result, dict): execution_result.setdefault("decision_trace", decision_trace) execution_result.setdefault("workflow_route", workflow_route.key) execution_result.setdefault("workflow_action", workflow_route.action) resolved_route = str(execution_result.get("resolved_route") or "").strip().lower() if resolved_route: work_item.intent = self.route_resolver.normalize_route_key(resolved_route) work_item.execution_result = execution_result if _safe_int(execution_result.get("project_id")) is not None: work_item.project_id = _safe_int(execution_result.get("project_id")) if _safe_int(execution_result.get("signal_id")) is not None: work_item.signal_id = _safe_int(execution_result.get("signal_id")) work_item.review_summary = await self.review_service.review_work_item( work_item=work_item, intent_name=intent.name, execution_result=execution_result, world_model=world_model, verification=verification, simulation=decision_payload["simulation"], ) await self._maybe_emit_risk_cadence_snapshot( project_id=_safe_int(work_item.project_id), review_summary=work_item.review_summary, ) decision_row.execution_result = execution_result issues = verification.get("issues") if isinstance(verification.get("issues"), list) else [] if verification["verified"]: work_item.state = STATE_SUCCEEDED await self._project_success(intent=intent, execution_result=execution_result) elif any( issue in {"waiting_manual", "followup_ambiguous", "signal_source_pending_manual"} for issue in issues ) or str(execution_result.get("status") or "").strip().lower() in {"pending_manual", "pending_confirmation"}: work_item.state = STATE_WAITING_MANUAL else: work_item.state = STATE_FAILED decision_row.final_outcome = work_item.state render = self.reply_renderer.render( work_state=work_item.state, intent=intent, decision_record={ "approval_requirement": decision_payload["policy"].get("approval_requirement"), }, execution_result=execution_result, ) work_item.reply_template_key = str(render["template_key"]) direct_reply_required = await self._claim_reply_receipt( work_item_id=int(work_item.id), source_system=source_system, source_message_id=synthetic_message_id, receipt_phase=str(render["receipt_phase"]), template_key=str(render["template_key"]), payload=render["payload"], ) ingest_event.status = "done" ingest_event.processed_at = _utc_now_naive() await self._write_processing_log( source_message_id=synthetic_message_id, work_item=work_item, message_text=clean_text, route=f"ai_pmo.{work_item.intent or intent.name}", operation_code=self._primary_operation_code(str(work_item.intent or intent.name)), operation_result=execution_result, processing_status="success" if work_item.state in {STATE_SUCCEEDED, STATE_WAITING_MANUAL} else "failed", duration_ms=int((time.perf_counter() - started) * 1000), ) final_project_id = _safe_int(work_item.project_id) or resolved_project_for_session final_intent_name = intent.name or fallback_intent.name await self._update_session( tenant_id=self.tenant_id, sender=sender, current_project_id=final_project_id, last_intent=final_intent_name, last_message_text=clean_text[:200], ) await self.db.commit() execution_route_key = RouteResolver.normalize_route_key( str((execution_result or {}).get("workflow_route") or "") ) if isinstance(execution_result, dict) else "" should_run_signal_planner = ( work_item.state == STATE_SUCCEEDED and ( intent.name == INTENT_SIGNAL_CREATE or execution_route_key == ROUTE_PROJECT_SIGNAL_COMMIT ) ) if should_run_signal_planner: signal_ids = _normalize_signal_id_list(execution_result.get("signal_ids")) signal_id = _safe_int(execution_result.get("signal_id")) if signal_id is not None and signal_id not in signal_ids: signal_ids = [signal_id, *signal_ids] for planner_signal_id in signal_ids: asyncio.create_task( SignalPlannerService.analyze( signal_id=int(planner_signal_id), tenant_id=int(self.tenant_id), source_message_id=synthetic_message_id, source_chat_id=channel_ref, dedupe_seed=f"signal-planner-failed:{int(planner_signal_id)}", session_factory=self._background_session_factory, ) ) stage_started_at = time.perf_counter() capture_result = self._build_capture_result( work_item=work_item, intent=intent, execution_result=execution_result, render=render, direct_reply_required=direct_reply_required, status="answered" if intent.name in {INTENT_QUERY, INTENT_FOLLOWUP_CONFIRM, INTENT_FOLLOWUP_CONTINUE} and work_item.state == STATE_SUCCEEDED else "captured", classification=( "question" if intent.name == INTENT_QUERY else ("signal" if intent.name == INTENT_SIGNAL_CREATE else "command") ), ) _log_perf("_build_capture_result", stage_started_at) return capture_result except Exception as exc: await self.db.rollback() logger.warning( "ai_pmo_kernel_failed", tenant_id=self.tenant_id, source_system=source_system, source_message_id=synthetic_message_id, error=str(exc), ) return {"handled": False, "reason": "kernel_exception", "error": str(exc)}