From 7c9d274a6cb35284bc66da581232538dae2d669d Mon Sep 17 00:00:00 2001 From: Yuuuu <1569752928@qq.com> Date: Wed, 11 Mar 2026 16:41:29 +0800 Subject: [PATCH] =?UTF-8?q?=E5=88=86=E6=9E=90LLM=E8=B0=83=E7=94=A8?= =?UTF-8?q?=E8=B6=85=E6=97=B6=E9=85=8D=E7=BD=AE=E5=8F=98=E5=8C=96=E5=BD=B1?= =?UTF-8?q?=E5=93=8D=E8=AF=A5=E8=AF=B7=E6=B1=82?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .openclaw/pmflow-routing-rules.json | 16 +-- CLAUDE.md | 1 + ...0260311_shift_all_timestamps_to_beijing.py | 3 +- backend/app/api/ai_steward.py | 6 +- backend/app/api/budget.py | 3 +- backend/app/api/notifications.py | 5 +- backend/app/api/skill_tags.py | 5 +- backend/app/api/sstc.py | 21 +-- backend/app/api/users.py | 7 +- backend/app/api/webhooks.py | 3 +- backend/app/dispatcher.py | 3 +- backend/app/events.py | 3 +- backend/app/main.py | 3 +- backend/app/services/adaptive_rag_service.py | 25 ++-- .../services/ai_orchestration_run_service.py | 4 +- .../app/services/ai_pmo_cadence_service.py | 4 +- backend/app/services/ai_pmo_kernel_service.py | 62 ++++++++- backend/app/services/ai_service.py | 3 +- backend/app/services/auditor.py | 3 +- backend/app/services/auth.py | 9 +- backend/app/services/auto_merger.py | 15 ++- backend/app/services/blocker_detection.py | 3 +- .../services/budget_enforcement_service.py | 7 +- backend/app/services/codegen.py | 7 +- backend/app/services/context_cache.py | 11 +- backend/app/services/daily_briefing.py | 5 +- backend/app/services/domain.py | 5 +- backend/app/services/event_workflow.py | 3 +- .../app/services/experiment_sync_service.py | 7 +- backend/app/services/feishu_integration.py | 66 +++++++++- .../app/services/feishu_long_connection.py | 23 +++- .../app/services/feishu_signal_pipeline.py | 4 +- backend/app/services/integration.py | 37 +++--- backend/app/services/llm.py | 121 +++++++++++++++++- .../app/services/model_capability_service.py | 3 +- backend/app/services/notification_service.py | 5 +- .../services/openclaw_commander_service.py | 4 +- .../app/services/project_knowledge_graph.py | 2 +- .../app/services/prompt_profile_service.py | 3 +- backend/app/services/signal_capture.py | 19 +-- .../signal_source_orchestration_service.py | 2 +- backend/app/services/skill_score_service.py | 5 +- backend/app/services/skill_service.py | 3 +- backend/app/services/sstc_service.py | 5 +- backend/app/services/symbol_index_service.py | 9 +- backend/app/services/task_persistence.py | 3 +- .../app/services/workload_alert_service.py | 3 +- backend/app/services/workload_service.py | 3 +- backend/app/websocket.py | 15 ++- scripts/ai_steward_eval.py | 5 +- scripts/ai_steward_pressure_coach.py | 5 +- 51 files changed, 436 insertions(+), 161 deletions(-) diff --git a/.openclaw/pmflow-routing-rules.json b/.openclaw/pmflow-routing-rules.json index 30e79615..649fa1f4 100644 --- a/.openclaw/pmflow-routing-rules.json +++ b/.openclaw/pmflow-routing-rules.json @@ -1,6 +1,6 @@ { "version": "v2", - "updated_at": "2026-03-11T00:00:00Z", + "updated_at": "2026-03-11T08:35:00Z", "source_of_truth": "openclaw", "description": "PMFlow 三引擎统一路由：所有主链路统一经 OpenClaw OpenAI-Compatible 网关。", "rules": [ @@ -10,7 +10,7 @@ "priority": 320, "tier": "high", "provider": "openai", - "model": "MiniMax-M2.5", + "model": "GLM-5", "api_env": "OPENAI_API_KEY", "function_exact": "classify_intent_complex" }, @@ -32,12 +32,12 @@ ] }, { - "name": "engine_high_reasoning_minimax", + "name": "engine_high_reasoning_glm5", "enabled": true, "priority": 220, "tier": "high", "provider": "openai", - "model": "MiniMax-M2.5", + "model": "GLM-5", "api_env": "OPENAI_API_KEY", "match_any": [ "adaptive_rag_answer", @@ -62,7 +62,7 @@ "priority": 200, "tier": "high", "provider": "openai", - "model": "MiniMax-M2.5", + "model": "GLM-5", "api_env": "OPENAI_API_KEY", "match_any": [ "analyze_food_calories", @@ -80,9 +80,9 @@ "enabled": true, "priority": 50, "tier": "safety", - "provider": "openai", - "model": "deepseek-chat", - "api_env": "OPENAI_API_KEY", + "provider": "glm", + "model": "GLM-4.7", + "api_env": "GLM_API_KEY", "match_all": true } ] diff --git a/CLAUDE.md b/CLAUDE.md index 570856eb..89477318 100644 --- a/CLAUDE.md +++ b/CLAUDE.md @@ -174,3 +174,4 @@ uvicorn app.main:app --reload --host 0.0.0.0 --port 8000 - 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*` 写入入口均已切换到北京时间 +- 后端与脚本层已清零 `datetime.utcnow()/timezone.utc`：统一使用 `backend/app/utils/timezone.py`（`now_cn_aware/now_cn_naive/CN_TZ`）管理时间语义 diff --git a/backend/alembic/versions/20260311_shift_all_timestamps_to_beijing.py b/backend/alembic/versions/20260311_shift_all_timestamps_to_beijing.py index ab993abb..87ad0f92 100644 --- a/backend/alembic/versions/20260311_shift_all_timestamps_to_beijing.py +++ b/backend/alembic/versions/20260311_shift_all_timestamps_to_beijing.py @@ -67,4 +67,5 @@ def downgrade() -> None: if not _is_postgresql(): return _shift_all_timestamp_columns(hours=-8) - _set_database_timezone("UTC") + # 统一策略：即使降级历史数据，也保持数据库会话时区为北京时间。 + _set_database_timezone("Asia/Shanghai") diff --git a/backend/app/api/ai_steward.py b/backend/app/api/ai_steward.py index db1686ad..a5cd63cf 100644 --- a/backend/app/api/ai_steward.py +++ b/backend/app/api/ai_steward.py @@ -33,9 +33,9 @@ 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 +from ..utils.timezone import now_cn_aware, now_cn_naive router = APIRouter(prefix="/ai-steward", tags=["AI Steward"]) @@ -264,7 +264,7 @@ async def onboard_signal_source( { "source": "pmflow_manual_signal_source_onboard", "instruction": request.instruction[:2000], - "created_at": datetime.now(timezone.utc).isoformat(), + "created_at": now_cn_aware().isoformat(), }, ensure_ascii=False, ), @@ -1103,7 +1103,7 @@ async def get_ai_steward_stats( 'active_blockers': active_blockers, 'active_projects': active_projects, 'hours_saved': hours_saved, - 'last_updated': datetime.now(timezone.utc).isoformat() + 'last_updated': now_cn_aware().isoformat() } return stats diff --git a/backend/app/api/budget.py b/backend/app/api/budget.py index cdbfc2f3..74e439e0 100644 --- a/backend/app/api/budget.py +++ b/backend/app/api/budget.py @@ -20,6 +20,7 @@ from ..schemas.budget import ( ) from ..services.budget_enforcement_service import BudgetEnforcementService +from ..utils.timezone import now_cn_naive router = APIRouter(prefix="/budget", tags=["budget"]) @@ -84,7 +85,7 @@ async def create_budget( db: AsyncSession = Depends(get_db), ): _ = current_user - now = datetime.utcnow() + now = now_cn_naive() period_start, period_end = _period_window(now, request.period) new_budget = Budget( tenant_id=tenant_id, diff --git a/backend/app/api/notifications.py b/backend/app/api/notifications.py index c91d8853..9e44091e 100644 --- a/backend/app/api/notifications.py +++ b/backend/app/api/notifications.py @@ -15,6 +15,7 @@ from ..dependencies import get_current_user, get_tenant_id from ..models import Notification, User +from ..utils.timezone import now_cn_naive router = APIRouter(prefix="/notifications", tags=["notifications"]) @@ -136,7 +137,7 @@ async def mark_notification_read( if not row.is_read: row.is_read = True - row.read_at = datetime.now(timezone.utc) + row.read_at = now_cn_naive() await db.commit() return success_response(data=_serialize_notification(row)) @@ -160,7 +161,7 @@ async def mark_all_notifications_read( ) ).scalars().all() - now = datetime.now(timezone.utc) + now = now_cn_naive() for row in rows: row.is_read = True row.read_at = now diff --git a/backend/app/api/skill_tags.py b/backend/app/api/skill_tags.py index 5250e3e4..241b6d84 100644 --- a/backend/app/api/skill_tags.py +++ b/backend/app/api/skill_tags.py @@ -17,6 +17,7 @@ from ..services.ai_pmo_kernel_service import AIPMOKernelService from ..services import skill_service from ..services.auth import has_any_role +from ..utils.timezone import now_cn_naive router = APIRouter(tags=["skill-tags"]) @@ -141,7 +142,7 @@ async def update_skill_tag( tag.description = body.description if body.sort_order is not None: tag.sort_order = body.sort_order - tag.updated_at = datetime.utcnow() + tag.updated_at = now_cn_naive() await db.commit() await db.refresh(tag) return _to_tag_payload(tag) @@ -285,7 +286,7 @@ async def update_user_skill( row.quality_score = body.quality_score if body.efficiency_score is not None: row.efficiency_score = body.efficiency_score - row.updated_at = datetime.utcnow() + row.updated_at = now_cn_naive() await db.commit() await AIPMOKernelService.invalidate_world_model_cache_for_user(db=db, tenant_id=tenant_id, user_id=user_id) return {"message": "Skill updated"} diff --git a/backend/app/api/sstc.py b/backend/app/api/sstc.py index e664b917..fb320d22 100644 --- a/backend/app/api/sstc.py +++ b/backend/app/api/sstc.py @@ -79,6 +79,7 @@ from ..dependencies import get_current_user, require_project_member_or_manager from ..utils.title import compact_title +from ..utils.timezone import CN_TZ, now_cn_aware, now_cn_naive router = APIRouter(tags=["sstc"]) _FEISHU_PLACEHOLDER_RE = re.compile(r"^飞书用户\(([^)]+)\)$") _SIGNAL_PREFIX_RE = re.compile(r"^\[(.*?)\]\s*:\s*") @@ -866,7 +867,7 @@ async def _build_ai_first_charter_content( fallback_model = "同供应商低成本模型" source_refs = [f"signal:{sid}" for sid in signal_ids] - now_iso = datetime.now(timezone.utc).isoformat() + now_iso = now_cn_aware().isoformat() validations_m1 = _default_ai_validations("模块1：触发与目标") validations_m2 = _default_ai_validations("模块2：范围与对齐") validations_m3 = _default_ai_validations("模块3：模型与依赖") @@ -1397,7 +1398,7 @@ def _normalize_review_content(raw: dict[str, Any] | None) -> dict[str, Any]: "what_went_well": [str(x).strip() for x in what_went_well if str(x).strip()], "root_causes": [str(x).strip() for x in root_causes if str(x).strip()], "improvements": normalized_improvements, - "last_updated_at": datetime.now(timezone.utc).isoformat(), + "last_updated_at": now_cn_aware().isoformat(), } @@ -1806,7 +1807,7 @@ async def list_signals( } created_at_dt = s.created_at if created_at_dt is not None and created_at_dt.tzinfo is not None: - created_at_dt = created_at_dt.astimezone(timezone.utc).replace(tzinfo=None) + created_at_dt = created_at_dt.astimezone(CN_TZ).replace(tzinfo=None) if not _match_signal_filters( row=row, status_filter=normalized_status_filter, @@ -1911,7 +1912,7 @@ async def upload_signal_attachment( "image_key": None, "media_id": None, "source": "manual_upload", - "occurred_at": datetime.now(timezone.utc).isoformat(), + "occurred_at": now_cn_aware().isoformat(), "download_url": _build_signal_attachment_download_url(pid, sid, attachment_id), } @@ -2918,7 +2919,7 @@ async def get_task_board( risk_count = 0 if (t.status or "").lower() == "blocked": risk_count += 1 - if t.due_date and t.due_date < datetime.now(timezone.utc).replace(tzinfo=None) and (t.status or "").lower() not in DONE_STATUSES: + if t.due_date and t.due_date < now_cn_aware().replace(tzinfo=None) and (t.status or "").lower() not in DONE_STATUSES: risk_count += 1 rows.append( { @@ -2960,7 +2961,7 @@ async def patch_task_status( before = {"status": task.status} task.status = body.status if body.status in DONE_STATUSES: - task.completed_at = datetime.now(timezone.utc).replace(tzinfo=None) + task.completed_at = now_cn_naive() await log_operation( db, tenant_id=tenant_id, @@ -3420,7 +3421,7 @@ async def patch_release_status( release_node.status = body.status content = _as_dict(release_node.content) content["status_note"] = body.note - content["last_status_changed_at"] = datetime.now(timezone.utc).isoformat() + content["last_status_changed_at"] = now_cn_aware().isoformat() release_node.content = content await log_operation( db, @@ -3525,7 +3526,7 @@ async def patch_release_verification_item( target["status"] = body.status if body.status == "done": target["verified_by"] = str(current_user.id) - target["verified_at"] = datetime.now(timezone.utc).isoformat() + target["verified_at"] = now_cn_aware().isoformat() else: target["verified_by"] = None target["verified_at"] = None @@ -3574,7 +3575,7 @@ async def complete_release_verification( incomplete = [item for item in items if str(item.get("status") or "").lower() != "done"] if incomplete: raise HTTPException(status_code=409, detail="仍有未完成验证项，请先逐项完成") - now_str = datetime.now(timezone.utc).isoformat() + now_str = now_cn_aware().isoformat() content["verification_items"] = items content["verification_done"] = True content["verification_progress"] = {"done": len(items), "total": len(items)} @@ -3912,7 +3913,7 @@ async def archive_review( review_node.status = "archived" if body.archive else "draft" content = _as_dict(review_node.content) content["archive_note"] = body.note - content["archived_at"] = datetime.now(timezone.utc).isoformat() if body.archive else None + content["archived_at"] = now_cn_aware().isoformat() if body.archive else None review_node.content = content await log_operation( db, diff --git a/backend/app/api/users.py b/backend/app/api/users.py index 1d272079..b3053d8d 100644 --- a/backend/app/api/users.py +++ b/backend/app/api/users.py @@ -43,6 +43,7 @@ from ..services.ai_pmo_kernel_service import AIPMOKernelService from ..services.operation_log_service import log_operation from ..services import skill_service +from ..utils.timezone import now_cn_naive router = APIRouter(prefix="/users", tags=["users"]) COMPLETED_TASK_STATUSES = {"done", "completed", "closed", "resolved", "archived"} @@ -650,7 +651,7 @@ async def reset_password( user.password_hash = AuthService.hash_password(body.new_password) except ValueError as exc: raise HTTPException(status_code=400, detail=str(exc)) from exc - user.updated_at = datetime.utcnow() + user.updated_at = now_cn_naive() await log_operation( db, tenant_id=tenant_id, @@ -696,7 +697,7 @@ async def patch_user_status( Task.assignee_id == user_id, ) ) - .values(assignee_id=None, updated_at=datetime.utcnow()) + .values(assignee_id=None, updated_at=now_cn_naive()) ) workload_result = await db.execute( select(UserWorkload).where( @@ -986,7 +987,7 @@ async def update_user_workload( raise HTTPException(status_code=404, detail="User not found") user.workload = max(0, min(200, int(workload))) - user.updated_at = datetime.utcnow() + user.updated_at = now_cn_naive() await db.commit() await db.refresh(user) return _serialize_user(user) diff --git a/backend/app/api/webhooks.py b/backend/app/api/webhooks.py index fb101410..e42eb5a6 100644 --- a/backend/app/api/webhooks.py +++ b/backend/app/api/webhooks.py @@ -13,6 +13,7 @@ from ..models import Signal from ..services.llm import llm_service from ..services.plane import plane_service +from ..utils.timezone import now_cn_aware router = APIRouter(prefix="/webhooks", tags=["webhooks"]) @@ -94,7 +95,7 @@ async def handle_git_webhook( t["status"] = "Done" t["audit_feedback"] = audit_result.get("reason") t["completed_at"] = ( - datetime.now(timezone.utc).isoformat() + now_cn_aware().isoformat() ) target_signal.tasks_generated = json.dumps(tasks) await db.commit() diff --git a/backend/app/dispatcher.py b/backend/app/dispatcher.py index 9515db88..b98a1041 100644 --- a/backend/app/dispatcher.py +++ b/backend/app/dispatcher.py @@ -12,6 +12,7 @@ import re import logging from pydantic import BaseModel, Field +from .utils.timezone import now_cn_aware logger = logging.getLogger(__name__) @@ -356,7 +357,7 @@ class RoutingResult(BaseModel): confidence: float details: Dict[str, Any] signal_id: Optional[str] = None - timestamp: float = Field(default_factory=lambda: datetime.utcnow().timestamp()) + timestamp: float = Field(default_factory=lambda: now_cn_aware().timestamp()) # 全局路由器实例 diff --git a/backend/app/events.py b/backend/app/events.py index fdf55e86..fc7afdbb 100644 --- a/backend/app/events.py +++ b/backend/app/events.py @@ -12,6 +12,7 @@ import asyncio import logging from contextlib import asynccontextmanager +from .utils.timezone import now_cn_aware logger = logging.getLogger(__name__) @@ -74,7 +75,7 @@ class Event: type: EventType data: Dict[str, Any] - timestamp: float = field(default_factory=lambda: datetime.utcnow().timestamp()) + timestamp: float = field(default_factory=lambda: now_cn_aware().timestamp()) source: Optional[str] = None correlation_id: Optional[str] = None metadata: Dict[str, Any] = field(default_factory=dict) diff --git a/backend/app/main.py b/backend/app/main.py index bb3fb20a..0dbeee73 100644 --- a/backend/app/main.py +++ b/backend/app/main.py @@ -29,6 +29,7 @@ from .services.feishu_long_connection import init_feishu_long_connection_manager from .services.feishu_integration import dispatch_pending_feishu_replies from .services.context_cache import register_context_cache_invalidation_hooks +from .utils.timezone import now_cn_aware configure_logging(settings.debug) logger = get_logger(__name__) register_context_cache_invalidation_hooks() @@ -422,7 +423,7 @@ async def _get_recent_active_projects( limit_per_tenant: int, max_projects: int, ) -> list[tuple[int, int]]: - since = (datetime.now(timezone.utc) - timedelta(days=max(1, int(days)))).replace(tzinfo=None) + since = (now_cn_aware() - timedelta(days=max(1, int(days)))).replace(tzinfo=None) per_tenant_limit = max(1, int(limit_per_tenant)) total_limit = max(per_tenant_limit, int(max_projects)) diff --git a/backend/app/services/adaptive_rag_service.py b/backend/app/services/adaptive_rag_service.py index 8c5a413f..1db7df50 100644 --- a/backend/app/services/adaptive_rag_service.py +++ b/backend/app/services/adaptive_rag_service.py @@ -54,6 +54,7 @@ from .signal_source_orchestration_service import SignalSourceOrchestrationServic from .symbol_index_service import SymbolIndexService +from ..utils.timezone import CN_TZ, now_cn_aware, now_cn_naive def _env_int(name: str, default: int, *, minimum: int, maximum: int) -> int: raw = str(os.getenv(name, "") or "").strip() if not raw: @@ -1408,7 +1409,7 @@ class AdaptiveRAGService: "original_question": str(question or "")[:4000], "normalized_question": summary, "write_mode": "ack_first", - "created_at": datetime.now(timezone.utc).isoformat(), + "created_at": now_cn_aware().isoformat(), } signal = Signal( source="PMFlow", @@ -2366,7 +2367,7 @@ class AdaptiveRAGService: history = session_state.rewrite_history if isinstance(session_state.rewrite_history, list) else [] history.append( { - "at": datetime.now(timezone.utc).isoformat(), + "at": now_cn_aware().isoformat(), "question": question[:500], "effective_question": effective_question[:500], "query_type": str(query_type or QUERY_TYPE_UNKNOWN), @@ -3116,7 +3117,7 @@ class AdaptiveRAGService: symbol_snapshot=symbol_snapshot, kg_overview=kg_overview, ), - "generated_at": datetime.now(timezone.utc).isoformat(), + "generated_at": now_cn_aware().isoformat(), } async def entity_resolution( @@ -3516,7 +3517,7 @@ class AdaptiveRAGService: "python_symbols": 0, "typescript_symbols": 0, "top_modules": [], - "generated_at": datetime.now(timezone.utc).isoformat(), + "generated_at": now_cn_aware().isoformat(), } candidates = [ @@ -3568,7 +3569,7 @@ class AdaptiveRAGService: "python_symbols": int(python_symbols), "typescript_symbols": int(ts_symbols), "top_modules": top_modules, - "generated_at": datetime.now(timezone.utc).isoformat(), + "generated_at": now_cn_aware().isoformat(), } @staticmethod @@ -4730,7 +4731,7 @@ class AdaptiveRAGService: file_url=(file_url or "")[:1024] or None, uploaded_by=uploaded_by, status="processing", - last_fetched_at=datetime.now(timezone.utc) if source_type == "url" else None, + last_fetched_at=now_cn_naive() if source_type == "url" else None, ) self.db.add(doc) await self.db.commit() @@ -4814,7 +4815,7 @@ class AdaptiveRAGService: doc.status = "complete" doc.error_message = None if source_type == "url": - doc.last_fetched_at = datetime.now(timezone.utc) + doc.last_fetched_at = now_cn_naive() await self.db.commit() except Exception as exc: doc.status = "failed" @@ -4906,7 +4907,7 @@ class AdaptiveRAGService: "sender": sender, "chat_id": chat_id, "message_id": message_id, - "captured_at": (captured_at or datetime.now(timezone.utc)).isoformat(), + "captured_at": (captured_at or now_cn_aware()).isoformat(), }, ) ) @@ -5071,7 +5072,7 @@ class AdaptiveRAGService: "project_id": int(project_id) if project_id is not None else None, "target_url": target_url, "title": title[:255], - "updated_at": datetime.now(timezone.utc).isoformat(), + "updated_at": now_cn_aware().isoformat(), }, ) ) @@ -5143,7 +5144,7 @@ class AdaptiveRAGService: "risk_level": risk, "target_url": None, "title": title, - "updated_at": datetime.now(timezone.utc).isoformat(), + "updated_at": now_cn_aware().isoformat(), } return record_id, title, content, metadata @@ -5268,7 +5269,7 @@ class AdaptiveRAGService: return None if value.tzinfo is None: return value - return value.astimezone(timezone.utc).replace(tzinfo=None) + return value.astimezone(CN_TZ).replace(tzinfo=None) @staticmethod def _normalize_session_title_raw(text: str, *, max_len: int = 30) -> str: @@ -5305,7 +5306,7 @@ class AdaptiveRAGService: if asked_by is None: return (None, 1) - now_utc = datetime.now(timezone.utc).replace(tzinfo=None) + now_utc = now_cn_naive() chosen: Optional[QASession] = None preferred_id = int(preferred_session_id) if preferred_session_id else 0 if preferred_id > 0: diff --git a/backend/app/services/ai_orchestration_run_service.py b/backend/app/services/ai_orchestration_run_service.py index 946311c6..6fac0b8f 100644 --- a/backend/app/services/ai_orchestration_run_service.py +++ b/backend/app/services/ai_orchestration_run_service.py @@ -9,9 +9,9 @@ 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 now_cn_naive() diff --git a/backend/app/services/ai_pmo_cadence_service.py b/backend/app/services/ai_pmo_cadence_service.py index dbfb796d..27632c5b 100644 --- a/backend/app/services/ai_pmo_cadence_service.py +++ b/backend/app/services/ai_pmo_cadence_service.py @@ -9,9 +9,9 @@ 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 now_cn_naive() diff --git a/backend/app/services/ai_pmo_kernel_service.py b/backend/app/services/ai_pmo_kernel_service.py index 043feb8d..90d1e534 100644 --- a/backend/app/services/ai_pmo_kernel_service.py +++ b/backend/app/services/ai_pmo_kernel_service.py @@ -72,8 +72,8 @@ 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 +from ..utils.timezone import now_cn_naive logger = get_logger(__name__) INTENT_QUERY = "query.read_only" @@ -103,6 +103,18 @@ RECEIPT_PHASE_ACTION = "action_applied" RECEIPT_PHASE_PENDING = "pending_manual" RECEIPT_PHASE_ERROR = "error" + +def _looks_like_upstream_timeout_message(text: str) -> bool: + lowered = str(text or "").strip().lower() + if not lowered: + return False + timeout_signals = ( + "request timed out before a response was generated", + "agents.defaults.timeoutseconds", + "timeoutseconds", + ) + return any(signal in lowered for signal in timeout_signals) + QUESTION_HINTS = ( "?", "？", @@ -1035,6 +1047,19 @@ class AIPMOKernelService: project_hint_id: Optional[int] = None, ) -> dict[str, Any]: started = time.perf_counter() + perf_work_item_id: Optional[int] = None + perf_intent_name = "" + + def _log_perf(stage_name: str, stage_started_at: float) -> None: + elapsed_seconds = time.perf_counter() - stage_started_at + perf_line = ( + f"[PERF] {stage_name} took {elapsed_seconds:.2f}s " + f"work_item={perf_work_item_id if perf_work_item_id is not None else 'N/A'} " + f"intent={perf_intent_name or 'N/A'}" + ) + logger.info(perf_line) + print(perf_line, flush=True) + clean_text = self._normalize_text(message_text) if not clean_text: return {"handled": True, "status": "ignored", "classification": "empty", "direct_reply_required": False} @@ -1096,6 +1121,7 @@ class AIPMOKernelService: source_message_id=synthetic_message_id, owner_user_id=_safe_int(getattr(operator, "id", None)), ) + perf_work_item_id = int(work_item.id) session = await self._get_or_create_session(tenant_id=self.tenant_id, sender=sender) try: @@ -1106,6 +1132,7 @@ class AIPMOKernelService: query_tables=world_model_query_tables, session=session, ) + stage_started_at = time.perf_counter() world_model = await self.build_world_model( tenant_id=self.tenant_id, project_id=_safe_int(getattr(project, "id", None)), @@ -1113,13 +1140,17 @@ class AIPMOKernelService: project=project, pre_queried_tables=world_model_query_tables, ) + _log_perf("_build_world_model", stage_started_at) self._schedule_portfolio_snapshot_persist(world_model=world_model) work_item.state = STATE_MODELED + stage_started_at = time.perf_counter() intent, intent_context_source = await self._resolve_intent( message_text=clean_text, session=session, world_model=world_model, ) + perf_intent_name = intent.name + _log_perf("_resolve_intent", stage_started_at) if intent_context_source: world_model["context_source"] = intent_context_source work_item.context_snapshot = world_model @@ -1162,11 +1193,13 @@ class AIPMOKernelService: work_item.intent = intent.name work_item.project_id = intent.project_id + stage_started_at = time.perf_counter() decision_payload = self.decision_engine.decide( intent=intent, world_model=world_model, message_text=clean_text, ) + _log_perf("DecisionEngine.decide", stage_started_at) 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() @@ -1291,7 +1324,8 @@ class AIPMOKernelService: last_message_text=clean_text[:200], ) await self.db.commit() - return self._build_capture_result( + stage_started_at = time.perf_counter() + capture_result = self._build_capture_result( work_item=work_item, intent=intent, execution_result=work_item.execution_result or {}, @@ -1300,11 +1334,14 @@ class AIPMOKernelService: status="captured", classification="command", ) + _log_perf("_build_capture_result", stage_started_at) + return capture_result work_item.state = STATE_PLANNED work_item.action_bundle = decision_payload["execution_plan"] work_item.state = STATE_EXECUTING if intent.name in {INTENT_FOLLOWUP_CONFIRM, INTENT_FOLLOWUP_CONTINUE}: + stage_started_at = time.perf_counter() execution_result, verification = await self._execute_followup_intent( work_item=work_item, intent=intent, @@ -1317,7 +1354,9 @@ class AIPMOKernelService: source_sender_open_id=source_sender_open_id, world_model=world_model, ) + _log_perf("TypedExecutor.execute", stage_started_at) else: + stage_started_at = time.perf_counter() execution_result = await self.executor.execute( intent=intent, message_text=clean_text, @@ -1326,6 +1365,7 @@ class AIPMOKernelService: source_system=source_system, source_message_id=synthetic_message_id, ) + _log_perf("TypedExecutor.execute", stage_started_at) 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) @@ -1399,7 +1439,8 @@ class AIPMOKernelService: last_message_text=clean_text[:200], ) await self.db.commit() - return self._build_capture_result( + stage_started_at = time.perf_counter() + capture_result = self._build_capture_result( work_item=work_item, intent=intent, execution_result=execution_result, @@ -1412,6 +1453,8 @@ class AIPMOKernelService: else ("signal" if intent.name == INTENT_SIGNAL_CREATE else "command") ), ) + _log_perf("_build_capture_result", stage_started_at) + return capture_result except Exception as exc: await self.db.rollback() logger.warning( @@ -3160,6 +3203,17 @@ class AIPMOKernelService: classification: str, ) -> dict[str, Any]: linked_project_ids = [int(work_item.project_id)] if work_item.project_id is not None else [] + direct_reply_text = str(render.get("text") or "").strip() + if _looks_like_upstream_timeout_message(direct_reply_text): + logger.warning( + "ai_pmo_upstream_timeout_reply_detected", + tenant_id=int(self.tenant_id), + work_item_id=int(work_item.id), + intent=str(intent.name or ""), + source_message_id=str(work_item.source_message_id or "")[:128], + observed_at=now_cn_naive().isoformat(), + reply_preview=direct_reply_text[:260], + ) result = { "handled": True, "status": status, @@ -3179,7 +3233,7 @@ class AIPMOKernelService: ], "intent": intent.name, "direct_reply_required": bool(direct_reply_required), - "direct_reply_text": str(render.get("text") or "").strip(), + "direct_reply_text": direct_reply_text, "reply_template_key": str(render.get("template_key") or "").strip(), "work_item_id": int(work_item.id), "decision_level": work_item.decision_level, diff --git a/backend/app/services/ai_service.py b/backend/app/services/ai_service.py index 4b584a6b..246f42a2 100644 --- a/backend/app/services/ai_service.py +++ b/backend/app/services/ai_service.py @@ -18,6 +18,7 @@ from typing import Any, AsyncGenerator, Optional, Union from sqlalchemy import and_, select # LangChain & OpenAI +from ..utils.timezone import now_cn_naive try: from langchain_openai import ChatOpenAI from langchain_core.messages import HumanMessage, AIMessage, SystemMessage @@ -524,7 +525,7 @@ class AIService: if b: b.quality_score = min(100, max(0, (b.quality_score or 0) + quality_score)) b.efficiency_score = min(100, max(0, (b.efficiency_score or 0) + efficiency_score)) - b.last_used_at = datetime.utcnow() + b.last_used_at = now_cn_naive() await db.commit() updates.append({ "user_id": str(uid), diff --git a/backend/app/services/auditor.py b/backend/app/services/auditor.py index 09203e8e..08c3ba3e 100644 --- a/backend/app/services/auditor.py +++ b/backend/app/services/auditor.py @@ -12,6 +12,7 @@ from pydantic import BaseModel, Field from datetime import datetime from enum import Enum +from ..utils.timezone import now_cn_aware logger = logging.getLogger(__name__) @@ -45,7 +46,7 @@ class AuditResult(BaseModel): issues: List[AuditIssue] = Field(default_factory=list) summary: str = "" details: Dict[str, Any] = Field(default_factory=dict) - timestamp: float = Field(default_factory=lambda: datetime.utcnow().timestamp()) + timestamp: float = Field(default_factory=lambda: now_cn_aware().timestamp()) class SecurityIssue(BaseModel): diff --git a/backend/app/services/auth.py b/backend/app/services/auth.py index fe4a0492..11798729 100644 --- a/backend/app/services/auth.py +++ b/backend/app/services/auth.py @@ -11,6 +11,7 @@ import hmac import os # Try to import passlib, fallback to simple hashing +from ..utils.timezone import now_cn_aware, now_cn_naive try: from passlib.context import CryptContext from jose import JWTError, jwt @@ -106,15 +107,15 @@ class AuthService: import json token_data = { **data, - "exp": (datetime.utcnow() + (expires_delta or timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS))).timestamp() + "exp": (now_cn_naive() + (expires_delta or timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS))).timestamp() } return base64.b64encode(json.dumps(token_data).encode()).decode() to_encode = data.copy() if expires_delta: - expire = datetime.utcnow() + expires_delta + expire = now_cn_naive() + expires_delta else: - expire = datetime.utcnow() + timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS) + expire = now_cn_naive() + timedelta(hours=ACCESS_TOKEN_EXPIRE_HOURS) to_encode.update({"exp": expire}) return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM) @@ -131,7 +132,7 @@ class AuthService: import json try: token_data = json.loads(base64.b64decode(token).decode()) - if token_data.get("exp", 0) > datetime.utcnow().timestamp(): + if token_data.get("exp", 0) > now_cn_aware().timestamp(): return token_data.get("sub") return None except: diff --git a/backend/app/services/auto_merger.py b/backend/app/services/auto_merger.py index df92096a..df610894 100644 --- a/backend/app/services/auto_merger.py +++ b/backend/app/services/auto_merger.py @@ -16,6 +16,7 @@ from datetime import datetime, timedelta from enum import Enum from ..events import Event, EventType +from ..utils.timezone import now_cn_aware, now_cn_naive logger = logging.getLogger(__name__) @@ -138,12 +139,12 @@ class AutoMergeService: # 4. 创建人工确认请求 if self.HUMAN_CONFIRMATION_REQUIRED: confirmation_token = self._generate_confirmation_token() - expires_at = datetime.utcnow().timestamp() + self.CONFIRMATION_TIMEOUT + expires_at = now_cn_aware().timestamp() + self.CONFIRMATION_TIMEOUT confirmation = ConfirmationRequest( token=confirmation_token, merge_request=request, - created_at=datetime.utcnow().timestamp(), + created_at=now_cn_aware().timestamp(), expires_at=expires_at, ) self._pending_confirmations[confirmation_token] = confirmation @@ -179,7 +180,7 @@ class AutoMergeService: ) # 检查是否过期 - if datetime.utcnow().timestamp() > confirmation.expires_at: + if now_cn_aware().timestamp() > confirmation.expires_at: del self._pending_confirmations[confirmation_token] return MergeResponse( status=MergeStatus.REJECTED, @@ -401,7 +402,7 @@ class AutoMergeService: if self._last_failure_time is None: return False - elapsed = (datetime.utcnow() - self._last_failure_time).total_seconds() + elapsed = (now_cn_naive() - self._last_failure_time).total_seconds() return elapsed < self.FAILURE_COOLDOWN_SECONDS def _get_cooldown_remaining(self) -> float: @@ -409,7 +410,7 @@ class AutoMergeService: if self._last_failure_time is None: return 0.0 - elapsed = (datetime.utcnow() - self._last_failure_time).total_seconds() + elapsed = (now_cn_naive() - self._last_failure_time).total_seconds() return max(0, self.FAILURE_COOLDOWN_SECONDS - elapsed) def _all_tests_passed(self, test_results: Dict[str, Any]) -> bool: @@ -448,7 +449,7 @@ class AutoMergeService: def _record_failure(self): """记录失败""" self._failure_count += 1 - self._last_failure_time = datetime.utcnow() + self._last_failure_time = now_cn_naive() # 检查是否需要进入熔断期 if self._failure_count >= self.MAX_CONSECUTIVE_FAILURES: @@ -478,7 +479,7 @@ class AutoMergeService: async def get_pending_confirmations(self) -> List[ConfirmationRequest]: """获取待确认的请求""" # 清理过期的确认请求 - now = datetime.utcnow().timestamp() + now = now_cn_aware().timestamp() expired = [ token for token, conf in self._pending_confirmations.items() if now > conf.expires_at diff --git a/backend/app/services/blocker_detection.py b/backend/app/services/blocker_detection.py index 19eda2ca..3431b85a 100644 --- a/backend/app/services/blocker_detection.py +++ b/backend/app/services/blocker_detection.py @@ -30,6 +30,7 @@ from .llm import LLMService # 阻塞模式正则表达式 +from ..utils.timezone import now_cn_naive BLOCKER_PATTERNS = [ r'等待.*接口', r'等待.*API', @@ -222,7 +223,7 @@ class BlockerDetectionService: raise ValueError(f"阻塞 {blocker_id} 不存在") blocker.status = 'resolved' - blocker.resolved_at = datetime.now(timezone.utc) + blocker.resolved_at = now_cn_naive() blocker.description += f"\n\n解决方案：{resolution_note}" await self.db.commit() diff --git a/backend/app/services/budget_enforcement_service.py b/backend/app/services/budget_enforcement_service.py index f6bf65ce..67e57ac4 100644 --- a/backend/app/services/budget_enforcement_service.py +++ b/backend/app/services/budget_enforcement_service.py @@ -18,6 +18,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from ..models import Budget, BudgetAlert +from ..utils.timezone import now_cn_naive class BudgetEnforcementService: """预算强制执行服务""" @@ -39,7 +40,7 @@ class BudgetEnforcementService: "budget_checked": False, } - now = datetime.utcnow() + now = now_cn_naive() await self._ensure_current_period(budget, now) current_usage = float(budget.current_usage_usd or 0.0) @@ -122,7 +123,7 @@ class BudgetEnforcementService: if not budget: return {"updated": False, "reason": "no_budget_configured"} - now = datetime.utcnow() + now = now_cn_naive() await self._ensure_current_period(budget, now) cost = max(float(actual_cost_usd or 0.0), 0.0) @@ -255,7 +256,7 @@ class BudgetEnforcementService: if not budget: raise ValueError(f"预算 {budget_id} 不存在") - now = datetime.utcnow() + now = now_cn_naive() expires_at = now + timedelta(hours=max(duration_hours, 1)) history = dict(budget.override_history or {}) records = list(history.get("records", [])) diff --git a/backend/app/services/codegen.py b/backend/app/services/codegen.py index eea6bad0..bbf6641c 100644 --- a/backend/app/services/codegen.py +++ b/backend/app/services/codegen.py @@ -15,6 +15,7 @@ from pydantic import BaseModel, Field from datetime import datetime from pathlib import Path +from ..utils.timezone import now_cn_aware logger = logging.getLogger(__name__) @@ -91,7 +92,7 @@ class CodeDraftService: """ 生成代码草案（激进策略） """ - start_time = datetime.utcnow().timestamp() + start_time = now_cn_aware().timestamp() try: # 1. 读取目标文件 @@ -130,7 +131,7 @@ class CodeDraftService: # 10. 收集警告 warnings = self._collect_warnings(generated_files, file_contexts) - generation_time = datetime.utcnow().timestamp() - start_time + generation_time = now_cn_aware().timestamp() - start_time return CodeDraftResponse( success=True, @@ -158,7 +159,7 @@ class CodeDraftService: confidence=0.0, warnings=[f"生成过程出错: {str(e)}"], review_points=[], - generation_time=datetime.utcnow().timestamp() - start_time, + generation_time=now_cn_aware().timestamp() - start_time, ) async def _read_target_files( diff --git a/backend/app/services/context_cache.py b/backend/app/services/context_cache.py index ed720ec1..0fde86c8 100644 --- a/backend/app/services/context_cache.py +++ b/backend/app/services/context_cache.py @@ -15,6 +15,7 @@ from sqlalchemy.orm import Session from ..models import Project, ProjectSignalLink, Signal, Task, User, UserProject, UserWorkload +from ..utils.timezone import now_cn_aware try: import redis.asyncio as redis_async from redis import Redis as RedisSync @@ -178,7 +179,7 @@ class ContextCacheService: "projects": projects.get("items", []), "team_members": team, "pending_signals": pending, - "snapshot_at": datetime.now(timezone.utc).isoformat(), + "snapshot_at": now_cn_aware().isoformat(), } @staticmethod @@ -558,7 +559,7 @@ class ContextCacheService: "backend_root": str(backend_root), "app_python_files": app_python_files, "feature_catalog_source": str(feature_catalog.get("source") or "unknown"), - "scanned_at": datetime.now(timezone.utc).isoformat(), + "scanned_at": now_cn_aware().isoformat(), }, } @@ -654,7 +655,7 @@ class ContextCacheService: payload = { "items": items, "count": len(items), - "snapshot_at": datetime.now(timezone.utc).isoformat(), + "snapshot_at": now_cn_aware().isoformat(), } await self._set_json(key, payload, PROJECT_LIST_TTL_SECONDS) return payload @@ -759,7 +760,7 @@ class ContextCacheService: "p0_count": int(p0_count or 0), "p1_count": int(p1_count or 0), "p2_count": int(p2_count or 0), - "snapshot_at": datetime.now(timezone.utc).isoformat(), + "snapshot_at": now_cn_aware().isoformat(), } await self._set_json(key, payload, SIGNAL_PENDING_TTL_SECONDS) return payload @@ -892,7 +893,7 @@ class ContextCacheService: ], "pending_signals": pending_signals, "workload_ratio": workload_ratio, - "snapshot_at": datetime.now(timezone.utc).isoformat(), + "snapshot_at": now_cn_aware().isoformat(), } async def get_session_messages(self, *, tenant_id: int, session_id: Optional[int]) -> list[dict[str, str]]: diff --git a/backend/app/services/daily_briefing.py b/backend/app/services/daily_briefing.py index aa80e2d3..4fb1aa07 100644 --- a/backend/app/services/daily_briefing.py +++ b/backend/app/services/daily_briefing.py @@ -28,6 +28,7 @@ from ..models import ( from .llm import LLMService +from ..utils.timezone import now_cn_aware def _parse_involved_user_ids(raw_value: Any) -> List[int]: """兼容 involved_user_ids 的 list / json-string / 单值字符串。""" if raw_value is None: @@ -86,7 +87,7 @@ class DailyBriefingService: raise ValueError(f"User {user_id} not found") # 计算时间范围 - now = datetime.now(timezone.utc) + now = now_cn_aware() yesterday = now - timedelta(days=1) # 根据角色收集相关信息 @@ -456,7 +457,7 @@ class DailyBriefingService: return { 'project_id': project_id, 'project_name': project.title, - 'generated_at': datetime.now(timezone.utc).isoformat(), + 'generated_at': now_cn_aware().isoformat(), 'task_summary': { 'total': len(tasks), 'by_status': status_counts, diff --git a/backend/app/services/domain.py b/backend/app/services/domain.py index 02f324de..6ceef0ff 100644 --- a/backend/app/services/domain.py +++ b/backend/app/services/domain.py @@ -13,6 +13,7 @@ from datetime import datetime from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession +from ..utils.timezone import now_cn_naive logger = logging.getLogger(__name__) @@ -216,7 +217,7 @@ class ProjectDomainService: await self._clear_default_domains() domain.is_default = request.is_default - domain.updated_at = datetime.utcnow() + domain.updated_at = now_cn_naive() await self.db.commit() await self.db.refresh(domain) @@ -308,7 +309,7 @@ class ProjectDomainService: raise ValueError("Domain is not active") # 更新最后同步时间 - domain.last_synced_at = datetime.utcnow() + domain.last_synced_at = now_cn_naive() await self.db.commit() # 设置活动域 diff --git a/backend/app/services/event_workflow.py b/backend/app/services/event_workflow.py index 01596669..74894891 100644 --- a/backend/app/services/event_workflow.py +++ b/backend/app/services/event_workflow.py @@ -30,6 +30,7 @@ from .llm import LLMService from ..events import EventType, Event, get_event_bus +from ..utils.timezone import now_cn_aware class WorkflowEventType: """工作流事件类型""" HIGH_IMPACT_SIGNAL = "high_impact_signal" @@ -92,7 +93,7 @@ class EventDrivenWorkflowEngine: { 'workflow_type': event_type, 'result': result, - 'timestamp': datetime.now(timezone.utc).isoformat() + 'timestamp': now_cn_aware().isoformat() } ) diff --git a/backend/app/services/experiment_sync_service.py b/backend/app/services/experiment_sync_service.py index 0e2a13f0..4e1e3ef0 100644 --- a/backend/app/services/experiment_sync_service.py +++ b/backend/app/services/experiment_sync_service.py @@ -9,6 +9,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from ..models import Experiment, ExperimentTruthNodeSync, TruthNode +from ..utils.timezone import now_cn_aware, now_cn_naive class ExperimentSyncService: """实验结果 -> 宪章同步，支持自动应用、人工审核与回滚。""" @@ -147,7 +148,7 @@ class ExperimentSyncService: return {"sync_id": record.id, "apply_status": record.apply_status} record.reviewed_by = reviewer_id - record.reviewed_at = datetime.now(timezone.utc) + record.reviewed_at = now_cn_naive() record.review_notes = review_notes if action == "approve": @@ -192,7 +193,7 @@ class ExperimentSyncService: record.apply_status = "requires_review" record.reviewed_by = reviewer_id - record.reviewed_at = datetime.now(timezone.utc) + record.reviewed_at = now_cn_naive() record.review_notes = rollback_notes or "manual rollback" await self.db.commit() @@ -293,7 +294,7 @@ class ExperimentSyncService: "sync_type": record.sync_type, "confidence_score": record.confidence_score, "experiment_summary": (record.sync_data or {}).get("experiment_summary", {}), - "applied_at": datetime.now(timezone.utc).isoformat(), + "applied_at": now_cn_aware().isoformat(), } content["experiment_sync"] = sync_section truth_node.content = content diff --git a/backend/app/services/feishu_integration.py b/backend/app/services/feishu_integration.py index f63c834a..e7f9952e 100644 --- a/backend/app/services/feishu_integration.py +++ b/backend/app/services/feishu_integration.py @@ -41,10 +41,12 @@ from .notification_service import NotificationService from .signal_capture import ProactiveSignalCaptureService from .llm import LLMService +from ..utils.timezone import now_cn_naive CN_TZ = timezone(timedelta(hours=8)) WEEKDAY_CN = ["周一", "周二", "周三", "周四", "周五", "周六", "周日"] OUTBOX_MAX_ATTEMPTS = 3 OUTBOX_STALE_SENDING_SECONDS = 120 +_OUTBOX_PERF_T0_BY_ID: dict[int, float] = {} class FeishuIntegrationService: @@ -213,6 +215,15 @@ class FeishuIntegrationService: chat = event.get('chat', {}) if isinstance(event.get('chat', {}), dict) else {} sender = event.get('sender', {}) if isinstance(event.get('sender', {}), dict) else {} message_id = (message.get("message_id") or "").strip() or "unknown" + perf_t0_raw = event.get("_perf_t0") + perf_t0: Optional[float] = None + if isinstance(perf_t0_raw, (int, float)): + perf_t0 = float(perf_t0_raw) + else: + try: + perf_t0 = float(str(perf_t0_raw).strip()) + except Exception: + perf_t0 = None self._emit_state_transition( trace_id=state_trace_id, state="RECEIVED", @@ -254,6 +265,7 @@ class FeishuIntegrationService: text="我已收到你的消息，但目前无法直接解析纯图片内容。请补充 1-2 句文字需求，我会立即给出分派建议。", trace_id=state_trace_id, source_message_id=message_id, + perf_t0=perf_t0, ) self._emit_state_transition( trace_id=state_trace_id, @@ -318,6 +330,7 @@ class FeishuIntegrationService: text=self._build_error_reply_text(), trace_id=state_trace_id, source_message_id=message_id, + perf_t0=perf_t0, ) self._emit_state_transition( trace_id=state_trace_id, @@ -401,6 +414,7 @@ class FeishuIntegrationService: text=deterministic_reply, trace_id=state_trace_id, source_message_id=message_id, + perf_t0=perf_t0, ) reply_route = "execute_template" elif scene == "qa": @@ -417,6 +431,7 @@ class FeishuIntegrationService: text=str(result.get("direct_reply_text") or "已收到，正在处理。"), trace_id=state_trace_id, source_message_id=message_id, + perf_t0=perf_t0, ) reply_route = "qa_direct" elif ( @@ -434,6 +449,7 @@ class FeishuIntegrationService: capture_result=result, trace_id=state_trace_id, source_message_id=message_id, + perf_t0=perf_t0, ) reply_route = "qa_llm" else: @@ -451,6 +467,7 @@ class FeishuIntegrationService: text=self._build_error_reply_text(), trace_id=state_trace_id, source_message_id=message_id, + perf_t0=perf_t0, ) reply_route = "error_template" @@ -589,6 +606,7 @@ class FeishuIntegrationService: capture_result: Dict[str, Any], trace_id: Optional[str] = None, source_message_id: Optional[str] = None, + perf_t0: Optional[float] = None, ) -> None: """被 @ 时回复一条有人味的执行反馈。""" deterministic_time_reply = self._build_time_truth_reply(message_content) @@ -600,6 +618,7 @@ class FeishuIntegrationService: text=deterministic_time_reply, trace_id=trace_id, source_message_id=source_message_id, + perf_t0=perf_t0, ) return @@ -619,6 +638,7 @@ class FeishuIntegrationService: text=execution_receipt, trace_id=trace_id, source_message_id=source_message_id, + perf_t0=perf_t0, ) return @@ -631,6 +651,7 @@ class FeishuIntegrationService: text=butler_reply, trace_id=trace_id, source_message_id=source_message_id, + perf_t0=perf_t0, ) return @@ -683,6 +704,7 @@ class FeishuIntegrationService: text=reply_text, trace_id=trace_id, source_message_id=source_message_id, + perf_t0=perf_t0, ) async def _safe_reply_text( @@ -694,6 +716,7 @@ class FeishuIntegrationService: mention_name: Optional[str] = None, trace_id: Optional[str] = None, source_message_id: Optional[str] = None, + perf_t0: Optional[float] = None, ) -> None: try: plain_text = self._normalize_outgoing_reply_text(text) @@ -702,12 +725,27 @@ class FeishuIntegrationService: display_name=mention_name, ) final_text = f"{mention_prefix}{plain_text}".strip() - await self._enqueue_reply_outbox( + outbox_id = await self._enqueue_reply_outbox( chat_id=chat_id, message_text=final_text, trace_id=trace_id, source_message_id=source_message_id, ) + perf_t1 = time.perf_counter() + if outbox_id > 0 and perf_t0 is not None: + _OUTBOX_PERF_T0_BY_ID[int(outbox_id)] = float(perf_t0) + if perf_t0 is not None: + print( + f"[PERF] outbox_written ts={now_cn_naive().isoformat()} T1={perf_t1:.6f} " + f"delta={perf_t1 - perf_t0:.2f}s " + f"outbox_id={outbox_id}" + ) + else: + print( + f"[PERF] outbox_written ts={now_cn_naive().isoformat()} T1={perf_t1:.6f} " + f"delta=N/A " + f"outbox_id={outbox_id}" + ) except Exception as exc: print(f"[Feishu] enqueue reply outbox failed: chat_id={chat_id}, error={exc}") # 统一由 drain_loop 消费 outbox，禁止业务路径直发，避免双发送。 @@ -783,7 +821,7 @@ class FeishuIntegrationService: fetch_limit = max(1, min(200, int(limit))) dispatched = 0 for _ in range(fetch_limit): - claimed = await cls._claim_next_outbox_entry(now=datetime.utcnow()) + claimed = await cls._claim_next_outbox_entry(now=now_cn_naive()) if claimed is None: break await cls._dispatch_claimed_outbox_entry(claimed=claimed) @@ -883,10 +921,11 @@ class FeishuIntegrationService: ).scalar_one_or_none() if row is not None and str(row.status or "").lower() == "sending": row.status = "sent" - row.sent_at = datetime.utcnow() + row.sent_at = now_cn_naive() row.next_retry_at = None row.last_error = f"dedupe_skip_sent:{int(sent_duplicate)}" await db_skip.commit() + _OUTBOX_PERF_T0_BY_ID.pop(outbox_id, None) return send_error: Optional[str] = None @@ -914,18 +953,33 @@ class FeishuIntegrationService: attempts = int(row.attempts or claimed.get("attempts") or 0) if send_error is None: row.status = "sent" - row.sent_at = datetime.utcnow() + row.sent_at = now_cn_naive() row.next_retry_at = None row.last_error = None await db_update.commit() + perf_t2 = time.perf_counter() + perf_t0 = _OUTBOX_PERF_T0_BY_ID.pop(outbox_id, None) + if perf_t0 is not None: + print( + f"[PERF] reply_dispatched ts={now_cn_naive().isoformat()} T2={perf_t2:.6f} " + f"delta={perf_t2 - perf_t0:.2f}s " + f"outbox_id={outbox_id}" + ) + else: + print( + f"[PERF] reply_dispatched ts={now_cn_naive().isoformat()} T2={perf_t2:.6f} " + f"delta=N/A " + f"outbox_id={outbox_id}" + ) return if attempts >= OUTBOX_MAX_ATTEMPTS: row.status = "failed" row.next_retry_at = None + _OUTBOX_PERF_T0_BY_ID.pop(outbox_id, None) else: row.status = "retrying" delay_seconds = min(30, 2 * (2 ** max(0, attempts - 1))) - row.next_retry_at = datetime.utcnow() + timedelta(seconds=delay_seconds) + row.next_retry_at = now_cn_naive() + timedelta(seconds=delay_seconds) row.last_error = (send_error or "unknown_send_error")[:1000] await db_update.commit() @@ -1466,7 +1520,7 @@ class FeishuIntegrationService: ts = int(str(raw).strip()) if ts > 10_000_000_000: ts = ts / 1000.0 - return datetime.fromtimestamp(ts, tz=timezone.utc) + return datetime.fromtimestamp(ts, tz=CN_TZ) except Exception: return None diff --git a/backend/app/services/feishu_long_connection.py b/backend/app/services/feishu_long_connection.py index 44e50544..1ebaadc4 100644 --- a/backend/app/services/feishu_long_connection.py +++ b/backend/app/services/feishu_long_connection.py @@ -18,6 +18,7 @@ from ..core.logging import get_logger from ..models import Integration from .feishu_integration import FeishuIntegrationService +from ..utils.timezone import CN_TZ, now_cn_aware logger = get_logger(__name__) DEFAULT_FEISHU_LONG_CONNECTION_RETRY_MAX_ATTEMPTS = 3 DEFAULT_FEISHU_LONG_CONNECTION_RETRY_BASE_SECONDS = 1.0 @@ -144,7 +145,7 @@ class FeishuLongConnectionManager: "last_disconnect_at": self._last_disconnect_at, "restart_attempts": self._restart_attempts, "next_retry_at": ( - datetime.fromtimestamp(self._next_retry_at, tz=timezone.utc).isoformat() + datetime.fromtimestamp(self._next_retry_at, tz=CN_TZ).isoformat() if self._next_retry_at > 0 else None ), @@ -249,7 +250,7 @@ class FeishuLongConnectionManager: try: await self._preflight_auth(tenant_id=self._tenant_id) - self._last_auth_at = datetime.now(timezone.utc).isoformat() + self._last_auth_at = now_cn_aware().isoformat() except Exception as exc: code, retryable = self._classify_error(exc) self._set_error(str(exc), code=code, retryable=retryable) @@ -266,7 +267,7 @@ class FeishuLongConnectionManager: return False self._clear_error() - self._last_connect_at = datetime.now(timezone.utc).isoformat() + self._last_connect_at = now_cn_aware().isoformat() self._last_disconnect_at = None self._restart_attempts = 0 self._next_retry_at = 0.0 @@ -403,7 +404,7 @@ class FeishuLongConnectionManager: logger.warning(f"feishu_long_connection_client_cleanup_failed: {exc}") self._client = None if self._running and not self._stopping: - self._last_disconnect_at = datetime.now(timezone.utc).isoformat() + self._last_disconnect_at = now_cn_aware().isoformat() if not self._last_error_code: self._set_error("feishu long connection thread exited", code="thread_exited", retryable=True) self._schedule_retry() @@ -416,6 +417,7 @@ class FeishuLongConnectionManager: return try: + perf_t0 = time.perf_counter() payload = self._to_dict(data) message_id = "" event_type = "" @@ -432,6 +434,13 @@ class FeishuLongConnectionManager: ).strip() except Exception: pass + if isinstance(payload, dict): + payload["_perf_t0"] = float(perf_t0) + print( + f"[PERF] message_received ts={now_cn_aware().isoformat()} T0={perf_t0:.6f} " + f"message_id={message_id or 'unknown'} " + f"event_type={event_type or 'unknown'}" + ) logger.info( "feishu_long_connection_event_received: " f"event_type={event_type or 'unknown'}, message_id={message_id or 'unknown'}" @@ -468,7 +477,7 @@ class FeishuLongConnectionManager: try: await self._process_payload(payload) - self._last_event_at = datetime.now(timezone.utc).isoformat() + self._last_event_at = now_cn_aware().isoformat() except Exception as exc: self._set_error(str(exc), code="event_process_failed", retryable=True) logger.warning(f"feishu_long_connection_process_failed: {exc}") @@ -506,6 +515,10 @@ class FeishuLongConnectionManager: if event_type: event_data = {**event_data, "type": event_type} + perf_t0 = payload.get("_perf_t0") + if isinstance(perf_t0, (int, float)): + event_data = {**event_data, "_perf_t0": float(perf_t0)} + if event_data: event_data = {**event_data, "_ingress_source": "long_connection"} diff --git a/backend/app/services/feishu_signal_pipeline.py b/backend/app/services/feishu_signal_pipeline.py index a93a0810..340fdb0e 100644 --- a/backend/app/services/feishu_signal_pipeline.py +++ b/backend/app/services/feishu_signal_pipeline.py @@ -42,8 +42,8 @@ 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 +from ..utils.timezone import CN_TZ, now_cn_aware logger = get_logger(__name__) PROJECT_PRIORITY_ORDER = { @@ -4048,7 +4048,7 @@ class FeishuSignalPipelineService: pending_rows = self._dedupe_queue_items(pending_rows) 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 报错。 + # DB 列是 timestamp without time zone，比较值必须使用 naive 北京时间，避免 asyncpg aware/naive 报错。 today_start = _utc_now().astimezone(CN_TZ).replace( hour=0, minute=0, diff --git a/backend/app/services/integration.py b/backend/app/services/integration.py index 12c2b7e9..61057f26 100644 --- a/backend/app/services/integration.py +++ b/backend/app/services/integration.py @@ -16,6 +16,7 @@ from datetime import datetime, timedelta from enum import Enum import httpx +from ..utils.timezone import now_cn_naive logger = logging.getLogger(__name__) @@ -118,12 +119,12 @@ class IntegrationService: IntegrationType.PLANE: IntegrationHealth( type=IntegrationType.PLANE, status=IntegrationStatus.PENDING, - last_check=datetime.utcnow(), + last_check=now_cn_naive(), ), IntegrationType.LARK: IntegrationHealth( type=IntegrationType.LARK, status=IntegrationStatus.PENDING, - last_check=datetime.utcnow(), + last_check=now_cn_naive(), ), } @@ -180,7 +181,7 @@ class IntegrationService: errors=[f"Plane integration is {health.status.value}"], ) - start_time = datetime.utcnow() + start_time = now_cn_naive() try: # 获取 Plane 服务 @@ -198,7 +199,7 @@ class IntegrationService: await self._record_failure(integration_type, ", ".join(result.get("errors", []))) # 计算延迟 - latency = (datetime.utcnow() - start_time).total_seconds() * 1000 + latency = (now_cn_naive() - start_time).total_seconds() * 1000 health.latency_ms = latency return TaskSyncResult( @@ -245,7 +246,7 @@ class IntegrationService: error=f"Lark integration is {health.status.value}", ) - start_time = datetime.utcnow() + start_time = now_cn_naive() try: # 获取飞书服务 @@ -259,7 +260,7 @@ class IntegrationService: await self._record_success(integration_type) # 计算延迟 - latency = (datetime.utcnow() - start_time).total_seconds() * 1000 + latency = (now_cn_naive() - start_time).total_seconds() * 1000 health.latency_ms = latency # 发送事件 @@ -384,12 +385,12 @@ class IntegrationService: """检查 Plane 集成健康""" integration_type = IntegrationType.PLANE health = self._health_status[integration_type] - start_time = datetime.utcnow() + start_time = now_cn_naive() if not self.config.plane.enabled or not self.config.plane.api_key: health.status = IntegrationStatus.INACTIVE health.message = "Plane integration is disabled" - health.last_check = datetime.utcnow() + health.last_check = now_cn_naive() return health try: @@ -399,14 +400,14 @@ class IntegrationService: health.status = IntegrationStatus.ACTIVE health.message = "Connected" - health.last_check = datetime.utcnow() - health.latency_ms = (datetime.utcnow() - start_time).total_seconds() * 1000 + health.last_check = now_cn_naive() + health.latency_ms = (now_cn_naive() - start_time).total_seconds() * 1000 except Exception as e: logger.warning(f"[Integration] Plane health check failed: {e}") health.status = IntegrationStatus.ERROR health.message = f"Health check failed: {str(e)}" - health.last_check = datetime.utcnow() + health.last_check = now_cn_naive() return health @@ -414,12 +415,12 @@ class IntegrationService: """检查飞书集成健康""" integration_type = IntegrationType.LARK health = self._health_status[integration_type] - start_time = datetime.utcnow() + start_time = now_cn_naive() if not self.config.lark.enabled or not self.config.lark.app_id: health.status = IntegrationStatus.INACTIVE health.message = "Lark integration is disabled" - health.last_check = datetime.utcnow() + health.last_check = now_cn_naive() return health try: @@ -434,14 +435,14 @@ class IntegrationService: health.status = IntegrationStatus.DEGRADED health.message = "Running in simulation mode" - health.last_check = datetime.utcnow() - health.latency_ms = (datetime.utcnow() - start_time).total_seconds() * 1000 + health.last_check = now_cn_naive() + health.latency_ms = (now_cn_naive() - start_time).total_seconds() * 1000 except Exception as e: logger.warning(f"[Integration] Lark health check failed: {e}") health.status = IntegrationStatus.ERROR health.message = f"Health check failed: {str(e)}" - health.last_check = datetime.utcnow() + health.last_check = now_cn_naive() return health @@ -494,7 +495,7 @@ class IntegrationService: async def _record_success(self, integration_type: IntegrationType) -> None: """记录成功操作""" health = self._health_status[integration_type] - health.last_success = datetime.utcnow() + health.last_success = now_cn_naive() health.consecutive_failures = 0 if health.status == IntegrationStatus.ERROR: health.status = IntegrationStatus.ACTIVE @@ -505,7 +506,7 @@ class IntegrationService: health.last_error = error health.error_count += 1 health.consecutive_failures += 1 - health.last_check = datetime.utcnow() + health.last_check = now_cn_naive() # 达到降级阈值 if health.consecutive_failures >= self.MAX_CONSECUTIVE_FAILURES: diff --git a/backend/app/services/llm.py b/backend/app/services/llm.py index ec1b219d..9a1c6386 100644 --- a/backend/app/services/llm.py +++ b/backend/app/services/llm.py @@ -31,6 +31,7 @@ from ..config.ai_prompts import build_default_role_prompt_bundle from ..database import AsyncSessionLocal from ..models import AICallLog, Integration +from ..utils.timezone import now_cn_aware try: import redis.asyncio as redis_async @@ -290,10 +291,28 @@ def _cn_now_text() -> str: return datetime.now(cn_tz).strftime("%Y-%m-%d %H:%M:%S") +def _cn_now_text_ms() -> str: + cn_tz = timezone(timedelta(hours=8)) + return datetime.now(cn_tz).strftime("%Y-%m-%d %H:%M:%S.%f")[:-3] + + def _runtime_log(scope: str, message: str) -> None: print(f"[{_cn_now_text()}][{scope}] {message}") +def _looks_like_openclaw_timeout_message(text: str) -> bool: + lowered = str(text or "").strip().lower() + if not lowered: + return False + timeout_signals = ( + "request timed out before a response was generated", + "agents.defaults.timeoutseconds", + "timeoutseconds", + "before a response was generated", + ) + return any(signal in lowered for signal in timeout_signals) + + @dataclass class ChatCompletionResult: text: str @@ -460,7 +479,7 @@ class LLMService: raise ValueError("api_key 不能为空") integration, bundle = await self._get_or_init_bundle(db, tenant_id=tenant_id) - now = datetime.now(timezone.utc).isoformat() + now = now_cn_aware().isoformat() engine_id = str(uuid4()) defaults = PROVIDER_DEFAULTS[provider] engine = { @@ -521,7 +540,7 @@ class LLMService: if base_url is not None: target["base_url"] = (base_url or "").strip().rstrip("/") or defaults["base_url"] - target["updated_at"] = datetime.now(timezone.utc).isoformat() + target["updated_at"] = now_cn_aware().isoformat() integration.config = json.loads(json.dumps(bundle)) await db.commit() return self._sanitize_engine_for_response(target) @@ -1101,6 +1120,21 @@ class LLMService: if cleaned_system_prompt: messages.append({"role": "system", "content": cleaned_system_prompt}) messages.append({"role": "user", "content": prompt}) + request_trace_id = uuid4().hex[:12] + request_started_at_perf = time.perf_counter() + function_label = str(function_name or "llm_generate") + openclaw_agent_timeout = int(openclaw_agents_timeout_seconds) if is_openclaw_gateway_request else None + + if is_openclaw_gateway_request: + _runtime_log( + "OpenClaw调用", + ( + f"start ts={_cn_now_text_ms()} trace_id={request_trace_id} " + f"function={function_label} provider={provider_name} model={model} " + f"request_timeout={float(request_timeout):.2f}s " + f"agent_timeout={openclaw_agent_timeout if openclaw_agent_timeout is not None else 'N/A'}" + ), + ) try: if provider_name == "ollama": @@ -1150,7 +1184,26 @@ class LLMService: } async with httpx.AsyncClient(timeout=request_timeout, trust_env=False) as client: resp = await client.post(endpoint, headers=headers, json=payload) - resp.raise_for_status() + if is_openclaw_gateway_request: + elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0 + _runtime_log( + "OpenClaw调用", + ( + f"http ts={_cn_now_text_ms()} trace_id={request_trace_id} " + f"function={function_label} status={resp.status_code} elapsed_ms={elapsed_ms:.0f}" + ), + ) + if resp.status_code >= 400: + body_preview = str(resp.text or "").replace("\n", " ").strip()[:260] + if is_openclaw_gateway_request: + _runtime_log( + "OpenClaw调用", + ( + f"http_error ts={_cn_now_text_ms()} trace_id={request_trace_id} " + f"function={function_label} status={resp.status_code} body={body_preview}" + ), + ) + resp.raise_for_status() data = resp.json() choices = data.get("choices") or [] if not choices: @@ -1174,10 +1227,72 @@ class LLMService: "LLM 返回空内容: " f"message_keys={msg_keys}, choice_keys={choice_keys}, choice_preview={choice_preview}" ) + if _looks_like_openclaw_timeout_message(text): + elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0 + _runtime_log( + "OpenClaw超时" if is_openclaw_gateway_request else "LLM超时文本", + ( + f"detected ts={_cn_now_text_ms()} trace_id={request_trace_id} " + f"function={function_label} elapsed_ms={elapsed_ms:.0f} " + f"text={text.replace(chr(10), ' ')[:260]}" + ), + ) + logger.warning( + "llm_timeout_payload_detected", + trace_id=request_trace_id, + function_name=function_label, + provider=provider_name, + model=model, + base_url=str(base_url or "")[:120], + request_timeout_seconds=float(request_timeout), + agent_timeout_seconds=openclaw_agent_timeout, + text_preview=text.replace("\n", " ")[:260], + ) + if is_openclaw_gateway_request: + elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0 + _runtime_log( + "OpenClaw调用", + ( + f"end ts={_cn_now_text_ms()} trace_id={request_trace_id} " + f"function={function_label} elapsed_ms={elapsed_ms:.0f} text_len={len(text)}" + ), + ) if use_provider_circuit_breaker: self._record_provider_success(circuit_key) return ChatCompletionResult(text=text, raw_response=data) except Exception as exc: + if _looks_like_openclaw_timeout_message(str(exc)): + elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0 + _runtime_log( + "LLM超时异常", + ( + f"detected ts={_cn_now_text_ms()} trace_id={request_trace_id} " + f"function={function_label} elapsed_ms={elapsed_ms:.0f} " + f"error={type(exc).__name__}:{str(exc)[:260]}" + ), + ) + logger.warning( + "llm_timeout_exception_detected", + trace_id=request_trace_id, + function_name=function_label, + provider=provider_name, + model=model, + base_url=str(base_url or "")[:120], + request_timeout_seconds=float(request_timeout), + agent_timeout_seconds=openclaw_agent_timeout, + error_type=type(exc).__name__, + error=str(exc)[:260], + ) + if is_openclaw_gateway_request: + elapsed_ms = (time.perf_counter() - request_started_at_perf) * 1000.0 + _runtime_log( + "OpenClaw调用", + ( + f"exception ts={_cn_now_text_ms()} trace_id={request_trace_id} " + f"function={function_label} elapsed_ms={elapsed_ms:.0f} " + f"error={type(exc).__name__}:{str(exc)[:260]}" + ), + ) if use_provider_circuit_breaker: self._record_provider_failure(circuit_key, exc) raise diff --git a/backend/app/services/model_capability_service.py b/backend/app/services/model_capability_service.py index 64c9826b..df795d34 100644 --- a/backend/app/services/model_capability_service.py +++ b/backend/app/services/model_capability_service.py @@ -26,6 +26,7 @@ from ..models import ( from ..services.ai_service import AIService +from ..utils.timezone import now_cn_naive class ModelCapabilityService: """模型能力矩阵核心服务""" @@ -649,7 +650,7 @@ class ModelCapabilityService: raise ValueError(f"能力差距 {gap_id} 不存在") gap.status = "resolved" - gap.resolved_at = datetime.utcnow() + gap.resolved_at = now_cn_naive() gap.resolution_notes = resolution_notes await self.db.commit() diff --git a/backend/app/services/notification_service.py b/backend/app/services/notification_service.py index c1e1babe..ff91d86e 100644 --- a/backend/app/services/notification_service.py +++ b/backend/app/services/notification_service.py @@ -10,6 +10,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from ..models import Notification, User +from ..utils.timezone import CN_TZ, now_cn_aware class NotificationService: def __init__(self, db: AsyncSession): self.db = db @@ -215,9 +216,9 @@ class NotificationService: @staticmethod def _format_cn_time(dt: Optional[datetime]) -> str: if not dt: - dt = datetime.now(timezone.utc) + dt = now_cn_aware() if dt.tzinfo is None: - dt = dt.replace(tzinfo=timezone.utc) + dt = dt.replace(tzinfo=CN_TZ) cn_tz = timezone(timedelta(hours=8)) return dt.astimezone(cn_tz).strftime("%Y-%m-%d %H:%M:%S") diff --git a/backend/app/services/openclaw_commander_service.py b/backend/app/services/openclaw_commander_service.py index f62cb107..1a10e41b 100644 --- a/backend/app/services/openclaw_commander_service.py +++ b/backend/app/services/openclaw_commander_service.py @@ -41,9 +41,9 @@ 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 +from ..utils.timezone import now_cn_naive SUPPORTED_COMMANDER_OPERATION_CODES = ( "project.create", "project.update", @@ -2025,7 +2025,7 @@ class OperationExecutorService: task.status = status if status in DONE_TASK_STATUSES: - task.completed_at = datetime.now(timezone.utc).replace(tzinfo=None) + task.completed_at = now_cn_naive() await log_operation( self.db, diff --git a/backend/app/services/project_knowledge_graph.py b/backend/app/services/project_knowledge_graph.py index a22bc461..422fc5eb 100644 --- a/backend/app/services/project_knowledge_graph.py +++ b/backend/app/services/project_knowledge_graph.py @@ -32,11 +32,11 @@ 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 +from ..utils.timezone import now_cn_naive def _utc_now_naive() -> datetime: return now_cn_naive() diff --git a/backend/app/services/prompt_profile_service.py b/backend/app/services/prompt_profile_service.py index 73b6f15c..f6eecd24 100644 --- a/backend/app/services/prompt_profile_service.py +++ b/backend/app/services/prompt_profile_service.py @@ -13,6 +13,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from ..config.ai_prompts import build_default_role_prompt_bundle from ..models import Integration +from ..utils.timezone import now_cn_aware INTEGRATION_TYPE = "ai_prompt_profiles" ROLE_KEY_RE = re.compile(r"^[a-z0-9][a-z0-9_-]{1,31}$") MAX_ROLES = 12 @@ -35,7 +36,7 @@ def _safe_json(value: Any) -> dict[str, Any]: def _now_iso() -> str: - return datetime.now(timezone.utc).isoformat() + return now_cn_aware().isoformat() class PromptProfileService: diff --git a/backend/app/services/signal_capture.py b/backend/app/services/signal_capture.py index b53e7db5..18b29adc 100644 --- a/backend/app/services/signal_capture.py +++ b/backend/app/services/signal_capture.py @@ -46,6 +46,7 @@ from ..utils.project_members import ensure_project_default_members from ..utils.sequence import ensure_project_id_sequence, ensure_signal_id_sequence +from ..utils.timezone import CN_TZ, now_cn_aware EMOJI_UNICODE_RE = re.compile( r"[\U0001F300-\U0001F6FF\U0001F700-\U0001F77F\U0001F780-\U0001F7FF" r"\U0001F800-\U0001F8FF\U0001F900-\U0001F9FF\U0001FA00-\U0001FAFF" @@ -196,9 +197,9 @@ class ProactiveSignalCaptureService: - 分享技术文章 """ trace_id = f"feishu-{uuid4().hex[:12]}" - captured_dt = captured_at or datetime.now(timezone.utc) + captured_dt = captured_at or now_cn_aware() if captured_dt.tzinfo is None: - captured_dt = captured_dt.replace(tzinfo=timezone.utc) + captured_dt = captured_dt.replace(tzinfo=CN_TZ) proposer_name = self._normalize_single_line_text(sender or "未知用户", max_len=80) or "未知用户" proposer_open_id = self._normalize_single_line_text(proposer_open_id or "", max_len=128) or None @@ -803,7 +804,7 @@ class ProactiveSignalCaptureService: processed=True, strategic_analysis=json.dumps({ 'url': url, - 'captured_at': datetime.now(timezone.utc).isoformat(), + 'captured_at': now_cn_aware().isoformat(), 'user_id': user_id, 'ai_analysis': analysis.get('analysis', {}) }), @@ -862,7 +863,7 @@ class ProactiveSignalCaptureService: 'repo_url': repo_url, 'event_type': event_type, 'event_data': event_data, - 'captured_at': datetime.now(timezone.utc).isoformat(), + 'captured_at': now_cn_aware().isoformat(), 'ai_analysis': analysis.get('analysis', {}) }), tasks_generated=json.dumps([]), @@ -922,7 +923,7 @@ class ProactiveSignalCaptureService: strategic_analysis=json.dumps({ 'hn_url': story.get('url'), 'hn_points': story.get('points'), - 'captured_at': datetime.now(timezone.utc).isoformat(), + 'captured_at': now_cn_aware().isoformat(), 'ai_analysis': analysis.get('analysis', {}) }), tasks_generated=json.dumps(analysis.get('suggested_tasks', [])), @@ -970,7 +971,7 @@ class ProactiveSignalCaptureService: logger.info("signal_capture_ai_think", stage=stage, **safe_payload) try: stage_zh = AI_THINK_STAGE_ZH.get(stage, stage) - now_text = self._format_cn_time(datetime.now(timezone.utc)) + now_text = self._format_cn_time(now_cn_aware()) print( f"[{now_text}][AI思考][SignalCapture] {stage_zh} " + json.dumps({"stage": stage, **safe_payload}, ensure_ascii=False) @@ -1851,7 +1852,7 @@ class ProactiveSignalCaptureService: def _format_cn_time(dt: datetime) -> str: safe_dt = dt if safe_dt.tzinfo is None: - safe_dt = safe_dt.replace(tzinfo=timezone.utc) + safe_dt = safe_dt.replace(tzinfo=CN_TZ) cn_tz = timezone(timedelta(hours=8)) return safe_dt.astimezone(cn_tz).strftime("%Y-%m-%d %H:%M:%S") @@ -2135,9 +2136,9 @@ URL: {url} requested_at: Optional[datetime] = None, ) -> Optional[int]: """当分类为 new_project 时，自动创建项目并关联当前信号。""" - requested_dt = requested_at or datetime.now(timezone.utc) + requested_dt = requested_at or now_cn_aware() if requested_dt.tzinfo is None: - requested_dt = requested_dt.replace(tzinfo=timezone.utc) + requested_dt = requested_dt.replace(tzinfo=CN_TZ) project_brief = await self._draft_new_project_brief( message=message, tenant_id=tenant_id, diff --git a/backend/app/services/signal_source_orchestration_service.py b/backend/app/services/signal_source_orchestration_service.py index de60cb1e..7febc3a8 100644 --- a/backend/app/services/signal_source_orchestration_service.py +++ b/backend/app/services/signal_source_orchestration_service.py @@ -28,11 +28,11 @@ 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 +from ..utils.timezone import now_cn_naive SOURCE_ONBOARDING_ACTION_RE = re.compile( r"(新增|增加|添加|接入|配置|注册|建立|加个|加一个|加).{0,8}(信号源|来源|监控源|channel|source|webhook|飞书|github|url|rss)", re.IGNORECASE, diff --git a/backend/app/services/skill_score_service.py b/backend/app/services/skill_score_service.py index d2a9fe06..2c27f1c1 100644 --- a/backend/app/services/skill_score_service.py +++ b/backend/app/services/skill_score_service.py @@ -14,6 +14,7 @@ from ..database import AsyncSessionLocal from ..models import AICallLog, Task, UserSkillBinding from .llm import llm_service +from ..utils.timezone import now_cn_naive PRIORITY_WEIGHT = { "critical": 1.5, "high": 1.2, @@ -117,8 +118,8 @@ async def _apply_skill_deltas( float(row.efficiency_score or 0.0) * 0.5 ) row.proficiency = int(_clamp(round(combined), 0, 100)) - row.last_used_at = datetime.utcnow() - row.updated_at = datetime.utcnow() + row.last_used_at = now_cn_naive() + row.updated_at = now_cn_naive() touched += 1 return touched diff --git a/backend/app/services/skill_service.py b/backend/app/services/skill_service.py index eed98e12..26ae3398 100644 --- a/backend/app/services/skill_service.py +++ b/backend/app/services/skill_service.py @@ -21,6 +21,7 @@ from .ai_pmo_kernel_service import AIPMOKernelService # 系统预设 AI 开发专属技能标签 +from ..utils.timezone import now_cn_naive SYSTEM_SKILL_TAGS = [ {"name": "Python", "slug": "python", "category": "language", "sort_order": 1}, {"name": "TypeScript", "slug": "typescript", "category": "language", "sort_order": 2}, @@ -196,7 +197,7 @@ async def update_skill_tag( tag.description = description if sort_order is not None: tag.sort_order = sort_order - tag.updated_at = datetime.utcnow() + tag.updated_at = now_cn_naive() await db.commit() await db.refresh(tag) return tag diff --git a/backend/app/services/sstc_service.py b/backend/app/services/sstc_service.py index cacab6df..0355a0e7 100644 --- a/backend/app/services/sstc_service.py +++ b/backend/app/services/sstc_service.py @@ -22,6 +22,7 @@ from ..models import ( ) from ..utils.project_members import ensure_project_default_members +from ..utils.timezone import now_cn_aware DONE_STATUSES = {"done", "completed"} ACTIVE_STATUSES = {"in_progress", "review"} BLOCKED_STATUSES = {"blocked"} @@ -248,7 +249,7 @@ class SSTCService: [ t for t in tasks - if t.due_date and t.due_date < datetime.now(timezone.utc).replace(tzinfo=None) and (t.status or "").lower() not in DONE_STATUSES + if t.due_date and t.due_date < now_cn_aware().replace(tzinfo=None) and (t.status or "").lower() not in DONE_STATUSES ] ) return { @@ -453,7 +454,7 @@ class SSTCService: overdue_tasks = [ t for t in tasks - if t.due_date and t.due_date < datetime.now(timezone.utc).replace(tzinfo=None) and (t.status or "").lower() not in DONE_STATUSES + if t.due_date and t.due_date < now_cn_aware().replace(tzinfo=None) and (t.status or "").lower() not in DONE_STATUSES ] risk_items = [ { diff --git a/backend/app/services/symbol_index_service.py b/backend/app/services/symbol_index_service.py index a8a9a1d6..f133457c 100644 --- a/backend/app/services/symbol_index_service.py +++ b/backend/app/services/symbol_index_service.py @@ -22,6 +22,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from ..models import SymbolDefinition, SymbolReference +from ..utils.timezone import CN_TZ, now_cn_aware, now_cn_naive TS_SYMBOL_RE = re.compile( r"^\s*(?:export\s+)?(?:default\s+)?(?:async\s+)?(?:function|class|interface|type|const)\s+([A-Za-z_][A-Za-z0-9_]*)", re.MULTILINE, @@ -127,7 +128,7 @@ class SymbolIndexService: text = file_path.read_text(encoding="utf-8", errors="ignore") rel_path = self._to_rel_path(file_path=file_path, workspace_root=workspace_root, backend_root=backend_root) file_hash = hashlib.sha1(text.encode("utf-8")).hexdigest() - mtime = datetime.fromtimestamp(file_path.stat().st_mtime, tz=timezone.utc).replace(tzinfo=None) + mtime = datetime.fromtimestamp(file_path.stat().st_mtime, tz=CN_TZ).replace(tzinfo=None) if not force_reindex: latest_hash = await self._get_latest_file_hash(rel_path=rel_path) if latest_hash and latest_hash == file_hash: @@ -156,7 +157,7 @@ class SymbolIndexService: await self.db.commit() stats["errors"] = stats["errors"][:20] - stats["finished_at"] = datetime.now(timezone.utc).isoformat() + stats["finished_at"] = now_cn_aware().isoformat() return stats async def search_symbols( @@ -327,7 +328,7 @@ class SymbolIndexService: {"module": str(path), "symbols": int(cnt or 0)} for path, cnt in file_groups ], - "snapshot_at": datetime.now(timezone.utc).isoformat(), + "snapshot_at": now_cn_aware().isoformat(), } async def _get_latest_file_hash(self, *, rel_path: str) -> Optional[str]: @@ -391,7 +392,7 @@ class SymbolIndexService: source_mtime=source_mtime, status="active", payload=item.payload, - last_indexed_at=datetime.now(timezone.utc).replace(tzinfo=None), + last_indexed_at=now_cn_naive(), ) self.db.add(row) created_symbols.append(row) diff --git a/backend/app/services/task_persistence.py b/backend/app/services/task_persistence.py index a4e0ca3b..c753ee5a 100644 --- a/backend/app/services/task_persistence.py +++ b/backend/app/services/task_persistence.py @@ -14,6 +14,7 @@ from ..utils.title import compact_title from . import workload_service from .operation_log_service import log_operation +from ..utils.timezone import CN_TZ TASK_PRIORITY_MAP = { "p0": "high", "critical": "high", @@ -98,7 +99,7 @@ def normalize_due_date(value: Any) -> Optional[datetime]: return None if isinstance(value, datetime): if value.tzinfo is not None: - return value.astimezone(timezone.utc).replace(tzinfo=None) + return value.astimezone(CN_TZ).replace(tzinfo=None) return value text = str(value).strip() if not text: diff --git a/backend/app/services/workload_alert_service.py b/backend/app/services/workload_alert_service.py index d451af3e..42ce766c 100644 --- a/backend/app/services/workload_alert_service.py +++ b/backend/app/services/workload_alert_service.py @@ -11,6 +11,7 @@ from sqlalchemy.ext.asyncio import AsyncSession from ..models import OperationLog, Task, User, UserProject, UserWorkload from .lark import lark_service +from ..utils.timezone import now_cn_naive DONE_STATUSES = {"done", "completed", "closed", "resolved", "archived"} @@ -200,7 +201,7 @@ async def process_tenant_unresolved_overloads( user_rows = rows.all() created = 0 - now = datetime.utcnow() + now = now_cn_naive() for user_id_raw, ratio_raw in user_rows: user_id = int(user_id_raw) ratio = float(ratio_raw or 0.0) diff --git a/backend/app/services/workload_service.py b/backend/app/services/workload_service.py index 6f91ae3e..7022309c 100644 --- a/backend/app/services/workload_service.py +++ b/backend/app/services/workload_service.py @@ -13,6 +13,7 @@ from ..config.ai_prompts import PROMPT_WORKLOAD_OPTIMIZATION from ..models import AICallLog, Task, User, UserProject, UserSkillBinding, UserWorkload from .llm import llm_service +from ..utils.timezone import now_cn_naive LEVEL_HOURS_MULTIPLIER = { "expert": 0.8, "advanced": 0.9, @@ -101,7 +102,7 @@ async def recalculate_user_workload_ratio( ) workload.workload_ratio = ratio user.workload = int(max(0, min(200, ratio))) - user.updated_at = datetime.utcnow() + user.updated_at = now_cn_naive() try: from . import workload_alert_service diff --git a/backend/app/websocket.py b/backend/app/websocket.py index f97e5e90..12d298f4 100644 --- a/backend/app/websocket.py +++ b/backend/app/websocket.py @@ -11,6 +11,7 @@ import logging from fastapi import WebSocket, WebSocketDisconnect from enum import Enum +from .utils.timezone import now_cn_aware, now_cn_naive logger = logging.getLogger(__name__) @@ -29,7 +30,7 @@ class WebSocketConnection: self.websocket = websocket self.client_id = client_id self.subscriptions: Set[str] = set() - self.connected_at = datetime.utcnow() + self.connected_at = now_cn_naive() self.last_ping: Optional[datetime] = None async def send(self, data: Dict[str, Any]) -> bool: @@ -44,8 +45,8 @@ class WebSocketConnection: async def ping(self) -> bool: """发送 ping""" try: - await self.websocket.send_json({"type": "ping", "timestamp": datetime.utcnow().timestamp()}) - self.last_ping = datetime.utcnow() + await self.websocket.send_json({"type": "ping", "timestamp": now_cn_aware().timestamp()}) + self.last_ping = now_cn_naive() return True except Exception as e: logger.error(f"Error pinging client {self.client_id}: {e}") @@ -113,7 +114,7 @@ class WebSocketManager: await connection.send({ "type": ConnectionEventType.CONNECTED, "client_id": client_id, - "timestamp": datetime.utcnow().timestamp(), + "timestamp": now_cn_aware().timestamp(), }) logger.info(f"WebSocket client connected: {client_id}") @@ -135,7 +136,7 @@ class WebSocketManager: await connection.send({ "type": ConnectionEventType.SUBSCRIBED, "event_types": list(connection.subscriptions), - "timestamp": datetime.utcnow().timestamp(), + "timestamp": now_cn_aware().timestamp(), }) async def unsubscribe(self, client_id: str, event_types: List[str]): @@ -148,7 +149,7 @@ class WebSocketManager: await connection.send({ "type": ConnectionEventType.UNSUBSCRIBED, "event_types": list(connection.subscriptions), - "timestamp": datetime.utcnow().timestamp(), + "timestamp": now_cn_aware().timestamp(), }) async def broadcast(self, event_type: str, data: Dict[str, Any]) -> int: @@ -166,7 +167,7 @@ class WebSocketManager: "type": "event", "event_type": event_type, "data": data, - "timestamp": datetime.utcnow().timestamp(), + "timestamp": now_cn_aware().timestamp(), } success_count = 0 diff --git a/scripts/ai_steward_eval.py b/scripts/ai_steward_eval.py index 8c2501f1..f46bbf4e 100755 --- a/scripts/ai_steward_eval.py +++ b/scripts/ai_steward_eval.py @@ -17,6 +17,7 @@ from datetime import datetime, timezone from pathlib import Path from typing import Any +from backend.app.utils.timezone import now_cn_aware try: import httpx except Exception: # pragma: no cover @@ -259,7 +260,7 @@ def _build_report( return { "version": "v1", - "generated_at": datetime.now(timezone.utc).isoformat(), + "generated_at": now_cn_aware().isoformat(), "mode": mode, "api_url": api_url, "thresholds": thresholds, @@ -297,7 +298,7 @@ def _build_report( def _write_report(report: dict[str, Any], output_dir: Path) -> Path: output_dir.mkdir(parents=True, exist_ok=True) - ts = datetime.now(timezone.utc).strftime("%Y%m%dT%H%M%SZ") + ts = now_cn_aware().strftime("%Y%m%dT%H%M%SZ") report_path = output_dir / f"ai_steward_eval_{ts}.json" report_path.write_text(json.dumps(report, ensure_ascii=False, indent=2), encoding="utf-8") latest = output_dir / "latest.json" diff --git a/scripts/ai_steward_pressure_coach.py b/scripts/ai_steward_pressure_coach.py index 0f972507..65a314c9 100755 --- a/scripts/ai_steward_pressure_coach.py +++ b/scripts/ai_steward_pressure_coach.py @@ -26,6 +26,7 @@ from typing import Any, Iterable import httpx +from backend.app.utils.timezone import now_cn_aware DEFAULT_TEMPLATE_LEAK_PATTERNS = [ "当前系统采用前后端分层架构", "fastapi + sqlalchemy", @@ -823,7 +824,7 @@ async def _amain() -> int: args.sleep_ms = max(0, min(args.sleep_ms, 10000)) args.timeout_seconds = max(5.0, min(args.timeout_seconds, 300.0)) - started_at = datetime.now(timezone.utc).isoformat() + started_at = now_cn_aware().isoformat() scenarios = _build_scenarios(args.mode) token = await _login_token( base_url=args.base_url, @@ -853,7 +854,7 @@ async def _amain() -> int: ] results = await asyncio.gather(*tasks) metrics = _merge_metrics(results) - finished_at = datetime.now(timezone.utc).isoformat() + finished_at = now_cn_aware().isoformat() report = _build_report( started_at=started_at,