From cf638e0a620daa0f36347fd9a89d6e91aefaaef7 Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Tue, 10 Mar 2026 19:09:59 +0800 Subject: [PATCH] =?UTF-8?q?=E6=8E=92=E6=9F=A5=E9=A3=9E=E4=B9=A6=E6=9C=BA?= =?UTF-8?q?=E5=99=A8=E4=BA=BA=E9=87=8D=E5=A4=8D=E5=9B=9E=E5=A4=8D=E9=97=AE?= =?UTF-8?q?=E9=A2=98=E3=80=8D=20note:=20wait=20must=20be=2018?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- AGENTS.md | 69 ++- CLAUDE.md | 59 +++ backend/app/api/projects.py | 3 + backend/app/api/skill_tags.py | 4 + backend/app/api/sstc.py | 3 + backend/app/api/users.py | 24 + backend/app/schemas/world_model.py | 136 ++++++ backend/app/services/ai_pmo_kernel_service.py | 418 +++++++++++++++--- backend/app/services/context_cache.py | 18 + backend/app/services/skill_service.py | 6 + 10 files changed, 673 insertions(+), 67 deletions(-) create mode 100644 backend/app/schemas/world_model.py diff --git a/AGENTS.md b/AGENTS.md index 3faff990..e42c471b 100644 --- a/AGENTS.md +++ b/AGENTS.md @@ -16,7 +16,12 @@ 3. **数据库规范**：使用 SQLAlchemy 2.0，所有模型继承 Base 基类，必须包含 `tenant_id` 字段，所有查询必须自动过滤 `tenant_id`（使用依赖注入实现）。 -4. **AI 能力规范**：所有 AI 能力必须封装在独立的 service 层，使用 LangChain 实现，Prompt 放在 `./backend/app/config/ai_prompts.py` 独立管理，禁止硬编码。 +4. **AI 能力规范**： + - 所有 AI 能力必须封装在独立的 service 层 + - AI 能力通过 OpenClaw 路由层调用 + - 路由规则参考 `backend/.openclaw/pmflow-routing-rules.json` + - Prompt 放在 `./backend/app/config/ai_prompts.py` 独立管理 + - 禁止直接硬编码模型名称 5. **AI 调用规范**：所有 AI 调用必须： - 记录日志到 `ai_call_logs` 表 @@ -98,12 +103,37 @@ frontend/ # React 前端 - 飞书回调字段可能缺失文件名/大小，仅保证 `attachment_id/key/type` 可用 - `agents.defaults.timeoutSeconds` 在 OpenClaw 网关行为未验证，当前以 `LLM_REQUEST_TIMEOUT` 环境变量为准 -- World Model 每次全量构建，高负载场景下存在性能风险，暂未缓存，改动前需评估影响 +- World Model 静态层已接入 Redis 缓存（key: `wm:static:{tenant_id}:{project_id}`，TTL 600s） +- 写操作后主动 invalidate，动态层不缓存 ## 12. 前端组件约束 - `ui/` 目录下组件禁止修改，仅使用 -- `sstc/` 目录下组件已模块化拆分，修改前必须执行 `FIND_USAGE` +- `frontend/src/components/sstc/` 目录已模块化拆分，新增文件与职责如下： + - `SSTCCharterStagePanel.tsx`：宪章阶段主面板（内容编辑、版本/状态展示） + - `SSTCExecutionBoardPanel.tsx`：执行看板阶段主面板（执行视图聚合） + - `SSTCReleaseCreateDialog.tsx`：发布创建弹窗 + - `SSTCReleaseStagePanel.tsx`：发布阶段主面板（发布列表与校验项） + - `SSTCReviewCreateDialog.tsx`：复盘创建弹窗 + - `SSTCReviewStagePanel.tsx`：复盘阶段主面板（复盘内容与改进项） + - `SSTCSignalStagePanel.tsx`：信号阶段主面板（信号录入/状态流转） + - `SSTCTaskCreateDialog.tsx`：任务创建弹窗 + - `SSTCTaskStagePanel.tsx`：任务阶段主面板（任务列表、筛选、状态流转） + - `sstcUtils.ts`：SSTC 跨阶段工具函数与转换逻辑 + - `useSSTCWorkspaceOrchestration.ts`：SSTC 工作区编排 Hook（状态聚合、事件编排、数据装配） +- 修改前必须执行 `FIND_USAGE` 的组件/模块清单（强制）： + - `SSTCWorkspace.tsx` + - `useSSTCWorkspaceOrchestration.ts` + - `sstcUtils.ts` + - `SSTCCharterStagePanel.tsx` + - `SSTCSignalStagePanel.tsx` + - `SSTCTaskStagePanel.tsx` + - `SSTCExecutionBoardPanel.tsx` + - `SSTCReleaseStagePanel.tsx` + - `SSTCReviewStagePanel.tsx` + - `SSTCReleaseCreateDialog.tsx` + - `SSTCReviewCreateDialog.tsx` + - `SSTCTaskCreateDialog.tsx` - 新增业务组件放入对应功能目录，禁止在 `pages/` 内写内联组件 - 样式只用 Tailwind utility class，禁止内联 `style` @@ -112,3 +142,36 @@ frontend/ # React 前端 - 后端改动：`python -m py_compile` 通过 + 相关单测通过 - 前端改动：`npm run build` 通过 + `tsc --noEmit` 零错误 - 新增接口：必须在 `backend/tests/` 补充对应单测 + +## 14. 文档同步规则（强制） + +每次任务完成后，若以下内容发生变化，必须同步更新 `CLAUDE.md` 对应章节： + +- 新增/删除/重命名文件或目录 + → 更新第3-5节目录结构 +- 新增/修改服务边界或禁止操作 + → 更新第8节服务边界约束 +- 新增已知 `BOUNDARY` 或 `ASSUMPTION` + → 更新第11节已知 `BOUNDARY` +- 新增/修改 API 接口 + → 更新第9节参考模式 + +文档更新必须在 `[Task:Done]` 之前完成。 +未更新文档的任务视为未完成。 + +## 15. 反迎合约束（强制） + +若检测到 Host 已预设方案倾向： +- 必须先输出 `[CRITIQUE: 当前方案的根本性风险]` +- 再输出执行路径 + +若 Host 方案是次优解： +- 必须声明 `[ALT: 更优方案]` + 理由 +- Host 确认坚持后方可执行 + +## 16. Context 管理约定 + +- 单任务涉及文件超过5个 → 拆分为独立子任务 +- 单任务预估改动超过200行 → 拆分执行 +- 发生 context 不足时 → 停止并输出已完成部分 + 声明 `[CONTEXT: LIMIT]`，不要截断输出 diff --git a/CLAUDE.md b/CLAUDE.md index ab4a195d..3de4f19c 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -62,6 +62,41 @@ frontend/ └── vite.config.ts ``` +### 5.1 `frontend/src/components/sstc/` 目录说明（模块化拆分后） +- `SSTCWorkspace.tsx`：SSTC 工作区主容器（组装各阶段组件） +- `useSSTCWorkspaceOrchestration.ts`：工作区编排 Hook（状态聚合、请求编排、事件路由） +- `sstcUtils.ts`：SSTC 工具函数（数据转换、状态判断、通用 helper） +- `SSTCSignalStagePanel.tsx`：信号阶段面板 +- `SSTCCharterStagePanel.tsx`：宪章阶段面板 +- `SSTCTaskStagePanel.tsx`：任务阶段面板 +- `SSTCExecutionBoardPanel.tsx`：执行看板阶段面板 +- `SSTCReleaseStagePanel.tsx`：发布阶段面板 +- `SSTCReviewStagePanel.tsx`：复盘阶段面板 +- `SSTCTaskCreateDialog.tsx`：任务创建弹窗 +- `SSTCReleaseCreateDialog.tsx`：发布创建弹窗 +- `SSTCReviewCreateDialog.tsx`：复盘创建弹窗 +- `AIVisibilityPanel.tsx`：AI 可视化信息面板 +- `SSTCGlobalHeader.tsx`：SSTC 全局头部 +- `SSTCStageNav.tsx`：阶段导航 +- `SSTCChainView.tsx`：链路视图 +- `SSTCRiskFloatPanel.tsx`：风险浮层 +- `SignalContentHoverCard.tsx`：信号内容悬浮卡片 +- `charterModules.ts`：宪章模块配置与映射 + +### 5.2 修改前必须执行 `FIND_USAGE` 的 SSTC 组件清单 +- `SSTCWorkspace.tsx` +- `useSSTCWorkspaceOrchestration.ts` +- `sstcUtils.ts` +- `SSTCCharterStagePanel.tsx` +- `SSTCSignalStagePanel.tsx` +- `SSTCTaskStagePanel.tsx` +- `SSTCExecutionBoardPanel.tsx` +- `SSTCReleaseStagePanel.tsx` +- `SSTCReviewStagePanel.tsx` +- `SSTCReleaseCreateDialog.tsx` +- `SSTCReviewCreateDialog.tsx` +- `SSTCTaskCreateDialog.tsx` + ## 6. 运行命令（高频） ### 根目录 ```bash @@ -87,3 +122,27 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - 规则变更：只改 `AGENTS.md` - 结构变更：同步改 `CLAUDE.md` - API/契约变更：后端 `schemas` + 前端 `types` 同步更新 + +## 8. 服务边界约束（禁止越界） +- 数据库写操作必须通过 Alembic migration，禁止直接 `ALTER TABLE` +- 新增接口必须遵循响应包络：`{code, message, data}`，不得裸返回 +- 禁止在 API 层写业务逻辑，业务逻辑归属 `services/` +- 禁止在前端组件内直接调用 `fetch`，必须通过 `frontend/src/api/` 封装层 +- 前端状态管理：服务端数据用 React Query，本地 UI 状态用 `useState` + +## 9. 新增代码的参考模式 +- 新增后端接口：参考 `backend/app/api/sstc.py` 的结构 +- 新增前端 API 调用：参考 `frontend/src/api/sstc.ts` +- 新增数据库表：参考 `alembic/versions/` 最新 migration +- 新增飞书事件处理：参考 `backend/app/services/feishu_integration.py` + +## 10. 幂等与事务约束 +- 所有入站事件必须经过 `ingest_events` 幂等去重 +- 飞书回复必须经过 `feishu_reply_outbox`，禁止同步直接发送 +- 跨表写操作必须在同一个数据库事务内完成 + +## 11. 已知 BOUNDARY（不要幻构行为） +- 飞书回调字段可能缺失文件名/大小，仅保证 `attachment_id/key/type` 可用 +- `agents.defaults.timeoutSeconds` 在 OpenClaw 网关行为未验证，当前以 `LLM_REQUEST_TIMEOUT` 环境变量为准 +- World Model 静态层已接入 Redis 缓存（key: `wm:static:{tenant_id}:{project_id}`，TTL 600s） +- 写操作后主动 invalidate，动态层不缓存 diff --git a/backend/app/api/projects.py b/backend/app/api/projects.py index adc542e4..d7052759 100644 --- a/backend/app/api/projects.py +++ b/backend/app/api/projects.py @@ -13,6 +13,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from ..database import get_db from ..dependencies import get_current_user, get_tenant_id, require_roles from ..models import Project, ProjectSignalLink, Signal, Task, User +from ..services.ai_pmo_kernel_service import AIPMOKernelService from ..services.context_cache import context_cache_service from ..utils.project_members import ensure_project_default_members from ..utils.sequence import ensure_project_id_sequence @@ -227,6 +228,7 @@ async def create_project( await db.commit() await db.refresh(project) await context_cache_service.invalidate_system_context(tenant_id=tenant_id) + await AIPMOKernelService.invalidate_world_model_cache(tenant_id=tenant_id, project_id=int(project.id)) return _serialize_project_row(project) @@ -256,6 +258,7 @@ async def update_project( await db.commit() await db.refresh(project) await context_cache_service.invalidate_system_context(tenant_id=tenant_id) + await AIPMOKernelService.invalidate_world_model_cache(tenant_id=tenant_id, project_id=int(project.id)) return _serialize_project_row(project) diff --git a/backend/app/api/skill_tags.py b/backend/app/api/skill_tags.py index 8a156c20..5250e3e4 100644 --- a/backend/app/api/skill_tags.py +++ b/backend/app/api/skill_tags.py @@ -13,6 +13,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from ..database import get_db from ..dependencies import get_current_user, get_tenant_id, require_roles from ..models import SkillTag, User, UserSkillBinding +from ..services.ai_pmo_kernel_service import AIPMOKernelService from ..services import skill_service from ..services.auth import has_any_role @@ -235,6 +236,7 @@ async def bind_user_skill( ) db.add(row) await db.commit() + await AIPMOKernelService.invalidate_world_model_cache_for_user(db=db, tenant_id=tenant_id, user_id=user_id) return {"message": "Skill bound"} @@ -285,6 +287,7 @@ async def update_user_skill( row.efficiency_score = body.efficiency_score row.updated_at = datetime.utcnow() await db.commit() + await AIPMOKernelService.invalidate_world_model_cache_for_user(db=db, tenant_id=tenant_id, user_id=user_id) return {"message": "Skill updated"} @@ -317,4 +320,5 @@ async def unbind_user_skill( await db.delete(row) await db.commit() + await AIPMOKernelService.invalidate_world_model_cache_for_user(db=db, tenant_id=tenant_id, user_id=user_id) return {"message": "Skill unbound"} diff --git a/backend/app/api/sstc.py b/backend/app/api/sstc.py index 79e233e2..e664b917 100644 --- a/backend/app/api/sstc.py +++ b/backend/app/api/sstc.py @@ -58,6 +58,7 @@ from ..schemas.sstc import ( ) from ..services.llm import llm_service from ..services.ai_service import AIService +from ..services.ai_pmo_kernel_service import AIPMOKernelService from ..services.operation_log_service import log_operation from ..services.auth import has_any_role from ..services.skill_service import recommend_personnel @@ -2382,6 +2383,7 @@ async def update_charter( ) await db.commit() await db.refresh(new_node) + await AIPMOKernelService.invalidate_world_model_cache(tenant_id=tenant_id, project_id=pid) _sync_pmflow_rag_async( tenant_id=tenant_id, record_type="charter", @@ -2424,6 +2426,7 @@ async def update_charter( ) await db.commit() await db.refresh(charter) + await AIPMOKernelService.invalidate_world_model_cache(tenant_id=tenant_id, project_id=pid) _sync_pmflow_rag_async( tenant_id=tenant_id, record_type="charter", diff --git a/backend/app/api/users.py b/backend/app/api/users.py index 2626e886..1d272079 100644 --- a/backend/app/api/users.py +++ b/backend/app/api/users.py @@ -39,6 +39,7 @@ from ..schemas import ( UserUpdate, ) from ..services.auth import AuthService +from ..services.ai_pmo_kernel_service import AIPMOKernelService from ..services.operation_log_service import log_operation from ..services import skill_service @@ -211,6 +212,15 @@ async def _replace_user_projects( user_id: int, project_ids: list[int], ) -> None: + existing_rows = await db.execute( + select(UserProject.project_id).where( + and_( + UserProject.tenant_id == tenant_id, + UserProject.user_id == user_id, + ) + ) + ) + existing_project_ids = {int(pid) for pid in existing_rows.scalars().all() if pid is not None} await db.execute( delete(UserProject).where( and_( @@ -239,6 +249,10 @@ async def _replace_user_projects( cache_payload["invalidate_project_list"] = True cache_payload["user_ids"].add(int(user_id)) + affected_project_ids = sorted(existing_project_ids.union({int(pid) for pid in unique_project_ids})) + for project_id in affected_project_ids: + await AIPMOKernelService.invalidate_world_model_cache(tenant_id=tenant_id, project_id=project_id) + async def _create_user_with_defaults( db: AsyncSession, @@ -750,6 +764,11 @@ async def delete_user( ) before_data = user.to_dict() + await AIPMOKernelService.invalidate_world_model_cache_for_user( + db=db, + tenant_id=tenant_id, + user_id=user_id, + ) await db.execute( delete(UserSkillBinding).where( and_( @@ -842,6 +861,11 @@ async def bulk_delete_users( ) continue + await AIPMOKernelService.invalidate_world_model_cache_for_user( + db=db, + tenant_id=tenant_id, + user_id=user_id, + ) await db.execute( delete(UserSkillBinding).where( and_( diff --git a/backend/app/schemas/world_model.py b/backend/app/schemas/world_model.py new file mode 100644 index 00000000..f002aa23 --- /dev/null +++ b/backend/app/schemas/world_model.py @@ -0,0 +1,136 @@ +"""AI PMO World Model 类型契约。""" + +from __future__ import annotations + +from typing import Any, NotRequired, Optional, TypedDict + + +class ProjectSnapshot(TypedDict): + id: int + name: str + title: str + description: str + status: str + owner_id: Optional[int] + + +class ProjectDomainSnapshot(TypedDict): + id: int + name: str + slug: str + description: str + owner_id: Optional[int] + + +class PortfolioSummary(TypedDict): + active_project_count: int + focus_project_id: Optional[int] + + +class BudgetSummary(TypedDict): + budget_id: Optional[int] + usage_ratio: float + hard_limit_enabled: bool + budget_type: NotRequired[str] + period: NotRequired[str] + budget_amount_usd: NotRequired[float] + current_usage_usd: NotRequired[float] + + +class CapacityItem(TypedDict): + user_id: int + current_assigned_hours: float + max_available_hours_per_week: float + workload_ratio: float + + +class SkillGraphItem(TypedDict): + user_id: int + skill_tag_id: int + proficiency_level: str + quality_score: float + efficiency_score: float + + +class CharterItem(TypedDict): + id: int + title: str + status: str + content: dict[str, Any] + + +class BlockedTask(TypedDict): + id: int + title: str + + +class TaskSummary(TypedDict): + total: int + by_status: dict[str, int] + overdue_count: int + uncovered_skill_tag_ids: list[int] + blocked_tasks: list[BlockedTask] + required_skill_tag_ids: NotRequired[list[int]] + + +class RiskSummary(TypedDict): + open_risk_count: int + + +class DependencySummary(TypedDict): + pressure: str + blocked_tasks: list[BlockedTask] + overdue_count: int + charter_ready: bool + + +class RecentSignalItem(TypedDict): + id: int + title: str + status: str + signal_type: str + + +class DecisionContextSnapshot(TypedDict): + captured_at: str + project_id: Optional[int] + budget: BudgetSummary + capacity: list[CapacityItem] + skill_graph_size: int + dependency_pressure: Optional[str] + + +class WorldModel(TypedDict): + sender: str + resolved_project_id: Optional[int] + project: Optional[ProjectSnapshot] + project_domain: Optional[ProjectDomainSnapshot] + portfolio: PortfolioSummary + budget: BudgetSummary + capacity: list[CapacityItem] + member_count: int + skill_graph: list[SkillGraphItem] + charter: list[CharterItem] + task_summary: TaskSummary + risk_summary: RiskSummary + dependencies: DependencySummary + recent_signals: list[RecentSignalItem] + decision_context_snapshot: DecisionContextSnapshot + + +__all__ = [ + "ProjectSnapshot", + "ProjectDomainSnapshot", + "PortfolioSummary", + "BudgetSummary", + "CapacityItem", + "SkillGraphItem", + "CharterItem", + "BlockedTask", + "TaskSummary", + "RiskSummary", + "DependencySummary", + "RecentSignalItem", + "DecisionContextSnapshot", + "WorldModel", +] diff --git a/backend/app/services/ai_pmo_kernel_service.py b/backend/app/services/ai_pmo_kernel_service.py index 2e4f22e1..8b46b543 100644 --- a/backend/app/services/ai_pmo_kernel_service.py +++ b/backend/app/services/ai_pmo_kernel_service.py @@ -2,6 +2,7 @@ from __future__ import annotations +import asyncio import hashlib import json import re @@ -15,6 +16,7 @@ from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from ..core.logging import get_logger +from ..database import AsyncSessionLocal from ..models import ( AIProcessingLog, Budget, @@ -36,10 +38,26 @@ from ..models import ( UserWorkload, WorkItem, ) +from ..schemas.world_model import ( + BlockedTask, + BudgetSummary, + CapacityItem, + CharterItem, + DependencySummary, + PortfolioSummary, + ProjectDomainSnapshot, + ProjectSnapshot, + RecentSignalItem, + RiskSummary, + SkillGraphItem, + TaskSummary, + WorldModel, +) from .adaptive_rag_service import AdaptiveRAGService from .ai_pmo_cadence_service import AIPMOCadenceService from .ai_steward import AIStewardService from .ai_pmo_review_service import AIPMOReviewService +from .context_cache import ContextCacheService from .notification_service import NotificationService from .openclaw_commander_service import OpenClawCommanderService from .ai_pmo_domain_workflow import SignalSourceDomainWorkflow @@ -103,6 +121,9 @@ POLICY_GATE_RULES: tuple[tuple[str, tuple[str, ...]], ...] = ( ("critical_path_change", ("关键路径", "里程碑改动", "影响多个项目", "跨项目依赖调整")), ) +WORLD_MODEL_STATIC_CACHE_TTL_SECONDS = 10 * 60 +WORLD_MODEL_STATIC_CACHE_KEY_PREFIX = "wm:static" + def _utc_now_naive() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) @@ -817,6 +838,7 @@ class AIPMOKernelService: def __init__(self, db: AsyncSession, tenant_id: int = 1) -> None: self.db = db self.tenant_id = int(tenant_id) + self.context_cache = ContextCacheService() self.intent_resolver = IntentResolver() self.decision_engine = DecisionEngine() self.reply_renderer = ReplyRenderer() @@ -923,11 +945,20 @@ class AIPMOKernelService: ) try: - world_model = await self._build_world_model( + world_model_query_tables: set[str] = set() + project = await self._resolve_project( message_text=clean_text, - sender=sender, project_hint_id=project_hint_id, + query_tables=world_model_query_tables, ) + world_model = await self.build_world_model( + tenant_id=self.tenant_id, + project_id=_safe_int(getattr(project, "id", None)), + sender=sender, + project=project, + pre_queried_tables=world_model_query_tables, + ) + self._schedule_portfolio_snapshot_persist(world_model=world_model) work_item.state = STATE_MODELED work_item.context_snapshot = world_model intent = self.intent_resolver.resolve( @@ -1239,64 +1270,262 @@ class AIPMOKernelService: await self.db.flush() return row + async def build_world_model( + self, + *, + tenant_id: int, + project_id: Optional[int], + sender: str = "", + project: Optional[Project] = None, + pre_queried_tables: Optional[set[str]] = None, + ) -> WorldModel: + if int(tenant_id) != int(self.tenant_id): + raise ValueError(f"tenant mismatch for world model build: input={tenant_id} service={self.tenant_id}") + + started = time.perf_counter() + queried_tables: set[str] = set(pre_queried_tables or set()) + resolved_project_id: Optional[int] = project_id + status = "failed" + try: + if project is None and project_id is not None: + self._record_world_model_query_table(queried_tables, "projects") + project = ( + await self.db.execute( + select(Project).where( + and_( + Project.tenant_id == self.tenant_id, + Project.id == int(project_id), + ) + ).limit(1) + ) + ).scalar_one_or_none() + + resolved_project_id = _safe_int(getattr(project, "id", None)) if project is not None else project_id + project_snapshot = self._serialize_project(project) + domain: Optional[ProjectDomainSnapshot] = None + skill_graph: list[SkillGraphItem] = [] + charter: list[CharterItem] = [] + + cached_static = await self._get_static_world_model_cache(tenant_id=self.tenant_id, project_id=resolved_project_id) + if cached_static is not None: + cached_project = cached_static.get("project") + if isinstance(cached_project, dict): + project_snapshot = cached_project + cached_domain = cached_static.get("project_domain") + if isinstance(cached_domain, dict): + domain = cached_domain + cached_skill_graph = cached_static.get("skill_graph") + if isinstance(cached_skill_graph, list): + skill_graph = [item for item in cached_skill_graph if isinstance(item, dict)] + cached_charter = cached_static.get("charter") + if isinstance(cached_charter, list): + charter = [item for item in cached_charter if isinstance(item, dict)] + else: + skill_graph = await self._build_skill_graph(project_id=resolved_project_id, query_tables=queried_tables) + charter = await self._build_charter_snapshot(project_id=resolved_project_id, query_tables=queried_tables) + domain = await self._build_project_domain_snapshot(project_id=resolved_project_id, query_tables=queried_tables) + await self._set_static_world_model_cache( + tenant_id=self.tenant_id, + project_id=resolved_project_id, + data={ + "project": project_snapshot, + "project_domain": domain, + "skill_graph": skill_graph, + "charter": charter, + }, + ) + + portfolio_summary = await self._build_portfolio_summary(project_id=resolved_project_id, query_tables=queried_tables) + budget = await self._build_budget_summary(project_id=resolved_project_id, query_tables=queried_tables) + capacity = await self._build_capacity_summary(project_id=resolved_project_id, query_tables=queried_tables) + task_summary = await self._build_task_summary( + project_id=resolved_project_id, + skill_graph=skill_graph, + query_tables=queried_tables, + ) + risk_summary = await self._build_risk_summary(project_id=resolved_project_id, query_tables=queried_tables) + dependencies = self._build_dependency_summary(charter=charter, task_summary=task_summary) + recent_signals = await self._build_recent_signal_summary(project_id=resolved_project_id, query_tables=queried_tables) + + world_model: WorldModel = { + "sender": sender, + "resolved_project_id": resolved_project_id, + "project": project_snapshot, + "project_domain": domain, + "portfolio": portfolio_summary, + "budget": budget, + "capacity": capacity, + "member_count": len(capacity), + "skill_graph": skill_graph, + "charter": charter, + "task_summary": task_summary, + "risk_summary": risk_summary, + "dependencies": dependencies, + "recent_signals": recent_signals, + "decision_context_snapshot": { + "captured_at": _utc_now_naive().isoformat(), + "project_id": resolved_project_id, + "budget": budget, + "capacity": capacity, + "skill_graph_size": len(skill_graph), + "dependency_pressure": dependencies.get("pressure"), + }, + } + status = "success" + return world_model + finally: + duration_ms = int((time.perf_counter() - started) * 1000) + logger.info( + "ai_pmo_world_model_build", + tenant_id=self.tenant_id, + project_id=resolved_project_id, + status=status, + duration_ms=duration_ms, + queried_tables=sorted(queried_tables), + query_table_count=len(queried_tables), + ) + + def _schedule_portfolio_snapshot_persist(self, *, world_model: WorldModel) -> None: + portfolio_summary = world_model.get("portfolio") + if not isinstance(portfolio_summary, dict): + return + project_id = _safe_int(world_model.get("resolved_project_id")) + asyncio.create_task( + self._persist_portfolio_snapshot(world_model=world_model, portfolio_summary=portfolio_summary), + name=f"ai-pmo-portfolio-snapshot:{self.tenant_id}:{project_id or 'none'}", + ) + + async def _persist_portfolio_snapshot( + self, + *, + world_model: WorldModel, + portfolio_summary: PortfolioSummary, + ) -> None: + try: + async with AsyncSessionLocal() as bg_db: + bg_db.add( + PortfolioSnapshot( + tenant_id=self.tenant_id, + scope_type="tenant", + scope_ref=f"tenant:{self.tenant_id}", + summary=portfolio_summary, + payload=world_model, + ) + ) + await bg_db.commit() + except Exception as exc: + logger.warning( + "ai_pmo_portfolio_snapshot_persist_failed", + tenant_id=self.tenant_id, + project_id=_safe_int(world_model.get("resolved_project_id")), + error=str(exc), + ) + + @staticmethod + def _world_model_static_cache_key(*, tenant_id: int, project_id: int) -> str: + return f"{WORLD_MODEL_STATIC_CACHE_KEY_PREFIX}:{int(tenant_id)}:{int(project_id)}" + + async def _get_static_world_model_cache( + self, + *, + tenant_id: int, + project_id: Optional[int], + ) -> Optional[dict[str, Any]]: + if project_id is None: + return None + key = self._world_model_static_cache_key(tenant_id=tenant_id, project_id=int(project_id)) + cached = await self.context_cache.get_json_key(key) + if not isinstance(cached, dict): + logger.debug(f"[WM_CACHE] MISS {tenant_id}:{project_id}") + return None + required_keys = {"project", "project_domain", "skill_graph", "charter"} + if not required_keys.issubset(set(cached.keys())): + logger.debug(f"[WM_CACHE] MISS {tenant_id}:{project_id}") + return None + logger.debug(f"[WM_CACHE] HIT {tenant_id}:{project_id}") + return cached + + async def _set_static_world_model_cache( + self, + *, + tenant_id: int, + project_id: Optional[int], + data: dict[str, Any], + ) -> None: + if project_id is None: + return + key = self._world_model_static_cache_key(tenant_id=tenant_id, project_id=int(project_id)) + await self.context_cache.set_json_key(key, data, WORLD_MODEL_STATIC_CACHE_TTL_SECONDS) + + async def _invalidate_world_model_cache(self, *, tenant_id: int, project_id: Optional[int]) -> None: + await self.invalidate_world_model_cache(tenant_id=tenant_id, project_id=project_id) + + @classmethod + async def invalidate_world_model_cache(cls, *, tenant_id: int, project_id: Optional[int]) -> None: + if project_id is None: + return + cache = ContextCacheService() + key = cls._world_model_static_cache_key(tenant_id=int(tenant_id), project_id=int(project_id)) + await cache.delete_keys(key) + + @classmethod + async def invalidate_world_model_cache_for_user( + cls, + *, + db: AsyncSession, + tenant_id: int, + user_id: int, + ) -> list[int]: + rows = ( + await db.execute( + select(UserProject.project_id).where( + and_( + UserProject.tenant_id == int(tenant_id), + UserProject.user_id == int(user_id), + ) + ) + ) + ).scalars().all() + project_ids = sorted({int(pid) for pid in rows if pid is not None}) + for pid in project_ids: + await cls.invalidate_world_model_cache(tenant_id=tenant_id, project_id=pid) + return project_ids + async def _build_world_model( self, *, message_text: str, sender: str, project_hint_id: Optional[int], - ) -> dict[str, Any]: - project = await self._resolve_project(message_text=message_text, project_hint_id=project_hint_id) - project_id = _safe_int(getattr(project, "id", None)) - portfolio_summary = await self._build_portfolio_summary(project_id=project_id) - budget = await self._build_budget_summary(project_id=project_id) - capacity = await self._build_capacity_summary(project_id=project_id) - skill_graph = await self._build_skill_graph(project_id=project_id) - charter = await self._build_charter_snapshot(project_id=project_id) - task_summary = await self._build_task_summary(project_id=project_id) - risk_summary = await self._build_risk_summary(project_id=project_id) - domain = await self._build_project_domain_snapshot(project_id=project_id) - dependencies = self._build_dependency_summary(charter=charter, task_summary=task_summary) - recent_signals = await self._build_recent_signal_summary(project_id=project_id) - - world_model = { - "sender": sender, - "resolved_project_id": project_id, - "project": self._serialize_project(project) if project is not None else None, - "project_domain": domain, - "portfolio": portfolio_summary, - "budget": budget, - "capacity": capacity, - "member_count": len(capacity), - "skill_graph": skill_graph, - "charter": charter, - "task_summary": task_summary, - "risk_summary": risk_summary, - "dependencies": dependencies, - "recent_signals": recent_signals, - "decision_context_snapshot": { - "captured_at": _utc_now_naive().isoformat(), - "project_id": project_id, - "budget": budget, - "capacity": capacity, - "skill_graph_size": len(skill_graph), - "dependency_pressure": dependencies.get("pressure"), - }, - } - self.db.add( - PortfolioSnapshot( - tenant_id=self.tenant_id, - scope_type="tenant", - scope_ref=f"tenant:{self.tenant_id}", - summary=portfolio_summary, - payload=world_model, - ) + ) -> WorldModel: + query_tables: set[str] = set() + project = await self._resolve_project( + message_text=message_text, + project_hint_id=project_hint_id, + query_tables=query_tables, + ) + return await self.build_world_model( + tenant_id=self.tenant_id, + project_id=_safe_int(getattr(project, "id", None)), + sender=sender, + project=project, + pre_queried_tables=query_tables, ) - await self.db.flush() - return world_model - async def _resolve_project(self, *, message_text: str, project_hint_id: Optional[int]) -> Optional[Project]: + @staticmethod + def _record_world_model_query_table(query_tables: Optional[set[str]], table_name: str) -> None: + if query_tables is not None: + query_tables.add(table_name) + + async def _resolve_project( + self, + *, + message_text: str, + project_hint_id: Optional[int], + query_tables: Optional[set[str]] = None, + ) -> Optional[Project]: if project_hint_id is not None: + self._record_world_model_query_table(query_tables, "projects") return ( await self.db.execute( select(Project).where( @@ -1311,6 +1540,7 @@ class AIPMOKernelService: id_match = re.search(r"(?:项目(?:id|ID)?|project(?:id)?)\s*[:#：]?\s*(\d+)", str(message_text or "")) if id_match: project_id = int(id_match.group(1)) + self._record_world_model_query_table(query_tables, "projects") return ( await self.db.execute( select(Project).where( @@ -1322,6 +1552,7 @@ class AIPMOKernelService: ) ).scalar_one_or_none() + self._record_world_model_query_table(query_tables, "projects") projects = ( await self.db.execute( select(Project).where(Project.tenant_id == self.tenant_id).order_by(Project.updated_at.desc(), Project.id.desc()).limit(50) @@ -1335,7 +1566,13 @@ class AIPMOKernelService: return project return None - async def _build_portfolio_summary(self, *, project_id: Optional[int]) -> dict[str, Any]: + async def _build_portfolio_summary( + self, + *, + project_id: Optional[int], + query_tables: Optional[set[str]] = None, + ) -> PortfolioSummary: + self._record_world_model_query_table(query_tables, "projects") active_project_count = ( await self.db.execute( select(func.count(Project.id)).where( @@ -1351,7 +1588,13 @@ class AIPMOKernelService: "focus_project_id": project_id, } - async def _build_budget_summary(self, *, project_id: Optional[int]) -> dict[str, Any]: + async def _build_budget_summary( + self, + *, + project_id: Optional[int], + query_tables: Optional[set[str]] = None, + ) -> BudgetSummary: + self._record_world_model_query_table(query_tables, "budgets") stmt = select(Budget).where(Budget.tenant_id == self.tenant_id) if project_id is not None: stmt = stmt.where(or_(Budget.project_id == int(project_id), Budget.project_id.is_(None))) @@ -1371,9 +1614,15 @@ class AIPMOKernelService: "hard_limit_enabled": bool(row.hard_limit_enabled), } - async def _build_capacity_summary(self, *, project_id: Optional[int]) -> list[dict[str, Any]]: + async def _build_capacity_summary( + self, + *, + project_id: Optional[int], + query_tables: Optional[set[str]] = None, + ) -> list[CapacityItem]: user_ids: list[int] = [] if project_id is not None: + self._record_world_model_query_table(query_tables, "user_projects") rows = ( await self.db.execute( select(UserProject.user_id).where( @@ -1386,6 +1635,7 @@ class AIPMOKernelService: ).scalars().all() user_ids = [int(uid) for uid in rows] if not user_ids and project_id is not None: + self._record_world_model_query_table(query_tables, "projects") owner_id = ( await self.db.execute( select(Project.owner_id).where( @@ -1400,6 +1650,7 @@ class AIPMOKernelService: user_ids = [int(owner_id)] if not user_ids: return [] + self._record_world_model_query_table(query_tables, "user_workloads") rows = ( await self.db.execute( select(UserWorkload).where( @@ -1420,9 +1671,15 @@ class AIPMOKernelService: for row in rows ] - async def _build_skill_graph(self, *, project_id: Optional[int]) -> list[dict[str, Any]]: + async def _build_skill_graph( + self, + *, + project_id: Optional[int], + query_tables: Optional[set[str]] = None, + ) -> list[SkillGraphItem]: user_ids: list[int] = [] if project_id is not None: + self._record_world_model_query_table(query_tables, "user_projects") rows = ( await self.db.execute( select(UserProject.user_id).where( @@ -1436,6 +1693,7 @@ class AIPMOKernelService: user_ids = [int(uid) for uid in rows] if not user_ids: return [] + self._record_world_model_query_table(query_tables, "user_skill_bindings") rows = ( await self.db.execute( select(UserSkillBinding).where( @@ -1457,9 +1715,16 @@ class AIPMOKernelService: for row in rows ] - async def _build_project_domain_snapshot(self, *, project_id: Optional[int]) -> Optional[dict[str, Any]]: + async def _build_project_domain_snapshot( + self, + *, + project_id: Optional[int], + query_tables: Optional[set[str]] = None, + ) -> Optional[ProjectDomainSnapshot]: if project_id is None: return None + self._record_world_model_query_table(query_tables, "projects") + self._record_world_model_query_table(query_tables, "project_domains") domain_row = ( await self.db.execute( select(ProjectDomain) @@ -1484,9 +1749,15 @@ class AIPMOKernelService: "owner_id": domain_row.owner_id, } - async def _build_charter_snapshot(self, *, project_id: Optional[int]) -> list[dict[str, Any]]: + async def _build_charter_snapshot( + self, + *, + project_id: Optional[int], + query_tables: Optional[set[str]] = None, + ) -> list[CharterItem]: if project_id is None: return [] + self._record_world_model_query_table(query_tables, "truth_nodes") rows = ( await self.db.execute( select(TruthNode).where( @@ -1508,9 +1779,16 @@ class AIPMOKernelService: for row in rows ] - async def _build_task_summary(self, *, project_id: Optional[int]) -> dict[str, Any]: + async def _build_task_summary( + self, + *, + project_id: Optional[int], + skill_graph: list[SkillGraphItem], + query_tables: Optional[set[str]] = None, + ) -> TaskSummary: if project_id is None: return {"total": 0, "by_status": {}, "overdue_count": 0, "uncovered_skill_tag_ids": [], "blocked_tasks": []} + self._record_world_model_query_table(query_tables, "tasks") tasks = ( await self.db.execute( select(Task).where(and_(Task.tenant_id == self.tenant_id, Task.project_id == int(project_id))) @@ -1518,7 +1796,7 @@ class AIPMOKernelService: ).scalars().all() by_status: dict[str, int] = {} required_skill_tag_ids: set[int] = set() - blocked_tasks: list[dict[str, Any]] = [] + blocked_tasks: list[BlockedTask] = [] overdue_count = 0 now = _utc_now_naive() for row in tasks: @@ -1533,7 +1811,6 @@ class AIPMOKernelService: due_date = row.due_date.replace(tzinfo=None) if row.due_date.tzinfo else row.due_date if due_date < now: overdue_count += 1 - skill_graph = await self._build_skill_graph(project_id=project_id) available_skill_ids = {int(item["skill_tag_id"]) for item in skill_graph if item.get("skill_tag_id") is not None} uncovered_skill_ids = sorted(skill_id for skill_id in required_skill_tag_ids if skill_id not in available_skill_ids) return { @@ -1545,9 +1822,16 @@ class AIPMOKernelService: "blocked_tasks": blocked_tasks[:8], } - async def _build_recent_signal_summary(self, *, project_id: Optional[int]) -> list[dict[str, Any]]: + async def _build_recent_signal_summary( + self, + *, + project_id: Optional[int], + query_tables: Optional[set[str]] = None, + ) -> list[RecentSignalItem]: if project_id is None: return [] + self._record_world_model_query_table(query_tables, "signals") + self._record_world_model_query_table(query_tables, "project_signal_links") rows = ( await self.db.execute( select(Signal).where( @@ -1578,7 +1862,13 @@ class AIPMOKernelService: for row in rows ] - async def _build_risk_summary(self, *, project_id: Optional[int]) -> dict[str, Any]: + async def _build_risk_summary( + self, + *, + project_id: Optional[int], + query_tables: Optional[set[str]] = None, + ) -> RiskSummary: + self._record_world_model_query_table(query_tables, "notifications") stmt = select(func.count(Notification.id)).where( and_( Notification.tenant_id == self.tenant_id, @@ -1592,7 +1882,7 @@ class AIPMOKernelService: return {"open_risk_count": int(open_risk_count or 0)} @staticmethod - def _build_dependency_summary(*, charter: list[dict[str, Any]], task_summary: dict[str, Any]) -> dict[str, Any]: + def _build_dependency_summary(*, charter: list[CharterItem], task_summary: TaskSummary) -> DependencySummary: blocked_tasks = task_summary.get("blocked_tasks") if isinstance(task_summary.get("blocked_tasks"), list) else [] overdue_count = int(task_summary.get("overdue_count") or 0) pressure = "high" if blocked_tasks or overdue_count > 0 else ("medium" if not charter else "low") @@ -2287,7 +2577,7 @@ class AIPMOKernelService: return "已按长连接模式处理飞书接入，当前不会走 webhook；这条说明不会再被当成信号源类型变更。" @staticmethod - def _serialize_project(project: Optional[Project]) -> Optional[dict[str, Any]]: + def _serialize_project(project: Optional[Project]) -> Optional[ProjectSnapshot]: if project is None: return None return { diff --git a/backend/app/services/context_cache.py b/backend/app/services/context_cache.py index ead979b0..ed720ec1 100644 --- a/backend/app/services/context_cache.py +++ b/backend/app/services/context_cache.py @@ -106,6 +106,24 @@ class ContextCacheService: except Exception: pass + async def get_json_key(self, key: str) -> Optional[dict[str, Any]]: + return await self._get_json(key) + + async def set_json_key(self, key: str, value: dict[str, Any], ttl: int) -> None: + await self._set_json(key, value, ttl) + + async def delete_keys(self, *keys: str) -> None: + valid_keys = [str(key).strip() for key in keys if str(key).strip()] + if not valid_keys: + return + redis = await self._get_redis() + if redis is None: + return + try: + await redis.delete(*valid_keys) + except Exception: + pass + @staticmethod def _is_system_snapshot_complete(payload: Optional[dict[str, Any]]) -> bool: if not isinstance(payload, dict): diff --git a/backend/app/services/skill_service.py b/backend/app/services/skill_service.py index c6d0eed7..eed98e12 100644 --- a/backend/app/services/skill_service.py +++ b/backend/app/services/skill_service.py @@ -17,6 +17,7 @@ from sqlalchemy.orm import selectinload from ..models import SkillTag, UserSkillBinding, UserWorkload, User, UserStatus, Task from ..database import _is_sqlite +from .ai_pmo_kernel_service import AIPMOKernelService # 系统预设 AI 开发专属技能标签 @@ -337,6 +338,11 @@ async def batch_update_user_skill_bindings( await db.commit() for b in result_bindings: await db.refresh(b) + await AIPMOKernelService.invalidate_world_model_cache_for_user( + db=db, + tenant_id=tenant_id, + user_id=user_id, + ) return result_bindings