From 2e358bc2a47f272f92df37deb28330930ef1f91e Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Wed, 11 Mar 2026 14:39:42 +0800 Subject: [PATCH] =?UTF-8?q?=E4=BD=BF=E7=94=A8=E5=8C=97=E4=BA=AC=E6=97=B6?= =?UTF-8?q?=E9=97=B4=E6=9B=B4=E6=96=B0=E6=95=B0=E6=8D=AE=E5=BA=93=E5=86=99?= =?UTF-8?q?=E5=85=A5=E3=80=8D}Sorry=20seems=20odd=20maybe?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .env.example | 4 +- .openclaw/pmflow-routing-rules.json | 11 + CLAUDE.md | 11 + ...0260311_shift_all_timestamps_to_beijing.py | 70 ++++ backend/app/api/ai_pmo.py | 191 +++++++++- backend/app/api/ai_steward.py | 3 +- backend/app/api/projects.py | 209 +++++++++-- backend/app/database.py | 2 + backend/app/services/ai_agent_kernel.py | 263 ++++++++++++++ .../services/ai_orchestration_run_service.py | 3 +- .../app/services/ai_pmo_cadence_service.py | 3 +- backend/app/services/ai_pmo_kernel_service.py | 226 +++++++++++- backend/app/services/ai_steward.py | 89 ++++- .../app/services/feishu_signal_pipeline.py | 77 +++- backend/app/services/llm.py | 12 + .../services/openclaw_commander_service.py | 3 +- .../app/services/project_knowledge_graph.py | 3 +- .../signal_source_orchestration_service.py | 3 +- backend/app/utils/timezone.py | 18 + backend/tests/test_ai_agent_kernel.py | 86 +++++ backend/tests/test_ai_pmo_kernel_service.py | 76 ++++ .../test_ai_pmo_work_item_decision_api.py | 143 ++++++++ .../tests/test_ai_steward_title_extract.py | 31 ++ backend/tests/test_projects_delete_api.py | 338 ++++++++++++++++++ docker-compose.dev.yml | 4 + docker-compose.yml | 4 + frontend/src/api/aiPmo.ts | 12 + frontend/src/pages/InboxPage.tsx | 16 +- frontend/src/types/inbox.ts | 1 + scripts/agent_capability_replay.py | 155 ++++++++ 30 files changed, 2008 insertions(+), 59 deletions(-) create mode 100644 backend/alembic/versions/20260311_shift_all_timestamps_to_beijing.py create mode 100644 backend/app/services/ai_agent_kernel.py create mode 100644 backend/app/utils/timezone.py create mode 100644 backend/tests/test_ai_agent_kernel.py create mode 100644 backend/tests/test_ai_pmo_work_item_decision_api.py create mode 100644 backend/tests/test_ai_steward_title_extract.py create mode 100644 backend/tests/test_projects_delete_api.py create mode 100644 scripts/agent_capability_replay.py diff --git a/.env.example b/.env.example index eddafbe9..01d57b4c 100644 --- a/.env.example +++ b/.env.example @@ -17,11 +17,11 @@ PMFLOW_EXPECTED_DB_DIALECT=postgresql CORS_ORIGINS=http://localhost:5174,http://127.0.0.1:5174 LLM_PROVIDER=glm LLM_FALLBACK_PROVIDER=glm -OPENCLAW_ROUTING_ENABLED=false +OPENCLAW_ROUTING_ENABLED=true OPENCLAW_ROUTING_RULES_PATH=../.openclaw/pmflow-routing-rules.json OPENCLAW_AGENTS_DEFAULT_TIMEOUT_SECONDS=120 LLM_DIRECT_EMERGENCY_ENABLED=true -LLM_MIXED_ROUTING_ENABLED=true +LLM_MIXED_ROUTING_ENABLED=false LLM_ROUTING_DEFAULT_TIER=active LLM_LOW_TO_HIGH_ESCALATION_ENABLED=true LLM_REQUEST_TIMEOUT=90 diff --git a/.openclaw/pmflow-routing-rules.json b/.openclaw/pmflow-routing-rules.json index 18749fb3..30e79615 100644 --- a/.openclaw/pmflow-routing-rules.json +++ b/.openclaw/pmflow-routing-rules.json @@ -4,6 +4,16 @@ "source_of_truth": "openclaw", "description": "PMFlow 三引擎统一路由：所有主链路统一经 OpenClaw OpenAI-Compatible 网关。", "rules": [ + { + "name": "engine_high_intent_complex_exact", + "enabled": true, + "priority": 320, + "tier": "high", + "provider": "openai", + "model": "MiniMax-M2.5", + "api_env": "OPENAI_API_KEY", + "function_exact": "classify_intent_complex" + }, { "name": "engine_low_execute_glm_flash", "enabled": true, @@ -31,6 +41,7 @@ "api_env": "OPENAI_API_KEY", "match_any": [ "adaptive_rag_answer", + "classify_intent_complex", "feishu_project_prefill", "feishu_task_orchestration", "generate_task_breakdown", diff --git a/CLAUDE.md b/CLAUDE.md index 492238d8..570856eb 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -51,6 +51,10 @@ backend/ └── tests/ ``` - 新增：`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/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`） - 移除：`backend/.openclaw/pmflow-routing-rules.json`（路由规则统一到根目录 `.openclaw/`） @@ -144,6 +148,8 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 ## 9. 新增代码的参考模式 - 新增后端接口：参考 `backend/app/api/sstc.py` 的结构 - 配置读取类接口：参考 `backend/app/api/llm_providers.py`（`GET /ai-steward/llm/providers`） +- 项目删除接口（`DELETE /projects/{project_id}`）：参考 `backend/app/api/projects.py`，先清理强依赖外键、再置空弱依赖外键、最后删除主记录（单事务提交） +- 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 - 新增飞书事件处理：参考 `backend/app/services/feishu_integration.py` @@ -163,3 +169,8 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - 环境变量以根目录 `.env` / `.env.example` 为唯一配置源；`backend/.env` 已移除且不再读取 - OpenClaw 路由规则统一由根目录 `.openclaw/pmflow-routing-rules.json` 管理，容器通过 `OPENCLAW_ROUTING_RULES_PATH` 相对路径 `../.openclaw/pmflow-routing-rules.json` 读取 - 模型命名统一使用裸模型名（如 `GLM-4.7`、`MiniMax-M2.5`），不使用供应商前缀模型名 +- 双 Agent 路由治理：默认启用 `OPENCLAW_ROUTING_ENABLED=true`、关闭 `LLM_MIXED_ROUTING_ENABLED`，复杂意图使用 `classify_intent_complex` 走 high-tier +- 写操作链路统一产出 `decision_trace`（intent/tier/confidence/gate_code/correction_action）用于离线回放与分桶评分 +- AI PMO 意图仲裁默认开启（`AI_PMO_INTENT_ARBITRATION_ENABLED=true`）：规则与 LLM 写操作冲突时按 `AI_PMO_INTENT_CONFLICT_POLICY=pending_manual` 强制进入人工确认，不直接自动执行 +- `project.create` 启用槽位完整性闸门：无法从指令提取有效项目名时标记 `slot_incomplete` 并进入人工确认 +- 数据库与核心写入链路统一使用北京时间（`Asia/Shanghai`）：PostgreSQL 会话时区、数据库默认时区与核心 `_utc_now*` 写入入口均已切换到北京时间 diff --git a/backend/alembic/versions/20260311_shift_all_timestamps_to_beijing.py b/backend/alembic/versions/20260311_shift_all_timestamps_to_beijing.py new file mode 100644 index 00000000..ab993abb --- /dev/null +++ b/backend/alembic/versions/20260311_shift_all_timestamps_to_beijing.py @@ -0,0 +1,70 @@ +"""shift all naive timestamps to beijing and set db timezone + +Revision ID: shift_all_timestamps_to_beijing +Revises: add_conversation_sessions_table +Create Date: 2026-03-11 +""" + +from __future__ import annotations + +from alembic import op +import sqlalchemy as sa + + +revision = "shift_all_timestamps_to_beijing" +down_revision = "add_conversation_sessions_table" +branch_labels = None +depends_on = None + + +def _is_postgresql() -> bool: + bind = op.get_bind() + return str(getattr(getattr(bind, "dialect", None), "name", "")).lower().startswith("postgres") + + +def _list_timestamp_columns() -> list[tuple[str, str]]: + bind = op.get_bind() + rows = bind.execute( + sa.text( + """ + SELECT table_name, column_name + FROM information_schema.columns + WHERE table_schema = 'public' + AND data_type = 'timestamp without time zone' + ORDER BY table_name, column_name + """ + ) + ).fetchall() + return [(str(row[0]), str(row[1])) for row in rows] + + +def _shift_all_timestamp_columns(*, hours: int) -> None: + delta_expr = f"INTERVAL '{int(hours)} hours'" if int(hours) >= 0 else f"- INTERVAL '{abs(int(hours))} hours'" + for table_name, column_name in _list_timestamp_columns(): + if int(hours) >= 0: + sql = f'UPDATE "{table_name}" SET "{column_name}" = "{column_name}" + {delta_expr} WHERE "{column_name}" IS NOT NULL' + else: + sql = f'UPDATE "{table_name}" SET "{column_name}" = "{column_name}" {delta_expr} WHERE "{column_name}" IS NOT NULL' + op.execute(sa.text(sql)) + + +def _set_database_timezone(timezone_name: str) -> None: + bind = op.get_bind() + database_name = str(bind.execute(sa.text("SELECT current_database()")).scalar_one()) + role_name = str(bind.execute(sa.text("SELECT current_user")).scalar_one()) + op.execute(sa.text(f'ALTER DATABASE "{database_name}" SET timezone TO \'{timezone_name}\'')) + op.execute(sa.text(f'ALTER ROLE "{role_name}" SET timezone TO \'{timezone_name}\'')) + + +def upgrade() -> None: + if not _is_postgresql(): + return + _set_database_timezone("Asia/Shanghai") + _shift_all_timestamp_columns(hours=8) + + +def downgrade() -> None: + if not _is_postgresql(): + return + _shift_all_timestamp_columns(hours=-8) + _set_database_timezone("UTC") diff --git a/backend/app/api/ai_pmo.py b/backend/app/api/ai_pmo.py index d18fedd1..67cfe490 100644 --- a/backend/app/api/ai_pmo.py +++ b/backend/app/api/ai_pmo.py @@ -3,7 +3,7 @@ from __future__ import annotations from datetime import datetime -from typing import Any, Optional +from typing import Any, Literal, Optional from fastapi import APIRouter, Depends, HTTPException from pydantic import BaseModel, Field @@ -38,6 +38,13 @@ class PMOResumeWorkItemRequest(BaseModel): source_message_id: Optional[str] = Field(default=None, max_length=128) +class PMOWorkItemDecisionRequest(BaseModel): + action: Literal["approve", "reject"] = Field(..., description="审批动作：approve=继续执行，reject=标记失效") + project_id: Optional[int] = Field(default=None) + source_message_id: Optional[str] = Field(default=None, max_length=128) + reason: Optional[str] = Field(default=None, max_length=500) + + @router.post("/process-text") async def process_text( request: PMOProcessTextRequest, @@ -218,7 +225,7 @@ async def resume_work_item( if target is None: raise HTTPException(status_code=404, detail="work_item_not_found") service = AIPMOKernelService(db, tenant_id=tenant_id) - result = await service.process_ingress( + raw_result = await service.process_ingress( source_system="ai_pmo_ui", source_message_id=request.source_message_id or f"resume-{int(work_item_id)}-{int(datetime.now().timestamp() * 1000)}", message_text=f"继续处理 work_item #{int(work_item_id)}", @@ -229,8 +236,54 @@ async def resume_work_item( chat_type="manual", project_hint_id=request.project_id or _safe_int(target.project_id), ) - if not result.get("handled"): - raise HTTPException(status_code=400, detail=str(result.get("reason") or "resume_failed")) + result = _normalize_resume_result(raw=raw_result, target_work_item=target) + return success_response(result) + + +@router.post("/work-items/{work_item_id}/decision") +async def decide_work_item( + work_item_id: int, + request: PMOWorkItemDecisionRequest, + current_user: User = Depends(get_current_user), + db: AsyncSession = Depends(get_db), + tenant_id: int = Depends(get_tenant_id), +): + target = ( + await db.execute( + select(WorkItem).where( + and_( + WorkItem.tenant_id == tenant_id, + WorkItem.id == int(work_item_id), + ) + ).limit(1) + ) + ).scalar_one_or_none() + if target is None: + raise HTTPException(status_code=404, detail="work_item_not_found") + + if request.action == "reject": + result = await _reject_work_item( + db=db, + tenant_id=tenant_id, + target=target, + current_user=current_user, + reason=request.reason, + ) + return success_response(result) + + service = AIPMOKernelService(db, tenant_id=tenant_id) + raw_result = await service.process_ingress( + source_system="ai_pmo_ui", + source_message_id=request.source_message_id or f"decision-approve-{int(work_item_id)}-{int(datetime.now().timestamp() * 1000)}", + message_text=f"继续处理 work_item #{int(work_item_id)}", + sender=str(current_user.name or current_user.username or "PMFlow用户"), + channel_ref="ai_pmo_ui", + source_sender_open_id=None, + captured_at=None, + chat_type="manual", + project_hint_id=request.project_id or _safe_int(target.project_id), + ) + result = _normalize_resume_result(raw=raw_result, target_work_item=target) return success_response(result) @@ -415,3 +468,133 @@ def _serialize_work_item_summary(*, row: WorkItem, ingest: IngestEvent, project: "created_at": row.created_at.isoformat() if row.created_at else None, "updated_at": row.updated_at.isoformat() if row.updated_at else None, } + + +def _normalize_resume_result(*, raw: dict[str, Any], target_work_item: WorkItem) -> dict[str, Any]: + if not bool(raw.get("handled")): + reason = str(raw.get("error") or raw.get("reason") or "继续执行失败").strip() or "继续执行失败" + return { + "handled": False, + "status": "failed_safe", + "followup_reply": reason, + "target_work_item_id": int(target_work_item.id), + "target_state": str(target_work_item.state or "").strip() or None, + "target_intent": str(target_work_item.intent or "").strip() or None, + "project_id": _safe_int(target_work_item.project_id), + "signal_id": _safe_int(target_work_item.signal_id), + "auto_tasks": [], + } + + state = None + rows = raw.get("results") if isinstance(raw.get("results"), list) else [] + if rows and isinstance(rows[0], dict): + state = str(rows[0].get("state") or "").strip() or None + if not state: + state = str(target_work_item.state or "").strip() or None + + if state == "succeeded": + status = "succeeded" + elif state == "waiting_manual": + status = "pending_manual" + elif state == "ignored": + status = "invalidated" + else: + status = "failed_safe" + + followup_reply = str(raw.get("direct_reply_text") or "").strip() + if not followup_reply: + if status == "succeeded": + followup_reply = "已继续执行并完成。" + elif status == "pending_manual": + followup_reply = "继续执行后仍需补充信息或人工确认。" + elif status == "invalidated": + followup_reply = "该工作项已失效，不再继续执行。" + else: + followup_reply = "继续执行失败，请查看详情。" + + linked_project_ids = raw.get("linked_project_ids") if isinstance(raw.get("linked_project_ids"), list) else [] + project_id = _safe_int(linked_project_ids[0]) if linked_project_ids else _safe_int(target_work_item.project_id) + return { + "handled": True, + "status": status, + "followup_reply": followup_reply, + "target_work_item_id": int(target_work_item.id), + "target_state": state, + "target_intent": str(target_work_item.intent or "").strip() or None, + "project_id": project_id, + "signal_id": _safe_int(raw.get("signal_id")) or _safe_int(target_work_item.signal_id), + "auto_tasks": raw.get("auto_tasks") if isinstance(raw.get("auto_tasks"), list) else [], + } + + +async def _reject_work_item( + *, + db: AsyncSession, + tenant_id: int, + target: WorkItem, + current_user: User, + reason: Optional[str], +) -> dict[str, Any]: + current_state = str(target.state or "").strip() + if current_state == "ignored": + return { + "handled": True, + "status": "invalidated", + "followup_reply": "该工作项已处于失效状态。", + "target_work_item_id": int(target.id), + "target_state": "ignored", + "target_intent": str(target.intent or "").strip() or None, + "project_id": _safe_int(target.project_id), + "signal_id": _safe_int(target.signal_id), + "auto_tasks": [], + } + + reject_reason = str(reason or "").strip() or "人工审批拒绝，工作项已失效。" + existing_execution = target.execution_result if isinstance(target.execution_result, dict) else {} + target.state = "ignored" + target.execution_result = { + **existing_execution, + "status": "invalidated", + "manual_decision": "reject", + "manual_reason": reject_reason, + "operator_user_id": _safe_int(current_user.id), + "processed_at": datetime.now().isoformat(), + } + target.review_summary = { + "verified": False, + "issues": ["rejected_by_user"], + "next_step": "该工作项已进入失效状态", + "operator": str(current_user.name or current_user.username or "").strip() or None, + "reason": reject_reason, + } + target.reply_template_key = "action_rejected" + + decision_row = ( + await db.execute( + select(DecisionRecord) + .where( + and_( + DecisionRecord.tenant_id == tenant_id, + DecisionRecord.work_item_id == int(target.id), + ) + ) + .order_by(DecisionRecord.id.desc()) + .limit(1) + ) + ).scalar_one_or_none() + if decision_row is not None: + decision_row.execution_result = target.execution_result + decision_row.final_outcome = "ignored" + + await db.commit() + return { + "handled": True, + "status": "invalidated", + "followup_reply": reject_reason, + "target_work_item_id": int(target.id), + "target_state": "ignored", + "target_intent": str(target.intent or "").strip() or None, + "project_id": _safe_int(target.project_id), + "signal_id": _safe_int(target.signal_id), + "auto_tasks": [], + } diff --git a/backend/app/api/ai_steward.py b/backend/app/api/ai_steward.py index 8378a1f4..db1686ad 100644 --- a/backend/app/api/ai_steward.py +++ b/backend/app/api/ai_steward.py @@ -33,13 +33,14 @@ from ..services.ai_orchestration_run_service import AIOrchestrationRunService from ..services.openclaw_commander_service import OpenClawCommanderService from ..services.signal_source_orchestration_service import SignalSourceOrchestrationService from ..schemas.rag import RAGAskRequest +from ..utils.timezone import now_cn_naive router = APIRouter(prefix="/ai-steward", tags=["AI Steward"]) def _utc_now_naive() -> datetime: - return datetime.now(timezone.utc).replace(tzinfo=None) + return now_cn_naive() async def _ensure_project_signal_link( diff --git a/backend/app/api/projects.py b/backend/app/api/projects.py index d7052759..caf3cbf5 100644 --- a/backend/app/api/projects.py +++ b/backend/app/api/projects.py @@ -7,12 +7,30 @@ from typing import Any, Optional from fastapi import APIRouter, Depends, HTTPException, Query from pydantic import BaseModel -from sqlalchemy import and_, delete, func, select +from sqlalchemy import and_, delete, func, select, update +from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from ..database import get_db from ..dependencies import get_current_user, get_tenant_id, require_roles -from ..models import Project, ProjectSignalLink, Signal, Task, User +from ..models import ( + AIOrchestrationRun, + AIWorkerExecution, + Budget, + ConversationSession, + CostTracking, + Experiment, + OperationExecutionLedger, + OperationLog, + Project, + ProjectSignalLink, + Signal, + SignalSourceRegistry, + Task, + User, + UserProject, + WorkItem, +) 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 @@ -154,6 +172,148 @@ async def _get_project_or_404( return project +async def _cleanup_project_dependencies( + db: AsyncSession, + *, + tenant_id: int, + project_id: int, +) -> None: + task_ids_subquery = ( + select(Task.id) + .where( + and_( + Task.tenant_id == tenant_id, + Task.project_id == project_id, + ) + ) + .scalar_subquery() + ) + + # 先断开任务子依赖，避免删除 tasks 时触发约束错误。 + await db.execute( + update(CostTracking) + .where( + and_( + CostTracking.tenant_id == tenant_id, + CostTracking.task_id.in_(task_ids_subquery), + ) + ) + .values(task_id=None) + ) + + # 非空外键：必须先删依赖记录。 + await db.execute( + delete(AIWorkerExecution).where( + and_( + AIWorkerExecution.tenant_id == tenant_id, + AIWorkerExecution.project_id == project_id, + ) + ) + ) + await db.execute( + delete(AIOrchestrationRun).where( + and_( + AIOrchestrationRun.tenant_id == tenant_id, + AIOrchestrationRun.project_id == project_id, + ) + ) + ) + await db.execute( + delete(SignalSourceRegistry).where( + and_( + SignalSourceRegistry.tenant_id == tenant_id, + SignalSourceRegistry.project_id == project_id, + ) + ) + ) + await db.execute( + delete(ProjectSignalLink).where( + and_( + ProjectSignalLink.tenant_id == tenant_id, + ProjectSignalLink.project_id == project_id, + ) + ) + ) + await db.execute( + delete(Task).where( + and_( + Task.tenant_id == tenant_id, + Task.project_id == project_id, + ) + ) + ) + await db.execute( + delete(UserProject).where( + and_( + UserProject.tenant_id == tenant_id, + UserProject.project_id == project_id, + ) + ) + ) + + # 可空外键：保留记录，将 project 解绑为 null。 + await db.execute( + update(WorkItem) + .where( + and_( + WorkItem.tenant_id == tenant_id, + WorkItem.project_id == project_id, + ) + ) + .values(project_id=None) + ) + await db.execute( + update(OperationLog) + .where( + and_( + OperationLog.tenant_id == tenant_id, + OperationLog.project_id == project_id, + ) + ) + .values(project_id=None) + ) + await db.execute( + update(OperationExecutionLedger) + .where( + and_( + OperationExecutionLedger.tenant_id == tenant_id, + OperationExecutionLedger.project_id == project_id, + ) + ) + .values(project_id=None) + ) + await db.execute( + update(Experiment) + .where( + and_( + Experiment.tenant_id == tenant_id, + Experiment.project_id == project_id, + ) + ) + .values(project_id=None) + ) + await db.execute( + update(Budget) + .where( + and_( + Budget.tenant_id == tenant_id, + Budget.project_id == project_id, + ) + ) + .values(project_id=None) + ) + await db.execute( + update(ConversationSession) + .where( + and_( + ConversationSession.tenant_id == tenant_id, + ConversationSession.current_project_id == project_id, + ) + ) + .values(current_project_id=None) + ) + + @router.get("/") async def list_projects( summary: Optional[int] = Query(default=0), @@ -270,33 +430,30 @@ async def delete_project( ): tenant_id = int(current_user.tenant_id) _ = current_user - await _get_project_or_404(db, tenant_id=tenant_id, project_id=project_id) - await db.execute( - delete(ProjectSignalLink).where( - and_( - ProjectSignalLink.tenant_id == tenant_id, - ProjectSignalLink.project_id == project_id, + try: + await _get_project_or_404(db, tenant_id=tenant_id, project_id=project_id) + await _cleanup_project_dependencies( + db, + tenant_id=tenant_id, + project_id=project_id, + ) + await db.execute( + delete(Project).where( + and_( + Project.tenant_id == tenant_id, + Project.id == project_id, + ) ) ) - ) - await db.execute( - delete(Task).where( - and_( - Task.tenant_id == tenant_id, - Task.project_id == project_id, - ) - ) - ) - await db.execute( - delete(Project).where( - and_( - Project.tenant_id == tenant_id, - Project.id == project_id, - ) - ) - ) - await db.commit() + await db.commit() + except IntegrityError as exc: + await db.rollback() + raise HTTPException(status_code=409, detail="Project has related records and cannot be deleted") 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=project_id) return {"message": "Project deleted"} diff --git a/backend/app/database.py b/backend/app/database.py index 3d60e8b9..f8fa98b4 100644 --- a/backend/app/database.py +++ b/backend/app/database.py @@ -40,6 +40,8 @@ if not _is_sqlite(): "max_overflow": max(0, int(settings.db_max_overflow)), "pool_timeout": max(5, int(settings.db_pool_timeout)), "pool_recycle": max(60, int(settings.db_pool_recycle)), + # 强制数据库会话时区为北京时间，确保 now()/CURRENT_TIMESTAMP 为 Asia/Shanghai。 + "connect_args": {"server_settings": {"timezone": "Asia/Shanghai"}}, } ) diff --git a/backend/app/services/ai_agent_kernel.py b/backend/app/services/ai_agent_kernel.py new file mode 100644 index 00000000..3e1f5d99 --- /dev/null +++ b/backend/app/services/ai_agent_kernel.py @@ -0,0 +1,263 @@ +"""Shared decision kernel for dual-agent routing and guardrails.""" + +from __future__ import annotations + +import os +import re +from dataclasses import dataclass +from typing import Iterable, Sequence + +BOOL_TRUE_VALUES = {"1", "true", "yes", "y", "on", "是"} +INTENT_FUNCTION_SIMPLE = "classify_intent_ai_pmo_kernel" +INTENT_FUNCTION_COMPLEX = "classify_intent_complex" + +BACKGROUND_MARKERS = ("因为", "所以", "用来", "为了", "方便", "希望", "背景", "场景") +MULTI_SEGMENT_MARKERS = (",", "，", "、", ";", "；", "并且", "同时", "然后") +INTENT_GROUP_KEYWORDS: dict[str, tuple[str, ...]] = { + "project": ("项目", "立项", "创建项目", "新增项目", "新建项目"), + "signal": ("信号", "信号源", "接入", "onboard", "source"), + "task": ("任务", "分配", "指派", "执行", "部署"), + "change": ("修改", "更新", "改为", "改成", "撤销", "回滚", "归档"), +} + +CORRECTION_ACTION_PATTERNS: tuple[tuple[str, tuple[str, ...]], ...] = ( + ("undo_last", ("撤销", "撤回", "undo", "作废上一步", "取消上一步")), + ("rollback", ("回滚", "恢复到", "还原", "回退")), + ("replace_target", ("改为", "改成", "替换为", "不是", "不对", "纠正")), + ("reroute", ("重新识别", "重新判断", "重新路由", "再判断一次")), +) + + +def _env_float(name: str, default: float) -> float: + raw = os.getenv(name) + if raw is None: + return float(default) + try: + value = float(str(raw).strip()) + except Exception: + return float(default) + return float(default) if value < 0 else value + + +def _env_bool(name: str, default: bool = False) -> bool: + raw = os.getenv(name) + if raw is None: + return default + return str(raw).strip().lower() in BOOL_TRUE_VALUES + + +def _normalize_text(text: str) -> str: + return re.sub(r"\s+", "", str(text or "")).lower() + + +def _normalize_name(name: str) -> str: + return re.sub(r"[\s_\-]+", "", str(name or "")).lower() + + +@dataclass(slots=True) +class IntentRouteDecision: + function_name: str + tier: str + is_complex: bool + reasons: list[str] + + +@dataclass(slots=True) +class GateDecision: + action: str + gate_code: str + reason: str + auto_execute: bool + + +@dataclass(slots=True) +class CorrectionDecision: + action: str + confidence: float + matched_keywords: list[str] + + +def _count_intent_groups(text: str) -> int: + normalized = _normalize_text(text) + hits = 0 + for tokens in INTENT_GROUP_KEYWORDS.values(): + if any(_normalize_name(token) in normalized for token in tokens): + hits += 1 + return hits + + +def is_complex_instruction(text: str, *, project_names: Sequence[str] | None = None) -> tuple[bool, list[str]]: + normalized = _normalize_text(text) + if not normalized: + return False, [] + + reasons: list[str] = [] + length_threshold = int(_env_float("AGENT_COMPLEX_LENGTH_THRESHOLD", 20)) + if len(normalized) > length_threshold: + reasons.append("length_gt_threshold") + + has_multi_segment = any(marker in text for marker in MULTI_SEGMENT_MARKERS) + if has_multi_segment: + reasons.append("multi_segment") + + has_background = any(marker in text for marker in BACKGROUND_MARKERS) + if has_background: + reasons.append("has_background") + + intent_group_hits = _count_intent_groups(text) + if intent_group_hits >= 3 or ( + intent_group_hits >= 2 + and ( + len(normalized) > max(14, length_threshold) + or has_multi_segment + or has_background + ) + ): + reasons.append("multi_intent_groups") + + for name in project_names or []: + candidate = _normalize_name(name) + if len(candidate) >= 2 and candidate in normalized: + reasons.append("contains_project_name") + break + + return bool(reasons), reasons + + +def select_intent_route(text: str, *, project_names: Sequence[str] | None = None) -> IntentRouteDecision: + is_complex, reasons = is_complex_instruction(text, project_names=project_names) + if is_complex: + return IntentRouteDecision( + function_name=INTENT_FUNCTION_COMPLEX, + tier="high", + is_complex=True, + reasons=reasons, + ) + return IntentRouteDecision( + function_name=INTENT_FUNCTION_SIMPLE, + tier="low", + is_complex=False, + reasons=["default_simple"], + ) + + +def resolve_correction(text: str) -> CorrectionDecision: + normalized = _normalize_text(text) + if not normalized: + return CorrectionDecision(action="none", confidence=0.0, matched_keywords=[]) + + for action, tokens in CORRECTION_ACTION_PATTERNS: + matched = [token for token in tokens if _normalize_name(token) in normalized] + if matched: + confidence = 0.9 if len(matched) >= 2 else 0.78 + return CorrectionDecision( + action=action, + confidence=float(min(1.0, confidence)), + matched_keywords=matched, + ) + return CorrectionDecision(action="none", confidence=0.0, matched_keywords=[]) + + +def gate_execution( + *, + confidence: float, + risk_level: str = "medium", + correction_action: str = "none", + force_manual: bool = False, + force_manual_code: str = "manual_review_required", + force_manual_reason: str = "", +) -> GateDecision: + auto_threshold = _env_float("AGENT_GATE_AUTO_THRESHOLD", 0.82) + review_threshold = _env_float("AGENT_GATE_REVIEW_THRESHOLD", 0.60) + strict_high_risk = _env_bool("AGENT_GATE_STRICT_HIGH_RISK", True) + require_review_below_auto = _env_bool("AGENT_GATE_REVIEW_BELOW_AUTO", False) + normalized_risk = str(risk_level or "medium").strip().lower() + safe_confidence = max(0.0, min(1.0, float(confidence or 0.0))) + + if force_manual: + return GateDecision( + action="pending_manual", + gate_code=str(force_manual_code or "manual_review_required").strip() or "manual_review_required", + reason=str(force_manual_reason or "命中人工复核策略，需人工确认后执行。").strip(), + auto_execute=False, + ) + + if correction_action in {"undo_last", "rollback", "replace_target", "reroute"}: + return GateDecision( + action="pending_manual", + gate_code="correction_review", + reason="检测到修正/回滚语义，需人工确认后执行。", + auto_execute=False, + ) + + if safe_confidence < review_threshold: + return GateDecision( + action="reject", + gate_code="confidence_too_low", + reason="置信度低于安全阈值，已阻断自动执行。", + auto_execute=False, + ) + + if normalized_risk in {"high", "critical"} and strict_high_risk and safe_confidence < max(auto_threshold, 0.9): + return GateDecision( + action="pending_manual", + gate_code="high_risk_manual_review", + reason="高风险写操作需要更高置信度，当前进入待人工确认。", + auto_execute=False, + ) + + if safe_confidence < auto_threshold and require_review_below_auto: + return GateDecision( + action="pending_manual", + gate_code="confidence_review_required", + reason="置信度未达自动执行阈值，进入待人工确认。", + auto_execute=False, + ) + + return GateDecision( + action="auto_execute", + gate_code="auto_execute_allowed", + reason="置信度达标，允许自动执行。", + auto_execute=True, + ) + + +def build_decision_trace( + *, + intent: str, + confidence: float, + route_tier: str, + route_function: str, + gate_action: str, + gate_code: str, + correction_action: str = "none", + correction_keywords: Iterable[str] | None = None, + complexity_reasons: Iterable[str] | None = None, +) -> dict[str, object]: + return { + "intent": str(intent or "").strip(), + "confidence": round(max(0.0, min(1.0, float(confidence or 0.0))), 4), + "route_tier": str(route_tier or "").strip(), + "route_function": str(route_function or "").strip(), + "gate_action": str(gate_action or "").strip(), + "gate_code": str(gate_code or "").strip(), + "correction_action": str(correction_action or "none").strip(), + "correction_keywords": [str(token) for token in (correction_keywords or []) if str(token).strip()], + "complexity_reasons": [str(token) for token in (complexity_reasons or []) if str(token).strip()], + } + + +def compute_capability_score( + *, + intent_accuracy: float, + gate_precision: float, + exec_success: float, + correction_success: float, +) -> float: + score = ( + 0.30 * max(0.0, min(1.0, float(intent_accuracy or 0.0))) + + 0.30 * max(0.0, min(1.0, float(gate_precision or 0.0))) + + 0.25 * max(0.0, min(1.0, float(exec_success or 0.0))) + + 0.15 * max(0.0, min(1.0, float(correction_success or 0.0))) + ) + return round(score, 4) diff --git a/backend/app/services/ai_orchestration_run_service.py b/backend/app/services/ai_orchestration_run_service.py index 3b9b44d1..946311c6 100644 --- a/backend/app/services/ai_orchestration_run_service.py +++ b/backend/app/services/ai_orchestration_run_service.py @@ -9,10 +9,11 @@ from sqlalchemy import and_, select from sqlalchemy.ext.asyncio import AsyncSession from ..models import AIOrchestrationRun +from ..utils.timezone import now_cn_naive def _utc_now_naive() -> datetime: - return datetime.now(timezone.utc).replace(tzinfo=None) + return now_cn_naive() class AIOrchestrationRunService: diff --git a/backend/app/services/ai_pmo_cadence_service.py b/backend/app/services/ai_pmo_cadence_service.py index c989c8b6..dbfb796d 100644 --- a/backend/app/services/ai_pmo_cadence_service.py +++ b/backend/app/services/ai_pmo_cadence_service.py @@ -9,10 +9,11 @@ from sqlalchemy import and_, func, or_, select from sqlalchemy.ext.asyncio import AsyncSession from ..models import Budget, Notification, PortfolioSnapshot, Project, ProjectDomain, ProjectSignalLink, Signal, SignalSourceRegistry, Task, UserProject, UserWorkload, WorkItem +from ..utils.timezone import now_cn_naive def _utc_now_naive() -> datetime: - return datetime.now(timezone.utc).replace(tzinfo=None) + return now_cn_naive() class AIPMOCadenceService: diff --git a/backend/app/services/ai_pmo_kernel_service.py b/backend/app/services/ai_pmo_kernel_service.py index 71b7812f..043feb8d 100644 --- a/backend/app/services/ai_pmo_kernel_service.py +++ b/backend/app/services/ai_pmo_kernel_service.py @@ -57,6 +57,12 @@ from ..schemas.world_model import ( WorldModel, ) from .adaptive_rag_service import AdaptiveRAGService +from .ai_agent_kernel import ( + build_decision_trace, + gate_execution as kernel_gate_execution, + resolve_correction as kernel_resolve_correction, + select_intent_route, +) from .ai_pmo_cadence_service import AIPMOCadenceService from .ai_steward import AIStewardService from .ai_pmo_review_service import AIPMOReviewService @@ -66,6 +72,7 @@ from .openclaw_commander_service import OpenClawCommanderService from .ai_pmo_domain_workflow import SignalSourceDomainWorkflow from .llm import llm_service from .signal_source_orchestration_service import SignalSourceOrchestrationService +from ..utils.timezone import now_cn_naive logger = get_logger(__name__) @@ -195,13 +202,21 @@ INTENT_CANDIDATES = ( INTENT_FOLLOWUP_CONTINUE, INTENT_IGNORE, ) +WRITE_INTENTS = { + INTENT_PROJECT_CREATE, + INTENT_SIGNAL_CREATE, + INTENT_SIGNAL_SOURCE_ONBOARD, + INTENT_FOLLOWUP_CONFIRM, + INTENT_FOLLOWUP_CONTINUE, +} +INTENT_CONFLICT_POLICIES = {"pending_manual", "prefer_rule", "prefer_llm"} WORLD_MODEL_STATIC_CACHE_TTL_SECONDS = 10 * 60 WORLD_MODEL_STATIC_CACHE_KEY_PREFIX = "wm:static" def _utc_now_naive() -> datetime: - return datetime.now(timezone.utc).replace(tzinfo=None) + return now_cn_naive() def _env_bool(name: str, default: bool = False) -> bool: @@ -1152,6 +1167,32 @@ class AIPMOKernelService: world_model=world_model, message_text=clean_text, ) + correction_action = str((intent.metadata or {}).get("correction_action") or "none").strip() or "none" + requires_manual_review = bool((intent.metadata or {}).get("requires_manual_review")) + manual_review_code = str((intent.metadata or {}).get("manual_review_code") or "manual_review_required").strip() + manual_review_reason = str((intent.metadata or {}).get("manual_review_reason") or "").strip() + llm_function = str((intent.metadata or {}).get("llm_function") or INTENT_LLM_FUNCTION_NAME).strip() + route_tier = str((intent.metadata or {}).get("route_tier") or "rule").strip() + route_reasons = (intent.metadata or {}).get("route_reasons") + gate = kernel_gate_execution( + confidence=float(intent.confidence or 0.0), + risk_level=self._risk_level_for_intent(intent.name), + correction_action=correction_action, + force_manual=requires_manual_review, + force_manual_code=manual_review_code, + force_manual_reason=manual_review_reason, + ) + decision_trace = build_decision_trace( + intent=intent.name, + confidence=float(intent.confidence or 0.0), + route_tier=route_tier, + route_function=llm_function, + gate_action=gate.action, + gate_code=gate.gate_code, + correction_action=correction_action, + correction_keywords=(intent.metadata or {}).get("correction_keywords") if isinstance(intent.metadata, dict) else [], + complexity_reasons=route_reasons if isinstance(route_reasons, list) else [], + ) work_item.state = STATE_DECIDED work_item.decision_level = str((decision_payload["strategic"] or {}).get("decision_level") or "portfolio") @@ -1174,12 +1215,22 @@ class AIPMOKernelService: ) policy = decision_payload["policy"] + if not gate.auto_execute: + policy["auto_execute"] = False + policy["policy_gate_hit"] = True + policy["policy_gate_code"] = gate.gate_code + policy["approval_requirement"] = { + "required": True, + "reason": gate.gate_code, + "message": gate.reason, + } if not bool(policy.get("auto_execute")): work_item.state = STATE_WAITING_MANUAL work_item.action_bundle = decision_payload["execution_plan"] work_item.execution_result = { "status": "waiting_manual", "reason": ((policy.get("approval_requirement") or {}).get("message") or "").strip(), + "decision_trace": decision_trace, } manual_issues = ["waiting_manual"] policy_gate_code = str(policy.get("policy_gate_code") or "").strip() @@ -1276,6 +1327,8 @@ class AIPMOKernelService: source_message_id=synthetic_message_id, ) verification = await self._verify_execution(intent=intent, execution_result=execution_result, world_model=world_model) + if isinstance(execution_result, dict): + execution_result.setdefault("decision_trace", decision_trace) work_item.execution_result = execution_result if _safe_int(execution_result.get("project_id")) is not None: work_item.project_id = _safe_int(execution_result.get("project_id")) @@ -1841,6 +1894,121 @@ class AIPMOKernelService: def _intent_llm_enabled() -> bool: return _env_bool("AI_PMO_INTENT_LLM_ENABLED", True) + @staticmethod + def _intent_arbitration_enabled() -> bool: + return _env_bool("AI_PMO_INTENT_ARBITRATION_ENABLED", True) + + @staticmethod + def _intent_conflict_policy() -> str: + raw = str(os.getenv("AI_PMO_INTENT_CONFLICT_POLICY", "pending_manual") or "").strip().lower() + return raw if raw in INTENT_CONFLICT_POLICIES else "pending_manual" + + @staticmethod + def _copy_intent_with_metadata(intent: CanonicalIntent, **metadata_updates: Any) -> CanonicalIntent: + metadata = dict(intent.metadata or {}) + metadata.update(metadata_updates) + return CanonicalIntent( + intent.name, + float(intent.confidence or 0.0), + intent.project_id, + intent.signal_title, + intent.signal_summary, + metadata, + ) + + @staticmethod + def _is_default_rule_query(intent: CanonicalIntent) -> bool: + if intent.name != INTENT_QUERY: + return False + metadata = intent.metadata if isinstance(intent.metadata, dict) else {} + return bool(metadata.get("fallback")) and str(metadata.get("reason") or "") == "default_query_fallback" + + @staticmethod + def _has_required_slots(*, intent: CanonicalIntent, message_text: str) -> tuple[bool, dict[str, Any]]: + if intent.name != INTENT_PROJECT_CREATE: + return True, {} + extracted_title = AIStewardService._extract_project_title_from_instruction(message_text) + has_title = bool(str(extracted_title or "").strip()) + return has_title, { + "slot_check": "project_title", + "slot_ok": has_title, + "extracted_project_title": extracted_title[:50] if extracted_title else "", + } + + def _apply_slot_guard(self, *, intent: CanonicalIntent, message_text: str) -> CanonicalIntent: + slot_ok, slot_meta = self._has_required_slots(intent=intent, message_text=message_text) + if slot_ok: + if slot_meta: + return self._copy_intent_with_metadata(intent, **slot_meta) + return intent + return self._copy_intent_with_metadata( + intent, + **slot_meta, + requires_manual_review=True, + manual_review_code="slot_incomplete", + manual_review_reason="关键信息不完整，需人工确认后执行。", + ) + + def _arbitrate_intent( + self, + *, + rule_intent: CanonicalIntent, + llm_intent: CanonicalIntent, + message_text: str, + ) -> tuple[CanonicalIntent, str]: + if rule_intent.name == llm_intent.name: + return self._copy_intent_with_metadata( + llm_intent, + arbitration="consensus", + arbitration_rule_intent=rule_intent.name, + arbitration_llm_intent=llm_intent.name, + ), "llm" + + policy = self._intent_conflict_policy() + rule_is_write = rule_intent.name in WRITE_INTENTS + llm_is_write = llm_intent.name in WRITE_INTENTS + has_write_conflict = rule_is_write or llm_is_write + prefer_rule_by_default = ( + not self._is_default_rule_query(rule_intent) + and ( + rule_intent.confidence >= llm_intent.confidence + or self._is_default_rule_query(llm_intent) + ) + ) + + if policy == "prefer_rule": + chosen = rule_intent + source = "rule_fallback" + elif policy == "prefer_llm": + chosen = llm_intent + source = "llm" + else: + chosen = rule_intent if prefer_rule_by_default else llm_intent + source = "rule_fallback" if chosen is rule_intent else "llm" + + if has_write_conflict and policy == "pending_manual": + chosen = self._copy_intent_with_metadata( + chosen, + arbitration="conflict_manual_review", + arbitration_rule_intent=rule_intent.name, + arbitration_llm_intent=llm_intent.name, + arbitration_rule_confidence=float(rule_intent.confidence or 0.0), + arbitration_llm_confidence=float(llm_intent.confidence or 0.0), + requires_manual_review=True, + manual_review_code="intent_conflict_write", + manual_review_reason="规则与LLM意图冲突，且涉及写操作，需人工确认后执行。", + ) + return chosen, source + + return self._copy_intent_with_metadata( + chosen, + arbitration="prefer_rule" if source == "rule_fallback" else "prefer_llm", + arbitration_rule_intent=rule_intent.name, + arbitration_llm_intent=llm_intent.name, + arbitration_rule_confidence=float(rule_intent.confidence or 0.0), + arbitration_llm_confidence=float(llm_intent.confidence or 0.0), + ), source + @classmethod def _is_short_confirmation_message( cls, @@ -1945,6 +2113,13 @@ class AIPMOKernelService: session: Optional[ConversationSession], world_model: dict[str, Any], ) -> Optional[CanonicalIntent]: + project_snapshot = world_model.get("project") if isinstance(world_model.get("project"), dict) else {} + project_names = [ + str(project_snapshot.get("title") or "").strip(), + str(project_snapshot.get("name") or "").strip(), + ] + route_decision = select_intent_route(message_text, project_names=project_names) + correction_decision = kernel_resolve_correction(message_text) 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 "", @@ -1958,7 +2133,7 @@ class AIPMOKernelService: prompt, db=self.db, tenant_id=self.tenant_id, - function_name=INTENT_LLM_FUNCTION_NAME, + function_name=route_decision.function_name, use_prompt_profile=False, temperature=0.0, max_tokens=260, @@ -2002,6 +2177,17 @@ class AIPMOKernelService: 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, @@ -2010,7 +2196,13 @@ class AIPMOKernelService: metadata={ "context_source": "llm", "reasoning": reasoning[:300], - "llm_function": INTENT_LLM_FUNCTION_NAME, + "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, }, ) @@ -2022,9 +2214,10 @@ class AIPMOKernelService: world_model: dict[str, Any], ) -> tuple[CanonicalIntent, str]: project_id = _safe_int(world_model.get("resolved_project_id")) + rule_intent = self.intent_resolver.resolve(text=message_text, project_id=project_id) if self._is_short_confirmation_message(message_text=message_text, session=session): if session is None: - return self.intent_resolver.resolve(text=message_text, project_id=project_id), "rule_fallback" + return rule_intent, "rule_fallback" return ( self._build_short_confirm_intent( message_text=message_text, @@ -2041,9 +2234,17 @@ class AIPMOKernelService: world_model=world_model, ) if llm_intent is not None: - return llm_intent, "llm" + if self._intent_arbitration_enabled(): + selected_intent, source = self._arbitrate_intent( + rule_intent=rule_intent, + llm_intent=llm_intent, + message_text=message_text, + ) + else: + selected_intent, source = llm_intent, "llm" + return self._apply_slot_guard(intent=selected_intent, message_text=message_text), source - return self.intent_resolver.resolve(text=message_text, project_id=project_id), "rule_fallback" + return self._apply_slot_guard(intent=rule_intent, message_text=message_text), "rule_fallback" async def _build_portfolio_summary( self, @@ -2983,6 +3184,11 @@ class AIPMOKernelService: "work_item_id": int(work_item.id), "decision_level": work_item.decision_level, "review_summary": work_item.review_summary, + "decision_trace": ( + execution_result.get("decision_trace") + if isinstance(execution_result, dict) + else ((intent.metadata or {}).get("decision_trace") if isinstance(intent.metadata, dict) else None) + ), } if intent.name == INTENT_QUERY and isinstance(execution_result.get("rag_result"), dict): result["rag_result"] = execution_result["rag_result"] @@ -3009,6 +3215,14 @@ class AIPMOKernelService: return "work_item.resume" return None + @staticmethod + def _risk_level_for_intent(intent_name: str) -> str: + if intent_name in {INTENT_PROJECT_CREATE, INTENT_SIGNAL_SOURCE_ONBOARD}: + return "high" + if intent_name in {INTENT_SIGNAL_CREATE, INTENT_FOLLOWUP_CONFIRM, INTENT_FOLLOWUP_CONTINUE}: + return "medium" + return "low" + @staticmethod def _extract_followup_work_item_id(message: str) -> Optional[int]: match = FOLLOWUP_WORK_ITEM_ID_RE.search(str(message or "")) diff --git a/backend/app/services/ai_steward.py b/backend/app/services/ai_steward.py index 939eb742..a6cbdf92 100644 --- a/backend/app/services/ai_steward.py +++ b/backend/app/services/ai_steward.py @@ -4,6 +4,7 @@ from __future__ import annotations import asyncio import json +import re from typing import Any, Dict, List from sqlalchemy.ext.asyncio import AsyncSession @@ -13,6 +14,7 @@ from ..database import get_db from ..models import CognitiveConflict, Project, ProjectSignalLink, Signal, TruthNode from ..utils.project_members import ensure_project_default_members from ..utils.sequence import ensure_project_id_sequence +from ..utils.title import compact_title from .adaptive_rag_service import AdaptiveRAGService from .llm import LLMService from .task_persistence import ( @@ -33,6 +35,71 @@ class AIStewardService: self.db = db self.llm = LLMService() + @staticmethod + def _normalize_project_title_candidate(candidate: str) -> str: + cleaned = re.sub(r"\s+", " ", str(candidate or "")).strip(" []【】()（）:：\"'“”") + if not cleaned: + return "" + normalized = re.sub(r"\s+", "", cleaned).lower() + generic_tokens = { + "项目", + "一个项目", + "新项目", + "项目列表", + "项目库", + "pmflow", + "pmflow项目列表", + } + if normalized in generic_tokens: + return "" + if any(token in normalized for token in ("创建项目", "新建项目", "新增项目", "不是创建信号")): + return "" + return compact_title(cleaned, max_len=24) + + @classmethod + def _extract_project_title_from_instruction(cls, text: str) -> str: + cleaned = re.sub(r"\s+", " ", str(text or "")).strip() + if not cleaned: + return "" + # 优先去掉“不是X，是Y”中的前置纠正片段，避免把纠正话术落到标题里。 + cleaned = re.sub(r"^.*?不是[^，。,;；]*信号[^，。,;；]*[，,;； ]*是", "", cleaned) + quoted_match = re.search(r"(?:项目|新项目)[^\"“”']{0,10}[\"“”']([^\"“”']{2,40})[\"“”']", cleaned) + if quoted_match: + return cls._normalize_project_title_candidate(quoted_match.group(1)) + for pattern in ( + 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"(?:因为|所以|用来|为了|方便).*$", "", candidate).strip() + normalized = cls._normalize_project_title_candidate(candidate) + if normalized: + return normalized + return "" + + @staticmethod + def _looks_like_instruction_sentence(text: str) -> bool: + normalized = re.sub(r"\s+", "", str(text or "")).lower() + if not normalized: + return False + return "项目" in normalized and any( + token in normalized + for token in ("创建", "新建", "新增", "增加", "添加", "帮我", "给我", "不是", "信号") + ) + + @classmethod + def _compact_project_description(cls, *, input_text: str, title: str, draft_description: str) -> str: + raw = str(draft_description or "").strip() + if raw and not cls._looks_like_instruction_sentence(raw): + return raw[:4000] + if title: + return f"项目：{title}"[:4000] + return str(input_text or "").strip()[:4000] + async def create_project_from_natural_language( self, input_text: str, @@ -207,10 +274,26 @@ class AIStewardService: } ) + extracted_title = self._extract_project_title_from_instruction(input_text) + draft_title = str(project_draft.get("title") or "").strip() + if self._looks_like_instruction_sentence(draft_title): + draft_title = "" + normalized_title = ( + self._normalize_project_title_candidate(draft_title) + or extracted_title + or compact_title(str(input_text or "").strip(), max_len=24) + or "未命名项目" + ) + draft_name = str(project_draft.get("name") or "").strip() + normalized_name = self._normalize_project_title_candidate(draft_name) or normalized_title normalized_project = { - "name": str(project_draft.get("name") or "未命名项目").strip()[:50] or "未命名项目", - "title": str(project_draft.get("title") or input_text[:50] or "未命名项目").strip()[:50] or "未命名项目", - "description": str(project_draft.get("description") or input_text).strip()[:4000] or input_text[:4000], + "name": normalized_name[:50] or "未命名项目", + "title": normalized_title[:50] or "未命名项目", + "description": self._compact_project_description( + input_text=input_text, + title=normalized_title, + draft_description=str(project_draft.get("description") or ""), + ), "assigned_role": str(project_draft.get("assigned_role") or "backend").strip()[:50] or "backend", "domain_id": analysis.get("domain_id"), } diff --git a/backend/app/services/feishu_signal_pipeline.py b/backend/app/services/feishu_signal_pipeline.py index 1759756f..a93a0810 100644 --- a/backend/app/services/feishu_signal_pipeline.py +++ b/backend/app/services/feishu_signal_pipeline.py @@ -33,10 +33,16 @@ from ..models import ( ) from .llm import LLMService from .adaptive_rag_service import AdaptiveRAGService +from .ai_agent_kernel import ( + build_decision_trace, + gate_execution as kernel_gate_execution, + resolve_correction as kernel_resolve_correction, +) from .notification_service import NotificationService from .openclaw_commander_service import OpenClawCommanderService, OperationExecutorService from ..utils.project_members import ensure_project_default_members from ..utils.sequence import ensure_project_id_sequence +from ..utils.timezone import CN_TZ, now_cn_aware logger = get_logger(__name__) @@ -344,7 +350,7 @@ def _first_non_empty(data: dict[str, Any], keys: tuple[str, ...]) -> Any: def _utc_now() -> datetime: - return datetime.now(timezone.utc) + return now_cn_aware() def _compact_text(value: Any, max_len: int = 160) -> str: @@ -366,7 +372,7 @@ def _build_undo_deadline(value: Optional[datetime], *, minutes: int = 10) -> Opt if value is None: return None if value.tzinfo is not None: - value = value.astimezone(timezone.utc).replace(tzinfo=None) + value = value.astimezone(CN_TZ).replace(tzinfo=None) return value + timedelta(minutes=minutes) @@ -570,6 +576,28 @@ class FeishuSignalPipelineService: correction_match = recall_rows[0] except Exception as exc: logger.warning(f"intent_correction_recall_failed: tenant_id={self.tenant_id}, error={exc}") + correction_decision = kernel_resolve_correction(text) + if correction_decision.action != "none": + decision_trace = build_decision_trace( + intent=INTENT_CLARIFICATION_REQUIRED, + confidence=correction_decision.confidence, + route_tier="rule", + route_function="_resolve_intent_gate", + gate_action="pending_manual", + gate_code="correction_review", + correction_action=correction_decision.action, + correction_keywords=correction_decision.matched_keywords, + complexity_reasons=[], + ) + return { + "intent": INTENT_CLARIFICATION_REQUIRED, + "canonical_action": "clarify", + "clarification_template": "检测到修正/回滚语义，已进入待确认队列。请明确目标对象后继续执行。", + "risk_level": "high", + "fields": {}, + "correction_match": correction_match, + "decision_trace": decision_trace, + } if self._is_healthcheck_message(text): return { @@ -1578,7 +1606,7 @@ class FeishuSignalPipelineService: ) -> Optional[int]: occurred_at = captured_at or _utc_now() if occurred_at.tzinfo is not None: - occurred_at = occurred_at.astimezone(timezone.utc).replace(tzinfo=None) + occurred_at = occurred_at.astimezone(CN_TZ).replace(tzinfo=None) item = FeishuItem( tenant_id=self.tenant_id, feishu_message_id=feishu_message_id, @@ -1792,7 +1820,7 @@ class FeishuSignalPipelineService: ) occurred_at = captured_at or _utc_now() if occurred_at.tzinfo is None: - occurred_at = occurred_at.replace(tzinfo=timezone.utc) + occurred_at = occurred_at.replace(tzinfo=CN_TZ) operator_user = await self._resolve_command_operator_user(sender=sender) if not fields.get("ready"): @@ -1998,7 +2026,7 @@ class FeishuSignalPipelineService: ) occurred_at = captured_at or _utc_now() if occurred_at.tzinfo is None: - occurred_at = occurred_at.replace(tzinfo=timezone.utc) + occurred_at = occurred_at.replace(tzinfo=CN_TZ) operator_user = await self._resolve_command_operator_user(sender=sender) if not fields.get("ready"): @@ -2188,7 +2216,7 @@ class FeishuSignalPipelineService: ) -> dict[str, Any]: occurred_at = captured_at or _utc_now() if occurred_at.tzinfo is None: - occurred_at = occurred_at.replace(tzinfo=timezone.utc) + occurred_at = occurred_at.replace(tzinfo=CN_TZ) fields = intent_gate.get("fields") if isinstance(intent_gate.get("fields"), dict) else {} blocking_reason = str(intent_gate.get("clarification_template") or "").strip() or "外部系统接入默认进入待决策队列。" queued = await self._queue_command_intent_item( @@ -2726,7 +2754,7 @@ class FeishuSignalPipelineService: routing_policy = await self._load_routing_policy() occurred_at = captured_at or _utc_now() if occurred_at.tzinfo is None: - occurred_at = occurred_at.replace(tzinfo=timezone.utc) + occurred_at = occurred_at.replace(tzinfo=CN_TZ) results: list[dict[str, Any]] = [] parent_item_id: Optional[int] = None @@ -3785,7 +3813,7 @@ class FeishuSignalPipelineService: return None occurred_at = captured_at or _utc_now() if occurred_at.tzinfo is not None: - occurred_at = occurred_at.astimezone(timezone.utc).replace(tzinfo=None) + occurred_at = occurred_at.astimezone(CN_TZ).replace(tzinfo=None) item = FeishuItem( tenant_id=self.tenant_id, feishu_message_id=feishu_message_id, @@ -4021,7 +4049,7 @@ class FeishuSignalPipelineService: pending_rows = [row for row in pending_rows if not self._should_hide_pending_item_from_inbox(row)] # DB 列是 timestamp without time zone，比较值必须使用 naive UTC，避免 asyncpg aware/naive 报错。 - today_start = _utc_now().astimezone(timezone.utc).replace( + today_start = _utc_now().astimezone(CN_TZ).replace( hour=0, minute=0, second=0, @@ -4732,7 +4760,23 @@ class FeishuSignalPipelineService: signal_auto_enter_threshold: float, ) -> Optional[int]: item.matched_project_id = int(matched_project_id) if matched_project_id else None - if cold_start_enabled or confidence < signal_auto_enter_threshold or not matched_project_id: + gate = kernel_gate_execution( + confidence=float(confidence or 0.0), + risk_level=("medium" if matched_project_id else "high"), + correction_action="none", + ) + decision_trace = build_decision_trace( + intent=INTENT_SIGNAL_INTAKE, + confidence=float(confidence or 0.0), + route_tier="rule", + route_function="_route_signal_item", + gate_action=gate.action, + gate_code=gate.gate_code, + correction_action="none", + correction_keywords=[], + complexity_reasons=[], + ) + if cold_start_enabled or gate.action != "auto_execute" or confidence < signal_auto_enter_threshold or not matched_project_id: item.card_type = "signal" item.route_status = ROUTE_STATUS_QUEUED item.queue_visible = True @@ -4747,10 +4791,20 @@ class FeishuSignalPipelineService: item.blocking_reason = "当前处于冷启动保护期，信号入库暂不自动执行。" item.next_step_text = "确认后，AI 将继续完成信号入库并推进后续动作。" item.decision_reason = "保守自动化已拦截自动入库，避免早期误关联。" + elif gate.action == "reject": + item.blocking_reason = gate.reason + item.next_step_text = "请补充更明确的执行对象后重试，或人工确认继续处理。" + item.decision_reason = "共享决策内核判定置信度不足，自动执行被拒绝。" else: - item.blocking_reason = "当前置信度不足以自动入库，需要你拍板后再继续。" + item.blocking_reason = gate.reason or "当前置信度不足以自动入库，需要你拍板后再继续。" item.next_step_text = "确认后，AI 将把信号入库并进入后续生成流程。" item.decision_reason = "信号虽然可执行，但归属判断仍存在不确定性。" + item.result_payload = { + "result_summary": "信号已进入待确认队列。", + "result_count": 0, + "impact_objects": [signal.title or f"信号#{signal.id}"], + "decision_trace": decision_trace, + } return matched_project_id await self._ensure_signal_link(signal_id=int(signal.id), project_id=int(matched_project_id)) @@ -4771,6 +4825,7 @@ class FeishuSignalPipelineService: "impact_objects": [signal.title or f"信号#{signal.id}", f"项目#{int(matched_project_id)}"], "target_url": f"/projects/{int(matched_project_id)}/board", "project_id": int(matched_project_id), + "decision_trace": decision_trace, } await self._notify_signal_auto_entered( project_id=int(matched_project_id), diff --git a/backend/app/services/llm.py b/backend/app/services/llm.py index b855c16f..ec1b219d 100644 --- a/backend/app/services/llm.py +++ b/backend/app/services/llm.py @@ -90,6 +90,7 @@ DEFAULT_TIER_MODEL_OVERRIDES = { "safety": OPENCLAW_DEFAULT_SAFETY_MODEL, } DEFAULT_HIGH_TIER_HINTS = ( + "classify_intent_complex", "project_from_text", "feishu_project_prefill", "feishu_task_orchestration", @@ -225,6 +226,17 @@ STRUCTURED_OBJECT_SCHEMAS: dict[str, dict[str, Any]] = { "project_id": ("number_or_none",), }, }, + "classify_intent_complex": { + "required_keys": ("intent", "confidence", "reasoning"), + "type_checks": { + "intent": "str", + "confidence": "number", + "reasoning": "str", + }, + "optional_type_checks": { + "project_id": ("number_or_none",), + }, + }, } diff --git a/backend/app/services/openclaw_commander_service.py b/backend/app/services/openclaw_commander_service.py index 2ba75481..f62cb107 100644 --- a/backend/app/services/openclaw_commander_service.py +++ b/backend/app/services/openclaw_commander_service.py @@ -41,6 +41,7 @@ from .sstc_service import ALIGNED_STATUSES from .task_persistence import build_task_create_draft, persist_task from ..utils.project_members import ensure_project_default_members from ..utils.sequence import ensure_project_id_sequence +from ..utils.timezone import now_cn_naive from ..utils.title import compact_title SUPPORTED_COMMANDER_OPERATION_CODES = ( @@ -66,7 +67,7 @@ SOURCE_ONBOARDING_FALLBACK_RE = re.compile( def _utc_now_naive() -> datetime: - return datetime.now(timezone.utc).replace(tzinfo=None) + return now_cn_naive() class OpenClawCommanderService: diff --git a/backend/app/services/project_knowledge_graph.py b/backend/app/services/project_knowledge_graph.py index 2c6ec012..a22bc461 100644 --- a/backend/app/services/project_knowledge_graph.py +++ b/backend/app/services/project_knowledge_graph.py @@ -32,12 +32,13 @@ from ..models import ( Task, User, ) +from ..utils.timezone import now_cn_naive from .context_cache import context_cache_service from .symbol_index_service import SymbolIndexService def _utc_now_naive() -> datetime: - return datetime.now(timezone.utc).replace(tzinfo=None) + return now_cn_naive() class ProjectKnowledgeGraphService: diff --git a/backend/app/services/signal_source_orchestration_service.py b/backend/app/services/signal_source_orchestration_service.py index 9c8a71c7..de60cb1e 100644 --- a/backend/app/services/signal_source_orchestration_service.py +++ b/backend/app/services/signal_source_orchestration_service.py @@ -28,6 +28,7 @@ from ..models import ( User, ) from ..database import AsyncSessionLocal +from ..utils.timezone import now_cn_naive from .ai_orchestration_run_service import AIOrchestrationRunService from .prompt_profile_service import prompt_profile_service from .llm import llm_service @@ -52,7 +53,7 @@ COMPLETE_WORKER_WAIT_TIMEOUT_SECONDS = 6.0 def _utc_now_naive() -> datetime: - return datetime.now(timezone.utc).replace(tzinfo=None) + return now_cn_naive() class SignalSourceOrchestrationService: diff --git a/backend/app/utils/timezone.py b/backend/app/utils/timezone.py new file mode 100644 index 00000000..6143e131 --- /dev/null +++ b/backend/app/utils/timezone.py @@ -0,0 +1,18 @@ +"""PMFlow 时间工具：统一使用北京时间（Asia/Shanghai）。""" + +from __future__ import annotations + +from datetime import datetime, timedelta, timezone + +CN_TZ = timezone(timedelta(hours=8), name="Asia/Shanghai") + + +def now_cn_aware() -> datetime: + """返回带时区的北京时间。""" + return datetime.now(CN_TZ) + + +def now_cn_naive() -> datetime: + """返回不带时区的北京时间（用于 TIMESTAMP WITHOUT TIME ZONE）。""" + return now_cn_aware().replace(tzinfo=None) + diff --git a/backend/tests/test_ai_agent_kernel.py b/backend/tests/test_ai_agent_kernel.py new file mode 100644 index 00000000..c07080ee --- /dev/null +++ b/backend/tests/test_ai_agent_kernel.py @@ -0,0 +1,86 @@ +import os +import unittest + +from app.services.ai_agent_kernel import ( + compute_capability_score, + gate_execution, + resolve_correction, + select_intent_route, +) + + +class AIAgentKernelTest(unittest.TestCase): + def test_select_intent_route_prefers_complex_for_multi_clause_instruction(self) -> None: + decision = select_intent_route( + "给我新增个项目，openclaw自动化部署，为了后续方便统一发布。", + project_names=["OpenClaw 自动化部署"], + ) + self.assertEqual(decision.function_name, "classify_intent_complex") + self.assertEqual(decision.tier, "high") + self.assertTrue(decision.is_complex) + + def test_select_intent_route_simple_instruction(self) -> None: + decision = select_intent_route("归档项目#51") + self.assertEqual(decision.function_name, "classify_intent_ai_pmo_kernel") + self.assertEqual(decision.tier, "low") + self.assertFalse(decision.is_complex) + + def test_resolve_correction_detects_rollback(self) -> None: + result = resolve_correction("这条不是新增项目，回滚刚才那次操作。") + self.assertIn(result.action, {"rollback", "replace_target"}) + self.assertGreaterEqual(result.confidence, 0.78) + self.assertTrue(result.matched_keywords) + + def test_gate_execution_respects_threshold_and_risk(self) -> None: + old_auto = os.environ.get("AGENT_GATE_AUTO_THRESHOLD") + old_review = os.environ.get("AGENT_GATE_REVIEW_THRESHOLD") + old_review_below_auto = os.environ.get("AGENT_GATE_REVIEW_BELOW_AUTO") + os.environ["AGENT_GATE_AUTO_THRESHOLD"] = "0.82" + os.environ["AGENT_GATE_REVIEW_THRESHOLD"] = "0.60" + try: + low = gate_execution(confidence=0.42, risk_level="low") + self.assertEqual(low.action, "reject") + pending = gate_execution(confidence=0.70, risk_level="medium") + self.assertEqual(pending.action, "auto_execute") + high_risk = gate_execution(confidence=0.86, risk_level="high") + self.assertEqual(high_risk.action, "pending_manual") + allowed = gate_execution(confidence=0.93, risk_level="high") + self.assertEqual(allowed.action, "auto_execute") + os.environ["AGENT_GATE_REVIEW_BELOW_AUTO"] = "1" + pending_with_switch = gate_execution(confidence=0.70, risk_level="medium") + self.assertEqual(pending_with_switch.action, "pending_manual") + forced = gate_execution( + confidence=0.99, + risk_level="low", + force_manual=True, + force_manual_code="intent_conflict_write", + force_manual_reason="规则与LLM冲突", + ) + self.assertEqual(forced.action, "pending_manual") + self.assertEqual(forced.gate_code, "intent_conflict_write") + finally: + if old_auto is None: + os.environ.pop("AGENT_GATE_AUTO_THRESHOLD", None) + else: + os.environ["AGENT_GATE_AUTO_THRESHOLD"] = old_auto + if old_review is None: + os.environ.pop("AGENT_GATE_REVIEW_THRESHOLD", None) + else: + os.environ["AGENT_GATE_REVIEW_THRESHOLD"] = old_review + if old_review_below_auto is None: + os.environ.pop("AGENT_GATE_REVIEW_BELOW_AUTO", None) + else: + os.environ["AGENT_GATE_REVIEW_BELOW_AUTO"] = old_review_below_auto + + def test_compute_capability_score(self) -> None: + score = compute_capability_score( + intent_accuracy=0.82, + gate_precision=0.80, + exec_success=0.78, + correction_success=0.76, + ) + self.assertAlmostEqual(score, 0.795, places=3) + + +if __name__ == "__main__": + unittest.main() diff --git a/backend/tests/test_ai_pmo_kernel_service.py b/backend/tests/test_ai_pmo_kernel_service.py index 8c198376..7290d96b 100644 --- a/backend/tests/test_ai_pmo_kernel_service.py +++ b/backend/tests/test_ai_pmo_kernel_service.py @@ -26,6 +26,7 @@ from app.services.ai_pmo_cadence_service import AIPMOCadenceService from app.services.ai_pmo_kernel_service import ( AIPMOKernelService, INTENT_PROJECT_CREATE, + INTENT_SIGNAL_CREATE, INTENT_SIGNAL_SOURCE_ONBOARD, IntentResolver, ) @@ -464,6 +465,81 @@ class AIPMOKernelServiceTest(unittest.IsolatedAsyncioTestCase): else: os.environ["AI_PMO_INTENT_LLM_ENABLED"] = prev + async def test_intent_arbitration_conflict_write_requires_manual_review(self) -> None: + old_llm = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") + old_arb = os.environ.get("AI_PMO_INTENT_ARBITRATION_ENABLED") + old_policy = os.environ.get("AI_PMO_INTENT_CONFLICT_POLICY") + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "1" + os.environ["AI_PMO_INTENT_ARBITRATION_ENABLED"] = "1" + os.environ["AI_PMO_INTENT_CONFLICT_POLICY"] = "pending_manual" + try: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + with patch( + "app.services.ai_pmo_kernel_service.llm_service.generate_structured_object", + AsyncMock( + return_value=StructuredExtractionResult( + payload={ + "intent": INTENT_SIGNAL_CREATE, + "confidence": 0.95, + "project_id": None, + "reasoning": "用户像是在提新增记录", + }, + source_field="content", + provider="openai", + model="MiniMax-M2.5", + raw_text='{"intent":"signal.create"}', + validation="ok", + ) + ), + ): + intent, source = await service._resolve_intent( + message_text="给我新增个项目，openclaw自动化部署。", + session=None, + world_model={"resolved_project_id": None, "project": {}}, + ) + + self.assertIn(source, {"llm", "rule_fallback"}) + self.assertTrue(bool((intent.metadata or {}).get("requires_manual_review"))) + self.assertEqual((intent.metadata or {}).get("manual_review_code"), "intent_conflict_write") + self.assertEqual((intent.metadata or {}).get("arbitration"), "conflict_manual_review") + finally: + if old_llm is None: + os.environ.pop("AI_PMO_INTENT_LLM_ENABLED", None) + else: + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = old_llm + if old_arb is None: + os.environ.pop("AI_PMO_INTENT_ARBITRATION_ENABLED", None) + else: + os.environ["AI_PMO_INTENT_ARBITRATION_ENABLED"] = old_arb + if old_policy is None: + os.environ.pop("AI_PMO_INTENT_CONFLICT_POLICY", None) + else: + os.environ["AI_PMO_INTENT_CONFLICT_POLICY"] = old_policy + + async def test_project_create_slot_guard_requires_title(self) -> None: + old_llm = os.environ.get("AI_PMO_INTENT_LLM_ENABLED") + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = "0" + try: + async with self.session_factory() as db: + service = AIPMOKernelService(db, tenant_id=1) + intent, source = await service._resolve_intent( + message_text="请创建项目。", + session=None, + world_model={"resolved_project_id": None, "project": {}}, + ) + + self.assertEqual(source, "rule_fallback") + self.assertEqual(intent.name, INTENT_PROJECT_CREATE) + self.assertFalse(bool((intent.metadata or {}).get("slot_ok"))) + self.assertTrue(bool((intent.metadata or {}).get("requires_manual_review"))) + self.assertEqual((intent.metadata or {}).get("manual_review_code"), "slot_incomplete") + finally: + if old_llm is None: + os.environ.pop("AI_PMO_INTENT_LLM_ENABLED", None) + else: + os.environ["AI_PMO_INTENT_LLM_ENABLED"] = old_llm + async def test_policy_gate_blocks_archive_like_request(self) -> None: async with self.session_factory() as db: service = AIPMOKernelService(db, tenant_id=1) diff --git a/backend/tests/test_ai_pmo_work_item_decision_api.py b/backend/tests/test_ai_pmo_work_item_decision_api.py new file mode 100644 index 00000000..265e4cb5 --- /dev/null +++ b/backend/tests/test_ai_pmo_work_item_decision_api.py @@ -0,0 +1,143 @@ +import os +import unittest +from tempfile import NamedTemporaryFile +from types import SimpleNamespace + +from sqlalchemy import and_, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.api.ai_pmo import PMOWorkItemDecisionRequest, _normalize_resume_result, decide_work_item +from app.models import Base, DecisionRecord, IngestEvent, User, WorkItem + + +class AIPMOWorkItemDecisionApiTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + tmp = NamedTemporaryFile(prefix="pmflow-ai-pmo-decision-api-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 + + def test_normalize_resume_result_maps_answered_to_succeeded(self) -> None: + target = SimpleNamespace( + id=99, + state="waiting_manual", + intent="signal.create", + project_id=7, + signal_id=None, + ) + raw = { + "handled": True, + "status": "answered", + "direct_reply_text": "已继续处理：信号已入库。", + "results": [{"state": "succeeded"}], + "linked_project_ids": [7], + "signal_id": 13, + "auto_tasks": [], + } + + normalized = _normalize_resume_result(raw=raw, target_work_item=target) + + self.assertTrue(normalized["handled"]) + self.assertEqual(normalized["status"], "succeeded") + self.assertEqual(normalized["target_work_item_id"], 99) + self.assertEqual(normalized["target_state"], "succeeded") + self.assertEqual(normalized["project_id"], 7) + self.assertEqual(normalized["signal_id"], 13) + + async def test_decision_reject_marks_waiting_manual_item_ignored(self) -> None: + async with self.session_factory() as db: + user = User( + tenant_id=1, + username="decision_boss", + email="decision_boss@example.com", + password_hash="hash", + name="决策人", + role="BOSS", + status="active", + is_active=True, + ) + db.add(user) + await db.flush() + + ingest = IngestEvent( + tenant_id=1, + channel="ai_pmo_ui_message", + event_key="decision-reject-1", + status="done", + payload={"message_text": "删除项目id 52"}, + ) + db.add(ingest) + await db.flush() + + work_item = WorkItem( + tenant_id=1, + ingest_event_id=int(ingest.id), + source_system="ai_pmo_ui", + source_message_id="decision-reject-item-1", + state="waiting_manual", + intent="query.read_only", + project_id=52, + signal_id=None, + execution_result={"status": "waiting_manual"}, + ) + db.add(work_item) + await db.flush() + + decision = DecisionRecord( + tenant_id=1, + work_item_id=int(work_item.id), + decision_object_type="project", + decision_object_id=52, + decision_level="project", + policy_gate_hit=True, + policy_gate_code="delete_or_archive", + approval_requirement={"required": True}, + ) + db.add(decision) + await db.commit() + + response = await decide_work_item( + work_item_id=int(work_item.id), + request=PMOWorkItemDecisionRequest(action="reject", reason="不同意执行，标记失效"), + current_user=user, + db=db, + tenant_id=1, + ) + + refreshed = ( + await db.execute( + select(WorkItem).where( + and_( + WorkItem.tenant_id == 1, + WorkItem.id == int(work_item.id), + ) + ) + ) + ).scalar_one() + refreshed_decision = ( + await db.execute( + select(DecisionRecord).where( + and_( + DecisionRecord.tenant_id == 1, + DecisionRecord.work_item_id == int(work_item.id), + ) + ) + ) + ).scalar_one() + + self.assertEqual(response["code"], 0) + self.assertEqual(response["data"]["status"], "invalidated") + self.assertEqual(refreshed.state, "ignored") + self.assertEqual(refreshed.reply_template_key, "action_rejected") + self.assertEqual(refreshed_decision.final_outcome, "ignored") diff --git a/backend/tests/test_ai_steward_title_extract.py b/backend/tests/test_ai_steward_title_extract.py new file mode 100644 index 00000000..e00e5059 --- /dev/null +++ b/backend/tests/test_ai_steward_title_extract.py @@ -0,0 +1,31 @@ +import unittest + +from app.services.ai_steward import AIStewardService + + +class AIStewardTitleExtractTest(unittest.IsolatedAsyncioTestCase): + async def test_extract_title_from_plain_project_create_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自动化部署") + + async def test_normalize_project_draft_should_not_use_instruction_sentence_as_title(self) -> None: + service = AIStewardService(db=None) # type: ignore[arg-type] + normalized = await service._normalize_project_creation_draft( + draft={ + "analysis": {}, + "project": { + "title": "不是创建信号，是创建项目名为openclaw自动化部署。", + "description": "不是创建信号，是创建项目名为openclaw自动化部署。", + }, + "tasks": [], + }, + input_text="不是创建信号，是创建项目名为openclaw自动化部署。", + creator_id=1, + tenant_id=1, + ) + self.assertEqual(normalized["project"]["title"], "openclaw自动化部署") + self.assertEqual(normalized["project"]["description"], "项目：openclaw自动化部署") diff --git a/backend/tests/test_projects_delete_api.py b/backend/tests/test_projects_delete_api.py new file mode 100644 index 00000000..98757536 --- /dev/null +++ b/backend/tests/test_projects_delete_api.py @@ -0,0 +1,338 @@ +import os +import unittest +from tempfile import NamedTemporaryFile +from unittest.mock import AsyncMock, patch + +from sqlalchemy import and_, select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine + +from app.api.projects import delete_project +from app.models import ( + AIOrchestrationRun, + AIWorkerExecution, + Base, + Budget, + ConversationSession, + CostTracking, + Experiment, + IngestEvent, + OperationExecutionLedger, + OperationLog, + Project, + SignalSourceRegistry, + Task, + TruthNode, + User, + UserProject, + WorkItem, +) + + +class DeleteProjectApiTest(unittest.IsolatedAsyncioTestCase): + async def asyncSetUp(self) -> None: + tmp = NamedTemporaryFile(prefix="pmflow-delete-project-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 test_delete_project_cleans_fk_dependencies(self) -> None: + async with self.session_factory() as db: + user = User( + tenant_id=1, + username="boss-delete-project", + email="boss-delete-project@example.com", + password_hash="x", + name="Boss", + role="BOSS", + status="active", + is_active=True, + ) + db.add(user) + await db.flush() + + project = Project( + tenant_id=1, + name="P-Delete", + title="P-Delete", + description="for delete test", + status="active", + owner_id=int(user.id), + signal_count=0, + ) + db.add(project) + await db.flush() + + truth_node = TruthNode( + tenant_id=1, + node_type="task", + project_id=int(project.id), + created_by=int(user.id), + title="TN", + content={"k": "v"}, + version=1, + status="active", + ) + db.add(truth_node) + await db.flush() + + task = Task( + tenant_id=1, + truth_node_id=int(truth_node.id), + project_id=int(project.id), + assignee_id=int(user.id), + title="task-a", + description="d", + required_skill_tag_ids=[1], + status="todo", + priority="p1", + ) + db.add(task) + await db.flush() + + cost_tracking = CostTracking( + tenant_id=1, + truth_node_id=int(truth_node.id), + task_id=int(task.id), + period="project", + estimated_tokens=1, + actual_tokens=1, + estimated_cost_usd=0.1, + actual_cost_usd=0.1, + roi_analysis={}, + optimization_suggestions={}, + ) + db.add(cost_tracking) + + registry = SignalSourceRegistry( + tenant_id=1, + project_id=int(project.id), + source_kind="github", + source_name="repo", + source_identifier="github:org/repo", + ) + db.add(registry) + await db.flush() + + worker_execution = AIWorkerExecution( + tenant_id=1, + project_id=int(project.id), + signal_source_registry_id=int(registry.id), + trace_id="trace-worker-delete", + worker_role_key="pm_dispatcher", + worker_role_name="PM Dispatcher", + ) + db.add(worker_execution) + + orchestration_run = AIOrchestrationRun( + tenant_id=1, + trace_id="trace-run-delete", + run_type="signal_source_onboarding", + project_id=int(project.id), + response_mode="fast", + status="queued", + worker_total=0, + worker_completed=0, + worker_failed=0, + ) + db.add(orchestration_run) + + ingest_event = IngestEvent( + tenant_id=1, + channel="feishu", + event_key="delete-project-event", + status="done", + ) + db.add(ingest_event) + await db.flush() + + work_item = WorkItem( + tenant_id=1, + ingest_event_id=int(ingest_event.id), + source_system="feishu", + source_message_id="msg-delete-project", + state="received", + project_id=int(project.id), + ) + db.add(work_item) + + operation_log = OperationLog( + tenant_id=1, + user_id=int(user.id), + project_id=int(project.id), + entity_type="project", + entity_id=str(project.id), + action="delete", + ) + db.add(operation_log) + + operation_ledger = OperationExecutionLedger( + tenant_id=1, + operation_code="project.delete", + project_id=int(project.id), + idempotency_key="project-delete-key", + status="processing", + ) + db.add(operation_ledger) + + experiment = Experiment( + tenant_id=1, + title="exp", + status="draft", + project_id=int(project.id), + ) + db.add(experiment) + + budget = Budget( + tenant_id=1, + budget_type="project", + project_id=int(project.id), + period="monthly", + budget_amount_usd=100.0, + alert_threshold_percent=80.0, + hard_limit_enabled=True, + current_usage_usd=10.0, + status="active", + override_history={}, + ) + db.add(budget) + + conversation = ConversationSession( + tenant_id=1, + sender="ou_delete_project", + current_project_id=int(project.id), + last_intent="delete_or_archive", + ) + db.add(conversation) + + user_project = UserProject( + tenant_id=1, + user_id=int(user.id), + project_id=int(project.id), + ) + db.add(user_project) + + 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()), + ): + result = await delete_project( + project_id=int(project.id), + db=db, + current_user=user, + ) + + self.assertEqual(result, {"message": "Project deleted"}) + + project_row = ( + await db.execute( + select(Project).where( + and_(Project.tenant_id == 1, Project.id == int(project.id)) + ) + ) + ).scalar_one_or_none() + self.assertIsNone(project_row) + + worker_rows = ( + await db.execute( + select(AIWorkerExecution).where( + and_( + AIWorkerExecution.tenant_id == 1, + AIWorkerExecution.project_id == int(project.id), + ) + ) + ) + ).scalars().all() + self.assertEqual(len(worker_rows), 0) + + run_rows = ( + await db.execute( + select(AIOrchestrationRun).where( + and_( + AIOrchestrationRun.tenant_id == 1, + AIOrchestrationRun.project_id == int(project.id), + ) + ) + ) + ).scalars().all() + self.assertEqual(len(run_rows), 0) + + registry_rows = ( + await db.execute( + select(SignalSourceRegistry).where( + and_( + SignalSourceRegistry.tenant_id == 1, + SignalSourceRegistry.project_id == int(project.id), + ) + ) + ) + ).scalars().all() + self.assertEqual(len(registry_rows), 0) + + user_project_rows = ( + await db.execute( + select(UserProject).where( + and_( + UserProject.tenant_id == 1, + UserProject.project_id == int(project.id), + ) + ) + ) + ).scalars().all() + self.assertEqual(len(user_project_rows), 0) + + task_rows = ( + await db.execute( + select(Task).where(and_(Task.tenant_id == 1, Task.project_id == int(project.id))) + ) + ).scalars().all() + self.assertEqual(len(task_rows), 0) + + refreshed_cost_tracking = ( + await db.execute(select(CostTracking).where(CostTracking.id == int(cost_tracking.id))) + ).scalar_one() + self.assertIsNone(refreshed_cost_tracking.task_id) + + refreshed_work_item = ( + await db.execute(select(WorkItem).where(WorkItem.id == int(work_item.id))) + ).scalar_one() + self.assertIsNone(refreshed_work_item.project_id) + + refreshed_operation_log = ( + await db.execute(select(OperationLog).where(OperationLog.id == int(operation_log.id))) + ).scalar_one() + self.assertIsNone(refreshed_operation_log.project_id) + + refreshed_ledger = ( + await db.execute( + select(OperationExecutionLedger).where(OperationExecutionLedger.id == int(operation_ledger.id)) + ) + ).scalar_one() + self.assertIsNone(refreshed_ledger.project_id) + + refreshed_experiment = ( + await db.execute(select(Experiment).where(Experiment.id == int(experiment.id))) + ).scalar_one() + self.assertIsNone(refreshed_experiment.project_id) + + refreshed_budget = ( + await db.execute(select(Budget).where(Budget.id == int(budget.id))) + ).scalar_one() + self.assertIsNone(refreshed_budget.project_id) + + refreshed_conversation = ( + await db.execute(select(ConversationSession).where(ConversationSession.id == int(conversation.id))) + ).scalar_one() + self.assertIsNone(refreshed_conversation.current_project_id) diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index b88a9760..aafc9f0e 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -7,6 +7,8 @@ services: POSTGRES_DB: ${POSTGRES_DB:-pmflow} POSTGRES_USER: ${POSTGRES_USER:-pmflow} POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-pmflow_pass} + TZ: Asia/Shanghai + PGTZ: Asia/Shanghai ports: - "${DB_PORT:-5435}:5432" volumes: @@ -55,6 +57,8 @@ services: redis: condition: service_healthy environment: + TZ: Asia/Shanghai + PGTZ: Asia/Shanghai DATABASE_URL: postgresql+asyncpg://${POSTGRES_USER:?missing_POSTGRES_USER}:${POSTGRES_PASSWORD:?missing_POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB:?missing_POSTGRES_DB} REDIS_URL: redis://:${REDIS_PASSWORD:?missing_REDIS_PASSWORD}@redis:6379/0 PMFLOW_EXPECTED_DB_DIALECT: ${PMFLOW_EXPECTED_DB_DIALECT:?missing_PMFLOW_EXPECTED_DB_DIALECT} diff --git a/docker-compose.yml b/docker-compose.yml index 454ceffc..b06049f3 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -7,6 +7,8 @@ services: POSTGRES_DB: ${POSTGRES_DB:-pmflow} POSTGRES_USER: ${POSTGRES_USER:-pmflow} POSTGRES_PASSWORD: ${POSTGRES_PASSWORD:-pmflow_pass} + TZ: Asia/Shanghai + PGTZ: Asia/Shanghai ports: - "${DB_PORT:-5435}:5432" volumes: @@ -55,6 +57,8 @@ services: redis: condition: service_healthy environment: + TZ: Asia/Shanghai + PGTZ: Asia/Shanghai DATABASE_URL: postgresql+asyncpg://${POSTGRES_USER:?missing_POSTGRES_USER}:${POSTGRES_PASSWORD:?missing_POSTGRES_PASSWORD}@db:5432/${POSTGRES_DB:?missing_POSTGRES_DB} REDIS_URL: redis://:${REDIS_PASSWORD:?missing_REDIS_PASSWORD}@redis:6379/0 PMFLOW_EXPECTED_DB_DIALECT: ${PMFLOW_EXPECTED_DB_DIALECT:?missing_PMFLOW_EXPECTED_DB_DIALECT} diff --git a/frontend/src/api/aiPmo.ts b/frontend/src/api/aiPmo.ts index d2a96a7f..e1eb7449 100644 --- a/frontend/src/api/aiPmo.ts +++ b/frontend/src/api/aiPmo.ts @@ -102,6 +102,13 @@ export interface AIPMOResumeResult { auto_tasks?: Array>; } +export interface AIPMOWorkItemDecisionRequest { + action: "approve" | "reject"; + project_id?: number | null; + source_message_id?: string; + reason?: string; +} + export interface AIPMOCadenceSnapshot { id: number; scope_type: string; @@ -151,6 +158,11 @@ export const aiPmoApi = { return unwrap(data); }, + async decideWorkItem(workItemId: number, body: AIPMOWorkItemDecisionRequest): Promise { + const { data } = await api.post>(`/ai-pmo/work-items/${workItemId}/decision`, body); + return unwrap(data); + }, + async getLatestCadence(params: { cadenceType: string; projectId?: number }): Promise { const { data } = await api.get>("/ai-pmo/cadence/latest", { params: { diff --git a/frontend/src/pages/InboxPage.tsx b/frontend/src/pages/InboxPage.tsx index 11680363..6a52307b 100644 --- a/frontend/src/pages/InboxPage.tsx +++ b/frontend/src/pages/InboxPage.tsx @@ -440,7 +440,8 @@ export default function InboxPage() { return; } if (action.kind === "confirm" && item.source_kind === "ai_pmo_work_item" && item.source_work_item) { - const result = await aiPmoApi.resumeWorkItem(item.source_work_item.id, { + const result = await aiPmoApi.decideWorkItem(item.source_work_item.id, { + action: "approve", project_id: item.project_id ?? undefined, }); await refreshAll(); @@ -465,6 +466,19 @@ export default function InboxPage() { }); return; } + if (action.kind === "reject" && item.source_kind === "ai_pmo_work_item" && item.source_work_item) { + const result = await aiPmoApi.decideWorkItem(item.source_work_item.id, { + action: "reject", + reason: "人工审批拒绝，已标记失效", + }); + await refreshAll(); + const followupReply = String(result.followup_reply || "").trim(); + toast({ title: followupReply || "已标记失效，不再继续执行", variant: "default" }); + if (workItemDetailItem?.id === item.id) { + setWorkItemDetailItem(null); + } + return; + } if (action.kind === "mark_read" && item.source_kind === "notification" && item.source_notification) { await notificationsApi.markRead(item.source_notification.id); await refreshAll(); diff --git a/frontend/src/types/inbox.ts b/frontend/src/types/inbox.ts index 383243e2..8dec220c 100644 --- a/frontend/src/types/inbox.ts +++ b/frontend/src/types/inbox.ts @@ -6,6 +6,7 @@ export type InboxDisplayKind = "action_required" | "auto_running" | "auto_done" export type InboxSectionKey = "all" | InboxDisplayKind; export type InboxActionKind = | "confirm" + | "reject" | "open_detail" | "open_target" | "mark_read" diff --git a/scripts/agent_capability_replay.py b/scripts/agent_capability_replay.py new file mode 100644 index 00000000..e7e8beb8 --- /dev/null +++ b/scripts/agent_capability_replay.py @@ -0,0 +1,155 @@ +#!/usr/bin/env python3 +"""Offline replay scorer for dual-agent capability metrics.""" + +from __future__ import annotations + +import argparse +import json +from dataclasses import dataclass +from pathlib import Path +from typing import Any + + +@dataclass(slots=True) +class MetricAccumulator: + count: int = 0 + intent_sum: float = 0.0 + gate_sum: float = 0.0 + exec_sum: float = 0.0 + correction_sum: float = 0.0 + + def append(self, *, intent: float, gate: float, exec_success: float, correction: float) -> None: + self.count += 1 + self.intent_sum += intent + self.gate_sum += gate + self.exec_sum += exec_success + self.correction_sum += correction + + def summary(self) -> dict[str, Any]: + if self.count <= 0: + return { + "count": 0, + "intent_accuracy": 0.0, + "gate_precision": 0.0, + "exec_success": 0.0, + "correction_success": 0.0, + "score": 0.0, + } + intent_accuracy = self.intent_sum / self.count + gate_precision = self.gate_sum / self.count + exec_success = self.exec_sum / self.count + correction_success = self.correction_sum / self.count + return { + "count": self.count, + "intent_accuracy": round(intent_accuracy, 4), + "gate_precision": round(gate_precision, 4), + "exec_success": round(exec_success, 4), + "correction_success": round(correction_success, 4), + "score": _compute_capability_score( + intent_accuracy=intent_accuracy, + gate_precision=gate_precision, + exec_success=exec_success, + correction_success=correction_success, + ), + } + + +def _compute_capability_score( + *, + intent_accuracy: float, + gate_precision: float, + exec_success: float, + correction_success: float, +) -> float: + score = ( + 0.30 * max(0.0, min(1.0, float(intent_accuracy or 0.0))) + + 0.30 * max(0.0, min(1.0, float(gate_precision or 0.0))) + + 0.25 * max(0.0, min(1.0, float(exec_success or 0.0))) + + 0.15 * max(0.0, min(1.0, float(correction_success or 0.0))) + ) + return round(score, 4) + + +def _to_binary(value: Any) -> float: + if isinstance(value, bool): + return 1.0 if value else 0.0 + if isinstance(value, (int, float)): + return 1.0 if float(value) >= 1.0 else (1.0 if float(value) >= 0.5 else 0.0) + text = str(value or "").strip().lower() + return 1.0 if text in {"1", "true", "yes", "y", "ok", "pass", "passed", "success", "succeeded"} else 0.0 + + +def _load_jsonl(path: Path) -> list[dict[str, Any]]: + rows: list[dict[str, Any]] = [] + for raw in path.read_text(encoding="utf-8").splitlines(): + line = raw.strip() + if not line: + continue + parsed = json.loads(line) + if isinstance(parsed, dict): + rows.append(parsed) + return rows + + +def _bucket_of(row: dict[str, Any]) -> str: + bucket = str(row.get("bucket") or "").strip().lower() + if bucket in {"simple", "complex"}: + return bucket + trace = row.get("decision_trace") if isinstance(row.get("decision_trace"), dict) else {} + route_tier = str(trace.get("route_tier") or row.get("route_tier") or "").strip().lower() + return "complex" if route_tier == "high" else "simple" + + +def _metric(row: dict[str, Any], key: str, fallback_keys: list[str]) -> float: + if key in row: + return _to_binary(row.get(key)) + for name in fallback_keys: + if name in row: + return _to_binary(row.get(name)) + return 0.0 + + +def main() -> None: + parser = argparse.ArgumentParser(description="Replay dual-agent metrics from JSONL samples.") + parser.add_argument("--input", required=True, help="Path to JSONL file.") + parser.add_argument("--output", default="", help="Optional output path; default prints to stdout.") + args = parser.parse_args() + + rows = _load_jsonl(Path(args.input)) + overall = MetricAccumulator() + simple = MetricAccumulator() + complex_bucket = MetricAccumulator() + + for row in rows: + intent = _metric(row, "intent_correct", ["intent_accuracy", "intent_success"]) + gate = _metric(row, "gate_correct", ["gate_precision", "gate_success"]) + exec_success = _metric(row, "exec_success", ["execution_success", "succeeded"]) + correction = _metric(row, "correction_success", ["correction_correct"]) + overall.append(intent=intent, gate=gate, exec_success=exec_success, correction=correction) + bucket_name = _bucket_of(row) + if bucket_name == "complex": + complex_bucket.append(intent=intent, gate=gate, exec_success=exec_success, correction=correction) + else: + simple.append(intent=intent, gate=gate, exec_success=exec_success, correction=correction) + + payload = { + "overall": overall.summary(), + "simple": simple.summary(), + "complex": complex_bucket.summary(), + "thresholds": { + "overall_score_target": 0.80, + "bucket_score_target": 0.78, + "intent_target": 0.80, + "gate_target": 0.78, + "correction_target": 0.75, + }, + } + text = json.dumps(payload, ensure_ascii=False, indent=2) + if args.output: + Path(args.output).write_text(text + "\n", encoding="utf-8") + return + print(text) + + +if __name__ == "__main__": + main()