From 0cda45238abb80ac3a6410c0ee6fc3af377b4f79 Mon Sep 17 00:00:00 2001 From: JiuContinent Date: Thu, 9 Jul 2026 17:26:19 +0800 Subject: [PATCH] =?UTF-8?q?```=20feat:=20=E6=B7=BB=E5=8A=A0AI=E8=AE=B0?= =?UTF-8?q?=E5=BF=86=E6=A8=A1=E5=9D=97=E5=92=8C=E4=BA=8B=E4=BB=B6=E8=B0=83?= =?UTF-8?q?=E5=BA=A6=E7=B3=BB=E7=BB=9F?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增AI记忆模块，支持本地记忆召回和自动写入功能 - 实现事件调度系统，支持批量处理待定事件和重试机制 - 集成心跳监控机制，跟踪API、调度器和工作节点状态 - 扩展仪表板数据统计，包含AI记忆条目和心跳概要 - 添加企业运营分析报告功能，提供财务、采购等多维度分析 - 更新配置设置，增加事件调度和AI记忆相关参数 - 优化任务队列，添加事件分发任务类型 - 扩展审计日志，记录AI记忆操作和事件调度行为 - 实现领域事件模型，支持事件持久化和状态管理 - 添加观察性服务，监控系统组件健康状况 ``` --- alembic/env.py | 4 + ...90001_v3_readonly_operations_completion.py | 139 +++++++++ app/api/router.py | 2 + app/core/config.py | 29 ++ app/core/constants.py | 1 + app/core/scheduler.py | 83 +++++- app/core/task_queue.py | 26 ++ app/modules/ai_agent/constants.py | 5 + app/modules/ai_agent/service.py | 48 ++- app/modules/ai_memory/__init__.py | 1 + app/modules/ai_memory/constants.py | 55 ++++ app/modules/ai_memory/models.py | 33 +++ app/modules/ai_memory/routes.py | 44 +++ app/modules/ai_memory/schemas.py | 29 ++ app/modules/ai_memory/service.py | 277 ++++++++++++++++++ app/modules/audit/constants.py | 12 + app/modules/dashboard/service.py | 10 +- app/modules/events/constants.py | 12 + app/modules/events/models.py | 8 + app/modules/events/routes.py | 18 +- app/modules/events/service.py | 205 ++++++++++++- app/modules/observability/constants.py | 21 ++ app/modules/observability/models.py | 23 ++ app/modules/observability/service.py | 114 ++++++- app/modules/reports/constants.py | 22 ++ app/modules/reports/routes.py | 18 ++ app/modules/reports/service.py | 182 +++++++++++- app/modules/workflows/constants.py | 4 + app/tasks.py | 25 ++ app/tools/init_db.py | 4 + app/tools/run_scheduler.py | 17 ++ docker-compose.yml | 18 ++ tests/test_smoke.py | 136 ++++++++- 33 files changed, 1591 insertions(+), 34 deletions(-) create mode 100644 alembic/versions/202607090001_v3_readonly_operations_completion.py create mode 100644 app/modules/ai_memory/__init__.py create mode 100644 app/modules/ai_memory/constants.py create mode 100644 app/modules/ai_memory/models.py create mode 100644 app/modules/ai_memory/routes.py create mode 100644 app/modules/ai_memory/schemas.py create mode 100644 app/modules/ai_memory/service.py create mode 100644 app/modules/observability/models.py create mode 100644 app/tools/run_scheduler.py diff --git a/alembic/env.py b/alembic/env.py index bb1772c..6a91fe4 100644 --- a/alembic/env.py +++ b/alembic/env.py @@ -5,10 +5,12 @@ from sqlalchemy import engine_from_config, pool from app.core.config import get_settings from app.core.db_base import Base +from app.modules.ai_memory import models as ai_memory_models from app.modules.audit import models as audit_models from app.modules.business import models as business_models from app.modules.events import models as event_models from app.modules.feishu import models as feishu_models +from app.modules.observability import models as observability_models from app.modules.workflows import models as workflow_models config = context.config @@ -21,10 +23,12 @@ settings = get_settings() # Keep imports referenced so SQLAlchemy model classes register with Base.metadata. _REGISTERED_MODEL_MODULES = ( + ai_memory_models, audit_models, business_models, event_models, feishu_models, + observability_models, workflow_models, ) diff --git a/alembic/versions/202607090001_v3_readonly_operations_completion.py b/alembic/versions/202607090001_v3_readonly_operations_completion.py new file mode 100644 index 0000000..7870341 --- /dev/null +++ b/alembic/versions/202607090001_v3_readonly_operations_completion.py @@ -0,0 +1,139 @@ +"""Complete V3 read-only operations foundation. + +Revision ID: 202607090001 +Revises: 202607080002 +Create Date: 2026-07-09 +""" + +from alembic import op +import sqlalchemy as sa +from sqlalchemy import inspect + + +revision = "202607090001" +down_revision = "202607080002" +branch_labels = None +depends_on = None + +DOMAIN_EVENTS_TABLE = "domain_events" +AI_MEMORY_ENTRIES_TABLE = "ai_memory_entries" +SYSTEM_HEARTBEATS_TABLE = "system_heartbeats" + + +def _table_exists(table_name: str) -> bool: + inspector = inspect(op.get_bind()) + return table_name in inspector.get_table_names() + + +def _column_names(table_name: str) -> set[str]: + inspector = inspect(op.get_bind()) + if table_name not in inspector.get_table_names(): + return set() + return {column["name"] for column in inspector.get_columns(table_name)} + + +def _add_column_if_missing(table_name: str, column: sa.Column) -> None: + if column.name not in _column_names(table_name): + op.add_column(table_name, column) + + +def _create_indexes(table_name: str, indexes: list[tuple[str, bool]]) -> None: + for column_name, unique in indexes: + op.create_index( + op.f(f"ix_{table_name}_{column_name}"), + table_name, + [column_name], + unique=unique, + if_not_exists=True, + ) + + +def upgrade() -> None: + for column in [ + sa.Column("next_attempt_at", sa.DateTime(), nullable=True), + sa.Column("locked_until", sa.DateTime(), nullable=True), + sa.Column("locked_by", sa.String(length=128), nullable=True), + sa.Column("max_attempts", sa.Integer(), nullable=True), + ]: + _add_column_if_missing(DOMAIN_EVENTS_TABLE, column) + _create_indexes( + DOMAIN_EVENTS_TABLE, + [ + ("next_attempt_at", False), + ("locked_until", False), + ("locked_by", False), + ], + ) + + if not _table_exists(AI_MEMORY_ENTRIES_TABLE): + op.create_table( + AI_MEMORY_ENTRIES_TABLE, + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("code", sa.String(length=64), nullable=False), + sa.Column("scope", sa.String(length=64), nullable=False), + sa.Column("subject", sa.String(length=128), nullable=False), + sa.Column("content", sa.Text(), nullable=False), + sa.Column("summary", sa.Text(), nullable=True), + sa.Column("tags", sa.JSON(), nullable=True), + sa.Column("source", sa.String(length=64), nullable=False), + sa.Column("importance", sa.Integer(), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("actor", sa.String(length=128), nullable=False), + sa.Column("last_used_at", sa.DateTime(), nullable=True), + sa.Column("expires_at", sa.DateTime(), nullable=True), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + _create_indexes( + AI_MEMORY_ENTRIES_TABLE, + [ + ("code", True), + ("scope", False), + ("subject", False), + ("source", False), + ("importance", False), + ("status", False), + ("actor", False), + ("last_used_at", False), + ("expires_at", False), + ("created_at", False), + ], + ) + + if not _table_exists(SYSTEM_HEARTBEATS_TABLE): + op.create_table( + SYSTEM_HEARTBEATS_TABLE, + sa.Column("id", sa.Integer(), autoincrement=True, nullable=False), + sa.Column("component", sa.String(length=128), nullable=False), + sa.Column("instance_id", sa.String(length=128), nullable=False), + sa.Column("status", sa.String(length=32), nullable=False), + sa.Column("last_seen_at", sa.DateTime(), nullable=False), + sa.Column("created_at", sa.DateTime(), nullable=False), + sa.Column("updated_at", sa.DateTime(), nullable=False), + sa.PrimaryKeyConstraint("id"), + ) + _create_indexes( + SYSTEM_HEARTBEATS_TABLE, + [ + ("component", False), + ("instance_id", False), + ("status", False), + ("last_seen_at", False), + ("created_at", False), + ], + ) + + +def downgrade() -> None: + for table_name in [SYSTEM_HEARTBEATS_TABLE, AI_MEMORY_ENTRIES_TABLE]: + if _table_exists(table_name): + op.drop_table(table_name) + + event_columns = _column_names(DOMAIN_EVENTS_TABLE) + for column_name in ["locked_by", "locked_until", "next_attempt_at"]: + if column_name in event_columns: + op.drop_index(op.f(f"ix_{DOMAIN_EVENTS_TABLE}_{column_name}"), table_name=DOMAIN_EVENTS_TABLE) + op.drop_column(DOMAIN_EVENTS_TABLE, column_name) + if "max_attempts" in _column_names(DOMAIN_EVENTS_TABLE): + op.drop_column(DOMAIN_EVENTS_TABLE, "max_attempts") diff --git a/app/api/router.py b/app/api/router.py index 0603031..b1109e5 100644 --- a/app/api/router.py +++ b/app/api/router.py @@ -2,6 +2,7 @@ from fastapi import APIRouter from app.core.constants import ApiResponseKey, ApiStatus from app.modules.ai_agent.routes import router as ai_router +from app.modules.ai_memory.routes import router as ai_memory_router from app.modules.audit.routes import router as audit_router from app.modules.business.routes import router as business_router from app.modules.dashboard.routes import router as dashboard_router @@ -28,6 +29,7 @@ api_router.include_router(dashboard_router, prefix="/dashboard", tags=["dashboar api_router.include_router(legacy_mysql_router, prefix="/integrations/mysql", tags=["mysql"]) api_router.include_router(feishu_router, prefix="/integrations/feishu", tags=["feishu"]) api_router.include_router(ai_router, prefix="/ai", tags=["ai"]) +api_router.include_router(ai_memory_router, prefix="/ai", tags=["ai-memory"]) api_router.include_router(reports_router, prefix="/reports", tags=["reports"]) api_router.include_router(risk_router, prefix="/risks", tags=["risks"]) api_router.include_router(audit_router, prefix="/audit", tags=["audit"]) diff --git a/app/core/config.py b/app/core/config.py index 5ac9856..7d53907 100644 --- a/app/core/config.py +++ b/app/core/config.py @@ -80,6 +80,34 @@ class Settings(BaseSettings): legacy_project_sync_cron_minute: int = 0 legacy_task_sync_cron_hour: int = 2 legacy_task_sync_cron_minute: int = 30 + event_dispatch_enabled: bool = True + event_dispatch_batch_size: int = 100 + event_dispatch_max_attempts: int = 3 + event_dispatch_retry_delay_seconds: int = 300 + event_dispatch_lock_seconds: int = 300 + event_dispatch_cron_minute: str = "*/5" + heartbeat_interval_seconds: int = 60 + ai_memory_enabled: bool = True + ai_memory_auto_write_enabled: bool = True + ai_memory_recall_limit: int = 5 + ai_memory_forbidden_keys: list[str] = Field( + default_factory=lambda: [ + "authorization", + "api_key", + "apikey", + "access_token", + "tenant_access_token", + "token", + "secret", + "password", + "openclaw_gateway_token", + "hermes_api_key", + "direct_llm_api_key", + "feishu_app_secret", + "feishu_verification_token", + ] + ) + @field_validator("cors_origins", mode="before") @classmethod def parse_cors_origins(cls, value: Any) -> list[str]: @@ -98,6 +126,7 @@ class Settings(BaseSettings): @field_validator( "openclaw_allowed_tools", "openclaw_allowed_actions", + "ai_memory_forbidden_keys", mode="before", ) @classmethod diff --git a/app/core/constants.py b/app/core/constants.py index 82d34ee..f3fb75f 100644 --- a/app/core/constants.py +++ b/app/core/constants.py @@ -6,6 +6,7 @@ class ActorValue(StrEnum): AUDITOR = "auditor" SYSTEM = "system" SCHEDULER = "scheduler" + WORKER = "worker" FEISHU = "feishu" diff --git a/app/core/scheduler.py b/app/core/scheduler.py index c5977ec..93964b8 100644 --- a/app/core/scheduler.py +++ b/app/core/scheduler.py @@ -1,8 +1,12 @@ +from socket import gethostname +from typing import Any + from fastapi import FastAPI -from app.core.constants import ActorValue from app.core.config import get_settings +from app.core.constants import ActorValue from app.modules.feishu.constants import FeishuReceiveIdType +from app.modules.observability.constants import HeartbeatComponent def attach_scheduler(app: FastAPI) -> None: @@ -12,35 +16,53 @@ def attach_scheduler(app: FastAPI) -> None: if not settings.scheduler_enabled: return + scheduler = create_scheduler(app) + + @app.on_event("startup") + def start_scheduler() -> None: + scheduler.start() + + @app.on_event("shutdown") + def stop_scheduler() -> None: + scheduler.shutdown(wait=False) + + +def create_scheduler(app: FastAPI | None = None) -> Any: + """Create the V2/V3 scheduler without requiring a FastAPI process.""" + from apscheduler.schedulers.background import BackgroundScheduler from app.core.database import SessionLocal from app.core.task_queue import ( enqueue_daily_brief_push, + enqueue_event_dispatch, enqueue_legacy_project_sync, enqueue_legacy_task_sync, enqueue_project_weekly_push, ) + from app.modules.observability.service import ObservabilityService from app.modules.reports.service import ReportService + settings = get_settings() scheduler = BackgroundScheduler(timezone="Asia/Shanghai") def run_daily_brief() -> None: db = SessionLocal() try: report = ReportService(db).daily_brief() - app.state.last_daily_brief = report + _set_state(app, "last_daily_brief", report) if ( settings.feishu_app_id and settings.feishu_app_secret and settings.feishu_default_chat_id ): if settings.task_queue_enabled: - app.state.last_daily_brief_dispatch = enqueue_daily_brief_push( + dispatch = enqueue_daily_brief_push( receive_id=settings.feishu_default_chat_id, receive_id_type=FeishuReceiveIdType.CHAT_ID, actor=ActorValue.SCHEDULER, ) + _set_state(app, "last_daily_brief_dispatch", dispatch) return ReportService(db).push_report( report, @@ -55,18 +77,19 @@ def attach_scheduler(app: FastAPI) -> None: db = SessionLocal() try: report = ReportService(db).project_weekly() - app.state.last_project_weekly = report + _set_state(app, "last_project_weekly", report) if ( settings.feishu_app_id and settings.feishu_app_secret and settings.feishu_default_chat_id ): if settings.task_queue_enabled: - app.state.last_project_weekly_dispatch = enqueue_project_weekly_push( + dispatch = enqueue_project_weekly_push( receive_id=settings.feishu_default_chat_id, receive_id_type=FeishuReceiveIdType.CHAT_ID, actor=ActorValue.SCHEDULER, ) + _set_state(app, "last_project_weekly_dispatch", dispatch) return ReportService(db).push_report( report, @@ -78,14 +101,31 @@ def attach_scheduler(app: FastAPI) -> None: db.close() def run_legacy_project_sync() -> None: - app.state.last_legacy_project_sync_dispatch = enqueue_legacy_project_sync( - actor=ActorValue.SCHEDULER, - ) + dispatch = enqueue_legacy_project_sync(actor=ActorValue.SCHEDULER) + _set_state(app, "last_legacy_project_sync_dispatch", dispatch) def run_legacy_task_sync() -> None: - app.state.last_legacy_task_sync_dispatch = enqueue_legacy_task_sync( + dispatch = enqueue_legacy_task_sync(actor=ActorValue.SCHEDULER) + _set_state(app, "last_legacy_task_sync_dispatch", dispatch) + + def run_event_dispatch() -> None: + dispatch = enqueue_event_dispatch( + limit=settings.event_dispatch_batch_size, actor=ActorValue.SCHEDULER, ) + _set_state(app, "last_event_dispatch", dispatch) + + def record_scheduler_heartbeat() -> None: + db = SessionLocal() + try: + heartbeat = ObservabilityService(db).record_heartbeat( + component=HeartbeatComponent.SCHEDULER, + instance_id=gethostname(), + actor=ActorValue.SCHEDULER, + ) + _set_state(app, "last_scheduler_heartbeat", heartbeat) + finally: + db.close() scheduler.add_job( run_daily_brief, @@ -104,6 +144,21 @@ def attach_scheduler(app: FastAPI) -> None: id="project_weekly_push", replace_existing=True, ) + if settings.event_dispatch_enabled: + scheduler.add_job( + run_event_dispatch, + trigger="cron", + minute=settings.event_dispatch_cron_minute, + id="event_dispatch", + replace_existing=True, + ) + scheduler.add_job( + record_scheduler_heartbeat, + trigger="interval", + seconds=settings.heartbeat_interval_seconds, + id="scheduler_heartbeat", + replace_existing=True, + ) if settings.legacy_sync_enabled and settings.legacy_project_query: scheduler.add_job( run_legacy_project_sync, @@ -122,11 +177,9 @@ def attach_scheduler(app: FastAPI) -> None: id="legacy_task_sync", replace_existing=True, ) + return scheduler - @app.on_event("startup") - def start_scheduler() -> None: - scheduler.start() - @app.on_event("shutdown") - def stop_scheduler() -> None: - scheduler.shutdown(wait=False) +def _set_state(app: FastAPI | None, key: str, value: Any) -> None: + if app is not None: + setattr(app.state, key, value) diff --git a/app/core/task_queue.py b/app/core/task_queue.py index dbeac2f..a391a02 100644 --- a/app/core/task_queue.py +++ b/app/core/task_queue.py @@ -9,6 +9,7 @@ TASK_PUSH_PROJECT_WEEKLY = "reports.push_project_weekly" TASK_GENERATE_RISK_EVENTS = "risks.generate_events" TASK_SYNC_LEGACY_PROJECTS = "legacy.sync_projects" TASK_SYNC_LEGACY_TASKS = "legacy.sync_tasks" +TASK_DISPATCH_PENDING_EVENTS = "events.dispatch_pending" def dispatch_task( @@ -191,6 +192,31 @@ def enqueue_risk_event_generation(actor: str = "scheduler") -> dict[str, Any]: ) +def enqueue_event_dispatch( + limit: int | None = None, + actor: str = "scheduler", +) -> dict[str, Any]: + def inline() -> Any: + from app.core.config import get_settings + from app.core.database import SessionLocal + from app.modules.events.service import EventService + + db = SessionLocal() + try: + return EventService(db).dispatch_pending( + limit=limit or get_settings().event_dispatch_batch_size, + worker_id=actor, + ) + finally: + db.close() + + return dispatch_task( + TASK_DISPATCH_PENDING_EVENTS, + {"limit": limit, "actor": actor}, + inline, + ) + + def enqueue_legacy_project_sync( source_query: str | None = None, source_query_name: str | None = None, diff --git a/app/modules/ai_agent/constants.py b/app/modules/ai_agent/constants.py index bf54fe3..5ffe134 100644 --- a/app/modules/ai_agent/constants.py +++ b/app/modules/ai_agent/constants.py @@ -29,6 +29,8 @@ class AIResponseKey(StrEnum): TEXT = "text" PIPELINE = "pipeline" HERMES_RECALL = "hermes_recall" + LOCAL_MEMORY = "local_memory" + MEMORY_WRITE = "memory_write" HERMES_ANSWER = "hermes_answer" HERMES_REMEMBER = "hermes_remember" OPENCLAW = "openclaw" @@ -49,6 +51,9 @@ class AIContextKey(StrEnum): OPENCLAW_SESSION_KEY = "openclaw_session_key" AGENT_PIPELINE = "agent_pipeline" HERMES_MEMORY = "hermes_memory" + LOCAL_MEMORY = "local_memory" + MEMORY_SCOPE = "memory_scope" + MEMORY_SUBJECT = "memory_subject" OPENCLAW = "openclaw" MODE = "mode" USER_PROMPT = "user_prompt" diff --git a/app/modules/ai_agent/service.py b/app/modules/ai_agent/service.py index 54aa2f6..f6bbd76 100644 --- a/app/modules/ai_agent/service.py +++ b/app/modules/ai_agent/service.py @@ -14,10 +14,13 @@ from app.modules.ai_agent.constants import ( AI_AUDIT_SENSITIVE_KEYS, AI_AUDIT_TRUNCATED_VALUE, AIToolAuditKey, + AIContextKey, AIProviderName, AIRequestKey, AIResponseKey, ) +from app.modules.ai_memory.constants import AIMemoryPayloadKey, AIMemoryScope +from app.modules.ai_memory.service import AIMemoryService from app.modules.ai_agent.skills import AISkillId, get_ai_skill from app.modules.audit.constants import ( AuditAction, @@ -44,11 +47,37 @@ class AIService: source: str = AuditSource.API, ) -> dict[str, Any]: adapter = get_adapter() - result = adapter.ask(prompt, context or {}) + original_context = context or {} + adapter_context = dict(original_context) + memory_service = AIMemoryService(self.db) + local_memory = memory_service.recall( + query=prompt, + scope=_memory_scope(original_context), + subject=_memory_subject(original_context), + actor=actor, + ) + if local_memory: + adapter_context[AIContextKey.LOCAL_MEMORY] = local_memory + result = adapter.ask(prompt, adapter_context) + answer = result[AIResponseKey.ANSWER] + raw = dict(result.get(AIResponseKey.RAW, {})) + if local_memory: + raw[AIResponseKey.LOCAL_MEMORY] = local_memory + memory_record = memory_service.auto_write( + prompt=prompt, + context=original_context, + answer=answer, + actor=actor, + ) + if memory_record is not None: + raw[AIResponseKey.MEMORY_WRITE] = { + AIMemoryPayloadKey.CODE: memory_record.code, + AIMemoryPayloadKey.STATUS: memory_record.status, + } response = { AIResponseKey.PROVIDER: adapter.provider_name, - AIResponseKey.ANSWER: result[AIResponseKey.ANSWER], - AIResponseKey.RAW: result.get(AIResponseKey.RAW, {}), + AIResponseKey.ANSWER: answer, + AIResponseKey.RAW: raw, } self.audit.log( AuditLogCreate( @@ -197,3 +226,16 @@ def _audit_safe_payload(value: Any, depth: int = 0) -> Any: if isinstance(value, str) and len(value) > AI_AUDIT_MAX_TEXT_LENGTH: return value[:AI_AUDIT_MAX_TEXT_LENGTH] + AI_AUDIT_TRUNCATED_VALUE return value + + +def _memory_scope(context: dict[str, Any]) -> str: + return str( + context.get(AIContextKey.MEMORY_SCOPE) + or context.get(AIMemoryPayloadKey.SCOPE) + or AIMemoryScope.GLOBAL + ) + + +def _memory_subject(context: dict[str, Any]) -> str | None: + value = context.get(AIContextKey.MEMORY_SUBJECT) or context.get(AIMemoryPayloadKey.SUBJECT) + return str(value) if value else None diff --git a/app/modules/ai_memory/__init__.py b/app/modules/ai_memory/__init__.py new file mode 100644 index 0000000..8b13789 --- /dev/null +++ b/app/modules/ai_memory/__init__.py @@ -0,0 +1 @@ + diff --git a/app/modules/ai_memory/constants.py b/app/modules/ai_memory/constants.py new file mode 100644 index 0000000..042d6fc --- /dev/null +++ b/app/modules/ai_memory/constants.py @@ -0,0 +1,55 @@ +from enum import StrEnum + + +class AIMemoryScope(StrEnum): + GLOBAL = "global" + PROJECT = "project" + DEPARTMENT = "department" + USER = "user" + + +class AIMemoryStatus(StrEnum): + ACTIVE = "active" + ARCHIVED = "archived" + REJECTED = "rejected" + + +class AIMemorySource(StrEnum): + AUTO = "auto" + HERMES = "hermes" + API = "api" + + +class AIMemoryResponseKey(StrEnum): + ITEMS = "items" + DATA = "data" + TOTAL = "total" + + +class AIMemoryPayloadKey(StrEnum): + CODE = "code" + SCOPE = "scope" + SUBJECT = "subject" + CONTENT = "content" + SUMMARY = "summary" + TAGS = "tags" + SOURCE = "source" + IMPORTANCE = "importance" + STATUS = "status" + QUERY = "query" + LIMIT = "limit" + COUNT = "count" + REJECTED_REASON = "rejected_reason" + + +class AIMemoryText(StrEnum): + DEFAULT_SCOPE = "global" + DEFAULT_SUBJECT = "company" + AUTO_TAG = "auto" + REJECTED_SECRET = "secret-like content rejected" + + +AI_MEMORY_CODE_PREFIX = "MEM" +AI_MEMORY_MAX_CONTENT_LENGTH = 2000 +AI_MEMORY_MAX_SUMMARY_LENGTH = 500 +AI_MEMORY_MIN_AUTO_WRITE_LENGTH = 12 diff --git a/app/modules/ai_memory/models.py b/app/modules/ai_memory/models.py new file mode 100644 index 0000000..e63d707 --- /dev/null +++ b/app/modules/ai_memory/models.py @@ -0,0 +1,33 @@ +from datetime import datetime + +from sqlalchemy import JSON, DateTime, Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.constants import ActorValue +from app.core.db_base import Base +from app.core.time import utc_now +from app.modules.ai_memory.constants import AIMemoryScope, AIMemorySource, AIMemoryStatus + + +class AIMemoryEntry(Base): + __tablename__ = "ai_memory_entries" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + code: Mapped[str] = mapped_column(String(64), unique=True, index=True) + scope: Mapped[str] = mapped_column(String(64), default=AIMemoryScope.GLOBAL, index=True) + subject: Mapped[str] = mapped_column(String(128), index=True) + content: Mapped[str] = mapped_column(Text) + summary: Mapped[str | None] = mapped_column(Text, nullable=True) + tags: Mapped[list | None] = mapped_column(JSON, nullable=True) + source: Mapped[str] = mapped_column(String(64), default=AIMemorySource.AUTO, index=True) + importance: Mapped[int] = mapped_column(Integer, default=1, index=True) + status: Mapped[str] = mapped_column(String(32), default=AIMemoryStatus.ACTIVE, index=True) + actor: Mapped[str] = mapped_column(String(128), default=ActorValue.SYSTEM, index=True) + last_used_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, index=True) + expires_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, index=True) + created_at: Mapped[datetime] = mapped_column(DateTime, default=utc_now, index=True) + updated_at: Mapped[datetime] = mapped_column( + DateTime, + default=utc_now, + onupdate=utc_now, + ) diff --git a/app/modules/ai_memory/routes.py b/app/modules/ai_memory/routes.py new file mode 100644 index 0000000..ae121bb --- /dev/null +++ b/app/modules/ai_memory/routes.py @@ -0,0 +1,44 @@ +from fastapi import APIRouter, Depends, Query +from sqlalchemy.orm import Session + +from app.core.database import get_db +from app.core.security import ApiPrincipal, require_api_key +from app.modules.ai_memory.constants import AIMemoryResponseKey, AIMemoryStatus +from app.modules.ai_memory.schemas import AIMemoryRecallRequest +from app.modules.ai_memory.service import AIMemoryService + +router = APIRouter(dependencies=[Depends(require_api_key)]) + + +@router.get("/memory") +def list_memory( + scope: str | None = None, + subject: str | None = None, + status: str = AIMemoryStatus.ACTIVE, + limit: int = Query(default=100, ge=1, le=500), + db: Session = Depends(get_db), +) -> dict: + return { + AIMemoryResponseKey.ITEMS: AIMemoryService(db).list_entries( + scope=scope, + subject=subject, + status_filter=status, + limit=limit, + ) + } + + +@router.post("/memory/recall") +def recall_memory( + payload: AIMemoryRecallRequest, + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + items = AIMemoryService(db).recall( + query=payload.query, + scope=payload.scope, + subject=payload.subject, + limit=payload.limit, + actor=principal.actor, + ) + return {AIMemoryResponseKey.ITEMS: items} diff --git a/app/modules/ai_memory/schemas.py b/app/modules/ai_memory/schemas.py new file mode 100644 index 0000000..9baf64f --- /dev/null +++ b/app/modules/ai_memory/schemas.py @@ -0,0 +1,29 @@ +from typing import Any + +from pydantic import BaseModel, Field + +from app.modules.ai_memory.constants import AIMemoryScope + + +class AIMemoryRecallRequest(BaseModel): + query: str = Field(..., min_length=1) + scope: str = AIMemoryScope.GLOBAL + subject: str | None = None + limit: int = Field(default=5, ge=1, le=50) + + +class AIMemoryRead(BaseModel): + code: str + scope: str + subject: str + content: str + summary: str | None + tags: list[Any] | None + source: str + importance: int + status: str + actor: str + last_used_at: str | None + expires_at: str | None + created_at: str + updated_at: str diff --git a/app/modules/ai_memory/service.py b/app/modules/ai_memory/service.py new file mode 100644 index 0000000..f6690a7 --- /dev/null +++ b/app/modules/ai_memory/service.py @@ -0,0 +1,277 @@ +from typing import Any + +from sqlalchemy import func, or_, select +from sqlalchemy.orm import Session + +from app.core.config import get_settings +from app.core.constants import ActorValue +from app.core.pagination import bounded_limit +from app.core.time import utc_now +from app.modules.ai_memory.constants import ( + AI_MEMORY_CODE_PREFIX, + AI_MEMORY_MAX_CONTENT_LENGTH, + AI_MEMORY_MAX_SUMMARY_LENGTH, + AI_MEMORY_MIN_AUTO_WRITE_LENGTH, + AIMemoryPayloadKey, + AIMemoryScope, + AIMemorySource, + AIMemoryStatus, + AIMemoryText, +) +from app.modules.ai_memory.models import AIMemoryEntry +from app.modules.audit.constants import ( + AuditAction, + AuditRiskLevel, + AuditSource, + AuditTargetType, +) +from app.modules.audit.schemas import AuditLogCreate +from app.modules.audit.service import AuditService +from app.modules.business.service import serialize_model +from app.modules.events.constants import EventAggregateType, EventSource, EventType +from app.modules.events.service import EventService + + +class AIMemoryService: + """Store and recall audited local AI memory for read-only operations.""" + + def __init__(self, db: Session): + self.db = db + self.audit = AuditService(db) + + def list_entries( + self, + scope: str | None = None, + subject: str | None = None, + status_filter: str = AIMemoryStatus.ACTIVE, + limit: int = 100, + ) -> list[dict[str, Any]]: + stmt = ( + select(AIMemoryEntry) + .where(AIMemoryEntry.status == status_filter) + .order_by(AIMemoryEntry.importance.desc(), AIMemoryEntry.id.desc()) + .limit(bounded_limit(limit)) + ) + if scope: + stmt = stmt.where(AIMemoryEntry.scope == scope) + if subject: + stmt = stmt.where(AIMemoryEntry.subject == subject) + return [serialize_model(item) for item in self.db.execute(stmt).scalars()] + + def recall( + self, + query: str, + scope: str = AIMemoryScope.GLOBAL, + subject: str | None = None, + limit: int | None = None, + actor: str = ActorValue.API, + ) -> list[dict[str, Any]]: + settings = get_settings() + if not settings.ai_memory_enabled: + return [] + limit_value = bounded_limit(limit or settings.ai_memory_recall_limit) + now = utc_now() + stmt = ( + select(AIMemoryEntry) + .where( + AIMemoryEntry.status == AIMemoryStatus.ACTIVE, + or_(AIMemoryEntry.expires_at.is_(None), AIMemoryEntry.expires_at > now), + AIMemoryEntry.scope.in_({AIMemoryScope.GLOBAL, scope}), + ) + .order_by(AIMemoryEntry.importance.desc(), AIMemoryEntry.id.desc()) + .limit(limit_value * 3) + ) + if subject: + stmt = stmt.where( + or_( + AIMemoryEntry.subject == subject, + AIMemoryEntry.scope == AIMemoryScope.GLOBAL, + ) + ) + candidates = list(self.db.execute(stmt).scalars()) + items = [item for item in candidates if _matches_query(item, query)] + if not items: + items = candidates[:limit_value] + items = items[:limit_value] + for item in items: + item.last_used_at = now + self.db.commit() + result = [serialize_model(item) for item in items] + self.audit.log( + AuditLogCreate( + actor=actor, + source=AuditSource.AI_MEMORY, + action=AuditAction.AI_MEMORY_RECALL, + target_type=AuditTargetType.AI_MEMORY, + risk_level=AuditRiskLevel.LOW, + request_payload={ + AIMemoryPayloadKey.QUERY: query, + AIMemoryPayloadKey.SCOPE: scope, + AIMemoryPayloadKey.SUBJECT: subject, + AIMemoryPayloadKey.LIMIT: limit_value, + }, + response_payload={AIMemoryPayloadKey.COUNT: len(result)}, + ) + ) + return result + + def auto_write( + self, + prompt: str, + context: dict[str, Any], + answer: str, + actor: str = ActorValue.API, + ) -> AIMemoryEntry | None: + settings = get_settings() + if not settings.ai_memory_enabled or not settings.ai_memory_auto_write_enabled: + return None + content = _build_memory_content(prompt, context, answer) + if len(content) < AI_MEMORY_MIN_AUTO_WRITE_LENGTH: + return None + scope = str(context.get(AIMemoryPayloadKey.SCOPE) or AIMemoryText.DEFAULT_SCOPE) + subject = str(context.get(AIMemoryPayloadKey.SUBJECT) or AIMemoryText.DEFAULT_SUBJECT) + if _contains_forbidden_value( + { + "prompt": prompt, + "context": context, + "answer": answer, + }, + settings.ai_memory_forbidden_keys, + ): + record = self._create_entry( + scope=scope, + subject=subject, + content=str(AIMemoryText.REJECTED_SECRET), + summary=str(AIMemoryText.REJECTED_SECRET), + tags=[str(AIMemoryText.AUTO_TAG)], + source=AIMemorySource.AUTO, + importance=0, + status_value=AIMemoryStatus.REJECTED, + actor=actor, + ) + return record + summary = _truncate(answer, AI_MEMORY_MAX_SUMMARY_LENGTH) + record = self._create_entry( + scope=scope, + subject=subject, + content=_truncate(content, AI_MEMORY_MAX_CONTENT_LENGTH), + summary=summary, + tags=[str(AIMemoryText.AUTO_TAG)], + source=AIMemorySource.AUTO, + importance=1, + status_value=AIMemoryStatus.ACTIVE, + actor=actor, + ) + return record + + def count_by_status(self) -> dict[str, int]: + rows = self.db.execute( + select(AIMemoryEntry.status, func.count()).group_by(AIMemoryEntry.status) + ).all() + return {str(status_value): int(count) for status_value, count in rows} + + def _create_entry( + self, + scope: str, + subject: str, + content: str, + summary: str | None, + tags: list[str], + source: str, + importance: int, + status_value: str, + actor: str, + ) -> AIMemoryEntry: + record = AIMemoryEntry( + code=f"{AI_MEMORY_CODE_PREFIX}-{utc_now():%Y%m%d%H%M%S%f}", + scope=scope, + subject=subject, + content=content, + summary=summary, + tags=tags, + source=source, + importance=importance, + status=status_value, + actor=actor, + ) + self.db.add(record) + self.db.commit() + self.db.refresh(record) + self.audit.log( + AuditLogCreate( + actor=actor, + source=AuditSource.AI_MEMORY, + action=AuditAction.AI_MEMORY_WRITE, + target_type=AuditTargetType.AI_MEMORY, + target_id=record.code, + risk_level=AuditRiskLevel.LOW, + request_payload={ + AIMemoryPayloadKey.SCOPE: scope, + AIMemoryPayloadKey.SUBJECT: subject, + AIMemoryPayloadKey.SOURCE: source, + AIMemoryPayloadKey.STATUS: status_value, + }, + response_payload={AIMemoryPayloadKey.CODE: record.code}, + ) + ) + EventService(self.db).emit( + event_type=EventType.AI_MEMORY_WRITTEN, + source=EventSource.AI_MEMORY, + aggregate_type=EventAggregateType.AI_MEMORY_ENTRY, + aggregate_id=record.code, + actor=actor, + payload={ + AIMemoryPayloadKey.CODE: record.code, + AIMemoryPayloadKey.SCOPE: scope, + AIMemoryPayloadKey.SUBJECT: subject, + AIMemoryPayloadKey.STATUS: status_value, + }, + idempotency_key=f"ai-memory:{record.code}", + dispatch=True, + ) + return record + + +def _build_memory_content(prompt: str, context: dict[str, Any], answer: str) -> str: + context_text = ", ".join( + f"{key}={value}" for key, value in sorted(context.items(), key=lambda item: str(item[0])) + ) + return f"prompt: {prompt}\ncontext: {context_text}\nanswer: {answer}" + + +def _contains_forbidden_value(value: Any, forbidden_keys: list[str]) -> bool: + forbidden = {item.lower() for item in forbidden_keys} + if isinstance(value, dict): + for key, item in value.items(): + if str(key).lower() in forbidden: + return True + if _contains_forbidden_value(item, forbidden_keys): + return True + return False + if isinstance(value, (list, tuple, set)): + return any(_contains_forbidden_value(item, forbidden_keys) for item in value) + if isinstance(value, str): + lowered = value.lower() + return any(item in lowered for item in forbidden) + return False + + +def _matches_query(entry: AIMemoryEntry, query: str) -> bool: + query_text = query.lower().strip() + if not query_text: + return True + text = " ".join( + [ + entry.subject or "", + entry.content or "", + entry.summary or "", + " ".join(str(item) for item in (entry.tags or [])), + ] + ).lower() + return any(token in text for token in query_text.split()) + + +def _truncate(value: str, max_length: int) -> str: + if len(value) <= max_length: + return value + return value[:max_length] diff --git a/app/modules/audit/constants.py b/app/modules/audit/constants.py index 3545850..84930fd 100644 --- a/app/modules/audit/constants.py +++ b/app/modules/audit/constants.py @@ -14,6 +14,11 @@ class AuditAction(StrEnum): LEGACY_SYNC_TASKS = "sync_tasks" RISK_EVENT_ACTION = "risk_event_action" REPORT_PUSH = "report_push" + AI_MEMORY_RECALL = "ai.memory_recall" + AI_MEMORY_WRITE = "ai.memory_write" + ENTERPRISE_ANALYTICS = "enterprise_analytics" + EVENT_DISPATCH = "event.dispatch" + HEARTBEAT = "heartbeat" class AuditRiskLevel(StrEnum): @@ -29,6 +34,9 @@ class AuditSource(StrEnum): FEISHU = "feishu" LEGACY_MYSQL = "legacy_mysql" REPORTS = "reports" + EVENTS = "events" + AI_MEMORY = "ai_memory" + OBSERVABILITY = "observability" class AuditTargetType(StrEnum): @@ -36,6 +44,10 @@ class AuditTargetType(StrEnum): OPENCLAW_TOOL = "openclaw_tool" RISK_EVENTS = "risk-events" WORK_REPORTS = "work-reports" + ENTERPRISE_ANALYTICS = "enterprise-analytics" + AI_MEMORY = "ai-memory" + DOMAIN_EVENT = "domain-event" + HEARTBEAT = "heartbeat" class AuditStatus(StrEnum): diff --git a/app/modules/dashboard/service.py b/app/modules/dashboard/service.py index 1dfead2..8ffe0a4 100644 --- a/app/modules/dashboard/service.py +++ b/app/modules/dashboard/service.py @@ -3,6 +3,8 @@ from typing import Any from sqlalchemy import func, select from sqlalchemy.orm import Session +from app.modules.ai_memory.constants import AIMemoryStatus +from app.modules.ai_memory.models import AIMemoryEntry from app.modules.audit.models import AuditLog from app.modules.business.constants import DONE_STATUSES, PROJECT_CLOSED_STATUSES, StatusValue from app.modules.business.models import ( @@ -16,6 +18,8 @@ from app.modules.business.models import ( from app.modules.business.service import serialize_model from app.modules.events.constants import EventStatus from app.modules.events.models import DomainEvent +from app.modules.observability.constants import ObservabilityMetricKey +from app.modules.observability.service import ObservabilityService from app.modules.reports.constants import ReportPushStatus from app.modules.risk.service import RiskService from app.modules.workflows.constants import WorkflowStatus @@ -23,7 +27,7 @@ from app.modules.workflows.models import WorkflowInstance class DashboardService: - """Build lightweight operational dashboard data for V2.""" + """Build lightweight operational dashboard data for V2/V3.""" def __init__(self, db: Session): self.db = db @@ -49,6 +53,8 @@ class DashboardService: WorkflowInstance, WorkflowInstance.status == WorkflowStatus.FAILED, ) + active_ai_memory = self._count(AIMemoryEntry, AIMemoryEntry.status == AIMemoryStatus.ACTIVE) + heartbeat_summary = ObservabilityService(self.db).heartbeat_summary() latest_reports = self.db.execute( select(WorkReport).order_by(WorkReport.id.desc()).limit(5) ).scalars() @@ -73,6 +79,8 @@ class DashboardService: "failed_events": failed_events, "running_workflows": running_workflows, "failed_workflows": failed_workflows, + "active_ai_memory": active_ai_memory, + "stale_heartbeats": heartbeat_summary[ObservabilityMetricKey.STALE], "risk_level": risk_summary["risk_level"], "risk_score": float(risk_summary["risk_score"]), }, diff --git a/app/modules/events/constants.py b/app/modules/events/constants.py index 8530215..ac3976f 100644 --- a/app/modules/events/constants.py +++ b/app/modules/events/constants.py @@ -11,20 +11,29 @@ class EventType(StrEnum): RISK_ACTION_RECORDED = "risk.action_recorded" REPORT_PUSH_SUCCEEDED = "report.push_succeeded" REPORT_PUSH_FAILED = "report.push_failed" + REPORT_GENERATED = "report.generated" LEGACY_SYNC_COMPLETED = "legacy.sync_completed" + LEGACY_SYNC_FAILED = "legacy.sync_failed" + AI_MEMORY_WRITTEN = "ai.memory_written" + ENTERPRISE_ANALYTICS_GENERATED = "enterprise.analytics_generated" class EventSource(StrEnum): RISK = "risk" REPORTS = "reports" LEGACY_MYSQL = "legacy_mysql" + AI_MEMORY = "ai_memory" + ANALYTICS = "analytics" API = "api" class EventAggregateType(StrEnum): RISK_EVENT = "risk-event" REPORT_PUSH_RUN = "report-push-run" + WORK_REPORT = "work-report" LEGACY_SYNC_RUN = "legacy-sync-run" + AI_MEMORY_ENTRY = "ai-memory-entry" + ENTERPRISE_ANALYTICS = "enterprise-analytics" class EventResponseKey(StrEnum): @@ -44,10 +53,13 @@ class EventPayloadKey(StrEnum): UPDATED = "updated" SKIPPED = "skipped" ERROR_MESSAGE = "error_message" + ATTEMPTS = "attempts" + HANDLED = "handled" class EventErrorDetail(StrEnum): EVENT_NOT_FOUND = "Domain event not found" + EVENT_NOT_RETRYABLE = "Domain event is not retryable" EVENT_CODE_PREFIX = "EVT" diff --git a/app/modules/events/models.py b/app/modules/events/models.py index 25e4111..b820681 100644 --- a/app/modules/events/models.py +++ b/app/modules/events/models.py @@ -29,5 +29,13 @@ class DomainEvent(Base): index=True, ) last_error: Mapped[str | None] = mapped_column(Text, nullable=True) + next_attempt_at: Mapped[datetime | None] = mapped_column( + DateTime, + nullable=True, + index=True, + ) + locked_until: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, index=True) + locked_by: Mapped[str | None] = mapped_column(String(128), nullable=True, index=True) + max_attempts: Mapped[int] = mapped_column(Integer, default=3) created_at: Mapped[datetime] = mapped_column(DateTime, default=utc_now, index=True) processed_at: Mapped[datetime | None] = mapped_column(DateTime, nullable=True, index=True) diff --git a/app/modules/events/routes.py b/app/modules/events/routes.py index 1e71294..54188d4 100644 --- a/app/modules/events/routes.py +++ b/app/modules/events/routes.py @@ -2,8 +2,7 @@ from fastapi import APIRouter, Depends, Query from sqlalchemy.orm import Session from app.core.database import get_db -from app.core.operation_guard import require_operations_enabled -from app.core.security import require_api_key +from app.core.security import ApiPrincipal, require_api_key from app.modules.events.constants import EventResponseKey from app.modules.events.service import EventService, _serialize_event @@ -31,14 +30,25 @@ def dispatch_event( event_id: str, db: Session = Depends(get_db), ) -> dict: - require_operations_enabled() return {EventResponseKey.EVENT: _serialize_event(EventService(db).dispatch_event(event_id))} +@router.post("/{event_id}/retry") +def retry_event( + event_id: str, + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + return { + EventResponseKey.EVENT: _serialize_event( + EventService(db).retry_event(event_id, actor=principal.actor) + ) + } + + @router.post("/dispatch-pending") def dispatch_pending( limit: int = Query(default=100, ge=1, le=500), db: Session = Depends(get_db), ) -> dict: - require_operations_enabled() return {EventResponseKey.ITEMS: EventService(db).dispatch_pending(limit=limit)} diff --git a/app/modules/events/service.py b/app/modules/events/service.py index bc6b0bf..9f9face 100644 --- a/app/modules/events/service.py +++ b/app/modules/events/service.py @@ -1,12 +1,23 @@ +from datetime import timedelta from typing import Any +from uuid import uuid4 from fastapi import HTTPException, status -from sqlalchemy import func, select +from sqlalchemy import func, or_, select from sqlalchemy.orm import Session +from app.core.config import get_settings from app.core.constants import ActorValue from app.core.pagination import bounded_limit from app.core.time import utc_now +from app.modules.audit.constants import ( + AuditAction, + AuditRiskLevel, + AuditSource, + AuditTargetType, +) +from app.modules.audit.schemas import AuditLogCreate +from app.modules.audit.service import AuditService from app.modules.events.constants import ( EVENT_CODE_PREFIX, EventAggregateType, @@ -51,8 +62,10 @@ class EventService: return self.dispatch_event(existing.event_id) return existing + settings = get_settings() + now = utc_now() record = DomainEvent( - event_id=f"{EVENT_CODE_PREFIX}-{utc_now():%Y%m%d%H%M%S%f}", + event_id=f"{EVENT_CODE_PREFIX}-{now:%Y%m%d%H%M%S%f}", event_type=event_type, source=source, aggregate_type=aggregate_type, @@ -60,6 +73,8 @@ class EventService: actor=actor, payload=payload or {}, idempotency_key=idempotency_key, + next_attempt_at=now, + max_attempts=settings.event_dispatch_max_attempts, ) self.db.add(record) self.db.commit() @@ -98,39 +113,120 @@ class EventService: ) return record - def dispatch_event(self, event_id: str) -> DomainEvent: + def dispatch_event(self, event_id: str, worker_id: str | None = None) -> DomainEvent: record = self.get_event(event_id) if record.status == EventStatus.PROCESSED: return record + if not self._can_attempt(record): + return record + settings = get_settings() + now = utc_now() + lock_owner = worker_id or f"api:{uuid4().hex}" + record.locked_by = lock_owner + record.locked_until = now + timedelta(seconds=settings.event_dispatch_lock_seconds) + record.status = EventStatus.PENDING record.attempts += 1 try: self._handle_event(record) except Exception as exc: - record.status = EventStatus.FAILED + retryable = record.attempts < self._max_attempts(record) + record.status = EventStatus.PENDING if retryable else EventStatus.FAILED record.last_error = str(exc) + record.next_attempt_at = ( + utc_now() + timedelta(seconds=settings.event_dispatch_retry_delay_seconds) + if retryable + else None + ) + record.locked_by = None + record.locked_until = None self.db.commit() self.db.refresh(record) + self._audit_dispatch(record) return record record.status = EventStatus.PROCESSED record.last_error = None record.processed_at = utc_now() + record.next_attempt_at = None + record.locked_by = None + record.locked_until = None self.db.commit() self.db.refresh(record) + self._audit_dispatch(record) return record - def dispatch_pending(self, limit: int = 100) -> list[dict[str, Any]]: + def dispatch_pending( + self, + limit: int = 100, + worker_id: str | None = None, + ) -> list[dict[str, Any]]: + now = utc_now() stmt = ( select(DomainEvent) - .where(DomainEvent.status == EventStatus.PENDING) + .where( + DomainEvent.status == EventStatus.PENDING, + or_( + DomainEvent.next_attempt_at.is_(None), + DomainEvent.next_attempt_at <= now, + ), + or_( + DomainEvent.locked_until.is_(None), + DomainEvent.locked_until <= now, + ), + or_( + DomainEvent.max_attempts.is_(None), + DomainEvent.attempts < DomainEvent.max_attempts, + ), + ) .order_by(DomainEvent.id.asc()) .limit(bounded_limit(limit)) ) records = list(self.db.execute(stmt).scalars()) - return [_serialize_event(self.dispatch_event(record.event_id)) for record in records] + lock_owner = worker_id or f"worker:{uuid4().hex}" + return [ + _serialize_event(self.dispatch_event(record.event_id, worker_id=lock_owner)) + for record in records + ] + + def retry_event(self, event_id: str, actor: str = ActorValue.API) -> DomainEvent: + record = self.get_event(event_id) + if record.status == EventStatus.PROCESSED or not self._can_attempt(record): + raise HTTPException( + status_code=status.HTTP_409_CONFLICT, + detail=EventErrorDetail.EVENT_NOT_RETRYABLE, + ) + record.status = EventStatus.PENDING + record.actor = actor + record.last_error = None + record.locked_by = None + record.locked_until = None + record.next_attempt_at = utc_now() + self.db.commit() + self.db.refresh(record) + return record def _handle_event(self, record: DomainEvent) -> None: if record.event_type == EventType.RISK_ACTION_RECORDED: self._handle_risk_action(record) + return + if record.event_type in { + EventType.REPORT_PUSH_SUCCEEDED, + EventType.REPORT_PUSH_FAILED, + EventType.REPORT_GENERATED, + }: + self._handle_report_event(record) + return + if record.event_type in { + EventType.LEGACY_SYNC_COMPLETED, + EventType.LEGACY_SYNC_FAILED, + }: + self._handle_legacy_sync_event(record) + return + if record.event_type == EventType.AI_MEMORY_WRITTEN: + self._handle_ai_memory_event(record) + return + if record.event_type == EventType.ENTERPRISE_ANALYTICS_GENERATED: + self._handle_enterprise_analytics_event(record) + return def _handle_risk_action(self, record: DomainEvent) -> None: from app.modules.risk.constants import RiskEventActionValue @@ -154,3 +250,98 @@ class EventService: actor=record.actor, payload=payload, ) + + def _handle_report_event(self, record: DomainEvent) -> None: + from app.modules.workflows.constants import WorkflowStatus, WorkflowType + + workflow_status = ( + WorkflowStatus.FAILED + if record.event_type == EventType.REPORT_PUSH_FAILED + else WorkflowStatus.COMPLETED + ) + self._track_operational_workflow( + record, + workflow_type=WorkflowType.REPORT_DELIVERY, + workflow_status=workflow_status, + ) + + def _handle_legacy_sync_event(self, record: DomainEvent) -> None: + from app.modules.workflows.constants import WorkflowStatus, WorkflowType + + workflow_status = ( + WorkflowStatus.FAILED + if record.event_type == EventType.LEGACY_SYNC_FAILED + else WorkflowStatus.COMPLETED + ) + self._track_operational_workflow( + record, + workflow_type=WorkflowType.LEGACY_SYNC_MONITOR, + workflow_status=workflow_status, + ) + + def _handle_ai_memory_event(self, record: DomainEvent) -> None: + from app.modules.ai_memory.constants import AIMemoryPayloadKey, AIMemoryStatus + from app.modules.workflows.constants import WorkflowStatus, WorkflowType + + payload = record.payload or {} + workflow_status = ( + WorkflowStatus.BLOCKED + if payload.get(AIMemoryPayloadKey.STATUS) == AIMemoryStatus.REJECTED + else WorkflowStatus.COMPLETED + ) + self._track_operational_workflow( + record, + workflow_type=WorkflowType.AI_MEMORY_CAPTURE, + workflow_status=workflow_status, + ) + + def _handle_enterprise_analytics_event(self, record: DomainEvent) -> None: + from app.modules.workflows.constants import WorkflowStatus, WorkflowType + + self._track_operational_workflow( + record, + workflow_type=WorkflowType.ENTERPRISE_ANALYTICS, + workflow_status=WorkflowStatus.COMPLETED, + ) + + def _track_operational_workflow( + self, + record: DomainEvent, + workflow_type: str, + workflow_status: str, + ) -> None: + from app.modules.workflows.service import WorkflowService + + WorkflowService(self.db).start_or_update( + workflow_type=workflow_type, + aggregate_type=record.aggregate_type, + aggregate_id=record.aggregate_id, + status_value=workflow_status, + action=record.event_type, + actor=record.actor, + payload=record.payload or {}, + ) + + def _audit_dispatch(self, record: DomainEvent) -> None: + AuditService(self.db).log( + AuditLogCreate( + actor=record.actor, + source=AuditSource.EVENTS, + action=AuditAction.EVENT_DISPATCH, + target_type=AuditTargetType.DOMAIN_EVENT, + target_id=record.event_id, + risk_level=AuditRiskLevel.LOW, + response_payload={ + EventPayloadKey.STATUS: record.status, + EventPayloadKey.ATTEMPTS: record.attempts, + EventPayloadKey.ERROR_MESSAGE: record.last_error, + }, + ) + ) + + def _can_attempt(self, record: DomainEvent) -> bool: + return record.attempts < self._max_attempts(record) + + @staticmethod + def _max_attempts(record: DomainEvent) -> int: + return record.max_attempts or get_settings().event_dispatch_max_attempts diff --git a/app/modules/observability/constants.py b/app/modules/observability/constants.py index 82aa4a7..16876ef 100644 --- a/app/modules/observability/constants.py +++ b/app/modules/observability/constants.py @@ -9,6 +9,10 @@ class ObservabilityKey(StrEnum): REDIS = "redis" EVENTS = "events" WORKFLOWS = "workflows" + AI_MEMORY = "ai_memory" + HEARTBEATS = "heartbeats" + SCHEDULER = "scheduler" + WORKER = "worker" class ObservabilityStatus(StrEnum): @@ -20,6 +24,23 @@ class ObservabilityStatus(StrEnum): class ObservabilityMetricKey(StrEnum): ERROR = "error" + ITEMS = "items" + COMPONENT = "component" + INSTANCE_ID = "instance_id" PENDING = "pending" FAILED = "failed" RUNNING = "running" + TOTAL = "total" + ACTIVE = "active" + STALE = "stale" + LAST_SEEN_AT = "last_seen_at" + + +class HeartbeatComponent(StrEnum): + API = "api" + SCHEDULER = "scheduler" + WORKER = "worker" + + +class HeartbeatStatus(StrEnum): + OK = "ok" diff --git a/app/modules/observability/models.py b/app/modules/observability/models.py new file mode 100644 index 0000000..609a910 --- /dev/null +++ b/app/modules/observability/models.py @@ -0,0 +1,23 @@ +from datetime import datetime + +from sqlalchemy import DateTime, Integer, String +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.db_base import Base +from app.core.time import utc_now + + +class SystemHeartbeat(Base): + __tablename__ = "system_heartbeats" + + id: Mapped[int] = mapped_column(Integer, primary_key=True, autoincrement=True) + component: Mapped[str] = mapped_column(String(128), index=True) + instance_id: Mapped[str] = mapped_column(String(128), index=True) + status: Mapped[str] = mapped_column(String(32), index=True) + last_seen_at: Mapped[datetime] = mapped_column(DateTime, default=utc_now, index=True) + created_at: Mapped[datetime] = mapped_column(DateTime, default=utc_now, index=True) + updated_at: Mapped[datetime] = mapped_column( + DateTime, + default=utc_now, + onupdate=utc_now, + ) diff --git a/app/modules/observability/service.py b/app/modules/observability/service.py index f85e65d..6a212d8 100644 --- a/app/modules/observability/service.py +++ b/app/modules/observability/service.py @@ -1,16 +1,30 @@ +from datetime import timedelta from typing import Any -from sqlalchemy import text +from sqlalchemy import select, text from sqlalchemy.orm import Session from app.core.config import get_settings +from app.core.constants import ActorValue +from app.core.time import utc_now +from app.modules.ai_memory.service import AIMemoryService +from app.modules.audit.constants import ( + AuditAction, + AuditRiskLevel, + AuditSource, + AuditTargetType, +) +from app.modules.audit.schemas import AuditLogCreate +from app.modules.audit.service import AuditService from app.modules.events.constants import EventStatus from app.modules.events.service import EventService from app.modules.observability.constants import ( + HeartbeatStatus, ObservabilityKey, ObservabilityMetricKey, ObservabilityStatus, ) +from app.modules.observability.models import SystemHeartbeat from app.modules.workflows.constants import WorkflowStatus from app.modules.workflows.service import WorkflowService @@ -30,6 +44,7 @@ class ObservabilityService: ObservabilityKey.REDIS: self._redis_check(), ObservabilityKey.EVENTS: self._events_check(), ObservabilityKey.WORKFLOWS: self._workflows_check(), + ObservabilityKey.HEARTBEATS: self._heartbeats_check(), } degraded = any( item[ObservabilityKey.STATUS] @@ -48,9 +63,75 @@ class ObservabilityService: ObservabilityKey.METRICS: { ObservabilityKey.EVENTS: EventService(self.db).count_by_status(), ObservabilityKey.WORKFLOWS: WorkflowService(self.db).count_by_status(), + ObservabilityKey.AI_MEMORY: AIMemoryService(self.db).count_by_status(), + ObservabilityKey.HEARTBEATS: self.heartbeat_summary(), } } + def record_heartbeat( + self, + component: str, + instance_id: str, + status_value: str = HeartbeatStatus.OK, + actor: str = ActorValue.SYSTEM, + ) -> dict[str, Any]: + now = utc_now() + record = self.db.execute( + select(SystemHeartbeat).where( + SystemHeartbeat.component == component, + SystemHeartbeat.instance_id == instance_id, + ) + ).scalar_one_or_none() + if record is None: + record = SystemHeartbeat( + component=component, + instance_id=instance_id, + status=status_value, + last_seen_at=now, + ) + self.db.add(record) + else: + record.status = status_value + record.last_seen_at = now + record.updated_at = now + self.db.commit() + self.db.refresh(record) + AuditService(self.db).log( + AuditLogCreate( + actor=actor, + source=AuditSource.OBSERVABILITY, + action=AuditAction.HEARTBEAT, + target_type=AuditTargetType.HEARTBEAT, + target_id=f"{component}:{instance_id}", + risk_level=AuditRiskLevel.LOW, + response_payload={ + ObservabilityMetricKey.COMPONENT: component, + ObservabilityMetricKey.INSTANCE_ID: instance_id, + ObservabilityKey.STATUS: status_value, + ObservabilityMetricKey.LAST_SEEN_AT: record.last_seen_at.isoformat(), + }, + ) + ) + return self._serialize_heartbeat(record) + + def heartbeat_summary(self) -> dict[str, Any]: + records = list(self.db.execute(select(SystemHeartbeat)).scalars()) + threshold = self._heartbeat_stale_threshold() + stale = [item for item in records if item.last_seen_at < threshold] + active = len(records) - len(stale) + last_seen_at = max((item.last_seen_at for item in records), default=None) + return { + ObservabilityMetricKey.TOTAL: len(records), + ObservabilityMetricKey.ACTIVE: active, + ObservabilityMetricKey.STALE: len(stale), + ObservabilityMetricKey.LAST_SEEN_AT: ( + last_seen_at.isoformat() if last_seen_at else None + ), + ObservabilityMetricKey.ITEMS: [ + self._serialize_heartbeat(item) for item in records + ], + } + def _database_check(self) -> dict[str, Any]: try: self.db.execute(text("select 1")).scalar() @@ -97,3 +178,34 @@ class ObservabilityService: ObservabilityMetricKey.RUNNING: counts.get(WorkflowStatus.RUNNING, 0), ObservabilityMetricKey.FAILED: failed, } + + def _heartbeats_check(self) -> dict[str, Any]: + summary = self.heartbeat_summary() + total = summary[ObservabilityMetricKey.TOTAL] + stale = summary[ObservabilityMetricKey.STALE] + if total == 0: + return {ObservabilityKey.STATUS: ObservabilityStatus.SKIPPED} + return { + ObservabilityKey.STATUS: ( + ObservabilityStatus.DEGRADED if stale else ObservabilityStatus.OK + ), + ObservabilityMetricKey.TOTAL: total, + ObservabilityMetricKey.STALE: stale, + ObservabilityMetricKey.LAST_SEEN_AT: summary[ + ObservabilityMetricKey.LAST_SEEN_AT + ], + } + + @staticmethod + def _serialize_heartbeat(record: SystemHeartbeat) -> dict[str, Any]: + return { + ObservabilityMetricKey.COMPONENT: record.component, + ObservabilityMetricKey.INSTANCE_ID: record.instance_id, + ObservabilityKey.STATUS: record.status, + ObservabilityMetricKey.LAST_SEEN_AT: record.last_seen_at.isoformat(), + } + + @staticmethod + def _heartbeat_stale_threshold() -> Any: + settings = get_settings() + return utc_now() - timedelta(seconds=settings.heartbeat_interval_seconds * 3) diff --git a/app/modules/reports/constants.py b/app/modules/reports/constants.py index ecd8157..42d06f3 100644 --- a/app/modules/reports/constants.py +++ b/app/modules/reports/constants.py @@ -13,6 +13,7 @@ class ReportTitle(StrEnum): WORK_DAILY = "经营日报" WORK_WEEKLY = "经营周报" PROJECT_LIFECYCLE = "项目全生命周期报告" + ENTERPRISE_ANALYTICS = "企业只读运营分析" class ReportStatus(StrEnum): @@ -70,6 +71,19 @@ class LifecycleResponseKey(StrEnum): PROJECT_LIFECYCLE_REPORT = "project_lifecycle_report" +class EnterpriseAnalyticsKey(StrEnum): + CODE = "code" + TITLE = "title" + FILTERS = "filters" + FINANCE = "finance" + PROCUREMENT = "procurement" + PERFORMANCE = "performance" + OPERATIONS = "operations" + RECOMMENDATIONS = "recommendations" + LINES = "lines" + CONTENT = "content" + + class ReportResponseKey(StrEnum): REPORT = "report" DATA = "data" @@ -142,6 +156,14 @@ class MetricKey(StrEnum): EVENTS_BY_LEVEL = "events_by_level" SCORE = "score" LEVEL = "level" + READINESS_SCORE = "readiness_score" + PAYMENT_EXPOSURE = "payment_exposure" + DELIVERY_RISK = "delivery_risk" + CONFIRMED = "confirmed" + CONFIRMED_RATE = "confirmed_rate" + AVERAGE_AUTO_SCORE = "average_auto_score" + AVERAGE_CONFIRMED_SCORE = "average_confirmed_score" + WEIGHT_TOTAL = "weight_total" class WorkReportMetricKey(StrEnum): diff --git a/app/modules/reports/routes.py b/app/modules/reports/routes.py index 0d19f9b..42f7da9 100644 --- a/app/modules/reports/routes.py +++ b/app/modules/reports/routes.py @@ -52,6 +52,24 @@ def project_lifecycle_report( ) +@router.get("/enterprise-analytics") +def enterprise_analytics( + project_code: str | None = None, + owner: str | None = None, + period_start: date | None = None, + period_end: date | None = None, + db: Session = Depends(get_db), + principal: ApiPrincipal = Depends(require_api_key), +) -> dict: + return ReportService(db).enterprise_analytics( + project_code=project_code, + owner=owner, + period_start=period_start, + period_end=period_end, + actor=principal.actor, + ) + + @router.get("/attendance-summary") def attendance_summary( work_date: date | None = None, diff --git a/app/modules/reports/service.py b/app/modules/reports/service.py index 326a557..3502528 100644 --- a/app/modules/reports/service.py +++ b/app/modules/reports/service.py @@ -25,6 +25,7 @@ from app.modules.business.models import ( AttendanceRecord, Expense, FundAccount, + PerformanceMetric, Procurement, Project, ReportPushRun, @@ -55,13 +56,14 @@ from app.modules.reports.constants import ( LifecycleResponseKey, LifecycleSection, MetricKey, - ReportResponseKey, - ReportPushStatus, ReportErrorDetail, + ReportPushStatus, + ReportResponseKey, ReportStatus, ReportText, ReportTitle, ReportType, + EnterpriseAnalyticsKey, WorkReportMetricKey, ) from app.modules.risk.constants import RiskSummaryKey, risk_level_for_score @@ -387,6 +389,169 @@ class ReportService: report[LifecycleResponseKey.AI_ANALYSIS] = self._lifecycle_ai_analysis(report, actor) return report + def enterprise_analytics( + self, + project_code: str | None = None, + owner: str | None = None, + period_start: date | None = None, + period_end: date | None = None, + actor: str = ActorValue.API, + ) -> dict[str, Any]: + """Build V3 read-only finance, procurement, performance, and operations analytics.""" + + code = _next_code("ANALYTICS") + lifecycle = self.project_lifecycle_report( + project_code=project_code, + owner=owner, + period_start=period_start, + period_end=period_end, + include_ai=False, + actor=actor, + ) + metrics = lifecycle[LifecycleResponseKey.METRICS] + projects = metrics[LifecycleSection.PROJECTS] + procurements = metrics[LifecycleSection.PROCUREMENTS] + expenses = metrics[LifecycleSection.EXPENSES] + funds = metrics[LifecycleSection.FUNDS] + tasks = metrics[LifecycleSection.TASKS] + risks = metrics[LifecycleSection.RISKS] + health = metrics[LifecycleSection.HEALTH] + finance = { + MetricKey.BUDGET_TOTAL: projects[MetricKey.BUDGET_TOTAL], + MetricKey.ACTUAL_TOTAL: projects[MetricKey.ACTUAL_TOTAL], + MetricKey.BUDGET_USAGE_RATE: projects[MetricKey.BUDGET_USAGE_RATE], + MetricKey.CURRENT_BALANCE_TOTAL: funds[MetricKey.CURRENT_BALANCE_TOTAL], + MetricKey.NET_POSITION: funds[MetricKey.NET_POSITION], + MetricKey.RISK_ACCOUNTS: funds[MetricKey.RISK_ACCOUNTS], + MetricKey.PAYMENT_EXPOSURE: ( + procurements[MetricKey.ACTUAL_TOTAL] + expenses[MetricKey.AMOUNT_TOTAL] + ), + } + procurement = { + MetricKey.TOTAL: procurements[MetricKey.TOTAL], + MetricKey.PENDING_APPROVAL: procurements[MetricKey.PENDING_APPROVAL], + MetricKey.PENDING_DELIVERY: procurements[MetricKey.PENDING_DELIVERY], + MetricKey.UNPAID: procurements[MetricKey.UNPAID], + MetricKey.EXPECTED_TOTAL: procurements[MetricKey.EXPECTED_TOTAL], + MetricKey.ACTUAL_TOTAL: procurements[MetricKey.ACTUAL_TOTAL], + MetricKey.DELIVERY_RISK: procurements[MetricKey.PENDING_DELIVERY], + } + performance = self._enterprise_performance_stats() + operations = { + MetricKey.READINESS_SCORE: health[MetricKey.SCORE], + MetricKey.LEVEL: health[MetricKey.LEVEL], + MetricKey.COMPLETION_RATE: tasks[MetricKey.COMPLETION_RATE], + MetricKey.OVERDUE_TASKS: risks[MetricKey.OVERDUE_TASKS], + MetricKey.DELAYED_PROJECTS: risks[MetricKey.DELAYED_PROJECTS], + MetricKey.OVER_BUDGET_PROJECTS: risks[MetricKey.OVER_BUDGET_PROJECTS], + MetricKey.OPEN_EVENTS: risks[MetricKey.OPEN_EVENTS], + MetricKey.HIGH_EVENTS: risks[MetricKey.HIGH_EVENTS], + } + recommendations = lifecycle[LifecycleResponseKey.RECOMMENDATIONS] + lines = self._enterprise_analytics_lines( + lifecycle[LifecycleResponseKey.FILTERS], + finance, + procurement, + performance, + operations, + recommendations, + ) + report = _json_safe( + { + EnterpriseAnalyticsKey.CODE: code, + EnterpriseAnalyticsKey.TITLE: ReportTitle.ENTERPRISE_ANALYTICS, + EnterpriseAnalyticsKey.FILTERS: lifecycle[LifecycleResponseKey.FILTERS], + EnterpriseAnalyticsKey.FINANCE: finance, + EnterpriseAnalyticsKey.PROCUREMENT: procurement, + EnterpriseAnalyticsKey.PERFORMANCE: performance, + EnterpriseAnalyticsKey.OPERATIONS: operations, + EnterpriseAnalyticsKey.RECOMMENDATIONS: recommendations, + EnterpriseAnalyticsKey.LINES: lines, + EnterpriseAnalyticsKey.CONTENT: "\n".join(lines), + } + ) + AuditService(self.db).log( + AuditLogCreate( + actor=actor, + source=AuditSource.REPORTS, + action=AuditAction.ENTERPRISE_ANALYTICS, + target_type=AuditTargetType.ENTERPRISE_ANALYTICS, + target_id=code, + response_payload=report, + ) + ) + EventService(self.db).emit( + event_type=EventType.ENTERPRISE_ANALYTICS_GENERATED, + source=EventSource.ANALYTICS, + aggregate_type=EventAggregateType.ENTERPRISE_ANALYTICS, + aggregate_id=code, + actor=actor, + payload={ + EventPayloadKey.CODE: code, + EventPayloadKey.STATUS: ReportStatus.GENERATED, + }, + idempotency_key=f"enterprise-analytics:{code}", + dispatch=True, + ) + return report + + def _enterprise_performance_stats(self) -> dict[str, Any]: + total = self._count(PerformanceMetric) + confirmed = self._count(PerformanceMetric, PerformanceMetric.confirmed_score.is_not(None)) + return { + MetricKey.TOTAL: total, + MetricKey.CONFIRMED: confirmed, + MetricKey.CONFIRMED_RATE: _rate(confirmed, total), + MetricKey.AVERAGE_AUTO_SCORE: self._avg(PerformanceMetric.auto_score), + MetricKey.AVERAGE_CONFIRMED_SCORE: self._avg(PerformanceMetric.confirmed_score), + MetricKey.WEIGHT_TOTAL: self._sum(PerformanceMetric.weight), + MetricKey.BY_STATUS: self._group_counts(PerformanceMetric, PerformanceMetric.status), + } + + def _enterprise_analytics_lines( + self, + filters: dict[str, Any], + finance: dict[str, Any], + procurement: dict[str, Any], + performance: dict[str, Any], + operations: dict[str, Any], + recommendations: list[str], + ) -> list[str]: + scope = ( + "、".join(f"{key}={value}" for key, value in filters.items() if value) + or ReportText.DEFAULT_SCOPE + ) + lines = [ + f"- 范围：{scope}", + ( + f"- 财务：预算 {_money(finance[MetricKey.BUDGET_TOTAL])}，" + f"实际 {_money(finance[MetricKey.ACTUAL_TOTAL])}，" + f"净头寸 {_money(finance[MetricKey.NET_POSITION])}，" + f"支付暴露 {_money(finance[MetricKey.PAYMENT_EXPOSURE])}" + ), + ( + f"- 采购：总数 {procurement[MetricKey.TOTAL]}，" + f"待审批 {procurement[MetricKey.PENDING_APPROVAL]}，" + f"待交付 {procurement[MetricKey.PENDING_DELIVERY]}，" + f"未付款 {procurement[MetricKey.UNPAID]}" + ), + ( + f"- 绩效：指标 {performance[MetricKey.TOTAL]}，" + f"已确认 {performance[MetricKey.CONFIRMED]}，" + f"确认率 {performance[MetricKey.CONFIRMED_RATE]}%，" + f"平均自动分 {performance[MetricKey.AVERAGE_AUTO_SCORE]}" + ), + ( + f"- 运营：准备度 {operations[MetricKey.READINESS_SCORE]}，" + f"任务完成率 {operations[MetricKey.COMPLETION_RATE]}%，" + f"逾期任务 {operations[MetricKey.OVERDUE_TASKS]}，" + f"打开风险 {operations[MetricKey.OPEN_EVENTS]}" + ), + ReportText.ACTION_HEADER, + ] + lines.extend(f" - {item}" for item in recommendations) + return lines + def _lifecycle_filters( self, project_code: str | None, @@ -985,6 +1150,19 @@ class ReportService: response_payload=record_data, ) ) + EventService(self.db).emit( + event_type=EventType.REPORT_GENERATED, + source=EventSource.REPORTS, + aggregate_type=EventAggregateType.WORK_REPORT, + aggregate_id=record.code, + actor=actor, + payload={ + EventPayloadKey.CODE: record.code, + EventPayloadKey.STATUS: record.status, + }, + idempotency_key=f"report-generated:{record.code}", + dispatch=True, + ) return {ReportResponseKey.REPORT: report, ReportResponseKey.DATA: record_data} diff --git a/app/modules/workflows/constants.py b/app/modules/workflows/constants.py index c192cac..f386dfe 100644 --- a/app/modules/workflows/constants.py +++ b/app/modules/workflows/constants.py @@ -3,6 +3,10 @@ from enum import StrEnum class WorkflowType(StrEnum): RISK_EVENT_REVIEW = "risk_event_review" + REPORT_DELIVERY = "report_delivery" + LEGACY_SYNC_MONITOR = "legacy_sync_monitor" + ENTERPRISE_ANALYTICS = "enterprise_analytics" + AI_MEMORY_CAPTURE = "ai_memory_capture" class WorkflowStatus(StrEnum): diff --git a/app/tasks.py b/app/tasks.py index 6708b97..d9d2cd2 100644 --- a/app/tasks.py +++ b/app/tasks.py @@ -1,4 +1,5 @@ from typing import Any +from socket import gethostname from celery import Celery @@ -40,6 +41,30 @@ def push_daily_brief( db.close() +@celery_app.task(name="events.dispatch_pending") +def dispatch_pending_events( + limit: int | None = None, + actor: str = ActorValue.WORKER, +) -> list[dict[str, Any]]: + from app.modules.events.service import EventService + from app.modules.observability.constants import HeartbeatComponent + from app.modules.observability.service import ObservabilityService + + db = SessionLocal() + try: + ObservabilityService(db).record_heartbeat( + component=HeartbeatComponent.WORKER, + instance_id=gethostname(), + actor=actor, + ) + return EventService(db).dispatch_pending( + limit=limit or settings.event_dispatch_batch_size, + worker_id=f"{actor}:{gethostname()}", + ) + finally: + db.close() + + @celery_app.task(name="reports.push_project_weekly") def push_project_weekly( receive_id: str | None = None, diff --git a/app/tools/init_db.py b/app/tools/init_db.py index 223f5de..7203264 100644 --- a/app/tools/init_db.py +++ b/app/tools/init_db.py @@ -1,4 +1,5 @@ from app.core.database import Base, engine +from app.modules.ai_memory.models import AIMemoryEntry from app.modules.audit.models import AuditLog from app.modules.business.models import ( AttendanceRecord, @@ -19,6 +20,7 @@ from app.modules.business.models import ( ) from app.modules.feishu.models import FeishuEventReceipt from app.modules.events.models import DomainEvent +from app.modules.observability.models import SystemHeartbeat from app.modules.workflows.models import WorkflowAction, WorkflowInstance _MODELS = [ @@ -42,6 +44,8 @@ _MODELS = [ DomainEvent, WorkflowInstance, WorkflowAction, + AIMemoryEntry, + SystemHeartbeat, ] diff --git a/app/tools/run_scheduler.py b/app/tools/run_scheduler.py new file mode 100644 index 0000000..65772b4 --- /dev/null +++ b/app/tools/run_scheduler.py @@ -0,0 +1,17 @@ +from time import sleep + +from app.core.scheduler import create_scheduler + + +def main() -> None: + scheduler = create_scheduler() + scheduler.start() + try: + while True: + sleep(60) + except KeyboardInterrupt: + scheduler.shutdown(wait=False) + + +if __name__ == "__main__": + main() diff --git a/docker-compose.yml b/docker-compose.yml index 634e62a..aa4f750 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -81,6 +81,24 @@ services: migrate: condition: service_completed_successfully + scheduler: + build: . + env_file: + - .env + environment: + DATABASE_URL: postgresql+psycopg://${POSTGRES_USER:-company_ai}:${POSTGRES_PASSWORD:?POSTGRES_PASSWORD is required}@db:5432/${POSTGRES_DB:-company_ai} + REDIS_URL: redis://redis:6379/0 + SCHEDULER_ENABLED: "true" + TASK_QUEUE_ENABLED: "true" + command: ["python", "-m", "app.tools.run_scheduler"] + depends_on: + db: + condition: service_healthy + redis: + condition: service_healthy + migrate: + condition: service_completed_successfully + volumes: postgres_data: redis_data: diff --git a/tests/test_smoke.py b/tests/test_smoke.py index 169db46..16c7df7 100644 --- a/tests/test_smoke.py +++ b/tests/test_smoke.py @@ -35,6 +35,11 @@ from app.core.pagination import bounded_limit, bounded_offset from app.core.security import require_api_key, require_audit_api_key from app.main import _allow_cors_credentials, app from app.modules.audit.constants import AUDIT_REDACTED_VALUE +from app.modules.ai_memory.constants import ( + AIMemoryPayloadKey, + AIMemoryResponseKey, + AIMemoryStatus, +) from app.modules.events.constants import ( EventAggregateType, EventPayloadKey, @@ -46,7 +51,13 @@ from app.modules.events.service import EventService from app.modules.business.registry import get_domain_model from app.modules.business.service import _model_payload, serialize_model from app.modules.legacy_mysql.service import LegacyMySQLService +from app.modules.observability.constants import ( + HeartbeatComponent, + ObservabilityKey, +) +from app.modules.observability.service import ObservabilityService from app.modules.reports.constants import ( + EnterpriseAnalyticsKey, LifecycleAttentionKey, LifecycleResponseKey, LifecycleSection, @@ -168,6 +179,16 @@ def test_feishu_webhook_routes_message_event() -> None: def test_v3_request_id_health_and_metrics() -> None: + db = SessionLocal() + try: + ObservabilityService(db).record_heartbeat( + component=HeartbeatComponent.WORKER, + instance_id="pytest-worker", + actor="pytest", + ) + finally: + db.close() + response = client.get("/api/v1/health/live", headers={"X-Request-ID": "rid-v3-smoke"}) assert response.status_code == 200 assert response.headers["X-Request-ID"] == "rid-v3-smoke" @@ -179,7 +200,9 @@ def test_v3_request_id_health_and_metrics() -> None: response = client.get("/api/v1/metrics", headers=headers) assert response.status_code == 200 - assert "metrics" in response.json() + metrics = response.json()["metrics"] + assert ObservabilityKey.EVENTS in metrics + assert ObservabilityKey.HEARTBEATS in metrics def test_v3_event_idempotency_and_workflow_dispatch() -> None: @@ -228,6 +251,80 @@ def test_v3_event_idempotency_and_workflow_dispatch() -> None: db.close() +def test_v3_event_retry_and_dispatch_pending_route() -> None: + db = SessionLocal() + try: + event = EventService(db).emit( + event_type=EventType.REPORT_PUSH_FAILED, + source=EventSource.REPORTS, + aggregate_type=EventAggregateType.REPORT_PUSH_RUN, + aggregate_id="push-v3-retry", + actor="pytest", + payload={ + EventPayloadKey.CODE: "push-v3-retry", + EventPayloadKey.STATUS: ReportPushStatus.FAILED, + }, + idempotency_key="v3-report-push-retry", + ) + event.status = EventStatus.FAILED + event.last_error = "transient" + db.commit() + event_id = event.event_id + finally: + db.close() + + retry_response = client.post(f"/api/v1/events/{event_id}/retry", headers=headers) + assert retry_response.status_code == 200 + assert retry_response.json()["event"]["status"] == EventStatus.PENDING + + dispatch_response = client.post("/api/v1/events/dispatch-pending", headers=headers) + assert dispatch_response.status_code == 200 + dispatched = [ + item for item in dispatch_response.json()["items"] if item["event_id"] == event_id + ] + assert dispatched + assert dispatched[0]["status"] == EventStatus.PROCESSED + + +def test_v3_ai_memory_recall_and_auto_write() -> None: + response = client.post( + "/api/v1/ai/ask", + headers=headers, + json={ + "prompt": "Summarize quarterly cash planning for project memory smoke", + "context": { + AIMemoryPayloadKey.SCOPE: "project", + AIMemoryPayloadKey.SUBJECT: "P-MEM-SMOKE", + }, + }, + ) + assert response.status_code == 200 + data = response.json() + assert data[AIResponseKey.PROVIDER] == AIProviderName.NOOP + assert data[AIResponseKey.RAW][AIResponseKey.MEMORY_WRITE][AIMemoryPayloadKey.STATUS] == ( + AIMemoryStatus.ACTIVE + ) + + list_response = client.get( + "/api/v1/ai/memory?scope=project&subject=P-MEM-SMOKE", + headers=headers, + ) + assert list_response.status_code == 200 + assert list_response.json()[AIMemoryResponseKey.ITEMS] + + recall_response = client.post( + "/api/v1/ai/memory/recall", + headers=headers, + json={ + "query": "quarterly cash planning", + "scope": "project", + "subject": "P-MEM-SMOKE", + }, + ) + assert recall_response.status_code == 200 + assert recall_response.json()[AIMemoryResponseKey.ITEMS] + + def test_v3_risk_action_routes_are_disabled_in_read_only_mode() -> None: response = create_business_record( "risk-events", @@ -648,6 +745,43 @@ def test_project_lifecycle_report_summarizes_progress_cost_and_risk() -> None: ) +def test_v3_enterprise_analytics_returns_read_only_sections() -> None: + performance_response = create_business_record( + "performance-metrics", + { + "code": "PERF-V3-001", + "name": "V3 delivery score", + "weight": 20, + "auto_score": 82, + "confirmed_score": 78, + "status": StatusValue.REVIEWED, + }, + ) + assert performance_response.status_code == 200 + + response = client.get("/api/v1/reports/enterprise-analytics", headers=headers) + assert response.status_code == 200 + data = response.json() + assert data[EnterpriseAnalyticsKey.TITLE] == ReportTitle.ENTERPRISE_ANALYTICS + assert EnterpriseAnalyticsKey.FINANCE in data + assert EnterpriseAnalyticsKey.PROCUREMENT in data + assert EnterpriseAnalyticsKey.PERFORMANCE in data + assert EnterpriseAnalyticsKey.OPERATIONS in data + assert data[EnterpriseAnalyticsKey.PERFORMANCE][MetricKey.CONFIRMED] >= 1 + + db = SessionLocal() + try: + workflow = db.execute( + select(WorkflowInstance).where( + WorkflowInstance.workflow_type == WorkflowType.ENTERPRISE_ANALYTICS, + WorkflowInstance.aggregate_id == data[EnterpriseAnalyticsKey.CODE], + ) + ).scalar_one() + assert workflow.status == WorkflowStatus.COMPLETED + finally: + db.close() + + def test_work_report_counts_pending_approval_backlog_outside_period() -> None: today = date.today() project_code = "P-BACKLOG-001"