"""Operating run kernel for PMFlow AI multi-agent runtime.""" from __future__ import annotations import json import time import re from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from ...models import User from ...schemas.pmflow_ai_operating_runtime import ( PMFlowAIAgentAssignment, PMFlowAIAgentWorkProduct, PMFlowAIAutonomyDecision, PMFlowAICustomerSafeClaim, PMFlowAICustomerSafeClaimLedger, PMFlowAIEvidenceLedgerEntry, PMFlowAIOperatingCycleState, PMFlowAIOperatingCommandPayload, PMFlowAIOperatingMission, PMFlowAIOperatingMissionCriticResult, PMFlowAIOperatingPlan, PMFlowAIOperatingRunEvent, PMFlowAIOperatingRunPacket, PMFlowAIProductObjectProjection, PMFlowAIProjectWorldFacet, PMFlowAIProjectWorldStatePacket, new_operating_id, utc_now_iso, ) from ...schemas.rag import RAGAskRequest from ..ai_function_runtime_profile_service import AIFunctionRuntimeProfileService from ..llm_service import llm_service from ..pmflow_ai_agent_runtime import PMFlowAIAgentRunService from ..pmflow_ai_agent_runtime.agent_evidence_quality import ( is_usable_research_claim_text, looks_like_source_noise, normalize_evidence_text, ) from .agent_registry import PMFlowAIOperatingAgentRegistry from .autonomy import PMFlowAIAutonomyPolicyService from .council import PMFlowAIAgentCouncilService from .cycle import OperatingCycleService from .mission_critic import PMFlowAIOperatingMissionCriticService, merge_missing_roles_into_mission from .mission_service import PMFlowAIOperatingMissionService from .project_world_state_service import PMFlowAIProjectWorldStateService from .specialist_agents import PMFlowAISpecialistAgentService from ..prompt_runtime_service import PromptRuntimeService from ..runtime_audit_context import bind_agent_audit_context PROMPT_KEY_PMFLOW_AI_OPERATING_ANSWER_COMPOSER = "PROMPT_PMFLOW_AI_OPERATING_ANSWER_COMPOSER" FUNCTION_NAME_PMFLOW_AI_OPERATING_ANSWER_COMPOSER = "pmflow_ai_operating_answer_composer" SCENE_KEY_PMFLOW_AI_OPERATING_RUNTIME = "space_pmflow_ai_operating_runtime" class PMFlowAIOperatingRunService: """Create mission-first multi-agent operating runs.""" def __init__( self, db: AsyncSession, *, tenant_id: int, mission_service: PMFlowAIOperatingMissionService | None = None, agent_registry: PMFlowAIOperatingAgentRegistry | None = None, project_world_state_service: PMFlowAIProjectWorldStateService | None = None, specialist_agent_service: PMFlowAISpecialistAgentService | None = None, mission_critic_service: PMFlowAIOperatingMissionCriticService | None = None, council_service: PMFlowAIAgentCouncilService | None = None, autonomy_policy_service: PMFlowAIAutonomyPolicyService | None = None, operating_cycle_service: OperatingCycleService | None = None, delivery_runtime: Any | None = None, ) -> None: self.db = db self.tenant_id = int(tenant_id) self.agent_registry = agent_registry or PMFlowAIOperatingAgentRegistry() self.mission_service = mission_service or PMFlowAIOperatingMissionService( db=db, tenant_id=self.tenant_id, agent_registry=self.agent_registry, ) 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( db=db, tenant_id=self.tenant_id, agent_registry=self.agent_registry, ) self.mission_critic_service = mission_critic_service or PMFlowAIOperatingMissionCriticService( db=db, tenant_id=self.tenant_id, agent_registry=self.agent_registry, ) 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() self.delivery_runtime = delivery_runtime or PMFlowAIAgentRunService(db, tenant_id=self.tenant_id) async def _compose_operating_final_answer( self, *, request: RAGAskRequest, mission: PMFlowAIOperatingMission, customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, work_products: list[PMFlowAIAgentWorkProduct], council_review: Any, autonomy_gate: Any, operating_plan: PMFlowAIOperatingPlan, product_object_projections: list[PMFlowAIProductObjectProjection], delivery_answer: str, ) -> str: generate = getattr(llm_service, "generate", None) if not callable(generate) or not callable(getattr(self.db, "execute", None)): return "" try: action_context = { "prepared_action_packets": list(operating_plan.prepared_action_packets or [])[:8], "confirmation_requests": list(operating_plan.confirmation_requests or [])[:8], "execution_preparations": list(operating_plan.execution_preparations or [])[:8], "product_object_projections": [ item.model_dump(mode="json", exclude_none=True) for item in product_object_projections[:8] ], } operating_decision = { "council_review": council_review.model_dump(mode="json", exclude_none=True) if hasattr(council_review, "model_dump") else dict(council_review or {}), "autonomy_gate": autonomy_gate.model_dump(mode="json", exclude_none=True) if hasattr(autonomy_gate, "model_dump") else dict(autonomy_gate or {}), } prompt_result = await PromptRuntimeService(self.db).render( tenant_id=self.tenant_id, prompt_key=PROMPT_KEY_PMFLOW_AI_OPERATING_ANSWER_COMPOSER, question=str(request.question or "").strip()[:1600], mission_json=_json_for_prompt(mission.model_dump(mode="json", exclude_none=True), limit=7000), claim_ledger_json=_json_for_prompt( customer_safe_claim_ledger.model_dump(mode="json", exclude_none=True), limit=7000, ), work_products_json=_json_for_prompt( [item.model_dump(mode="json", exclude_none=True) for item in work_products], limit=14000, ), operating_decision_json=_json_for_prompt(operating_decision, limit=7000), action_context_json=_json_for_prompt(action_context, limit=8000), delivery_answer=str(delivery_answer or "").strip()[:5000], ) runtime_profile = await AIFunctionRuntimeProfileService(self.db).resolve_profile( function_name=FUNCTION_NAME_PMFLOW_AI_OPERATING_ANSWER_COMPOSER, scene_key=SCENE_KEY_PMFLOW_AI_OPERATING_RUNTIME, prompt_key=PROMPT_KEY_PMFLOW_AI_OPERATING_ANSWER_COMPOSER, ) with bind_agent_audit_context(agent_step_key="operating_answer_composer"): raw = await generate( prompt_result.prompt, db=self.db, tenant_id=self.tenant_id, function_name=FUNCTION_NAME_PMFLOW_AI_OPERATING_ANSWER_COMPOSER, 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_answer_runtime", "scene_contract": "post_operating_final_answer", **runtime_profile.routing_meta(), **prompt_result.routing_meta(), }, ) answer = _plain_customer_text(raw, limit=8000) if not answer or _looks_like_internal_answer_text(answer): return "" return answer except Exception: return "" async def run_turn( self, *, request: RAGAskRequest, current_user: User, conversation_context: dict[str, Any] | None, tenant_project_context: dict[str, Any] | None, started_at: float | None = None, event_sink: Any | None = None, ) -> dict[str, Any]: started = started_at or time.perf_counter() delivery_kwargs = { "request": request, "current_user": current_user, "conversation_context": conversation_context, "tenant_project_context": tenant_project_context, "started_at": started, } if event_sink is not None: delivery_kwargs["event_sink"] = event_sink delivery_payload = await self.delivery_runtime.run_turn(**delivery_kwargs) if _agent_os_delivery_can_stand_alone(delivery_payload): return _standalone_agent_os_delivery_payload( delivery_payload=delivery_payload, started=started, ) mission = await self.mission_service.build_mission( request=request, current_user=current_user, conversation_context=conversation_context, tenant_project_context=tenant_project_context, ) initial_critic = await self.mission_critic_service.review( user_question=str(request.question or ""), mission=mission, artifact_slots=[], evidence_lanes=[], current_object_binding=_current_object_binding_from_context( conversation_context=conversation_context, tenant_project_context=tenant_project_context, ), ) mission = merge_missing_roles_into_mission(mission=mission, critic=initial_critic) 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) events = [ PMFlowAIOperatingRunEvent( event_type="mission_created", summary="已建立本轮 Operating Mission，先定义要推进的目标和确认边界。", ), PMFlowAIOperatingRunEvent( event_type="agents_assigned", summary=f"已分派 {len(assignments)} 个专业 agent 产出结构化工作结果。", ), PMFlowAIOperatingRunEvent( event_type="project_world_state_read", summary=_project_world_state_event_summary(project_world_state), ), ] if _delivery_is_terminal_boundary(delivery_payload): return _boundary_delivery_payload( delivery_payload=delivery_payload, mission=mission, started=started, ) customer_safe_claim_ledger = _build_customer_safe_claim_ledger(delivery_payload, mission=mission) project_world_state = _merge_delivery_research_evidence( project_world_state, delivery_payload, customer_safe_claim_ledger=customer_safe_claim_ledger, ) if _delivery_can_finish_without_deep_operating_chain( delivery_payload=delivery_payload, mission=mission, project_world_state=project_world_state, customer_safe_claim_ledger=customer_safe_claim_ledger, ): return _shallow_operating_delivery_payload( delivery_payload=delivery_payload, mission=mission, project_world_state=project_world_state, customer_safe_claim_ledger=customer_safe_claim_ledger, started=started, ) post_evidence_critic = await self.mission_critic_service.review( user_question=str(request.question or ""), mission=mission, artifact_slots=_artifact_slots_from_delivery(delivery_payload), evidence_lanes=_evidence_lanes_from_world_state( project_world_state=project_world_state, delivery_payload=delivery_payload, customer_safe_claim_ledger=customer_safe_claim_ledger, ), current_object_binding=_current_object_binding_from_delivery( delivery_payload=delivery_payload, conversation_context=conversation_context, tenant_project_context=tenant_project_context, ), ) mission = merge_missing_roles_into_mission(mission=mission, critic=post_evidence_critic) assignments = self.agent_registry.assignments_for_mission(mission) async_producer = getattr(self.specialist_agent_service, "produce_work_products_async", None) if callable(async_producer): work_products = await async_producer( mission=mission, assignments=assignments, project_world_state=project_world_state, learning_ledger={}, ) else: 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) product_object_projections = [ PMFlowAIProductObjectProjection.model_validate(item) for item in list(operating_plan.product_object_projections or []) if isinstance(item, dict) ] command_payload = _build_command_payload( mission=mission, assignments=assignments, 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, product_object_projections=product_object_projections, delivery_payload=delivery_payload, customer_safe_claim_ledger=customer_safe_claim_ledger, ) status = _operating_status(operating_plan) specialist_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, product_object_projections=product_object_projections, operating_plan=operating_plan, ) operating_run = PMFlowAIOperatingRunPacket( run_id=new_operating_id("orun"), mission=mission, status=status, assignments=assignments, project_world_state=project_world_state, customer_safe_claim_ledger=customer_safe_claim_ledger, work_products=work_products, agent_council_review=council_review, autonomy_gate=autonomy_gate, operating_cycle=operating_cycle, product_object_projections=product_object_projections, operating_plan=operating_plan, events=[ *events, *_mission_critic_events( initial_critic=initial_critic, post_evidence_critic=post_evidence_critic, ), *[ PMFlowAIOperatingRunEvent( event_type="specialist_work_product_ready", summary=f"{product.product_type} 已生成。", agent_id=product.agent_id, assignment_id=product.assignment_id, ) for product in work_products ], PMFlowAIOperatingRunEvent( event_type="agent_council_review_ready", summary=council_review.resolution_summary, ), PMFlowAIOperatingRunEvent( event_type="operating_plan_compiled", summary="已把 Agent Council 会审结果编译进 Operating Plan；回答交付只作为客户侧摘要适配。", ), PMFlowAIOperatingRunEvent( event_type="autonomy_gate_applied", summary=_autonomy_gate_event_summary(autonomy_gate), ), PMFlowAIOperatingRunEvent( event_type="learning_ledger_updated", summary=_learning_event_summary(operating_cycle), ), ], delivery_ref={ "adapter": "answer_delivery", "agent_run_id": _agent_run_id(delivery_payload), "answer_mode": str(delivery_payload.get("answer_mode") or ""), }, completed_at=utc_now_iso(), ) delivery_answer = _delivery_customer_answer(delivery_payload) if _answer_conflicts_with_evidence( delivery_answer, project_world_state=project_world_state, delivery_payload=delivery_payload, customer_safe_claim_ledger=customer_safe_claim_ledger, ): delivery_answer = "" if _answer_conflicts_with_evidence( specialist_answer, project_world_state=project_world_state, delivery_payload=delivery_payload, customer_safe_claim_ledger=customer_safe_claim_ledger, ): specialist_answer = "" operating_answer = await self._compose_operating_final_answer( request=request, mission=mission, customer_safe_claim_ledger=customer_safe_claim_ledger, work_products=work_products, council_review=council_review, autonomy_gate=autonomy_gate, operating_plan=operating_plan, product_object_projections=product_object_projections, delivery_answer=delivery_answer, ) if not _should_keep_delivery_open_reasoning_answer( delivery_payload=delivery_payload, project_world_state=project_world_state, ) else "" final_answer = ( operating_answer or _compose_answer_from_terminal_and_artifacts( delivery_payload=delivery_payload, specialist_answer=specialist_answer, project_world_state=project_world_state, customer_safe_claim_ledger=customer_safe_claim_ledger, ) or specialist_answer or delivery_answer ) payload = dict(delivery_payload or {}) payload["answer"] = final_answer payload["reply_text"] = final_answer if isinstance(payload.get("answer_canvas"), dict): payload["answer_canvas"] = { **dict(payload["answer_canvas"]), "direct_answer": final_answer, "headline": "PMFlow AI", } else: payload["answer_canvas"] = { "direct_answer": final_answer, "headline": "PMFlow AI", "next_actions": [], "action_candidates": [], } payload["answer_canvas"] = _merge_operating_canvas( payload["answer_canvas"], mission=mission, assignments=assignments, work_products=work_products, operating_plan=operating_plan, product_object_projections=product_object_projections, ) payload["processing_stage"] = "multi_agent_operating_runtime" payload["answer_mode"] = "operating_runtime" payload["policy_version"] = "pmflow-multi-agent-operating-runtime-v1" payload["operating_summary"] = specialist_answer payload["operating_mission"] = mission.model_dump(mode="json", exclude_none=True) payload["operating_mission_critic"] = post_evidence_critic.model_dump(mode="json", exclude_none=True) payload["operating_mission_critic_reviews"] = [ initial_critic.model_dump(mode="json", exclude_none=True), post_evidence_critic.model_dump(mode="json", exclude_none=True), ] payload["project_world_state_packet"] = project_world_state.model_dump(mode="json", exclude_none=True) payload["customer_safe_claim_ledger"] = customer_safe_claim_ledger.model_dump(mode="json", exclude_none=True) payload["evidence_ledger"] = [item.model_dump(mode="json", exclude_none=True) for item in project_world_state.evidence_ledger] payload["agent_assignments"] = [item.model_dump(mode="json", exclude_none=True) for item in assignments] payload["agent_work_products"] = [item.model_dump(mode="json", exclude_none=True) for item in work_products] payload["agent_council_review"] = council_review.model_dump(mode="json", exclude_none=True) payload["conflict_ledger"] = [item.model_dump(mode="json", exclude_none=True) for item in council_review.conflict_ledger] payload["autonomy_gate"] = autonomy_gate.model_dump(mode="json", exclude_none=True) payload["prepared_action_packets"] = list(operating_plan.prepared_action_packets or []) payload["confirmation_requests"] = list(operating_plan.confirmation_requests or []) payload["execution_preparations"] = list(operating_plan.execution_preparations or []) payload["product_object_projections"] = [item.model_dump(mode="json", exclude_none=True) for item in product_object_projections] payload["operating_cycle"] = operating_cycle.model_dump(mode="json", exclude_none=True) payload["consumed_learning_ledger"] = operating_cycle.consumed_learning_ledger.model_dump(mode="json", exclude_none=True) payload["outcome_observations"] = [item.model_dump(mode="json", exclude_none=True) for item in operating_cycle.outcome_observations] payload["learning_ledger"] = operating_cycle.learning_ledger.model_dump(mode="json", exclude_none=True) payload["learning_updates"] = list(operating_plan.learning_updates or []) payload["next_cycle_triggers"] = list(operating_plan.next_cycle_triggers or []) payload["operating_plan"] = operating_plan.model_dump(mode="json", exclude_none=True) payload["operating_run_packet"] = operating_run.model_dump(mode="json", exclude_none=True) payload["command_payload"] = command_payload.model_dump(mode="json", exclude_none=True) payload["answer_process_packet"] = _merge_operating_process( payload.get("answer_process_packet"), mission=mission, assignments=assignments, work_products=work_products, project_world_state=project_world_state, council_review=council_review.model_dump(mode="json", exclude_none=True), autonomy_gate=autonomy_gate.model_dump(mode="json", exclude_none=True), operating_cycle=operating_cycle.model_dump(mode="json", exclude_none=True), initial_critic=initial_critic, post_evidence_critic=post_evidence_critic, customer_safe_claim_ledger=customer_safe_claim_ledger, total_duration_ms=int((time.perf_counter() - started) * 1000), ) return payload def _merge_delivery_research_evidence( project_world_state: PMFlowAIProjectWorldStatePacket, delivery_payload: dict[str, Any], *, customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, ) -> PMFlowAIProjectWorldStatePacket: """Promote inner read-loop evidence into the operating world state.""" deliverable = delivery_payload.get("deliverable_packet") if isinstance(delivery_payload.get("deliverable_packet"), dict) else {} workspace = delivery_payload.get("agent_workspace") if isinstance(delivery_payload.get("agent_workspace"), dict) else {} observations = _delivery_observations(delivery_payload, workspace=workspace) source_rows = _external_source_rows(observations) research_lanes = _external_research_lanes(observations) claims = [item.claim_text for item in list(customer_safe_claim_ledger.claims or []) if item.claim_text] gaps = _gap_texts(workspace, deliverable, list(project_world_state.remaining_gaps or [])) if not source_rows and not claims and not gaps: return project_world_state.model_copy(update={"customer_safe_claim_ledger": customer_safe_claim_ledger}) source_count = len(source_rows) claim_count = len(claims) status = "ready" if claim_count >= 3 and source_count >= 2 else "partial" summary = f"已完成外部研究读取：{source_count} 个候选来源、{claim_count} 条可用判断、{len(gaps)} 个剩余证据缺口。" facet = PMFlowAIProjectWorldFacet( facet_key="external_research_evidence", display_label="外部研究证据", status=status, summary=summary, facts=[ {"fact_key": "source_candidates", "value": source_rows[:8]}, {"fact_key": "usable_claims", "value": claims[:8]}, {"fact_key": "evidence_gaps", "value": gaps[:8]}, {"fact_key": "research_lanes", "value": research_lanes[:8]}, ], evidence_refs=_dedupe([str(item.get("observation_id") or item.get("id") or "") for item in observations])[:12] or ["agent_research_delivery"], gaps=gaps[:12], ) coverage = dict(project_world_state.coverage or {}) covered = _dedupe([*list(coverage.get("covered_facets") or []), facet.facet_key]) missing = [item for item in list(coverage.get("missing_facets") or []) if item != facet.facet_key] remaining = [gap for gap in gaps if gap and not _looks_like_internal_token(str(gap))] if not remaining and status != "ready": remaining = ["还缺少跨来源验证的公开材料"] return project_world_state.model_copy( update={ "status": "ready" if project_world_state.status == "partial" and status == "ready" else project_world_state.status, "facets": [*[item for item in project_world_state.facets if item.facet_key != facet.facet_key], facet], "customer_safe_claim_ledger": customer_safe_claim_ledger, "evidence_ledger": [ *list(project_world_state.evidence_ledger or []), PMFlowAIEvidenceLedgerEntry( project_ref=project_world_state.primary_project_ref or "space:external_research", facet_key=facet.facet_key, source_ref="agent_research_delivery", summary=summary, confidence="high" if status == "ready" else "medium", ), ], "coverage": { **coverage, "covered_facets": covered, "missing_facets": missing, "coverage_score": max(float(coverage.get("coverage_score") or 0.0), 0.45 if status == "partial" else 0.65), }, "remaining_gaps": remaining[:24], } ) def _agent_os_delivery_can_stand_alone(delivery_payload: dict[str, Any]) -> bool: answer = _delivery_customer_answer(delivery_payload) if not answer: return False if isinstance(delivery_payload.get("interaction_request"), dict) and delivery_payload.get("interaction_request"): return False packet = delivery_payload.get("agent_run_packet") if isinstance(delivery_payload.get("agent_run_packet"), dict) else {} stop_reason = str(packet.get("stop_reason") or "").strip() if stop_reason not in {"planner_ready_to_answer", "ready_to_answer", "completed"}: return False workspace = delivery_payload.get("agent_workspace") if isinstance(delivery_payload.get("agent_workspace"), dict) else {} goal_frame = workspace.get("goal_frame") if isinstance(workspace.get("goal_frame"), dict) else {} if str(goal_frame.get("source_policy") or "").strip() == "open_reasoning": return True tool_steps = list(packet.get("steps") or []) if any(str(item.get("tool_name") or "").strip() in {"external_search", "inspect_source_detail"} for item in tool_steps if isinstance(item, dict)): return True return False def _standalone_agent_os_delivery_payload( *, delivery_payload: dict[str, Any], started: float, ) -> dict[str, Any]: payload = dict(delivery_payload or {}) answer = _delivery_customer_answer(payload) payload["answer"] = answer payload["reply_text"] = answer payload["processing_stage"] = "pmflow_agent_runtime" payload["answer_mode"] = "pmflow_agent_runtime" payload["policy_version"] = str(payload.get("policy_version") or "pmflow-agent-os-v2") payload["agent_assignments"] = [] payload["agent_work_products"] = [] payload["prepared_action_packets"] = [] payload["confirmation_requests"] = [] payload["execution_preparations"] = [] payload["product_object_projections"] = [] packet = payload.get("answer_process_packet") if isinstance(payload.get("answer_process_packet"), dict) else {} packet["packet_version"] = "customer_answer_process.v1" packet["visibility"] = "customer" packet["total_duration_ms"] = max(0, int((time.perf_counter() - started) * 1000)) packet["wall_clock_duration_ms"] = packet["total_duration_ms"] payload["answer_process_packet"] = packet return payload def _external_source_rows(observations: list[dict[str, Any]]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for observation in observations: source = str(observation.get("source_kind") or observation.get("source") or "").strip() facts = _dicts(observation.get("facts")) if source not in {"external_search", "inspect_source_detail"} and not any(str(item.get("type") or "") in {"external_evidence", "external_source_text"} for item in facts): continue for fact in facts: fact_type = str(fact.get("type") or "").strip() if fact_type not in {"external_evidence", "external_source_text"}: continue url = str(fact.get("url") or "").strip() title = str(fact.get("title") or url or "公开来源").strip() summary = str(fact.get("summary") or fact.get("text") or "").strip() row_summary = _external_source_row_summary(summary) if not url and not row_summary: continue if looks_like_source_noise(title): continue if not row_summary: continue rows.append({"title": title[:220], "url": url[:500], "summary": row_summary[:420], "source": source}) return _dedupe_dicts_local(rows, key="url")[:12] def _external_research_lanes(observations: list[dict[str, Any]]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] for observation in observations: for fact in _dicts(observation.get("facts")): fact_type = str(fact.get("type") or "") if fact_type == "external_search_query": for lane in _dicts(fact.get("research_lanes")): item = { "lane": _plain_customer_text(lane.get("lane") or lane.get("name") or lane.get("title"), limit=160), "purpose": _plain_customer_text(lane.get("purpose") or lane.get("goal"), limit=220), "minimum_useful_signal": _plain_customer_text( lane.get("minimum_useful_signal") or lane.get("success_criteria"), limit=260, ), } if item["lane"]: rows.append({key: value for key, value in item.items() if value}) continue if fact_type != "external_search_hypotheses": continue for branch in _dicts(fact.get("branch_searches")): item = { "lane": _plain_customer_text(branch.get("meaning") or branch.get("purpose"), limit=160), "purpose": _plain_customer_text(branch.get("purpose"), limit=220), "query": _plain_customer_text(branch.get("query"), limit=220), } if item["lane"]: rows.append({key: value for key, value in item.items() if value}) return _dedupe_dicts_local(rows, key="lane")[:12] def _external_source_row_summary(summary: Any) -> str: text = normalize_evidence_text(summary) if not text: return "" if not looks_like_source_noise(text): return text[:420] fragments = re.split(r"(?<=[。！？.!?])\s+|[\n\r]+|；|;", text) for fragment in [*fragments, *_research_signal_windows(text)]: fragment = fragment.strip(" \t|-—:：;；,，") if len(fragment) < 36: continue if looks_like_source_noise(fragment): continue if is_usable_research_claim_text(fragment): return fragment[:420] return "" def _research_signal_windows(text: str) -> list[str]: value = re.sub(r"\s+", " ", str(text or "")).strip() if not value: return [] pattern = re.compile( r"\b(?:19|20)\d{2}(?:\b|年)|\d[\d,]*(?:\.\d+)?\s*(?:%|万|亿|千|百|million|billion|trillion|k|m|bn)(?:\b|元|人|次|个|条|家)?|\$|€|£|¥|人民币|美元|营收|收入|成本|融资|投资|增长|下降|需求|供给|采用|参会|预算|赞助|转化|销售|reported|grew|declined|reached|increased|decreased|expected|shows|forecast|projected|valued|budget|sponsor|attendance|attendee|attendees|adoption|revenue", flags=re.I, ) rows: list[str] = [] for match in pattern.finditer(value): fragment = value[max(0, match.start() - 220) : min(len(value), match.end() + 260)] fragment = _trim_page_shell_edges(fragment=fragment, signal_text=match.group(0)) if len(fragment) > 520: fragment = fragment[:520].rsplit(" ", 1)[0] or fragment[:520] if 36 <= len(fragment) <= 520: rows.append(fragment) return _dedupe(rows)[:20] def _trim_page_shell_edges(*, fragment: str, signal_text: str) -> str: value = str(fragment or "").strip(" \t|-—:：;；,，") lowered = value.lower() signal_index = lowered.find(str(signal_text or "").lower()) if signal_index < 0: return value markers = ( "subscribe", "sign in", "newsletter", "share", "privacy", "terms", "advertisement", "contact us", "read more", "premium", "site map", "copyright", "登录", "注册", "隐私政策", "用户协议", "广告", "联系我们", "订阅", ) cut = 0 for marker in markers: idx = lowered.rfind(marker, 0, signal_index) if idx >= 0: cut = max(cut, idx + len(marker)) if cut: value = value[cut:].strip(" \t|-—:：;；,，") lowered = value.lower() suffix_cut = len(value) for marker in markers: idx = lowered.find(marker, max(0, len(str(signal_text or "")) + 24)) if idx >= 0: suffix_cut = min(suffix_cut, idx) return value[:suffix_cut].strip(" \t|-—:：;；,，") def _delivery_observations(delivery_payload: dict[str, Any], *, workspace: dict[str, Any]) -> list[dict[str, Any]]: rows: list[dict[str, Any]] = [] rows.extend(_dicts(delivery_payload.get("agent_os_observations"))) packet = delivery_payload.get("agent_run_packet") if isinstance(delivery_payload.get("agent_run_packet"), dict) else {} rows.extend(_dicts(packet.get("observations"))) rows.extend(_dicts(workspace.get("observations"))) rows.extend(_dicts(workspace.get("observation_ledger"))) normalized: list[dict[str, Any]] = [] for row in rows: item = dict(row) if "source_kind" not in item and item.get("source"): item["source_kind"] = item.get("source") if "id" not in item and item.get("observation_id"): item["id"] = item.get("observation_id") normalized.append(item) return _dedupe_dicts_local(normalized, key="id")[:24] def _build_customer_safe_claim_ledger( delivery_payload: dict[str, Any], *, mission: PMFlowAIOperatingMission, ) -> PMFlowAICustomerSafeClaimLedger: deliverable = delivery_payload.get("deliverable_packet") if isinstance(delivery_payload.get("deliverable_packet"), dict) else {} workspace = delivery_payload.get("agent_workspace") if isinstance(delivery_payload.get("agent_workspace"), dict) else {} terminal_state = delivery_payload.get("terminal_answer_state") if isinstance(delivery_payload.get("terminal_answer_state"), dict) else {} answer_runtime_packet = delivery_payload.get("answer_runtime_packet") if isinstance(delivery_payload.get("answer_runtime_packet"), dict) else {} observations = _delivery_observations(delivery_payload, workspace=workspace) terminal_status = str(terminal_state.get("terminal_status") or "").strip() blocked_statuses = {"waiting_for_evidence", "blocked", "refused_by_boundary", "waiting_for_user"} if terminal_status in blocked_statuses: return PMFlowAICustomerSafeClaimLedger( mission_id=mission.mission_id, status="blocked", blocked_reason=_claim_ledger_blocked_reason(terminal_status), ) entries: list[PMFlowAICustomerSafeClaim] = [] for index, row in enumerate(_dicts(terminal_state.get("claims"))): _append_customer_safe_claim( entries, row=row, source_kind="terminal_claim", source_label="本轮回答判断", source_ref=f"terminal_claim:{index}", ) for index, row in enumerate(_dicts(workspace.get("claim_ledger")) + _dicts(workspace.get("claims"))): _append_customer_safe_claim( entries, row=row, source_kind="workspace_claim", source_label="工作区判断", source_ref=f"workspace_claim:{index}", ) section_index = 0 for section in _dicts(deliverable.get("sections")): if str(section.get("section_kind") or "") not in {"claims", "findings", "summary"}: continue for item in _dicts(section.get("items")): _append_customer_safe_claim( entries, row=item, source_kind="deliverable_claim", source_label="结构化交付判断", source_ref=f"deliverable_claim:{section_index}", ) section_index += 1 for key in ("claims", "claim_ledger", "customer_safe_claims"): for index, row in enumerate(_dicts(answer_runtime_packet.get(key))): _append_customer_safe_claim( entries, row=row, source_kind="answer_claim", source_label="答案质量门判断", source_ref=f"answer_claim:{key}:{index}", ) for observation in observations: observation_ref = str(observation.get("observation_id") or observation.get("id") or "").strip() for index, fact in enumerate(_dicts(observation.get("facts"))): fact_type = str(fact.get("type") or "").strip() if fact_type not in {"normalized_claim", "claim", "finding"}: continue _append_customer_safe_claim( entries, row=fact, source_kind="observation_claim", source_label="已归一化观察判断", source_ref=observation_ref or f"observation_claim:{index}", ) deduped = _dedupe_customer_safe_claims(entries) source_counts: dict[str, int] = {} source_refs: list[dict[str, Any]] = [] evidence_refs: list[str] = [] for claim in deduped: source_counts[claim.source_kind] = source_counts.get(claim.source_kind, 0) + 1 if claim.source_ref: source_refs.append({"source_kind": claim.source_kind, "source_label": claim.source_label, "source_ref": claim.source_ref}) evidence_refs.extend(list(claim.evidence_refs or [])) status = "ready" if len(deduped) >= 3 else ("partial" if deduped else "empty") return PMFlowAICustomerSafeClaimLedger( mission_id=mission.mission_id, status=status, claims=deduped[:16], claim_count=len(deduped[:16]), source_counts=source_counts, source_refs=_dedupe_dicts_local(source_refs, key="source_ref")[:24], evidence_refs=_dedupe(evidence_refs)[:30], ) def _should_keep_delivery_open_reasoning_answer( *, delivery_payload: dict[str, Any], project_world_state: PMFlowAIProjectWorldStatePacket, ) -> bool: if project_world_state.project_refs: return False observations = _delivery_observations( delivery_payload, workspace=delivery_payload.get("agent_workspace") if isinstance(delivery_payload.get("agent_workspace"), dict) else {}, ) if _external_source_rows(observations): return False terminal_state = delivery_payload.get("terminal_answer_state") if isinstance(delivery_payload.get("terminal_answer_state"), dict) else {} claims = _dicts(terminal_state.get("claims")) workspace = delivery_payload.get("agent_workspace") if isinstance(delivery_payload.get("agent_workspace"), dict) else {} claims.extend(_dicts(workspace.get("claim_ledger"))) return any(str(item.get("claim_type") or "").strip() == "controlled_hypothesis_basis" for item in claims) def _delivery_can_finish_without_deep_operating_chain( *, delivery_payload: dict[str, Any], mission: PMFlowAIOperatingMission, project_world_state: PMFlowAIProjectWorldStatePacket, customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, ) -> bool: if project_world_state.project_refs or list(getattr(mission.project_scope, "object_refs", []) or []): return False answer = _delivery_customer_answer(delivery_payload) if not answer: return False if _answer_conflicts_with_evidence( answer, project_world_state=project_world_state, delivery_payload=delivery_payload, customer_safe_claim_ledger=customer_safe_claim_ledger, ): return False if isinstance(delivery_payload.get("interaction_request"), dict) and delivery_payload.get("interaction_request"): return False if str(delivery_payload.get("grounding_status") or "") in {"not_grounded", "insufficient_evidence"}: return False workspace = delivery_payload.get("agent_workspace") if isinstance(delivery_payload.get("agent_workspace"), dict) else {} goal_frame = workspace.get("goal_frame") if isinstance(workspace.get("goal_frame"), dict) else {} answer_policy = str(goal_frame.get("answer_policy") or "").strip() missing_policy = str(goal_frame.get("missing_evidence_policy") or "").strip() latency_policy = goal_frame.get("latency_policy") if isinstance(goal_frame.get("latency_policy"), dict) else {} latency_mode = str(latency_policy.get("mode") or "").strip() if answer_policy not in {"", "directional_answer", "partial_answer"}: return False if missing_policy not in {"", "answer_with_boundaries"}: return False if latency_mode not in {"", "fast", "standard"}: return False observations = _delivery_observations(delivery_payload, workspace=workspace) has_external_sources = bool(_external_source_rows(observations)) has_open_reasoning_basis = _payload_has_controlled_hypothesis_basis(delivery_payload) terminal = delivery_payload.get("terminal_answer_state") if isinstance(delivery_payload.get("terminal_answer_state"), dict) else {} terminal_status = str(terminal.get("terminal_status") or "").strip() agent_run_packet = delivery_payload.get("agent_run_packet") if isinstance(delivery_payload.get("agent_run_packet"), dict) else {} stop_reason = str(agent_run_packet.get("stop_reason") or "").strip() verifier_passed = any( str(item.get("result") or "").strip() == "passed" for item in _dicts(agent_run_packet.get("verifications")) ) has_claims = bool(customer_safe_claim_ledger.claims) runtime_ready = stop_reason in {"planner_ready_to_answer", "ready_to_answer", "completed"} or verifier_passed terminal_ready = terminal_status in {"", "answerable", "partially_answerable", "answered", "completed"} if not terminal_status and not runtime_ready and not has_open_reasoning_basis: return False return terminal_ready and (has_external_sources or has_open_reasoning_basis or has_claims) def _payload_has_controlled_hypothesis_basis(delivery_payload: dict[str, Any]) -> bool: terminal_state = delivery_payload.get("terminal_answer_state") if isinstance(delivery_payload.get("terminal_answer_state"), dict) else {} claims = _dicts(terminal_state.get("claims")) workspace = delivery_payload.get("agent_workspace") if isinstance(delivery_payload.get("agent_workspace"), dict) else {} claims.extend(_dicts(workspace.get("claim_ledger"))) return any(str(item.get("claim_type") or "").strip() == "controlled_hypothesis_basis" for item in claims) def _append_customer_safe_claim( entries: list[PMFlowAICustomerSafeClaim], *, row: dict[str, Any], source_kind: str, source_label: str, source_ref: str, ) -> None: if row.get("allowed_to_state") is False: return text = str(row.get("claim_text") or row.get("summary") or row.get("text") or "").strip() if not _is_customer_safe_claim_text(text): return entries.append( PMFlowAICustomerSafeClaim( claim_text=text, source_kind=source_kind, source_label=source_label, source_ref=source_ref, confidence=_normalize_claim_confidence(row.get("confidence")), evidence_refs=_claim_evidence_refs(row), object_refs=_claim_object_refs(row), allowed_to_state=True, ) ) def _dedupe_customer_safe_claims(entries: list[PMFlowAICustomerSafeClaim]) -> list[PMFlowAICustomerSafeClaim]: merged: dict[str, PMFlowAICustomerSafeClaim] = {} for claim in entries: key = normalize_evidence_text(claim.claim_text) if not key: continue existing = merged.get(key) if existing is None: merged[key] = claim continue merged[key] = existing.model_copy( update={ "evidence_refs": _dedupe([*list(existing.evidence_refs or []), *list(claim.evidence_refs or [])])[:16], "object_refs": _dedupe([*list(existing.object_refs or []), *list(claim.object_refs or [])])[:16], "source_ref": existing.source_ref or claim.source_ref, "source_kind": existing.source_kind or claim.source_kind, "source_label": existing.source_label or claim.source_label, "confidence": _highest_confidence(existing.confidence, claim.confidence), } ) return list(merged.values()) def _claim_evidence_refs(row: dict[str, Any]) -> list[str]: values = row.get("evidence_refs") or row.get("source_refs") or row.get("sources") or [] if not isinstance(values, list): values = [values] out: list[str] = [] for value in values[:16]: if isinstance(value, dict): text = str(value.get("evidence_id") or value.get("source_ref") or value.get("url") or value.get("title") or "").strip() else: text = str(value or "").strip() if text and text not in out: out.append(text[:180]) return out def _claim_object_refs(row: dict[str, Any]) -> list[str]: values = row.get("object_refs") or row.get("target_refs") or row.get("project_refs") or [] if not isinstance(values, list): values = [values] return _dedupe([str(item or "").strip() for item in values if str(item or "").strip()])[:16] def _normalize_claim_confidence(value: Any) -> str: text = str(value or "").strip().lower() if text in {"high", "medium", "low"}: return text return "unknown" def _highest_confidence(left: str, right: str) -> str: order = {"unknown": 0, "low": 1, "medium": 2, "high": 3} return left if order.get(left, 0) >= order.get(right, 0) else right def _claim_ledger_blocked_reason(terminal_status: str) -> str: mapping = { "waiting_for_evidence": "本轮仍在等待证据，不能把临时材料写成客户安全判断。", "blocked": "本轮答案状态被阻断，不能生成客户安全判断。", "refused_by_boundary": "本轮触发披露或权限边界，不能生成业务判断。", "waiting_for_user": "本轮需要用户补充，不能生成客户安全判断。", } return mapping.get(terminal_status, "本轮未形成可客户展示的判断。") def _claim_texts( workspace: dict[str, Any], deliverable: dict[str, Any], *, terminal_state: dict[str, Any] | None = None, answer_runtime_packet: dict[str, Any] | None = None, observations: list[dict[str, Any]] | None = None, ) -> list[str]: terminal_status = str((terminal_state or {}).get("terminal_status") or "").strip() if terminal_status in {"waiting_for_evidence", "blocked", "refused_by_boundary", "waiting_for_user"}: return [] claims: list[str] = [] for row in _dicts((terminal_state or {}).get("claims")): if row.get("allowed_to_state") is False: continue claims.append(str(row.get("claim_text") or row.get("summary") or row.get("text") or "").strip()) for row in _dicts(workspace.get("claim_ledger")) + _dicts(workspace.get("claims")): if row.get("allowed_to_state") is False: continue claims.append(str(row.get("claim_text") or row.get("summary") or row.get("text") or "").strip()) for section in _dicts(deliverable.get("sections")): if str(section.get("section_kind") or "") not in {"claims", "findings", "summary"}: continue for item in _dicts(section.get("items")): if item.get("allowed_to_state") is False: continue claims.append(str(item.get("summary") or item.get("claim_text") or item.get("text") or "").strip()) for key in ("claims", "claim_ledger", "customer_safe_claims"): for row in _dicts((answer_runtime_packet or {}).get(key)): if row.get("allowed_to_state") is False: continue claims.append(str(row.get("claim_text") or row.get("summary") or row.get("text") or "").strip()) for observation in observations or []: for fact in _dicts(observation.get("facts")): if fact.get("allowed_to_state") is False: continue fact_type = str(fact.get("type") or "").strip() if fact_type not in {"normalized_claim", "claim", "finding"}: continue claims.append(str(fact.get("claim_text") or fact.get("summary") or fact.get("text") or "").strip()) return _dedupe([item for item in claims if _is_customer_safe_claim_text(item)])[:12] def _is_customer_safe_claim_text(text: Any) -> bool: value = normalize_evidence_text(text) if not value: return False if is_usable_research_claim_text(value): return True if len(value) < 18: return False if looks_like_source_noise(value) or _looks_like_internal_token(value): return False if re.search(r"#{2,}|---|^\s*[-*]\s+\*\*|\|\s*[-:]{2,}\s*\|", value): return False return True def _gap_texts(workspace: dict[str, Any], deliverable: dict[str, Any], existing_gaps: list[str]) -> list[str]: gaps = [str(item or "").strip() for item in existing_gaps if str(item or "").strip()] for row in _dicts(workspace.get("gap_ledger")) + _dicts(workspace.get("remaining_gaps")): gaps.append(str(row.get("summary") or row.get("gap_text") or row.get("text") or "").strip()) for section in _dicts(deliverable.get("sections")): if str(section.get("section_kind") or "") not in {"gaps", "boundaries", "limitations"}: continue for item in _dicts(section.get("items")): gaps.append(str(item.get("summary") or item.get("text") or "").strip()) return _dedupe([item for item in gaps if item])[:16] def _artifact_slots_from_delivery(delivery_payload: dict[str, Any]) -> list[dict[str, Any]]: canvas = delivery_payload.get("answer_canvas") if isinstance(delivery_payload.get("answer_canvas"), dict) else {} artifacts = _dicts(canvas.get("artifacts")) slots: list[dict[str, Any]] = [] for artifact in artifacts[:8]: artifact_kind = str(artifact.get("artifact_kind") or artifact.get("kind") or "").strip() artifact_title = str(artifact.get("title") or "").strip() for slot in _dicts(artifact.get("slots"))[:12]: slot_key = str(slot.get("slot_key") or slot.get("key") or "").strip() summary = _plain_customer_text(slot.get("summary") or slot.get("display_goal") or slot.get("title"), limit=500) if not slot_key and not summary: continue slots.append( { "artifact_kind": artifact_kind, "artifact_title": artifact_title, "slot_key": slot_key, "display_goal": summary, "must_cover": [ _plain_customer_text(item, limit=260) for item in list(slot.get("must_cover") or []) if _plain_customer_text(item, limit=260) ][:8], } ) deliverable = delivery_payload.get("deliverable_packet") if isinstance(delivery_payload.get("deliverable_packet"), dict) else {} for section in _dicts(deliverable.get("sections"))[:8]: kind = str(section.get("section_kind") or "").strip() if not kind: continue summary = _plain_customer_text(section.get("title") or section.get("summary") or kind, limit=420) slots.append({"artifact_kind": "deliverable_packet", "slot_key": kind, "display_goal": summary, "must_cover": []}) return _dedupe_dicts_local(slots, key="slot_key")[:24] def _evidence_lanes_from_world_state( *, project_world_state: PMFlowAIProjectWorldStatePacket, delivery_payload: dict[str, Any], customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, ) -> list[dict[str, Any]]: lanes: list[dict[str, Any]] = [] for facet in list(project_world_state.facets or [])[:12]: facts = [ {"fact_key": fact.get("fact_key"), "count": len(fact.get("value") or []) if isinstance(fact.get("value"), list) else 1} for fact in list(facet.facts or [])[:8] if isinstance(fact, dict) ] lanes.append( { "facet_key": facet.facet_key, "display_label": facet.display_label, "status": facet.status, "summary": facet.summary, "facts": facts, } ) ledger_claims = [ _plain_customer_text(item.claim_text, limit=300) for item in list(customer_safe_claim_ledger.claims or []) if item.claim_text ] if ledger_claims: lanes.append( { "lane": "customer_safe_claim_ledger", "status": customer_safe_claim_ledger.status, "summary": f"CustomerSafeClaimLedger 已形成 {len(ledger_claims)} 条客户安全判断。", "claims": ledger_claims[:8], } ) return lanes[:24] def _current_object_binding_from_context( *, conversation_context: dict[str, Any] | None, tenant_project_context: dict[str, Any] | None, ) -> dict[str, Any]: for container in (tenant_project_context, conversation_context): if not isinstance(container, dict): continue surface_context = container.get("surface_context") if isinstance(container.get("surface_context"), dict) else {} current = surface_context.get("current_focus_object") if isinstance(surface_context.get("current_focus_object"), dict) else {} if current: return dict(current) visible = list(surface_context.get("visible_objects") or []) if visible and isinstance(visible[0], dict): return dict(visible[0]) return {} def _current_object_binding_from_delivery( *, delivery_payload: dict[str, Any], conversation_context: dict[str, Any] | None, tenant_project_context: dict[str, Any] | None, ) -> dict[str, Any]: packet = ( delivery_payload.get("customer_interaction_context_packet") if isinstance(delivery_payload.get("customer_interaction_context_packet"), dict) else {} ) current = packet.get("current_focus_object") if isinstance(packet.get("current_focus_object"), dict) else {} if current: return dict(current) return _current_object_binding_from_context(conversation_context=conversation_context, tenant_project_context=tenant_project_context) def _mission_critic_events( *, initial_critic: PMFlowAIOperatingMissionCriticResult, post_evidence_critic: PMFlowAIOperatingMissionCriticResult, ) -> list[PMFlowAIOperatingRunEvent]: rows: list[PMFlowAIOperatingRunEvent] = [] for key, critic in (("mission_critic_initial", initial_critic), ("mission_critic_post_evidence", post_evidence_critic)): if not critic.needs_replan and not critic.missing_capability_roles: continue rows.append( PMFlowAIOperatingRunEvent( event_type=key, summary=critic.reason or f"OperatingMissionCritic 补齐 {len(critic.missing_capability_roles)} 个 specialist coverage 缺口。", ) ) return rows def _answer_conflicts_with_evidence( answer: str, *, project_world_state: PMFlowAIProjectWorldStatePacket, delivery_payload: dict[str, Any], customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, ) -> bool: text = str(answer or "") if not text.strip(): return False source_count, claim_count = _world_source_claim_counts( project_world_state=project_world_state, delivery_payload=delivery_payload, customer_safe_claim_ledger=customer_safe_claim_ledger, ) source_zero = bool(re.search(r"0\s*(?:个)?(?:候选)?来源", text)) claim_zero = bool(re.search(r"0\s*条(?:可用)?(?:判断|结论|证据|信息)", text)) if source_zero and source_count > 0: return True if claim_zero and claim_count > 0: return True return False def _world_source_claim_counts( *, project_world_state: PMFlowAIProjectWorldStatePacket, delivery_payload: dict[str, Any], customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, ) -> tuple[int, int]: source_count = 0 claim_count = len(list(customer_safe_claim_ledger.claims or [])) for facet in list(project_world_state.facets or []): for fact in list(facet.facts or []): if not isinstance(fact, dict): continue key = str(fact.get("fact_key") or "") value = fact.get("value") if key == "source_candidates" and isinstance(value, list): source_count += len(value) if key == "usable_claims" and isinstance(value, list): claim_count += len(value) return source_count, claim_count def _compose_answer_from_terminal_and_artifacts( *, delivery_payload: dict[str, Any], specialist_answer: str, project_world_state: PMFlowAIProjectWorldStatePacket, customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, ) -> str: claims = [ _plain_customer_text(item.claim_text, limit=420) for item in list(customer_safe_claim_ledger.claims or []) if item.claim_text ] claims = _dedupe([item for item in claims if item])[:4] canvas = delivery_payload.get("answer_canvas") if isinstance(delivery_payload.get("answer_canvas"), dict) else {} artifacts = _dicts(canvas.get("artifacts")) blueprint = next( ( item for item in artifacts if str(item.get("artifact_kind") or "").strip() == "product_blueprint" or "蓝图" in str(item.get("title") or "") or "blueprint" in str(item.get("artifact_kind") or "").lower() ), {}, ) if not claims and not blueprint: return "" parts: list[str] = [] if claims: parts.append("可以先给一个有边界的判断：" + "；".join(claims[:2]) + "。") if blueprint: slots = [] for slot in _dicts(blueprint.get("slots"))[:5]: title = _plain_customer_text(slot.get("title") or slot.get("slot_key"), limit=120) summary = _plain_customer_text(slot.get("summary") or slot.get("display_goal"), limit=260) if title and summary: slots.append(f"{title}：{summary}") elif summary: slots.append(summary) if slots: parts.append("我会把它收敛成产品蓝图来推进：" + "；".join(slots[:4]) + "。") actions = _dicts(canvas.get("action_candidates")) if actions: labels = [_customer_action_label(item.get("title") or item.get("action_label")) for item in actions[:2]] action_text = "；".join(item for item in labels if item) if action_text: parts.append(f"下一步可以准备待确认动作：{action_text}。") gaps = _customer_research_gaps(list(project_world_state.remaining_gaps or [])) if gaps: parts.append("证据边界是：" + "；".join(gaps[:2]) + "。") if specialist_answer: parts.append(_plain_customer_text(specialist_answer, limit=800)) return "".join(parts).strip() def _dedupe(rows: list[str]) -> list[str]: out: list[str] = [] for item in rows: text = " ".join(str(item or "").split())[:700] if text and text not in out: out.append(text) return out def _dedupe_dicts_local(rows: list[dict[str, Any]], *, key: str) -> list[dict[str, Any]]: seen: set[str] = set() out: list[dict[str, Any]] = [] for row in rows: value = str(row.get(key) or row.get("title") or row.get("summary") or "").strip() if not value or value in seen: continue seen.add(value) out.append(row) return out def _dicts(value: Any) -> list[dict[str, Any]]: if not isinstance(value, list): return [] return [item for item in value if isinstance(item, dict)] def _json_for_prompt(value: Any, *, limit: int) -> str: return json.dumps(value, ensure_ascii=False, default=str)[: max(500, int(limit))] def _build_command_payload( *, mission: PMFlowAIOperatingMission, assignments: list[PMFlowAIAgentAssignment], project_world_state: PMFlowAIProjectWorldStatePacket, work_products: list[PMFlowAIAgentWorkProduct], council_review: Any, autonomy_gate: PMFlowAIAutonomyDecision, operating_cycle: PMFlowAIOperatingCycleState, operating_plan: PMFlowAIOperatingPlan, product_object_projections: list[PMFlowAIProductObjectProjection], delivery_payload: dict[str, Any], customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, ) -> PMFlowAIOperatingCommandPayload: context_packet = ( delivery_payload.get("customer_interaction_context_packet") if isinstance(delivery_payload.get("customer_interaction_context_packet"), dict) else {} ) return PMFlowAIOperatingCommandPayload( mission={ "mission_id": mission.mission_id, "goal": mission.mission_goal, "desired_outcome": mission.desired_outcome, "autonomy_level": mission.autonomy_level, }, project_focus=context_packet.get("current_focus_object") if isinstance(context_packet.get("current_focus_object"), dict) else {}, project_world_state=_project_world_state_command_summary(project_world_state), agent_council=[ { "agent_id": assignment.agent_id, "status": _product_for_agent(work_products=work_products, agent_id=assignment.agent_id).status if _product_for_agent(work_products=work_products, agent_id=assignment.agent_id) is not None else "blocked", "work_product": assignment.expected_work_product, "summary": _summary_for_agent(work_products=work_products, agent_id=assignment.agent_id), } for assignment in assignments ], agent_council_review=council_review.model_dump(mode="json", exclude_none=True), conflict_ledger=[item.model_dump(mode="json", exclude_none=True) for item in council_review.conflict_ledger], priority_order=list(council_review.priority_order or []), risk_vetoes=list(council_review.risk_vetoes or []), autonomy_gate=autonomy_gate.model_dump(mode="json", exclude_none=True), 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), strategic_read=operating_plan.strategic_read, operating_plan=operating_plan.model_dump(mode="json", exclude_none=True), prepared_actions=list(operating_plan.prepared_action_packets or []), confirmation_requests=list(operating_plan.confirmation_requests or []), execution_preparations=list(operating_plan.execution_preparations or []), product_object_projections=[item.model_dump(mode="json", exclude_none=True) for item in product_object_projections], blocked_actions=list(operating_plan.blocked_actions or []), evidence_boundary=list(operating_plan.evidence_boundary or []), evidence_ledger=[item.model_dump(mode="json", exclude_none=True) for item in project_world_state.evidence_ledger[:20]], customer_safe_claim_ledger=customer_safe_claim_ledger.model_dump(mode="json", exclude_none=True), learning_updates=list(operating_plan.learning_updates or []), next_cycle={ "status": "ready_for_next_operating_step", "summary": "下一步应让专业 agent 基于当前项目世界状态和 learning ledger 原生产出更深入的经营计划。", "triggers": list(operating_plan.next_cycle_triggers or []), }, ) def _merge_operating_process( process_packet: Any, *, mission: PMFlowAIOperatingMission, assignments: list[PMFlowAIAgentAssignment], work_products: list[PMFlowAIAgentWorkProduct], project_world_state: PMFlowAIProjectWorldStatePacket, council_review: dict[str, Any], autonomy_gate: dict[str, Any], operating_cycle: dict[str, Any], initial_critic: PMFlowAIOperatingMissionCriticResult, post_evidence_critic: PMFlowAIOperatingMissionCriticResult, customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, total_duration_ms: int, ) -> dict[str, Any]: packet = dict(process_packet or {}) if isinstance(process_packet, dict) else {} existing_steps = list(packet.get("steps") or []) if isinstance(packet.get("steps"), list) else [] packet["packet_version"] = "customer_answer_process.v1" packet["visibility"] = "customer" packet["answer_type"] = packet.get("answer_type") or "agent_runtime" packet["route_display"] = packet.get("route_display") or "agent_runtime" packet["steps"] = [ *_operating_process_steps( mission=mission, assignments=assignments, work_products=work_products, project_world_state=project_world_state, council_review=council_review, autonomy_gate=autonomy_gate, operating_cycle=operating_cycle, initial_critic=initial_critic, post_evidence_critic=post_evidence_critic, customer_safe_claim_ledger=customer_safe_claim_ledger, ), *existing_steps, ] packet["total_duration_ms"] = max(0, int(total_duration_ms or packet.get("total_duration_ms") or 0)) return packet def _operating_process_steps( *, mission: PMFlowAIOperatingMission, assignments: list[PMFlowAIAgentAssignment], work_products: list[PMFlowAIAgentWorkProduct], project_world_state: PMFlowAIProjectWorldStatePacket, council_review: dict[str, Any], autonomy_gate: dict[str, Any], operating_cycle: dict[str, Any], initial_critic: PMFlowAIOperatingMissionCriticResult, post_evidence_critic: PMFlowAIOperatingMissionCriticResult, customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, ) -> list[dict[str, Any]]: critic_missing = _dedupe([*initial_critic.missing_capability_roles, *post_evidence_critic.missing_capability_roles]) return [ { "key": "operating_mission", "label": "确定 Operating Mission", "step_type": "operating_mission", "status": "completed", "display_summary": mission.desired_business_outcome or mission.desired_outcome or mission.mission_goal, }, { "key": "operating_mission_critic", "label": "校验 Mission 覆盖", "step_type": "operating_mission_critic", "status": "completed", "display_summary": ( f"Mission Critic 根据工作产品槽位和证据通道补齐 {len(critic_missing)} 个专业能力。" if critic_missing else "Mission Critic 已确认当前专业能力覆盖足以支撑本轮交付。" ), }, { "key": "specialist_registry", "label": "选择专业 Agent", "step_type": "specialist_registry", "status": "completed" if assignments else "blocked", "display_summary": f"本轮根据 mission 和能力合同选择 {len(assignments)} 个专业 agent，产出 {len(work_products)} 个工作产品。", }, { "key": "project_world_state", "label": "读取项目/空间状态", "step_type": "project_world_state", "status": "completed" if project_world_state.status in {"ready", "partial"} else "blocked", "display_summary": _project_world_state_event_summary(project_world_state), }, { "key": "customer_safe_claim_ledger", "label": "汇总客户安全判断", "step_type": "customer_safe_claim_ledger", "status": "completed" if customer_safe_claim_ledger.status in {"ready", "partial"} else customer_safe_claim_ledger.status, "display_summary": _claim_ledger_process_summary(customer_safe_claim_ledger), }, { "key": "agent_council", "label": "会审专业产物", "step_type": "agent_council", "status": "completed" if str(council_review.get("status") or "") != "blocked" else "blocked", "display_summary": str(council_review.get("resolution_summary") or "")[:600], }, { "key": "autonomy_gate", "label": "应用确认门禁", "step_type": "autonomy_gate", "status": "completed" if str(autonomy_gate.get("status") or "") != "blocked" else "blocked", "display_summary": _autonomy_gate_process_summary(autonomy_gate), }, { "key": "learning_loop", "label": "写入下一轮学习触发", "step_type": "learning_loop", "status": "completed", "display_summary": _learning_process_summary(operating_cycle), }, ] def _merge_operating_canvas( canvas: dict[str, Any], *, mission: PMFlowAIOperatingMission, assignments: list[PMFlowAIAgentAssignment], work_products: list[PMFlowAIAgentWorkProduct], operating_plan: PMFlowAIOperatingPlan, product_object_projections: list[PMFlowAIProductObjectProjection], ) -> dict[str, Any]: out = dict(canvas or {}) artifacts = [item for item in list(out.get("artifacts") or []) if isinstance(item, dict)] artifacts.extend( _operating_canvas_artifacts( mission=mission, assignments=assignments, work_products=work_products, operating_plan=operating_plan, product_object_projections=product_object_projections, ) ) out["artifacts"] = _dedupe_dicts_local(artifacts, key="title")[:6] return out def _operating_canvas_artifacts( *, mission: PMFlowAIOperatingMission, assignments: list[PMFlowAIAgentAssignment], work_products: list[PMFlowAIAgentWorkProduct], operating_plan: PMFlowAIOperatingPlan, product_object_projections: list[PMFlowAIProductObjectProjection], ) -> list[dict[str, Any]]: slots = [] for product in work_products[:8]: if product.status == "blocked" and not product.summary: continue slots.append( { "slot_key": product.agent_id, "title": _customer_work_product_title(product), "summary": _plain_customer_text(product.summary, limit=500), "coverage_status": product.status, } ) for projection in product_object_projections[:4]: slots.append( { "slot_key": projection.object_type, "title": projection.title, "summary": projection.summary, "coverage_status": "prepared", } ) if not slots: return [] return [ { "artifact_id": f"operating_{mission.mission_id}", "artifact_kind": "operating_mission", "title": "本轮 Operating Mission 与专业产物", "summary": operating_plan.strategic_read or mission.desired_business_outcome or mission.desired_outcome, "slots": slots[:8], "confidence": "medium", "agent_count": len(assignments), } ] def _operating_status(operating_plan: PMFlowAIOperatingPlan) -> str: if operating_plan.requires_confirmation: return "waiting_for_confirmation" return "completed" def _agent_run_id(payload: dict[str, Any]) -> str: packet = payload.get("agent_run_packet") if isinstance(payload.get("agent_run_packet"), dict) else {} return str(packet.get("run_id") or "").strip() def _delivery_customer_answer(delivery_payload: dict[str, Any]) -> str: """Prefer the already gated Agent Answer Runtime output as the customer answer. The operating layer enriches the run with mission/work-product/command packets. It must not replace a usable natural answer with its own process summary; that summary is kept separately for audit and command surfaces. """ for key in ("answer", "reply_text"): text = _plain_customer_text(delivery_payload.get(key), limit=5000) if text and not _looks_like_internal_answer_text(text): return text canvas = delivery_payload.get("answer_canvas") if isinstance(delivery_payload.get("answer_canvas"), dict) else {} text = _plain_customer_text(canvas.get("direct_answer"), limit=5000) if text and not _looks_like_internal_answer_text(text): return text return "" def _delivery_is_terminal_boundary(delivery_payload: dict[str, Any]) -> bool: packet = delivery_payload.get("agent_run_packet") if isinstance(delivery_payload.get("agent_run_packet"), dict) else {} terminal = delivery_payload.get("terminal_answer_state") if isinstance(delivery_payload.get("terminal_answer_state"), dict) else {} stop_reason = str(packet.get("stop_reason") or "").strip() terminal_status = str(terminal.get("terminal_status") or "").strip() if stop_reason == "protected_internal_refusal": return True return terminal_status in {"refused_by_boundary"} and bool(_delivery_customer_answer(delivery_payload)) def _boundary_delivery_payload( *, delivery_payload: dict[str, Any], mission: PMFlowAIOperatingMission, started: float, ) -> dict[str, Any]: payload = dict(delivery_payload or {}) answer = _delivery_customer_answer(payload) payload["answer"] = answer payload["reply_text"] = answer payload["processing_stage"] = str(payload.get("processing_stage") or "agent_runtime") payload["answer_mode"] = str(payload.get("answer_mode") or "agent_runtime") payload["policy_version"] = "pmflow-multi-agent-operating-runtime-v1" payload["operating_mission"] = mission.model_dump(mode="json", exclude_none=True) payload["agent_assignments"] = [] payload["agent_work_products"] = [] payload["prepared_action_packets"] = [] payload["confirmation_requests"] = [] payload["execution_preparations"] = [] payload["product_object_projections"] = [] packet = payload.get("answer_process_packet") if isinstance(payload.get("answer_process_packet"), dict) else {} packet["total_duration_ms"] = max(0, int((time.perf_counter() - started) * 1000)) payload["answer_process_packet"] = packet return payload def _shallow_operating_delivery_payload( *, delivery_payload: dict[str, Any], mission: PMFlowAIOperatingMission, project_world_state: PMFlowAIProjectWorldStatePacket, customer_safe_claim_ledger: PMFlowAICustomerSafeClaimLedger, started: float, ) -> dict[str, Any]: payload = dict(delivery_payload or {}) answer = _delivery_customer_answer(payload) payload["answer"] = answer payload["reply_text"] = answer if isinstance(payload.get("answer_canvas"), dict): payload["answer_canvas"] = { **dict(payload["answer_canvas"]), "direct_answer": answer, "headline": "PMFlow AI", } else: payload["answer_canvas"] = { "direct_answer": answer, "headline": "PMFlow AI", "next_actions": [], "action_candidates": [], } payload["processing_stage"] = "multi_agent_operating_runtime" payload["answer_mode"] = "agent_runtime_operating_shallow" payload["policy_version"] = "pmflow-multi-agent-operating-runtime-v1" payload["operating_summary"] = answer payload["operating_mission"] = mission.model_dump(mode="json", exclude_none=True) payload["project_world_state_packet"] = project_world_state.model_dump(mode="json", exclude_none=True) payload["customer_safe_claim_ledger"] = customer_safe_claim_ledger.model_dump(mode="json", exclude_none=True) payload["agent_assignments"] = [] payload["agent_work_products"] = [] payload["prepared_action_packets"] = [] payload["confirmation_requests"] = [] payload["execution_preparations"] = [] payload["product_object_projections"] = [] packet = payload.get("answer_process_packet") if isinstance(payload.get("answer_process_packet"), dict) else {} steps = list(packet.get("steps") or []) if isinstance(packet.get("steps"), list) else [] packet["packet_version"] = "customer_answer_process.v1" packet["visibility"] = "customer" packet["steps"] = [ { "key": "operating_runtime_scope_gate", "label": "判断是否需要深度编排", "step_type": "operating_scope_gate", "status": "completed", "display_summary": "当前材料已足够先给出带边界的方向判断，本轮不再追加更重的专业产物链。", "duration_ms": 0, "blocking": False, }, *steps, ] packet["total_duration_ms"] = max(0, int((time.perf_counter() - started) * 1000)) packet["wall_clock_duration_ms"] = packet["total_duration_ms"] payload["answer_process_packet"] = packet return payload def _looks_like_internal_answer_text(text: str) -> bool: value = str(text or "").strip() if not value: return True lowered = value.lower() internal_markers = ( "agent os", "turnstate", "goalframe", "researchnote", "claim_ledger", "gap_ledger", "answer_runtime_packet", "runtime", "trace", "prompt", ) if any(marker in lowered for marker in internal_markers): return True return _looks_like_internal_token(value) 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 _autonomy_gate_event_summary(autonomy_gate: PMFlowAIAutonomyDecision) -> str: return ( f"Autonomy Gate 已生成 {len(autonomy_gate.prepared_action_packets)} 个准备动作包、" f"{len(autonomy_gate.confirmation_requests)} 个确认请求和 " f"{len(autonomy_gate.execution_preparations)} 个执行准备包；写入和 handoff 均未执行。" ) def _autonomy_gate_process_summary(autonomy_gate: dict[str, Any]) -> str: prepared_count = len(list(autonomy_gate.get("prepared_action_packets") or [])) confirmation_count = len(list(autonomy_gate.get("confirmation_requests") or [])) execution_count = len(list(autonomy_gate.get("execution_preparations") or [])) return ( f"已准备 {prepared_count} 个动作包、{confirmation_count} 个确认请求、" f"{execution_count} 个执行交接包；写入/执行/发布/外部动作全部需要确认后再进入专用链路。" ) def _learning_event_summary(operating_cycle: PMFlowAIOperatingCycleState) -> str: return ( f"已消费 {len(operating_cycle.consumed_learning_ledger.learning_items)} 条历史学习，" f"记录 {len(operating_cycle.outcome_observations)} 条结果观察，" f"准备 {len(operating_cycle.next_cycle_triggers)} 个下一轮触发合同。" ) def _learning_process_summary(operating_cycle: dict[str, Any]) -> str: consumed = operating_cycle.get("consumed_learning_ledger") if isinstance(operating_cycle.get("consumed_learning_ledger"), dict) else {} learning = operating_cycle.get("learning_ledger") if isinstance(operating_cycle.get("learning_ledger"), dict) else {} observations = list(operating_cycle.get("outcome_observations") or []) triggers = list(operating_cycle.get("next_cycle_triggers") or []) return ( f"已读取 {len(list(consumed.get('learning_items') or []))} 条历史学习，" f"本轮记录 {len(observations)} 条结果观察，" f"学习账本累计 {len(list(learning.get('learning_items') or []))} 条，" f"下一轮可由 {len(triggers)} 类事件触发。" ) def _claim_ledger_process_summary(ledger: PMFlowAICustomerSafeClaimLedger) -> str: if ledger.status == "blocked": return ledger.blocked_reason or "本轮未形成可进入回答和项目世界状态的客户安全判断。" source_count = len(dict(ledger.source_counts or {})) return f"已从 {source_count} 类客户安全来源汇总 {len(list(ledger.claims or []))} 条判断，供项目世界状态、答案仲裁和后续产品蓝图共用。" def _project_world_state_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, "project_refs": list(project_world_state.project_refs or []), "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": list(project_world_state.remaining_gaps or [])[:12], } def _compose_specialist_answer( *, mission: PMFlowAIOperatingMission, project_world_state: PMFlowAIProjectWorldStatePacket, work_products: list[PMFlowAIAgentWorkProduct], council_review: Any, autonomy_gate: PMFlowAIAutonomyDecision, operating_cycle: PMFlowAIOperatingCycleState, product_object_projections: list[PMFlowAIProductObjectProjection], operating_plan: PMFlowAIOperatingPlan, ) -> str: strategist = _product_for_agent(work_products=work_products, agent_id="product_strategy_agent") execution = _product_for_agent(work_products=work_products, agent_id="pm_breakdown_agent") risk = _product_for_agent(work_products=work_products, agent_id="qa_risk_agent") project_ref = _display_scope_ref(project_world_state.primary_project_ref or "") if project_world_state.status not in {"ready", "partial"}: gaps = "、".join(project_world_state.remaining_gaps[:3]) or "缺少可安全绑定的项目对象" return f"我还不能给出完整推进判断：{project_ref} 的项目世界状态没有读完整，当前缺口是 {gaps}。这轮只能先保留本轮目标和对象候选，不能用空间概览或单段回答冒充项目计划。" lead = operating_plan.highest_leverage_move or mission.desired_outcome strategy_text = _product_summary(strategist) execution_text = _product_summary(execution) research_product = _product_for_type(work_products=work_products, product_type="ResearchNote") if research_product is not None and (project_world_state.primary_project_ref == "space:external_research" or not project_world_state.project_refs): return _compose_research_operating_answer( mission=mission, project_world_state=project_world_state, research_product=research_product, operating_plan=operating_plan, autonomy_gate=autonomy_gate, operating_cycle=operating_cycle, ) risk_blocks = list(council_review.risk_vetoes or []) or (list(risk.blocked_actions or []) if risk is not None else []) risk_text = "" if risk_blocks: first_block = risk_blocks[0] risk_text = f" 不能直接做的是：{first_block.get('title')}，原因是 {first_block.get('reason')}。" prepared = list(operating_plan.prepared_action_packets or []) prepared_text = "" if prepared: prepared_text = " 我已经准备好候选动作包：" + "；".join(str(item.get("title") or "") for item in prepared[:3] if item.get("title")) + "，这些都需要你确认后才会写入或执行。" projection_text = "" if product_object_projections: projection = product_object_projections[0] projection_text = f" 本轮已经形成一个待确认的{projection.object_type}：{projection.title}。" confirmation_text = "" if autonomy_gate.confirmation_requests: confirmation_text = f" 本轮有 {len(autonomy_gate.confirmation_requests)} 个确认请求；确认前只展示候选，不会写入项目资料、改动任务或交给执行环境。" learning_text = "" if operating_cycle.consumed_learning_ledger.learning_items: learning_text = f" 我也读取了上一轮 {len(operating_cycle.consumed_learning_ledger.learning_items)} 条历史约束，避免重复使用已失败的推进方式。" council_text = "" if council_review.conflict_ledger: council_text = f" Agent Council 已记录 {len(council_review.conflict_ledger)} 个冲突/边界，并收敛成单一优先级。" fallback_context = project_world_state.facets[0].summary if project_world_state.facets else "" context = strategy_text or execution_text or fallback_context return f"{project_ref} 的最高杠杆动作是：{lead}。{context}{council_text}{projection_text}{prepared_text}{confirmation_text}{learning_text}{risk_text}".strip() def _compose_research_operating_answer( *, mission: PMFlowAIOperatingMission, project_world_state: PMFlowAIProjectWorldStatePacket, research_product: PMFlowAIAgentWorkProduct | None, operating_plan: PMFlowAIOperatingPlan, autonomy_gate: PMFlowAIAutonomyDecision, operating_cycle: PMFlowAIOperatingCycleState, ) -> str: output = research_product.structured_output if research_product is not None and isinstance(research_product.structured_output, dict) else {} evidence_state = output.get("evidence_state") if isinstance(output.get("evidence_state"), dict) else {} source_count = int(evidence_state.get("source_count") or 0) claims = _gated_research_claims(output) claim_count = len(claims) ready = bool(evidence_state.get("ready_for_opportunity_brief")) and claim_count > 0 gaps = _customer_research_gaps(list(output.get("research_gaps") or project_world_state.remaining_gaps or [])) source_plan = [item for item in list(output.get("source_plan") or []) if isinstance(item, dict)] lead = operating_plan.highest_leverage_move or "继续补齐研究证据" if ready and claims: answer = "按目前能核验的材料，可以先形成一个初步判断：" + "；".join(claims[:3]) if gaps: answer += f"。仍需保留的边界是：{'；'.join(gaps[:2])}。" elif claims: answer = ( "现在可以先给一个有边界的初判：" + "；".join(claims[:3]) + f"。但还不能升级成机会简报或执行动作：当前是 {source_count} 个候选来源、{claim_count} 条可用判断，" "还需要跨来源补证。" ) if gaps: answer += f" 还缺：{'；'.join(gaps[:3])}。" else: answer = ( "现在还不能直接下市场结论。" f"已整理 {source_count} 个候选来源和 {claim_count} 条可用判断，" f"接下来最值得补的是：{lead}。" ) if gaps: answer += f"主要缺口：{'；'.join(gaps[:3])}。" if source_plan: lanes = [] for item in source_plan[:3]: lane = str(item.get("lane") or item.get("goal") or "").strip() query = str(item.get("query") or "").strip() label = lane or query if label: lanes.append(label) if lanes: answer += " 下一步只推进前置研究补证：" + "；".join(lanes) + "。" if autonomy_gate.confirmation_requests: executable = [ item for item in list(autonomy_gate.confirmation_requests or []) if isinstance(item, dict) and str(item.get("action_type") or "").strip() ] if executable: labels = [_customer_action_label(item.get("action_label") or item.get("title")) for item in executable[:3]] action_text = "；".join(item for item in labels if item) if action_text: answer += " 可继续的动作是：" + action_text + "。" answer += "确认前只展示候选，不会写入项目资料、改动任务或交给执行环境。" if operating_cycle.consumed_learning_ledger.learning_items: answer += f" 这轮沿用了上一轮 {len(operating_cycle.consumed_learning_ledger.learning_items)} 条历史约束，但只用于避免重复走失败路径，不把历史失败当成结论。" return answer.strip() def _gated_research_claims(output: dict[str, Any]) -> list[str]: rows = list(output.get("market_claims") or output.get("claims") or []) out: list[str] = [] for item in rows[:12]: text = _plain_customer_text(item, limit=420) if not text or not is_usable_research_claim_text(text): continue if text not in out: out.append(text) return out def _customer_research_gaps(rows: list[Any]) -> list[str]: out: list[str] = [] for item in rows[:12]: text = _plain_customer_text(item, limit=260) if not text: continue if _looks_like_internal_token(text): text = "还缺少可直接支撑结论的公开来源材料" if text not in out: out.append(text) return out def _customer_action_label(value: object) -> str: text = _plain_customer_text(value, limit=80) if not text: return "" if _looks_like_internal_token(text): return "" latin_count = sum(1 for ch in text if "a" <= ch.lower() <= "z") cjk_count = sum(1 for ch in text if "\u4e00" <= ch <= "\u9fff") if latin_count > 16 and cjk_count == 0: return "" return text def _plain_customer_text(value: object, *, limit: int) -> str: text = str(value or "").replace("\r\n", "\n").replace("\r", "\n").strip() if not text: return "" lines = [re.sub(r"[ \t]+", " ", line).strip() for line in text.split("\n")] text = "\n".join(lines) text = re.sub(r"\n{3,}", "\n\n", text).strip() text = _repair_numbered_markdown_table_continuations(text) text = _repair_fragmented_markdown_tables(text) text = _remove_orphan_markdown_pipe_lines(text) text = _repair_inline_bold_boundaries(text) text = _repair_inline_markdown_tables(text) text = _repair_raw_markdown_headings(text) text = _repair_inline_markdown_lists(text) text = _repair_inline_section_breaks(text) text = text[:limit].strip() return text.replace("Project Memory", "项目资料").replace("handoff", "执行交接") def _repair_inline_markdown_tables(text: str) -> str: value = str(text or "").strip() if not value or "|" not in value: return value paragraphs = re.split(r"\n\s*\n+", value) repaired: list[str] = [] for paragraph in paragraphs: item = paragraph.strip() if not item: continue if "|" in item and "---" not in item and item.count("|") >= 4: recovered = _recover_malformed_pipe_rows(item) if recovered: repaired.append(recovered) continue if not _looks_like_inline_markdown_table(item): repaired.append(item) continue first_pipe = item.find("|") if first_pipe > 0: lead = item[:first_pipe].strip() if lead: repaired.append(lead) item = item[first_pipe:].strip() reconstructed = _reconstruct_inline_markdown_table(item) if reconstructed: repaired.append(reconstructed) continue table = re.sub(r"\s*\|\|\s*", " |\n| ", item) table = re.sub(r"\s+(\|\s*:?-{3,}:?\s*\|)", r"\n\1", table) table = re.sub(r"(\|\s*:?-{3,}:?\s*\|[ \t]*(?:\|[ \t]*:?-{3,}:?[ \t]*)*)\s+\|", r"\1\n|", table) table = re.sub(r"\|\s+(:?-{3,}:?)", r"|\1", table) table = re.sub(r"(:?-{3,}:?)\s+\|", r"\1|", table) repaired.append("\n".join(line.strip() for line in table.splitlines() if line.strip())) return "\n\n".join(repaired).strip() def _remove_orphan_markdown_pipe_lines(text: str) -> str: value = str(text or "").strip() if not value or "|" not in value: return value lines = [line for line in value.split("\n") if not re.fullmatch(r"[ \t]*\|+[ \t]*", line)] return re.sub(r"\n{3,}", "\n\n", "\n".join(lines)).strip() def _repair_inline_bold_boundaries(text: str) -> str: value = str(text or "").strip() if "**" not in value: return value def repl(match: re.Match[str]) -> str: body = match.group(1).strip() before = value[match.start() - 1] if match.start() > 0 else "" after = value[match.end()] if match.end() < len(value) else "" prefix = " " if before and not re.fullmatch(r"\s|[(（\[【{《“]", before) else "" suffix = " " if after and not re.fullmatch(r"\s|[)，)\]】}。；：，、！？,.!?:》”]", after) else "" return f"{prefix}**{body}**{suffix}" value = re.sub(r"\*\*([^*\n]+?)\*\*", repl, value) return re.sub(r"[ \t]{2,}", " ", value).strip() def _repair_fragmented_markdown_tables(text: str) -> str: value = str(text or "").strip() if not value or "|" not in value or "---" not in value: return value lines = value.split("\n") out: list[str] = [] index = 0 while index < len(lines): current = lines[index].strip() next_line = lines[index + 1].strip() if index + 1 < len(lines) else "" if not _is_markdown_pipe_line(current) or not _is_markdown_table_separator_line(next_line): out.append(lines[index]) index += 1 continue first_pipe = current.find("|") lead = current[:first_pipe].strip() if first_pipe > 0 else "" header_line = current[first_pipe:].strip() if first_pipe > 0 else current segment = [header_line, next_line] cursor = index + 2 crossed_blank_line = False while cursor < len(lines): candidate = lines[cursor].strip() if not candidate: next_table_index = _next_non_empty_line_index(lines, cursor + 1) if next_table_index >= 0 and _is_markdown_pipe_line(lines[next_table_index].strip()): crossed_blank_line = True cursor += 1 continue break if not _is_markdown_pipe_line(candidate): break segment.append(candidate) cursor += 1 reconstructed = _reconstruct_fragmented_markdown_table(segment) if reconstructed and (crossed_blank_line or reconstructed != "\n".join(segment)): if lead: out.append(lead) out.append(reconstructed) index = cursor continue out.append(lines[index]) index += 1 return "\n".join(out).strip() def _repair_numbered_markdown_table_continuations(text: str) -> str: value = str(text or "").strip() if not value or "|" not in value or "---" not in value: return value lines = value.split("\n") out: list[str] = [] index = 0 while index < len(lines): current = lines[index].strip() next_line = lines[index + 1].strip() if index + 1 < len(lines) else "" if not _is_markdown_pipe_line(current) or not _is_markdown_table_separator_line(next_line): out.append(lines[index]) index += 1 continue header = _markdown_pipe_cells(current, preserve_internal_empty=False) separators = _markdown_pipe_cells(next_line, preserve_internal_empty=False) column_count = len(header) if column_count < 2 or len(separators) != column_count: out.append(lines[index]) index += 1 continue segment: list[str] = [] cursor = index + 2 saw_bare_continuation = False while cursor < len(lines): candidate = lines[cursor].strip() if not candidate: next_table_index = _next_non_empty_line_index(lines, cursor + 1) next_candidate = lines[next_table_index].strip() if next_table_index >= 0 else "" if next_candidate and ( _is_markdown_pipe_line(next_candidate) or next_candidate.startswith("|") or _is_numbered_table_continuation_line(next_candidate) or (saw_bare_continuation and not _looks_like_new_markdown_block(next_candidate)) ): segment.append("") cursor += 1 continue break if _is_standalone_markdown_heading(candidate): break if _is_markdown_pipe_line(candidate) or (saw_bare_continuation and candidate.startswith("|")): segment.append(candidate) cursor += 1 continue if _is_numbered_table_continuation_line(candidate) or (segment and "|" in candidate): saw_bare_continuation = True segment.append(candidate) cursor += 1 continue if saw_bare_continuation and segment and not _looks_like_new_markdown_block(candidate): segment.append(candidate) cursor += 1 continue break if not saw_bare_continuation or not segment: out.append(lines[index]) index += 1 continue repaired = _reconstruct_numbered_continuation_table(header, separators, segment) if not repaired: out.append(lines[index]) index += 1 continue out.append(repaired) index = cursor return "\n".join(out).strip() def _is_standalone_markdown_heading(line: str) -> bool: value = str(line or "").strip() return bool(re.fullmatch(r"#{1,6}\s+\S.*", value) or re.fullmatch(r"\*\*[^*\n]{2,80}\*\*", value)) def _looks_like_new_markdown_block(line: str) -> bool: value = str(line or "").strip() return _is_standalone_markdown_heading(value) or bool(re.fullmatch(r"[-*+]\s+\S.*", value)) def _is_numbered_table_continuation_line(line: str) -> bool: value = str(line or "").strip() return bool(re.match(r"^\|?\s*\d+[.)、]\s+\S", value) or re.match(r"^\|{2,}\s*\d+[.)、]\s+\S", value)) def _reconstruct_numbered_continuation_table( header: list[str], separators: list[str], lines: list[str], ) -> str: column_count = len(header) normalized = "\n".join(lines) normalized = normalized.replace("||", "\n| ") normalized = re.sub(r"\n{2,}", "\n", normalized).strip() chunks = [ chunk.strip() for chunk in re.split(r"(?=(?:^|\n)\s*\|?\s*\d+[.)、]\s+\S)", normalized) if chunk.strip() ] rows: list[list[str]] = [] malformed = False for chunk in chunks: row = _row_from_numbered_table_chunk(chunk, column_count) if not row: malformed = True continue if not chunk.strip().startswith("|") or len(_markdown_pipe_cells(chunk, preserve_internal_empty=False)) != column_count: malformed = True rows.append(row) if not malformed or len(rows) < 2: return "" return "\n".join( [ "| " + " | ".join(header) + " |", "| " + " | ".join(separators) + " |", *["| " + " | ".join(row) + " |" for row in rows], ] ) def _row_from_numbered_table_chunk(chunk: str, column_count: int) -> list[str] | None: cleaned = re.sub(r"^\|+", "", str(chunk or "")) cleaned = re.sub(r"\|+$", "", cleaned).strip() cells = [_normalize_markdown_table_cell(cell) for cell in cleaned.split("|")] cells = [cell for cell in cells if cell] if len(cells) >= column_count: return [*cells[: column_count - 1], "；".join(cells[column_count - 1 :])][:column_count] if len(cells) == column_count - 1 and column_count >= 3: split = _split_numbered_table_lead_cell(cells[0]) if split: cells = [split[0], split[1], *cells[1:]] if len(cells) == column_count: return cells return None def _split_numbered_table_lead_cell(value: str) -> tuple[str, str] | None: text = _normalize_markdown_table_cell(value) match = re.match(r"^((?:\d+[.)、]\s*)?\*\*[^*]{2,60}\*\*)\s+(.+)$", text) if match: return match.group(1).strip(), match.group(2).strip() match = re.match(r"^(\d+[.)、]\s+[^，。；：:、|]{2,28}?)[ \t]+(.+)$", text) if match: return match.group(1).strip(), match.group(2).strip() return None def _next_non_empty_line_index(lines: list[str], start: int) -> int: for index in range(start, len(lines)): if lines[index].strip(): return index return -1 def _is_markdown_pipe_line(line: str) -> bool: value = str(line or "").strip() return value.startswith("|") and value.count("|") >= 2 def _is_markdown_table_separator_line(line: str) -> bool: cells = _markdown_pipe_cells(line, preserve_internal_empty=False) return len(cells) >= 2 and all(re.fullmatch(r":?-{3,}:?", cell) for cell in cells) def _markdown_pipe_cells(line: str, *, preserve_internal_empty: bool) -> list[str]: cells = [_normalize_markdown_table_cell(cell) for cell in str(line or "").split("|")] if cells and cells[0] == "": cells = cells[1:] if cells and cells[-1] == "": cells = cells[:-1] if preserve_internal_empty: return cells return [cell for cell in cells if cell] def _normalize_markdown_table_cell(value: str) -> str: cell = re.sub(r"", "；", str(value or ""), flags=re.IGNORECASE) cell = re.sub(r"\s+", " ", cell) cell = re.sub(r"；{2,}", "；", cell) return cell.strip() def _reconstruct_fragmented_markdown_table(lines: list[str]) -> str: if len(lines) < 3: return "" if any(re.search(r"#{2,6}\s*\S", line) for line in lines[2:]): return "" header = _markdown_pipe_cells(lines[0], preserve_internal_empty=False) separators = _markdown_pipe_cells(lines[1], preserve_internal_empty=False) column_count = len(header) if column_count < 2 or len(separators) != column_count: return "" if not all(re.fullmatch(r":?-{3,}:?", cell) for cell in separators): return "" rows: list[list[str]] = [] buffer: list[str] = [] for line in lines[2:]: cells = _markdown_pipe_cells(line, preserve_internal_empty=True) non_empty_cells = [cell for cell in cells if cell] if not non_empty_cells: continue can_use_whole_row = ( not buffer and len(cells) == column_count and bool(cells[0]) and len(non_empty_cells) >= max(2, column_count - 1) ) if can_use_whole_row: rows.append(cells) continue buffer.extend(non_empty_cells) while len(buffer) >= column_count: rows.append(buffer[:column_count]) buffer = buffer[column_count:] if not rows: return "" table = "\n".join( [ "| " + " | ".join(header) + " |", "| " + " | ".join(separators) + " |", *["| " + " | ".join(row) + " |" for row in rows], ] ) trailing = " ".join(buffer).strip() return "\n\n".join(part for part in [table, trailing] if part).strip() def _recover_malformed_pipe_rows(item: str) -> str: first_pipe = item.find("|") lead = item[:first_pipe].strip() if first_pipe > 0 else "" body = item[first_pipe:] if first_pipe >= 0 else item cells, trailing = _split_heading_from_pipe_cells(body) rows = _rows_from_pipe_cells(cells) if not rows: return "" table = "\n".join( [ "| 优先级 | 项目 | 说明 |", "| :--- | :--- | :--- |", *[f"| {row['priority']} | {row['title']} | {row['summary']} |" for row in rows], ] ) return "\n\n".join(part for part in [lead, table, trailing] if part).strip() def _split_heading_from_pipe_cells(body: str) -> tuple[list[str], str]: cells = [cell.strip() for cell in str(body or "").split("|")] cells = [cell for cell in cells if cell] heading_index = next((index for index, cell in enumerate(cells) if re.search(r"#{2,6}\s*\S", cell)), -1) if heading_index < 0: return cells, "" heading = re.sub(r"^#{2,6}\s*", "", cells[heading_index]).strip() return cells[:heading_index], _format_markdown_heading_body(heading) if heading else "" def _rows_from_pipe_cells(cells: list[str]) -> list[dict[str, str]]: rows: list[dict[str, str]] = [] priority_re = re.compile(r"^(?:P\d+|高|中|低|High|Medium|Low)$", re.IGNORECASE) index = 0 while index + 2 < len(cells): if priority_re.fullmatch(cells[index]): rows.append({"priority": cells[index], "title": cells[index + 1], "summary": cells[index + 2]}) index += 3 continue if priority_re.fullmatch(cells[index + 2]): rows.append({"priority": cells[index + 2], "title": cells[index], "summary": cells[index + 1]}) index += 3 continue break return [row for row in rows if row["title"] and row["summary"]] def _reconstruct_inline_markdown_table(item: str) -> str: cells = [cell.strip() for cell in str(item or "").split("|")] cells = [cell for cell in cells if cell] sep_index = next((index for index, cell in enumerate(cells) if re.fullmatch(r":?-{3,}:?", cell)), -1) if sep_index < 2: return "" column_count = sep_index separators = cells[sep_index : sep_index + column_count] if len(separators) < column_count or not all(re.fullmatch(r":?-{3,}:?", cell) for cell in separators): return "" header = cells[:column_count] rest = cells[sep_index + column_count :] rows = [rest[index : index + column_count] for index in range(0, len(rest), column_count)] rows = [row for row in rows if len(row) == column_count and any(cell for cell in row)] if not rows: return "" return "\n".join( [ "| " + " | ".join(header) + " |", "| " + " | ".join(separators) + " |", *["| " + " | ".join(row) + " |" for row in rows], ] ) def _looks_like_inline_markdown_table(text: str) -> bool: item = str(text or "").strip() if "\n" in item or item.count("|") < 8: return False return bool(re.search(r"\|\s*:?-{3,}:?\s*\|", item)) def _repair_inline_markdown_lists(text: str) -> str: value = str(text or "").strip() if not value: return "" value = re.sub(r"([：:])\s+((?:[-*+–—]|\d+[.)、])\s+)", r"\1\n\n\2", value) value = re.sub(r"([。！？；）)])\s*((?:[-*+–—]|\d+[.)、])\s+)", r"\1\n\n\2", value) value = re.sub(r"(? str: value = str(text or "").strip() if not value: return "" value = re.sub(r"^(?![ \t]*#{2,6}\s)([^。！？\n|]{3,36}?)[ \t]+[—–-][ \t]+(?=\S{2,24}[：:])", r"**\1**\n\n", value) value = re.sub(r"([。！？])[ \t]+([^。！？\n|]{3,36}?)[ \t]+[—–-][ \t]+(?=\S{2,24}[：:])", r"\1\n\n**\2**\n\n", value) return re.sub(r"\n{3,}", "\n\n", value).strip() def _repair_raw_markdown_headings(text: str) -> str: value = str(text or "").replace(r"\#", "#") value = re.sub(r"([^\n])\s+(#{2,6}\s+\S)", r"\1\n\n\2", value) return re.sub( r"^[ \t]*#{2,6}\s+(.+)$", lambda match: _format_markdown_heading_body(match.group(1)), value, flags=re.MULTILINE, ).strip() def _format_markdown_heading_body(value: str) -> str: body = re.sub(r"^#{2,6}\s+", "", str(value or "").strip()) if not body: return "" separator = re.search(r"\s+[—–-]\s+", body) if not separator: return f"**{body}**" title = body[: separator.start()].strip() rest = body[separator.end() :].strip() if not title: return rest if not rest: return f"**{title}**" return f"**{title}**\n\n{rest}" def _looks_like_internal_token(text: str) -> bool: lowered = text.lower().strip() if re.search(r"\b[a-z]+(?:_[a-z0-9]+){1,}\b", lowered): return True if re.search(r"\b(step|run|packet|trace|prompt|function)_[a-z0-9_]+\b", lowered): return True return False def _project_world_state_event_summary(project_world_state: PMFlowAIProjectWorldStatePacket) -> str: coverage = 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 "没有强项目对象，本轮不会用空间项目数量或 active 状态冒充项目世界状态。" return f"已读取 {len(project_world_state.project_refs)} 个项目对象的世界状态；覆盖 {covered} 项、部分覆盖 {partial} 项、缺失 {missing} 项。" def _display_scope_ref(value: str) -> str: text = str(value or "").strip() if text == "space:external_research": return "这个机会研究" if text.startswith("project:"): return "当前项目" if not text: return "当前对象" return text def _summary_for_agent(*, work_products: list[PMFlowAIAgentWorkProduct], agent_id: str) -> str: product = _product_for_agent(work_products=work_products, agent_id=agent_id) return product.summary if product else "" 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 _product_for_type(*, work_products: list[PMFlowAIAgentWorkProduct], product_type: str) -> PMFlowAIAgentWorkProduct | None: return next((item for item in work_products if item.product_type == product_type), None) def _product_summary(product: PMFlowAIAgentWorkProduct | None) -> str: return str(product.summary if product is not None else "").strip()[:500] def _customer_work_product_title(product: PMFlowAIAgentWorkProduct) -> str: output = product.structured_output if isinstance(product.structured_output, dict) else {} for key in ("title", "artifact_title", "summary_title", "research_question", "idea_title"): title = _plain_customer_text(output.get(key), limit=120) if title: return title return _plain_customer_text(product.product_type, limit=120) or "专业工作产品" __all__ = ["PMFlowAIOperatingRunService"]