"""Specialist work-product layer for the durable PMFlow Agent OS runtime.""" from __future__ import annotations from dataclasses import dataclass, field from typing import Any from ...schemas.pmflow_agent_runtime import ( PMFlowAgentArtifactPacket, PMFlowAgentStepPacket, PMFlowAgentVerificationPacket, new_pmflow_agent_id, ) from ...schemas.pmflow_ai_operating_runtime import ( PMFlowAIAgentAssignment, PMFlowAIAgentCouncilReview, PMFlowAIAgentWorkProduct, PMFlowAIAutonomyDecision, PMFlowAIOperatingCycleState, PMFlowAIOperatingMission, PMFlowAIOperatingPlan, PMFlowAIOperatingScope, PMFlowAIProjectWorldStatePacket, ) from ...schemas.rag import RAGAskRequest from ..pmflow_ai_operating_runtime.agent_registry import PMFlowAIOperatingAgentRegistry from ..pmflow_ai_operating_runtime.autonomy import PMFlowAIAutonomyPolicyService from ..pmflow_ai_operating_runtime.council import PMFlowAIAgentCouncilService from ..pmflow_ai_operating_runtime.cycle import OperatingCycleService from ..pmflow_ai_operating_runtime.project_world_state_service import PMFlowAIProjectWorldStateService from ..pmflow_ai_operating_runtime.specialist_agents import PMFlowAISpecialistAgentService @dataclass(slots=True) class PMFlowAgentSpecialistResult: mission: PMFlowAIOperatingMission | None = None project_world_state: PMFlowAIProjectWorldStatePacket | None = None assignments: list[PMFlowAIAgentAssignment] = field(default_factory=list) work_products: list[PMFlowAIAgentWorkProduct] = field(default_factory=list) council_review: PMFlowAIAgentCouncilReview | None = None autonomy_gate: PMFlowAIAutonomyDecision | None = None operating_cycle: PMFlowAIOperatingCycleState | None = None operating_plan: PMFlowAIOperatingPlan | None = None command_payload: dict[str, Any] = field(default_factory=dict) steps: list[PMFlowAgentStepPacket] = field(default_factory=list) artifacts: list[PMFlowAgentArtifactPacket] = field(default_factory=list) verifications: list[PMFlowAgentVerificationPacket] = field(default_factory=list) answer: str | None = None skipped_reason: str | None = None def customer_safe_payload(self) -> dict[str, Any]: return { "packet_version": "pmflow_agent_specialists.v1", "visibility": "customer", "skipped_reason": self.skipped_reason, "mission": _dump_model(self.mission), "project_world_state": _dump_model(self.project_world_state), "assignments": [_dump_model(item) for item in self.assignments], "work_products": [_dump_model(item) for item in self.work_products], "agent_council_review": _dump_model(self.council_review), "autonomy_gate": _dump_model(self.autonomy_gate), "operating_cycle": _dump_model(self.operating_cycle), "consumed_learning_ledger": _dump_model(getattr(self.operating_cycle, "consumed_learning_ledger", None)), "outcome_observations": [ _dump_model(item) for item in list(getattr(self.operating_cycle, "outcome_observations", []) or []) ], "learning_ledger": _dump_model(getattr(self.operating_cycle, "learning_ledger", None)), "next_cycle_triggers": list(getattr(self.operating_cycle, "next_cycle_triggers", []) or []), "operating_plan": _dump_model(self.operating_plan), "command_payload": dict(self.command_payload or {}), "steps": [item.model_dump(mode="json", exclude_none=True) for item in self.steps], "artifacts": [item.model_dump(mode="json", exclude_none=True) for item in self.artifacts], "verifications": [item.model_dump(mode="json", exclude_none=True) for item in self.verifications], } class PMFlowAgentSpecialistRuntimeService: """Run role-specific agents over a strong project anchor. This service is not a question classifier. It only starts when the durable Agent OS already has a structured project anchor, then consumes one shared project world-state packet to produce role work products. """ def __init__( self, *, db: Any, tenant_id: int, agent_registry: Any | None = None, project_world_state_service: Any | None = None, specialist_agent_service: Any | None = None, council_service: Any | None = None, autonomy_policy_service: Any | None = None, operating_cycle_service: Any | None = None, ) -> None: self.db = db self.tenant_id = int(tenant_id or 1) self.agent_registry = agent_registry or PMFlowAIOperatingAgentRegistry() self.project_world_state_service = project_world_state_service or PMFlowAIProjectWorldStateService( db, tenant_id=self.tenant_id, ) self.specialist_agent_service = specialist_agent_service or PMFlowAISpecialistAgentService() self.council_service = council_service or PMFlowAIAgentCouncilService() self.autonomy_policy_service = autonomy_policy_service or PMFlowAIAutonomyPolicyService() self.operating_cycle_service = operating_cycle_service or OperatingCycleService() async def run( self, *, request: RAGAskRequest, current_user: Any, conversation_context: dict[str, Any] | None, tenant_project_context: dict[str, Any] | None, ) -> PMFlowAgentSpecialistResult: project_ref = _strong_project_ref( request=request, conversation_context=conversation_context, tenant_project_context=tenant_project_context, ) if not project_ref: return PMFlowAgentSpecialistResult(skipped_reason="本轮没有强项目锚点，未运行项目专业 agent。") mission = _build_mission( request=request, current_user=current_user, conversation_context=conversation_context, project_ref=project_ref, ) project_world_state = await self.project_world_state_service.build_packet( mission=mission, current_user=current_user, tenant_project_context=tenant_project_context, ) assignments = self.agent_registry.assignments_for_mission(mission) work_products = self.specialist_agent_service.produce_work_products( mission=mission, assignments=assignments, project_world_state=project_world_state, ) council_review, operating_plan = self.council_service.review_and_compile( mission=mission, project_world_state=project_world_state, work_products=work_products, ) autonomy_gate, operating_plan = self.autonomy_policy_service.apply( mission=mission, project_world_state=project_world_state, work_products=work_products, council_review=council_review, operating_plan=operating_plan, ) operating_cycle = self.operating_cycle_service.build_cycle_state( request=request, mission=mission, project_world_state=project_world_state, council_review=council_review, autonomy_gate=autonomy_gate, operating_plan=operating_plan, conversation_context=conversation_context, ) operating_plan = _attach_cycle_to_plan( operating_plan=operating_plan, operating_cycle=operating_cycle, ) command_payload = _build_command_payload( mission=mission, project_world_state=project_world_state, assignments=assignments, work_products=work_products, council_review=council_review, autonomy_gate=autonomy_gate, operating_cycle=operating_cycle, operating_plan=operating_plan, ) result = PMFlowAgentSpecialistResult( mission=mission, project_world_state=project_world_state, assignments=assignments, work_products=work_products, council_review=council_review, autonomy_gate=autonomy_gate, operating_cycle=operating_cycle, operating_plan=operating_plan, command_payload=command_payload, ) result.steps = _specialist_steps(result) result.artifacts = _specialist_artifacts(result) result.verifications = _specialist_verifications(result) result.answer = _compose_specialist_answer( mission=mission, project_world_state=project_world_state, work_products=work_products, council_review=council_review, autonomy_gate=autonomy_gate, operating_cycle=operating_cycle, operating_plan=operating_plan, ) return result def _build_mission( *, request: RAGAskRequest, current_user: Any, conversation_context: dict[str, Any] | None, project_ref: str, ) -> PMFlowAIOperatingMission: question = _text(getattr(request, "question", ""), 1600) project_id = _project_id_from_ref(project_ref) return PMFlowAIOperatingMission( mission_goal=f"围绕 {project_ref} 推进用户本轮目标：{question or '继续推进当前项目'}", desired_outcome="形成面向当前项目对象的经营判断、专业 agent 产物、可确认动作和下一步推进计划。", original_user_input=question, project_scope=PMFlowAIOperatingScope( tenant_id=int(getattr(current_user, "tenant_id", 0) or 0) or None, user_id=int(getattr(current_user, "id", 0) or 0) or None, session_id=_safe_int(getattr(request, "session_id", None)) or _safe_int((conversation_context or {}).get("session_id")), object_refs=[project_ref], current_project_id=project_id, surface="pmflow_ai", ), success_criteria=[ "专业 agent 产物必须来自同一份项目世界状态。", "每个角色产物必须有结构化输出、依据引用和角色检查结果。", "用户可见回答必须从角色产物和会审后的推进顺序编译。", ], autonomy_level="prepare_changes", decision_boundary=[ "Project Memory 写入、ChangeSet commit、任务变更、Execution Bridge handoff、发布和外部动作必须经过用户确认。", ], ) def _specialist_steps(result: PMFlowAgentSpecialistResult) -> list[PMFlowAgentStepPacket]: world_state = result.project_world_state world_status = str(getattr(world_state, "status", "") or "unknown") council_status = str(getattr(result.council_review, "status", "") or "unknown") autonomy_status = str(getattr(result.autonomy_gate, "status", "") or "unknown") cycle_status = "completed" if result.operating_cycle is not None else "blocked" return [ PMFlowAgentStepPacket( step_id=new_pmflow_agent_id("spec"), kind="specialist_assignment", status="completed", label="组织专业 Agent：分派项目经营角色", summary=f"已分派 {len(result.assignments)} 个角色，共同消费同一份项目世界状态。", ), PMFlowAgentStepPacket( step_id=new_pmflow_agent_id("spec"), kind="specialist_work_products", status="completed" if world_status in {"ready", "partial"} else "blocked", label="生成专业产物：策略、用户验证、产品、技术、执行和风险", summary=f"已生成 {len(result.work_products)} 个结构化工作产物，项目状态为 {world_status}。", ), PMFlowAgentStepPacket( step_id=new_pmflow_agent_id("spec"), kind="specialist_council_compile", status="completed" if council_status != "blocked" else "blocked", label="会审产物：收敛优先级和边界", summary=_text(getattr(result.council_review, "resolution_summary", "") or "已把角色产物收敛成单一推进顺序。", 600), ), PMFlowAgentStepPacket( step_id=new_pmflow_agent_id("spec"), kind="autonomy_gate_prepared_actions", status="completed" if autonomy_status != "blocked" else "blocked", label="确认门禁：准备动作包并阻断直接写入/执行", summary=_autonomy_gate_summary(result.autonomy_gate), ), PMFlowAgentStepPacket( step_id=new_pmflow_agent_id("spec"), kind="operating_cycle_learning_loop", status=cycle_status, label="学习闭环：记录结果和下一轮触发", summary=_operating_cycle_summary(result.operating_cycle), ), ] def _specialist_artifacts(result: PMFlowAgentSpecialistResult) -> list[PMFlowAgentArtifactPacket]: artifacts: list[PMFlowAgentArtifactPacket] = [] if result.mission is not None: artifacts.append( PMFlowAgentArtifactPacket( kind="operating_mission", status="created", title="本轮项目经营目标", summary=result.mission.mission_goal, payload=_dump_model(result.mission), ) ) if result.project_world_state is not None: artifacts.append( PMFlowAgentArtifactPacket( kind="project_world_state", status=result.project_world_state.status, title="项目世界状态", summary=_project_world_summary(result.project_world_state), payload=_dump_model(result.project_world_state), ) ) for product in result.work_products[:8]: artifacts.append( PMFlowAgentArtifactPacket( kind="specialist_work_product", status=product.status, title=_work_product_title(product), summary=product.summary, payload=_dump_model(product), ) ) if result.council_review is not None: artifacts.append( PMFlowAgentArtifactPacket( kind="agent_council_review", status=result.council_review.status, title="专业 Agent 会审", summary=result.council_review.resolution_summary, payload=_dump_model(result.council_review), ) ) if result.autonomy_gate is not None: artifacts.append( PMFlowAgentArtifactPacket( kind="autonomy_gate", status=result.autonomy_gate.status, title="确认门禁", summary=_autonomy_gate_summary(result.autonomy_gate), payload=_dump_model(result.autonomy_gate), ) ) for packet in list(result.autonomy_gate.prepared_action_packets or [])[:8]: if isinstance(packet, dict): artifacts.append( PMFlowAgentArtifactPacket( kind="prepared_action_packet", status=str(packet.get("status") or "prepared_waiting_confirmation")[:32], title=_text(packet.get("title"), 240) or "待确认动作包", summary=_prepared_action_summary(packet), payload=packet, ) ) for request in list(result.autonomy_gate.confirmation_requests or [])[:8]: if isinstance(request, dict): artifacts.append( PMFlowAgentArtifactPacket( kind="confirmation_request", status=str(request.get("status") or "waiting_for_user_confirmation")[:32], title=_text(request.get("title"), 240) or "确认请求", summary=_text(request.get("confirmation_reason"), 1000), payload=request, ) ) for preparation in list(result.autonomy_gate.execution_preparations or [])[:4]: if isinstance(preparation, dict): artifacts.append( PMFlowAgentArtifactPacket( kind="execution_preparation", status=str(preparation.get("readiness") or "prepared_waiting_confirmation")[:32], title="执行交接准备包", summary="已准备执行交接材料，但未发起 handoff。", payload=preparation, ) ) if result.operating_cycle is not None: artifacts.append( PMFlowAgentArtifactPacket( kind="operating_cycle", status="created", title="持续推进闭环", summary=_operating_cycle_summary(result.operating_cycle), payload=_dump_model(result.operating_cycle), ) ) artifacts.append( PMFlowAgentArtifactPacket( kind="learning_ledger", status="updated", title="项目学习账本", summary=_learning_ledger_summary(result.operating_cycle), payload=_dump_model(result.operating_cycle.learning_ledger), ) ) if result.command_payload: artifacts.append( PMFlowAgentArtifactPacket( kind="agent_command_payload", status="created", title="PMFlow AI 指挥产物", summary=_text(result.command_payload.get("strategic_read"), 1000), payload=result.command_payload, ) ) return artifacts[:24] def _specialist_verifications(result: PMFlowAgentSpecialistResult) -> list[PMFlowAgentVerificationPacket]: verifications: list[PMFlowAgentVerificationPacket] = [] for product in result.work_products[:12]: verification = product.verification if isinstance(product.verification, dict) else {} findings = [{"type": "role_contract", "summary": _customer_gap_text(item)} for item in list(verification.get("findings") or [])[:8]] verifications.append( PMFlowAgentVerificationPacket( target_ref=f"specialist_work_product:{product.agent_id}", result=_text(verification.get("status") or product.status or "unknown", 32), retry_would_change_state=False, findings=findings, ) ) if result.council_review is not None: verifications.append( PMFlowAgentVerificationPacket( target_ref="agent_council_review", result=result.council_review.status, retry_would_change_state=False, findings=[ {"type": "council_boundary", "summary": _customer_gap_text(item.summary)} for item in list(result.council_review.conflict_ledger or [])[:8] ], ) ) if result.autonomy_gate is not None: findings = _autonomy_verification_findings(result.autonomy_gate) verifications.append( PMFlowAgentVerificationPacket( target_ref="autonomy_gate", result="passed" if not findings else "blocked", retry_would_change_state=False, findings=findings, ) ) if result.operating_cycle is not None: findings = _learning_loop_verification_findings(result.operating_cycle) verifications.append( PMFlowAgentVerificationPacket( target_ref="operating_cycle_learning_loop", result="passed" if not findings else "blocked", retry_would_change_state=False, findings=findings, ) ) return verifications[:16] def _build_command_payload( *, mission: PMFlowAIOperatingMission, project_world_state: PMFlowAIProjectWorldStatePacket, assignments: list[PMFlowAIAgentAssignment], work_products: list[PMFlowAIAgentWorkProduct], council_review: PMFlowAIAgentCouncilReview, autonomy_gate: PMFlowAIAutonomyDecision, operating_cycle: PMFlowAIOperatingCycleState, operating_plan: PMFlowAIOperatingPlan, ) -> dict[str, Any]: return { "packet_version": "pmflow_agent_os_command_payload.v1", "visibility": "customer", "mission": { "mission_id": mission.mission_id, "goal": mission.mission_goal, "autonomy_level": mission.autonomy_level, }, "project_world_state": _project_world_command_summary(project_world_state), "specialist_agents": [ { "agent_id": assignment.agent_id, "work_product": assignment.expected_work_product, "status": _product_for_agent(work_products, assignment.agent_id).status if _product_for_agent(work_products, assignment.agent_id) is not None else "blocked", "summary": _product_for_agent(work_products, assignment.agent_id).summary if _product_for_agent(work_products, assignment.agent_id) is not None else "", } for assignment in assignments ], "priority_order": list(operating_plan.priority_order or []), "safe_suggestions": list(operating_plan.safe_suggestions or []), "prepared_changes": list(operating_plan.prepared_changes or []), "requires_confirmation": list(operating_plan.requires_confirmation or []), "prepared_action_packets": list(operating_plan.prepared_action_packets or []), "confirmation_requests": list(operating_plan.confirmation_requests or []), "execution_preparations": list(operating_plan.execution_preparations or []), "blocked_actions": list(operating_plan.blocked_actions or []), "evidence_boundary": [_customer_gap_text(item) for item in list(operating_plan.evidence_boundary or [])[:12]], "strategic_read": operating_plan.strategic_read, "highest_leverage_move": operating_plan.highest_leverage_move, "agent_council_review": _dump_model(council_review), "autonomy_gate": _dump_model(autonomy_gate), "operating_cycle": _dump_model(operating_cycle), "consumed_learning_ledger": _dump_model(operating_cycle.consumed_learning_ledger), "outcome_observations": [_dump_model(item) for item in list(operating_cycle.outcome_observations or [])], "learning_ledger": _dump_model(operating_cycle.learning_ledger), "learning_updates": list(operating_plan.learning_updates or []), "next_cycle": { "status": "ready_for_next_operating_step", "summary": _operating_cycle_summary(operating_cycle), "triggers": list(operating_plan.next_cycle_triggers or []), }, "prohibited_side_effects": list(autonomy_gate.prohibited_side_effects or []), "work_product_refs": [product.work_product_id for product in work_products], } def _compose_specialist_answer( *, mission: PMFlowAIOperatingMission, project_world_state: PMFlowAIProjectWorldStatePacket, work_products: list[PMFlowAIAgentWorkProduct], council_review: PMFlowAIAgentCouncilReview, autonomy_gate: PMFlowAIAutonomyDecision, operating_cycle: PMFlowAIOperatingCycleState, operating_plan: PMFlowAIOperatingPlan, ) -> str: project_label = _project_label(project_world_state) if project_world_state.status not in {"ready", "partial"}: gaps = "、".join(_customer_gap_text(item) for item in list(project_world_state.remaining_gaps or [])[:3]) boundary = gaps or "项目运行信息还没有读完整" return ( f"我这轮已经确认对象是 {project_label}，但还不能给完整推进方案：{boundary}。" "我会保留本轮目标和对象锚点，下一步先补齐项目状态读取，再让策略、产品、执行和风险角色会审。" ) lead = _text(operating_plan.highest_leverage_move or mission.desired_outcome, 500) first_priority = _first_action(operating_plan.priority_order) first_block = _first_action(operating_plan.blocked_actions) strategist = _product_for_agent(work_products, "project_strategist") product = _product_for_agent(work_products, "product_manager") execution = _product_for_agent(work_products, "execution_planner") strategy_read = _text((strategist.summary if strategist else "") or operating_plan.strategic_read, 500) product_read = _text(product.summary if product else "", 360) execution_read = _text(execution.summary if execution else "", 360) first_sentence = f"{project_label} 现在最应该先推进：{lead}。" if first_priority and first_priority != lead: first_sentence += f"我把几个专业角色的建议收敛后，第一步落到“{first_priority}”。" else: first_sentence += "这个结论是策略、产品、执行、技术和风险角色基于同一份项目状态交叉后的结果。" support_parts = [item for item in (strategy_read, product_read, execution_read) if item] support = "；".join(support_parts[:2]) second_sentence = f"具体看，{support}。" if support else "具体执行前，还需要继续补齐可验证的项目状态和验收证据。" boundary = "确认前我不会写项目记忆、改任务、发起执行 handoff 或发布。" if first_block: boundary = f"先不要直接做“{first_block}”；{boundary}" prepared_count = len(list(autonomy_gate.prepared_action_packets or [])) confirmation_count = len(list(autonomy_gate.confirmation_requests or [])) execution_count = len(list(autonomy_gate.execution_preparations or [])) if prepared_count or execution_count: boundary += f" 我已经把 {prepared_count} 个候选动作包和 {execution_count} 个执行交接包整理好，全部停在确认门禁。" if confirmation_count: boundary += f" 当前有 {confirmation_count} 个确认请求，确认后也只会进入对应专用链路，不会在这里直接执行。" if council_review.conflict_ledger: boundary += f" 本轮会审还记录了 {len(council_review.conflict_ledger)} 个冲突或边界，已经进入后续确认。" learning_sentence = _learning_answer_sentence(operating_cycle) if learning_sentence: boundary += f" {learning_sentence}" return f"{first_sentence}\n\n{second_sentence}\n\n{boundary}".strip() def _attach_cycle_to_plan( *, operating_plan: PMFlowAIOperatingPlan, operating_cycle: PMFlowAIOperatingCycleState, ) -> PMFlowAIOperatingPlan: learning_updates = [ { "learning_key": item.get("learning_key"), "summary": item.get("summary"), "outcome_type": item.get("outcome_type"), "outcome_status": item.get("outcome_status"), } for item in list(operating_cycle.learning_ledger.learning_items or [])[-12:] if isinstance(item, dict) ] return operating_plan.model_copy( update={ "operating_cycle": operating_cycle.model_dump(mode="json", exclude_none=True), "consumed_learning_ledger": operating_cycle.consumed_learning_ledger.model_dump(mode="json", exclude_none=True), "learning_updates": learning_updates, "next_cycle_triggers": list(operating_cycle.next_cycle_triggers or []), } ) def _operating_cycle_summary(operating_cycle: PMFlowAIOperatingCycleState | None) -> str: if operating_cycle is None: return "未建立持续推进闭环。" consumed_count = len(list(operating_cycle.consumed_learning_ledger.learning_items or [])) observation_count = len(list(operating_cycle.outcome_observations or [])) learning_count = len(list(operating_cycle.learning_ledger.learning_items or [])) trigger_count = len(list(operating_cycle.next_cycle_triggers or [])) return ( f"已读取 {consumed_count} 条历史学习，本轮记录 {observation_count} 条结果观察，" f"学习账本累计 {learning_count} 条，后续可由 {trigger_count} 类事件继续触发。" ) def _learning_ledger_summary(operating_cycle: PMFlowAIOperatingCycleState) -> str: ledger = operating_cycle.learning_ledger return ( f"学习项 {len(list(ledger.learning_items or []))} 条，" f"决策结果 {len(list(ledger.decision_outcomes or []))} 条，" f"执行结果 {len(list(ledger.execution_outcomes or []))} 条，" f"实验结果 {len(list(ledger.experiment_results or []))} 条。" ) def _learning_loop_verification_findings(operating_cycle: PMFlowAIOperatingCycleState) -> list[dict[str, Any]]: findings: list[dict[str, Any]] = [] if not operating_cycle.learning_ledger.scope_ref: findings.append({"type": "learning_scope_missing", "summary": "学习账本缺少项目对象范围。"}) if operating_cycle.learning_ledger.scope_ref != operating_cycle.consumed_learning_ledger.scope_ref: findings.append({"type": "learning_scope_mismatch", "summary": "读取的学习账本和更新后的账本范围不一致。"}) if not list(operating_cycle.outcome_observations or []): findings.append({"type": "outcome_observation_missing", "summary": "本轮没有记录结果观察，后续运行无法学习。"}) if not list(operating_cycle.next_cycle_triggers or []): findings.append({"type": "next_cycle_trigger_missing", "summary": "本轮没有生成下一轮触发合同。"}) return findings[:8] def _learning_answer_sentence(operating_cycle: PMFlowAIOperatingCycleState) -> str: consumed_count = len(list(operating_cycle.consumed_learning_ledger.learning_items or [])) observation_count = len(list(operating_cycle.outcome_observations or [])) if not consumed_count: return "" parts: list[str] = [] parts.append(f"我也读取了上一轮 {consumed_count} 条学习记录来约束本轮计划。") if observation_count: parts.append(f"这轮已记录 {observation_count} 条结果观察，后面继续问、确认或有执行结果回流时，会接着这份状态推进。") return "".join(parts) def _autonomy_gate_summary(autonomy_gate: PMFlowAIAutonomyDecision | None) -> str: if autonomy_gate is None: return "未生成确认门禁。" prepared_count = len(list(autonomy_gate.prepared_action_packets or [])) confirmation_count = len(list(autonomy_gate.confirmation_requests or [])) execution_count = len(list(autonomy_gate.execution_preparations or [])) blocked_count = len(list(autonomy_gate.blocked_actions or [])) return ( f"已准备 {prepared_count} 个候选动作包、{confirmation_count} 个确认请求、" f"{execution_count} 个执行交接包；{blocked_count} 个动作被边界或风险阻断，未执行任何写入。" ) def _prepared_action_summary(packet: dict[str, Any]) -> str: family = _text(packet.get("action_family"), 120) title = _text(packet.get("title"), 240) if family == "task_change": return f"任务变更候选：{title}。确认前不会改任务。" if family == "experiment_plan": return f"用户验证实验候选：{title}。确认前不会写入项目事实。" if family == "project_memory_candidate": return f"项目记忆候选：{title}。确认前不会写 Project Memory。" return f"待确认动作：{title}。确认前不会执行。" def _autonomy_verification_findings(autonomy_gate: PMFlowAIAutonomyDecision) -> list[dict[str, Any]]: findings: list[dict[str, Any]] = [] prepared_packets = [dict(item) for item in list(autonomy_gate.prepared_action_packets or []) if isinstance(item, dict)] confirmation_requests = [dict(item) for item in list(autonomy_gate.confirmation_requests or []) if isinstance(item, dict)] execution_preparations = [dict(item) for item in list(autonomy_gate.execution_preparations or []) if isinstance(item, dict)] requested_ids = { str(requested_for) for request in confirmation_requests for requested_for in list(request.get("requested_for") or []) } for packet in prepared_packets: prepared_id = str(packet.get("prepared_action_id") or "") if not bool(packet.get("confirmation_required")): findings.append({"type": "confirmation_boundary", "summary": f"{packet.get('title') or '候选动作'} 缺少确认要求。"}) if prepared_id and prepared_id not in requested_ids: findings.append({"type": "confirmation_request_missing", "summary": f"{packet.get('title') or '候选动作'} 没有关联确认请求。"}) for request in confirmation_requests: if bool(request.get("can_execute_after_confirmation")): findings.append({"type": "direct_execution_block", "summary": f"{request.get('title') or '确认请求'} 不能在确认后直接执行，必须进入专用链路。"}) for preparation in execution_preparations: if str(preparation.get("readiness") or "") == "prepared_waiting_confirmation" and not list(preparation.get("required_confirmations") or []): findings.append({"type": "execution_confirmation_missing", "summary": "执行交接准备包缺少确认请求。"}) prohibited = set(str(item) for item in list(autonomy_gate.prohibited_side_effects or [])) for required in ("write_project_memory", "execute_handoff"): if required not in prohibited: findings.append({"type": "prohibited_side_effect_missing", "summary": f"确认门禁缺少 {required} 禁止项。"}) return findings[:12] def _project_world_command_summary(project_world_state: PMFlowAIProjectWorldStatePacket) -> dict[str, Any]: return { "packet_version": project_world_state.packet_version, "world_state_id": project_world_state.world_state_id, "status": project_world_state.status, "primary_project_ref": project_world_state.primary_project_ref, "coverage": dict(project_world_state.coverage or {}), "facet_summaries": [ { "facet_key": facet.facet_key, "display_label": facet.display_label, "status": facet.status, "summary": facet.summary, } for facet in list(project_world_state.facets or [])[:12] ], "remaining_gaps": [_customer_gap_text(item) for item in list(project_world_state.remaining_gaps or [])[:12]], } def _project_world_summary(project_world_state: PMFlowAIProjectWorldStatePacket) -> str: coverage = dict(project_world_state.coverage or {}) covered = len(list(coverage.get("covered_facets") or [])) partial = len(list(coverage.get("partial_facets") or [])) missing = len(list(coverage.get("missing_facets") or [])) if project_world_state.status == "missing_scope": return "缺少强项目对象，未读取项目世界状态。" return f"已读取 {len(project_world_state.project_refs)} 个项目对象；覆盖 {covered} 项、部分覆盖 {partial} 项、缺失 {missing} 项。" def _project_label(project_world_state: PMFlowAIProjectWorldStatePacket) -> str: for packet in list(project_world_state.project_operating_evidence_packets or [])[:4]: if not isinstance(packet, dict): continue title = _text(packet.get("project_title") or packet.get("title"), 120) if title: return f"项目“{title}”" for facet in list(project_world_state.facets or [])[:6]: for fact in list(facet.facts or [])[:12]: if not isinstance(fact, dict): continue if fact.get("key") == "project_title" and _text(fact.get("value"), 120): return f"项目“{_text(fact.get('value'), 120)}”" return project_world_state.primary_project_ref or "这个项目" def _first_action(rows: list[dict[str, Any]]) -> str: for item in list(rows or [])[:8]: if not isinstance(item, dict): continue title = _text(item.get("title") or item.get("display_label"), 240) if title: return title return "" def _work_product_title(product: PMFlowAIAgentWorkProduct) -> str: titles = { "project_strategist": "策略判断", "customer_discovery": "用户验证计划", "product_manager": "产品范围计划", "technical_architect": "技术路径评估", "execution_planner": "执行推进计划", "risk_reviewer": "风险审查", } return titles.get(product.agent_id, product.product_type or "专业工作产物") def _product_for_agent(work_products: list[PMFlowAIAgentWorkProduct], agent_id: str) -> PMFlowAIAgentWorkProduct | None: return next((item for item in work_products if item.agent_id == agent_id), None) def _strong_project_ref( *, request: RAGAskRequest, conversation_context: dict[str, Any] | None, tenant_project_context: dict[str, Any] | None, ) -> str | None: request_project_id = _safe_int(getattr(request, "project_id", None)) if request_project_id is not None: return f"project:{request_project_id}" context_project_id = _safe_int((conversation_context or {}).get("current_project_id")) if context_project_id is not None: return f"project:{context_project_id}" surface = _surface_context(tenant_project_context) focus = surface.get("current_focus_object") if isinstance(focus, dict): object_ref = _text(focus.get("object_ref"), 160) object_type = _text(focus.get("object_type"), 80) object_id = _safe_int(focus.get("object_id")) if object_ref.startswith("project:"): return object_ref if object_type == "project" and object_id is not None: return f"project:{object_id}" return None def _surface_context(tenant_project_context: dict[str, Any] | None) -> dict[str, Any]: surface = dict(tenant_project_context or {}).get("surface_context") return dict(surface) if isinstance(surface, dict) else {} def _project_id_from_ref(project_ref: str) -> int | None: prefix, _, raw_id = str(project_ref or "").partition(":") if prefix == "project" and raw_id.strip().isdigit(): return int(raw_id) return None def _customer_gap_text(value: Any) -> str: text = _text(value, 500) normalized = text.lower() if "_" in normalized: if "evidence" in normalized and "missing" in normalized: return "这个角色产物还缺少可追溯依据" if "structured" in normalized and "empty" in normalized: return "这个角色产物没有形成结构化结果" if "execution" in normalized and "missing" in normalized: return "执行摘要暂时缺失" if "truth" in normalized and "missing" in normalized: return "项目真实状态暂时缺失" if "project" in normalized and ("world" in normalized or "state" in normalized): return "项目运行信息还没有读完整" return text.replace("_", " ") if len(text) < 80 and "_" in text else text def _dump_model(value: Any) -> dict[str, Any]: if value is None: return {} if hasattr(value, "model_dump"): dumped = value.model_dump(mode="json", exclude_none=True) return dumped if isinstance(dumped, dict) else {} return dict(value) if isinstance(value, dict) else {} def _text(value: Any, limit: int) -> str: return " ".join(str(value or "").strip().split())[:limit] def _safe_int(value: Any) -> int | None: try: if value is None or value == "": return None return int(value) except Exception: return None __all__ = ["PMFlowAgentSpecialistResult", "PMFlowAgentSpecialistRuntimeService"]