From a3a4ce61ffdb8a96927528bd0f752357101dfc73 Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Thu, 12 Mar 2026 09:10:36 +0800 Subject: [PATCH] Update AI steward and audit logic --- .openclaw/pmflow-routing-rules.json | 17 +- CLAUDE.md | 19 + .../20260311_add_ai_pmo_work_items_view.py | 59 +++ ...20260311_add_operation_log_audit_fields.py | 79 +++ ...11_expand_user_workload_ratio_precision.py | 74 +++ backend/app/api/ai_steward.py | 43 ++ backend/app/api/audit.py | 68 +++ backend/app/api/projects.py | 162 +++++- backend/app/api/router.py | 2 + backend/app/database.py | 165 ++++++- backend/app/dependencies/auth.py | 46 +- backend/app/dependencies/tenant.py | 8 +- backend/app/main.py | 18 +- backend/app/middleware/__init__.py | 5 + backend/app/middleware/audit_context.py | 44 ++ backend/app/models/operation_log.py | 9 + backend/app/models/user_workload.py | 2 +- backend/app/services/ai_pmo_kernel_service.py | 343 ++++++++++--- backend/app/services/ai_steward.py | 152 +++++- backend/app/services/audit_query_service.py | 74 +++ backend/app/services/codegen.py | 3 +- backend/app/services/llm.py | 1 + .../services/openclaw_commander_service.py | 18 +- backend/app/services/operation_log_service.py | 12 + .../app/services/project_member_service.py | 240 +++++++++ .../services/signal_source_repair_service.py | 465 ++++++++++++++++++ backend/app/services/sstc_service.py | 91 +++- backend/app/utils/audited_write.py | 112 +++++ backend/app/utils/project_members.py | 80 +-- backend/neural_sync.db-shm | Bin 0 -> 32768 bytes backend/neural_sync.db-wal | 0 backend/tests/test_ai_pmo_kernel_service.py | 205 ++++++++ .../tests/test_ai_steward_title_extract.py | 37 ++ .../tests/test_openclaw_commander_service.py | 8 + .../test_projects_member_assignment_api.py | 203 ++++++++ docker-compose.dev.yml | 2 +- docker-compose.yml | 2 +- frontend/src/api/aiSteward.ts | 26 + frontend/src/api/audit.ts | 61 +++ frontend/src/api/projects.ts | 24 + frontend/src/api/sstc.ts | 7 + .../src/components/sstc/SSTCGlobalHeader.tsx | 61 ++- 42 files changed, 2840 insertions(+), 207 deletions(-) create mode 100644 backend/alembic/versions/20260311_add_ai_pmo_work_items_view.py create mode 100644 backend/alembic/versions/20260311_add_operation_log_audit_fields.py create mode 100644 backend/alembic/versions/20260311_expand_user_workload_ratio_precision.py create mode 100644 backend/app/api/audit.py create mode 100644 backend/app/middleware/__init__.py create mode 100644 backend/app/middleware/audit_context.py create mode 100644 backend/app/services/audit_query_service.py create mode 100644 backend/app/services/project_member_service.py create mode 100644 backend/app/services/signal_source_repair_service.py create mode 100644 backend/app/utils/audited_write.py create mode 100644 backend/neural_sync.db-shm create mode 100644 backend/neural_sync.db-wal create mode 100644 backend/tests/test_projects_member_assignment_api.py diff --git a/.openclaw/pmflow-routing-rules.json b/.openclaw/pmflow-routing-rules.json index 649fa1f4..bfa6d13c 100644 --- a/.openclaw/pmflow-routing-rules.json +++ b/.openclaw/pmflow-routing-rules.json @@ -1,9 +1,24 @@ { "version": "v2", - "updated_at": "2026-03-11T08:35:00Z", + "updated_at": "2026-03-11T09:10:00Z", "source_of_truth": "openclaw", "description": "PMFlow 三引擎统一路由：所有主链路统一经 OpenClaw OpenAI-Compatible 网关。", "rules": [ + { + "name": "engine_low_project_draft_glm47_fallback", + "enabled": true, + "priority": 340, + "tier": "low", + "provider": "glm", + "model": "GLM-4.7", + "api_env": "GLM_API_KEY", + "match_any": [ + "project_draft_analysis_glm47_fallback", + "project_draft_signal_glm47_fallback", + "project_draft_charter_glm47_fallback", + "project_draft_task_chain_glm47_fallback" + ] + }, { "name": "engine_high_intent_complex_exact", "enabled": true, diff --git a/CLAUDE.md b/CLAUDE.md index 89477318..1731f694 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -33,6 +33,7 @@ backend/ │ ├── api/ # 路由层（含 endpoints） │ ├── core/ # 配置、日志、核心能力 │ ├── dependencies/ # 鉴权/租户等依赖注入 +│ ├── middleware/ # 请求中间件（审计上下文注入等） │ ├── models/ # SQLAlchemy 模型 │ ├── schemas/ # Pydantic 模型 │ ├── services/ # 业务服务（AI/SSTC/集成） @@ -43,6 +44,9 @@ backend/ │ ├── 20260310_add_feishu_outbox_reply_hash.py # 飞书 outbox 回执哈希幂等字段 │ ├── 20260311_add_pg_trgm_text_indexes.py # pg_trgm 扩展与文本检索索引 │ ├── 20260311_add_kb_chunk_ivfflat_index.py # knowledge_base_chunks 向量 IVFFlat 索引 +│ ├── 20260311_add_operation_log_audit_fields.py # operation_logs 审计字段扩展(trace/path/caller) +│ ├── 20260311_add_ai_pmo_work_items_view.py # ai_pmo_work_items 兼容视图（映射 work_items） +│ ├── 20260311_expand_user_workload_ratio_precision.py # user_workloads.workload_ratio 精度扩容(10,2) │ └── 20260311_add_conversation_sessions_table.py # 飞书多轮会话上下文持久化表 ├── scripts/ # 后端运维/数据修复脚本 │ └── backfill_orphan_signal_links.py # 回填孤立 signal 的 project_signal_links @@ -53,6 +57,11 @@ backend/ - 新增：`backend/app/api/llm_providers.py`（AI 管家 LLM provider preset 配置读取接口） - 新增：`backend/app/services/ai_agent_kernel.py`（双 Agent 共享决策内核：复杂度路由/执行门控/修正识别） - 新增：`backend/app/utils/timezone.py`（统一北京时间工具：`now_cn_aware/now_cn_naive`） +- 新增：`backend/app/utils/audited_write.py`（写接口审计上下文、caller 归一、北京时间格式化） +- 新增：`backend/app/middleware/audit_context.py`（请求级 trace/interface/caller 注入） +- 新增：`backend/app/api/audit.py`（审计查询接口：`GET /audit/operations`） +- 新增：`backend/app/services/audit_query_service.py`（operation_logs 过滤与序列化） +- 新增：`backend/app/services/signal_source_repair_service.py`（脏信号源回填服务） - 新增：`backend/alembic/versions/20260311_shift_all_timestamps_to_beijing.py`（全库历史时间 +8h，数据库默认时区切到 Asia/Shanghai） - 新增：`backend/tests/test_ai_pmo_work_item_decision_api.py`（AI PMO 审批动作接口回归测试） - 移除：`backend/.env.example`（配置模板统一到根目录 `.env.example`） @@ -139,16 +148,23 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 ## 8. 服务边界约束（禁止越界） - 数据库写操作必须通过 Alembic migration，禁止直接 `ALTER TABLE` +- 运行期禁止 DDL：`database.py` 钩子统一拦截 `CREATE/ALTER/DROP/...`，仅迁移路径允许执行 - 新增接口必须遵循响应包络：`{code, message, data}`，不得裸返回 - 禁止在 API 层写业务逻辑，业务逻辑归属 `services/` - 禁止在前端组件内直接调用 `fetch`，必须通过 `frontend/src/api/` 封装层 - 前端状态管理：服务端数据用 React Query，本地 UI 状态用 `useState` - 飞书回执发送仅允许 `main.py` 的 drain_loop 作为消费者，禁止在业务路径即时派发 outbox 发送任务 +- 所有写接口必须落审计日志（`operation_logs`）：至少包含 `trace_id/interface_path/caller` +- `created_at_beijing` 为计算属性返回，禁止新增数据库物理列 ## 9. 新增代码的参考模式 - 新增后端接口：参考 `backend/app/api/sstc.py` 的结构 +- 审计查询接口：参考 `backend/app/api/audit.py`（`GET /audit/operations`） +- 脏数据修复接口：参考 `backend/app/api/ai_steward.py`（`POST /ai-steward/signal-sources/repair-dirty`） - 配置读取类接口：参考 `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` +- SSTC 总览接口（`GET /projects/{project_id}/sstc/overview`）：`workload_health` 返回成员明细 `members[]`（姓名/角色/负载比例/状态）用于头部卡片展示 - AI PMO 审批接口（`POST /ai-pmo/work-items/{work_item_id}/decision`）：显式 `approve/reject`，`reject` 必须落 `ignored`（失效）并同步 `decision_records.final_outcome` - 新增前端 API 调用：参考 `frontend/src/api/sstc.ts` - 新增数据库表：参考 `alembic/versions/` 最新 migration @@ -175,3 +191,6 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - `project.create` 启用槽位完整性闸门：无法从指令提取有效项目名时标记 `slot_incomplete` 并进入人工确认 - 数据库与核心写入链路统一使用北京时间（`Asia/Shanghai`）：PostgreSQL 会话时区、数据库默认时区与核心 `_utc_now*` 写入入口均已切换到北京时间 - 后端与脚本层已清零 `datetime.utcnow()/timezone.utc`：统一使用 `backend/app/utils/timezone.py`（`now_cn_aware/now_cn_naive/CN_TZ`）管理时间语义 +- `operation_logs.created_at_beijing` 为应用层计算字段（非 DB 列），审计查询接口返回时统一补齐 +- 历史脏信号源修复优先复用同项目已存在 `signal_id`，仅无法复用时才触发 `capture-feishu` 补录 +- 兼容 SQL 查询入口：`ai_pmo_work_items` 为视图别名（底层映射 `work_items`），避免历史脚本直接查旧表名失败 diff --git a/backend/alembic/versions/20260311_add_ai_pmo_work_items_view.py b/backend/alembic/versions/20260311_add_ai_pmo_work_items_view.py new file mode 100644 index 00000000..343849a5 --- /dev/null +++ b/backend/alembic/versions/20260311_add_ai_pmo_work_items_view.py @@ -0,0 +1,59 @@ +"""add ai_pmo_work_items compatibility view + +Revision ID: add_ai_pmo_work_items_view +Revises: add_operation_log_audit_fields +Create Date: 2026-03-11 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "add_ai_pmo_work_items_view" +down_revision = "add_operation_log_audit_fields" +branch_labels = None +depends_on = None + + +def _table_exists(bind, table_name: str) -> bool: + return sa.inspect(bind).has_table(table_name) + + +def upgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "work_items"): + return + + op.execute("DROP VIEW IF EXISTS ai_pmo_work_items") + op.execute( + """ + CREATE VIEW ai_pmo_work_items AS + SELECT + id, + ingest_event_id, + source_system, + source_message_id, + state, + intent, + decision_level, + project_id, + signal_id, + owner_user_id, + dedupe_key, + context_snapshot, + action_bundle, + execution_result, + review_summary, + reply_template_key, + tenant_id, + created_at, + updated_at + FROM work_items + """ + ) + + +def downgrade() -> None: + op.execute("DROP VIEW IF EXISTS ai_pmo_work_items") diff --git a/backend/alembic/versions/20260311_add_operation_log_audit_fields.py b/backend/alembic/versions/20260311_add_operation_log_audit_fields.py new file mode 100644 index 00000000..f131545c --- /dev/null +++ b/backend/alembic/versions/20260311_add_operation_log_audit_fields.py @@ -0,0 +1,79 @@ +"""add operation log audit fields + +Revision ID: add_operation_log_audit_fields +Revises: shift_all_timestamps_to_beijing +Create Date: 2026-03-11 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "add_operation_log_audit_fields" +down_revision = "shift_all_timestamps_to_beijing" +branch_labels = None +depends_on = None + + +def _table_exists(bind, table_name: str) -> bool: + return sa.inspect(bind).has_table(table_name) + + +def _column_exists(bind, table_name: str, column_name: str) -> bool: + inspector = sa.inspect(bind) + if not inspector.has_table(table_name): + return False + return any(col.get("name") == column_name for col in inspector.get_columns(table_name)) + + +def _index_exists(bind, table_name: str, index_name: str) -> bool: + inspector = sa.inspect(bind) + if not inspector.has_table(table_name): + return False + return any(idx.get("name") == index_name for idx in inspector.get_indexes(table_name)) + + +def upgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "operation_logs"): + return + + with op.batch_alter_table("operation_logs") as batch_op: + if not _column_exists(bind, "operation_logs", "trace_id"): + batch_op.add_column(sa.Column("trace_id", sa.String(length=64), nullable=True)) + if not _column_exists(bind, "operation_logs", "interface_path"): + batch_op.add_column(sa.Column("interface_path", sa.String(length=255), nullable=True)) + if not _column_exists(bind, "operation_logs", "caller"): + batch_op.add_column( + sa.Column("caller", sa.String(length=16), nullable=False, server_default=sa.text("'system'")) + ) + + if not _index_exists(bind, "operation_logs", "ix_operation_logs_trace_id"): + op.create_index("ix_operation_logs_trace_id", "operation_logs", ["trace_id"], unique=False) + if not _index_exists(bind, "operation_logs", "ix_operation_logs_interface_path"): + op.create_index("ix_operation_logs_interface_path", "operation_logs", ["interface_path"], unique=False) + if not _index_exists(bind, "operation_logs", "ix_operation_logs_caller"): + op.create_index("ix_operation_logs_caller", "operation_logs", ["caller"], unique=False) + + +def downgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "operation_logs"): + return + + if _index_exists(bind, "operation_logs", "ix_operation_logs_caller"): + op.drop_index("ix_operation_logs_caller", table_name="operation_logs") + if _index_exists(bind, "operation_logs", "ix_operation_logs_interface_path"): + op.drop_index("ix_operation_logs_interface_path", table_name="operation_logs") + if _index_exists(bind, "operation_logs", "ix_operation_logs_trace_id"): + op.drop_index("ix_operation_logs_trace_id", table_name="operation_logs") + + with op.batch_alter_table("operation_logs") as batch_op: + if _column_exists(bind, "operation_logs", "caller"): + batch_op.drop_column("caller") + if _column_exists(bind, "operation_logs", "interface_path"): + batch_op.drop_column("interface_path") + if _column_exists(bind, "operation_logs", "trace_id"): + batch_op.drop_column("trace_id") diff --git a/backend/alembic/versions/20260311_expand_user_workload_ratio_precision.py b/backend/alembic/versions/20260311_expand_user_workload_ratio_precision.py new file mode 100644 index 00000000..e3384bf8 --- /dev/null +++ b/backend/alembic/versions/20260311_expand_user_workload_ratio_precision.py @@ -0,0 +1,74 @@ +"""expand user_workloads workload_ratio precision + +Revision ID: expand_user_workload_ratio_precision +Revises: add_ai_pmo_work_items_view +Create Date: 2026-03-11 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "expand_user_workload_ratio_precision" +down_revision = "add_ai_pmo_work_items_view" +branch_labels = None +depends_on = None + + +def _table_exists(bind, table_name: str) -> bool: + return sa.inspect(bind).has_table(table_name) + + +def _column_numeric_meta(bind, table_name: str, column_name: str) -> tuple[int | None, int | None] | None: + inspector = sa.inspect(bind) + if not inspector.has_table(table_name): + return None + for column in inspector.get_columns(table_name): + if str(column.get("name")) == column_name: + col_type = column.get("type") + return ( + int(getattr(col_type, "precision", 0) or 0) or None, + int(getattr(col_type, "scale", 0) or 0) or None, + ) + return None + + +def _alter_workload_ratio(bind, *, precision: int, scale: int) -> None: + current = _column_numeric_meta(bind, "user_workloads", "workload_ratio") + if current == (precision, scale): + return + current_precision = int((current or (4, 2))[0] or 4) + current_scale = int((current or (4, 2))[1] or 2) + with op.batch_alter_table("user_workloads") as batch_op: + batch_op.alter_column( + "workload_ratio", + existing_type=sa.Numeric(current_precision, current_scale), + type_=sa.Numeric(precision, scale), + existing_nullable=False, + ) + + +def upgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "user_workloads"): + return + _alter_workload_ratio(bind, precision=10, scale=2) + + +def downgrade() -> None: + bind = op.get_bind() + if not _table_exists(bind, "user_workloads"): + return + op.execute( + """ + UPDATE user_workloads + SET workload_ratio = CASE + WHEN workload_ratio > 99.99 THEN 99.99 + WHEN workload_ratio < -99.99 THEN -99.99 + ELSE workload_ratio + END + """ + ) + _alter_workload_ratio(bind, precision=4, scale=2) diff --git a/backend/app/api/ai_steward.py b/backend/app/api/ai_steward.py index a5cd63cf..9a01d285 100644 --- a/backend/app/api/ai_steward.py +++ b/backend/app/api/ai_steward.py @@ -32,7 +32,13 @@ from ..services.adaptive_rag_service import AdaptiveRAGService from ..services.ai_orchestration_run_service import AIOrchestrationRunService from ..services.openclaw_commander_service import OpenClawCommanderService from ..services.signal_source_orchestration_service import SignalSourceOrchestrationService +from ..services.signal_source_repair_service import ( + SignalSourceRepairMissingSignalError, + SignalSourceRepairService, + SignalSourceRepairTimeoutError, +) from ..schemas.rag import RAGAskRequest +from ..utils.audited_write import audited_write from ..utils.timezone import now_cn_aware, now_cn_naive @@ -166,6 +172,12 @@ class SignalSourceListResponse(BaseModel): items: List[Dict[str, Any]] +class SignalSourceRepairRequest(BaseModel): + project_id: int = Field(..., ge=1) + registry_ids: List[int] = Field(default_factory=list, min_length=1, max_length=50) + sender: str = Field(default="PMFlow修复器", min_length=1, max_length=64) + + def _parse_related_users(raw_value: Any) -> List[str]: """兼容 involved_user_ids 的 list / json-string / 单值字符串。""" if raw_value is None: @@ -219,6 +231,7 @@ async def ask_ai_steward_rag( @router.post("/command/dispatch") +@audited_write(caller="user") async def dispatch_openclaw_command( request: OpenClawCommandDispatchRequest, current_user: User = Depends(require_manager_user), @@ -246,6 +259,7 @@ async def dispatch_openclaw_command( @router.post("/signal-sources/onboard") +@audited_write(caller="user") async def onboard_signal_source( request: SignalSourceOnboardRequest, current_user: User = Depends(require_manager_user), @@ -341,6 +355,34 @@ async def list_project_signal_sources( ) return success_response(SignalSourceListResponse(items=items).model_dump()) + +@router.post("/signal-sources/repair-dirty") +@audited_write(caller="system") +async def repair_dirty_signal_sources( + request: SignalSourceRepairRequest, + current_user: User = Depends(require_manager_user), + db: AsyncSession = Depends(get_db), + tenant_id: int = Depends(get_tenant_id), +): + effective_tenant_id = int(getattr(current_user, "tenant_id", tenant_id) or tenant_id) + try: + rows = await SignalSourceRepairService(db).repair_registries( + tenant_id=effective_tenant_id, + project_id=int(request.project_id), + registry_ids=[int(item) for item in request.registry_ids], + sender=str(request.sender or "PMFlow修复器").strip()[:64] or "PMFlow修复器", + ) + except SignalSourceRepairTimeoutError as exc: + raise HTTPException(status_code=504, detail=str(exc)) from exc + except SignalSourceRepairMissingSignalError as exc: + raise HTTPException(status_code=502, detail=str(exc)) from exc + return success_response( + { + "project_id": int(request.project_id), + "items": rows, + } + ) + @router.post("/project-from-text", response_model=ProjectFromTextResponse) async def create_project_from_text( request: ProjectFromTextRequest, @@ -384,6 +426,7 @@ async def create_project_from_text( # ==================== Signal Capture Endpoints ==================== @router.post("/signal/capture-feishu") +@audited_write(caller="ai") async def capture_feishu_signal( request: FeishuCaptureRequest, current_user: User = Depends(get_current_user), diff --git a/backend/app/api/audit.py b/backend/app/api/audit.py new file mode 100644 index 00000000..b659f157 --- /dev/null +++ b/backend/app/api/audit.py @@ -0,0 +1,68 @@ +"""审计日志查询 API。""" + +from __future__ import annotations + +from datetime import datetime +from typing import Optional + +from fastapi import APIRouter, Depends, HTTPException, Query +from sqlalchemy.ext.asyncio import AsyncSession + +from ..core.response import success_response +from ..database import get_db +from ..dependencies.auth import get_optional_current_user +from ..dependencies.tenant import get_tenant_id +from ..models import User +from ..services.audit_query_service import AuditQueryService +from ..services.auth import has_any_role + +router = APIRouter(prefix="/audit", tags=["Audit"]) + + +def _effective_tenant_id(current_user: User, header_tenant_id: int) -> int: + try: + return int(getattr(current_user, "tenant_id") or header_tenant_id or 1) + except Exception: + return int(header_tenant_id or 1) + + +def _user_role(user: User) -> str: + role = getattr(user, "role", "") + return str(getattr(role, "value", role) or "").strip() + + +@router.get("/operations") +async def list_audit_operations( + tenant_id: Optional[int] = Query(default=None, ge=1), + entity_type: Optional[str] = Query(default=None), + start_time: Optional[datetime] = Query(default=None), + end_time: Optional[datetime] = Query(default=None), + limit: int = Query(default=100, ge=1, le=100), + current_user: Optional[User] = Depends(get_optional_current_user), + db: AsyncSession = Depends(get_db), + header_tenant_id: int = Depends(get_tenant_id), +): + if current_user is not None and not has_any_role(_user_role(current_user), ["BOSS", "ADMIN"]): + raise HTTPException(status_code=403, detail="Access denied. Required roles: BOSS, ADMIN") + if current_user is None: + effective_tenant_id = int(header_tenant_id or 1) + else: + effective_tenant_id = _effective_tenant_id(current_user, header_tenant_id) + if tenant_id is not None and int(tenant_id) != effective_tenant_id: + raise HTTPException(status_code=403, detail="cross-tenant query is forbidden") + scoped_tenant_id = int(tenant_id or effective_tenant_id) + + items = await AuditQueryService(db).list_operations( + tenant_id=scoped_tenant_id, + entity_type=entity_type, + start_time=start_time, + end_time=end_time, + limit=limit, + ) + return success_response( + { + "items": items, + "count": len(items), + "tenant_id": scoped_tenant_id, + } + ) diff --git a/backend/app/api/projects.py b/backend/app/api/projects.py index caf3cbf5..10b7a2f3 100644 --- a/backend/app/api/projects.py +++ b/backend/app/api/projects.py @@ -6,7 +6,7 @@ import json from typing import Any, Optional from fastapi import APIRouter, Depends, HTTPException, Query -from pydantic import BaseModel +from pydantic import BaseModel, Field from sqlalchemy import and_, delete, func, select, update from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession @@ -33,7 +33,13 @@ from ..models import ( ) 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 ..services.operation_log_service import log_operation +from ..services.project_member_service import ( + ProjectMemberAssignmentError, + notify_member_assignment_failure, + replace_project_members, + resolve_project_member_ids, +) from ..utils.sequence import ensure_project_id_sequence from ..utils.title import compact_title @@ -60,6 +66,7 @@ class ProjectCreateRequest(BaseModel): status: str = "active" domain_id: Optional[int] = None assigned_role: Optional[str] = None + member_user_ids: list[int] = Field(default_factory=list) class ProjectUpdateRequest(BaseModel): @@ -75,6 +82,10 @@ class LinkSignalRequest(BaseModel): relevance_score: Optional[float] = 1.0 +class ProjectMembersAssignRequest(BaseModel): + member_user_ids: list[int] = Field(default_factory=list) + + def _serialize_project_row(project: Project) -> dict: return { "id": str(project.id), @@ -151,6 +162,23 @@ async def _build_project_summary( return base +async def _fetch_project_member_user_ids( + db: AsyncSession, + *, + tenant_id: int, + project_id: int, +) -> list[int]: + rows = await db.execute( + select(UserProject.user_id).where( + and_( + UserProject.tenant_id == int(tenant_id), + UserProject.project_id == int(project_id), + ) + ) + ) + return sorted(int(user_id) for user_id in rows.scalars().all() if user_id is not None) + + async def _get_project_or_404( db: AsyncSession, *, @@ -364,6 +392,28 @@ async def create_project( title = compact_title(body.title, max_len=24) if not title: raise HTTPException(status_code=400, detail="title required") + try: + member_user_ids = await resolve_project_member_ids( + db, + tenant_id=tenant_id, + owner_id=int(current_user.id), + assigned_role=body.assigned_role, + requested_member_user_ids=body.member_user_ids, + require_requested_members=False, + include_owner=True, + ) + except ProjectMemberAssignmentError as exc: + await notify_member_assignment_failure( + db, + tenant_id=tenant_id, + actor_user_id=int(current_user.id), + project_title=title, + reason=str(exc), + interface_path="/projects/", + commit=True, + ) + raise HTTPException(status_code=400, detail=str(exc)) from exc + await ensure_project_id_sequence(db) project = Project( tenant_id=tenant_id, @@ -377,21 +427,111 @@ async def create_project( signal_count=0, ) db.add(project) - await db.flush() - await ensure_project_default_members( - db, - tenant_id=tenant_id, - project_id=int(project.id), - owner_id=int(current_user.id), - assigned_role=project.assigned_role, - ) - await db.commit() + try: + await db.flush() + await replace_project_members( + db, + tenant_id=tenant_id, + project_id=int(project.id), + member_user_ids=member_user_ids, + ) + await log_operation( + db, + tenant_id=tenant_id, + user_id=int(current_user.id), + project_id=int(project.id), + entity_type="project", + entity_id=str(project.id), + action="project.create", + before_data=None, + after_data={ + "title": project.title, + "status": project.status, + "member_user_ids": member_user_ids, + }, + interface_path="/projects/", + caller="user", + ) + await db.commit() + except Exception: + await db.rollback() + raise 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) +@router.put("/{project_id}/members") +async def assign_project_members( + project_id: int, + body: ProjectMembersAssignRequest, + db: AsyncSession = Depends(get_db), + current_user: User = Depends(require_roles("BOSS", "ADMIN")), +): + tenant_id = int(current_user.tenant_id) + project = await _get_project_or_404(db, tenant_id=tenant_id, project_id=project_id) + before_member_user_ids = await _fetch_project_member_user_ids( + db, + tenant_id=tenant_id, + project_id=int(project.id), + ) + + try: + member_user_ids = await resolve_project_member_ids( + db, + tenant_id=tenant_id, + owner_id=None, + assigned_role=project.assigned_role, + requested_member_user_ids=body.member_user_ids, + require_requested_members=True, + include_owner=False, + ) + await replace_project_members( + db, + tenant_id=tenant_id, + project_id=int(project.id), + member_user_ids=member_user_ids, + ) + await log_operation( + db, + tenant_id=tenant_id, + user_id=int(current_user.id), + project_id=int(project.id), + entity_type="project", + entity_id=str(project.id), + action="project.members.assign", + before_data={"member_user_ids": before_member_user_ids}, + after_data={"member_user_ids": member_user_ids}, + interface_path=f"/projects/{project_id}/members", + caller="user", + ) + await db.commit() + except ProjectMemberAssignmentError as exc: + await db.rollback() + await notify_member_assignment_failure( + db, + tenant_id=tenant_id, + actor_user_id=int(current_user.id), + project_id=int(project.id), + project_title=project.title or project.name, + reason=str(exc), + interface_path=f"/projects/{project_id}/members", + commit=True, + ) + raise HTTPException(status_code=400, detail=str(exc)) from exc + except Exception: + await db.rollback() + raise + + 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 { + "project_id": str(project.id), + "member_user_ids": [str(user_id) for user_id in member_user_ids], + } + + @router.patch("/{project_id}") async def update_project( project_id: int, diff --git a/backend/app/api/router.py b/backend/app/api/router.py index c248d46f..a879274d 100644 --- a/backend/app/api/router.py +++ b/backend/app/api/router.py @@ -34,6 +34,7 @@ from .recipes import router as recipes_router from .signal_sources import router as signal_sources_router from .operation_catalog import router as operation_catalog_router from .ai_pmo import router as ai_pmo_router +from .audit import router as audit_router # 创建主 API 路由器 api_router = APIRouter() @@ -67,3 +68,4 @@ api_router.include_router(recipes_router) api_router.include_router(signal_sources_router) api_router.include_router(operation_catalog_router) api_router.include_router(ai_pmo_router) +api_router.include_router(audit_router) diff --git a/backend/app/database.py b/backend/app/database.py index f8fa98b4..546f37b0 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -2,20 +2,46 @@ from __future__ import annotations +import os +import traceback from collections.abc import AsyncGenerator +from fastapi import Request +from sqlalchemy import event from sqlalchemy.ext.asyncio import ( AsyncEngine, AsyncSession, async_sessionmaker, create_async_engine, ) -from sqlalchemy import event +from sqlalchemy.orm import Session from sqlalchemy.pool import NullPool from .core.config import settings +from .core.logging import get_logger +from .models import OperationLog +from .utils.audited_write import get_audit_context DATABASE_URL = settings.database_url +logger = get_logger(__name__) + +AUDIT_ENFORCE_KEY = "_pmflow_audit_enforce" +AUDIT_WRITTEN_KEY = "_pmflow_audit_written" +AUDIT_BUSINESS_WRITE_KEY = "_pmflow_business_write" +AUDIT_METHOD_KEY = "_pmflow_request_method" +AUDIT_PATH_KEY = "_pmflow_interface_path" +AUDIT_TENANT_KEY = "_pmflow_tenant_id" + +DDL_PREFIXES = ( + "create ", + "alter ", + "drop ", + "truncate ", + "rename ", + "comment ", + "grant ", + "revoke ", +) def _is_sqlite() -> bool: @@ -61,6 +87,126 @@ if _is_sqlite(): finally: cursor.close() + +def _strip_sql_leading_comments(statement: str) -> str: + text = str(statement or "") + while True: + stripped = text.lstrip() + if stripped.startswith("/*") and "*/" in stripped: + text = stripped.split("*/", 1)[1] + continue + if stripped.startswith("--") and "\n" in stripped: + text = stripped.split("\n", 1)[1] + continue + return stripped + + +def _is_ddl_statement(statement: str) -> bool: + normalized = _strip_sql_leading_comments(statement).lower() + return any(normalized.startswith(prefix) for prefix in DDL_PREFIXES) + + +def _is_migration_stack() -> bool: + for frame in traceback.extract_stack(limit=50): + filename = str(frame.filename or "").replace("\\", "/").lower() + if "/alembic/" in filename: + return True + return False + + +@event.listens_for(engine.sync_engine, "before_cursor_execute", retval=True) +def _block_runtime_ddl(_conn, _cursor, statement, parameters, _context, _executemany): # pragma: no cover + if os.getenv("PMFLOW_ALLOW_RUNTIME_DDL", "0").lower() in {"1", "true", "yes"}: + return statement, parameters + if not _is_ddl_statement(str(statement or "")): + return statement, parameters + if _is_migration_stack(): + return statement, parameters + context = get_audit_context() + logger.error( + "runtime_ddl_blocked", + trace_id=context.trace_id, + interface_path=context.interface_path, + statement_preview=str(statement or "")[:200], + ) + raise RuntimeError("运行期禁止 DDL，请使用 Alembic migration。") + + +def _is_mutation_method(method: str) -> bool: + return str(method or "").upper() in {"POST", "PUT", "PATCH", "DELETE"} + + +def _is_business_change(session: Session, instance: object) -> bool: + if isinstance(instance, OperationLog): + session.info[AUDIT_WRITTEN_KEY] = True + return False + if instance in session.dirty and not session.is_modified(instance, include_collections=False): + return False + return True + + +def _collect_business_write_flag(session: Session) -> None: + if not session.info.get(AUDIT_ENFORCE_KEY): + return + if session.info.get(AUDIT_BUSINESS_WRITE_KEY): + return + changed = set(session.new).union(session.dirty).union(session.deleted) + if any(_is_business_change(session, row) for row in changed): + session.info[AUDIT_BUSINESS_WRITE_KEY] = True + + +def _build_auto_audit_log(session: Session) -> OperationLog: + context = get_audit_context() + tenant_id = int(context.tenant_id or session.info.get(AUDIT_TENANT_KEY) or 1) + user_id = int(context.user_id or os.getenv("PMFLOW_AUDIT_SYSTEM_USER_ID", "1")) + path = str(context.interface_path or session.info.get(AUDIT_PATH_KEY) or "unknown")[:255] + method = str(context.method or session.info.get(AUDIT_METHOD_KEY) or "UNKNOWN").upper()[:16] + trace_id = str(context.trace_id or "").strip()[:64] or None + caller = str(context.caller or "system").strip().lower()[:16] or "system" + entity_id = trace_id or f"{path}:{method.lower()}" + return OperationLog( + tenant_id=tenant_id, + user_id=user_id, + project_id=None, + entity_type="interface_write", + entity_id=entity_id[:128], + action=f"auto:{method.lower()}", + before_data=None, + after_data={ + "method": method, + "path": path, + "trace_id": trace_id, + }, + remark="auto generated audit log for write transaction", + trace_id=trace_id, + interface_path=path, + caller=caller, + ) + + +@event.listens_for(Session, "before_flush") +def _mark_session_business_write(session: Session, _flush_context, _instances) -> None: # pragma: no cover + _collect_business_write_flag(session) + + +@event.listens_for(Session, "before_commit") +def _ensure_request_audit_log(session: Session) -> None: # pragma: no cover + if not session.info.get(AUDIT_ENFORCE_KEY): + return + if not session.info.get(AUDIT_BUSINESS_WRITE_KEY): + return + if session.info.get(AUDIT_WRITTEN_KEY): + return + session.add(_build_auto_audit_log(session)) + session.info[AUDIT_WRITTEN_KEY] = True + + +@event.listens_for(Session, "after_commit") +@event.listens_for(Session, "after_rollback") +def _clear_audit_flags(session: Session) -> None: # pragma: no cover + session.info.pop(AUDIT_BUSINESS_WRITE_KEY, None) + session.info.pop(AUDIT_WRITTEN_KEY, None) + AsyncSessionLocal = async_sessionmaker( bind=engine, class_=AsyncSession, @@ -69,7 +215,22 @@ AsyncSessionLocal = async_sessionmaker( ) -async def get_db() -> AsyncGenerator[AsyncSession, None]: +def _resolve_tenant_from_request(request: Request) -> int: + raw = str(request.headers.get("X-Tenant-ID") or "").strip() + if not raw: + return 1 + try: + return int(raw) + except ValueError: + return 1 + + +async def get_db(request: Request) -> AsyncGenerator[AsyncSession, None]: """FastAPI 依赖：提供数据库会话。""" async with AsyncSessionLocal() as session: + method = str(request.method or "").upper() + session.info[AUDIT_ENFORCE_KEY] = _is_mutation_method(method) + session.info[AUDIT_METHOD_KEY] = method[:16] + session.info[AUDIT_PATH_KEY] = str(request.url.path or "")[:255] or "/" + session.info[AUDIT_TENANT_KEY] = _resolve_tenant_from_request(request) yield session diff --git a/backend/app/dependencies/auth.py b/backend/app/dependencies/auth.py index d1198e73..d4bfcc85 100644 --- a/backend/app/dependencies/auth.py +++ b/backend/app/dependencies/auth.py @@ -15,9 +15,11 @@ from ..database import get_db from ..dependencies.tenant import get_tenant_id from ..models import User, UserStatus # noqa: E402 from ..services.auth import AuthService, has_any_role +from ..utils.audited_write import update_audit_context oauth2_scheme = OAuth2PasswordBearer(tokenUrl="auth/login") +oauth2_scheme_optional = OAuth2PasswordBearer(tokenUrl="auth/login", auto_error=False) async def get_current_user( @@ -41,6 +43,45 @@ async def get_current_user( if user is None: raise credentials_exception + try: + update_audit_context( + caller="user", + user_id=int(user.id), + tenant_id=int(getattr(user, "tenant_id", 1) or 1), + ) + except Exception: + pass + + return user + + +async def get_optional_current_user( + token: str | None = Depends(oauth2_scheme_optional), + db: AsyncSession = Depends(get_db), +) -> User | None: + """可选认证用户：无 token 时返回 None，有 token 时按 get_current_user 逻辑校验。""" + if not token: + return None + credentials_exception = HTTPException( + status_code=status.HTTP_401_UNAUTHORIZED, + detail="Could not validate credentials", + headers={"WWW-Authenticate": "Bearer"}, + ) + username = AuthService.verify_token(token) + if username is None: + raise credentials_exception + result = await db.execute(select(User).where(User.username == username)) + user = result.scalar_one_or_none() + if user is None: + raise credentials_exception + try: + update_audit_context( + caller="user", + user_id=int(user.id), + tenant_id=int(getattr(user, "tenant_id", 1) or 1), + ) + except Exception: + pass return user @@ -82,7 +123,10 @@ async def get_current_tenant_id( try: tenant_id = int(getattr(current_user, "tenant_id", 0) or 0) if tenant_id > 0: + update_audit_context(tenant_id=tenant_id) return tenant_id except (TypeError, ValueError): pass - return int(header_tenant_id or 1) + fallback = int(header_tenant_id or 1) + update_audit_context(tenant_id=fallback) + return fallback diff --git a/backend/app/dependencies/tenant.py b/backend/app/dependencies/tenant.py index 8f5c82ae..85d208f9 100644 --- a/backend/app/dependencies/tenant.py +++ b/backend/app/dependencies/tenant.py @@ -8,6 +8,8 @@ from typing import Annotated from fastapi import Header +from ..utils.audited_write import update_audit_context + async def get_tenant_id( x_tenant_id: Annotated[str | None, Header(alias="X-Tenant-ID")] = None, @@ -18,7 +20,11 @@ async def get_tenant_id( """ if x_tenant_id: try: - return int(x_tenant_id) + resolved = int(x_tenant_id) + update_audit_context(tenant_id=resolved) + return resolved except ValueError: + update_audit_context(tenant_id=1) return 1 + update_audit_context(tenant_id=1) return 1 diff --git a/backend/app/main.py b/backend/app/main.py index 0dbeee73..6f7be938 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -17,7 +17,8 @@ from .core.config import settings from .core.logging import configure_logging, get_logger from .core.response import success_response from .database import AsyncSessionLocal, engine -from .models import Base, Project, WorkItem +from .middleware import AuditContextMiddleware +from .models import Project, WorkItem from .services.scheduler_center import WorkloadRecalcScheduler from .services.scheduler_center import KnowledgeUrlRefreshScheduler from .services.scheduler_center import KnowledgeGraphIncrementalScheduler @@ -520,17 +521,20 @@ async def _warmup_world_models() -> None: @asynccontextmanager async def lifespan(_: FastAPI): """应用生命周期：初始化表与系统技能标签。""" - auto_create_tables = os.getenv("AUTO_CREATE_TABLES", "true").lower() in { + auto_create_tables = os.getenv("AUTO_CREATE_TABLES", "false").lower() in { "1", "true", "yes", } + if auto_create_tables: + logger.error( + "runtime_ddl_blocked", + auto_create_tables=auto_create_tables, + reason="runtime ddl is forbidden, run alembic migration instead", + ) + raise RuntimeError("AUTO_CREATE_TABLES=true 已被禁用；请使用 Alembic migration 变更数据库。") async with engine.begin() as conn: await conn.run_sync(_check_database_dialect_guard) - if auto_create_tables: - await conn.run_sync(Base.metadata.create_all) - await conn.run_sync(_sqlite_backfill_columns) - await conn.run_sync(_ensure_feishu_integrity_guards) async with AsyncSessionLocal() as db: try: @@ -591,6 +595,8 @@ app = FastAPI( lifespan=lifespan, ) +app.add_middleware(AuditContextMiddleware) + origins = [item.strip() for item in settings.cors_origins.split(",") if item.strip()] app.add_middleware( CORSMiddleware, diff --git a/backend/app/middleware/__init__.py b/backend/app/middleware/__init__.py new file mode 100644 index 00000000..6080a9fb --- /dev/null +++ b/backend/app/middleware/__init__.py @@ -0,0 +1,5 @@ +"""应用中间件。""" + +from .audit_context import AuditContextMiddleware + +__all__ = ["AuditContextMiddleware"] diff --git a/backend/app/middleware/audit_context.py b/backend/app/middleware/audit_context.py new file mode 100644 index 00000000..ee4cb0e0 --- /dev/null +++ b/backend/app/middleware/audit_context.py @@ -0,0 +1,44 @@ +"""请求审计上下文中间件。""" + +from __future__ import annotations + +from uuid import uuid4 + +from starlette.middleware.base import BaseHTTPMiddleware +from starlette.requests import Request +from starlette.responses import Response + +from ..utils.audited_write import AuditRequestContext, normalize_caller, reset_audit_context, set_audit_context + + +class AuditContextMiddleware(BaseHTTPMiddleware): + """为每个请求注入 trace/caller/path 审计上下文。""" + + @staticmethod + def _infer_caller(request: Request) -> str: + caller_header = request.headers.get("X-Caller") + if caller_header: + return normalize_caller(caller_header) + path = str(request.url.path or "") + if path.startswith(("/webhooks", "/feishu", "/wecom")): + return "ai" + if request.headers.get("Authorization"): + return "user" + return "system" + + async def dispatch(self, request: Request, call_next) -> Response: + trace_id = str(request.headers.get("X-Trace-Id") or f"req-{uuid4().hex[:24]}")[:64] + context = AuditRequestContext( + trace_id=trace_id, + interface_path=str(request.url.path or "")[:255] or "/", + caller=self._infer_caller(request), + method=str(request.method or "").upper()[:16], + ) + token = set_audit_context(context) + request.state.trace_id = trace_id + try: + response = await call_next(request) + finally: + reset_audit_context(token) + response.headers["X-Trace-Id"] = trace_id + return response diff --git a/backend/app/models/operation_log.py b/backend/app/models/operation_log.py index fbc144a5..9d4714c0 100644 --- a/backend/app/models/operation_log.py +++ b/backend/app/models/operation_log.py @@ -1,5 +1,6 @@ """操作日志模型（全链路可追溯）""" +from datetime import datetime from typing import Optional from sqlalchemy import BigInteger, Integer, ForeignKey, Text, VARCHAR @@ -7,6 +8,7 @@ from sqlalchemy.orm import Mapped, mapped_column from .base import Base from .types import JSONB +from ..utils.audited_write import format_beijing_datetime class OperationLog(Base): @@ -25,3 +27,10 @@ class OperationLog(Base): before_data: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) after_data: Mapped[Optional[dict]] = mapped_column(JSONB, nullable=True) remark: Mapped[Optional[str]] = mapped_column(Text, nullable=True) + trace_id: Mapped[Optional[str]] = mapped_column(VARCHAR(64), nullable=True, index=True) + interface_path: Mapped[Optional[str]] = mapped_column(VARCHAR(255), nullable=True, index=True) + caller: Mapped[str] = mapped_column(VARCHAR(16), nullable=False, default="system", index=True) + + @property + def created_at_beijing(self) -> Optional[str]: + return format_beijing_datetime(self.created_at if isinstance(self.created_at, datetime) else None) diff --git a/backend/app/models/user_workload.py b/backend/app/models/user_workload.py index 52176883..7c86d458 100644 --- a/backend/app/models/user_workload.py +++ b/backend/app/models/user_workload.py @@ -18,7 +18,7 @@ class UserWorkload(BaseNoCreatedAt): DECIMAL(5, 2), nullable=False, default=40.0 ) workload_ratio: Mapped[float] = mapped_column( - DECIMAL(4, 2), nullable=False, default=0.0 + DECIMAL(10, 2), nullable=False, default=0.0 ) user = relationship("User", back_populates="workload_rel") diff --git a/backend/app/services/ai_pmo_kernel_service.py b/backend/app/services/ai_pmo_kernel_service.py index 90d1e534..b1d8270a 100644 --- a/backend/app/services/ai_pmo_kernel_service.py +++ b/backend/app/services/ai_pmo_kernel_service.py @@ -203,7 +203,12 @@ SHORT_CONFIRM_HINTS = ( "可以", "没问题", ) -INTENT_LLM_TIMEOUT_SECONDS = 8.0 +INTENT_LLM_TIMEOUT_DEFAULT_SECONDS = 30.0 +INTENT_LLM_TIMEOUT_COMPLEX_DEFAULT_SECONDS = 30.0 +INTENT_LLM_TIMEOUT_FALLBACK_DEFAULT_SECONDS = 35.0 +INTENT_LLM_TOTAL_BUDGET_DEFAULT_SECONDS = 55.0 +INTENT_LLM_STAGE2_CONFIDENCE_DEFAULT = 0.72 +INTENT_LLM_STAGE2_MIN_REMAINING_DEFAULT_SECONDS = 12.0 INTENT_LLM_FUNCTION_NAME = "classify_intent_ai_pmo_kernel" INTENT_CANDIDATES = ( INTENT_SIGNAL_SOURCE_ONBOARD, @@ -238,6 +243,83 @@ def _env_bool(name: str, default: bool = False) -> bool: return str(raw).strip().lower() in BOOL_TRUE_VALUES +def _env_float(name: str, default: float) -> float: + raw = os.getenv(name) + if raw is None: + return float(default) + try: + return float(raw) + except (TypeError, ValueError): + return float(default) + + +def _intent_llm_timeout_seconds(*, function_name: str) -> float: + normalized = str(function_name or "").strip().lower() + if normalized == "classify_intent_complex": + default_timeout = INTENT_LLM_TIMEOUT_COMPLEX_DEFAULT_SECONDS + configured = _env_float( + "AI_PMO_INTENT_LLM_TIMEOUT_COMPLEX_SECONDS", + _env_float("AI_PMO_INTENT_LLM_TIMEOUT_SECONDS", default_timeout), + ) + elif normalized == str(INTENT_LLM_FUNCTION_NAME or "").strip().lower(): + default_timeout = INTENT_LLM_TIMEOUT_FALLBACK_DEFAULT_SECONDS + configured = _env_float( + "AI_PMO_INTENT_LLM_TIMEOUT_FALLBACK_SECONDS", + _env_float("AI_PMO_INTENT_LLM_TIMEOUT_SECONDS", default_timeout), + ) + else: + default_timeout = INTENT_LLM_TIMEOUT_DEFAULT_SECONDS + configured = _env_float("AI_PMO_INTENT_LLM_TIMEOUT_SECONDS", default_timeout) + return min(max(float(configured), 12.0), 120.0) + + +def _intent_llm_total_budget_seconds() -> float: + configured = _env_float("AI_PMO_INTENT_LLM_TOTAL_BUDGET_SECONDS", INTENT_LLM_TOTAL_BUDGET_DEFAULT_SECONDS) + return min(max(float(configured), 20.0), 120.0) + + +def _intent_llm_stage2_confidence_threshold() -> float: + configured = _env_float("AI_PMO_INTENT_LLM_STAGE2_CONFIDENCE_THRESHOLD", INTENT_LLM_STAGE2_CONFIDENCE_DEFAULT) + return min(max(float(configured), 0.40), 0.95) + + +def _intent_llm_stage2_min_remaining_seconds() -> float: + configured = _env_float( + "AI_PMO_INTENT_LLM_STAGE2_MIN_REMAINING_SECONDS", + INTENT_LLM_STAGE2_MIN_REMAINING_DEFAULT_SECONDS, + ) + return min(max(float(configured), 8.0), 30.0) + + +def _should_force_low_tier_intent_route(*, message_text: str, correction_action: str) -> bool: + if str(correction_action or "").strip().lower() != "none": + return True + normalized = re.sub(r"\s+", "", str(message_text or "")).lower() + if not normalized: + return False + return any(token in normalized for token in TRANSPORT_MODE_CORRECTION_HINTS) + + +def _should_escalate_to_high_intent_route( + *, + route_is_complex: bool, + low_intent: Optional["CanonicalIntent"], + rule_intent: Optional["CanonicalIntent"], +) -> tuple[bool, list[str]]: + reasons: list[str] = [] + if route_is_complex: + reasons.append("route_complex") + if low_intent is None: + reasons.append("stage1_empty") + else: + low_confidence = float(low_intent.confidence or 0.0) + if low_confidence < _intent_llm_stage2_confidence_threshold(): + reasons.append("stage1_low_confidence") + if rule_intent is not None and low_intent is not None and str(rule_intent.name or "") != str(low_intent.name or ""): + reasons.append("stage1_rule_conflict") + return bool(reasons), reasons + + def _safe_int(value: Any) -> Optional[int]: try: if value is None or value == "": @@ -2155,6 +2237,7 @@ class AIPMOKernelService: message_text: str, session: Optional[ConversationSession], world_model: dict[str, Any], + rule_intent: Optional[CanonicalIntent] = None, ) -> Optional[CanonicalIntent]: project_snapshot = world_model.get("project") if isinstance(world_model.get("project"), dict) else {} project_names = [ @@ -2163,6 +2246,10 @@ class AIPMOKernelService: ] route_decision = select_intent_route(message_text, project_names=project_names) correction_decision = kernel_resolve_correction(message_text) + forced_low_tier = _should_force_low_tier_intent_route( + message_text=message_text, + correction_action=correction_decision.action, + ) prompt = PROMPT_AI_PMO_INTENT_CLASSIFICATION.format( message_text=str(message_text or "")[:1200], session_last_intent=str(getattr(session, "last_intent", "") or "").strip() if session is not None else "", @@ -2170,84 +2257,190 @@ class AIPMOKernelService: world_model_project_id=_safe_int(world_model.get("resolved_project_id")), candidate_intents_json=json.dumps(list(INTENT_CANDIDATES), ensure_ascii=False), ).strip() - try: - extraction = await asyncio.wait_for( - llm_service.generate_structured_object( - prompt, - db=self.db, + total_budget_seconds = _intent_llm_total_budget_seconds() + stage_started_at = time.perf_counter() + + async def _invoke_intent_stage( + *, + function_name: str, + route_tier: str, + route_reasons: list[str], + stage_label: str, + timeout_cap_seconds: Optional[float] = None, + ) -> Optional[CanonicalIntent]: + llm_timeout_seconds = _intent_llm_timeout_seconds(function_name=function_name) + if timeout_cap_seconds is not None: + llm_timeout_seconds = min(llm_timeout_seconds, max(float(timeout_cap_seconds), 0.0)) + if llm_timeout_seconds <= 0.0: + return None + logger.info( + "ai_pmo_intent_llm_route", + tenant_id=self.tenant_id, + stage=stage_label, + function_name=function_name, + route_tier=route_tier, + timeout_seconds=llm_timeout_seconds, + forced_low_tier=forced_low_tier, + correction_action=correction_decision.action, + correction_keywords=correction_decision.matched_keywords, + route_reasons=route_reasons, + total_budget_seconds=total_budget_seconds, + ) + try: + extraction = await asyncio.wait_for( + llm_service.generate_structured_object( + prompt, + db=self.db, + tenant_id=self.tenant_id, + function_name=function_name, + use_prompt_profile=False, + temperature=0.0, + max_tokens=260, + response_mode="fast", + ), + timeout=llm_timeout_seconds, + ) + except TimeoutError: + logger.warning( + "ai_pmo_intent_llm_timeout", tenant_id=self.tenant_id, - function_name=route_decision.function_name, - use_prompt_profile=False, - temperature=0.0, - max_tokens=260, - response_mode="fast", - ), - timeout=INTENT_LLM_TIMEOUT_SECONDS, + stage=stage_label, + function_name=function_name, + timeout_seconds=llm_timeout_seconds, + ) + return None + except Exception as exc: + logger.warning( + "ai_pmo_intent_llm_failed", + tenant_id=self.tenant_id, + stage=stage_label, + function_name=function_name, + error=str(exc), + ) + return None + + payload = extraction.payload if extraction and isinstance(extraction.payload, dict) else None + if payload is None: + logger.warning( + "ai_pmo_intent_llm_empty_payload", + tenant_id=self.tenant_id, + stage=stage_label, + function_name=function_name, + error_code=str(getattr(extraction, "error_code", "") or ""), + provider=str(getattr(extraction, "provider", "") or ""), + model=str(getattr(extraction, "model", "") or ""), + ) + return None + + intent_name = str(payload.get("intent") or "").strip() + if intent_name not in INTENT_CANDIDATES: + return None + confidence = float(payload.get("confidence") or 0.0) + project_id = _safe_int(payload.get("project_id")) + if project_id is not None and int(project_id) <= 0: + project_id = None + if project_id is None: + project_id = _safe_int(world_model.get("resolved_project_id")) + reasoning = str(payload.get("reasoning") or "").strip() + decision_trace = build_decision_trace( + intent=intent_name, + confidence=confidence, + route_tier=route_tier, + route_function=function_name, + gate_action="unresolved", + gate_code="intent_only", + correction_action=correction_decision.action, + correction_keywords=correction_decision.matched_keywords, + complexity_reasons=route_reasons, ) - except TimeoutError: - logger.warning( - "ai_pmo_intent_llm_timeout", + return self._build_canonical_intent( + intent_name=intent_name, + message_text=message_text, + project_id=project_id, + confidence=confidence, + metadata={ + "context_source": "llm", + "stage": stage_label, + "reasoning": reasoning[:300], + "llm_function": function_name, + "route_tier": route_tier, + "route_complex": route_tier == "high", + "route_reasons": route_reasons, + "forced_low_tier": forced_low_tier, + "timeout_seconds": llm_timeout_seconds, + "total_budget_seconds": total_budget_seconds, + "correction_action": correction_decision.action, + "correction_keywords": correction_decision.matched_keywords, + "decision_trace": decision_trace, + }, + ) + + low_stage_reasons = ["stage1_low_first", *list(route_decision.reasons or [])] + if forced_low_tier: + low_stage_reasons.append("correction_forced_low_tier") + low_intent = await _invoke_intent_stage( + function_name=INTENT_LLM_FUNCTION_NAME, + route_tier="low", + route_reasons=low_stage_reasons, + stage_label="stage1_low", + ) + if forced_low_tier: + return low_intent + + should_escalate, escalate_reasons = _should_escalate_to_high_intent_route( + route_is_complex=bool(route_decision.is_complex), + low_intent=low_intent, + rule_intent=rule_intent, + ) + if not should_escalate: + return low_intent + + elapsed_seconds = time.perf_counter() - stage_started_at + remaining_budget_seconds = total_budget_seconds - elapsed_seconds + stage2_min_remaining_seconds = _intent_llm_stage2_min_remaining_seconds() + if remaining_budget_seconds < stage2_min_remaining_seconds: + logger.info( + "ai_pmo_intent_stage2_skipped", tenant_id=self.tenant_id, - timeout_seconds=INTENT_LLM_TIMEOUT_SECONDS, - ) - return None - except Exception as exc: - logger.warning( - "ai_pmo_intent_llm_failed", - tenant_id=self.tenant_id, - error=str(exc), + reason="budget_insufficient", + remaining_budget_seconds=max(0.0, remaining_budget_seconds), + min_remaining_seconds=stage2_min_remaining_seconds, + escalate_reasons=escalate_reasons, ) + if low_intent is not None: + return self._copy_intent_with_metadata( + low_intent, + stage2_skipped=True, + stage2_skip_reason="budget_insufficient", + stage2_reasons=escalate_reasons, + ) return None - payload = extraction.payload if extraction and isinstance(extraction.payload, dict) else None - if payload is None: - logger.warning( - "ai_pmo_intent_llm_empty_payload", - tenant_id=self.tenant_id, - error_code=str(getattr(extraction, "error_code", "") or ""), - provider=str(getattr(extraction, "provider", "") or ""), - model=str(getattr(extraction, "model", "") or ""), + high_stage_reasons = [*list(route_decision.reasons or []), *escalate_reasons, "stage2_high_retry"] + high_intent = await _invoke_intent_stage( + function_name="classify_intent_complex", + route_tier="high", + route_reasons=high_stage_reasons, + stage_label="stage2_high", + timeout_cap_seconds=remaining_budget_seconds, + ) + if high_intent is not None: + if low_intent is not None: + return self._copy_intent_with_metadata( + high_intent, + stage2_promoted=True, + stage1_intent=low_intent.name, + stage1_confidence=float(low_intent.confidence or 0.0), + ) + return high_intent + if low_intent is not None: + return self._copy_intent_with_metadata( + low_intent, + stage2_attempted=True, + stage2_failed=True, + stage2_reasons=escalate_reasons, ) - return None - - intent_name = str(payload.get("intent") or "").strip() - if intent_name not in INTENT_CANDIDATES: - return None - confidence = float(payload.get("confidence") or 0.0) - project_id = _safe_int(payload.get("project_id")) - if project_id is not None and int(project_id) <= 0: - project_id = None - if project_id is None: - project_id = _safe_int(world_model.get("resolved_project_id")) - reasoning = str(payload.get("reasoning") or "").strip() - decision_trace = build_decision_trace( - intent=intent_name, - confidence=confidence, - route_tier=route_decision.tier, - route_function=route_decision.function_name, - gate_action="unresolved", - gate_code="intent_only", - correction_action=correction_decision.action, - correction_keywords=correction_decision.matched_keywords, - complexity_reasons=route_decision.reasons, - ) - return self._build_canonical_intent( - intent_name=intent_name, - message_text=message_text, - project_id=project_id, - confidence=confidence, - metadata={ - "context_source": "llm", - "reasoning": reasoning[:300], - "llm_function": route_decision.function_name, - "route_tier": route_decision.tier, - "route_complex": route_decision.is_complex, - "route_reasons": route_decision.reasons, - "correction_action": correction_decision.action, - "correction_keywords": correction_decision.matched_keywords, - "decision_trace": decision_trace, - }, - ) + return None async def _resolve_intent( self, @@ -2275,6 +2468,7 @@ class AIPMOKernelService: message_text=message_text, session=session, world_model=world_model, + rule_intent=rule_intent, ) if llm_intent is not None: if self._intent_arbitration_enabled(): @@ -2286,6 +2480,11 @@ class AIPMOKernelService: else: selected_intent, source = llm_intent, "llm" return self._apply_slot_guard(intent=selected_intent, message_text=message_text), source + logger.info( + "ai_pmo_intent_rule_fallback", + tenant_id=self.tenant_id, + reason="intent_llm_empty_or_failed_or_timeout", + ) return self._apply_slot_guard(intent=rule_intent, message_text=message_text), "rule_fallback" diff --git a/backend/app/services/ai_steward.py b/backend/app/services/ai_steward.py index a6cbdf92..a8f579a3 100644 --- a/backend/app/services/ai_steward.py +++ b/backend/app/services/ai_steward.py @@ -4,7 +4,9 @@ from __future__ import annotations import asyncio import json +import os import re +import time from typing import Any, Dict, List from sqlalchemy.ext.asyncio import AsyncSession @@ -25,7 +27,47 @@ from .task_persistence import ( ) logger = get_logger(__name__) -PROJECT_CREATION_DRAFT_TIMEOUT_SECONDS = 8.0 + + +def _env_float(name: str, default: float) -> float: + raw = os.getenv(name) + if raw is None: + return float(default) + try: + return float(str(raw).strip()) + except Exception: + return float(default) + + +def _env_int(name: str, default: int) -> int: + raw = os.getenv(name) + if raw is None: + return int(default) + try: + return int(str(raw).strip()) + except Exception: + return int(default) + + +PROJECT_CREATION_DRAFT_TIMEOUT_SECONDS = max( + 8.0, + _env_float("AI_STEWARD_PROJECT_CREATION_STEP_TIMEOUT_SECONDS", 18.0), +) +PROJECT_CREATION_STEP_RETRY_COUNT = max( + 0, + _env_int("AI_STEWARD_PROJECT_CREATION_STEP_RETRY_COUNT", 1), +) +PROJECT_CREATION_STEP_FALLBACK_FUNCTIONS = { + "project_from_text_analysis": "project_draft_analysis_glm47_fallback", + "project_from_text_signal": "project_draft_signal_glm47_fallback", + "project_from_text_charter": "project_draft_charter_glm47_fallback", + "project_from_text_task_chain": "project_draft_task_chain_glm47_fallback", +} +PROJECT_CREATION_STUB_MARKERS = ( + "我暂时无法生成可靠回答", + "模型服务异常", + "请稍后重试", +) class AIStewardService: @@ -38,6 +80,14 @@ class AIStewardService: @staticmethod def _normalize_project_title_candidate(candidate: str) -> str: cleaned = re.sub(r"\s+", " ", str(candidate or "")).strip(" []【】()（）:：\"'“”") + if not cleaned: + return "" + cleaned = re.sub( + r"^(?:(?:项目|需求)(?:名字|名称|名)?(?:叫|是|为|名为|名称为|名字为)?|(?:项目|需求)?(?:名为|名称为|名字为|名称是|名字是|叫|为))\s*", + "", + cleaned, + flags=re.IGNORECASE, + ).strip(" []【】()（）:：\"'“”") if not cleaned: return "" normalized = re.sub(r"\s+", "", cleaned).lower() @@ -67,14 +117,20 @@ class AIStewardService: if quoted_match: return cls._normalize_project_title_candidate(quoted_match.group(1)) for pattern in ( + r"(?:项目|需求)(?:名字|名称|名)?(?:叫|名为|名称是|名称为|是|为)[：:，,\s]*(.+)", r"(?:帮我|请|麻烦|给我)?(?:新增|增加|添加|创建|新建|建立|立项|开个|开一个|建个|建一个|做个|做一个|加个|加一个)(?:一个|个)?项目(?:叫|名为|名称是|名称为|是|为)?[：:，,\s]*(.+)", - r"项目(?:叫|名为|名称是|名称为)[：:，,\s]*(.+)", ): match = re.search(pattern, cleaned, flags=re.IGNORECASE) if not match: continue candidate = re.sub(r"[，。,.;；!?？].*$", "", match.group(1)).strip(" []【】()（）:：\"'“”") - candidate = re.sub(r"^(?:一个|个|新项目|项目)\s*", "", candidate, flags=re.IGNORECASE) + candidate = re.sub(r"^(?:一个|个|新项目|项目|需求)\s*", "", candidate, flags=re.IGNORECASE) + candidate = re.sub( + r"^(?:(?:项目|需求)(?:名字|名称|名)?(?:叫|是|为|名为|名称为|名字为)?|(?:项目|需求)?(?:名为|名称为|名字为|名称是|名字是|叫|为))\s*", + "", + candidate, + flags=re.IGNORECASE, + ) candidate = re.sub(r"(?:因为|所以|用来|为了|方便).*$", "", candidate).strip() normalized = cls._normalize_project_title_candidate(candidate) if normalized: @@ -187,6 +243,24 @@ class AIStewardService: "risks": self._build_risk_alert_drafts(analysis=analysis), } + @staticmethod + def _looks_like_project_creation_stub_response(text: str) -> bool: + normalized = re.sub(r"\s+", "", str(text or "")).lower() + if not normalized: + return True + return any(marker in normalized for marker in PROJECT_CREATION_STUB_MARKERS) + + @staticmethod + def _project_creation_function_candidates(function_name: str) -> list[str]: + base = str(function_name or "").strip() + if not base: + return [] + candidates = [base] + fallback = str(PROJECT_CREATION_STEP_FALLBACK_FUNCTIONS.get(base) or "").strip() + if fallback and fallback != base: + candidates.append(fallback) + return candidates + async def _generate_project_creation_text( self, *, @@ -195,24 +269,62 @@ class AIStewardService: function_name: str, timeout_seconds: float = PROJECT_CREATION_DRAFT_TIMEOUT_SECONDS, ) -> str: - try: - return await asyncio.wait_for( - self.llm.generate( - prompt, - db=self.db, - tenant_id=tenant_id, - function_name=function_name, - ), - timeout=timeout_seconds, - ) - except Exception as exc: - logger.warning( - "ai_steward_project_creation_step_fallback", - tenant_id=int(tenant_id), - function_name=function_name, - error=str(exc), - ) + candidates = self._project_creation_function_candidates(function_name=function_name) + if not candidates: return "" + attempts_per_function = PROJECT_CREATION_STEP_RETRY_COUNT + 1 + for candidate_function in candidates: + for attempt_index in range(attempts_per_function): + started_at = time.perf_counter() + try: + response = await asyncio.wait_for( + self.llm.generate( + prompt, + db=self.db, + tenant_id=tenant_id, + function_name=candidate_function, + response_mode="fast", + ), + timeout=timeout_seconds, + ) + except Exception as exc: + logger.warning( + "ai_steward_project_creation_step_retry", + tenant_id=int(tenant_id), + function_name=candidate_function, + attempt=attempt_index + 1, + elapsed_seconds=round(time.perf_counter() - started_at, 2), + error=str(exc), + ) + continue + + cleaned_response = str(response or "").strip() + elapsed_seconds = round(time.perf_counter() - started_at, 2) + if self._looks_like_project_creation_stub_response(cleaned_response): + logger.warning( + "ai_steward_project_creation_step_stub_retry", + tenant_id=int(tenant_id), + function_name=candidate_function, + attempt=attempt_index + 1, + elapsed_seconds=elapsed_seconds, + ) + continue + logger.info( + "ai_steward_project_creation_step_ok", + tenant_id=int(tenant_id), + function_name=candidate_function, + attempt=attempt_index + 1, + elapsed_seconds=elapsed_seconds, + ) + return cleaned_response + + logger.warning( + "ai_steward_project_creation_step_fallback", + tenant_id=int(tenant_id), + function_name=function_name, + error="all_attempts_exhausted", + ) + return "" async def _normalize_project_creation_draft( self, diff --git a/backend/app/services/audit_query_service.py b/backend/app/services/audit_query_service.py new file mode 100644 index 00000000..d82196d6 --- /dev/null +++ b/backend/app/services/audit_query_service.py @@ -0,0 +1,74 @@ +"""审计日志查询服务。""" + +from __future__ import annotations + +from datetime import datetime +from typing import Any, Optional + +from sqlalchemy import and_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..models import OperationLog +from ..utils.audited_write import format_beijing_datetime +from ..utils.timezone import CN_TZ + + +def _to_beijing_naive(value: Optional[datetime]) -> Optional[datetime]: + if value is None: + return None + if value.tzinfo is None: + return value + return value.astimezone(CN_TZ).replace(tzinfo=None) + + +def _serialize_log(row: OperationLog) -> dict[str, Any]: + return { + "id": int(row.id), + "tenant_id": int(row.tenant_id), + "user_id": int(row.user_id), + "project_id": int(row.project_id) if row.project_id is not None else None, + "entity_type": str(row.entity_type or ""), + "entity_id": str(row.entity_id or ""), + "action": str(row.action or ""), + "trace_id": str(row.trace_id or ""), + "interface_path": str(row.interface_path or ""), + "caller": str(row.caller or "system"), + "remark": str(row.remark or ""), + "before_data": row.before_data if isinstance(row.before_data, dict) else None, + "after_data": row.after_data if isinstance(row.after_data, dict) else None, + "created_at": row.created_at.isoformat() if row.created_at else None, + "created_at_beijing": row.created_at_beijing or format_beijing_datetime(row.created_at), + } + + +class AuditQueryService: + def __init__(self, db: AsyncSession) -> None: + self.db = db + + async def list_operations( + self, + *, + tenant_id: int, + entity_type: Optional[str], + start_time: Optional[datetime], + end_time: Optional[datetime], + limit: int = 100, + ) -> list[dict[str, Any]]: + filters = [OperationLog.tenant_id == int(tenant_id)] + if entity_type: + filters.append(OperationLog.entity_type == str(entity_type).strip()[:64]) + normalized_start = _to_beijing_naive(start_time) + normalized_end = _to_beijing_naive(end_time) + if normalized_start is not None: + filters.append(OperationLog.created_at >= normalized_start) + if normalized_end is not None: + filters.append(OperationLog.created_at <= normalized_end) + + query = ( + select(OperationLog) + .where(and_(*filters)) + .order_by(OperationLog.created_at.desc(), OperationLog.id.desc()) + .limit(max(1, min(int(limit), 100))) + ) + rows = (await self.db.execute(query)).scalars().all() + return [_serialize_log(row) for row in rows] diff --git a/backend/app/services/codegen.py b/backend/app/services/codegen.py index bbf6641c..48e10d07 100644 --- a/backend/app/services/codegen.py +++ b/backend/app/services/codegen.py @@ -407,7 +407,8 @@ IMPORTANT: ) # 对于修改的文件 - old_content = file_contexts.get(file.path)?.content or "" + old_file_info = file_contexts.get(file.path) + old_content = old_file_info.content if old_file_info is not None else "" old_lines = old_content.split("\n") new_lines = file.content.split("\n") diff --git a/backend/app/services/llm.py b/backend/app/services/llm.py index 9a1c6386..4450dcc5 100644 --- a/backend/app/services/llm.py +++ b/backend/app/services/llm.py @@ -116,6 +116,7 @@ DEFAULT_LOW_TIER_HINTS = ( "openclaw_command_dispatch", "signal_source_ai_worker_execution", "extract_context_insights", + "project_draft_", "adaptive_rag_answer_normal", "rag_intent_classification", "classify_intent", diff --git a/backend/app/services/openclaw_commander_service.py b/backend/app/services/openclaw_commander_service.py index 1a10e41b..85c418ce 100644 --- a/backend/app/services/openclaw_commander_service.py +++ b/backend/app/services/openclaw_commander_service.py @@ -790,6 +790,14 @@ class OpenClawCommanderService: @staticmethod def _normalize_project_title_candidate(candidate: str) -> str: cleaned = re.sub(r"\s+", " ", str(candidate or "")).strip(" []【】()（）:：\"'“”") + if not cleaned: + return "" + cleaned = re.sub( + r"^(?:(?:项目|需求)(?:名字|名称|名)?(?:叫|是|为|名为|名称为|名字为)?|(?:项目|需求)?(?:名为|名称为|名字为|名称是|名字是|叫|为))\s*", + "", + cleaned, + flags=re.IGNORECASE, + ).strip(" []【】()（）:：\"'“”") if not cleaned: return "" normalized = re.sub(r"\s+", "", cleaned).lower() @@ -820,15 +828,21 @@ class OpenClawCommanderService: if quoted_match: return OpenClawCommanderService._normalize_project_title_candidate(quoted_match.group(1)) for pattern in ( + r"(?:项目|需求)(?:名字|名称|名)?(?:叫|名为|名称是|名称为|是|为)[：:，,\s]*(.+)", r"(?:帮我|请|麻烦|给我)?(?:新增|增加|添加|创建|新建|建立|开个|开一个|建个|建一个|做个|做一个|加个|加一个)(?:一个|个)?项目(?:叫|名为|名称是|名称为|是|为)?[：:，,\s]*(.+)", r"(?:新增|增加|添加|创建|新建|建立)项目[：:，,\s]*(.+)", - r"项目(?:叫|名为|名称是|名称为)[：:，,\s]*(.+)", ): match = re.search(pattern, cleaned, re.IGNORECASE) if not match: continue candidate = re.sub(r"[，。,.;；!?？].*$", "", match.group(1)).strip(" []【】()（）:：\"'“”") - candidate = re.sub(r"^(?:一个|个|新项目)\s*", "", candidate, flags=re.IGNORECASE) + candidate = re.sub(r"^(?:一个|个|新项目|项目|需求)\s*", "", candidate, flags=re.IGNORECASE) + candidate = re.sub( + r"^(?:(?:项目|需求)(?:名字|名称|名)?(?:叫|是|为|名为|名称为|名字为)?|(?:项目|需求)?(?:名为|名称为|名字为|名称是|名字是|叫|为))\s*", + "", + candidate, + flags=re.IGNORECASE, + ) if candidate: return OpenClawCommanderService._normalize_project_title_candidate(candidate) return "" diff --git a/backend/app/services/operation_log_service.py b/backend/app/services/operation_log_service.py index 75819850..8301a534 100644 --- a/backend/app/services/operation_log_service.py +++ b/backend/app/services/operation_log_service.py @@ -5,6 +5,7 @@ from typing import Optional from sqlalchemy.ext.asyncio import AsyncSession from ..models import OperationLog +from ..utils.audited_write import get_audit_context, normalize_caller async def log_operation( @@ -19,9 +20,16 @@ async def log_operation( before_data: Optional[dict] = None, after_data: Optional[dict] = None, remark: Optional[str] = None, + trace_id: Optional[str] = None, + interface_path: Optional[str] = None, + caller: Optional[str] = None, ) -> None: """记录操作日志，失败时不影响主流程。""" try: + context = get_audit_context() + resolved_trace = str(trace_id or context.trace_id or "").strip()[:64] or None + resolved_interface = str(interface_path or context.interface_path or "").strip()[:255] or None + resolved_caller = normalize_caller(caller or context.caller or "system") db.add( OperationLog( tenant_id=tenant_id, @@ -33,8 +41,12 @@ async def log_operation( before_data=before_data, after_data=after_data, remark=remark, + trace_id=resolved_trace, + interface_path=resolved_interface, + caller=resolved_caller, ) ) + db.sync_session.info["_pmflow_audit_written"] = True except Exception: # 日志失败不阻断主交易 return diff --git a/backend/app/services/project_member_service.py b/backend/app/services/project_member_service.py new file mode 100644 index 00000000..4c29d515 --- /dev/null +++ b/backend/app/services/project_member_service.py @@ -0,0 +1,240 @@ +"""项目成员分配服务。""" + +from __future__ import annotations + +import re +from typing import Iterable, Sequence + +from sqlalchemy import and_, delete, select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..models import User, UserProject +from .notification_service import NotificationService + +_ROLE_SPLIT_RE = re.compile(r"[,/|\s]+") +_ROLE_ALIAS_MAP = { + "AI": {"AI"}, + "AI_ENGINEER": {"AI"}, + "FRONTEND": {"FE"}, + "FE": {"FE"}, + "BACKEND": {"BE"}, + "BE": {"BE"}, + "BOSS": {"BOSS"}, + "ADMIN": {"ADMIN"}, +} + + +class ProjectMemberAssignmentError(ValueError): + """项目成员分配失败。""" + + def __init__(self, message: str, *, invalid_user_ids: Sequence[int] | None = None) -> None: + super().__init__(message) + self.invalid_user_ids = [int(uid) for uid in (invalid_user_ids or [])] + + +def _normalize_user_ids(user_ids: Iterable[int] | None) -> list[int]: + normalized: set[int] = set() + for raw in user_ids or []: + if raw is None: + continue + try: + normalized.add(int(raw)) + except (TypeError, ValueError): + continue + return sorted(normalized) + + +def _parse_role_tokens(raw: str | None) -> set[str]: + tokens = { + token.strip().upper() + for token in _ROLE_SPLIT_RE.split(str(raw or "").strip()) + if token.strip() + } + normalized: set[str] = set() + for token in tokens: + normalized.update(_ROLE_ALIAS_MAP.get(token, {token})) + return normalized + + +def _user_matches_project_role(user_role: str | None, expected_roles: set[str]) -> bool: + if not expected_roles: + return False + return bool(_parse_role_tokens(user_role).intersection(expected_roles)) + + +async def _load_active_user_ids( + db: AsyncSession, + *, + tenant_id: int, + candidate_user_ids: Sequence[int] | None = None, +) -> set[int]: + query = select(User.id).where( + and_( + User.tenant_id == int(tenant_id), + User.status == "active", + User.is_active.is_(True), + ) + ) + normalized_ids = _normalize_user_ids(candidate_user_ids) + if normalized_ids: + query = query.where(User.id.in_(normalized_ids)) + rows = await db.execute(query) + return {int(uid) for uid in rows.scalars().all() if uid is not None} + + +async def _load_role_member_ids( + db: AsyncSession, + *, + tenant_id: int, + assigned_role: str | None, +) -> set[int]: + expected_roles = _parse_role_tokens(assigned_role) + if not expected_roles: + return set() + + rows = await db.execute( + select(User.id, User.role).where( + and_( + User.tenant_id == int(tenant_id), + User.status == "active", + User.is_active.is_(True), + ) + ) + ) + role_member_ids: set[int] = set() + for user_id, user_role in rows.all(): + if _user_matches_project_role(user_role, expected_roles): + role_member_ids.add(int(user_id)) + return role_member_ids + + +async def resolve_project_member_ids( + db: AsyncSession, + *, + tenant_id: int, + owner_id: int | None, + assigned_role: str | None, + requested_member_user_ids: Sequence[int] | None = None, + require_requested_members: bool = False, + include_owner: bool = True, +) -> list[int]: + """解析项目成员列表，支持显式成员、角色兜底与 owner 兜底。""" + member_ids: set[int] = set() + requested_ids = _normalize_user_ids(requested_member_user_ids) + + if requested_ids: + active_requested = await _load_active_user_ids( + db, + tenant_id=tenant_id, + candidate_user_ids=requested_ids, + ) + invalid_user_ids = sorted(set(requested_ids) - active_requested) + if invalid_user_ids: + raise ProjectMemberAssignmentError( + "member_user_ids 包含不存在或非激活用户", + invalid_user_ids=invalid_user_ids, + ) + member_ids.update(active_requested) + elif require_requested_members: + raise ProjectMemberAssignmentError("至少分配一名项目成员") + + if include_owner and owner_id is not None: + owner_ids = await _load_active_user_ids( + db, + tenant_id=tenant_id, + candidate_user_ids=[int(owner_id)], + ) + member_ids.update(owner_ids) + + if not member_ids and not require_requested_members: + member_ids.update( + await _load_role_member_ids( + db, + tenant_id=tenant_id, + assigned_role=assigned_role, + ) + ) + + if not member_ids and not require_requested_members: + active_user_ids = await _load_active_user_ids(db, tenant_id=tenant_id) + if active_user_ids: + member_ids.add(min(active_user_ids)) + + if not member_ids: + raise ProjectMemberAssignmentError("当前租户没有可分配成员，已拒绝写入") + return sorted(member_ids) + + +async def replace_project_members( + db: AsyncSession, + *, + tenant_id: int, + project_id: int, + member_user_ids: Sequence[int], +) -> list[int]: + """全量替换项目成员。""" + normalized_ids = _normalize_user_ids(member_user_ids) + if not normalized_ids: + raise ProjectMemberAssignmentError("至少分配一名项目成员") + + await db.execute( + delete(UserProject).where( + and_( + UserProject.tenant_id == int(tenant_id), + UserProject.project_id == int(project_id), + ) + ) + ) + for user_id in normalized_ids: + db.add( + UserProject( + tenant_id=int(tenant_id), + user_id=int(user_id), + project_id=int(project_id), + ) + ) + return normalized_ids + + +async def notify_member_assignment_failure( + db: AsyncSession, + *, + tenant_id: int, + actor_user_id: int | None, + reason: str, + interface_path: str, + project_id: int | None = None, + project_title: str | None = None, + commit: bool = False, +) -> None: + """将成员分配失败事件写入消息台。""" + if actor_user_id is None: + return + + project_label = f"项目#{int(project_id)}" if project_id is not None else (project_title or "新项目") + content = ( + f"接口：{interface_path}\n" + f"项目：{project_label}\n" + f"原因：{reason}" + ) + await NotificationService(db).create_notification( + tenant_id=int(tenant_id), + user_id=int(actor_user_id), + category="system", + title="项目成员分配失败", + content=content, + source="project_member_service", + entity_type="project", + entity_id=int(project_id) if project_id is not None else None, + target_url=f"/projects/{int(project_id)}/board" if project_id is not None else "/projects", + kind="error", + priority="high", + status="failed", + blocking_reason=reason[:500], + recommended_action="retry", + dedupe_key=( + f"project-member-failed:{int(tenant_id)}:" + f"{int(actor_user_id)}:{interface_path}:{int(project_id or 0)}" + )[:255], + commit=commit, + ) diff --git a/backend/app/services/signal_source_repair_service.py b/backend/app/services/signal_source_repair_service.py new file mode 100644 index 00000000..987d66be --- /dev/null +++ b/backend/app/services/signal_source_repair_service.py @@ -0,0 +1,465 @@ +"""信号源脏数据修复服务。""" + +from __future__ import annotations + +import asyncio +import os +import re +from typing import Any, Optional + +from sqlalchemy import and_, select +from sqlalchemy.ext.asyncio import AsyncSession + +from ..models import SignalSourceRegistry, User +from ..utils.audited_write import get_audit_context +from ..utils.timezone import now_cn_naive +from .notification_service import NotificationService +from .operation_log_service import log_operation +from .signal_capture import ProactiveSignalCaptureService + + +class SignalSourceRepairError(RuntimeError): + """信号源修复失败。""" + + +class SignalSourceRepairTimeoutError(SignalSourceRepairError): + """信号源修复超时。""" + + def __init__(self, *, registry_id: int, stage: str, timeout_seconds: int) -> None: + super().__init__( + f"signal source repair timeout: registry_id={int(registry_id)}, " + f"stage={stage}, timeout_seconds={int(timeout_seconds)}" + ) + self.registry_id = int(registry_id) + self.stage = str(stage) + self.timeout_seconds = int(timeout_seconds) + + +class SignalSourceRepairMissingSignalError(SignalSourceRepairError): + """修复路径未拿到 signal_id。""" + + def __init__(self, *, registry_id: int, project_id: int, status: str) -> None: + super().__init__( + f"signal source repair failed: registry_id={int(registry_id)}, " + f"project_id={int(project_id)}, capture_status={status or 'unknown'}" + ) + self.registry_id = int(registry_id) + self.project_id = int(project_id) + self.status = str(status or "") + + +class SignalSourceRepairService: + """通过 capture-feishu 接口补录信号并回填历史 registry.signal_id。""" + + def __init__(self, db: AsyncSession) -> None: + self.db = db + + @staticmethod + def _safe_int(value: Any) -> Optional[int]: + try: + if value is None: + return None + return int(value) + except (TypeError, ValueError): + return None + + @staticmethod + def _resolve_timeout_seconds() -> int: + raw = ( + os.getenv("SIGNAL_SOURCE_REPAIR_TIMEOUT_SECONDS") + or os.getenv("LLM_REQUEST_TIMEOUT") + or "60" + ) + try: + timeout = int(float(raw)) + except (TypeError, ValueError): + timeout = 60 + return max(10, min(timeout, 300)) + + @staticmethod + def _build_repair_message(*, project_id: int, registry_id: int, source_name: str) -> str: + return ( + f"给项目ID: {project_id} 补录历史接入记录 registry#{registry_id}（{source_name}），" + "请仅执行入库，不需要二次处理。" + ) + + @staticmethod + def _build_explicit_repair_message(*, project_id: int, row: SignalSourceRegistry) -> str: + source_kind = str(row.source_kind or "").strip()[:40] + source_name = str(row.source_name or "历史信号源").strip()[:80] + source_identifier = str(row.source_identifier or "").strip()[:180] + return ( + f"项目 {int(project_id)} 新增信号源：{source_name}。" + f"source_kind={source_kind}; source_identifier={source_identifier}。" + "这是历史脏数据补录，请执行 signal_source.onboard 入库并生成消息台记录。" + ) + + @staticmethod + def _build_capture_message_id(*, project_id: int, registry_id: int, stage: str) -> str: + phase = "explicit" if str(stage or "").strip().lower() == "explicit" else "base" + return f"om_registry_repair_{int(project_id)}_{int(registry_id)}_{phase}"[:120] + + async def _load_registry( + self, + *, + tenant_id: int, + project_id: int, + registry_id: int, + ) -> Optional[SignalSourceRegistry]: + return ( + await self.db.execute( + select(SignalSourceRegistry).where( + and_( + SignalSourceRegistry.tenant_id == int(tenant_id), + SignalSourceRegistry.project_id == int(project_id), + SignalSourceRegistry.id == int(registry_id), + ) + ).limit(1) + ) + ).scalar_one_or_none() + + @staticmethod + def _build_group_chat_id(row: SignalSourceRegistry) -> str: + raw = (row.config or {}).get("chat_id") or row.source_identifier or f"repair_registry_{int(row.id)}" + return str(raw)[:120] + + @staticmethod + def _extract_route_status(capture_result: dict[str, Any]) -> str: + rows = capture_result.get("results") + if not isinstance(rows, list) or not rows: + return "" + first = rows[0] if isinstance(rows[0], dict) else {} + return str(first.get("route_status") or "") + + @classmethod + def _extract_signal_id(cls, capture_result: dict[str, Any]) -> Optional[int]: + candidates: list[Any] = [capture_result.get("signal_id")] + for key in ("signal_source_workflow", "signal_source_orchestration", "ai_execution"): + payload = capture_result.get(key) + if isinstance(payload, dict): + candidates.append(payload.get("signal_id")) + rows = capture_result.get("results") + if isinstance(rows, list) and rows and isinstance(rows[0], dict): + candidates.append(rows[0].get("signal_id")) + for item in candidates: + value = cls._safe_int(item) + if value is not None: + return value + return None + + @staticmethod + def _normalize_identifier_key(value: Any) -> str: + key = re.sub(r"\s+", "", str(value or "").strip().lower()) + if not key: + return "" + for prefix in ( + "feishu:chat:", + "wecom:chat:", + "feishu_chat_", + "wecom_chat_", + "feishu:", + "wecom:", + ): + if key.startswith(prefix): + key = key[len(prefix):] + break + return key[:180] + + async def _find_reusable_signal_id( + self, + *, + tenant_id: int, + project_id: int, + row: SignalSourceRegistry, + ) -> Optional[int]: + target_key = self._normalize_identifier_key(row.source_identifier) + if not target_key: + return None + candidates = ( + await self.db.execute( + select(SignalSourceRegistry).where( + and_( + SignalSourceRegistry.tenant_id == int(tenant_id), + SignalSourceRegistry.project_id == int(project_id), + SignalSourceRegistry.id != int(row.id), + SignalSourceRegistry.signal_id.is_not(None), + ) + ) + .order_by(SignalSourceRegistry.created_at.desc(), SignalSourceRegistry.id.desc()) + .limit(50) + ) + ).scalars().all() + for candidate in candidates: + current_key = self._normalize_identifier_key(candidate.source_identifier) + if current_key and current_key == target_key: + return self._safe_int(candidate.signal_id) + for candidate in candidates: + current_key = self._normalize_identifier_key(candidate.source_identifier) + if current_key and (current_key.endswith(target_key) or target_key.endswith(current_key)): + return self._safe_int(candidate.signal_id) + return None + + async def _capture_registry( + self, + *, + capture_service: ProactiveSignalCaptureService, + tenant_id: int, + project_id: int, + sender: str, + row: SignalSourceRegistry, + stage: str, + timeout_seconds: int, + ) -> dict[str, Any]: + message = ( + self._build_explicit_repair_message(project_id=project_id, row=row) + if stage == "explicit" + else self._build_repair_message( + project_id=project_id, + registry_id=int(row.id), + source_name=str(row.source_name or "信号源"), + ) + ) + coroutine = capture_service.capture_from_feishu( + group_chat_id=self._build_group_chat_id(row), + message_content=message, + sender=sender, + tenant_id=int(tenant_id), + feishu_message_id=self._build_capture_message_id( + project_id=project_id, + registry_id=int(row.id), + stage=stage, + ), + ) + try: + return await asyncio.wait_for(coroutine, timeout=timeout_seconds) + except asyncio.TimeoutError as exc: + raise SignalSourceRepairTimeoutError( + registry_id=int(row.id), + stage=stage, + timeout_seconds=timeout_seconds, + ) from exc + + async def _active_user_ids(self, *, tenant_id: int) -> list[int]: + rows = ( + await self.db.execute( + select(User.id).where( + and_( + User.tenant_id == int(tenant_id), + User.is_active.is_(True), + ) + ) + ) + ).scalars().all() + return [int(item) for item in rows if item is not None] + + async def _emit_visibility_records( + self, + *, + tenant_id: int, + project_id: int, + row: SignalSourceRegistry, + signal_id: int, + capture_status: str, + ) -> None: + dedupe_key = f"signal-source-repair:{int(tenant_id)}:{int(row.id)}:{int(signal_id)}" + await NotificationService(self.db).create_notifications( + tenant_id=int(tenant_id), + user_ids=await self._active_user_ids(tenant_id=tenant_id), + category="system", + title="AI 已修复历史信号源关联", + content=( + f"项目ID: {int(project_id)}\n" + f"信号源ID: {int(row.id)}\n" + f"signal_id: {int(signal_id)}\n" + f"status: {capture_status or 'relinked'}" + ), + source="ai_steward_repair", + entity_type="signal_source_registry", + entity_id=int(row.id), + kind="info", + priority="medium", + status="notified", + recommended_action="view_result", + dedupe_key=dedupe_key, + payload={ + "project_id": int(project_id), + "registry_id": int(row.id), + "signal_id": int(signal_id), + "capture_status": capture_status or "relinked", + }, + commit=False, + ) + context = get_audit_context() + await log_operation( + self.db, + tenant_id=int(tenant_id), + user_id=int(context.user_id or 1), + project_id=int(project_id), + entity_type="signal_source_registry", + entity_id=str(int(row.id)), + action="signal_source_registry.repair_link", + after_data={ + "signal_id": int(signal_id), + "capture_status": capture_status or "relinked", + }, + remark="repair dirty signal source registry link", + ) + + async def _persist_registry_link( + self, + *, + tenant_id: int, + project_id: int, + row: SignalSourceRegistry, + signal_id: int, + capture_status: str, + ) -> None: + row.signal_id = int(signal_id) + row.last_signal_at = now_cn_naive() + await self._emit_visibility_records( + tenant_id=tenant_id, + project_id=project_id, + row=row, + signal_id=signal_id, + capture_status=capture_status, + ) + await self.db.flush() + + def _build_item_result( + self, + *, + row: SignalSourceRegistry, + signal_id: Optional[int], + capture_status: str, + work_item_id: Optional[int], + route_status: str, + ) -> dict[str, Any]: + return { + "registry_id": int(row.id), + "signal_id": signal_id, + "capture_status": capture_status, + "work_item_id": work_item_id, + "route_status": route_status, + } + + async def _repair_single_registry( + self, + *, + capture_service: ProactiveSignalCaptureService, + tenant_id: int, + project_id: int, + registry_id: int, + sender: str, + timeout_seconds: int, + ) -> dict[str, Any]: + row = await self._load_registry( + tenant_id=tenant_id, + project_id=project_id, + registry_id=registry_id, + ) + if row is None: + return {"registry_id": int(registry_id), "status": "not_found"} + if row.signal_id is not None: + return self._build_item_result( + row=row, + signal_id=self._safe_int(row.signal_id), + capture_status="already_linked", + work_item_id=None, + route_status="", + ) + reusable_signal_id = await self._find_reusable_signal_id( + tenant_id=tenant_id, + project_id=project_id, + row=row, + ) + if reusable_signal_id is not None: + await self._persist_registry_link( + tenant_id=tenant_id, + project_id=project_id, + row=row, + signal_id=int(reusable_signal_id), + capture_status="relinked", + ) + return self._build_item_result( + row=row, + signal_id=int(reusable_signal_id), + capture_status="relinked", + work_item_id=None, + route_status="", + ) + capture_result = await self._capture_registry( + capture_service=capture_service, + tenant_id=tenant_id, + project_id=project_id, + sender=sender, + row=row, + stage="base", + timeout_seconds=timeout_seconds, + ) + signal_id = self._extract_signal_id(capture_result) + if signal_id is None: + capture_result = await self._capture_registry( + capture_service=capture_service, + tenant_id=tenant_id, + project_id=project_id, + sender=sender, + row=row, + stage="explicit", + timeout_seconds=timeout_seconds, + ) + signal_id = self._extract_signal_id(capture_result) + if signal_id is None: + signal_id = await self._find_reusable_signal_id( + tenant_id=tenant_id, + project_id=project_id, + row=row, + ) + if signal_id is None: + raise SignalSourceRepairMissingSignalError( + registry_id=int(row.id), + project_id=int(project_id), + status=str(capture_result.get("status") or ""), + ) + capture_status = str(capture_result.get("status") or "captured") + await self._persist_registry_link( + tenant_id=tenant_id, + project_id=project_id, + row=row, + signal_id=int(signal_id), + capture_status=capture_status, + ) + return self._build_item_result( + row=row, + signal_id=int(signal_id), + capture_status=capture_status, + work_item_id=self._safe_int(capture_result.get("work_item_id")), + route_status=self._extract_route_status(capture_result), + ) + + async def repair_registries( + self, + *, + tenant_id: int, + project_id: int, + registry_ids: list[int], + sender: str, + ) -> list[dict[str, Any]]: + capture_service = ProactiveSignalCaptureService(self.db) + timeout_seconds = self._resolve_timeout_seconds() + results: list[dict[str, Any]] = [] + try: + for registry_id in registry_ids: + item = await self._repair_single_registry( + capture_service=capture_service, + tenant_id=tenant_id, + project_id=project_id, + registry_id=int(registry_id), + sender=sender, + timeout_seconds=timeout_seconds, + ) + results.append(item) + await self.db.commit() + except Exception: + await self.db.rollback() + raise + return results diff --git a/backend/app/services/sstc_service.py b/backend/app/services/sstc_service.py index 0355a0e7..d9dfd87d 100644 --- a/backend/app/services/sstc_service.py +++ b/backend/app/services/sstc_service.py @@ -197,39 +197,78 @@ class SSTCService: } async def get_workload_health(self, assignee_ids: list[int]) -> dict[str, Any]: - if not assignee_ids: + member_profiles = await self._load_member_workload_profiles(assignee_ids) + if not member_profiles: return { "avg_ratio": 0.0, "overloaded_count": 0, "idle_count": 0, "member_count": 0, + "members": [], } - result = await self.db.execute( - select(UserWorkload).where( - and_( - UserWorkload.tenant_id == self.tenant_id, - UserWorkload.user_id.in_(assignee_ids), - ) - ) - ) - workloads = list(result.scalars().all()) - ratios = [] - for row in workloads: - ratio = float(row.workload_ratio or 0) - if ratio > 1.5: - ratio = ratio / 100.0 - ratios.append(ratio) - avg_ratio = _avg(ratios) + ratios = [float(item["workload_ratio"]) for item in member_profiles] overloaded_count = len([x for x in ratios if x > 1.0]) idle_count = len([x for x in ratios if x < 0.3]) return { - "avg_ratio": round(avg_ratio, 2), + "avg_ratio": round(_avg(ratios), 2), "overloaded_count": overloaded_count, "idle_count": idle_count, - "member_count": len(assignee_ids), + "member_count": len(member_profiles), + "members": member_profiles, } + @staticmethod + def _normalize_workload_ratio(raw_ratio: float | int | None) -> float: + ratio = float(raw_ratio or 0) + if ratio > 1.5: + ratio = ratio / 100.0 + return max(0.0, ratio) + + async def _load_member_workload_profiles(self, assignee_ids: list[int]) -> list[dict[str, Any]]: + if not assignee_ids: + return [] + + rows = await self.db.execute( + select(User.id, User.name, User.username, User.role, UserWorkload.workload_ratio) + .select_from(User) + .outerjoin( + UserWorkload, + and_( + UserWorkload.tenant_id == self.tenant_id, + UserWorkload.user_id == User.id, + ), + ) + .where( + and_( + User.tenant_id == self.tenant_id, + User.id.in_(assignee_ids), + User.is_active.is_(True), + ) + ) + .order_by(User.id.asc()) + ) + + members: list[dict[str, Any]] = [] + for user_id, name, username, role, workload_ratio in rows.all(): + ratio = self._normalize_workload_ratio(workload_ratio) + if ratio > 1.0: + status = "overloaded" + elif ratio < 0.3: + status = "idle" + else: + status = "normal" + members.append( + { + "user_id": str(int(user_id)), + "name": str(name or username or f"用户{int(user_id)}"), + "role": str(role or ""), + "workload_ratio": round(ratio, 2), + "status": status, + } + ) + return members + @staticmethod def _task_metrics(tasks: list[Task]) -> dict[str, Any]: total = len(tasks) @@ -272,7 +311,7 @@ class SSTCService: charters_ids = [int(c.id) for c in charters] alignment = await self.get_alignment_stats(charters_ids) metrics = self._task_metrics(tasks) - member_ids = await self.get_project_member_ids(project_id) + member_ids = await self.get_project_workload_member_ids(project_id) workload = await self.get_workload_health(member_ids) chain_id = str(charters[0].id) if charters else None return { @@ -568,7 +607,7 @@ class SSTCService: ) return int(result.scalar() or 0) > 0 - async def get_project_member_ids(self, project_id: int) -> list[int]: + async def _load_project_member_id_set(self, project_id: int) -> set[int]: member_result = await self.db.execute( select(UserProject.user_id).where( and_( @@ -577,7 +616,10 @@ class SSTCService: ) ) ) - member_ids = {int(x) for x in member_result.scalars().all() if x is not None} + return {int(x) for x in member_result.scalars().all() if x is not None} + + async def get_project_workload_member_ids(self, project_id: int) -> list[int]: + member_ids = await self._load_project_member_id_set(project_id) project = await self.get_project(project_id) if not member_ids and project: member_ids = set( @@ -590,6 +632,11 @@ class SSTCService: ) ) await self.db.commit() + return sorted(member_ids) + + async def get_project_member_ids(self, project_id: int) -> list[int]: + member_ids = set(await self.get_project_workload_member_ids(project_id)) + project = await self.get_project(project_id) if project and project.owner_id: member_ids.add(int(project.owner_id)) return sorted(member_ids) diff --git a/backend/app/utils/audited_write.py b/backend/app/utils/audited_write.py new file mode 100644 index 00000000..7223548d --- /dev/null +++ b/backend/app/utils/audited_write.py @@ -0,0 +1,112 @@ +"""写接口审计上下文工具。""" + +from __future__ import annotations + +from contextvars import ContextVar, Token +from dataclasses import dataclass, replace +from datetime import datetime +from functools import wraps +from inspect import iscoroutinefunction +from typing import Any, Awaitable, Callable, Optional, TypeVar, cast + +from .timezone import CN_TZ + + +T = TypeVar("T") + +VALID_CALLERS = {"system", "user", "ai"} + + +@dataclass(frozen=True) +class AuditRequestContext: + trace_id: Optional[str] = None + interface_path: Optional[str] = None + caller: str = "system" + method: Optional[str] = None + tenant_id: Optional[int] = None + user_id: Optional[int] = None + + +_audit_context: ContextVar[AuditRequestContext] = ContextVar( + "pmflow_audit_request_context", + default=AuditRequestContext(), +) + + +def normalize_caller(value: Optional[str]) -> str: + normalized = str(value or "").strip().lower() + return normalized if normalized in VALID_CALLERS else "system" + + +def get_audit_context() -> AuditRequestContext: + return _audit_context.get() + + +def set_audit_context(context: AuditRequestContext) -> Token[AuditRequestContext]: + return _audit_context.set( + AuditRequestContext( + trace_id=(str(context.trace_id)[:64] if context.trace_id else None), + interface_path=(str(context.interface_path)[:255] if context.interface_path else None), + caller=normalize_caller(context.caller), + method=(str(context.method)[:16] if context.method else None), + tenant_id=int(context.tenant_id) if context.tenant_id is not None else None, + user_id=int(context.user_id) if context.user_id is not None else None, + ) + ) + + +def reset_audit_context(token: Token[AuditRequestContext]) -> None: + _audit_context.reset(token) + + +def update_audit_context( + *, + trace_id: Optional[str] = None, + interface_path: Optional[str] = None, + caller: Optional[str] = None, + method: Optional[str] = None, + tenant_id: Optional[int] = None, + user_id: Optional[int] = None, +) -> AuditRequestContext: + current = get_audit_context() + next_ctx = replace( + current, + trace_id=current.trace_id if trace_id is None else trace_id, + interface_path=current.interface_path if interface_path is None else interface_path, + caller=current.caller if caller is None else normalize_caller(caller), + method=current.method if method is None else method, + tenant_id=current.tenant_id if tenant_id is None else int(tenant_id), + user_id=current.user_id if user_id is None else int(user_id), + ) + _audit_context.set(next_ctx) + return next_ctx + + +def format_beijing_datetime(dt: Optional[datetime]) -> Optional[str]: + if dt is None: + return None + localized = dt.replace(tzinfo=CN_TZ) if dt.tzinfo is None else dt.astimezone(CN_TZ) + return localized.strftime("%Y-%m-%d %H:%M:%S") + + +def audited_write(*, caller: Optional[str] = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]: + """写接口装饰器：覆写当前请求 caller 审计上下文。""" + + normalized = normalize_caller(caller) + + def _decorator(func: Callable[..., Any]) -> Callable[..., Any]: + if not iscoroutinefunction(func): + return func + + @wraps(func) + async def _async_wrapper(*args: Any, **kwargs: Any) -> Any: + current = get_audit_context() + token = set_audit_context(replace(current, caller=normalized)) + try: + return await cast(Callable[..., Awaitable[Any]], func)(*args, **kwargs) + finally: + reset_audit_context(token) + + return _async_wrapper + + return _decorator diff --git a/backend/app/utils/project_members.py b/backend/app/utils/project_members.py index 0fb148fd..d24786e6 100644 --- a/backend/app/utils/project_members.py +++ b/backend/app/utils/project_members.py @@ -1,41 +1,10 @@ from __future__ import annotations -import re - from sqlalchemy import and_, select from sqlalchemy.ext.asyncio import AsyncSession -from ..models import User, UserProject - -_ROLE_SPLIT_RE = re.compile(r"[,/|\s]+") -_ROLE_ALIAS_MAP = { - "AI": {"AI"}, - "AI_ENGINEER": {"AI"}, - "FRONTEND": {"FE"}, - "FE": {"FE"}, - "BACKEND": {"BE"}, - "BE": {"BE"}, - "BOSS": {"BOSS"}, - "ADMIN": {"ADMIN"}, -} - - -def _parse_role_tokens(raw: str | None) -> set[str]: - tokens = { - token.strip().upper() - for token in _ROLE_SPLIT_RE.split(str(raw or "").strip()) - if token.strip() - } - normalized: set[str] = set() - for token in tokens: - normalized.update(_ROLE_ALIAS_MAP.get(token, {token})) - return normalized - - -def _user_matches_project_role(user_role: str | None, expected_roles: set[str]) -> bool: - if not expected_roles: - return False - return bool(_parse_role_tokens(user_role).intersection(expected_roles)) +from ..models import UserProject +from ..services.project_member_service import resolve_project_member_ids async def ensure_project_default_members( @@ -45,37 +14,19 @@ async def ensure_project_default_members( project_id: int, owner_id: int | None, assigned_role: str | None, + requested_member_user_ids: list[int] | None = None, ) -> list[int]: - member_ids: set[int] = set() - if owner_id is not None: - member_ids.add(int(owner_id)) - - expected_roles = _parse_role_tokens(assigned_role) - if expected_roles: - rows = await db.execute( - select(User.id, User.role).where( - and_( - User.tenant_id == int(tenant_id), - User.status == "active", - User.is_active.is_(True), - ) - ) + member_ids = set( + await resolve_project_member_ids( + db, + tenant_id=int(tenant_id), + owner_id=int(owner_id) if owner_id is not None else None, + assigned_role=assigned_role, + requested_member_user_ids=requested_member_user_ids or [], + require_requested_members=False, + include_owner=True, ) - for user_id, user_role in rows.all(): - if _user_matches_project_role(user_role, expected_roles): - member_ids.add(int(user_id)) - - if not member_ids: - fallback_rows = await db.execute( - select(User.id).where( - and_( - User.tenant_id == int(tenant_id), - User.status == "active", - User.is_active.is_(True), - ) - ).order_by(User.id.asc()).limit(1) - ) - member_ids.update(int(user_id) for user_id in fallback_rows.scalars().all()) + ) existing_rows = await db.execute( select(UserProject.user_id).where( @@ -96,4 +47,7 @@ async def ensure_project_default_members( ) ) - return sorted(member_ids.union(existing_member_ids)) + all_member_ids = sorted(member_ids.union(existing_member_ids)) + if not all_member_ids: + raise ValueError("项目成员为空，已拒绝写入") + return all_member_ids diff --git a/backend/neural_sync.db-shm b/backend/neural_sync.db-shm new file mode 100644 index 0000000000000000000000000000000000000000..fe9ac2845eca6fe6da8a63cd096d9cf9e24ece10 GIT binary patch literal 32768 zcmeIuAr62r3 None: + self.assertGreaterEqual(_intent_llm_timeout_seconds(function_name="classify_intent_complex"), 30.0) + self.assertGreaterEqual(_intent_llm_timeout_seconds(function_name=INTENT_LLM_FUNCTION_NAME), 30.0) + + async def test_correction_message_forces_low_tier_glm47_intent_function(self) -> None: + prev = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "1" + captured_function_name: dict[str, str] = {} + try: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + + async def _mock_generate(_: str, **kwargs): + captured_function_name["value"] = str(kwargs.get("function_name") or "") + return StructuredExtractionResult( + payload={ + "intent": INTENT_PROJECT_CREATE, + "confidence": 0.93, + "project_id": None, + "reasoning": "否定纠错后回退项目创建", + }, + source_field="content", + provider="glm", + model="GLM-4.7", + validation="ok", + ) + + with patch( + "app.services.ai_pmo_kernel_service.select_intent_route", + return_value=SimpleNamespace( + function_name="classify_intent_complex", + tier="high", + reasons=["complex_multi_clause"], + is_complex=True, + ), + ), patch( + "app.services.ai_pmo_kernel_service.kernel_resolve_correction", + return_value=SimpleNamespace(action="rewrite", matched_keywords=["不是"], confidence=0.92), + ), patch( + "app.services.ai_pmo_kernel_service.llm_service.generate_structured_object", + AsyncMock(side_effect=_mock_generate), + ): + llm_intent = await service._resolve_intent_with_llm( + message_text="不是创建信号，是创建项目名为openclaw自动化部署。", + session=None, + world_model={"resolved_project_id": None, "project": {}}, + ) + + self.assertIsNotNone(llm_intent) + self.assertEqual(captured_function_name.get("value"), INTENT_LLM_FUNCTION_NAME) + self.assertTrue(bool((llm_intent.metadata or {}).get("forced_low_tier"))) + 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_two_stage_intent_escalates_to_high_on_conflict(self) -> None: + prev_enabled = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") + prev_budget = os.environ.get("AI_PMO_INTENT_LLM_TOTAL_BUDGET_SECONDS") + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "1" + os.environ["AI_PMO_INTENT_LLM_TOTAL_BUDGET_SECONDS"] = "80" + called_functions: list[str] = [] + try: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + + async def _mock_generate(_: str, **kwargs): + fn = str(kwargs.get("function_name") or "") + called_functions.append(fn) + if fn == INTENT_LLM_FUNCTION_NAME: + return StructuredExtractionResult( + payload={ + "intent": INTENT_SIGNAL_CREATE, + "confidence": 0.58, + "project_id": None, + "reasoning": "低阶先判成信号创建", + }, + source_field="content", + provider="glm", + model="GLM-4.7", + validation="ok", + ) + return StructuredExtractionResult( + payload={ + "intent": INTENT_PROJECT_CREATE, + "confidence": 0.93, + "project_id": None, + "reasoning": "高阶复判为项目创建", + }, + source_field="content", + provider="openai", + model="GLM-5", + validation="ok", + ) + + with patch( + "app.services.ai_pmo_kernel_service.select_intent_route", + return_value=SimpleNamespace( + function_name="classify_intent_complex", + tier="high", + reasons=["complex_multi_clause"], + is_complex=True, + ), + ), patch( + "app.services.ai_pmo_kernel_service.kernel_resolve_correction", + return_value=SimpleNamespace(action="none", matched_keywords=[], confidence=0.0), + ), patch( + "app.services.ai_pmo_kernel_service.llm_service.generate_structured_object", + AsyncMock(side_effect=_mock_generate), + ): + llm_intent = await service._resolve_intent_with_llm( + message_text="给我新增个项目，需求名字为openclaw自动化部署。", + session=None, + world_model={"resolved_project_id": None, "project": {}}, + rule_intent=service._build_canonical_intent( + intent_name=INTENT_PROJECT_CREATE, + message_text="给我新增个项目，需求名字为openclaw自动化部署。", + project_id=None, + confidence=0.88, + metadata={"context_source": "rule"}, + ), + ) + + self.assertIsNotNone(llm_intent) + self.assertEqual(called_functions, [INTENT_LLM_FUNCTION_NAME, "classify_intent_complex"]) + self.assertEqual(llm_intent.name, INTENT_PROJECT_CREATE) + self.assertEqual((llm_intent.metadata or {}).get("llm_function"), "classify_intent_complex") + self.assertTrue(bool((llm_intent.metadata or {}).get("stage2_promoted"))) + finally: + if prev_enabled is None: + os.environ.pop("AI_PMO_INTENT_LLM_ENABLED", None) + else: + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = prev_enabled + if prev_budget is None: + os.environ.pop("AI_PMO_INTENT_LLM_TOTAL_BUDGET_SECONDS", None) + else: + os.environ["AI_PMO_INTENT_LLM_TOTAL_BUDGET_SECONDS"] = prev_budget + + async def test_two_stage_intent_keeps_low_when_confident_and_no_conflict(self) -> None: + prev_enabled = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "1" + called_functions: list[str] = [] + try: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + + async def _mock_generate(_: str, **kwargs): + fn = str(kwargs.get("function_name") or "") + called_functions.append(fn) + return StructuredExtractionResult( + payload={ + "intent": INTENT_QUERY, + "confidence": 0.95, + "project_id": None, + "reasoning": "低阶高置信查询", + }, + source_field="content", + provider="glm", + model="GLM-4.7", + validation="ok", + ) + + with patch( + "app.services.ai_pmo_kernel_service.select_intent_route", + return_value=SimpleNamespace( + function_name=INTENT_LLM_FUNCTION_NAME, + tier="low", + reasons=["default_simple"], + is_complex=False, + ), + ), patch( + "app.services.ai_pmo_kernel_service.kernel_resolve_correction", + return_value=SimpleNamespace(action="none", matched_keywords=[], confidence=0.0), + ), patch( + "app.services.ai_pmo_kernel_service.llm_service.generate_structured_object", + AsyncMock(side_effect=_mock_generate), + ): + llm_intent = await service._resolve_intent_with_llm( + message_text="项目进展如何？", + session=None, + world_model={"resolved_project_id": None, "project": {}}, + rule_intent=service._build_canonical_intent( + intent_name=INTENT_QUERY, + message_text="项目进展如何？", + project_id=None, + confidence=0.90, + metadata={"context_source": "rule"}, + ), + ) + + self.assertIsNotNone(llm_intent) + self.assertEqual(called_functions, [INTENT_LLM_FUNCTION_NAME]) + self.assertEqual(llm_intent.name, INTENT_QUERY) + self.assertEqual((llm_intent.metadata or {}).get("stage"), "stage1_low") + finally: + if prev_enabled is None: + os.environ.pop("AI_PMO_INTENT_LLM_ENABLED", None) + else: + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = prev_enabled + async def test_intent_llm_failure_falls_back_to_rule_resolver(self) -> None: prev = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "1" diff --git a/backend/tests/test_ai_steward_title_extract.py b/backend/tests/test_ai_steward_title_extract.py index e00e5059..29a56683 100644 --- a/backend/tests/test_ai_steward_title_extract.py +++ b/backend/tests/test_ai_steward_title_extract.py @@ -1,4 +1,5 @@ import unittest +from unittest.mock import AsyncMock from app.services.ai_steward import AIStewardService @@ -8,6 +9,14 @@ class AIStewardTitleExtractTest(unittest.IsolatedAsyncioTestCase): title = AIStewardService._extract_project_title_from_instruction("给我新增个项目，openclaw自动化部署。") self.assertEqual(title, "openclaw自动化部署") + async def test_extract_title_from_requirement_name_instruction(self) -> None: + title = AIStewardService._extract_project_title_from_instruction("给我新增个项目，需求名字为openclaw自动化部署。") + self.assertEqual(title, "openclaw自动化部署") + + async def test_extract_title_from_project_name_instruction(self) -> None: + title = AIStewardService._extract_project_title_from_instruction("给我新增个项目，项目名为openclaw自动化部署。") + self.assertEqual(title, "openclaw自动化部署") + async def test_extract_title_from_correction_sentence(self) -> None: title = AIStewardService._extract_project_title_from_instruction("不是创建信号，是创建项目名为openclaw自动化部署。") self.assertEqual(title, "openclaw自动化部署") @@ -29,3 +38,31 @@ class AIStewardTitleExtractTest(unittest.IsolatedAsyncioTestCase): ) self.assertEqual(normalized["project"]["title"], "openclaw自动化部署") self.assertEqual(normalized["project"]["description"], "项目：openclaw自动化部署") + + async def test_project_creation_text_retries_with_glm47_fallback_function(self) -> None: + service = AIStewardService(db=None) # type: ignore[arg-type] + called_functions: list[str] = [] + + async def _mock_generate(_: str, **kwargs): + fn = str(kwargs.get("function_name") or "") + called_functions.append(fn) + if len(called_functions) <= 2: + return "我暂时无法生成可靠回答（模型服务异常）。请稍后重试。" + return '{"project_name":"openclaw自动化部署"}' + + service.llm.generate = AsyncMock(side_effect=_mock_generate) # type: ignore[method-assign] + text = await service._generate_project_creation_text( + prompt="生成分析结果", + tenant_id=1, + function_name="project_from_text_analysis", + timeout_seconds=5.0, + ) + self.assertEqual(text, '{"project_name":"openclaw自动化部署"}') + self.assertEqual( + called_functions, + [ + "project_from_text_analysis", + "project_from_text_analysis", + "project_draft_analysis_glm47_fallback", + ], + ) diff --git a/backend/tests/test_openclaw_commander_service.py b/backend/tests/test_openclaw_commander_service.py index 2230129e..36780cc5 100644 --- a/backend/tests/test_openclaw_commander_service.py +++ b/backend/tests/test_openclaw_commander_service.py @@ -89,6 +89,14 @@ class OpenClawCommanderServiceTest(unittest.TestCase): def setUp(self) -> None: self.executor = OperationExecutorService(db=None) # type: ignore[arg-type] + def test_extract_project_create_title_handles_requirement_prefix(self) -> None: + title = OpenClawCommanderService._extract_project_create_title("给我新增个项目，需求名字为openclaw自动化部署。") + self.assertEqual(title, "openclaw自动化部署") + + def test_extract_project_create_title_handles_project_name_prefix(self) -> None: + title = OpenClawCommanderService._extract_project_create_title("给我新增个项目，项目名为openclaw自动化部署。") + self.assertEqual(title, "openclaw自动化部署") + def test_plan_operations_accepts_tool_call_json(self) -> None: service = OpenClawCommanderService(db=None) # type: ignore[arg-type] service._commander_operation_catalog = lambda: [ # type: ignore[method-assign] diff --git a/backend/tests/test_projects_member_assignment_api.py b/backend/tests/test_projects_member_assignment_api.py new file mode 100644 index 00000000..25dc3fc8 --- /dev/null +++ b/backend/tests/test_projects_member_assignment_api.py @@ -0,0 +1,203 @@ +import os +import unittest +from tempfile import NamedTemporaryFile +from unittest.mock import AsyncMock, patch + +from fastapi import HTTPException +from sqlalchemy import and_, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.api.projects import ( + ProjectCreateRequest, + ProjectMembersAssignRequest, + assign_project_members, + create_project, +) +from app.models import Base, Notification, Project, User, UserProject + + +class ProjectsMemberAssignmentApiTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + tmp = NamedTemporaryFile(prefix="pmflow-project-members-test-", 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 _seed_users(self, db: AsyncSession) -> tuple[User, User, User]: + boss = User( + tenant_id=1, + username="boss-member-test", + email="boss-member-test@example.com", + password_hash="x", + name="Boss", + role="BOSS", + status="active", + is_active=True, + ) + member_a = User( + tenant_id=1, + username="member-a-test", + email="member-a-test@example.com", + password_hash="x", + name="Member A", + role="BE", + status="active", + is_active=True, + ) + member_b = User( + tenant_id=1, + username="member-b-test", + email="member-b-test@example.com", + password_hash="x", + name="Member B", + role="FE", + status="active", + is_active=True, + ) + db.add_all([boss, member_a, member_b]) + await db.flush() + return boss, member_a, member_b + + async def test_create_project_assigns_requested_members_and_owner(self) -> None: + async with self.session_factory() as db: + boss, member_a, member_b = await self._seed_users(db) + + with ( + patch("app.api.projects.context_cache_service.invalidate_system_context", AsyncMock()), + patch("app.api.projects.AIPMOKernelService.invalidate_world_model_cache", AsyncMock()), + ): + payload = await create_project( + body=ProjectCreateRequest( + title="成员分配测试项目", + member_user_ids=[int(member_a.id), int(member_b.id)], + ), + db=db, + current_user=boss, + ) + + project_id = int(payload["id"]) + assigned_rows = ( + await db.execute( + select(UserProject.user_id).where( + and_( + UserProject.tenant_id == 1, + UserProject.project_id == project_id, + ) + ) + ) + ).scalars().all() + self.assertEqual( + sorted(int(uid) for uid in assigned_rows), + sorted([int(boss.id), int(member_a.id), int(member_b.id)]), + ) + + async def test_assign_project_members_empty_should_notify_and_raise(self) -> None: + async with self.session_factory() as db: + boss, member_a, _ = await self._seed_users(db) + project = Project( + tenant_id=1, + name="P-members-empty", + title="P-members-empty", + description="members test", + owner_id=int(boss.id), + status="active", + assigned_role="backend", + signal_count=0, + ) + db.add(project) + await db.flush() + db.add( + UserProject( + tenant_id=1, + user_id=int(member_a.id), + project_id=int(project.id), + ) + ) + await db.commit() + + with self.assertRaises(HTTPException) as exc_ctx: + await assign_project_members( + project_id=int(project.id), + body=ProjectMembersAssignRequest(member_user_ids=[]), + db=db, + current_user=boss, + ) + self.assertEqual(exc_ctx.exception.status_code, 400) + + notifications = ( + await db.execute( + select(Notification).where( + and_( + Notification.tenant_id == 1, + Notification.user_id == int(boss.id), + Notification.title == "项目成员分配失败", + ) + ) + ) + ).scalars().all() + self.assertGreaterEqual(len(notifications), 1) + + async def test_assign_project_members_replaces_member_list(self) -> None: + async with self.session_factory() as db: + boss, member_a, member_b = await self._seed_users(db) + project = Project( + tenant_id=1, + name="P-members-replace", + title="P-members-replace", + description="members replace test", + owner_id=int(boss.id), + status="active", + assigned_role="backend", + signal_count=0, + ) + db.add(project) + await db.flush() + db.add( + UserProject( + tenant_id=1, + user_id=int(boss.id), + project_id=int(project.id), + ) + ) + await db.commit() + + with ( + patch("app.api.projects.context_cache_service.invalidate_system_context", AsyncMock()), + patch("app.api.projects.AIPMOKernelService.invalidate_world_model_cache", AsyncMock()), + ): + payload = await assign_project_members( + project_id=int(project.id), + body=ProjectMembersAssignRequest(member_user_ids=[int(member_a.id), int(member_b.id)]), + db=db, + current_user=boss, + ) + + self.assertEqual( + sorted(payload["member_user_ids"]), + sorted([str(int(member_a.id)), str(int(member_b.id))]), + ) + assigned_rows = ( + await db.execute( + select(UserProject.user_id).where( + and_( + UserProject.tenant_id == 1, + UserProject.project_id == int(project.id), + ) + ) + ) + ).scalars().all() + self.assertEqual( + sorted(int(uid) for uid in assigned_rows), + sorted([int(member_a.id), int(member_b.id)]), + ) diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index aafc9f0e..b604f3ad 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -80,7 +80,7 @@ services: LLM_ROUTING_SAFETY_MODEL: ${LLM_ROUTING_SAFETY_MODEL:?missing_LLM_ROUTING_SAFETY_MODEL} LLM_DIRECT_EMERGENCY_ENABLED: ${LLM_DIRECT_EMERGENCY_ENABLED:?missing_LLM_DIRECT_EMERGENCY_ENABLED} LLM_REQUEST_TIMEOUT: ${LLM_REQUEST_TIMEOUT:?missing_LLM_REQUEST_TIMEOUT} - AUTO_CREATE_TABLES: "true" + AUTO_CREATE_TABLES: "false" GLM_API_KEY: ${GLM_API_KEY:?missing_GLM_API_KEY} GLM_BASE_URL: ${GLM_BASE_URL:?missing_GLM_BASE_URL} GLM_MODEL: ${GLM_MODEL:?missing_GLM_MODEL} diff --git a/docker-compose.yml b/docker-compose.yml index b06049f3..fbe47ac5 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -80,7 +80,7 @@ services: LLM_ROUTING_SAFETY_MODEL: ${LLM_ROUTING_SAFETY_MODEL:?missing_LLM_ROUTING_SAFETY_MODEL} LLM_DIRECT_EMERGENCY_ENABLED: ${LLM_DIRECT_EMERGENCY_ENABLED:?missing_LLM_DIRECT_EMERGENCY_ENABLED} LLM_REQUEST_TIMEOUT: ${LLM_REQUEST_TIMEOUT:?missing_LLM_REQUEST_TIMEOUT} - AUTO_CREATE_TABLES: "true" + AUTO_CREATE_TABLES: "false" GLM_API_KEY: ${GLM_API_KEY:?missing_GLM_API_KEY} GLM_BASE_URL: ${GLM_BASE_URL:?missing_GLM_BASE_URL} GLM_MODEL: ${GLM_MODEL:?missing_GLM_MODEL} diff --git a/frontend/src/api/aiSteward.ts b/frontend/src/api/aiSteward.ts index 5f2bcf6f..04552b6e 100644 --- a/frontend/src/api/aiSteward.ts +++ b/frontend/src/api/aiSteward.ts @@ -16,6 +16,25 @@ export interface OpenClawCommandDispatchRequest { max_operations?: number; } +export interface SignalSourceRepairRequest { + project_id: number; + registry_ids: number[]; + sender?: string; +} + +export interface SignalSourceRepairItem { + registry_id: number; + signal_id: number | null; + capture_status: string; + work_item_id: number | null; + route_status: string; +} + +export interface SignalSourceRepairResponse { + project_id: number; + items: SignalSourceRepairItem[]; +} + export interface OpenClawCommandExecutionItem { index: number; operation_code: string; @@ -90,4 +109,11 @@ export const aiStewardApi = { const raw = await apiRequest>("/ai-steward/llm/providers"); return unwrap(raw); }, + async repairDirtySignalSources(body: SignalSourceRepairRequest): Promise { + const raw = await apiRequest>("/ai-steward/signal-sources/repair-dirty", { + method: "POST", + body: JSON.stringify(body), + }); + return unwrap(raw); + }, }; diff --git a/frontend/src/api/audit.ts b/frontend/src/api/audit.ts index 0491308a..1fbfd852 100644 --- a/frontend/src/api/audit.ts +++ b/frontend/src/api/audit.ts @@ -5,6 +5,15 @@ import { apiRequest } from './base' +type ApiEnvelope = { code: number; message: string; data: T } + +function unwrap(raw: ApiEnvelope | T): T { + if (raw && typeof raw === 'object' && 'code' in (raw as ApiEnvelope) && 'data' in (raw as ApiEnvelope)) { + return (raw as ApiEnvelope).data + } + return raw as T +} + export interface CodeAuditRequest { task_name: string commit_message: string @@ -35,6 +44,49 @@ export interface CodeAuditResponse { details: Record } +export interface AuditOperation { + id: number + tenant_id: number + user_id: number + project_id: number | null + entity_type: string + entity_id: string + action: string + trace_id: string + interface_path: string + caller: 'system' | 'user' | 'ai' | string + remark: string + before_data: Record | null + after_data: Record | null + created_at: string | null + created_at_beijing: string | null +} + +export interface AuditOperationsResponse { + items: AuditOperation[] + count: number + tenant_id: number +} + +export interface ListAuditOperationsParams { + tenant_id?: number + entity_type?: string + start_time?: string + end_time?: string + limit?: number +} + +function toQueryString(params: ListAuditOperationsParams): string { + const query = new URLSearchParams() + if (typeof params.tenant_id === 'number') query.set('tenant_id', String(params.tenant_id)) + if (params.entity_type) query.set('entity_type', params.entity_type) + if (params.start_time) query.set('start_time', params.start_time) + if (params.end_time) query.set('end_time', params.end_time) + if (typeof params.limit === 'number') query.set('limit', String(params.limit)) + const rendered = query.toString() + return rendered ? `?${rendered}` : '' +} + /** * 审计代码变更 */ @@ -44,3 +96,12 @@ export async function auditCodeChange(request: CodeAuditRequest): Promise { + const raw = await apiRequest>(`/audit/operations${toQueryString(params)}`, { + method: 'GET', + }) + return unwrap(raw) +} diff --git a/frontend/src/api/projects.ts b/frontend/src/api/projects.ts index 8c1007d6..985ab0b6 100644 --- a/frontend/src/api/projects.ts +++ b/frontend/src/api/projects.ts @@ -53,6 +53,7 @@ export interface ProjectCreateRequest { status?: ProjectStatus domain_id?: string assigned_role?: AssignedRole + member_user_ids?: Array } /** @@ -66,6 +67,13 @@ export interface ProjectUpdateRequest { assigned_role?: AssignedRole } +/** + * 分配项目成员请求（全量替换） + */ +export interface ProjectMembersAssignRequest { + member_user_ids: Array +} + /** * 关联信号请求 */ @@ -254,6 +262,22 @@ export const projectsApi = { }) }, + /** + * 分配项目成员（全量替换） + */ + async assignMembers( + projectId: string, + data: ProjectMembersAssignRequest + ): Promise<{ project_id: string; member_user_ids: string[] }> { + return apiRequest<{ project_id: string; member_user_ids: string[] }>( + `/projects/${projectId}/members`, + { + method: 'PUT', + body: JSON.stringify(data), + } + ) + }, + /** * 从本地路径导入项目 */ diff --git a/frontend/src/api/sstc.ts b/frontend/src/api/sstc.ts index 786100a9..50739d9e 100644 --- a/frontend/src/api/sstc.ts +++ b/frontend/src/api/sstc.ts @@ -201,6 +201,13 @@ export interface SSTCOverview { overloaded_count: number; idle_count: number; member_count: number; + members: Array<{ + user_id: string; + name: string; + role: string; + workload_ratio: number; + status: "idle" | "normal" | "overloaded"; + }>; }; } diff --git a/frontend/src/components/sstc/SSTCGlobalHeader.tsx b/frontend/src/components/sstc/SSTCGlobalHeader.tsx index aa59a473..74fe5758 100644 --- a/frontend/src/components/sstc/SSTCGlobalHeader.tsx +++ b/frontend/src/components/sstc/SSTCGlobalHeader.tsx @@ -5,7 +5,7 @@ import { Button } from "@/components/ui/button"; import { Card, CardContent, CardHeader, CardTitle } from "@/components/ui/card"; import { Badge } from "@/components/ui/badge"; import { compactTitle } from "@/lib/title"; -import { formatProjectNameWithId, toZhStatus } from "@/lib/display"; +import { formatProjectNameWithId, toZhRole, toZhStatus } from "@/lib/display"; import { formatBeijingDateTime } from "@/lib/datetime"; interface SSTCGlobalHeaderProps { @@ -36,6 +36,32 @@ function ratioColor(ratio: number): string { return "#16a34a"; } +function ratioClass(ratio: number): string { + if (ratio > 1.0) return "text-rose-600"; + if (ratio < 0.3) return "text-sky-600"; + return "text-emerald-600"; +} + +function memberStatusLabel(status: "idle" | "normal" | "overloaded"): string { + if (status === "overloaded") return "过载"; + if (status === "idle") return "空闲"; + return "正常"; +} + +function memberStatusClass(status: "idle" | "normal" | "overloaded"): string { + if (status === "overloaded") return "bg-rose-100 text-rose-700"; + if (status === "idle") return "bg-sky-100 text-sky-700"; + return "bg-emerald-100 text-emerald-700"; +} + +function normalizeRoleLabel(rawRole: string): string { + const primaryRole = String(rawRole || "") + .split(",") + .map((item) => item.trim()) + .find(Boolean); + return toZhRole(primaryRole || rawRole); +} + export function SSTCGlobalHeader({ overview, onOpenAi, @@ -46,6 +72,7 @@ export function SSTCGlobalHeader({ const alignmentColor = scoreColor(overview.alignment_score || 0); const progress = overview.task_progress.progress || 0; const ratio = overview.workload_health.avg_ratio || 0; + const workloadMembers = overview.workload_health.members || []; const projectLabel = formatProjectNameWithId({ id: overview.project.id, title: overview.project.title, @@ -133,7 +160,7 @@ export function SSTCGlobalHeader({ 团队负载健康度 - +

{(ratio * 100).toFixed(0)}%

@@ -153,6 +180,36 @@ export function SSTCGlobalHeader({ 覆盖成员 {overview.workload_health.member_count} 人

+
+ {workloadMembers.length === 0 ? ( +

暂无成员负载明细，请先分配项目成员

+ ) : ( + workloadMembers.slice(0, 4).map((member) => ( +
+
+

{member.name}

+

{normalizeRoleLabel(member.role)}

+
+
+ + {(Number(member.workload_ratio || 0) * 100).toFixed(0)}% + + + {memberStatusLabel(member.status)} + +
+
+ )) + )} + {workloadMembers.length > 4 ? ( +

其余 {workloadMembers.length - 4} 人请点击查看完整负载

+ ) : null} +