"""Specialist agent capability registry for PMFlow AI operating runs.""" from __future__ import annotations from typing import Iterable from ...schemas.pmflow_ai_operating_runtime import ( PMFlowAIAgentAssignment, PMFlowAIAgentCapabilityContract, PMFlowAIOperatingMission, ) class PMFlowAIOperatingAgentRegistry: """Registry of specialist operating agents. The registry is a capability contract surface. It is intentionally not a keyword router: selection is driven by mission contract and project scope, not by matching user text. """ def __init__(self, contracts: Iterable[PMFlowAIAgentCapabilityContract] | None = None) -> None: self._contracts = {contract.agent_id: contract for contract in (contracts or _default_contracts())} def list_contracts(self) -> list[PMFlowAIAgentCapabilityContract]: return list(self._contracts.values()) def assignments_for_mission(self, mission: PMFlowAIOperatingMission) -> list[PMFlowAIAgentAssignment]: requested = [agent_id for agent_id in mission.required_agents if agent_id in self._contracts] agent_ids = requested or [contract.agent_id for contract in self.list_contracts() if contract.agent_id in _CORE_STEP1_AGENT_IDS] input_refs = ["operating_mission", "project_world_state_packet", "evidence_ledger", "learning_ledger"] assignments: list[PMFlowAIAgentAssignment] = [] for agent_id in agent_ids: contract = self._contracts.get(agent_id) if contract is None: continue assignments.append( PMFlowAIAgentAssignment( agent_id=contract.agent_id, role_goal=_role_goal(contract=contract, mission=mission), input_refs=input_refs, expected_work_product=_expected_work_product(contract.agent_id), autonomy_limit=contract.autonomy_limit, ) ) return assignments def contract_for(self, agent_id: str) -> PMFlowAIAgentCapabilityContract | None: return self._contracts.get(str(agent_id or "").strip()) _CORE_STEP1_AGENT_IDS = { "idea_incubation_agent", "market_research_agent", "competitor_agent", "product_strategy_agent", "mvp_scope_agent", "tech_architect_agent", "pm_breakdown_agent", "qa_risk_agent", "growth_agent", "monetization_agent", "learning_agent", } def _default_contracts() -> list[PMFlowAIAgentCapabilityContract]: return [ PMFlowAIAgentCapabilityContract( agent_id="idea_incubation_agent", display_name="Idea Incubation Agent", responsibility="Compress raw ideas into target users, problem hypotheses, value hypotheses and first validation cuts.", input_contract=["OperatingMission", "ProjectWorldStatePacket", "LearningLedger"], output_contract=["IdeaSeed", "problem_hypotheses", "value_hypotheses", "first_validation_cut"], allowed_tools=["read_learning_ledger", "read_research_workspace"], write_permissions=[], autonomy_limit="propose_plan", critic_checks=["hypotheses must be marked as assumptions unless evidence-backed", "must identify first falsifiable test"], customer_safe_projection={"idea_seed": True, "hypotheses": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="market_research_agent", display_name="Market Research Agent", responsibility="Collect public market, trend, demand and timing evidence and turn it into a bounded research note.", input_contract=["OperatingMission", "external evidence ledger", "LearningLedger"], output_contract=["ResearchNote", "market_claims", "source_boundaries", "research_gaps"], allowed_tools=["external_search", "inspect_source_detail", "read_learning_ledger"], write_permissions=[], autonomy_limit="propose_plan", critic_checks=["market claims require evidence refs or explicit boundary", "must propose recovery when sources are weak"], customer_safe_projection={"research_note": True, "source_boundaries": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="competitor_agent", display_name="Competitor Agent", responsibility="Identify comparable products, substitutes and positioning pressure without inventing competitive facts.", input_contract=["OperatingMission", "external evidence ledger", "ProjectWorldStatePacket"], output_contract=["CompetitorMap", "substitutes", "differentiation_gaps"], allowed_tools=["external_search", "inspect_source_detail"], write_permissions=[], autonomy_limit="propose_plan", critic_checks=["competitor facts require source refs", "must separate substitutes from direct competitors"], customer_safe_projection={"competitor_map": True, "positioning_gaps": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="product_strategy_agent", display_name="Product Strategy Agent", responsibility="Convert evidence and project state into positioning, MVP direction, tradeoffs and opportunity decision.", input_contract=["OperatingMission", "ProjectWorldStatePacket", "ResearchNote", "LearningLedger"], output_contract=["OpportunityBrief", "ProductHypothesis", "positioning_options", "highest_leverage_move"], allowed_tools=["read_project_world_state", "read_learning_ledger", "read_research_workspace"], write_permissions=["prepare_object_only"], autonomy_limit="prepare_changes", critic_checks=["strategy must tie to mission outcome", "must separate assumptions from verified evidence"], customer_safe_projection={"opportunity_brief": True, "product_hypothesis": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="mvp_scope_agent", display_name="MVP Scope Agent", responsibility="Turn strategy into MVP options, requirement boundaries and out-of-scope decisions.", input_contract=["OperatingMission", "ProjectWorldStatePacket", "OpportunityBrief"], output_contract=["MVPOption", "RequirementDraft candidates", "out_of_scope_now"], allowed_tools=["read_project_world_state", "prepare_requirement_draft"], write_permissions=["prepare_change_only"], autonomy_limit="prepare_changes", critic_checks=["requirement drafts require confirmation", "scope must include acceptance criteria"], customer_safe_projection={"mvp_options": True, "requirement_candidates": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="tech_architect_agent", display_name="Tech Architect Agent", responsibility="Review technical path, architecture risk, dependencies, cost and handoff readiness.", input_contract=["OperatingMission", "ProjectWorldStatePacket", "ExecutionContext"], output_contract=["TechnicalArchitectureAssessment", "technical_bottlenecks", "handoff_readiness"], allowed_tools=["read_project_world_state", "read_execution_summary"], write_permissions=[], autonomy_limit="propose_plan", critic_checks=["must separate readiness from execution", "must not claim code execution"], customer_safe_projection={"technical_risks": True, "handoff_readiness": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="pm_breakdown_agent", display_name="PM Breakdown Agent", responsibility="Compile next-cycle tasks, priority changes, dependencies, blockers and execution package candidates.", input_contract=["OperatingMission", "ProjectWorldStatePacket", "specialist work products"], output_contract=["TaskDraft", "ExecutionPlan", "next_cycle_plan", "prepared_task_changes"], allowed_tools=["read_project_world_state", "prepare_task_change", "prepare_execution_package"], write_permissions=["prepare_change_only", "prepare_handoff_only"], autonomy_limit="prepare_changes", critic_checks=["task changes require confirmation", "blocked dependencies must be explicit"], customer_safe_projection={"next_cycle_plan": True, "prepared_actions": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="qa_risk_agent", display_name="QA Risk Agent", responsibility="Review privacy, compliance, safety, authorization and execution risks; veto unsafe prepared actions.", input_contract=["OperatingMission", "ProjectWorldStatePacket", "OperatingPlan draft"], output_contract=["QAPlan", "RiskReview", "blocked_actions", "confirmation_requirements"], allowed_tools=["read_project_world_state", "read_policy_boundary"], write_permissions=[], autonomy_limit="propose_plan", critic_checks=["high-risk action must be blocked or require confirmation", "risk veto must be visible"], customer_safe_projection={"blocked_actions": True, "confirmation_requirements": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="growth_agent", display_name="Growth Agent", responsibility="Translate opportunity or project state into channel, launch, feedback and growth experiment options.", input_contract=["OperatingMission", "ProjectWorldStatePacket", "ResearchNote", "ReleaseSignals"], output_contract=["GrowthExperiment", "ReleaseSignal candidates", "feedback_plan"], allowed_tools=["read_project_world_state", "read_release_growth_signals", "external_search"], write_permissions=["prepare_experiment_only"], autonomy_limit="prepare_changes", critic_checks=["growth experiments need success criteria", "must not claim launch execution"], customer_safe_projection={"growth_experiments": True, "feedback_plan": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="monetization_agent", display_name="Monetization Agent", responsibility="Assess pricing, cost, revenue path and commercial viability signals.", input_contract=["OperatingMission", "ProjectWorldStatePacket", "market evidence", "LearningLedger"], output_contract=["RevenueSignal", "pricing_hypotheses", "unit_economics_gaps"], allowed_tools=["external_search", "read_project_world_state", "read_learning_ledger"], write_permissions=[], autonomy_limit="propose_plan", critic_checks=["commercial claims require evidence or assumption marker", "must identify cost/revenue gaps"], customer_safe_projection={"revenue_signals": True, "pricing_hypotheses": True, "internal_reasoning": False}, ), PMFlowAIAgentCapabilityContract( agent_id="learning_agent", display_name="Learning Agent", responsibility="Extract accepted/rejected decisions, useful source patterns, user preference signals and next-cycle triggers.", input_contract=["OperatingMission", "OutcomeObservations", "LearningLedger"], output_contract=["LearningUpdate", "next_cycle_policy", "memory_candidates"], allowed_tools=["read_learning_ledger", "observe_outcome"], write_permissions=["prepare_learning_only"], autonomy_limit="prepare_changes", critic_checks=["learning memory writeback requires explicit policy", "must not store raw transcript as preference"], customer_safe_projection={"learning_updates": True, "next_cycle": True, "internal_reasoning": False}, ), ] def _role_goal(*, contract: PMFlowAIAgentCapabilityContract, mission: PMFlowAIOperatingMission) -> str: goal = mission.mission_goal or "推进当前 PMFlow operating mission" return f"{contract.display_name} 围绕本轮 mission 产出结构化 { _expected_work_product(contract.agent_id) }：{goal}" def _expected_work_product(agent_id: str) -> str: mapping = { "idea_incubation_agent": "IdeaSeed", "market_research_agent": "ResearchNote", "competitor_agent": "CompetitorMap", "product_strategy_agent": "OpportunityBrief", "mvp_scope_agent": "MVPOption", "tech_architect_agent": "TechnicalArchitectureAssessment", "pm_breakdown_agent": "TaskDraft", "qa_risk_agent": "QAPlan", "growth_agent": "GrowthExperiment", "monetization_agent": "RevenueSignal", "learning_agent": "LearningUpdate", } return mapping.get(agent_id, "AgentWorkProduct") __all__ = ["PMFlowAIOperatingAgentRegistry"]