feat(Agent):jobs and files

.env.example

@@ -94,6 +94,7 @@ JOB_WORKERS=2
 JOB_HEARTBEAT_INTERVAL_SECONDS=10
 JOB_STALE_AFTER_SECONDS=120
 JOB_MAX_ATTEMPTS=3
+JOB_CHAT_HISTORY_LIMIT=20
 
 MAX_UPLOAD_SIZE_MB=20
 

Agent/AGENTS.md

@@ -0,0 +1,25 @@
+# Agent/AGENTS.md
+
+生效目录：`Agent/` 及其子目录。
+
+负责约束的修改类型：LangGraph 图、State/路由、结构化输出、MCP server/tool、RAG、因果分析工具、后处理和报告输出。
+
+## 修改前必须阅读
+
+- 必须阅读 [`Document/architecture/agent-runtime.md`](../Document/architecture/agent-runtime.md)、[`Document/architecture/job-file-lifecycle.md`](../Document/architecture/job-file-lifecycle.md) 和 [`Document/development/testing.md`](../Document/development/testing.md)。
+- 必须从当前调用者开始检查 worker runtime、graph runner、对应 State、工具注册、错误路径和测试；不能只修改一个节点后假设运行时会自动适配。
+- 修改 MCP、RAG 或 worker 初始化时，必须检查 `app/agent/worker/runtime.py`、`bootstrap.py`、Docker Compose 环境变量和知识库挂载。
+
+## Graph、输出与工具规则
+
+- 条件路由必须读取显式 State 字段，例如 `route_decision`、`fold_decision`；禁止用展示消息文本猜测控制流。
+- 涉及到修改agent链路的，需要告诉用户修改后的逻辑和目前的逻辑区别。
+- 结构化输出必须使用统一的 `Agent/llm_structured_output.py` 入口和当前 function calling 约定；修改 thinking、tool choice 或 schema 时必须同步检查调用器和测试。
+- MCP planner 使用原生 Tool Calls
+- RAG 启动检查与完整加载是两个阶段；知识库不可用时必须遵守当前无知识库模式，不得在 worker 启动中擅自增加全量加载。
+- 因果工具必须记录输入限制、矩阵方向、边权语义和方法假设；修改 DirectLiNGAM 时必须保持连续数值 CSV 和 `target_to_source` 契约。
+
+## 修改后验证
+
+- 必须覆盖成功路径、工具不可用、结构化解析失败、超时/异常、路由字段缺失和用户事件脱敏。
+- Agent 测试通过不代表 Job/worker/API 已验证；跨层变更必须追加 `tests/unit/agent/`、`tests/unit/` 或 Docker 验证，并在结果中区分真实模型/MCP 是否运行。

Agent/CausalAgentMCP/mcp_server.py

@@ -1,39 +1,71 @@
-import os
-import logging
-import sys
+"""因果分析 MCP server；通过 Job 冻结身份读取 MySQL 文件正文。"""
 
-CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))      # .../Agent/CausalAgentMCP
-AGENT_DIR = os.path.dirname(CURRENT_DIR)                      # .../Agent
-PROJECT_ROOT = os.path.dirname(AGENT_DIR)                     # 项目根目录
+from __future__ import annotations
 
-for p in (PROJECT_ROOT, AGENT_DIR):
-    if p not in sys.path:
-        sys.path.insert(0, p)
+import logging
+import os
+import sys
+from typing import Any
 
+CURRENT_DIR = os.path.dirname(os.path.abspath(__file__))
+AGENT_DIR = os.path.dirname(CURRENT_DIR)
+PROJECT_ROOT = os.path.dirname(AGENT_DIR)
+for path in (PROJECT_ROOT, AGENT_DIR):
+    if path not in sys.path:
+        sys.path.insert(0, path)
 
 from mcp.server.fastmcp import FastMCP
+
 from Agent.causal.causalachieve import (
     run_direct_lingam_analysis,
     run_olc_analysis,
     run_pc_analysis,
 )
+from Database.agent_connect import require_frozen_file_for_job
 
 
-log_file_path = os.path.join(CURRENT_DIR, 'mcp_server.log')
+log_file_path = os.path.join(CURRENT_DIR, "mcp_server.log")
 logging.basicConfig(
     level=logging.INFO,
-    format='%(asctime)s - %(levelname)s - %(message)s',
-    handlers=[
-        logging.FileHandler(log_file_path, encoding='utf-8'),
-    ]
+    format="%(asctime)s - %(levelname)s - %(message)s",
+    handlers=[logging.FileHandler(log_file_path, encoding="utf-8")],
 )
-logging.info("MCP Server Script Started, Logging Initialized")
-
 
 mcp = FastMCP("causal-analyzer")
 
