"""Truth-bound project copilot answer service.""" from __future__ import annotations from typing import Any from sqlalchemy.ext.asyncio import AsyncSession from ..schemas.pmflow_os import ( ProjectCopilotActionLadderLevel, ProjectCopilotAllowedAction, ProjectCopilotAnswerCanvas, ProjectCopilotAskRequest, ProjectCopilotAskResponse, ProjectCopilotContextPacket, ProjectCopilotObjectAnchorPacket, ProjectTruthStateProjection, ) from .pmflow_os_projection_service import PMFlowOSProjectionService class ProjectCopilotService: """Build customer-safe project copilot answers from the project truth projection. This service intentionally stays read-only. It may explain, draft candidate direction, review risk, prepare handoff guidance, or route the user back to a PMFlow surface, but it never commits a ChangeSet, writes Project Memory, executes commands, or publishes. """ def __init__(self, db: AsyncSession, *, tenant_id: int): self.db = db self.tenant_id = int(tenant_id) self.projection_service = PMFlowOSProjectionService(db, tenant_id=tenant_id) async def ask( self, *, project_id: int, current_user_id: int, body: ProjectCopilotAskRequest, ) -> ProjectCopilotAskResponse: truth_state = await self.projection_service.get_truth_state( project_id=int(project_id), current_user_id=int(current_user_id), ) try: execution_summary = await self.projection_service.get_execution_summary( project_id=int(project_id), current_user_id=int(current_user_id), ) except Exception: execution_summary = None anchor = self._resolve_anchor(truth_state, body) packet = self._build_context_packet( truth_state, body=body, anchor=anchor, execution_summary=execution_summary.model_dump(mode="json") if execution_summary else {}, ) canvas = self._render_answer(body=body, packet=packet, truth_state=truth_state) return ProjectCopilotAskResponse( answer_canvas=canvas, project_copilot_context_packet=packet, customer_interaction_context_packet={ "visible_objects": [ { "object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}", "display_label": packet.object_anchor.title, "object_type": packet.object_anchor.type, "surface": packet.surface, } ], "current_focus_object": { "object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}", "display_label": packet.object_anchor.title, "surface": packet.surface, }, }, answer_process_packet={ "steps": [ {"label": "读取项目真相", "status": "completed", "blocking": True}, {"label": "绑定当前对象锚点", "status": "completed", "blocking": True}, {"label": "校验允许动作和禁止动作", "status": "completed", "blocking": True}, {"label": "生成低风险主动建议", "status": "completed", "blocking": False}, {"label": "生成客户安全回答", "status": "completed", "blocking": False}, ], "model_calls": 0, "source": "project_copilot_deterministic_renderer", }, allowed_actions=packet.allowed_actions, blocked_actions=packet.forbidden_actions, proactive_suggestions=packet.proactive_suggestions, warnings=[], ) def _resolve_anchor( self, truth_state: ProjectTruthStateProjection, body: ProjectCopilotAskRequest, ) -> ProjectCopilotObjectAnchorPacket: requested = body.object_anchor if requested: anchor = self._anchor_from_request(truth_state, requested.type, requested.id) if anchor: return anchor return self._default_anchor(truth_state, body.surface) def _anchor_from_request( self, truth_state: ProjectTruthStateProjection, anchor_type: str, anchor_id: str, ) -> ProjectCopilotObjectAnchorPacket | None: if not anchor_id: return None if anchor_type == "change_set": item = _find_by_id(truth_state.changeSets, anchor_id) if item: return ProjectCopilotObjectAnchorPacket( type="change_set", id=str(item.get("id", anchor_id)), title=str(item.get("headline") or item.get("summary") or f"变更集 {anchor_id}"), summary=str(item.get("summary") or "当前变更集。"), status_label=_change_set_status_label(item.get("status")), mode="committed" if str(item.get("status")) in {"committed", "partially_committed"} else "current", source_label="项目待确认链", risk_level=self._risk_for_change_set(truth_state, str(item.get("id", anchor_id))), evidence_ids=self._evidence_for_change_set(truth_state, str(item.get("id", anchor_id))), ) if anchor_type == "change_item": item = _find_by_id(truth_state.changeItems, anchor_id) if item: return ProjectCopilotObjectAnchorPacket( type="change_item", id=str(item.get("id", anchor_id)), title=str(item.get("title") or f"变更项 {anchor_id}"), summary=str(item.get("summary") or item.get("diff") or "当前变更项。"), status_label=_change_item_status_label(item.get("status")), mode="committed" if str(item.get("status")) in {"committed", "skipped"} else "current", source_label="项目待确认链", risk_level=_risk_level(item.get("riskLevel") or item.get("risk_level")), evidence_ids=[str(eid) for eid in item.get("evidenceIds", []) or item.get("evidence_ids", []) or []], ) if anchor_type == "task": task = next((item for item in truth_state.tasks if str(item.id) == str(anchor_id)), None) if task: return ProjectCopilotObjectAnchorPacket( type="task", id=str(task.id), title=task.title, summary=task.summary, status_label=_task_status_label(task.status), mode="historical" if task.status in {"done", "blocked"} else "current", source_label="项目工作图 / 开发执行", risk_level=self._risk_for_stage(truth_state, task.stage), evidence_ids=[], ) if anchor_type == "release": release = next((item for item in truth_state.releases if str(item.id) == str(anchor_id)), None) if release: return ProjectCopilotObjectAnchorPacket( type="release", id=str(release.id), title=release.title, summary=release.summary, status_label=_release_status_label(release.status), mode="current" if release.status in {"ready", "shipping", "live", "watch"} else "stale", source_label="发布与增长", risk_level="medium", evidence_ids=[], ) if anchor_type == "evidence": evidence = _find_by_id(truth_state.evidence, anchor_id) if evidence: return ProjectCopilotObjectAnchorPacket( type="evidence", id=str(evidence.get("id", anchor_id)), title=str(evidence.get("title") or f"证据 {anchor_id}"), summary=str(evidence.get("excerpt") or evidence.get("summary") or "当前证据。"), status_label="可引用", mode="current", source_label=str(evidence.get("sourceLabel") or evidence.get("source_label") or "项目证据"), risk_level="low", evidence_ids=[str(evidence.get("id", anchor_id))], ) if anchor_type == "project" and str(truth_state.project.id) == str(anchor_id): return self._project_anchor(truth_state) return None def _default_anchor(self, truth_state: ProjectTruthStateProjection, surface: str) -> ProjectCopilotObjectAnchorPacket: pending_items = [item for item in truth_state.changeItems if str(item.get("status")) == "pending_review"] if surface == "change_review" and pending_items: return self._anchor_from_request(truth_state, "change_item", str(pending_items[0].get("id"))) or self._project_anchor(truth_state) if surface in {"execution_bridge", "work_graph"} and truth_state.tasks: selected = next((item for item in truth_state.tasks if str(item.id) == str(truth_state.selectedTaskId)), None) or truth_state.tasks[0] return self._anchor_from_request(truth_state, "task", str(selected.id)) or self._project_anchor(truth_state) if surface == "release_growth" and truth_state.releases: return self._anchor_from_request(truth_state, "release", str(truth_state.releases[0].id)) or self._project_anchor(truth_state) if truth_state.activeChangeSetId: anchor = self._anchor_from_request(truth_state, "change_set", truth_state.activeChangeSetId) if anchor: return anchor return self._project_anchor(truth_state) def _project_anchor(self, truth_state: ProjectTruthStateProjection) -> ProjectCopilotObjectAnchorPacket: high_risk = any(getattr(item, "level", "low") == "high" for item in truth_state.risks) return ProjectCopilotObjectAnchorPacket( type="project", id=str(truth_state.project.id), title=truth_state.project.name, summary=truth_state.project.goal, status_label=_stage_label(truth_state.project.stage), mode="current", source_label="项目真相层", risk_level="high" if high_risk else "low", evidence_ids=[str(item.get("id")) for item in truth_state.evidence[:3] if item.get("id")], ) def _build_context_packet( self, truth_state: ProjectTruthStateProjection, *, body: ProjectCopilotAskRequest, anchor: ProjectCopilotObjectAnchorPacket, execution_summary: dict[str, Any], ) -> ProjectCopilotContextPacket: pending_items = [item for item in truth_state.changeItems if str(item.get("status")) == "pending_review"] accepted_or_edited = [ item for item in truth_state.changeItems if str(item.get("status")) in {"accepted", "edited"} ] high_risk_items = [ item for item in truth_state.changeItems if _risk_level(item.get("riskLevel") or item.get("risk_level")) == "high" ] selected_task = self._selected_task(truth_state, anchor) active_level = self._action_ladder(body.surface, anchor, selected_task, len(pending_items)) allowed_actions = self._allowed_actions(body.surface, anchor, selected_task, len(pending_items), active_level) forbidden_actions = self._forbidden_actions(body.surface, len(pending_items), selected_task) limitations = self._limitations(truth_state, body.surface, anchor, selected_task, len(pending_items)) release_gate = execution_summary.get("releaseGate") if isinstance(execution_summary.get("releaseGate"), dict) else {} release_signals = execution_summary.get("releaseSignals") if isinstance(execution_summary.get("releaseSignals"), dict) else {} release_signal_summary = release_signals.get("summary") if isinstance(release_signals.get("summary"), dict) else {} project_id = str(truth_state.project.id) return ProjectCopilotContextPacket( project_ref={ "id": project_id, "title": truth_state.project.name, "status": _stage_label(truth_state.project.stage), "summary": truth_state.project.goal, }, surface=body.surface, object_anchor=anchor, truth_projection={ "stage": _stage_label(truth_state.project.stage), "loop_stage": _loop_stage_label(truth_state.loopStage), "confirmed_requirements_count": len([item for item in truth_state.requirements if item.status == "confirmed"]), "confirmed_tasks_count": len([item for item in truth_state.tasks if item.status in {"confirmed", "in_progress", "blocked", "done"}]), "open_change_sets_count": len([item for item in truth_state.changeSets if str(item.get("status")) in {"open", "reviewed"}]), "pending_change_items_count": len(pending_items), "evidence_count": len(truth_state.evidence), "high_risk_count": len(high_risk_items), "latest_writeback_at": truth_state.lastWritebackAt, }, change_context={ "active_change_set_id": truth_state.activeChangeSetId, "selected_change_item_id": anchor.id if anchor.type == "change_item" else None, "pending_count": len(pending_items), "accepted_or_edited_count": len(accepted_or_edited), "high_risk_count": len(high_risk_items), "evidence_ids": list(dict.fromkeys(anchor.evidence_ids + self._pending_evidence_ids(pending_items)))[:6], "commit_gate": "blocked_pending_items" if pending_items else "ready_for_human_commit" if accepted_or_edited else "no_open_changes", }, execution_context={ "selected_task_id": str(selected_task.id) if selected_task else None, "selected_task_status": selected_task.status if selected_task else None, "formal_task_available": bool(selected_task and selected_task.status in {"confirmed", "in_progress", "blocked", "done"}), "execution_status": truth_state.execution.status, "prompt_package_ready": bool(truth_state.execution.prompt.strip()), "repo_binding_known": bool(truth_state.execution.commitLog or truth_state.execution.prStatus != "pending"), "artifact_completeness": _string_or_none(execution_summary.get("artifactCompleteness", {}).get("status") if isinstance(execution_summary.get("artifactCompleteness"), dict) else None), "release_readiness": _string_or_none(release_gate.get("status")), }, release_context={ "release_id": str(truth_state.releases[0].id) if truth_state.releases else None, "release_status": _release_status_label(truth_state.releases[0].status) if truth_state.releases else None, "release_gate_status": _string_or_none(release_gate.get("status")), "feedback_signal_count": int(release_signal_summary.get("feedbackCount") or 0), "metric_signal_count": len(truth_state.metricSignals), "gate_blocked_count": len(release_gate.get("blockingReasons", []) or []), }, allowed_actions=allowed_actions, forbidden_actions=forbidden_actions, automation_policy=_low_risk_automation_policy(forbidden_actions), proactive_suggestions=self._proactive_suggestions( project_id=project_id, surface=body.surface, anchor=anchor, selected_task=selected_task, pending_count=len(pending_items), accepted_or_edited_count=len(accepted_or_edited), release_gate=release_gate, allowed_actions=allowed_actions, ), limitations=limitations, action_ladder_level=active_level, ) def _render_answer( self, *, body: ProjectCopilotAskRequest, packet: ProjectCopilotContextPacket, truth_state: ProjectTruthStateProjection, ) -> ProjectCopilotAnswerCanvas: headline = f"{packet.object_anchor.title} · {self._intent_label(body.intent)}" if body.intent == "supplement": direct = ( f"补充信息会先挂到「{packet.object_anchor.title}」这个锚点下；如果它会改变项目事实，" "下一步只能生成候选变更并进入待确认，不能直接写入项目记忆。" ) elif body.intent == "command": direct = ( f"这条指令只能在当前项目和「{packet.object_anchor.title}」范围内生成动作建议。" "执行、发布、权限变更和项目记忆写入都必须回到对应主流程确认。" ) else: direct = self._explain_current_state(packet) evidence_blocks = [ { "display_label": str(item.get("title") or item.get("id")), "display_summary": str(item.get("excerpt") or item.get("summary") or ""), "business_values": {"source": str(item.get("sourceLabel") or item.get("source_label") or "项目证据")}, } for item in truth_state.evidence if str(item.get("id")) in set(packet.object_anchor.evidence_ids) ][:3] next_actions = list(packet.proactive_suggestions[:5]) if not next_actions: next_actions = [ { "title": self._allowed_action_label(action), "label": self._allowed_action_label(action), "action": action, "href": self._action_href(packet.surface, action, packet.project_ref["id"]), "automation_level": "suggest_only", "requires_user_confirm": True, "object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}", "object_anchor": _canvas_object_anchor(packet), } for action in packet.allowed_actions ] return ProjectCopilotAnswerCanvas( headline=headline, direct_answer=direct, evidence_blocks=evidence_blocks, object_refs=[ { "object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}", "display_label": packet.object_anchor.title, "display_summary": packet.object_anchor.summary, } ], next_actions=next_actions, limitations=packet.limitations, ) def _explain_current_state(self, packet: ProjectCopilotContextPacket) -> str: pending = int(packet.change_context.get("pending_count") or 0) high_risk = int(packet.change_context.get("high_risk_count") or 0) if packet.surface == "change_review": if pending: return f"当前有 {pending} 个待确认变更项，其中高风险 {high_risk} 个。副驾只能解释证据、风险和处理顺序，最终写入必须由你确认。" return "当前没有待确认变更项。副驾不会用示例对象补位，可以回到指挥台补充输入或查看已写入项目记忆。" if packet.surface == "execution_bridge": if packet.execution_context.get("formal_task_available"): return "当前可以准备执行上下文包，但副驾不会在 PMFlow 内执行本机命令；真实执行必须由运行端领取正式执行包并回传证据。" return "当前缺少已确认正式任务，候选任务不能直接进入执行桥。" if packet.surface == "release_growth": return "发布与反馈只能形成门禁判断或下一轮待确认变更，不能绕过项目记忆直接改正式对象。" if pending: return f"AI 已生成 {pending} 项待确认内容。下一步应核对证据、处理风险或进入待确认，而不是让副驾直接写入。" return "当前副驾可以解释项目真相、补充候选方向或路由到主流程；没有人工确认前不会写入、执行或发布。" def _action_ladder( self, surface: str, anchor: ProjectCopilotObjectAnchorPacket, selected_task: Any, pending_count: int, ) -> ProjectCopilotActionLadderLevel: if surface == "execution_bridge" or anchor.type == "task": return "handoff_prepare" if surface == "change_review" or anchor.risk_level == "high": return "human_review" if pending_count: return "draft_candidate" return "read_only" def _allowed_actions( self, surface: str, anchor: ProjectCopilotObjectAnchorPacket, selected_task: Any, pending_count: int, ladder: ProjectCopilotActionLadderLevel, ) -> list[ProjectCopilotAllowedAction]: actions: list[ProjectCopilotAllowedAction] = ["answer", "route_to_surface"] if surface in {"command_center", "structured_extraction", "release_growth"} or ladder == "draft_candidate": actions.append("draft_change") if surface in {"change_review", "team_layer"} or pending_count > 0 or anchor.risk_level == "high": actions.append("review_explain") if (surface == "execution_bridge" or anchor.type == "task") and selected_task and selected_task.status in {"confirmed", "in_progress", "blocked", "done"}: actions.append("prepare_execution_handoff") return list(dict.fromkeys(actions)) def _forbidden_actions(self, surface: str, pending_count: int, selected_task: Any) -> list[dict[str, str]]: forbidden = [ {"action": "cross_project_portfolio_ranking", "reason": "项目副驾只绑定当前项目，跨项目排序属于 Space AI。"}, {"action": "platform_provider_model_prompt_diagnosis", "reason": "provider/model/prompt 诊断只允许在平台 AI Ops 入口处理。"}, {"action": "project_memory_write", "reason": "项目记忆写入必须由 ChangeSet commit 主链执行。"}, {"action": "changeset_commit", "reason": "副驾只能解释、建议和路由，不能替用户提交变更集。"}, {"action": "external_command_execution", "reason": "外部命令必须由正式执行包和运行端确认链处理。"}, {"action": "publish_or_deploy", "reason": "发布、部署、删除和权限变更都必须经过发布门禁与人工确认。"}, ] if pending_count: forbidden.append({"action": "execution_handoff_from_pending_change", "reason": "仍有待确认变更项，不能从候选对象直接进入执行桥。"}) if not selected_task or selected_task.status not in {"confirmed", "in_progress", "blocked", "done"}: forbidden.append({"action": "prepare_execution_handoff", "reason": "缺少已确认正式任务，执行交接只能作为后续路由建议。"}) if surface == "team_layer": forbidden.append({"action": "team_layer_direct_commit_or_assignment", "reason": "团队层只做责任路由和阻塞解释，不直接拍板写入或分派执行。"}) return forbidden def _proactive_suggestions( self, *, project_id: str, surface: str, anchor: ProjectCopilotObjectAnchorPacket, selected_task: Any, pending_count: int, accepted_or_edited_count: int, release_gate: dict[str, Any], allowed_actions: list[ProjectCopilotAllowedAction], ) -> list[dict[str, Any]]: suggestions: list[dict[str, Any]] = [] object_ref = f"{anchor.type}:{anchor.id}" object_anchor = { "object_ref": object_ref, "object_type": anchor.type, "object_id": anchor.id, "surface": surface, "project_id": project_id, "project_title": anchor.title if anchor.type == "project" else "", "href": self._surface_href(surface, project_id=project_id, anchor=anchor, selected_task=selected_task), "parent_object_ref": None if anchor.type == "project" else f"project:{project_id}", } def push( *, title: str, reason: str, action: ProjectCopilotAllowedAction, href: str, automation_level: str = "suggest_only", ) -> None: if action not in allowed_actions: return suggestions.append( { "title": title, "label": title, "reason": reason, "why_now": reason, "action": action, "href": href, "automation_level": automation_level, "requires_user_confirm": True, "required_confirmation": "human_confirm_on_target_surface", "risk_level": anchor.risk_level, "source_objects": [object_ref], "audit_event": "project_copilot.proactive_suggestion.created", "object_ref": object_ref, "object_anchor": object_anchor, "blocked_automation": [ "project_memory_write", "changeset_commit", "execution_handoff", "external_command_execution", "publish_or_deploy", ], } ) if pending_count > 0: push( title="进入待确认处理本轮变更", reason=f"仍有 {pending_count} 个待确认项，AI 只能主动把你带到审查入口。", action="review_explain", href=f"/change-review?projectId={project_id}", ) if accepted_or_edited_count > 0 and pending_count == 0: push( title="检查写入门禁再确认提交", reason="已有人工处理过的候选项，但真正写入仍必须由你在 Change Review 提交。", action="route_to_surface", href=f"/change-review?projectId={project_id}", ) if selected_task and str(getattr(selected_task, "status", "")) in {"confirmed", "in_progress", "blocked", "done"}: push( title="准备执行上下文包预览", reason="只准备上下文和证据清单，不在 PMFlow 内执行本机命令。", action="prepare_execution_handoff", href=f"/execution-bridge?projectId={project_id}&taskId={getattr(selected_task, 'id', '')}", automation_level="prepare_only", ) if surface == "release_growth" or str(release_gate.get("status") or "").strip(): push( title="检查发布门禁证据", reason="发布只能先做证据检查和门禁解释，不能自动发布或部署。", action="route_to_surface", href=f"/release-growth?projectId={project_id}", ) if not suggestions: push( title="继续补充项目上下文", reason="AI 可以主动整理候选方向，但有效变更仍会进入待确认链。", action="draft_change", href=f"/command-center?projectId={project_id}", ) return suggestions[:4] def _surface_href( self, surface: str, *, project_id: str, anchor: ProjectCopilotObjectAnchorPacket, selected_task: Any, ) -> str: if surface == "change_review": if anchor.type == "change_item": return f"/change-review?projectId={project_id}&itemId={anchor.id}" if anchor.type == "change_set": return f"/change-review?projectId={project_id}&changeSetId={anchor.id}" return f"/change-review?projectId={project_id}" if surface == "execution_bridge": task_id = anchor.id if anchor.type == "task" else str(getattr(selected_task, "id", "") or "") suffix = f"&taskId={task_id}" if task_id else "" return f"/execution-bridge?projectId={project_id}{suffix}" if surface == "release_growth": suffix = f"&releaseId={anchor.id}" if anchor.type == "release" else "" return f"/release-growth?projectId={project_id}{suffix}" if surface == "team_layer": return f"/team-layer?projectId={project_id}" if surface == "project_memory": return f"/project-memory?projectId={project_id}" if surface == "work_graph": return f"/work-graph?projectId={project_id}" if surface == "structured_extraction": return f"/structured-extraction?projectId={project_id}" return f"/pmflow-os?projectId={project_id}" def _limitations( self, truth_state: ProjectTruthStateProjection, surface: str, anchor: ProjectCopilotObjectAnchorPacket, selected_task: Any, pending_count: int, ) -> list[str]: limitations: list[str] = [] if not truth_state.evidence: limitations.append("当前项目缺少可引用证据，副驾不能给出已证实结论。") if pending_count: limitations.append(f"仍有 {pending_count} 个待确认变更项，正式写入前只能生成候选建议或审查解释。") if (surface == "execution_bridge" or anchor.type == "task") and ( not selected_task or selected_task.status not in {"confirmed", "in_progress", "blocked", "done"} ): limitations.append("缺少已确认正式任务，不能准备真实执行交接。") if surface == "release_growth" and not truth_state.releases: limitations.append("当前缺少发布对象和门禁记录，发布判断只能提示需要补齐证据。") if anchor.mode in {"historical", "stale"}: limitations.append("当前锚点不是实时对象，建议先切回当前变更、任务或项目全局。") return limitations or ["所有建议仍需按当前页面主链确认后才会写入、执行或发布。"] def _selected_task(self, truth_state: ProjectTruthStateProjection, anchor: ProjectCopilotObjectAnchorPacket) -> Any: if anchor.type == "task": return next((item for item in truth_state.tasks if str(item.id) == str(anchor.id)), None) if truth_state.selectedTaskId: selected = next((item for item in truth_state.tasks if str(item.id) == str(truth_state.selectedTaskId)), None) if selected: return selected return truth_state.tasks[0] if truth_state.tasks else None def _risk_for_change_set(self, truth_state: ProjectTruthStateProjection, change_set_id: str) -> str: risks = [ _risk_level(item.get("riskLevel") or item.get("risk_level")) for item in truth_state.changeItems if str(item.get("changeSetId") or item.get("change_set_id")) == str(change_set_id) ] if "high" in risks: return "high" if "medium" in risks: return "medium" return "low" def _risk_for_stage(self, truth_state: ProjectTruthStateProjection, stage: str) -> str: stage_risks = [item.level for item in truth_state.risks if item.stage == stage] if "high" in stage_risks: return "high" if "medium" in stage_risks: return "medium" return "low" def _evidence_for_change_set(self, truth_state: ProjectTruthStateProjection, change_set_id: str) -> list[str]: evidence_ids: list[str] = [] for item in truth_state.changeItems: if str(item.get("changeSetId") or item.get("change_set_id")) != str(change_set_id): continue evidence_ids.extend(str(eid) for eid in item.get("evidenceIds", []) or item.get("evidence_ids", []) or []) return list(dict.fromkeys(evidence_ids))[:6] def _pending_evidence_ids(self, pending_items: list[dict[str, Any]]) -> list[str]: evidence_ids: list[str] = [] for item in pending_items: evidence_ids.extend(str(eid) for eid in item.get("evidenceIds", []) or item.get("evidence_ids", []) or []) return evidence_ids def _intent_label(self, intent: str) -> str: if intent == "supplement": return "补充" if intent == "command": return "指令" return "追问" def _allowed_action_label(self, action: str) -> str: return { "answer": "回答当前项目", "draft_change": "生成候选变更", "review_explain": "解释审查风险", "prepare_execution_handoff": "准备执行交接", "route_to_surface": "跳转主流程", }.get(action, "继续处理") def _action_href(self, surface: str, action: str, project_id: str) -> str: if action == "review_explain": return f"/change-review?projectId={project_id}" if action == "prepare_execution_handoff": return f"/execution-bridge?projectId={project_id}" if action == "draft_change": return f"/structured-extraction?projectId={project_id}" return f"/{surface.replace('_', '-')}?projectId={project_id}" def _find_by_id(items: list[dict[str, Any]], item_id: str) -> dict[str, Any] | None: return next((item for item in items if str(item.get("id")) == str(item_id)), None) def _low_risk_automation_policy(forbidden_actions: list[dict[str, str]]) -> dict[str, Any]: return { "policy_version": "ai_dual_04_low_risk_proactive.v1", "mode": "suggest_only", "allowed_automation_levels": ["suggest_only", "prepare_only"], "allowed_without_user_confirmation": [ "rank_attention", "prepare_navigation", "summarize_context", "prepare_execution_context_preview", ], "requires_user_confirmation_before": [ "project_memory_write", "changeset_commit", "execution_handoff", "external_command_execution", "publish_or_deploy", ], "forbidden_actions": [item.get("action") for item in forbidden_actions if item.get("action")], } def _canvas_object_anchor(packet: ProjectCopilotContextPacket) -> dict[str, Any]: return { "object_ref": f"{packet.object_anchor.type}:{packet.object_anchor.id}", "object_type": packet.object_anchor.type, "object_id": packet.object_anchor.id, "surface": packet.surface, "project_id": str(packet.project_ref.get("id") or ""), "project_title": str(packet.project_ref.get("title") or ""), "href": f"/{str(packet.surface).replace('_', '-')}?projectId={packet.project_ref.get('id')}", "parent_object_ref": None if packet.object_anchor.type == "project" else f"project:{packet.project_ref.get('id')}", } def _risk_level(value: Any) -> str: normalized = str(value or "low").lower() if normalized == "high": return "high" if normalized == "medium": return "medium" return "low" def _stage_label(value: str) -> str: return { "idea": "想法", "validation": "验证", "planning": "规划", "development": "开发", "testing": "测试", "release": "发布", "growth": "增长", }.get(str(value), str(value)) def _loop_stage_label(value: str) -> str: return { "command_idle": "等待输入", "ai_extracting": "AI 提取中", "extraction_ready": "提取待核对", "review_open": "待确认", "writeback_success": "写入完成", }.get(str(value), str(value)) def _change_set_status_label(value: Any) -> str: return { "committed": "已写入", "partially_committed": "部分写入", "reviewed": "已审查", "open": "待确认", }.get(str(value), "待确认") def _change_item_status_label(value: Any) -> str: return { "accepted": "已接受", "edited": "已修改", "deferred": "暂不落库", "committed": "已写入", "skipped": "已跳过", "pending_review": "待确认", }.get(str(value), "待确认") def _task_status_label(value: str) -> str: return { "blocked": "阻塞", "done": "完成", "in_progress": "执行中", "confirmed": "正式任务", "draft": "草稿", }.get(str(value), str(value)) def _release_status_label(value: str) -> str: return { "ready": "可发布", "shipping": "发布中", "live": "已发布", "watch": "观察中", "check_pending": "待检查", }.get(str(value), str(value)) def _string_or_none(value: Any) -> str | None: if value is None: return None text = str(value).strip() return text or None