From 8dde37448ec6f4cf9309a33390f20df7527d630f Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Fri, 13 Mar 2026 01:19:35 +0800 Subject: [PATCH] Update PMFlow docs and services --- CLAUDE.md | 6 + backend/app/api/ai_steward.py | 27 +- backend/app/config/ai_prompts.py | 3 + backend/app/services/ai_pmo_kernel_service.py | 129 ++++++-- backend/app/services/feishu_integration.py | 26 +- backend/app/services/llm.py | 293 +++++++++++++---- .../app/services/project_knowledge_graph.py | 308 +++++++++++++++++- backend/app/services/task_brief_service.py | 94 +++++- backend/tests/test_ai_pmo_kernel_service.py | 96 +++++- .../test_ai_steward_capture_feishu_api.py | 88 +++++ backend/tests/test_feishu_routing_receipts.py | 52 +++ .../tests/test_llm_structured_extraction.py | 149 ++++++++- ...test_project_knowledge_graph_guardrails.py | 117 +++++++ backend/tests/test_task_briefs_api.py | 67 +++- frontend/src/components/layout/Sidebar.tsx | 45 +-- frontend/src/components/layout/TopBar.tsx | 53 ++- 16 files changed, 1426 insertions(+), 127 deletions(-) create mode 100644 backend/tests/test_ai_steward_capture_feishu_api.py create mode 100644 backend/tests/test_project_knowledge_graph_guardrails.py diff --git a/CLAUDE.md b/CLAUDE.md index 71e18028..cb4b4dfb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -79,6 +79,9 @@ backend/ - 新增：`backend/app/api/task_briefs.py`（任务包中心接口） - 新增：`backend/tests/test_model_routing_api.py`（模型路由接口回归测试） - 新增：`backend/tests/test_task_briefs_api.py`（任务包中心接口回归测试） +- 新增：`backend/tests/test_project_knowledge_graph_guardrails.py`（双模型 DDL 治理知识图谱同步回归测试） +- 更新：`backend/app/services/llm.py`（按 function_name 注入 DDL 接口化硬约束，覆盖 low/high 意图模型与 OpenClaw 规划模型） +- 更新：`backend/app/services/project_knowledge_graph.py`（新增 `ai_model -> governance_policy -> interface` 治理图谱同步能力，增量任务新增 `model_guardrails` 阶段） - 移除：`backend/.env.example`（配置模板统一到根目录 `.env.example`） - 新增：`backend/.openclaw/pmflow-routing-rules.json`（容器内兼容同步副本，与根目录规则保持一致） @@ -197,6 +200,7 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - 飞书回执发送仅允许 `main.py` 的 drain_loop 作为消费者，禁止在业务路径即时派发 outbox 发送任务 - 所有写接口必须落审计日志（`operation_logs`）：至少包含 `trace_id/interface_path/caller` - `created_at_beijing` 为计算属性返回，禁止新增数据库物理列 +- low/high 双模型（`classify_intent_ai_pmo_kernel`、`classify_intent_complex`）与 `openclaw_command_dispatch` 必须遵循 DDL 接口化约束：禁止直连 SQL DDL，命中时强制输出 `ddl_api_required` ## 9. 新增代码的参考模式 - 新增后端接口：参考 `backend/app/api/sstc.py` 的结构 @@ -204,6 +208,7 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - 任务包中心接口：参考 `backend/app/api/task_briefs.py`（`/task-briefs/center/templates/runs/generate/contract-sync`） - 审计查询接口：参考 `backend/app/api/audit.py`（`GET /audit/operations`） - 脏数据修复接口：参考 `backend/app/api/ai_steward.py`（`POST /ai-steward/signal-sources/repair-dirty`） +- 飞书捕获模拟接口：`POST /ai-steward/signal/capture-feishu` 在 `direct_reply_required=true` 时会同步写入 `feishu_reply_outbox`，并返回 `reply_outbox_enqueued/reply_outbox_id` 便于链路自检 - 配置读取类接口：参考 `backend/app/api/llm_providers.py`（`GET /ai-steward/llm/providers`） - 项目删除接口（`DELETE /projects/{project_id}`）：参考 `backend/app/api/projects.py`，先清理强依赖外键、再置空弱依赖外键、最后删除主记录（单事务提交） - 项目成员分配接口（`PUT /projects/{project_id}/members`）：参考 `backend/app/api/projects.py` + `backend/app/services/project_member_service.py`，要求显式传入 `member_user_ids` 并写入 `user_projects` @@ -240,6 +245,7 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - Signal Planner 结构化调用采用 `25s` 硬超时：超时或结构化不合法时自动写入降级 `ai_planner(fallback_mode=true)` 并收敛为 `planner_status=completed` - 模型命名统一使用裸模型名（如 `GLM-4.7`、`MiniMax-M2.5`），不使用供应商前缀模型名 - 双 Agent 路由治理：默认启用 `OPENCLAW_ROUTING_ENABLED=true`、关闭 `LLM_MIXED_ROUTING_ENABLED`，复杂意图使用 `classify_intent_complex` 走 high-tier +- 双模型 DDL 治理图谱的模型来源优先读取 `OPENCLAW_ROUTING_RULES_PATH` 指向的路由规则；读取失败时回退 `LLM_ROUTING_LOW_MODEL/LLM_ROUTING_HIGH_MODEL` - 写操作链路统一产出 `decision_trace`（intent/tier/confidence/gate_code/correction_action）用于离线回放与分桶评分 - AI PMO 意图仲裁默认开启（`AI_PMO_INTENT_ARBITRATION_ENABLED=true`）：规则与 LLM 写操作冲突时按 `AI_PMO_INTENT_CONFLICT_POLICY=pending_manual` 强制进入人工确认，不直接自动执行 - `project.create` 启用槽位完整性闸门：无法从指令提取有效项目名时标记 `slot_incomplete` 并进入人工确认 diff --git a/backend/app/api/ai_steward.py b/backend/app/api/ai_steward.py index 9a01d285..43738851 100644 --- a/backend/app/api/ai_steward.py +++ b/backend/app/api/ai_steward.py @@ -30,6 +30,7 @@ from ..services.daily_briefing import get_daily_briefing_service from ..services.event_workflow import get_event_workflow_engine, WorkflowEventType from ..services.adaptive_rag_service import AdaptiveRAGService from ..services.ai_orchestration_run_service import AIOrchestrationRunService +from ..services.feishu_outbox_service import FeishuOutboxService from ..services.openclaw_commander_service import OpenClawCommanderService from ..services.signal_source_orchestration_service import SignalSourceOrchestrationService from ..services.signal_source_repair_service import ( @@ -448,7 +449,31 @@ async def capture_feishu_signal( tenant_id=tenant_id, feishu_message_id=request.feishu_message_id, ) - return result + response_payload = dict(result) if isinstance(result, dict) else {"result": result} + + should_enqueue_reply = bool(response_payload.get("direct_reply_required")) + reply_text = str(response_payload.get("direct_reply_text") or "").strip() or "已收到，正在处理。" + source_message_id = (str(request.feishu_message_id or "").strip()[:128] or None) + normalized_chat_id = str(request.group_chat_id or "").strip() + if should_enqueue_reply and normalized_chat_id: + try: + outbox_id = await FeishuOutboxService.enqueue( + db, + tenant_id=int(tenant_id), + chat_id=normalized_chat_id, + text=reply_text, + source_message_id=source_message_id, + dedupe_seed=f"capture-feishu:{normalized_chat_id}:{str(request.sender or '').strip()[:64]}", + trace_id=str(response_payload.get("ai_trace_id") or "").strip()[:64] or None, + ) + await db.commit() + response_payload["reply_outbox_enqueued"] = bool(outbox_id) + response_payload["reply_outbox_id"] = int(outbox_id) if outbox_id else None + except Exception as outbox_exc: + await db.rollback() + response_payload["reply_outbox_enqueued"] = False + response_payload["reply_outbox_error"] = str(outbox_exc)[:200] + return response_payload except Exception as e: raise HTTPException(status_code=500, detail=f"飞书信号捕获失败: {str(e)}") diff --git a/backend/app/config/ai_prompts.py b/backend/app/config/ai_prompts.py index e62de81c..45ebdb27 100644 --- a/backend/app/config/ai_prompts.py +++ b/backend/app/config/ai_prompts.py @@ -422,6 +422,8 @@ PROMPT_OPENCLAW_COMMAND_DISPATCH = """ 11. 字段未知时返回空字符串、空数组或 null。 12. 信息不足时也必须返回合法 JSON，operations=[]。 13. 不要 Markdown，不要解释，只返回 JSON。 +14. 严禁输出任何 DDL 直连数据库执行方案（CREATE/ALTER/DROP/TRUNCATE/RENAME/COMMENT/GRANT/REVOKE）。 +15. 若输入涉及数据库 DDL 变更，必须返回 operations=[]，并在 summary 明确写出 ddl_api_required（表示必须走受控接口/迁移流程）。 """ @@ -527,6 +529,7 @@ PROMPT_AI_PMO_INTENT_CLASSIFICATION = """ 5. 禁止输出 thinking、analysis、解释句、Markdown、代码块。 6. 若消息表达“新增/增加/添加/新建/创建/建立 + 项目”（如“给我新增个项目”），优先判定 project.create；仅当明确是“信号源/来源/webhook/rss/github 接入或创建信号”时才判 signal_source.onboard 或 signal.create。 7. 只输出 JSON，不要额外文本。 +8. 若消息要求直接执行数据库 DDL（建表/改表/删表等），禁止判定为可自动写入意图，必须返回 query.read_only，并在 reasoning 包含 ddl_api_required。 """ diff --git a/backend/app/services/ai_pmo_kernel_service.py b/backend/app/services/ai_pmo_kernel_service.py index a59d148a..8b5a0b19 100644 --- a/backend/app/services/ai_pmo_kernel_service.py +++ b/backend/app/services/ai_pmo_kernel_service.py @@ -376,6 +376,8 @@ class IntentResolver: OPERATION_VERBS = ( "新增", "增加", + "添加", + "加", "加一个", "加个", "新建", @@ -394,6 +396,26 @@ class IntentResolver: "执行", ) + @staticmethod + def _looks_like_negated_write_statement(*, text: str) -> bool: + normalized = re.sub(r"\s+", "", str(text or "")).lower() + if not normalized: + return False + negation_tokens = ("别", "别再", "别乱", "不要", "不要再", "先别", "先不要", "禁止", "不能", "无需", "不用") + if not any(token in normalized for token in negation_tokens): + return False + write_tokens = ("创建", "新建", "新增", "增加", "添加", "加", "接入", "执行", "分配", "指派", "立项") + object_tokens = ("信号", "需求", "任务", "项目", "事项", "工单", "流程", "信号源") + if any(w in normalized for w in write_tokens) and any(o in normalized for o in object_tokens): + return True + return bool( + re.search( + r"(?:别|不要|先别|先不要|禁止|不能).{0,8}(?:创建|新建|新增|增加|添加|加|接入|执行|分配|指派|立项)", + normalized, + re.IGNORECASE, + ) + ) + @staticmethod def resolve(*, text: str, project_id: Optional[int]) -> CanonicalIntent: normalized = re.sub(r"\s+", "", str(text or "")).lower() @@ -402,6 +424,15 @@ class IntentResolver: return CanonicalIntent(INTENT_IGNORE, 1.0, project_id, "", "", {"reason": "empty"}) if IntentResolver._is_smalltalk_message(pretty=pretty, normalized=normalized): return CanonicalIntent(INTENT_IGNORE, 0.95, project_id, "", "", {"reason": "smalltalk"}) + if IntentResolver._looks_like_negated_write_statement(text=pretty): + return CanonicalIntent( + INTENT_QUERY, + 0.9, + project_id, + "", + pretty[:300], + {"reason": "negated_write_statement"}, + ) if any(token in pretty for token in FOLLOWUP_CONFIRM_HINTS): return CanonicalIntent(INTENT_FOLLOWUP_CONFIRM, 0.82, project_id, "", "", {}) if any(token in pretty for token in FOLLOWUP_CONTINUE_HINTS): @@ -427,6 +458,15 @@ class IntentResolver: pretty[:300], {}, ) + if IntentResolver._looks_like_signal_create_request(text=pretty): + return CanonicalIntent( + INTENT_SIGNAL_CREATE, + 0.78, + project_id, + IntentResolver._build_signal_title(pretty, default="新信号"), + pretty[:300], + {}, + ) if IntentResolver._looks_like_query(text=pretty): return CanonicalIntent( INTENT_QUERY, @@ -436,15 +476,6 @@ class IntentResolver: pretty[:300], {}, ) - if any(token in pretty for token in SIGNAL_CREATE_HINTS) or any(token in pretty for token in ("新增", "增加", "录入", "跟踪", "关注")): - return CanonicalIntent( - INTENT_SIGNAL_CREATE, - 0.78, - project_id, - IntentResolver._build_signal_title(pretty, default="新信号"), - pretty[:300], - {}, - ) return CanonicalIntent( INTENT_QUERY, 0.6, @@ -454,6 +485,28 @@ class IntentResolver: {"fallback": True, "reason": "default_query_fallback"}, ) + @staticmethod + def _looks_like_signal_create_request(*, text: str) -> bool: + normalized = re.sub(r"\s+", "", str(text or "")).lower() + if not normalized: + return False + if IntentResolver._looks_like_negated_write_statement(text=normalized): + return False + if any(token in normalized for token in SIGNAL_CREATE_HINTS): + return True + signal_object_hints = ("信号", "需求", "任务", "事项", "问题", "改造", "优化") + has_operation = any(token in normalized for token in IntentResolver.OPERATION_VERBS) + has_signal_object = any(token in normalized for token in signal_object_hints) + if has_operation and has_signal_object: + return True + return bool( + re.search( + r"(?:给|帮|请).{0,20}(?:新增|增加|添加|补充|加).{0,20}(?:信号|需求|任务|事项)", + normalized, + re.IGNORECASE, + ) + ) + @staticmethod def _is_smalltalk_message(*, pretty: str, normalized: str) -> bool: compact = re.sub(r"[\\s\\?？!！,，。~～、]+", "", str(pretty or "")).strip().lower() @@ -472,6 +525,8 @@ class IntentResolver: normalized = re.sub(r"\s+", "", str(text or "")).lower() if not normalized: return False + if IntentResolver._looks_like_signal_create_request(text=normalized): + return False if any(token in normalized for token in QUESTION_HINTS): return True has_question_structure = ( @@ -1004,16 +1059,23 @@ class TypedExecutor: ) -> dict[str, Any]: if intent.name == INTENT_QUERY: rag_service = AdaptiveRAGService(self.db, tenant_id=self.tenant_id) - rag_result = await rag_service.answer_question( - question=message_text, - asked_by=operator_user_id, - asked_via=source_system, - project_id=intent.project_id, - source_chat_id=source_message_id, - conversation_key=source_message_id, - action_mode="read", - sync_requirement=False, - ) + try: + rag_result = await rag_service.answer_question( + question=message_text, + asked_by=operator_user_id, + asked_via=source_system, + project_id=intent.project_id, + source_chat_id=source_message_id, + conversation_key=source_message_id, + action_mode="read", + sync_requirement=False, + ) + except Exception as exc: + return { + "status": "failed_safe", + "error_message": self._build_query_error_message(exc), + "project_id": intent.project_id, + } return { "status": "succeeded", "answer": str(rag_result.get("answer") or "").strip(), @@ -1139,6 +1201,20 @@ class TypedExecutor: "error_message": "当前意图尚未迁移到 AI PMO kernel，已保留给兼容链路处理。", } + @staticmethod + def _build_query_error_message(error: Exception) -> str: + reason = re.sub(r"\s+", " ", str(error or "")).strip().strip("'\"") + reason = re.sub(r"^(runtimeerror|valueerror|exception)\s*:\s*", "", reason, flags=re.IGNORECASE) + if len(reason) > 120: + reason = f"{reason[:120]}..." + marker = "模型调用失败：" + while reason.startswith(marker): + reason = reason[len(marker):].strip() + if not reason: + reason = "unknown_error" + message = f"{marker}{reason}" + return message if "请稍后重试" in message else f"{message}。请稍后重试。" + async def _create_signal( self, *, @@ -1395,6 +1471,18 @@ class AIPMOKernelService: message_text=clean_text, closure_payload=closure_payload, ) + if intent.name in WRITE_INTENTS and IntentResolver._looks_like_negated_write_statement(text=clean_text): + metadata = dict(intent.metadata or {}) + metadata["guard_reason"] = "negated_write_statement" + metadata["guarded_from_intent"] = intent.name + intent = CanonicalIntent( + INTENT_QUERY, + max(float(intent.confidence or 0.0), 0.9), + intent.project_id, + "", + clean_text[:300], + metadata, + ) perf_intent_name = intent.name _log_perf("_resolve_intent", stage_started_at) if intent_context_source: @@ -2630,6 +2718,9 @@ class AIPMOKernelService: ) -> tuple[CanonicalIntent, str]: project_id = _safe_int(world_model.get("resolved_project_id")) rule_intent = self.intent_resolver.resolve(text=message_text, project_id=project_id) + # 查询/噪音场景优先走规则意图，避免额外触发 LLM 分类链路导致不必要失败重试。 + if rule_intent.name in {INTENT_QUERY, INTENT_IGNORE} and float(rule_intent.confidence or 0.0) >= 0.85: + return self._apply_slot_guard(intent=rule_intent, message_text=message_text), "rule_preferred" if self._is_short_confirmation_message(message_text=message_text, session=session): if session is None: return rule_intent, "rule_fallback" diff --git a/backend/app/services/feishu_integration.py b/backend/app/services/feishu_integration.py index e7f9952e..44446e8b 100644 --- a/backend/app/services/feishu_integration.py +++ b/backend/app/services/feishu_integration.py @@ -371,6 +371,7 @@ class FeishuIntegrationService: ) reply_route = "none" + capability_reply = self._build_capability_reply(message_content) if scene == "execute": if self._is_duplicate_capture_result(result): self._emit_state_transition( @@ -384,7 +385,7 @@ class FeishuIntegrationService: state="ROUTED", route="tool_first_execute", ) - deterministic_reply = str(result.get("direct_reply_text") or "").strip() + deterministic_reply = capability_reply or str(result.get("direct_reply_text") or "").strip() if not deterministic_reply: deterministic_reply = self._build_execution_receipt_reply( message_content=message_content, @@ -423,7 +424,18 @@ class FeishuIntegrationService: state="ROUTED", route="qa_consult", ) - if chat_id and bool(result.get("direct_reply_required")): + if chat_id and capability_reply: + await self._safe_reply_text( + chat_id=chat_id, + mention_open_id=sender_open_id, + mention_name=sender_name, + text=capability_reply, + trace_id=state_trace_id, + source_message_id=message_id, + perf_t0=perf_t0, + ) + reply_route = "qa_capability" + elif chat_id and bool(result.get("direct_reply_required")): await self._safe_reply_text( chat_id=chat_id, mention_open_id=sender_open_id, @@ -1024,6 +1036,16 @@ class FeishuIntegrationService: ) return any(token in normalized for token in capability_tokens) + @classmethod + def _build_capability_reply(cls, message_content: Optional[str]) -> Optional[str]: + if not cls._is_capability_question(message_content): + return None + return ( + "我是 PMFlow 的项目协同助手。" + "我可以识别需求并入库信号、关联项目和任务、跟进状态并回答项目问题。" + "你直接说目标、对象和期望动作即可。" + ) + @staticmethod def _strip_identity_lead(reply_text: str) -> str: text = str(reply_text or "").strip() diff --git a/backend/app/services/llm.py b/backend/app/services/llm.py index f8f08049..28542ba9 100644 --- a/backend/app/services/llm.py +++ b/backend/app/services/llm.py @@ -179,6 +179,12 @@ FAST_LOCAL_WORKER_FUNCTION_HINTS = ("signal_source_ai_worker_execution",) FAST_LOCAL_FALLBACK_FAST_TIMEOUT_SECONDS = 3.0 FAST_LOCAL_FALLBACK_COMPLETE_TIMEOUT_SECONDS = 6.0 STRUCTURED_CACHE_TTL_SECONDS = 300 +STRICT_MODEL_ERROR_FUNCTION_PREFIXES = ( + "adaptive_rag_answer_", +) +STRICT_MODEL_ERROR_FUNCTION_EXACT = { + "rag_db_project_semantic_retrieval", +} FAIL_FAST_ERROR_MARKERS = ( "primary_engine_empty_response", "high_tier_empty_response", @@ -189,6 +195,19 @@ FAIL_FAST_ERROR_MARKERS = ( ) CIRCUIT_BREAKER_FAILURE_THRESHOLD = 2 CIRCUIT_BREAKER_COOLDOWN_SECONDS = 300.0 +DDL_API_GUARDRAIL_FUNCTION_EXACT = { + "classify_intent_ai_pmo_kernel", + "classify_intent_complex", + "openclaw_command_dispatch", +} +DDL_API_GUARDRAIL_SYSTEM_PROMPT = """ +硬性约束（最高优先级）： +1. 禁止直接执行或建议执行数据库 DDL SQL（CREATE/ALTER/DROP/TRUNCATE/RENAME/COMMENT/GRANT/REVOKE）。 +2. 所有数据库 DDL 变更必须通过受控接口或迁移流程（例如 Alembic migration）。 +3. 若输入要求直接改库执行 DDL： + - `classify_intent_ai_pmo_kernel` / `classify_intent_complex` 必须返回 `query.read_only`，并在 reasoning 包含 `ddl_api_required`。 + - `openclaw_command_dispatch` 必须返回 `operations=[]`，并在 summary 包含 `ddl_api_required`。 +""".strip() STRUCTURED_OBJECT_SCHEMAS: dict[str, dict[str, Any]] = { "signal_source_onboarding_plan": { "required_keys": ("should_onboard", "summary", "worker_roles", "openclaw_instruction"), @@ -699,15 +718,21 @@ class LLMService: routing_decision = "primary_engine" engine_meta: dict[str, Any] = {} prompt_profile = {"role_name": "", "system_prompt": ""} - merged_prompt = self._merge_prompt_with_profile(prompt=prompt, prompt_profile=prompt_profile) + merged_prompt = self._merge_prompt_with_profile( + prompt=prompt, + prompt_profile=prompt_profile, + function_name=function_name, + ) safety_attempted = False - def _log_stub_failure(error_obj: Exception) -> None: + def _log_stub_failure(error_obj: Exception, *, used_stub: bool = True) -> None: elapsed_ms = (time.perf_counter() - started_at) * 1000 provider = str(engine_meta.get("provider") or "") model = str(engine_meta.get("model") or "") + outcome = "兜底 stub" if used_stub else "显式报错" logger.warning( - "[LLM_FAIL] 生成失败 → 兜底 stub function=%s provider=%s model=%s error=%r duration_ms=%.0f", + "[LLM_FAIL] 生成失败 → %s function=%s provider=%s model=%s error=%r duration_ms=%.0f", + outcome, function_name, provider, model, @@ -718,7 +743,11 @@ class LLMService: try: if use_prompt_profile: prompt_profile = await self._resolve_active_prompt_profile(db, tenant_id=tenant_id) - merged_prompt = self._merge_prompt_with_profile(prompt=prompt, prompt_profile=prompt_profile) + merged_prompt = self._merge_prompt_with_profile( + prompt=prompt, + prompt_profile=prompt_profile, + function_name=function_name, + ) engine = await self._resolve_engine_for_call( db, tenant_id=tenant_id, @@ -815,13 +844,18 @@ class LLMService: return response_text except Exception as exc: upgraded_exc: Exception = exc - if self._is_fail_fast_exception(exc): + strict_model_error = self._should_raise_model_error(function_name) + if self._is_fail_fast_exception(exc) and not strict_model_error: status = "failed_fast" response_text = self._stub_response(prompt) _runtime_log("LLM", f"生成快速失败，已跳过升级重试：{exc}") _log_stub_failure(exc) return response_text - if self._is_low_tier_function(function_name): + if self._is_fail_fast_exception(exc) and strict_model_error: + status = "failed_fast" + _runtime_log("LLM", f"生成快速失败，严格模式继续升级重试：{exc}") + should_try_high_tier_retry = self._is_low_tier_function(function_name) or strict_model_error + if should_try_high_tier_retry: if self._prefer_fast_local_fallback(function_name): raise upgraded_exc high_engine = await self._resolve_high_tier_engine_for_escalation( @@ -863,9 +897,29 @@ class LLMService: response_text = escalated_text return response_text except Exception as high_exc: + _runtime_log( + "LLM升级", + ( + "高阶兜底失败：" + f"function={function_name}, provider={high_provider}, " + f"model={high_model}, error={high_exc}" + ), + ) upgraded_exc = ValueError(f"high_tier_retry_failed:{high_exc}") engine_meta = high_engine + if strict_model_error: + status = "failed" + reason_text = self._build_model_error_message( + function_name=function_name, + provider=str(engine_meta.get("provider") or ""), + model=str(engine_meta.get("model") or ""), + error=upgraded_exc, + ) + response_text = reason_text + _log_stub_failure(upgraded_exc, used_stub=False) + raise RuntimeError(reason_text) from upgraded_exc + if not safety_attempted: safety_attempted = True safety_text, safety_engine = await self._try_safety_engine( @@ -931,7 +985,11 @@ class LLMService: return response_text finally: try: - merged_prompt = self._merge_prompt_with_profile(prompt=prompt, prompt_profile=prompt_profile) + merged_prompt = self._merge_prompt_with_profile( + prompt=prompt, + prompt_profile=prompt_profile, + function_name=function_name, + ) await self._log_ai_call( tenant_id=tenant_id, function_name=function_name, @@ -973,7 +1031,11 @@ class LLMService: try: if use_prompt_profile: prompt_profile = await self._resolve_active_prompt_profile(db, tenant_id=tenant_id) - merged_prompt = self._merge_prompt_with_profile(prompt=prompt, prompt_profile=prompt_profile) + merged_prompt = self._merge_prompt_with_profile( + prompt=prompt, + prompt_profile=prompt_profile, + function_name=function_name, + ) engine = await self._resolve_engine_for_call( db, tenant_id=tenant_id, @@ -1133,7 +1195,10 @@ class LLMService: raise ValueError(f"provider_circuit_open:{provider_name}:{model}") key = api_key.strip() messages: list[dict[str, str]] = [] - cleaned_system_prompt = str(system_prompt or "").strip() + cleaned_system_prompt = self._inject_function_guardrails( + system_prompt=system_prompt, + function_name=function_name, + ) if cleaned_system_prompt: messages.append({"role": "system", "content": cleaned_system_prompt}) messages.append({"role": "user", "content": prompt}) @@ -1183,67 +1248,96 @@ class LLMService: if not key: raise ValueError("未配置 LLM API Key") endpoint = f"{base_url.rstrip('/')}/chat/completions" - payload = { - "model": model, - "messages": messages, - "temperature": temperature, - "max_tokens": max_tokens, - } - if is_openclaw_gateway_request: - payload["agents"] = { - "defaults": { - "timeoutSeconds": int(openclaw_agents_timeout_seconds), - } - } headers = { "Authorization": f"Bearer {key}", "Content-Type": "application/json", } - async with httpx.AsyncClient(timeout=request_timeout, trust_env=False) as client: - resp = await client.post(endpoint, headers=headers, json=payload) + current_max_tokens = max(int(max_tokens or 0), 1) + empty_content_retry_budget = 1 + while True: + payload = { + "model": model, + "messages": messages, + "temperature": temperature, + "max_tokens": current_max_tokens, + } if is_openclaw_gateway_request: - elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0 - _runtime_log( - "OpenClaw调用", - ( - f"http ts={_cn_now_text_ms()} trace_id={request_trace_id} " - f"function={function_label} status={resp.status_code} elapsed_ms={elapsed_ms:.0f}" - ), - ) - if resp.status_code >= 400: - body_preview = str(resp.text or "").replace("\n", " ").strip()[:260] + payload["agents"] = { + "defaults": { + "timeoutSeconds": int(openclaw_agents_timeout_seconds), + } + } + async with httpx.AsyncClient(timeout=request_timeout, trust_env=False) as client: + resp = await client.post(endpoint, headers=headers, json=payload) if is_openclaw_gateway_request: + elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0 _runtime_log( "OpenClaw调用", ( - f"http_error ts={_cn_now_text_ms()} trace_id={request_trace_id} " - f"function={function_label} status={resp.status_code} body={body_preview}" + f"http ts={_cn_now_text_ms()} trace_id={request_trace_id} " + f"function={function_label} status={resp.status_code} elapsed_ms={elapsed_ms:.0f}" ), ) - resp.raise_for_status() - data = resp.json() - choices = data.get("choices") or [] - if not choices: - raise ValueError("LLM 响应无 choices") - message = choices[0].get("message") if isinstance(choices[0], dict) else {} - text = self._extract_text_from_chat_response( - message=message, - choice=choices[0] if isinstance(choices[0], dict) else {}, - response=data, - structured_output=structured_output, - ) - if not text and not structured_output: + if resp.status_code >= 400: + body_preview = str(resp.text or "").replace("\n", " ").strip()[:260] + if is_openclaw_gateway_request: + _runtime_log( + "OpenClaw调用", + ( + f"http_error ts={_cn_now_text_ms()} trace_id={request_trace_id} " + f"function={function_label} status={resp.status_code} body={body_preview}" + ), + ) + resp.raise_for_status() + data = resp.json() + choices = data.get("choices") or [] + if not choices: + raise ValueError("LLM 响应无 choices") + first_choice = choices[0] if isinstance(choices[0], dict) else {} + message = first_choice.get("message") if isinstance(first_choice, dict) else {} + text = self._extract_text_from_chat_response( + message=message, + choice=first_choice, + response=data, + structured_output=structured_output, + ) + if text or structured_output: + break + msg_keys = list(message.keys()) if isinstance(message, dict) else [] - choice_keys = list(choices[0].keys()) if isinstance(choices[0], dict) else [] + choice_keys = list(first_choice.keys()) if isinstance(first_choice, dict) else [] choice_preview = "" try: - choice_preview = json.dumps(choices[0], ensure_ascii=False)[:300] + choice_preview = json.dumps(first_choice, ensure_ascii=False)[:300] except Exception: - choice_preview = str(choices[0])[:300] - raise ValueError( + choice_preview = str(first_choice)[:300] + empty_error = ValueError( "LLM 返回空内容: " f"message_keys={msg_keys}, choice_keys={choice_keys}, choice_preview={choice_preview}" ) + can_retry_empty = ( + empty_content_retry_budget > 0 + and self._should_retry_empty_chat_response( + structured_output=structured_output, + message=message, + choice=first_choice, + current_max_tokens=current_max_tokens, + ) + ) + if not can_retry_empty: + raise empty_error + + empty_content_retry_budget -= 1 + next_max_tokens = self._next_retry_max_tokens_for_empty_chat(current_max_tokens) + _runtime_log( + "LLM", + ( + "空内容提额重试：" + f"function={function_label}, provider={provider_name}, model={model}, " + f"max_tokens={current_max_tokens}->{next_max_tokens}" + ), + ) + current_max_tokens = next_max_tokens if _looks_like_openclaw_timeout_message(text): elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0 _runtime_log( @@ -1894,8 +1988,38 @@ class LLMService: return self._looks_like_unanswerable_response(response_text) @staticmethod - def _merge_prompt_with_profile(*, prompt: str, prompt_profile: dict[str, str]) -> str: - system_prompt = str(prompt_profile.get("system_prompt") or "").strip() + def _requires_ddl_api_guardrail(function_name: str) -> bool: + fn = str(function_name or "").strip().lower() + if not fn: + return False + return fn in DDL_API_GUARDRAIL_FUNCTION_EXACT + + @classmethod + def _inject_function_guardrails(cls, *, system_prompt: str, function_name: str) -> str: + base_prompt = str(system_prompt or "").strip() + if not cls._requires_ddl_api_guardrail(function_name): + return base_prompt + extra = str(DDL_API_GUARDRAIL_SYSTEM_PROMPT or "").strip() + if not extra: + return base_prompt + if extra in base_prompt: + return base_prompt + if not base_prompt: + return extra + return f"{base_prompt}\n\n{extra}" + + @classmethod + def _merge_prompt_with_profile( + cls, + *, + prompt: str, + prompt_profile: dict[str, str], + function_name: str = "", + ) -> str: + system_prompt = cls._inject_function_guardrails( + system_prompt=str(prompt_profile.get("system_prompt") or "").strip(), + function_name=function_name, + ) if not system_prompt: return prompt role_name = str(prompt_profile.get("role_name") or "unknown").strip() or "unknown" @@ -1911,6 +2035,36 @@ class LLMService: return False return any(hint and hint in fn for hint in FAST_LOCAL_FALLBACK_FUNCTION_HINTS) + @staticmethod + def _should_raise_model_error(function_name: str) -> bool: + fn = str(function_name or "").strip().lower() + if not fn: + return False + if fn in STRICT_MODEL_ERROR_FUNCTION_EXACT: + return True + return any(prefix and fn.startswith(prefix) for prefix in STRICT_MODEL_ERROR_FUNCTION_PREFIXES) + + @staticmethod + def _build_model_error_message( + *, + function_name: str, + provider: str, + model: str, + error: Exception, + ) -> str: + clean_error = re.sub(r"\s+", " ", str(error or "")).strip() + if len(clean_error) > 160: + clean_error = f"{clean_error[:160]}..." + provider_text = str(provider or "unknown").strip() or "unknown" + model_text = str(model or "unknown").strip() or "unknown" + fn_text = str(function_name or "llm_generate").strip() or "llm_generate" + if not clean_error: + clean_error = "upstream_empty_or_invalid_response" + return ( + f"模型调用失败：function={fn_text}, provider={provider_text}, model={model_text}, " + f"reason={clean_error}" + ) + @staticmethod def _is_fast_local_planner_function(function_name: str) -> bool: fn = str(function_name or "").strip().lower() @@ -1968,6 +2122,34 @@ class LLMService: configured = OPENCLAW_AGENTS_DEFAULT_TIMEOUT_SECONDS return min(configured, OPENCLAW_AGENTS_TIMEOUT_SECONDS_MAX) + @staticmethod + def _should_retry_empty_chat_response( + *, + structured_output: bool, + message: Any, + choice: Any, + current_max_tokens: int, + ) -> bool: + if structured_output: + return False + if int(current_max_tokens or 0) >= 2800: + return False + finish_reason = str(choice.get("finish_reason") or "").strip().lower() if isinstance(choice, dict) else "" + if finish_reason not in {"length", "max_tokens"}: + return False + reasoning_text = "" + if isinstance(message, dict): + reasoning_text = str(message.get("reasoning_content") or message.get("reasoning") or "").strip() + if not reasoning_text and isinstance(choice, dict): + reasoning_text = str(choice.get("reasoning_content") or choice.get("reasoning") or "").strip() + return bool(reasoning_text) + + @staticmethod + def _next_retry_max_tokens_for_empty_chat(current_max_tokens: int) -> int: + current = max(int(current_max_tokens or 0), 1) + grown = max(current + 512, int(current * 1.8)) + return min(grown, 3000) + def _use_provider_circuit_breaker(self, function_name: str) -> bool: return not self._prefer_fast_local_fallback(function_name) @@ -2540,7 +2722,8 @@ class LLMService: if base_url_override: routed["base_url"] = base_url_override elif self._openclaw_routing_enabled() and provider == "openai": - routed["base_url"] = OPENCLAW_PRIMARY_GATEWAY_URL + configured_openai_base = str(os.getenv("OPENAI_BASE_URL", "") or "").strip().rstrip("/") + routed["base_url"] = configured_openai_base or OPENCLAW_PRIMARY_GATEWAY_URL if str(routed.get("id") or "") != str(active_engine.get("id") or ""): _runtime_log( diff --git a/backend/app/services/project_knowledge_graph.py b/backend/app/services/project_knowledge_graph.py index 422fc5eb..44c2d32c 100644 --- a/backend/app/services/project_knowledge_graph.py +++ b/backend/app/services/project_knowledge_graph.py @@ -10,6 +10,8 @@ from __future__ import annotations import hashlib +import json +import os from datetime import datetime, timedelta, timezone from pathlib import Path from typing import Any, Optional @@ -41,6 +43,20 @@ def _utc_now_naive() -> datetime: return now_cn_naive() +DDL_BLOCKED_PREFIXES = [ + "CREATE", + "ALTER", + "DROP", + "TRUNCATE", + "RENAME", + "COMMENT", + "GRANT", + "REVOKE", +] +DDL_GUARDRAIL_POLICY_KEY = "policy:ddl_via_interface_only" +DDL_GUARDRAIL_POLICY_NAME = "DDL 只能通过受控接口/迁移流程" + + class ProjectKnowledgeGraphService: def __init__(self, db: AsyncSession, *, tenant_id: int) -> None: self.db = db @@ -1093,6 +1109,284 @@ class ProjectKnowledgeGraphService: counters["evidences"] += 1 return counters + @staticmethod + def _routing_rule_candidate_paths() -> list[Path]: + candidates: list[Path] = [] + configured = str(os.getenv("OPENCLAW_ROUTING_RULES_PATH", "") or "").strip() + if configured: + candidates.append(Path(configured).expanduser()) + + backend_root = Path(__file__).resolve().parents[2] + project_root = Path(__file__).resolve().parents[3] + candidates.extend( + [ + project_root / ".openclaw" / "pmflow-routing-rules.json", + backend_root / ".openclaw" / "pmflow-routing-rules.json", + ] + ) + + deduped: list[Path] = [] + seen: set[str] = set() + for path in candidates: + key = str(path) + if key in seen: + continue + seen.add(key) + deduped.append(path) + return deduped + + @classmethod + def _resolve_guardrail_target_models(cls) -> dict[str, dict[str, str]]: + specs: dict[str, dict[str, str]] = { + "low": { + "tier": "low", + "provider": str(os.getenv("LLM_ROUTING_LOW_PROVIDER", "glm") or "glm").strip(), + "model": str(os.getenv("LLM_ROUTING_LOW_MODEL", "GLM-4.7") or "GLM-4.7").strip(), + "api_env": str(os.getenv("LLM_ROUTING_LOW_API_ENV", "GLM_API_KEY") or "GLM_API_KEY").strip(), + "source": "env_default", + "source_path": "", + "rule_name": "", + }, + "high": { + "tier": "high", + "provider": str(os.getenv("LLM_ROUTING_HIGH_PROVIDER", "openai") or "openai").strip(), + "model": str(os.getenv("LLM_ROUTING_HIGH_MODEL", "MiniMax-M2.5") or "MiniMax-M2.5").strip(), + "api_env": str(os.getenv("LLM_ROUTING_HIGH_API_ENV", "OPENAI_API_KEY") or "OPENAI_API_KEY").strip(), + "source": "env_default", + "source_path": "", + "rule_name": "", + }, + } + + for path in cls._routing_rule_candidate_paths(): + try: + content = path.read_text(encoding="utf-8") + payload = json.loads(content) + except Exception: + continue + rows = payload.get("rules") if isinstance(payload, dict) and isinstance(payload.get("rules"), list) else [] + if not rows: + continue + enabled_rows = [row for row in rows if isinstance(row, dict) and bool(row.get("enabled", True))] + enabled_rows.sort(key=lambda item: int(item.get("priority") or 0), reverse=True) + resolved_any = False + for tier in ("low", "high"): + for row in enabled_rows: + if str(row.get("tier") or "").strip().lower() != tier: + continue + model = str(row.get("model") or "").strip() + if not model: + continue + specs[tier] = { + "tier": tier, + "provider": str(row.get("provider") or specs[tier]["provider"]).strip(), + "model": model, + "api_env": str(row.get("api_env") or specs[tier]["api_env"]).strip(), + "source": "routing_rule", + "source_path": str(path), + "rule_name": str(row.get("name") or "").strip(), + } + resolved_any = True + break + if resolved_any and all(str(specs[tier].get("source") or "") == "routing_rule" for tier in ("low", "high")): + break + return specs + + async def _sync_model_guardrail_entities(self) -> dict[str, Any]: + counters = {"models": 0, "policies": 0, "interfaces": 0, "relations": 0, "evidences": 0} + model_specs = self._resolve_guardrail_target_models() + + policy_entity = await self.upsert_entity( + entity_type="governance_policy", + entity_key=DDL_GUARDRAIL_POLICY_KEY, + display_name=DDL_GUARDRAIL_POLICY_NAME, + payload={ + "policy_code": "ddl_via_interface_only", + "enforcement_level": "hard_block", + "blocked_sql_prefixes": DDL_BLOCKED_PREFIXES, + "required_execution_path": "api_or_migration_interface", + "enforced_by": [ + "backend/app/database.py", + "backend/app/services/llm.py", + "backend/app/config/ai_prompts.py", + ], + }, + source_type="governance", + source_ref="backend/app/database.py", + ) + counters["policies"] += 1 + await self.add_evidence( + evidence_type="policy_guardrail", + content_excerpt="Runtime DDL is blocked; model planning must route DDL via interface/migration only.", + entity_id=int(policy_entity.id), + source_type="governance", + source_ref="backend/app/database.py", + metadata={"blocked_sql_prefixes": DDL_BLOCKED_PREFIXES}, + ) + counters["evidences"] += 1 + + interface_specs = [ + { + "entity_key": "interface:api:/ai-steward/command/dispatch", + "display_name": "POST /ai-steward/command/dispatch", + "payload": {"path": "/ai-steward/command/dispatch", "kind": "api"}, + "source_ref": "backend/app/api/ai_steward.py", + }, + { + "entity_key": "interface:migration:alembic", + "display_name": "Alembic Migration Interface", + "payload": {"path": "alembic", "kind": "migration"}, + "source_ref": "backend/alembic/versions", + }, + ] + interface_ids: list[int] = [] + for item in interface_specs: + interface_entity = await self.upsert_entity( + entity_type="interface", + entity_key=str(item["entity_key"]), + display_name=str(item["display_name"]), + payload=dict(item["payload"]), + source_type="governance", + source_ref=str(item["source_ref"]), + ) + counters["interfaces"] += 1 + interface_ids.append(int(interface_entity.id)) + rel = await self.upsert_relation( + relation_type="policy_enforced_by_interface", + subject_entity_id=int(policy_entity.id), + object_entity_id=int(interface_entity.id), + source_type="governance", + source_ref=str(item["source_ref"]), + ) + counters["relations"] += 1 + await self.add_evidence( + evidence_type="policy_guardrail", + content_excerpt=f"{DDL_GUARDRAIL_POLICY_NAME} -> {item['display_name']}", + relation_id=int(rel.id), + source_type="governance", + source_ref=str(item["source_ref"]), + metadata=dict(item["payload"]), + ) + counters["evidences"] += 1 + + for tier in ("low", "high"): + spec = model_specs.get(tier) if isinstance(model_specs.get(tier), dict) else {} + model_name = str(spec.get("model") or "").strip() or ("GLM-4.7" if tier == "low" else "MiniMax-M2.5") + provider = str(spec.get("provider") or "").strip() + api_env = str(spec.get("api_env") or "").strip() + source = str(spec.get("source") or "env_default").strip() + source_ref = str(spec.get("source_path") or spec.get("rule_name") or "llm_routing_env").strip()[:255] + entity = await self.upsert_entity( + entity_type="ai_model", + entity_key=f"ai_model:{tier}:{model_name.lower()[:160]}", + display_name=model_name[:255], + payload={ + "tier": tier, + "provider": provider, + "model": model_name, + "api_env": api_env, + "source": source, + "routing_rule": str(spec.get("rule_name") or "").strip(), + }, + source_type="model_routing", + source_ref=source_ref or None, + ) + counters["models"] += 1 + rel = await self.upsert_relation( + relation_type="model_must_follow_policy", + subject_entity_id=int(entity.id), + object_entity_id=int(policy_entity.id), + source_type="governance", + source_ref=source_ref or None, + payload={"tier": tier, "policy_code": "ddl_via_interface_only"}, + ) + counters["relations"] += 1 + await self.add_evidence( + evidence_type="policy_guardrail", + content_excerpt=f"{tier} model {model_name} must follow {DDL_GUARDRAIL_POLICY_KEY}", + relation_id=int(rel.id), + source_type="governance", + source_ref=source_ref or None, + metadata={"tier": tier, "provider": provider, "model": model_name}, + ) + counters["evidences"] += 1 + + for interface_id in interface_ids: + rel = await self.upsert_relation( + relation_type="model_executes_via_interface", + subject_entity_id=int(entity.id), + object_entity_id=int(interface_id), + source_type="governance", + source_ref=source_ref or None, + payload={"tier": tier}, + ) + counters["relations"] += 1 + await self.add_evidence( + evidence_type="policy_guardrail", + content_excerpt=f"{tier} model {model_name} executes via controlled interface", + relation_id=int(rel.id), + source_type="governance", + source_ref=source_ref or None, + metadata={"tier": tier}, + ) + counters["evidences"] += 1 + + return counters + + async def sync_model_guardrails_if_due( + self, + *, + interval_hours: int = 6, + ) -> dict[str, Any]: + hours = max(1, int(interval_hours)) + offset_raw = await self.get_ingest_offset(channel="model_guardrail", offset_key="last_model_guardrail_sync_at") + last_sync_at = self._parse_iso_dt(offset_raw) + now = _utc_now_naive() + if last_sync_at is not None and now - last_sync_at < timedelta(hours=hours): + return { + "executed": False, + "reason": "not_due", + "last_sync_at": last_sync_at.isoformat(), + "next_due_at": (last_sync_at + timedelta(hours=hours)).isoformat(), + } + + payload = await self._sync_model_guardrail_entities() + await self.enqueue_ingest_event( + channel="model_guardrail", + event_key=f"model_guardrail:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}", + payload=payload, + status="done", + ) + await self.set_ingest_offset( + channel="model_guardrail", + offset_key="last_model_guardrail_sync_at", + offset_value=_utc_now_naive().isoformat(), + payload=payload, + ) + await self.db.commit() + return { + "executed": True, + "reason": "due", + "payload": payload, + } + + async def sync_model_guardrails_now(self) -> dict[str, Any]: + payload = await self._sync_model_guardrail_entities() + await self.enqueue_ingest_event( + channel="model_guardrail", + event_key=f"model_guardrail:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}", + payload=payload, + status="done", + ) + await self.set_ingest_offset( + channel="model_guardrail", + offset_key="last_model_guardrail_sync_at", + offset_value=_utc_now_naive().isoformat(), + payload=payload, + ) + await self.db.commit() + return payload + async def sync_feature_catalog_if_due( self, *, @@ -1403,8 +1697,18 @@ class ProjectKnowledgeGraphService: stage_errors["feature_catalog"] = str(exc)[:500] stage_results["feature_catalog"] = {"error": stage_errors["feature_catalog"]} + try: + stage_results["model_guardrails"] = await self.sync_model_guardrails_if_due( + interval_hours=feature_scan_hours, + ) + success_count += 1 + except Exception as exc: + await self.db.rollback() + stage_errors["model_guardrails"] = str(exc)[:500] + stage_results["model_guardrails"] = {"error": stage_errors["model_guardrails"]} + ended = _utc_now_naive() - if success_count == 6: + if success_count == 7: final_status = "success" elif success_count == 0: final_status = "failed" @@ -1622,6 +1926,8 @@ class ProjectKnowledgeGraphService: ) counters["evidences"] += 1 + counters["model_guardrails"] = await self._sync_model_guardrail_entities() + await self.enqueue_ingest_event( channel="db_delta", event_key=f"structured_sync:{_utc_now_naive().strftime('%Y%m%d%H%M%S')}", diff --git a/backend/app/services/task_brief_service.py b/backend/app/services/task_brief_service.py index 23a79a00..f1a06d10 100644 --- a/backend/app/services/task_brief_service.py +++ b/backend/app/services/task_brief_service.py @@ -3,6 +3,7 @@ from __future__ import annotations import json +import subprocess from datetime import datetime from pathlib import Path from typing import Any, Optional @@ -57,6 +58,14 @@ class TaskBriefService: def _project_root() -> Path: return Path(__file__).resolve().parents[3] + @staticmethod + def _normalize_template_text(content: str) -> str: + return "\n".join(line.rstrip() for line in str(content or "").strip().splitlines()).strip() + + @classmethod + def _is_default_template_fallback(cls, content: str) -> bool: + return cls._normalize_template_text(content) == cls._normalize_template_text(DEFAULT_TEMPLATE_FALLBACK) + @classmethod def _load_default_template_content(cls) -> str: template_path = cls._project_root() / "docs/architecture/TASK_BRIEF_TEMPLATE.md" @@ -67,6 +76,53 @@ class TaskBriefService: return DEFAULT_TEMPLATE_FALLBACK return DEFAULT_TEMPLATE_FALLBACK + @classmethod + def _resolve_git_ref(cls, ref: str) -> str: + try: + output = subprocess.check_output( + ["git", "rev-parse", "--short", ref], + cwd=str(cls._project_root()), + stderr=subprocess.DEVNULL, + text=True, + timeout=1.5, + ).strip() + return output or "-" + except Exception: + return "-" + + @staticmethod + def _to_markdown_list(items: list[str], *, empty_text: str) -> str: + cleaned = [str(item).strip() for item in items if str(item).strip()] + if not cleaned: + return f"- {empty_text}" + return "\n".join(f"- {item}" for item in cleaned) + + @classmethod + def _render_template_content( + cls, + *, + template_content: str, + generated_at: str, + base_ref: str, + head_ref: str, + changed_files: list[str], + scope_summary: str, + risks: list[str], + ) -> str: + content = str(template_content or "").strip() or cls._load_default_template_content() + replacements = { + "{{GENERATED_AT}}": str(generated_at or "-"), + "{{BASE_REF}}": str(base_ref or "-"), + "{{HEAD_REF}}": str(head_ref or "-"), + "{{CHANGED_FILES}}": cls._to_markdown_list(changed_files, empty_text="(none)"), + "{{SCOPE_SUMMARY}}": str(scope_summary or "-"), + "{{RISKS}}": cls._to_markdown_list(risks, empty_text="待补充风险评估"), + } + rendered = content + for token, value in replacements.items(): + rendered = rendered.replace(token, value) + return rendered.strip() + @classmethod def _build_default_template(cls) -> dict[str, Any]: now_text = now_cn_naive().isoformat() @@ -227,6 +283,26 @@ class TaskBriefService: if not existing_templates: bundle["templates"] = [cls._build_default_template()] changed = True + else: + upgraded_default_template = False + latest_default_content = cls._load_default_template_content() + for item in existing_templates: + if str(item.get("id") or "") != DEFAULT_TEMPLATE_ID: + continue + current_content = str(item.get("content") or "").strip() + if current_content and not cls._is_default_template_fallback(current_content): + break + item["name"] = str(item.get("name") or "").strip() or DEFAULT_TEMPLATE_NAME + item["description"] = str(item.get("description") or "").strip() or DEFAULT_TEMPLATE_DESCRIPTION + item["content"] = latest_default_content + item["updated_at"] = now_cn_naive().isoformat() + if item.get("updated_by") is None: + item["updated_by"] = 0 + upgraded_default_template = True + break + if upgraded_default_template: + bundle["templates"] = existing_templates + changed = True existing_briefs = [item for item in bundle.get("briefs", []) if isinstance(item, dict)] existing_runs = [item for item in bundle.get("runs", []) if isinstance(item, dict)] @@ -355,9 +431,25 @@ class TaskBriefService: brief_id = f"brief_{uuid4().hex[:12]}" run_id = f"run_{uuid4().hex[:12]}" - content = str((template_obj or {}).get("content") or "").strip() file_lines = [str(path).strip() for path in changed_files if str(path).strip()] summary = str(scope_summary or "").strip() or f"共 {len(file_lines)} 个文件发生变更" + risks = ( + [ + "跨文件改动可能引发契约/类型不一致，请补充联动核对。", + "合入前需完成 build/lint/test 并记录结果。", + ] + if file_lines + else ["未提供变更文件列表，风险评估信息不足，请人工补充。"] + ) + content = cls._render_template_content( + template_content=str((template_obj or {}).get("content") or "").strip(), + generated_at=now_text, + base_ref=cls._resolve_git_ref("HEAD~1"), + head_ref=cls._resolve_git_ref("HEAD"), + changed_files=file_lines, + scope_summary=summary, + risks=risks, + ) run = { "id": run_id, diff --git a/backend/tests/test_ai_pmo_kernel_service.py b/backend/tests/test_ai_pmo_kernel_service.py index d76bc47a..ab3ab4a0 100644 --- a/backend/tests/test_ai_pmo_kernel_service.py +++ b/backend/tests/test_ai_pmo_kernel_service.py @@ -6,7 +6,7 @@ from tempfile import NamedTemporaryFile from types import SimpleNamespace from unittest.mock import AsyncMock, patch -from sqlalchemy import and_, select +from sqlalchemy import and_, func, select from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine from app.models import ( @@ -206,6 +206,20 @@ class AIPMOKernelServiceTest(unittest.IsolatedAsyncioTestCase): ) self.assertEqual(intent.name, "ignore") + def test_intent_resolver_prefers_signal_create_for_operation_with_query_words(self) -> None: + intent = IntentResolver.resolve( + text="给AI劝退助手加一个需求，劝退完后第二天问情况如何跟进？", + project_id=1, + ) + self.assertEqual(intent.name, INTENT_SIGNAL_CREATE) + + def test_intent_resolver_negated_write_statement_prefers_query(self) -> None: + intent = IntentResolver.resolve( + text="别乱创建信号，说明你是谁", + project_id=1, + ) + self.assertEqual(intent.name, INTENT_QUERY) + async def test_build_world_model_includes_budget_capacity_and_skill_graph(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) @@ -270,6 +284,86 @@ class AIPMOKernelServiceTest(unittest.IsolatedAsyncioTestCase): self.assertTrue(bool(work_item.review_summary.get("verified"))) self.assertEqual(review_cycle.summary["work_item_id"], int(work_item.id)) + async def test_process_ingress_negated_write_statement_forces_query(self) -> None: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + with patch( + "app.services.ai_pmo_kernel_service.AdaptiveRAGService.answer_question", + AsyncMock(return_value={"answer": "我是 PMFlow 项目协同助手。", "sources": []}), + ): + result = await service.process_ingress( + source_system="pmflow_manual", + source_message_id="msg-negated-write-1", + message_text="别乱创建信号，说明你是谁", + sender="老板", + channel_ref="pmflow", + ) + + work_item = ( + await db.execute( + select(WorkItem).where( + and_( + WorkItem.tenant_id == 1, + WorkItem.source_message_id == "msg-negated-write-1", + ) + ) + ) + ).scalar_one() + signal_count = ( + await db.execute( + select(func.count()).select_from(Signal).where(Signal.tenant_id == 1) + ) + ).scalar_one() + + self.assertEqual(result["status"], "answered") + self.assertEqual(work_item.intent, INTENT_QUERY) + self.assertEqual(int(signal_count or 0), 0) + + async def test_process_query_message_returns_failed_safe_when_rag_raises(self) -> None: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + with patch( + "app.services.ai_pmo_kernel_service.AdaptiveRAGService.answer_question", + AsyncMock(side_effect=RuntimeError("模型调用失败：模型调用失败：function=adaptive_rag_answer_normal")), + ): + result = await service.process_ingress( + source_system="pmflow_manual", + source_message_id="msg-query-failed-safe-1", + message_text="项目ID: 1 现在状态如何？", + sender="老板", + channel_ref="pmflow", + project_hint_id=1, + ) + + self.assertTrue(result["handled"]) + self.assertEqual(result["status"], "captured") + self.assertEqual(result["classification"], "question") + reply_text = str(result.get("direct_reply_text") or "") + self.assertIn("模型调用失败：function=adaptive_rag_answer_normal", reply_text) + self.assertNotIn("模型调用失败：模型调用失败：", reply_text) + + async def test_resolve_intent_prefers_rule_for_query_without_llm_call(self) -> None: + prev = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "1" + try: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + with patch.object(service, "_resolve_intent_with_llm", AsyncMock(return_value=None)) as llm_mock: + intent, source = await service._resolve_intent( + message_text="你是谁？", + session=None, + world_model={"resolved_project_id": 1}, + ) + + self.assertEqual(intent.name, INTENT_QUERY) + self.assertEqual(source, "rule_preferred") + llm_mock.assert_not_awaited() + finally: + if prev is None: + os.environ.pop("AI_PMO_INTENT_LLM_ENABLED", None) + else: + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = prev + async def test_duplicate_message_is_filtered(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) diff --git a/backend/tests/test_ai_steward_capture_feishu_api.py b/backend/tests/test_ai_steward_capture_feishu_api.py new file mode 100644 index 00000000..975eaa0e --- /dev/null +++ b/backend/tests/test_ai_steward_capture_feishu_api.py @@ -0,0 +1,88 @@ +import os +import tempfile +import unittest +from unittest.mock import AsyncMock, patch + +from sqlalchemy import and_, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.api.ai_steward import FeishuCaptureRequest, capture_feishu_signal +from app.models import Base, FeishuReplyOutbox, User + + +class CaptureFeishuSignalApiTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + tmp = tempfile.NamedTemporaryFile(prefix="pmflow-ai-steward-capture-", suffix=".db", delete=False) + tmp.close() + self.db_path = tmp.name + self.engine = create_async_engine(f"sqlite+aiosqlite:///{self.db_path}", future=True) + self.session_factory = async_sessionmaker(self.engine, expire_on_commit=False, class_=AsyncSession) + async with self.engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + async def asyncTearDown(self) -> None: + await self.engine.dispose() + if getattr(self, "db_path", None): + try: + os.remove(self.db_path) + except FileNotFoundError: + pass + + async def test_capture_feishu_signal_enqueues_outbox_when_direct_reply_required(self) -> None: + async with self.session_factory() as db: + user = User( + tenant_id=1, + username="capture_admin", + email="capture_admin@example.com", + password_hash="hash", + name="捕获管理员", + role="ADMIN", + status="active", + is_active=True, + ) + db.add(user) + await db.commit() + await db.refresh(user) + + with patch( + "app.api.ai_steward.ProactiveSignalCaptureService.capture_from_feishu", + AsyncMock( + return_value={ + "handled": True, + "status": "captured", + "classification": "signal", + "ai_trace_id": "ai-pmo-123", + "direct_reply_required": True, + "direct_reply_text": "已收到，正在执行。", + } + ), + ): + payload = await capture_feishu_signal( + request=FeishuCaptureRequest( + group_chat_id="oc_test_001", + message_content="给 AI 劝退助手加一个复杂需求", + sender="老板", + feishu_message_id="om_test_001", + ), + current_user=user, + db=db, + tenant_id=1, + ) + + self.assertTrue(payload["reply_outbox_enqueued"]) + self.assertIsInstance(payload["reply_outbox_id"], int) + row = ( + await db.execute( + select(FeishuReplyOutbox).where( + and_( + FeishuReplyOutbox.tenant_id == 1, + FeishuReplyOutbox.source_message_id == "om_test_001", + ) + ) + ) + ).scalar_one_or_none() + self.assertIsNotNone(row) + assert row is not None + self.assertEqual(row.chat_id, "oc_test_001") + self.assertEqual(row.message_text, "已收到，正在执行。") + diff --git a/backend/tests/test_feishu_routing_receipts.py b/backend/tests/test_feishu_routing_receipts.py index 9babf0b6..7ecbec08 100644 --- a/backend/tests/test_feishu_routing_receipts.py +++ b/backend/tests/test_feishu_routing_receipts.py @@ -77,6 +77,11 @@ class FeishuRoutingReceiptsTest(unittest.TestCase): "系统处理中出现异常，本次未执行。请稍后重试，或补充更明确的目标、对象和动作。", ) + def test_build_capability_reply_returns_stable_text(self) -> None: + reply = FeishuIntegrationService._build_capability_reply("别乱创建信号，说明你是谁") + self.assertIsNotNone(reply) + self.assertIn("PMFlow 的项目协同助手", str(reply)) + def test_handle_message_receive_skips_reply_for_duplicate_event(self) -> None: service = FeishuIntegrationService(db=None) # type: ignore[arg-type] event = { @@ -121,6 +126,53 @@ class FeishuRoutingReceiptsTest(unittest.TestCase): reply_mock.assert_not_awaited() + def test_handle_message_receive_capability_reply_overrides_execute_direct_reply(self) -> None: + service = FeishuIntegrationService(db=None) # type: ignore[arg-type] + event = { + "message": { + "message_id": "om_capability_guard", + "message_type": "text", + "chat_id": "oc_test", + "chat_type": "group", + "content": '{"text":"@PMFlow 别乱创建信号，说明你是谁"}', + }, + "chat": { + "chat_id": "oc_test", + "chat_type": "group", + }, + "sender": { + "sender_type": "user", + "sender_id": {"open_id": "ou_test"}, + "sender_name": "林宇航", + }, + "type": "im.message.receive_v1", + } + + with ( + patch.object(service, "_record_ingress_trace", AsyncMock()), + patch.object(service, "_record_message_trace", AsyncMock()), + patch.object(service, "_resolve_sender_display_name", AsyncMock(return_value="林宇航")), + patch.object(service, "_safe_reply_text", AsyncMock()) as reply_mock, + patch( + "app.services.feishu_integration.ProactiveSignalCaptureService.capture_from_feishu", + AsyncMock( + return_value={ + "status": "captured", + "classification": "signal", + "direct_reply_required": True, + "direct_reply_text": "我在等 Interactive Card 的内容。", + "results": [{"route_status": "queued_for_decision"}], + } + ), + ), + ): + asyncio.run(service._handle_message_receive(event)) + + reply_mock.assert_awaited_once() + sent_text = str(reply_mock.await_args.kwargs.get("text") or "") + self.assertIn("PMFlow 的项目协同助手", sent_text) + self.assertNotIn("Interactive Card", sent_text) + def test_sync_feishu_recent_cache_backfills_missing_rag_turn(self) -> None: service = ProactiveSignalCaptureService(db=None) # type: ignore[arg-type] diff --git a/backend/tests/test_llm_structured_extraction.py b/backend/tests/test_llm_structured_extraction.py index d1e7cf25..5d98c23b 100644 --- a/backend/tests/test_llm_structured_extraction.py +++ b/backend/tests/test_llm_structured_extraction.py @@ -1,5 +1,6 @@ +import os import unittest -from unittest.mock import AsyncMock +from unittest.mock import AsyncMock, MagicMock, patch from app.services.llm import ChatCompletionResult, LLMService, StructuredExtractionResult @@ -203,5 +204,151 @@ class LLMStructuredCacheTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(len(calls), 1) +class LLMQueryFailureBehaviorTest(unittest.IsolatedAsyncioTestCase): + async def test_generate_strict_query_retries_high_tier_even_when_rule_not_low(self) -> None: + service = LLMService() + service._resolve_engine_for_call = AsyncMock( + return_value={ + "provider": "glm", + "model": "GLM-4.7", + "api_key": "fake", + "base_url": "http://unit.test", + } + ) + service._resolve_active_prompt_profile = AsyncMock(return_value={"role_name": "", "system_prompt": ""}) + service._is_low_tier_function = MagicMock(return_value=False) + service._resolve_high_tier_engine_for_escalation = AsyncMock( + return_value={ + "provider": "openai", + "model": "GLM-5", + "api_key": "fake-openai", + "base_url": "http://unit.test/openai", + } + ) + service._chat_completion = AsyncMock( + side_effect=[ValueError("provider_circuit_open:glm:GLM-4.7"), "高阶模型正常响应"] + ) + service._try_safety_engine = AsyncMock(return_value=("", None)) + service._log_ai_call = AsyncMock() + + text = await service.generate( + "query prompt", + db=object(), + tenant_id=1, + function_name="adaptive_rag_answer_normal", + use_prompt_profile=False, + ) + + self.assertEqual(text, "高阶模型正常响应") + self.assertEqual(service._chat_completion.await_count, 2) + + async def test_generate_raises_runtime_error_for_query_answer_fail_fast(self) -> None: + service = LLMService() + service._resolve_engine_for_call = AsyncMock( + return_value={ + "provider": "glm", + "model": "GLM-4.7", + "api_key": "fake", + "base_url": "http://unit.test", + } + ) + service._resolve_active_prompt_profile = AsyncMock(return_value={"role_name": "", "system_prompt": ""}) + service._chat_completion = AsyncMock(side_effect=ValueError("primary_engine_empty_response")) + service._resolve_high_tier_engine_for_escalation = AsyncMock(return_value=None) + service._try_safety_engine = AsyncMock(return_value=("", None)) + service._log_ai_call = AsyncMock() + + with self.assertRaises(RuntimeError) as ctx: + await service.generate( + "query prompt", + db=object(), + tenant_id=1, + function_name="adaptive_rag_answer_normal", + use_prompt_profile=False, + ) + + self.assertIn("模型调用失败：", str(ctx.exception)) + service._try_safety_engine.assert_not_awaited() + + async def test_generate_keeps_stub_behavior_for_non_query_function(self) -> None: + service = LLMService() + service._resolve_engine_for_call = AsyncMock( + return_value={ + "provider": "glm", + "model": "GLM-4.7", + "api_key": "fake", + "base_url": "http://unit.test", + } + ) + service._resolve_active_prompt_profile = AsyncMock(return_value={"role_name": "", "system_prompt": ""}) + service._chat_completion = AsyncMock(side_effect=ValueError("primary_engine_empty_response")) + service._log_ai_call = AsyncMock() + + text = await service.generate( + "reply prompt", + db=object(), + tenant_id=1, + function_name="feishu_humanized_reply", + use_prompt_profile=False, + ) + + self.assertIn("模型服务异常", text) + + +class LLMRoutingSelectionTest(unittest.TestCase): + def test_inject_function_guardrails_for_dual_model_functions(self) -> None: + injected = LLMService._inject_function_guardrails( + system_prompt="", + function_name="classify_intent_complex", + ) + self.assertIn("ddl_api_required", injected) + self.assertIn("DDL", injected) + + def test_merge_prompt_with_profile_appends_guardrail_for_openclaw_dispatch(self) -> None: + merged = LLMService._merge_prompt_with_profile( + prompt='{"instruction":"db ddl"}', + prompt_profile={"role_name": "planner", "system_prompt": "你是规划器。"}, + function_name="openclaw_command_dispatch", + ) + self.assertIn("你是规划器。", merged) + self.assertIn("ddl_api_required", merged) + + def test_resolve_tier_engine_keeps_configured_openai_base_url(self) -> None: + service = LLMService() + env_updates = { + "OPENCLAW_ROUTING_ENABLED": "1", + "OPENAI_BASE_URL": "https://api.edgefn.net/v1", + "LLM_ROUTING_HIGH_PROVIDER": "openai", + } + with patch.dict(os.environ, env_updates, clear=False): + engine = service._resolve_tier_engine( + tier="high", + engines=[], + active_engine={}, + ) + + self.assertIsNotNone(engine) + assert engine is not None + self.assertEqual(str(engine.get("base_url")), "https://api.edgefn.net/v1") + + def test_should_retry_empty_chat_response_for_reasoning_length(self) -> None: + should_retry = LLMService._should_retry_empty_chat_response( + structured_output=False, + message={"content": "", "reasoning_content": "先分析上下文"}, + choice={"finish_reason": "length"}, + current_max_tokens=900, + ) + self.assertTrue(should_retry) + + def test_should_not_retry_empty_chat_response_when_not_length(self) -> None: + should_retry = LLMService._should_retry_empty_chat_response( + structured_output=False, + message={"content": "", "reasoning_content": "先分析上下文"}, + choice={"finish_reason": "stop"}, + current_max_tokens=900, + ) + self.assertFalse(should_retry) + + if __name__ == "__main__": unittest.main() diff --git a/backend/tests/test_project_knowledge_graph_guardrails.py b/backend/tests/test_project_knowledge_graph_guardrails.py new file mode 100644 index 00000000..1bb968cd --- /dev/null +++ b/backend/tests/test_project_knowledge_graph_guardrails.py @@ -0,0 +1,117 @@ +import json +import os +import tempfile +import unittest + +from sqlalchemy import and_, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.models import Base, KGEntity, KGRelation +from app.services.project_knowledge_graph import DDL_GUARDRAIL_POLICY_KEY, ProjectKnowledgeGraphService + + +class ProjectKnowledgeGraphGuardrailSyncTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + self._old_rules_path = os.environ.get("OPENCLAW_ROUTING_RULES_PATH") + + tmp_db = tempfile.NamedTemporaryFile(prefix="pmflow-kg-guardrail-", suffix=".db", delete=False) + tmp_db.close() + self.db_path = tmp_db.name + self.engine = create_async_engine(f"sqlite+aiosqlite:///{self.db_path}", future=True) + self.session_factory = async_sessionmaker(self.engine, expire_on_commit=False, class_=AsyncSession) + async with self.engine.begin() as conn: + await conn.run_sync(Base.metadata.create_all) + + tmp_rules = tempfile.NamedTemporaryFile(prefix="pmflow-routing-rules-", suffix=".json", delete=False) + tmp_rules.write( + json.dumps( + { + "rules": [ + { + "name": "low-intent-rule", + "enabled": True, + "priority": 300, + "tier": "low", + "provider": "glm", + "model": "GLM-4.7", + "api_env": "GLM_API_KEY", + }, + { + "name": "high-intent-rule", + "enabled": True, + "priority": 280, + "tier": "high", + "provider": "openai", + "model": "GLM-5", + "api_env": "OPENAI_API_KEY", + }, + ] + } + ).encode("utf-8") + ) + tmp_rules.flush() + tmp_rules.close() + self.rules_path = tmp_rules.name + os.environ["OPENCLAW_ROUTING_RULES_PATH"] = self.rules_path + + async def asyncTearDown(self) -> None: + await self.engine.dispose() + for path in (getattr(self, "db_path", None), getattr(self, "rules_path", None)): + if path: + try: + os.remove(path) + except FileNotFoundError: + pass + + if self._old_rules_path is None: + os.environ.pop("OPENCLAW_ROUTING_RULES_PATH", None) + else: + os.environ["OPENCLAW_ROUTING_RULES_PATH"] = self._old_rules_path + + async def test_sync_model_guardrails_now_builds_dual_model_policy_graph(self) -> None: + async with self.session_factory() as db: + service = ProjectKnowledgeGraphService(db, tenant_id=1) + payload = await service.sync_model_guardrails_now() + + self.assertEqual(payload["models"], 2) + self.assertGreaterEqual(payload["relations"], 4) + + policy_entity = ( + await db.execute( + select(KGEntity).where( + and_( + KGEntity.tenant_id == 1, + KGEntity.entity_type == "governance_policy", + KGEntity.entity_key == DDL_GUARDRAIL_POLICY_KEY, + ) + ) + ) + ).scalar_one_or_none() + self.assertIsNotNone(policy_entity) + + model_entities = ( + await db.execute( + select(KGEntity).where( + and_( + KGEntity.tenant_id == 1, + KGEntity.entity_type == "ai_model", + ) + ) + ) + ).scalars().all() + self.assertEqual(len(model_entities), 2) + model_names = {str(row.display_name or "") for row in model_entities} + self.assertEqual(model_names, {"GLM-4.7", "GLM-5"}) + + policy_relations = ( + await db.execute( + select(KGRelation).where( + and_( + KGRelation.tenant_id == 1, + KGRelation.relation_type == "model_must_follow_policy", + KGRelation.object_entity_id == int(policy_entity.id), + ) + ) + ) + ).scalars().all() + self.assertEqual(len(policy_relations), 2) diff --git a/backend/tests/test_task_briefs_api.py b/backend/tests/test_task_briefs_api.py index 7c1ac86c..6532ab15 100644 --- a/backend/tests/test_task_briefs_api.py +++ b/backend/tests/test_task_briefs_api.py @@ -14,7 +14,8 @@ from app.api.task_briefs import ( list_task_brief_runs, list_task_brief_templates, ) -from app.models import Base, User +from app.models import Base, Integration, User +from app.services.task_brief_service import DEFAULT_TEMPLATE_FALLBACK class TaskBriefApiTest(unittest.IsolatedAsyncioTestCase): @@ -95,3 +96,67 @@ class TaskBriefApiTest(unittest.IsolatedAsyncioTestCase): self.assertEqual(sync_board["code"], 0) self.assertIn("all_synced", sync_board["data"]) self.assertIsInstance(sync_board["data"].get("files"), list) + + async def test_default_template_upgrade_and_placeholder_render(self) -> None: + async with self.session_factory() as db: + user = User( + tenant_id=1, + username="brief_admin_upgrade", + email="brief_admin_upgrade@example.com", + password_hash="hash", + name="模板管理员", + role="ADMIN", + status="active", + is_active=True, + ) + db.add(user) + await db.commit() + await db.refresh(user) + + db.add( + Integration( + tenant_id=1, + type="task_briefs", + is_active=True, + config={ + "templates": [ + { + "id": "tpl_default_signal_planner", + "name": "Signal Planner 默认模板", + "description": "系统初始化模板（可按需修改）", + "content": DEFAULT_TEMPLATE_FALLBACK, + "updated_at": "2026-01-01T00:00:00", + "updated_by": 0, + "created_at": "2026-01-01T00:00:00", + } + ], + "runs": [], + "briefs": [], + }, + ) + ) + await db.commit() + + templates = await list_task_brief_templates(db=db, current_user=user) + self.assertEqual(templates["code"], 0) + default_tpl = next((item for item in templates["data"]["items"] if item.get("id") == "tpl_default_signal_planner"), None) + self.assertIsNotNone(default_tpl) + self.assertIn("## Verification", str(default_tpl["content"])) + self.assertNotEqual(str(default_tpl["content"]).strip(), DEFAULT_TEMPLATE_FALLBACK.strip()) + + generated = await generate_task_brief( + body=GenerateTaskBriefRequest( + title="Default Template Render", + changed_files=["backend/app/services/task_brief_service.py"], + scope_summary="验证默认模板占位符渲染", + ), + db=db, + current_user=user, + ) + self.assertEqual(generated["code"], 0) + content = str(generated["data"]["brief"]["content"] or "") + self.assertIn("backend/app/services/task_brief_service.py", content) + self.assertIn("验证默认模板占位符渲染", content) + self.assertNotIn("{{CHANGED_FILES}}", content) + self.assertNotIn("{{SCOPE_SUMMARY}}", content) + self.assertNotIn("{{GENERATED_AT}}", content) diff --git a/frontend/src/components/layout/Sidebar.tsx b/frontend/src/components/layout/Sidebar.tsx index 4f7e04ec..1a9802c2 100644 --- a/frontend/src/components/layout/Sidebar.tsx +++ b/frontend/src/components/layout/Sidebar.tsx @@ -1,4 +1,4 @@ -import { NavLink, useLocation } from "react-router-dom"; +import { NavLink } from "react-router-dom"; import { LayoutDashboard, FolderKanban, @@ -9,11 +9,6 @@ import { Users, Sparkles, Sliders, - Files, - LayoutTemplate, - History, - ShieldCheck, - Settings2, X, } from "lucide-react"; import { cn } from "@/lib/utils"; @@ -31,28 +26,18 @@ const baseNavItems = [ { to: "/skills", label: "能力管理", icon: BarChart3 }, ]; -const configToolItems = [ - { to: "/ai-config/task-brief-center", label: "增量任务包中心", icon: Files }, - { to: "/ai-config/task-brief-templates", label: "任务包模板管理", icon: LayoutTemplate }, - { to: "/ai-config/task-brief-runs", label: "任务包生成记录", icon: History }, - { to: "/ai-config/contract-sync-board", label: "契约同步看板", icon: ShieldCheck }, - { to: "/ai-config/model-routing-config", label: "模型路由配置", icon: Settings2 }, -]; - interface SidebarProps { open: boolean; onClose: () => void; } export function Sidebar({ open, onClose }: SidebarProps) { - const location = useLocation(); const { theme, toggleTheme } = useTheme(); const user = useAuthStore((s) => s.user); const isManager = isManagerRole(user?.role); const navItems = isManager ? [...baseNavItems, { to: "/users", label: "用户管理", icon: Users }] : baseNavItems; - const showConfigTools = location.pathname.startsWith("/ai-config"); return (