"""OpenClaw 指挥层与操作执行服务。""" from __future__ import annotations import asyncio from datetime import datetime, timezone import hashlib import json import re import time from types import SimpleNamespace from typing import Any, Optional from uuid import uuid4 from sqlalchemy import and_, func, or_, select from sqlalchemy.exc import IntegrityError from sqlalchemy.ext.asyncio import AsyncSession from ..config.ai_prompts import PROMPT_OPENCLAW_COMMAND_DISPATCH from ..models import ( OperationExecutionLedger, Project, SkillTag, Task, TruthNode, User, UserProject, UserSkillBinding, UserWorkload, ) from . import workload_service from .adaptive_rag_service import AdaptiveRAGService from .ai_orchestration_run_service import AIOrchestrationRunService from .context_cache import context_cache_service from .llm import llm_service from .operation_catalog_service import operation_catalog_service from .operation_log_service import log_operation from .notification_service import NotificationService from .skill_score_service import process_task_completion 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.title import compact_title SUPPORTED_COMMANDER_OPERATION_CODES = ( "project.create", "project.update", "project.archive", "project.reopen", "task.create", "task.assign", "task.prioritize", "task.status_transition", "notification.create", "personnel.workload_update", ) PERSISTENT_IDEMPOTENCY_OPERATION_CODES = {"notification.create", "project.create"} SUPPORTED_PROJECT_STATUSES = {"active", "archived", "done"} SUPPORTED_TASK_STATUSES = {"todo", "in_progress", "review", "done", "blocked", "closed"} DONE_TASK_STATUSES = {"done", "completed", "closed"} SOURCE_ONBOARDING_FALLBACK_RE = re.compile( r"(新增|增加|添加|接入|配置|注册|建立).{0,10}(信号源|来源|监控源|github|飞书|webhook|rss|url|source|channel)", re.IGNORECASE, ) def _utc_now_naive() -> datetime: return datetime.now(timezone.utc).replace(tzinfo=None) class OpenClawCommanderService: """将自然语言指令转成可执行 operation plan。""" def __init__(self, db: AsyncSession) -> None: self.db = db async def dispatch_command( self, *, tenant_id: int, operator_user: User, project_id: int, instruction: str, dry_run: bool, max_operations: int, response_mode: str = "fast", trace_id: Optional[str] = None, persist_run: bool = True, ) -> dict[str, Any]: project_row = await self._load_project(tenant_id=tenant_id, project_id=project_id) if project_row is None: raise ValueError("project not found") project = self._build_project_snapshot(project_row) members = await self._load_project_team_context( tenant_id=tenant_id, project_id=project_id, owner_id=int(project.owner_id) if project.owner_id else None, ) tasks, task_summary = await self._load_project_tasks(tenant_id=tenant_id, project_id=project_id, members=members) planner_role = await llm_service._resolve_active_prompt_profile(self.db, tenant_id=tenant_id) trace_id = str(trace_id or f"openclaw-{tenant_id}-{project_id}-{int(time.time() * 1000)}-{uuid4().hex[:8]}") plan, planner_status, planner_diagnostics = await self._plan_operations( tenant_id=tenant_id, operator_user=operator_user, project=project, tasks=tasks, task_summary=task_summary, members=members, instruction=instruction, dry_run=dry_run, max_operations=max_operations, response_mode=response_mode, project_directory=[self._serialize_project_directory_item(project_row)], ) executor = OperationExecutorService(self.db) execution = await executor.execute_operations( tenant_id=tenant_id, operator_user=operator_user, project=project, operations=plan.get("operations") or [], dry_run=dry_run, trace_id=trace_id, ) openclaw_status = self._resolve_openclaw_status(execution=execution, dry_run=dry_run) run_status = openclaw_status if openclaw_status in {"completed", "partial", "failed"} else "completed" result = { "trace_id": trace_id, "response_mode": response_mode, "run_status": run_status, "planner_status": planner_status, "planner_diagnostics": planner_diagnostics, "openclaw_status": openclaw_status, "project": { "id": str(project.id), "title": project.title or project.name, "status": project.status, }, "prompt_role": { "role_key": str(planner_role.get("role_key") or ""), "role_name": str(planner_role.get("role_name") or ""), }, "instruction": instruction, "dry_run": bool(dry_run), "summary": str(plan.get("summary") or "").strip(), "confidence": self._to_float(plan.get("confidence"), default=0.0), "task_summary": task_summary, "member_count": len(members), "planned_operations": len(plan.get("operations") or []), "execution": execution, "pending_workers": [], } if persist_run: run_service = AIOrchestrationRunService(self.db) await run_service.create_run( tenant_id=tenant_id, trace_id=trace_id, run_type="openclaw_dispatch", project_id=int(project.id), signal_id=None, response_mode=response_mode, status=run_status, planner_status=planner_status, openclaw_status=openclaw_status, worker_total=0, worker_completed=0, worker_failed=0, summary=str(result.get("summary") or "").strip(), planner_diagnostics=planner_diagnostics, result_payload=dict(result), created_by_user_id=int(operator_user.id) if getattr(operator_user, "id", None) else None, completed_at=_utc_now_naive() if run_status in {"completed", "partial", "failed"} else None, ) await self.db.commit() return result async def dispatch_global_command( self, *, tenant_id: int, operator_user: User, instruction: str, dry_run: bool, max_operations: int, response_mode: str = "fast", trace_id: Optional[str] = None, persist_run: bool = False, ) -> dict[str, Any]: project = self._build_global_project_snapshot() project_directory = await self._load_project_directory(tenant_id=tenant_id) planner_role = await llm_service._resolve_active_prompt_profile(self.db, tenant_id=tenant_id) trace_id = str(trace_id or f"openclaw-{tenant_id}-global-{int(time.time() * 1000)}-{uuid4().hex[:8]}") plan, planner_status, planner_diagnostics = await self._plan_operations( tenant_id=tenant_id, operator_user=operator_user, project=project, tasks=[], task_summary={"total": 0, "active": 0, "done": 0, "blocked": 0, "sampled": 0}, members=[], instruction=instruction, dry_run=dry_run, max_operations=max_operations, response_mode=response_mode, project_directory=project_directory, ) executor = OperationExecutorService(self.db) execution = await executor.execute_operations( tenant_id=tenant_id, operator_user=operator_user, project=None, operations=plan.get("operations") or [], dry_run=dry_run, trace_id=trace_id, ) openclaw_status = self._resolve_openclaw_status(execution=execution, dry_run=dry_run) run_status = openclaw_status if openclaw_status in {"completed", "partial", "failed"} else "completed" project_ref = self._extract_project_reference_from_execution(execution) or { "id": None, "title": "", "status": "", } result = { "trace_id": trace_id, "response_mode": response_mode, "run_status": run_status, "planner_status": planner_status, "planner_diagnostics": planner_diagnostics, "openclaw_status": openclaw_status, "project": project_ref, "prompt_role": { "role_key": str(planner_role.get("role_key") or ""), "role_name": str(planner_role.get("role_name") or ""), }, "instruction": instruction, "dry_run": bool(dry_run), "summary": str(plan.get("summary") or "").strip(), "confidence": self._to_float(plan.get("confidence"), default=0.0), "task_summary": {"total": 0, "active": 0, "done": 0, "blocked": 0, "sampled": 0}, "member_count": 0, "planned_operations": len(plan.get("operations") or []), "execution": execution, "pending_workers": [], } resolved_project_id = self._to_int(project_ref.get("id")) if isinstance(project_ref, dict) else None if persist_run and resolved_project_id is not None and resolved_project_id > 0: run_service = AIOrchestrationRunService(self.db) await run_service.create_run( tenant_id=tenant_id, trace_id=trace_id, run_type="openclaw_dispatch", project_id=int(resolved_project_id), signal_id=None, response_mode=response_mode, status=run_status, planner_status=planner_status, openclaw_status=openclaw_status, worker_total=0, worker_completed=0, worker_failed=0, summary=str(result.get("summary") or "").strip(), planner_diagnostics=planner_diagnostics, result_payload=dict(result), created_by_user_id=int(operator_user.id) if getattr(operator_user, "id", None) else None, completed_at=_utc_now_naive() if run_status in {"completed", "partial", "failed"} else None, ) await self.db.commit() return result @staticmethod def _build_project_snapshot(project: Project) -> SimpleNamespace: return SimpleNamespace( id=int(project.id), owner_id=int(project.owner_id) if getattr(project, "owner_id", None) else None, title=str(project.title or "").strip(), name=str(project.name or "").strip(), status=str(project.status or "").strip(), assigned_role=str(project.assigned_role or "").strip(), description=str(project.description or "").strip(), ) @staticmethod def _build_global_project_snapshot() -> SimpleNamespace: return SimpleNamespace( id=0, owner_id=None, title="", name="", status="", assigned_role="", description="", ) @staticmethod def _serialize_project_directory_item(project: Project) -> dict[str, Any]: return { "project_id": int(project.id), "title": str(project.title or project.name or "").strip(), "status": str(project.status or "").strip(), } async def _plan_operations( self, *, tenant_id: int, operator_user: User, project: Project, tasks: list[dict[str, Any]], task_summary: dict[str, Any], members: list[dict[str, Any]], instruction: str, dry_run: bool, max_operations: int, response_mode: str, project_directory: Optional[list[dict[str, Any]]] = None, ) -> tuple[dict[str, Any], str, dict[str, Any]]: command_payload = { "instruction": instruction, "operator_user_id": int(operator_user.id), "operator_name": str(operator_user.name or operator_user.username or operator_user.id), "operator_role": str(operator_user.role or ""), "dry_run": bool(dry_run), "max_operations": int(max_operations), } project_payload = { "scope": "project" if self._to_int(project.id) and int(project.id) > 0 else "global", "project_id": int(project.id) if self._to_int(project.id) and int(project.id) > 0 else None, "title": str(project.title or project.name or "").strip(), "description": str(project.description or "").strip()[:400], "status": str(project.status or "").strip(), "assigned_role": str(project.assigned_role or "").strip(), "task_summary": task_summary, "candidate_projects": project_directory or [], } operation_catalog_payload = [ { "operation_code": item["operation_code"], "description": item["description"], "allowed_fields": sorted( list( ( (item.get("input_schema") or {}).get("properties") or {} ).keys() ) ), } for item in self._commander_operation_catalog() ] prompt = PROMPT_OPENCLAW_COMMAND_DISPATCH.format( command_json=json.dumps(command_payload, ensure_ascii=False), project_json=json.dumps(project_payload, ensure_ascii=False), tasks_json=json.dumps(tasks, ensure_ascii=False), members_json=json.dumps(members, ensure_ascii=False), operation_catalog_json=json.dumps(operation_catalog_payload, ensure_ascii=False), max_operations=int(max_operations), ) try: extraction = await llm_service.generate_structured_object( prompt, db=self.db, tenant_id=tenant_id, function_name="openclaw_command_dispatch", use_prompt_profile=False, temperature=0.0, max_tokens=520, response_mode=response_mode, ) diagnostics = self._build_planner_diagnostics(extraction) if extraction.payload is None: fallback = await self._build_fallback_plan( tenant_id=tenant_id, project_id=self._to_int(project.id), instruction=instruction, max_operations=max_operations, error=str(extraction.error_code or "structured_unknown"), project_directory=project_directory, ) return fallback, "fallback", diagnostics plan = extraction.payload normalized_ops = self._normalize_operations(plan.get("operations"), max_operations=max_operations) return ( { "summary": str(plan.get("summary") or "").strip() or "OpenClaw 已生成操作计划。", "confidence": self._to_float(plan.get("confidence"), default=0.0), "operations": normalized_ops, }, "planned", diagnostics, ) except Exception as exc: fallback = await self._build_fallback_plan( tenant_id=tenant_id, project_id=self._to_int(project.id), instruction=instruction, max_operations=max_operations, error=str(exc), project_directory=project_directory, ) return ( fallback, "fallback", { "used_fallback": True, "error_code": "structured_empty", "source_field": None, "provider": "", "model": "", }, ) async def _build_fallback_plan( self, *, tenant_id: int, project_id: Optional[int], instruction: str, max_operations: int, error: str, project_directory: Optional[list[dict[str, Any]]] = None, ) -> dict[str, Any]: instruction_text = str(instruction or "").strip() text = instruction_text.lower() project_ops = await self._build_project_fallback_operations( tenant_id=tenant_id, project_id=project_id, instruction=instruction_text, max_operations=max_operations, project_directory=project_directory, ) if project_ops: return { "summary": "OpenClaw 规划失败，已回退到项目命令最小执行闭环。", "confidence": 0.58, "operations": project_ops, } if self._looks_like_signal_source_instruction(instruction_text): if project_id is None: return { "summary": "当前信号源指令缺少项目上下文，系统未自动执行。", "confidence": 0.0, "operations": [], } operations = await self._build_signal_source_fallback_operations( tenant_id=tenant_id, project_id=int(project_id), instruction=instruction_text, max_operations=max_operations, ) if operations: return { "summary": "OpenClaw 规划失败，已回退为信号源接入最小闭环。", "confidence": 0.28, "operations": operations, } return { "summary": "OpenClaw 未生成可安全执行的信号源接入动作，等待人工补充上下文。", "confidence": 0.0, "operations": [], } unassigned_only = any(token in text for token in ("未分配", "没人负责", "unassigned")) rebalance = any(token in text for token in ("重平衡", "优化负载", "重新分配", "调度")) assign_like = rebalance or any(token in text for token in ("分配", "负责人", "指派", "接管")) if not assign_like: return { "summary": "当前指令缺少足够上下文，系统未执行任何动作。", "confidence": 0.0, "operations": [], } if project_id is None: return { "summary": "当前指令缺少项目上下文，系统未自动执行分配动作。", "confidence": 0.0, "operations": [], } proposals = await workload_service.build_ai_optimize_preview( self.db, tenant_id=tenant_id, project_id=int(project_id), max_moves=max_operations, unassigned_only=unassigned_only and not rebalance, ) operations = [ { "operation_code": "task.assign", "payload": { "task_id": int(item.get("task_id")), "assignee_id": int(item.get("new_assignee_id")), }, "reason": str(item.get("reason") or "fallback_workload_optimizer"), } for item in proposals[:max_operations] if self._to_int(item.get("task_id")) and self._to_int(item.get("new_assignee_id")) ] if operations: return { "summary": "OpenClaw 规划失败，已回退到内置负载优化器生成分配计划。", "confidence": 0.35, "operations": operations, } return { "summary": "系统未生成可安全执行的分配动作，等待补充更明确的负责人或任务上下文。", "confidence": 0.0, "operations": [], } @staticmethod def _build_planner_diagnostics(extraction: Any) -> dict[str, Any]: return { "used_fallback": extraction.error_code is not None, "error_code": extraction.error_code, "source_field": extraction.source_field or None, "provider": str(extraction.provider or ""), "model": str(extraction.model or ""), } @staticmethod def _resolve_openclaw_status(*, execution: dict[str, Any], dry_run: bool) -> str: if dry_run: return "planned" failed_count = int(execution.get("failed_count") or 0) applied_count = int(execution.get("applied_count") or 0) skipped_count = int(execution.get("skipped_count") or 0) if failed_count > 0 and applied_count == 0 and skipped_count == 0: return "failed" if failed_count > 0: return "partial" return "completed" @staticmethod def _looks_like_signal_source_instruction(text: str) -> bool: normalized = re.sub(r"\s+", "", str(text or "")) if len(normalized) < 6: return False if "信号源" in normalized and any(token in normalized for token in ("新增", "增加", "添加", "接入", "配置", "注册", "建立")): return True return bool(SOURCE_ONBOARDING_FALLBACK_RE.search(normalized)) async def _build_signal_source_fallback_operations( self, *, tenant_id: int, project_id: int, instruction: str, max_operations: int, ) -> list[dict[str, Any]]: if max_operations <= 0: return [] project = await self._load_project(tenant_id=tenant_id, project_id=project_id) if project is None: return [] source_name = self._extract_signal_source_name(instruction) charter = await self._load_latest_aligned_charter(tenant_id=tenant_id, project_id=project_id) recipient_id = await self._resolve_signal_source_notification_recipient( tenant_id=tenant_id, owner_id=int(project.owner_id) if project.owner_id else None, ) operations: list[dict[str, Any]] = [] if charter is not None and len(operations) < max_operations: operations.append( { "operation_code": "task.create", "payload": { "project_id": int(project_id), "title": f"接入{source_name}信号源"[:120], "description": self._build_signal_source_task_description( source_name=source_name, instruction=instruction, ), "priority": "high", }, "reason": "fallback_signal_source_onboarding", } ) if recipient_id is not None and len(operations) < max_operations: operations.append( { "operation_code": "notification.create", "payload": { "user_id": int(recipient_id), "title": f"{source_name}信号源接入待确认"[:120], "content": self._build_signal_source_notification_content( source_name=source_name, instruction=instruction, charter_ready=charter is not None, ), }, "reason": "fallback_signal_source_onboarding", } ) return operations[:max_operations] @staticmethod def _extract_signal_source_name(text: str) -> str: cleaned = re.sub(r"\s+", " ", str(text or "")).strip() if not cleaned: return "未命名" repo_match = re.search(r"\b([A-Za-z0-9_.-]+/[A-Za-z0-9_.-]+)\b", cleaned) if repo_match: return repo_match.group(1)[:80] quoted_match = re.search(r"[\"'“](.+?)[\"'”]", cleaned) if quoted_match: candidate = quoted_match.group(1).strip(" []【】()（）:：\"'“”") if candidate: return candidate[:80] for pattern in ( r"(?:新增|增加|添加|接入|配置|注册|建立)(?:一个|1个|一条|一组|新的|一个新的)?\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"^(?:一个|1个|一条|一组|一个新的|新的|新增的)\s+", "", candidate, flags=re.IGNORECASE, ) candidate = re.sub(r"(?:信号源|来源|监控源|source|channel)$", "", candidate, flags=re.IGNORECASE).strip() if candidate: return candidate[:80] lower = cleaned.lower() if "github" in lower: return "GitHub" if "飞书" in cleaned or "lark" in lower: return "Feishu" if "webhook" in lower: return "Webhook" return cleaned[:80] @staticmethod def _build_signal_source_task_description(*, source_name: str, instruction: str) -> str: return ( f"为信号源“{source_name}”完成最小接入任务。" "包含接入配置、权限校验、数据字段确认和回流验证。" f"原始指令：{instruction[:400]}" )[:5000] @staticmethod def _build_signal_source_notification_content(*, source_name: str, instruction: str, charter_ready: bool) -> str: suffix = ( "已自动生成接入任务，请确认负责人和接入边界。" if charter_ready else "当前项目尚未有 aligned charter，系统只先发出确认通知，避免无宪章自动建任务。" ) return f"OpenClaw 已识别到新增信号源请求：{source_name}。{suffix} 原始指令：{instruction[:300]}".strip() async def _resolve_signal_source_notification_recipient( self, *, tenant_id: int, owner_id: Optional[int], ) -> Optional[int]: if owner_id is not None: owner = ( await self.db.execute( select(User.id).where( and_( User.tenant_id == tenant_id, User.id == int(owner_id), User.is_active.is_(True), User.status == "active", ) ) ) ).scalar_one_or_none() if owner is not None: return int(owner) manager = ( await self.db.execute( select(User.id).where( and_( User.tenant_id == tenant_id, User.is_active.is_(True), User.status == "active", User.role.in_(["BOSS", "ADMIN"]), ) ).order_by(User.id.asc()).limit(1) ) ).scalar_one_or_none() return int(manager) if manager is not None else None def _commander_operation_catalog(self) -> list[dict[str, Any]]: items: list[dict[str, Any]] = [] for code in SUPPORTED_COMMANDER_OPERATION_CODES: row = operation_catalog_service.get_operation(code) if row is not None: items.append(row) return items async def _load_project(self, *, tenant_id: int, project_id: int) -> Optional[Project]: return ( await self.db.execute( select(Project).where( and_( Project.tenant_id == tenant_id, Project.id == project_id, ) ) ) ).scalar_one_or_none() async def _load_project_directory(self, *, tenant_id: int, limit: int = 20) -> list[dict[str, Any]]: rows = ( await self.db.execute( select(Project).where( and_( Project.tenant_id == tenant_id, Project.status != "inactive", ) ).order_by(Project.updated_at.desc(), Project.id.desc()).limit(limit) ) ).scalars().all() return [self._serialize_project_directory_item(row) for row in rows] async def _build_project_fallback_operations( self, *, tenant_id: int, project_id: Optional[int], instruction: str, max_operations: int, project_directory: Optional[list[dict[str, Any]]], ) -> list[dict[str, Any]]: if max_operations <= 0: return [] title = self._extract_project_create_title(instruction) if title: payload: dict[str, Any] = {"title": title} description = self._extract_project_create_description(instruction=instruction, title=title) if description: payload["description"] = description return [ { "operation_code": "project.create", "payload": payload, "reason": "explicit_project_create", } ] resolved_project_id = project_id or await self._resolve_project_id_from_instruction( tenant_id=tenant_id, instruction=instruction, project_directory=project_directory, ) if resolved_project_id is None: return [] status_operation = self._build_project_status_fallback_operation( project_id=int(resolved_project_id), instruction=instruction, ) return [status_operation] if status_operation else [] async def _resolve_project_id_from_instruction( self, *, tenant_id: int, instruction: str, project_directory: Optional[list[dict[str, Any]]], ) -> Optional[int]: directory = project_directory or await self._load_project_directory(tenant_id=tenant_id, limit=50) normalized = re.sub(r"\s+", "", str(instruction or "")).lower() for item in directory: title = re.sub(r"\s+", "", str(item.get("title") or "")).lower() project_id = self._to_int(item.get("project_id")) if project_id is None or not title or len(title) < 2: continue if title in normalized: return int(project_id) return None @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 normalized.startswith(("在项目列表", "到项目列表", "进项目列表", "在pmflow", "到pmflow", "进pmflow")): return "" return compact_title(cleaned, max_len=24) @staticmethod def _extract_project_create_title(text: str) -> str: cleaned = re.sub(r"\s+", " ", str(text or "")).strip() if not cleaned: return "" quoted_match = re.search(r"(?:项目|新项目)[^\"“”']{0,10}[\"“”']([^\"“”']{2,40})[\"“”']", cleaned) if quoted_match: return OpenClawCommanderService._normalize_project_title_candidate(quoted_match.group(1)) for pattern in ( 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) if candidate: return OpenClawCommanderService._normalize_project_title_candidate(candidate) return "" @staticmethod def _extract_project_create_description(*, instruction: str, title: str) -> str: cleaned = re.sub(r"\s+", " ", str(instruction or "")).strip() if not cleaned: return "" if title: cleaned = cleaned.replace(title, "", 1).strip(" ，,:：") cleaned = re.sub( r"^(?:帮我|请|麻烦)?(?:创建|新建|建立|开个|开一个|建个|建一个|做个|做一个)(?:一个|个)?项目(?:叫|名为|名称是|名称为|是|为)?", "", cleaned, flags=re.IGNORECASE, ).strip(" ，,:：") if not cleaned: return "" return cleaned[:4000] @staticmethod def _build_project_status_fallback_operation(*, project_id: int, instruction: str) -> Optional[dict[str, Any]]: normalized = re.sub(r"\s+", "", str(instruction or "")).lower() if any(token in normalized for token in ("归档", "archive", "停用项目", "项目归档", "状态改成archived", "状态改为archived")): return { "operation_code": "project.archive", "payload": {"project_id": int(project_id)}, "reason": "explicit_project_archive", } if any( token in normalized for token in ( "重新打开", "重开", "恢复active", "恢复为active", "恢复项目", "激活项目", "设为active", "设成active", "改成active", "改为active", "状态改成active", "状态改为active", "状态设为active", ) ): return { "operation_code": "project.reopen", "payload": {"project_id": int(project_id)}, "reason": "explicit_project_reopen", } if any( token in normalized for token in ( "设为done", "设成done", "改成done", "改为done", "状态改成done", "状态改为done", "状态设为done", "标记完成", ) ): return { "operation_code": "project.update", "payload": {"project_id": int(project_id), "status": "done"}, "reason": "explicit_project_done", } return None @staticmethod def _extract_project_reference_from_execution(execution: dict[str, Any]) -> Optional[dict[str, Any]]: results = execution.get("results") if isinstance(execution, dict) else None if not isinstance(results, list): return None for row in results: if not isinstance(row, dict): continue if str(row.get("entity_type") or "").strip() != "project": continue payload = row.get("result") if isinstance(row.get("result"), dict) else {} project_id = payload.get("project_id") or row.get("entity_id") title = payload.get("title") or payload.get("project_title") or "" status = payload.get("status") or "" if project_id is None: continue return { "id": str(project_id), "title": str(title or "").strip(), "status": str(status or "").strip(), } return None async def _load_latest_aligned_charter(self, *, tenant_id: int, project_id: int) -> Optional[TruthNode]: return ( await self.db.execute( select(TruthNode).where( and_( TruthNode.tenant_id == tenant_id, TruthNode.project_id == int(project_id), TruthNode.node_type == "charter", TruthNode.status.in_(tuple(ALIGNED_STATUSES)), ) ).order_by(TruthNode.updated_at.desc(), TruthNode.id.desc()) ) ).scalars().first() async def _load_project_tasks( self, *, tenant_id: int, project_id: int, members: list[dict[str, Any]], ) -> tuple[list[dict[str, Any]], dict[str, Any]]: task_rows = ( await self.db.execute( select(Task).where( and_( Task.tenant_id == tenant_id, Task.project_id == project_id, ) ).order_by(Task.id.asc()) ) ).scalars().all() member_name_map = {int(item["user_id"]): str(item.get("name") or "") for item in members if self._to_int(item.get("user_id"))} active_tasks = [task for task in task_rows if str(task.status or "").strip().lower() not in DONE_TASK_STATUSES] selected_tasks = active_tasks[:40] if active_tasks else task_rows[:20] serialized = [ { "task_id": int(task.id), "title": str(task.title or "").strip(), "status": str(task.status or "").strip().lower(), "priority": str(task.priority or "").strip().lower(), "assignee_id": int(task.assignee_id) if task.assignee_id else None, "assignee_name": member_name_map.get(int(task.assignee_id), "") if task.assignee_id else "", "required_skill_tag_ids": [int(item) for item in (task.required_skill_tag_ids or []) if self._to_int(item)], "estimated_hours": float(task.estimated_hours or 0.0), } for task in selected_tasks ] summary = { "total": len(task_rows), "active": len(active_tasks), "done": len([task for task in task_rows if str(task.status or "").strip().lower() in DONE_TASK_STATUSES]), "blocked": len([task for task in task_rows if str(task.status or "").strip().lower() == "blocked"]), "sampled": len(serialized), } return serialized, summary async def _load_project_team_context( self, *, tenant_id: int, project_id: int, owner_id: Optional[int], ) -> list[dict[str, Any]]: member_ids_result = await self.db.execute( select(UserProject.user_id).where( and_( UserProject.tenant_id == tenant_id, UserProject.project_id == project_id, ) ) ) member_ids = {int(uid) for uid in member_ids_result.scalars().all()} if owner_id: member_ids.add(int(owner_id)) if not member_ids: manager_result = await self.db.execute( select(User.id).where( and_( User.tenant_id == tenant_id, User.status == "active", User.is_active.is_(True), User.role.in_(["BOSS", "ADMIN"]), ) ).limit(5) ) member_ids = {int(uid) for uid in manager_result.scalars().all()} if not member_ids: return [] users = ( await self.db.execute( select(User).where( and_( User.tenant_id == tenant_id, User.id.in_(sorted(member_ids)), User.status == "active", User.is_active.is_(True), ) ) ) ).scalars().all() if not users: return [] user_ids = [int(user.id) for user in users] workload_rows = ( await self.db.execute( select(UserWorkload).where( and_( UserWorkload.tenant_id == tenant_id, UserWorkload.user_id.in_(user_ids), ) ) ) ).scalars().all() workload_map = {int(item.user_id): item for item in workload_rows} skill_rows = ( await self.db.execute( select( UserSkillBinding.user_id, SkillTag.id, SkillTag.name, SkillTag.slug, UserSkillBinding.proficiency, ) .join(SkillTag, SkillTag.id == UserSkillBinding.skill_tag_id) .where( and_( UserSkillBinding.tenant_id == tenant_id, UserSkillBinding.user_id.in_(user_ids), SkillTag.tenant_id == tenant_id, ) ) ) ).all() skill_map: dict[int, list[dict[str, Any]]] = {} for uid, sid, name, slug, proficiency in skill_rows: skill_map.setdefault(int(uid), []).append( { "skill_id": int(sid), "name": str(name or "").strip(), "slug": str(slug or "").strip(), "proficiency": int(proficiency or 0), } ) members: list[dict[str, Any]] = [] for user in users: uid = int(user.id) workload_row = workload_map.get(uid) member_skills = skill_map.get(uid, []) members.append( { "user_id": uid, "name": str(user.name or user.username or uid), "role": str(user.role or ""), "active": bool(user.is_active and str(user.status or "").strip().lower() == "active"), "workload_ratio": round( float(workload_row.workload_ratio if workload_row is not None else float(user.workload or 0.0)), 2, ), "current_assigned_hours": round( float(workload_row.current_assigned_hours or 0.0) if workload_row else 0.0, 2, ), "max_available_hours_per_week": round( float(workload_row.max_available_hours_per_week or 40.0) if workload_row else 40.0, 2, ), "skill_tag_ids": [int(item["skill_id"]) for item in member_skills], "skill_names": [str(item["name"]) for item in member_skills], } ) members.sort(key=lambda item: (float(item.get("workload_ratio") or 0.0), int(item.get("user_id") or 0))) return members @classmethod def _parse_plan_response(cls, raw_response: str) -> dict[str, Any]: content = (raw_response or "").strip() if not content: raise ValueError("empty llm response") if content.startswith("```"): parts = content.split("```") if len(parts) >= 2: content = parts[1].strip() if content.startswith("json"): content = content[4:].strip() try: parsed = json.loads(content) if isinstance(parsed, dict): return parsed except Exception: pass object_text = cls._extract_first_json_object(content) if not object_text: raise ValueError("llm json not found") parsed = json.loads(object_text) if not isinstance(parsed, dict): raise ValueError("llm json is not object") return parsed @classmethod def _extract_first_json_object(cls, text: str) -> Optional[str]: raw = str(text or "") start = raw.find("{") if start < 0: return None depth = 0 in_str = False escaped = False for idx in range(start, len(raw)): ch = raw[idx] if in_str: if escaped: escaped = False elif ch == "\\": escaped = True elif ch == '"': in_str = False continue if ch == '"': in_str = True continue if ch == "{": depth += 1 elif ch == "}": depth -= 1 if depth == 0: return raw[start : idx + 1] return None def _normalize_operations(self, raw_operations: Any, *, max_operations: int) -> list[dict[str, Any]]: if not isinstance(raw_operations, list): return [] normalized: list[dict[str, Any]] = [] for item in raw_operations: if not isinstance(item, dict): continue operation_code = str(item.get("operation_code") or "").strip().lower() payload = item.get("payload") if not operation_code or not isinstance(payload, dict): continue normalized.append( { "operation_code": operation_code, "payload": dict(payload), "reason": str(item.get("reason") or "").strip()[:300], } ) if len(normalized) >= max_operations: break return normalized @staticmethod def _to_int(value: Any) -> Optional[int]: if isinstance(value, bool): return None try: return int(str(value).strip()) except Exception: return None @staticmethod def _to_float(value: Any, *, default: float = 0.0) -> float: try: return float(value) except Exception: return float(default) class OperationExecutorService: """对白名单操作做 schema 校验并执行。""" def __init__(self, db: AsyncSession) -> None: self.db = db self._project_member_cache: dict[int, set[int]] = {} async def execute_operations( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], operations: list[dict[str, Any]], dry_run: bool, trace_id: str, ) -> dict[str, Any]: seen_idempotency_keys: set[str] = set() results: list[dict[str, Any]] = [] applied_count = 0 skipped_count = 0 failed_count = 0 context_project_id = self._project_context_id(project) for index, item in enumerate(operations, start=1): operation_code = str(item.get("operation_code") or "").strip().lower() raw_payload = item.get("payload") reason = str(item.get("reason") or "").strip() persistent_claim: OperationExecutionLedger | None = None try: payload = self._validate_and_normalize_payload(operation_code=operation_code, payload=raw_payload) idempotency_key = self._build_idempotency_key( operation_code=operation_code, payload=payload, tenant_id=tenant_id, project_id=context_project_id, ) if idempotency_key in seen_idempotency_keys: skipped_count += 1 results.append( { "index": index, "operation_code": operation_code, "payload": payload, "idempotency_key": idempotency_key, "status": "skipped", "message": "duplicate_in_request", } ) continue seen_idempotency_keys.add(idempotency_key) if dry_run: results.append( { "index": index, "operation_code": operation_code, "payload": payload, "idempotency_key": idempotency_key, "status": "planned", "message": reason or "dry_run", } ) continue persistent_claim, duplicate_result = await self._claim_persistent_idempotency( tenant_id=tenant_id, project_id=context_project_id if context_project_id > 0 else None, operation_code=operation_code, payload=payload, idempotency_key=idempotency_key, trace_id=trace_id, ) if duplicate_result is not None: skipped_count += 1 results.append( { "index": index, "operation_code": operation_code, "payload": payload, "idempotency_key": idempotency_key, **duplicate_result, } ) continue result = await self._execute_single_operation( tenant_id=tenant_id, operator_user=operator_user, project=project, operation_code=operation_code, payload=payload, reason=reason, trace_id=trace_id, idempotency_key=idempotency_key, persistent_claim=persistent_claim, ) status = str(result.get("status") or "applied") if status == "applied": applied_count += 1 elif status == "skipped": skipped_count += 1 results.append( { "index": index, "operation_code": operation_code, "payload": payload, "idempotency_key": idempotency_key, **result, } ) except Exception as exc: await self._safe_rollback() failed_count += 1 results.append( { "index": index, "operation_code": operation_code, "payload": raw_payload if isinstance(raw_payload, dict) else {}, "status": "failed", "message": str(exc), } ) return { "dry_run": bool(dry_run), "requested_count": len(operations), "applied_count": applied_count, "skipped_count": skipped_count, "failed_count": failed_count, "results": results, } async def _claim_persistent_idempotency( self, *, tenant_id: int, project_id: Optional[int], operation_code: str, payload: dict[str, Any], idempotency_key: str, trace_id: str, ) -> tuple[Optional[OperationExecutionLedger], Optional[dict[str, Any]]]: if operation_code not in PERSISTENT_IDEMPOTENCY_OPERATION_CODES: return None, None claim = OperationExecutionLedger( tenant_id=tenant_id, operation_code=operation_code, project_id=int(project_id) if project_id is not None and int(project_id) > 0 else None, idempotency_key=idempotency_key, trace_id=trace_id, status="processing", payload=dict(payload), ) self.db.add(claim) try: await self.db.flush() return claim, None except IntegrityError: await self._safe_rollback() existing = await self._load_execution_claim( tenant_id=tenant_id, operation_code=operation_code, idempotency_key=idempotency_key, ) if existing is None: raise if str(existing.status or "").strip().lower() == "failed": await self.db.delete(existing) await self.db.flush() return await self._claim_persistent_idempotency( tenant_id=tenant_id, project_id=project_id, operation_code=operation_code, payload=payload, idempotency_key=idempotency_key, trace_id=trace_id, ) return ( None, { "status": "skipped", "message": "duplicate_persistent" if str(existing.status or "").strip().lower() == "applied" else "duplicate_in_progress", "entity_type": str(existing.entity_type or "").strip() or None, "entity_id": str(existing.entity_id or "").strip() or None, "trace_id": str(existing.trace_id or "").strip() or None, }, ) async def _load_execution_claim( self, *, tenant_id: int, operation_code: str, idempotency_key: str, ) -> Optional[OperationExecutionLedger]: return ( await self.db.execute( select(OperationExecutionLedger).where( and_( OperationExecutionLedger.tenant_id == tenant_id, OperationExecutionLedger.operation_code == operation_code, OperationExecutionLedger.idempotency_key == idempotency_key, ) ) ) ).scalar_one_or_none() async def _safe_rollback(self) -> None: try: await self.db.rollback() except Exception: return def _validate_and_normalize_payload(self, *, operation_code: str, payload: Any) -> dict[str, Any]: row = operation_catalog_service.get_operation(operation_code) if row is None or operation_code not in SUPPORTED_COMMANDER_OPERATION_CODES: raise ValueError(f"unsupported operation_code: {operation_code}") if not isinstance(payload, dict): raise ValueError(f"{operation_code} payload must be object") input_schema = row.get("input_schema") if isinstance(row.get("input_schema"), dict) else {} properties = input_schema.get("properties") if isinstance(input_schema.get("properties"), dict) else {} required = [str(item) for item in (input_schema.get("required") or [])] normalized: dict[str, Any] = {} if input_schema.get("additionalProperties") is False: extra_keys = sorted(set(payload.keys()) - set(properties.keys())) if extra_keys: raise ValueError(f"{operation_code} payload has unsupported fields: {', '.join(extra_keys)}") for field, spec in properties.items(): if field not in payload: continue normalized[field] = self._coerce_value(field=field, value=payload[field], spec=spec if isinstance(spec, dict) else {}) missing = [field for field in required if field not in normalized] if missing: raise ValueError(f"{operation_code} missing required fields: {', '.join(missing)}") return normalized @classmethod def _coerce_value(cls, *, field: str, value: Any, spec: dict[str, Any]) -> Any: value_type = str(spec.get("type") or "").strip().lower() if value_type == "integer": parsed = cls._to_int(value) if parsed is None: raise ValueError(f"{field} must be integer") return parsed if value_type == "number": if isinstance(value, bool): raise ValueError(f"{field} must be number") try: parsed_float = float(value) except Exception as exc: raise ValueError(f"{field} must be number") from exc minimum = spec.get("minimum") maximum = spec.get("maximum") if minimum is not None and parsed_float < float(minimum): raise ValueError(f"{field} below minimum") if maximum is not None and parsed_float > float(maximum): raise ValueError(f"{field} above maximum") return parsed_float if value_type == "string": text = str(value or "").strip() if not text: raise ValueError(f"{field} must be non-empty string") enum_values = spec.get("enum") if isinstance(enum_values, list) and enum_values: enum_map = {str(item).strip().lower(): str(item) for item in enum_values} matched = enum_map.get(text.lower()) if matched is None: raise ValueError(f"{field} must be one of {', '.join(str(item) for item in enum_values)}") text = matched min_length = spec.get("minLength") max_length = spec.get("maxLength") if min_length is not None and len(text) < int(min_length): raise ValueError(f"{field} shorter than minLength") if max_length is not None and len(text) > int(max_length): raise ValueError(f"{field} longer than maxLength") return text if value_type == "array": if not isinstance(value, list): raise ValueError(f"{field} must be array") max_items = spec.get("maxItems") if max_items is not None and len(value) > int(max_items): raise ValueError(f"{field} too many items") item_spec = spec.get("items") if isinstance(spec.get("items"), dict) else {} return [ cls._coerce_value(field=field, value=item, spec=item_spec) for item in value ] return value async def _execute_single_operation( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], operation_code: str, payload: dict[str, Any], reason: str, trace_id: str, idempotency_key: str, persistent_claim: Optional[OperationExecutionLedger], ) -> dict[str, Any]: if operation_code == "project.create": return await self._execute_project_create( tenant_id=tenant_id, operator_user=operator_user, payload=payload, reason=reason, trace_id=trace_id, ) if operation_code == "project.update": return await self._execute_project_update( tenant_id=tenant_id, operator_user=operator_user, project=project, payload=payload, reason=reason, trace_id=trace_id, ) if operation_code == "project.archive": return await self._execute_project_archive( tenant_id=tenant_id, operator_user=operator_user, project=project, payload=payload, reason=reason, trace_id=trace_id, ) if operation_code == "project.reopen": return await self._execute_project_reopen( tenant_id=tenant_id, operator_user=operator_user, project=project, payload=payload, reason=reason, trace_id=trace_id, ) if operation_code == "task.assign": return await self._execute_task_assign( tenant_id=tenant_id, operator_user=operator_user, project=project, payload=payload, reason=reason, trace_id=trace_id, ) if operation_code == "task.create": return await self._execute_task_create( tenant_id=tenant_id, operator_user=operator_user, project=project, payload=payload, reason=reason, trace_id=trace_id, ) if operation_code == "task.prioritize": return await self._execute_task_prioritize( tenant_id=tenant_id, operator_user=operator_user, project=project, payload=payload, reason=reason, trace_id=trace_id, ) if operation_code == "task.status_transition": return await self._execute_task_status_transition( tenant_id=tenant_id, operator_user=operator_user, project=project, payload=payload, reason=reason, trace_id=trace_id, ) if operation_code == "notification.create": return await self._execute_notification_create( tenant_id=tenant_id, operator_user=operator_user, project=project, payload=payload, reason=reason, trace_id=trace_id, idempotency_key=idempotency_key, persistent_claim=persistent_claim, ) if operation_code == "personnel.workload_update": return await self._execute_personnel_workload_update( tenant_id=tenant_id, operator_user=operator_user, project=project, payload=payload, reason=reason, trace_id=trace_id, ) raise ValueError(f"unsupported operation_code: {operation_code}") async def _execute_project_create( self, *, tenant_id: int, operator_user: User, payload: dict[str, Any], reason: str, trace_id: str, ) -> dict[str, Any]: title = compact_title(str(payload["title"]), max_len=24) if not title: raise ValueError("project title required") existing = await self._load_project_by_title(tenant_id=tenant_id, title=title) if existing is not None: return { "status": "skipped", "message": "project_already_exists", "entity_type": "project", "entity_id": str(existing.id), "result": { "project_id": int(existing.id), "title": str(existing.title or existing.name or "").strip(), "status": str(existing.status or "").strip(), }, } owner_id = self._to_int(payload.get("owner_id")) or int(operator_user.id) if owner_id is not None: owner = await self._load_active_user(tenant_id=tenant_id, user_id=int(owner_id)) if owner is None: owner_id = int(operator_user.id) await ensure_project_id_sequence(self.db) project = Project( tenant_id=tenant_id, name=title, title=title, description=str(payload.get("description") or "").strip()[:4000] or None, owner_id=int(owner_id) if owner_id is not None else None, status="active", domain_id=None, assigned_role="ai_engineer", signal_count=0, ) self.db.add(project) await self.db.flush() await ensure_project_default_members( self.db, tenant_id=tenant_id, project_id=int(project.id), owner_id=int(project.owner_id) if project.owner_id is not None else None, assigned_role=project.assigned_role, ) await log_operation( self.db, tenant_id=tenant_id, user_id=int(operator_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, "owner_id": int(project.owner_id) if project.owner_id is not None else None, }, remark=f"{trace_id}:{reason}"[:1000] if reason else trace_id, ) await self.db.commit() await context_cache_service.invalidate_system_context(tenant_id=tenant_id) return { "status": "applied", "message": f"project created #{int(project.id)}", "entity_type": "project", "entity_id": str(project.id), "result": { "project_id": int(project.id), "title": str(project.title or project.name or "").strip(), "status": str(project.status or "").strip(), }, } async def _execute_project_update( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], payload: dict[str, Any], reason: str, trace_id: str, ) -> dict[str, Any]: target_project = await self._load_project_for_operation( tenant_id=tenant_id, project=project, payload_project_id=int(payload["project_id"]), operation_code="project.update", ) before = { "title": str(target_project.title or target_project.name or "").strip(), "description": str(target_project.description or "").strip(), "status": str(target_project.status or "").strip(), } changed = False if "title" in payload: next_title = compact_title(str(payload.get("title") or ""), max_len=24) if not next_title: raise ValueError("project title required") if next_title != str(target_project.title or target_project.name or "").strip(): target_project.title = next_title target_project.name = next_title changed = True if "description" in payload: next_description = str(payload.get("description") or "").strip()[:4000] or None if next_description != (target_project.description or None): target_project.description = next_description changed = True if "status" in payload: next_status = self._normalize_project_status(payload.get("status")) if next_status != str(target_project.status or "").strip().lower(): target_project.status = next_status changed = True if not changed: return { "status": "skipped", "message": "already_in_target_state", "entity_type": "project", "entity_id": str(target_project.id), "result": { "project_id": int(target_project.id), "title": str(target_project.title or target_project.name or "").strip(), "status": str(target_project.status or "").strip(), }, } await log_operation( self.db, tenant_id=tenant_id, user_id=int(operator_user.id), project_id=int(target_project.id), entity_type="project", entity_id=str(target_project.id), action="project.update", before_data=before, after_data={ "title": str(target_project.title or target_project.name or "").strip(), "description": str(target_project.description or "").strip(), "status": str(target_project.status or "").strip(), }, remark=f"{trace_id}:{reason}"[:1000] if reason else trace_id, ) await self.db.commit() await context_cache_service.invalidate_system_context(tenant_id=tenant_id) return { "status": "applied", "message": f"project updated #{int(target_project.id)}", "entity_type": "project", "entity_id": str(target_project.id), "result": { "project_id": int(target_project.id), "title": str(target_project.title or target_project.name or "").strip(), "status": str(target_project.status or "").strip(), }, } async def _execute_project_archive( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], payload: dict[str, Any], reason: str, trace_id: str, ) -> dict[str, Any]: return await self._execute_project_update( tenant_id=tenant_id, operator_user=operator_user, project=project, payload={"project_id": int(payload["project_id"]), "status": "archived"}, reason=reason, trace_id=trace_id, ) async def _execute_project_reopen( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], payload: dict[str, Any], reason: str, trace_id: str, ) -> dict[str, Any]: return await self._execute_project_update( tenant_id=tenant_id, operator_user=operator_user, project=project, payload={"project_id": int(payload["project_id"]), "status": "active"}, reason=reason, trace_id=trace_id, ) async def _execute_task_create( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], payload: dict[str, Any], reason: str, trace_id: str, ) -> dict[str, Any]: project = self._require_project_context(project=project, operation_code="task.create") project_id = int(payload["project_id"]) if project_id != int(project.id): raise ValueError("task.create project_id mismatch") charter = await self._load_latest_aligned_charter(tenant_id=tenant_id, project_id=project_id) if charter is None: raise ValueError("aligned charter required before task.create") assignee_id = self._to_int(payload.get("assignee_id")) touched_user_ids: list[int] = [] if assignee_id is not None: assignee = await self._load_active_user(tenant_id=tenant_id, user_id=assignee_id) if assignee is None: raise ValueError("assignee not found or inactive") member_ids = await self._get_project_member_ids( tenant_id=tenant_id, project_id=project_id, owner_id=int(project.owner_id) if project.owner_id else None, ) if int(assignee.id) not in member_ids: raise ValueError("assignee is not project member") touched_user_ids.append(int(assignee.id)) due_date = self._parse_due_date_value(payload.get("due_date")) required_skill_tag_ids = self._safe_int_list(payload.get("required_skill_tag_ids")) task = await persist_task( self.db, tenant_id=tenant_id, truth_node_id=int(charter.id), project_id=project_id, draft=build_task_create_draft( title=str(payload["title"]), description=str(payload.get("description") or "").strip() or None, assignee_id=assignee_id, required_skill_tag_ids=required_skill_tag_ids, priority=payload.get("priority"), status="todo", estimated_hours=self._to_float(payload.get("estimated_hours"), default=0.0) if payload.get("estimated_hours") is not None else None, due_date=due_date, ), actor_user_id=int(operator_user.id), action="task.create", remark=f"{trace_id}:{reason}"[:1000] if reason else trace_id, commit=not touched_user_ids, refresh_workload=bool(touched_user_ids), ) await context_cache_service.invalidate_system_context(tenant_id=tenant_id) for user_id in touched_user_ids: await context_cache_service.invalidate_user_context(tenant_id=tenant_id, user_id=user_id) self._sync_task_rag(task=task, tenant_id=tenant_id) return { "status": "applied", "message": f"task created #{int(task.id)}", "entity_type": "task", "entity_id": str(task.id), } async def _execute_task_assign( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], payload: dict[str, Any], reason: str, trace_id: str, ) -> dict[str, Any]: project = self._require_project_context(project=project, operation_code="task.assign") task = await self._load_task(tenant_id=tenant_id, task_id=int(payload["task_id"])) if task is None or int(task.project_id) != int(project.id): raise ValueError("task not found in target project") assignee = await self._load_active_user(tenant_id=tenant_id, user_id=int(payload["assignee_id"])) if assignee is None: raise ValueError("assignee not found or inactive") member_ids = await self._get_project_member_ids(tenant_id=tenant_id, project_id=int(project.id), owner_id=int(project.owner_id) if project.owner_id else None) if int(assignee.id) not in member_ids: raise ValueError("assignee is not project member") current_assignee_id = int(task.assignee_id) if task.assignee_id else None if current_assignee_id == int(assignee.id): return { "status": "skipped", "message": "already_assigned", "entity_type": "task", "entity_id": str(task.id), } before = { "assignee_id": current_assignee_id, "status": task.status, } task.assignee_id = int(assignee.id) await log_operation( self.db, tenant_id=tenant_id, user_id=int(operator_user.id), project_id=int(project.id), entity_type="task", entity_id=str(task.id), action="task.assign", before_data=before, after_data={"assignee_id": int(assignee.id), "status": task.status}, remark=f"{trace_id}:{reason}"[:1000] if reason else trace_id, ) touched_user_ids: list[int] = [] if current_assignee_id: touched_user_ids.append(current_assignee_id) touched_user_ids.append(int(assignee.id)) for uid in touched_user_ids: await workload_service.refresh_user_assigned_hours_from_tasks(self.db, tenant_id=tenant_id, user_id=uid) await context_cache_service.invalidate_system_context(tenant_id=tenant_id) for uid in sorted(set(touched_user_ids)): await context_cache_service.invalidate_user_context(tenant_id=tenant_id, user_id=uid) self._sync_task_rag(task=task, tenant_id=tenant_id) return { "status": "applied", "message": f"task assigned to user#{int(assignee.id)}", "entity_type": "task", "entity_id": str(task.id), } async def _execute_task_prioritize( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], payload: dict[str, Any], reason: str, trace_id: str, ) -> dict[str, Any]: project = self._require_project_context(project=project, operation_code="task.prioritize") task = await self._load_task(tenant_id=tenant_id, task_id=int(payload["task_id"])) if task is None or int(task.project_id) != int(project.id): raise ValueError("task not found in target project") before_priority = self._normalize_task_priority(task.priority) next_priority = self._normalize_task_priority(payload.get("priority")) if before_priority == next_priority: return { "status": "skipped", "message": "already_in_target_priority", "entity_type": "task", "entity_id": str(task.id), } task.priority = next_priority await log_operation( self.db, tenant_id=tenant_id, user_id=int(operator_user.id), project_id=int(project.id), entity_type="task", entity_id=str(task.id), action="task.prioritize", before_data={"priority": before_priority}, after_data={"priority": next_priority}, remark=f"{trace_id}:{reason}"[:1000] if reason else trace_id, ) await self.db.commit() await context_cache_service.invalidate_system_context(tenant_id=tenant_id) if task.assignee_id: await context_cache_service.invalidate_user_context(tenant_id=tenant_id, user_id=int(task.assignee_id)) self._sync_task_rag(task=task, tenant_id=tenant_id) return { "status": "applied", "message": f"task priority updated to {next_priority}", "entity_type": "task", "entity_id": str(task.id), } async def _execute_task_status_transition( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], payload: dict[str, Any], reason: str, trace_id: str, ) -> dict[str, Any]: project = self._require_project_context(project=project, operation_code="task.status_transition") task = await self._load_task(tenant_id=tenant_id, task_id=int(payload["task_id"])) if task is None or int(task.project_id) != int(project.id): raise ValueError("task not found in target project") status = str(payload["status"]).strip().lower() if status not in SUPPORTED_TASK_STATUSES: raise ValueError("unsupported task status") before_status = str(task.status or "").strip().lower() if before_status == status: return { "status": "skipped", "message": "already_in_target_status", "entity_type": "task", "entity_id": str(task.id), } task.status = status if status in DONE_TASK_STATUSES: task.completed_at = datetime.now(timezone.utc).replace(tzinfo=None) await log_operation( self.db, tenant_id=tenant_id, user_id=int(operator_user.id), project_id=int(project.id), entity_type="task", entity_id=str(task.id), action="task.status_transition", before_data={"status": before_status}, after_data={"status": status}, remark=f"{trace_id}:{reason}"[:1000] if reason else trace_id, ) if task.assignee_id: await workload_service.refresh_user_assigned_hours_from_tasks( self.db, tenant_id=tenant_id, user_id=int(task.assignee_id), ) else: await self.db.commit() await context_cache_service.invalidate_system_context(tenant_id=tenant_id) if task.assignee_id: await context_cache_service.invalidate_user_context(tenant_id=tenant_id, user_id=int(task.assignee_id)) self._sync_task_rag(task=task, tenant_id=tenant_id) if status in DONE_TASK_STATUSES and before_status not in DONE_TASK_STATUSES and task.assignee_id: asyncio.create_task( process_task_completion( task_id=int(task.id), tenant_id=tenant_id, operator_user_id=int(operator_user.id), ) ) return { "status": "applied", "message": f"task status updated to {status}", "entity_type": "task", "entity_id": str(task.id), } async def _execute_notification_create( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], payload: dict[str, Any], reason: str, trace_id: str, idempotency_key: str, persistent_claim: Optional[OperationExecutionLedger], ) -> dict[str, Any]: project = self._require_project_context(project=project, operation_code="notification.create") user = await self._load_active_user(tenant_id=tenant_id, user_id=int(payload["user_id"])) if user is None: raise ValueError("notification target user not found or inactive") notification = await NotificationService(self.db).create_notification( tenant_id=tenant_id, user_id=int(user.id), category="system", title=str(payload["title"]).strip(), content=str(payload["content"]).strip(), source="openclaw_commander", entity_type="project", entity_id=int(project.id), target_url=f"/projects/{int(project.id)}/board", commit=False, ) if persistent_claim is not None: persistent_claim.status = "applied" persistent_claim.entity_type = "notification" persistent_claim.entity_id = str(notification.id) persistent_claim.result_payload = { "status": "applied", "notification_id": int(notification.id), "idempotency_key": idempotency_key, } persistent_claim.error_message = None await log_operation( self.db, tenant_id=tenant_id, user_id=int(operator_user.id), project_id=int(project.id), entity_type="notification", entity_id=str(notification.id), action="notification.create", before_data=None, after_data={ "user_id": int(user.id), "title": notification.title, "source": notification.source, }, remark=f"{trace_id}:{reason}"[:1000] if reason else trace_id, ) await self.db.commit() return { "status": "applied", "message": f"notification created for user#{int(user.id)}", "entity_type": "notification", "entity_id": str(notification.id), } async def _execute_personnel_workload_update( self, *, tenant_id: int, operator_user: User, project: Optional[Project | SimpleNamespace], payload: dict[str, Any], reason: str, trace_id: str, ) -> dict[str, Any]: project = self._require_project_context(project=project, operation_code="personnel.workload_update") user = await self._load_active_user(tenant_id=tenant_id, user_id=int(payload["user_id"])) if user is None: raise ValueError("target user not found or inactive") member_ids = await self._get_project_member_ids( tenant_id=tenant_id, project_id=int(project.id), owner_id=int(project.owner_id) if project.owner_id else None, ) if int(user.id) not in member_ids: raise ValueError("target user is not project member") before_workload = await self._load_user_workload(tenant_id=tenant_id, user_id=int(user.id)) result = await workload_service.update_user_max_weekly_hours( self.db, tenant_id=tenant_id, user_id=int(user.id), max_weekly_hours=float(payload["max_weekly_hours"]), ) if result is None: raise ValueError("target user workload not found") await log_operation( self.db, tenant_id=tenant_id, user_id=int(operator_user.id), project_id=int(project.id), entity_type="user_workload", entity_id=str(user.id), action="personnel.workload_update", before_data=before_workload, after_data=result, remark=f"{trace_id}:{reason}"[:1000] if reason else trace_id, ) await self.db.commit() await context_cache_service.invalidate_system_context(tenant_id=tenant_id) await context_cache_service.invalidate_user_context(tenant_id=tenant_id, user_id=int(user.id)) return { "status": "applied", "message": f"user#{int(user.id)} max weekly hours updated", "entity_type": "user_workload", "entity_id": str(user.id), } async def _load_task(self, *, tenant_id: int, task_id: int) -> Optional[Task]: return ( await self.db.execute( select(Task).where( and_( Task.tenant_id == tenant_id, Task.id == int(task_id), ) ) ) ).scalar_one_or_none() async def _load_project_by_title(self, *, tenant_id: int, title: str) -> Optional[Project]: normalized = compact_title(str(title or ""), max_len=24) if not normalized: return None return ( await self.db.execute( select(Project).where( and_( Project.tenant_id == tenant_id, or_( func.lower(Project.title) == normalized.lower(), func.lower(Project.name) == normalized.lower(), ), Project.status != "inactive", ) ).limit(1) ) ).scalar_one_or_none() async def _load_project_for_operation( self, *, tenant_id: int, project: Optional[Project | SimpleNamespace], payload_project_id: int, operation_code: str, ) -> Project: context_project_id = self._project_context_id(project) if context_project_id > 0 and context_project_id != int(payload_project_id): raise ValueError(f"{operation_code} project_id mismatch") row = await self._load_project(tenant_id=tenant_id, project_id=int(payload_project_id)) if row is None: raise ValueError("project not found") return row async def _load_active_user(self, *, tenant_id: int, user_id: int) -> Optional[User]: return ( await self.db.execute( select(User).where( and_( User.tenant_id == tenant_id, User.id == int(user_id), User.is_active.is_(True), User.status == "active", ) ) ) ).scalar_one_or_none() async def _load_user_workload(self, *, tenant_id: int, user_id: int) -> Optional[dict[str, Any]]: row = ( await self.db.execute( select(UserWorkload).where( and_( UserWorkload.tenant_id == tenant_id, UserWorkload.user_id == int(user_id), ) ) ) ).scalar_one_or_none() if row is None: return None return { "user_id": str(user_id), "current_assigned_hours": float(row.current_assigned_hours or 0.0), "max_weekly_hours": float(row.max_available_hours_per_week or 40.0), "workload_ratio": float(row.workload_ratio or 0.0), } async def _get_project_member_ids(self, *, tenant_id: int, project_id: int, owner_id: Optional[int]) -> set[int]: cached = self._project_member_cache.get(int(project_id)) if cached is not None: return set(cached) rows = await self.db.execute( select(UserProject.user_id).where( and_( UserProject.tenant_id == tenant_id, UserProject.project_id == int(project_id), ) ) ) member_ids = {int(uid) for uid in rows.scalars().all()} if owner_id: member_ids.add(int(owner_id)) self._project_member_cache[int(project_id)] = set(member_ids) return member_ids async def _load_latest_aligned_charter(self, *, tenant_id: int, project_id: int) -> Optional[TruthNode]: return ( await self.db.execute( select(TruthNode).where( and_( TruthNode.tenant_id == tenant_id, TruthNode.project_id == int(project_id), TruthNode.node_type == "charter", TruthNode.status.in_(tuple(ALIGNED_STATUSES)), ) ).order_by(TruthNode.updated_at.desc(), TruthNode.id.desc()) ) ).scalars().first() @staticmethod def _project_context_id(project: Optional[Project | SimpleNamespace]) -> int: if project is None: return 0 try: return max(0, int(getattr(project, "id", 0) or 0)) except Exception: return 0 def _require_project_context( self, *, project: Optional[Project | SimpleNamespace], operation_code: str, ) -> Project | SimpleNamespace: if self._project_context_id(project) <= 0: raise ValueError(f"{operation_code} requires project context") if project is None: raise ValueError(f"{operation_code} requires project context") return project def _build_idempotency_key( self, *, operation_code: str, payload: dict[str, Any], tenant_id: int, project_id: int, ) -> str: row = operation_catalog_service.get_operation(operation_code) fields = [] if isinstance(row, dict): fields = [str(item) for item in ((row.get("idempotency_key") or {}).get("fields") or [])] parts: dict[str, Any] = {} for field in fields: if field == "tenant_id": parts[field] = int(tenant_id) elif field == "project_id": parts[field] = int(project_id) elif field == "content_hash": parts[field] = hashlib.sha256(str(payload.get("content") or "").strip().encode("utf-8")).hexdigest()[:16] else: parts[field] = payload.get(field) encoded = json.dumps(parts, ensure_ascii=False, sort_keys=True, default=str) return hashlib.sha256(f"{operation_code}|{encoded}".encode("utf-8")).hexdigest()[:24] def _sync_task_rag(self, *, task: Task, tenant_id: int) -> None: try: content = ( (task.description or "") + f"\n状态: {task.status}" + f"\n负责人: {int(task.assignee_id) if task.assignee_id else '未分配'}" ) asyncio.create_task( AdaptiveRAGService.upsert_pmflow_record_async( tenant_id=tenant_id, record_type="task", record_id=int(task.id), project_id=int(task.project_id), title=str(task.title or "任务"), content=content, ) ) except Exception: return @staticmethod def _to_int(value: Any) -> Optional[int]: if isinstance(value, bool): return None try: return int(str(value).strip()) except Exception: return None @staticmethod def _safe_int_list(value: Any) -> list[int]: if not isinstance(value, list): return [] items: list[int] = [] for raw in value: try: items.append(int(raw)) except Exception: continue return items @classmethod def _normalize_task_priority(cls, value: Any) -> str: text = str(value or "medium").strip().lower() if text in {"p0", "critical", "high"}: return "high" if text in {"p2", "p3", "low"}: return "low" return "medium" @staticmethod def _normalize_project_status(value: Any) -> str: text = str(value or "active").strip().lower() if text not in SUPPORTED_PROJECT_STATUSES: raise ValueError("unsupported project status") return text @staticmethod def _parse_due_date_value(value: Any) -> Optional[datetime]: text = str(value or "").strip() if not text: return None try: return datetime.fromisoformat(text.replace("Z", "+00:00")).replace(tzinfo=None) except ValueError as exc: raise ValueError("invalid due_date") from exc