"""Operating mission critic for specialist coverage and replan decisions.""" from __future__ import annotations import json import re from typing import Any from ...schemas.pmflow_ai_operating_runtime import ( PMFlowAIAgentCapabilityContract, PMFlowAIOperatingMission, PMFlowAIOperatingMissionCriticResult, ) from ..ai_function_runtime_profile_service import AIFunctionRuntimeProfileService from ..llm_service import llm_service from ..prompt_runtime_service import PromptRuntimeService from ..runtime_audit_context import bind_agent_audit_context from .agent_registry import PMFlowAIOperatingAgentRegistry PROMPT_KEY_PMFLOW_AI_OPERATING_MISSION_CRITIC = "PROMPT_PMFLOW_AI_OPERATING_MISSION_CRITIC" FUNCTION_NAME_PMFLOW_AI_OPERATING_MISSION_CRITIC = "pmflow_ai_operating_mission_critic" SCENE_KEY_PMFLOW_AI_OPERATING_RUNTIME = "space_pmflow_ai_operating_runtime" _GENERIC_CONTRACT_TOKENS = { "agent", "agents", "operating", "mission", "specialist", "work", "works", "product", "products", "prepared", "prepare", "changes", "change", "claim", "claims", "customer", "ledger", "customersafeclaimledger", "safe", "suggestions", "suggestion", "当前", "本轮", "目标", "对象", "范围", "专业", "产物", "结构化", "确认", "可确认", } class PMFlowAIOperatingMissionCriticService: """Check whether a mission has enough specialist capability coverage. The critic is capability-contract driven. Runtime decisions are not made from user-text keywords or industry enumerations; the optional LLM critic receives only mission, artifact/evidence contracts and registered agents. """ def __init__( self, *, db: Any | None = None, tenant_id: int | None = None, generator: Any | None = None, agent_registry: PMFlowAIOperatingAgentRegistry | None = None, ) -> None: self.db = db self.tenant_id = int(tenant_id or 1) self.generator = generator or llm_service self.agent_registry = agent_registry or PMFlowAIOperatingAgentRegistry() async def review( self, *, user_question: str, mission: PMFlowAIOperatingMission, artifact_slots: list[dict[str, Any]] | None = None, evidence_lanes: list[dict[str, Any]] | None = None, current_object_binding: dict[str, Any] | None = None, ) -> PMFlowAIOperatingMissionCriticResult: fallback = self._structural_review( mission=mission, artifact_slots=artifact_slots or [], evidence_lanes=evidence_lanes or [], current_object_binding=current_object_binding or {}, ) generate = getattr(self.generator, "generate", None) if not callable(generate) or not callable(getattr(self.db, "execute", None)): return fallback try: contracts = self.agent_registry.list_contracts() prompt_result = await PromptRuntimeService(self.db).render( tenant_id=self.tenant_id, prompt_key=PROMPT_KEY_PMFLOW_AI_OPERATING_MISSION_CRITIC, user_question=str(user_question or "")[:1600], operating_mission_json=_dump_json(mission.model_dump(mode="json", exclude_none=True), limit=7000), specialist_contracts_json=_dump_json([item.model_dump(mode="json", exclude_none=True) for item in contracts], limit=12000), artifact_slots_json=_dump_json(artifact_slots or [], limit=7000), evidence_lanes_json=_dump_json(evidence_lanes or [], limit=7000), current_object_binding_json=_dump_json(current_object_binding or {}, limit=4000), ) runtime_profile = await AIFunctionRuntimeProfileService(self.db).resolve_profile( function_name=FUNCTION_NAME_PMFLOW_AI_OPERATING_MISSION_CRITIC, scene_key=SCENE_KEY_PMFLOW_AI_OPERATING_RUNTIME, prompt_key=PROMPT_KEY_PMFLOW_AI_OPERATING_MISSION_CRITIC, ) with bind_agent_audit_context(agent_step_key="operating_mission_critic"): raw = await generate( prompt_result.prompt, db=self.db, tenant_id=self.tenant_id, function_name=FUNCTION_NAME_PMFLOW_AI_OPERATING_MISSION_CRITIC, use_prompt_profile=False, temperature=runtime_profile.temperature, response_mode=runtime_profile.response_mode, max_tokens=runtime_profile.max_tokens, routing_meta={ "scene_key": SCENE_KEY_PMFLOW_AI_OPERATING_RUNTIME, "runtime": "operating_mission_critic", "scene_contract": "mission_capability_coverage", **runtime_profile.routing_meta(), "allow_low_to_high_escalation": False, "disable_thinking": True, **prompt_result.routing_meta(), }, ) payload = _extract_json_object(raw) if not payload: return fallback return _repair_critic_result( result=PMFlowAIOperatingMissionCriticResult.model_validate(payload), fallback=fallback, contracts=self.agent_registry.list_contracts(), selected_agents=list(mission.required_agents or []), ) except Exception: return fallback def _structural_review( self, *, mission: PMFlowAIOperatingMission, artifact_slots: list[dict[str, Any]], evidence_lanes: list[dict[str, Any]], current_object_binding: dict[str, Any], ) -> PMFlowAIOperatingMissionCriticResult: selected = [str(item or "").strip() for item in mission.required_agents if str(item or "").strip()] missing: list[str] = [] desired_outputs = _desired_output_tokens( mission=mission, artifact_slots=artifact_slots, evidence_lanes=evidence_lanes, current_object_binding=current_object_binding, ) for contract in self.agent_registry.list_contracts(): if contract.agent_id in selected: continue if _contract_matches_desired_outputs(contract=contract, desired_outputs=desired_outputs): missing.append(contract.agent_id) missing = _dedupe([item for item in missing if item]) mission_ok = not missing return PMFlowAIOperatingMissionCriticResult( mission_ok=mission_ok, missing_capability_roles=missing, over_selected_roles=[], needs_replan=bool(missing), reason=( "当前 mission 与已有结构化材料需要补充专业能力产物。" if missing else "当前 mission 的专业能力覆盖足以支撑本轮交付。" ), evidence_alignment={ "used_artifact_slots": _slot_keys(artifact_slots), "used_evidence_lanes": _lane_keys(evidence_lanes), "desired_output_contracts": desired_outputs[:24], }, ) def merge_missing_roles_into_mission( *, mission: PMFlowAIOperatingMission, critic: PMFlowAIOperatingMissionCriticResult, ) -> PMFlowAIOperatingMission: selected = [str(item or "").strip() for item in mission.required_agents if str(item or "").strip()] merged = _dedupe([*selected, *list(critic.missing_capability_roles or [])]) if merged == selected: return mission return mission.model_copy( update={ "required_agents": merged[:16], "success_criteria": _dedupe( [ *list(mission.success_criteria or []), "OperatingMissionCritic 已补齐与业务目标、结构化交付槽位和证据通道一致的 specialist coverage。", ] )[:16], } ) def _desired_output_tokens( *, mission: PMFlowAIOperatingMission, artifact_slots: list[dict[str, Any]], evidence_lanes: list[dict[str, Any]], current_object_binding: dict[str, Any], ) -> list[str]: tokens: list[str] = [] tokens.append(str(mission.deliverable_type or "")) tokens.append(str(mission.stage or "")) tokens.extend(_split_capability_tokens(mission.desired_outcome)) tokens.extend(_split_capability_tokens(mission.desired_business_outcome)) for item in artifact_slots: if not isinstance(item, dict): continue slot_tokens = _split_capability_tokens(item.get("slot_key")) goal_tokens = _split_capability_tokens(item.get("display_goal") or item.get("title") or item.get("summary")) tokens.extend(slot_tokens) tokens.extend(goal_tokens) for row in list(item.get("must_cover") or []): row_tokens = _split_capability_tokens(row) tokens.extend(row_tokens) for item in evidence_lanes: if not isinstance(item, dict): continue lane_tokens = _split_capability_tokens(item.get("lane") or item.get("facet_key") or item.get("fact_key")) summary_tokens = _split_capability_tokens(item.get("summary") or item.get("display_label") or item.get("purpose")) tokens.extend(lane_tokens) tokens.extend(summary_tokens) tokens.extend(_split_capability_tokens(current_object_binding.get("object_type") if isinstance(current_object_binding, dict) else "")) return _dedupe([item for item in tokens if item])[:48] def _contract_matches_desired_outputs(*, contract: PMFlowAIAgentCapabilityContract, desired_outputs: list[str]) -> bool: if not desired_outputs: return False contract_tokens: set[str] = set() for value in [ contract.agent_id, contract.display_name, contract.responsibility, *list(contract.output_contract or []), *list(contract.critic_checks or []), ]: contract_tokens.update(_split_capability_tokens(value)) if not contract_tokens: return False desired = set(desired_outputs) if desired & contract_tokens: return True return False def _slot_keys(artifact_slots: list[dict[str, Any]]) -> list[str]: return _dedupe([str(item.get("slot_key") or "").strip() for item in artifact_slots if isinstance(item, dict)]) def _lane_keys(evidence_lanes: list[dict[str, Any]]) -> list[str]: rows: list[str] = [] for item in evidence_lanes: if not isinstance(item, dict): continue rows.append(str(item.get("lane") or item.get("facet_key") or item.get("fact_key") or "").strip()) return _dedupe(rows) def _split_capability_tokens(value: Any) -> list[str]: text = str(value or "").strip() if not text: return [] words = re.findall(r"[A-Za-z][A-Za-z0-9]+|[\u4e00-\u9fff]{2,}", text.replace("_", " ")) out: list[str] = [] for word in words: normalized = word.strip().lower() if not normalized or normalized in _GENERIC_CONTRACT_TOKENS: continue out.append(normalized) for part in re.findall(r"[A-Z]?[a-z]+|[A-Z]+(?=[A-Z]|$)|\d+", word): normalized_part = part.lower() if len(part) >= 3 and normalized_part not in _GENERIC_CONTRACT_TOKENS: out.append(normalized_part) return _dedupe(out) def _repair_critic_result( *, result: PMFlowAIOperatingMissionCriticResult, fallback: PMFlowAIOperatingMissionCriticResult, contracts: list[PMFlowAIAgentCapabilityContract], selected_agents: list[str], ) -> PMFlowAIOperatingMissionCriticResult: allowed = {item.agent_id for item in contracts} missing = [item for item in list(result.missing_capability_roles or []) if item in allowed and item not in selected_agents] over = [item for item in list(result.over_selected_roles or []) if item in allowed and item in selected_agents] if not missing and fallback.missing_capability_roles: missing = [item for item in fallback.missing_capability_roles if item in allowed and item not in selected_agents] return result.model_copy( update={ "missing_capability_roles": _dedupe(missing)[:16], "over_selected_roles": _dedupe(over)[:16], "mission_ok": bool(result.mission_ok) and not missing, "needs_replan": bool(result.needs_replan or missing), "reason": result.reason or fallback.reason, "evidence_alignment": result.evidence_alignment or fallback.evidence_alignment, } ) def _extract_json_object(raw: Any) -> dict[str, Any]: if isinstance(raw, dict): return raw text = str(raw or "").strip() if not text: return {} try: value = json.loads(text) return value if isinstance(value, dict) else {} except Exception: pass match = re.search(r"\{.*\}", text, flags=re.S) if not match: return {} try: value = json.loads(match.group(0)) return value if isinstance(value, dict) else {} except Exception: return {} def _dump_json(value: Any, *, limit: int) -> str: try: text = json.dumps(value, ensure_ascii=False, default=str) except Exception: text = str(value) return text[:limit] def _dedupe(rows: list[str]) -> list[str]: out: list[str] = [] for row in rows: text = " ".join(str(row or "").split())[:240] if text and text not in out: out.append(text) return out __all__ = ["PMFlowAIOperatingMissionCriticService", "merge_missing_roles_into_mission"]