+
+def _load_csv(
+    user_id: int,
+    job_id: str,
+    input_user_file_id: int,
+    input_object_id: int,
+) -> str:
+    """按可信运行时身份读取 Job  BLOB，仅在 MCP 进程内存中解码。"""
+    file_row = require_frozen_file_for_job(
+        user_id,
+        job_id,
+        input_user_file_id,
+        input_object_id,
+    )
+    return file_row["file_content"].decode("utf-8")
+
+
+def _error_result(message: str, error_type: str) -> dict[str, Any]:
+    """构造不包含文件正文或异常原文的 MCP 失败结果。"""
+    return {
+        "success": False,
+        "message": message,
+        "error_type": error_type,
+    }
+
+
 @mcp.tool()
-async def causal_pc(csv_data: str) -> dict:
+async def causal_pc(
+    user_id: int,
+    job_id: str,
+    input_user_file_id: int,
+    input_object_id: int,
+) -> dict:
     """
     使用PC算法对CSV数据执行因果发现分析。
 
@@ -50,31 +82,37 @@ async def causal_pc(csv_data: str) -> dict:
     - 仅提供条件独立性信息的场景
     - 使用快速邻接搜索优化的大规模问题
 
-    这是一个纯计算工具，不执行任何数据库或文件系统操作。
 
     Args:
-        csv_data: 一个包含完整CSV文件内容的字符串。
+        user_id: 由 Agent runtime 注入的当前用户 ID。
+        job_id: 由 Agent runtime 注入的当前分析任务 ID。
+        input_user_file_id: Job 创建时冻结的逻辑文件 ID。
+        input_object_id: Job 创建时冻结的不可变文件对象 ID。
+
+    文件正文由工具按 Job 冻结身份从文件库读取，模型不得传入 csv_data。
 
     Returns:
         一个包含分析结果的结构化字典，包括因果图结构和边的方向信息。
     """
-    logging.info(f"工具 'causal_pc' 已被调用，输入数据长度: {len(csv_data)}。")
+
     try:
-        # 工具的核心职责：执行分析
-        analysis_result = run_pc_analysis(csv_data)
-
-        return analysis_result
-
-    except Exception as e:
-        logging.error(f"'causal_pc' 工具执行出错: {e}", exc_info=True)
-        return {
-            "success": False,
-            "message": f"执行分析时发生内部错误: {e}",
-            "error_type": type(e).__name__,
-        }
+        csv_data = _load_csv(user_id, job_id, input_user_file_id, input_object_id)
+        logging.info("工具 causal_pc 读取文件字节数=%s", len(csv_data.encode("utf-8")))
+        return run_pc_analysis(csv_data)
+    except FileNotFoundError:
+        return _error_result("任务文件不可用", "FrozenFileNotFound")
+    except Exception:
+        logging.error("causal_pc 执行失败", exc_info=True)
+        return _error_result("执行分析时发生内部错误", "AnalysisError")
+
 
 @mcp.tool()
-async def causal_olc(csv_data: str) -> dict:
+async def causal_olc(
+    user_id: int,
+    job_id: str,
+    input_user_file_id: int,
+    input_object_id: int,
+) -> dict:
     """
     使用OLC算法对CSV数据执行因果发现分析，专门处理存在隐藏混杂因素的场景。
 
@@ -91,53 +129,65 @@ async def causal_olc(csv_data: str) -> dict:
     - 非加性噪声模型
     - 非常小的样本量（<200个样本）
 
-    这是一个纯计算工具，不执行任何数据库或文件系统操作。
-
     Args:
-        csv_data: 一个包含完整CSV文件内容的字符串，数据应为连续值变量。
+        user_id: 由 Agent runtime 注入的当前用户 ID。
+        job_id: 由 Agent runtime 注入的当前分析任务 ID。
+        input_user_file_id: Job 创建时冻结的逻辑文件 ID。
+        input_object_id: Job 创建时冻结的不可变文件对象 ID。
+
+    文件正文由工具按 Job 冻结身份从文件库读取，模型不得传入 csv_data。
 
     Returns:
         一个包含分析结果的结构化字典，包括因果图结构和潜在混杂因素信息。
     """
-    logging.info(f"工具 'causal_olc' 已被调用，输入数据长度: {len(csv_data)}。")
     try:
-        # 工具的核心职责：执行分析
-        analysis_result = run_olc_analysis(csv_data)
-
-        return analysis_result
-    except Exception as e:
-        logging.error(f"'causal_olc' 工具执行出错: {e}", exc_info=True)
-        return {
-            "success": False,
-            "message": f"执行分析时发生内部错误: {e}",
-            "error_type": type(e).__name__,
-        }
+        csv_data = _load_csv(user_id, job_id, input_user_file_id, input_object_id)
+        logging.info("工具 causal_olc 读取文件字节数=%s", len(csv_data.encode("utf-8")))
+        return run_olc_analysis(csv_data)
+    except FileNotFoundError:
+        return _error_result("任务文件不可用", "FrozenFileNotFound")
+    except Exception:
+        logging.error("causal_olc 执行失败", exc_info=True)
+        return _error_result("执行分析时发生内部错误", "AnalysisError")
 
 
 @mcp.tool()
-async def causal_direct_lingam(csv_data: str) -> dict:
-    """使用 DirectLiNGAM 对连续数值 CSV 数据执行因果发现分析。"""
-    logging.info(
-        "工具 'causal_direct_lingam' 已被调用，输入数据长度: %s。",
-        len(csv_data),
-    )
+async def causal_direct_lingam(
+    user_id: int,
+    job_id: str,
+    input_user_file_id: int,
+    input_object_id: int,
+) -> dict:
+    """使用 DirectLiNGAM 对连续数值 CSV 数据执行因果发现分析。
+
+    适用于线性、非高斯、误差独立且无潜在混杂的连续数值变量；文件正文由
+    工具按 Job 冻结身份读取，模型不得传入 csv_data。
+
+    Args:
+        user_id: 由 Agent runtime 注入的当前用户 ID。
+        job_id: 由 Agent runtime 注入的当前分析任务 ID。
+        input_user_file_id: Job 创建时冻结的逻辑文件 ID。
+        input_object_id: Job 创建时冻结的不可变文件对象 ID。
+    """
     try:
+        csv_data = _load_csv(user_id, job_id, input_user_file_id, input_object_id)
+        logging.info(
+            "工具 causal_direct_lingam 读取文件字节数=%s",
+            len(csv_data.encode("utf-8")),
+        )
         return run_direct_lingam_analysis(csv_data)
-    except Exception as exc:
-        logging.error("'causal_direct_lingam' 工具执行出错: %s", exc, exc_info=True)
-        return {
-            "success": False,
-            "algorithm": "direct_lingam",
-            "message": f"执行 DirectLiNGAM 分析时发生内部错误: {exc}",
-            "error_type": type(exc).__name__,
-        }
+    except FileNotFoundError:
+        return _error_result("任务文件不可用", "FrozenFileNotFound")
+    except Exception:
+        logging.error("causal_direct_lingam 执行失败", exc_info=True)
+        return _error_result("执行 DirectLiNGAM 分析时发生内部错误", "AnalysisError")
+
 
 if __name__ == "__main__":
     logging.info("MCP 因果分析服务器启动")
-
     try:
-        mcp.run(transport='stdio')
-    except Exception as e:
-        logging.error(f"MCP 服务器运行时出现致命错误: {e}", exc_info=True)
+        mcp.run(transport="stdio")
+    except Exception:
+        logging.error("MCP 服务器运行时出现致命错误", exc_info=True)
     finally:
-        logging.info("MCP 因果分析服务器关闭") 
+        logging.info("MCP 服务器关闭")

Agent/causal_agent/edges.py

@@ -3,7 +3,7 @@
 
 
 ROUTE_DECISIONS = {"fold", "postprocess", "normal_chat", "inquiry_answer"}
-FOLD_DECISIONS = {"preprocess", "agent"}
+FOLD_DECISIONS = {"preprocess", "agent", "normal_chat"}
 
 def decision_router(state: CausalAgentState) -> str:
     """

Agent/causal_agent/fault_tolerance.py

@@ -57,12 +57,22 @@ def timeout(run_timeout: float, idle_timeout: float | None = None) -> TimeoutPol
 
 
 def _error_message(error: NodeError) -> str:
-    return f"{error.node} 节点执行失败: {error.error}"
+    """生成不包含异常原文的节点失败摘要。"""
+    return f"{error.node} 节点执行失败"
 
 
 def sanitize_error(exc: BaseException) -> str:
-    """Return a display-safe error string without exposing implementation detail."""
-    return str(exc) or exc.__class__.__name__
+    """把异常归类为有限的公开错误，避免泄露路径、连接串或文件正文。"""
+    normalized = str(exc).lower()
+    if "timeout" in normalized:
+        return "调用超时"
+    if any(token in normalized for token in ("connection", "connect", "network")):
+        return "服务连接失败"
+    if any(token in normalized for token in ("permission", "auth")):
+        return "服务授权失败"
+    if any(token in normalized for token in ("rate", "limit")):
+        return "服务当前繁忙"
+    return "节点执行失败"
 
 
 def route_to_normal_chat(state: CausalAgentState, error: NodeError) -> Command:

Agent/causal_agent/graph.py

@@ -116,7 +116,8 @@ def build_graph(llm: "ChatOpenAI", mcp_tools: list, rag_tools: list, checkpointe
         edges.fold_router,#由fold_router函数决定路由
         {
             "preprocess": "preprocess",
-            "agent": "agent"
+            "agent": "agent",
+            "normal_chat": "normal_chat",
         }
     )