整体概览

基于 CoPaw v0.1.0 源码目录 src/copaw/ 完整分析 · 819 个文件 · 版本 v0.1.0.post1

全景图

深度解读

CoPaw 架构深度解读

结合灵工 SaaS 平台场景的架构分析与改造建议

第一层：接入层（Channel Adapters）

核心设计思想：渠道无关性

所有外部渠道都被一个统一的 AgentRequest 对象抹平，上层完全感知不到渠道差异：

BaseChannel 接口（可扩展自定义渠道）

class BaseChannel(ABC):
    channel: ChannelType          # 渠道标识字符串（如 "dingtalk"）
    uses_manager_queue: bool = True  # 是否使用 ChannelManager 消息队列

    def __init__(self, process, on_reply_sent=None,
                 show_tool_details=True,
                 filter_tool_messages=False,   # 是否过滤工具调用消息（用户不可见）
                 filter_thinking=False,         # 是否过滤 thinking 块
                 dm_policy="open",              # 私聊策略：open / deny
                 group_policy="open",           # 群聊策略：open / deny / mention_only
                 require_mention=False,         # 是否必须 @Bot 才响应
                 allow_from=None,               # 白名单 user_id 列表
                 ...):
        ...

# 实现自定义渠道只需：
# 1. 继承 BaseChannel
# 2. 实现 build_agent_request_from_native(payload) → AgentRequest
# 3. 实现 send_response(to_handle, response, meta) → None
# 4. 在 config.json 的 channels 节注册

灵工场景映射：接入层应至少覆盖两个渠道——Web 内嵌 Chat Widget（客户侧）和 API 接口（运营工具调用）。user_id 需要扩展为 tenant_id + user_id 的复合键，这是数据隔离的起点。

外网穿透：v0.1.0 已将 ngrok 替换为 Cloudflare Tunnel（cloudflared tunnel --url http://localhost:8088），自动获取 *.trycloudflare.com 公网域名，接收 Webhook 回调。

第二层：核心引擎

2.1 AgentRunner · query_handler

这是请求的调度入口，处理以下五个串行步骤：

1. approval 检查：是否有人工审批需求（对应 ToolGuard 的 TIMEOUT 状态）
2. /command 路由：识别 /help、/reset、/compact、/new 等指令，短路处理不进入 agent 循环
3. session load：从 Storage 加载历史 session（SafeJSONSession.load_state()），恢复上下文
4. agent 执行：调用 CoPawAgent 主体逻辑
5. session save：finally 块保证无论成功失败都落盘

关键设计：session 生命周期与 agent 生命周期解耦，agent 崩溃不会丢失历史记录。

2.2 CoPawAgent ★ 核心节点

MRO 链：CoPawAgent → ToolGuardMixin → ReActAgent，安全拦截通过 Python MRO 自动注入，无需修改父类代码。

灵工场景映射：在 rebuild_sys_prompt 阶段通过 env_context 参数注入当前企业的数据权限范围（如"你只能查询 tenant_id=X 的数据"），是隔离控制的最优位置，无需修改框架代码。

2.3 ReAct Loop — 推理执行循环

_reasoning() → LLM 决策下一步工具与参数
      ↓
ToolGuardMixin._acting() → 安全检查（deny/approve/timeout）
      ↓
_acting()    → 实际执行工具（Built-in / Skill / MCP）
      ↓
追加 memory  → 将工具返回结果追加到上下文
      ↓
循环         → LLM 再次 reasoning，直到任务完成或达到 max_iters

深层理解：LLM 永远不直接访问数据库，只通过工具的返回结果间接获取信息。这对数据隔离非常友好——只要工具层做好 tenant_id 过滤，LLM 就天然拿不到越权数据。

三类工具体系

Built-in Tools（13种，可按 Agent 独立开关）

Skill Tools（三层目录，动态加载）

基于 SKILL.md frontmatter 描述的软能力系统，v0.1.0 引入三层目录管理：

builtin_skills/      ← 代码包内置（12种），只读，含版本号
      ↓ customized 同名时优先
customized_skills/   ← 用户自定义（workspace_dir/），永不被 builtin 覆盖
      ↓ enable_skill() 同步
active_skills/       ← 运行时激活（workspace_dir/），Agent 实际使用

动态加载意味着可以按租户配置不同的 Skill 集合，即不同企业开放不同能力。

MCP Tools（Model Context Protocol）

支持 stdio、HTTP、SSE、StreamableHTTP 四种传输协议，支持热重载，v0.1.0 新增自动重连+rebuild_info 重建机制。

MCP 是当前标准化 agent 工具接口的主流协议。你们的业务系统（查工单、查结算、查合同）都应该封装成 MCP Server 对外暴露——每个企业一个 MCP Server，在 header 中注入 enterprise_id，强制数据隔离。

ToolGuard Engine — 安全拦截层

这是架构中最重要的安全边界，v0.1.0 全新模块，通过 MRO 机制对 CoPawAgent 透明注入。

灵工场景映射：数据查询类工具设为 guarded，写操作（发起付款、修改合同状态）设为需要人工 APPROVE 或直接 denied。这样 agent 能力开放可以渐进式推进，先只开放只读能力。

第三层：基础能力层

Memory（记忆系统）

Provider（模型提供商管理）

灵工场景映射：简单的数据查询意图理解用本地小模型（降低成本）；复杂的分析诊断切换云端大模型（提升质量）。这是 token 成本控制的关键机制。

Config（配置系统）

Storage（存储系统）

核心文件结构：
sessions/*.json    # 会话记录（{uid}_{sid}.json）
chats.json         # 会话元数据索引
token_usage.json   # token 消耗统计（日/Provider/Model 三维）
auth.json          # 认证（v0.1.0 新增，HMAC-SHA256）
jobs.json          # 定时任务（v0.1.0 新增）
providers.json     # LLM 配置
config.json        # Agent 运行配置

灵工场景映射：token_usage.json 对应你们的商业模式核心数据。每个租户的 token 消耗需要单独统计，用于计费或配额控制。纯文件存储在小规模部署时够用，规模化后需要迁移到数据库。

针对灵工 SaaS 的核心改造建议

① 数据隔离（三层叠加）

AgentRequest.tenant_id（v0.0.6 缺失，需新增）
      ↓
rebuild_sys_prompt 注入权限描述
（"你只能查询 tenant_id=X 的数据，不得跨租户"）
      ↓
MCP Tool 层强制过滤 tenant_id
（WHERE tenant_id = ? 硬编码，不可被 LLM 覆盖）

三层叠加确保隔离，任何一层都不能单独保障安全。

② 权限管控（ToolGuard 三级机制）

③ Token 费用控制（四层）

• 简单问题 → 本地小模型（Ollama/llama.cpp）降低成本
• 复杂分析 → 云端大模型提升质量
• 按租户统计消耗（TokenUsageManager 扩展 tenant_id 维度）
• 超配额降级或提示升级套餐

④ 商业模式设计（产品分层）

CoPaw 的 token_usage 数据 + Skill 按目录配置的机制天然支持产品分层：

关键结论

源码与文件

源码目录对应关系

配置文件映射深入解析

核心问题：每个文件何时被读、何时被写、映射到哪个运行时对象

重启影响矩阵

技术栈

功能清单与优先级

按"逆向研究价值"排序，★★★ = 最值得深挖

核心功能

渠道支持（v0.1.0 完整版）

Skill 内置清单（v0.1.0）

Skill 导入来源

• https://clawhub.ai/skill-name — 官方 Hub（默认）
• https://skills.sh/owner/repo/skill
• https://github.com/owner/repo/tree/branch/skill（含 SKILL.md 的目录）
• https://lobehub.com/skills/skill-name — ZIP 包格式
• https://modelscope.cn/skills/@owner/name
• https://skillsmp.com/skills/skill-name
• 本地 ZIP 文件上传（最大 200MB，自动安全扫描）

CoPaw 核心数据模型

基于 v0.1.0.post1 源码 Pydantic / dataclass 完整提取 · 共 9 大实体族

六大实体族总览

ER 实体关系图

实体关系说明

灵工 SaaS 多租户扩展设计

消息流转完整链路

基于 v0.1.0 源码 src/copaw/app/runner/runner.py + agents/react_agent.py 完整还原 · 15 步 · 4 阶段

完整 15 步交互时序图

六大核心设计规律

四个业务场景落地建议

关键设计决策解读

核心机制

CoPaw 最值得逆向研究的5个核心设计模式，每一个都对灵工 SaaS 场景有直接的工程价值。

身份记忆 PROFILE

优先级：★★★
源码位置：src/copaw/agents/prompt.py、src/copaw/agents/memory/agent_md_manager.py、src/copaw/agents/hooks/bootstrap.py
关键机制：PromptBuilder 每次对话前重建 system prompt，修改文件立即生效，无缓存

设计哲学：「越用越懂你」

不同于传统 AI 每次对话都重置，CoPaw 通过三层 Markdown 文件构建持久化的 Agent 身份记忆，且全部在对话开始前动态读取，无任何缓存。

PromptBuilder 完整实现（prompt.py 源码精读）

class PromptBuilder:
    HEARTBEAT_PATTERN = re.compile(
        r"<!-- heartbeat:start -->.*?<!-- heartbeat:end -->",
        re.DOTALL,
    )

    def _load_file(self, filename: str) -> None:
        file_path = self.working_dir / filename
        if not file_path.exists():
            return          # ★ 所有文件都是可选的，不存在就跳过，不报错

        content = file_path.read_text(encoding="utf-8").strip()

        # ① 自动剥离 YAML frontmatter（--- 开头的元数据块）
        if content.startswith("---"):
            parts = content.split("---", 2)
            if len(parts) >= 3:
                content = parts[2].strip()

        # ② AGENTS.md 特殊处理：heartbeat 段根据配置开关
        if filename == "AGENTS.md":
            if not self.heartbeat_enabled:
                content = self.HEARTBEAT_PATTERN.sub("", content)  # 移除整段
            else:
                content = content.replace("<!-- heartbeat:start -->", "")
                content = content.replace("<!-- heartbeat:end -->", "")

        if content:
            self.prompt_parts.append(f"# {filename}")   # 添加节标题
            self.prompt_parts.append(content)

    def build(self) -> str:
        files_to_load = self.enabled_files or PromptConfig.DEFAULT_FILES
        # DEFAULT_FILES = ["AGENTS.md", "SOUL.md", "PROFILE.md"]

        for filename in files_to_load:
            self._load_file(filename)     # 按序加载，全部可选

        if not self.prompt_parts:
            return DEFAULT_SYS_PROMPT     # 降级兜底

        return "

".join(self.prompt_parts)


def build_system_prompt_from_working_dir(
    working_dir, enabled_files=None,
    agent_id=None, heartbeat_enabled=False
) -> str:
    # 按 agent_id 加载独立配置（v0.1.0 多 Agent 支持）
    if agent_id:
        agent_config = load_agent_config(agent_id)
        enabled_files = agent_config.system_prompt_files

    builder = PromptBuilder(working_dir, enabled_files, heartbeat_enabled)
    prompt = builder.build()

    # ③ v0.1.0 新增：Agent Identity 前置注入
    if agent_id:
        identity = (
            f"# Agent Identity

"
            f"Your agent id is `{agent_id}`. "
            f"This is your unique identifier in the multi-agent system.

"
        )
        prompt = identity + prompt

    return prompt

系统提示词构建顺序（完整时序）

rebuild_sys_prompt() 执行时序：

① Agent Identity（agent_id 唯一标识，v0.1.0 新增）
   "Your agent id is `enterprise_A`. This is your unique identifier..."

② # AGENTS.md
   [文件内容，去除 frontmatter]
   [heartbeat 段根据 config.heartbeat.enabled 保留或删除]

③ # SOUL.md
   [文件内容，去除 frontmatter]

④ # PROFILE.md
   [文件内容，去除 frontmatter]
   （不存在时跳过，不影响其他文件）

⑤ env_context（可选，通过 CoPawAgent(env_context=...) 传入）
   ↑ 这里注入租户权限上下文！
   "你只能查询 tenant_id=enterprise_A 的数据。"

最终合并："

".join([① ② ③ ④ ⑤])

BootstrapHook — 首次初始化引导

当工作区存在 BOOTSTRAP.md 时，BootstrapHook 在首次用户交互时触发，引导用户完成 Agent 初始设置：

class BootstrapHook:
    async def __call__(self, agent, kwargs):
        bootstrap_path = self.working_dir / "BOOTSTRAP.md"
        completed_flag = self.working_dir / ".bootstrap_completed"

        if completed_flag.exists():   # 防止重复触发
            return None
        if not bootstrap_path.exists():
            return None

        messages = await agent.memory.get_memory()
        if not is_first_user_interaction(messages):  # 只对第一条消息触发
            return None

        # 在第一条 user 消息前注入引导文本（中/英/俄 三语）
        guidance = build_bootstrap_guidance(self.language)
        for msg in messages[system_count:]:
            if msg.role == "user":
                prepend_to_message_content(msg, guidance)
                break

        # 创建完成标记文件，防止下次再触发
        completed_flag.touch()
        return None

引导文本（中文版）包含：1. 阅读 BOOTSTRAP.md → 2. 按指示帮用户定义 AI 身份 → 3. 创建/更新 PROFILE.md/AGENTS.md 等文件 → 4. 完成后自动删除 BOOTSTRAP.md。

灵工场景映射：企业首次接入时，BOOTSTRAP.md 可引导管理员完成 AGENTS.md（业务权限声明）、SOUL.md（服务风格）配置，零开发介入，运营人员自助完成初始化。

PROFILE.md 自动更新机制

# post_reasoning hook — 对话结束后判断是否更新 PROFILE.md
async def post_reasoning_hook(agent, response):
    # 1. 追加写入 Session History（每轮对话必做）
    await memory_manager.append_session(response)

    # 2. Agent 主动写入 PROFILE.md（通过 write_file 工具）
    #    当 AI 在对话中发现用户新偏好时，可使用 write_file 工具更新 PROFILE.md
    #    下次对话 rebuild_sys_prompt() 时自动读取最新内容

PROFILE.md 是「越用越懂你」的实现载体。AI 可以通过 write_file 工具主动更新用户偏好，形成学习闭环。对于多用户场景，需要按 user_id 隔离：profiles/{user_id}/PROFILE.md。

灵工场景改造建议

Skill 触发机制

优先级：★★★
源码位置：src/copaw/agents/skills_manager.py、src/copaw/agents/skills_hub.py、src/copaw/security/skill_scanner/
关键机制：三层目录 + SKILL.md frontmatter description 字段是 LLM 触发决策的唯一依据

Skill 与 MCP Tool 的根本区别

三层目录架构（v0.1.0 重大升级）

# 目录优先级：customized > builtin（同名时 customized 覆盖 builtin）
# Agent 运行时只读取：active_skills/（builtin + customized 的合并结果）

builtin_skills/              ← 代码包内，只读，12种内置技能
  ├── cron/
  │   └── SKILL.md           ← metadata.builtin_skill_version = "1.0"
  ├── docx/
  │   ├── SKILL.md
  │   └── scripts/           ← Python/JS 执行脚本
  └── ... (共 12 种)

workspace_dir/
├── customized_skills/       ← 用户自定义（永不被 builtin 覆盖）
│   └── linggong_query/
│       └── SKILL.md
└── active_skills/           ← 运行时激活（Agent 实际加载此目录）
    ├── cron/SKILL.md        ← 从 builtin 同步（enable_skill()）
    └── linggong_query/SKILL.md  ← 从 customized 同步

# 生命周期（SkillService 接口）：
create_skill()     → 写入 customized_skills/
enable_skill()     → 同步到 active_skills/（含 SkillScanner 安全扫描）
disable_skill()    → 从 active_skills/ 移除
list_available_skills() → 读取 active_skills/ 目录列表

SKILL.md 完整格式（触发机制核心）

---
name: linggong_data_query         # ★ 技能标识（唯一，同目录名）
description: |                    # ★★★ 关键字段：LLM 根据此决定何时调用
  当用户询问灵工订单、结算数据、用工记录时使用本技能。
  触发词：查订单、查结算、查工人信息、数据分析、统计报表
metadata:
  builtin_skill_version: "1.0"    # 版本号（内置技能升级比对用）
  copaw:
    emoji: "📊"
    requires:
      bins: []                    # 依赖的外部二进制（无则为空）
    install: []                   # 安装方式声明（可选）
---

# 灵工数据查询规范

当用户发起数据查询时，按以下步骤执行：

## 第一步：确认查询范围
- 确认时间范围（如：本周/本月/自定义日期）
- 确认查询类型（订单/结算/用工记录/统计报表）

## 第二步：调用数据查询工具
使用 `query_worker_orders` MCP 工具，参数格式：
- date_from: "YYYY-MM-DD"
- date_to: "YYYY-MM-DD"
- status: "all" | "pending" | "completed"

## 第三步：结果展示
- 查询结果用 Markdown 表格展示
- 金额字段保留两位小数，加"元"单位
- 超过 20 条时自动分页，告知用户总数

Skill 注册与触发完整流程

# ── 1. 初始化时注册（CoPawAgent 构建时）────────────────────────────
for skill_name in list_available_skills(workspace_dir):
    skill_dir = active_skills / skill_name
    toolkit.register_agent_skill(str(skill_dir))
    # → 读取 SKILL.md，提取 description
    # → 将 Skill 内容注入 system prompt 的 <skill> 标签

# ── 2. system prompt 中的 Skill 格式 ──────────────────────────────
# 注入后的 system prompt 含有如下段落：
<skill name="linggong_data_query">
当用户询问灵工订单、结算数据... (SKILL.md 正文内容)
</skill>

# ── 3. LLM 推理时（_reasoning 阶段）─────────────────────────────
# LLM 读取 system prompt 中所有 <skill> 标签
# 根据用户意图匹配 description 最合适的 Skill
# 按 Skill 中定义的步骤自主编排工具调用序列

# ── 4. namesake_strategy（同名 Skill 处理策略）────────────────────
toolkit.register_agent_skill(
    str(skill_dir),
    namesake_strategy="skip"   # skip/override/raise/rename
)

SkillScanner 安全扫描（激活前自动触发）

# enable_skill() 内部流程：
def enable_skill(self, name: str, force: bool = False) -> bool:
    skill_dir = customized_skills / name

    # ① SkillScanner 扫描（mode="block" 时扫描失败则拒绝激活）
    scanner = SkillScanner()
    result = scanner.scan(skill_dir)
    if not result.is_safe and not force:
        raise SkillScannerError(f"Skill scan failed: {result.max_severity}")

    # ② 安全通过 → 同步到 active_skills/
    shutil.copytree(skill_dir, active_skills / name)
    return True

# SkillScanner 检测 17 种威胁（PatternAnalyzer）：
# PROMPT_INJECTION / COMMAND_INJECTION / HARDCODED_SECRETS
# DATA_EXFILTRATION / SUPPLY_CHAIN_ATTACK / UNICODE_STEGANOGRAPHY
# ...（见数据结构章节完整列表）

Skills Hub 在线安装（v0.1.0 新增）

# install_skill_from_hub(url) 完整流程：
URL
  → _resolve_bundle_from_url()    # 解析来源（clawhub/github/lobehub/modelscope...）
  → bundle(SKILL.md + files)
  → _normalize_bundle()           # 校验结构、symlink 防护、路径遍历防护
  → SkillService.create_skill()   # 写入 customized_skills/
  → SkillScanner.scan()           # PatternAnalyzer 17 种威胁检测
  → enable_skill()                # 同步到 active_skills/
  → HubInstallResult(name, enabled, source_url)

# 支持 7 种来源：
clawhub.ai / github.com / skills.sh / lobehub.com
modelscope.cn / skillsmp.com / 本地 ZIP（最大 200MB）

灵工场景映射：将灵工业务查询规范、诊断流程、报表规范封装为私有 Skill，通过 ZIP 上传方式部署到各企业工作区；按企业差异化配置 active_skills/，实现能力定制化。

工具调用链路

优先级：★★★
源码位置：src/copaw/agents/tool_guard_mixin.py、src/copaw/security/tool_guard/engine.py、src/copaw/agents/react_agent.py
关键机制：ToolGuardMixin 通过 MRO 透明注入，5步决策链（denied → guarded → always_run → approval → execute）

工具注册全流程（Agent 初始化时）

# CoPawAgent._create_toolkit() — 按序注册三类工具

# ① 内置工具（13种，按 Agent config 独立开关）
for tool_name, tool_func in BUILTIN_TOOL_FUNCTIONS.items():
    if agent_config.tools.builtin_tools[tool_name].enabled:
        toolkit.register_tool_function(
            tool_func,
            namesake_strategy="skip"
        )

# ② Skill 工具（从 active_skills/ 动态加载）
for skill_name in list_available_skills(workspace_dir):
    toolkit.register_agent_skill(str(active_skills / skill_name))

# ③ MCP 工具（异步注册，对话开始前）
await agent.register_mcp_clients(namesake_strategy="skip")
for client in mcp_clients:
    await toolkit.register_mcp_client(client)

# ④ memory_search 工具（可选，仅启用 MemoryManager 时）
if memory_manager:
    toolkit.register_tool_function(
        create_memory_search_tool(memory_manager)
    )

ToolGuardMixin 5步决策链（核心）

MRO 链：CoPawAgent → ToolGuardMixin → ReActAgent
_acting(tool_call) 覆写拦截点，每次工具调用都经过以下决策：

ToolGuardEngine 源码关键路径

# engine.py — guard() 核心方法
def guard(self, tool_name, params, *, only_always_run=False):
    if not self._enabled:
        return None

    result = ToolGuardResult(tool_name=tool_name, params=params)

    # 按 only_always_run 过滤 Guardian 列表
    guardians = (
        [g for g in self._guardians if g.always_run]
        if only_always_run else self._guardians
    )

    for guardian in guardians:
        try:
            findings = guardian.guard(tool_name, params)
            result.findings.extend(findings)
            result.guardians_used.append(guardian.name)
        except Exception as exc:
            result.guardians_failed.append({"name": guardian.name, "error": str(exc)})

    result.guard_duration_seconds = time.monotonic() - t0
    return result

# tool_guard_mixin.py — _acting() 决策（简化）
async def _acting(self, tool_call):
    self._ensure_tool_guard()
    async with self._tool_guard_lock:     # ← 串行化，防并发竞态
        action = await self._decide_guard_action(tool_call)

    if action:
        return await self._execute_guard_action(action, tool_call)

    return await super()._acting(tool_call)   # ← 无拦截，正常执行

斜杠命令系统（/compact /new /clear 等）

# CommandHandler.SYSTEM_COMMANDS（v0.1.0 完整命令集）
{
    "compact",        # 手动触发记忆压缩（同等于 MemoryCompactionHook 自动触发）
    "new",            # 新建会话（保留记忆摘要，清空对话历史）
    "clear",          # 清空当前会话（含压缩摘要一起清除）
    "history",        # 查看对话历史摘要（ReMeLight 生成）
    "compact_str",    # 显示当前压缩摘要内容（调试用）
    "await_summary",  # 等待异步摘要任务完成
    "message",        # 内部发送消息
    "dump_history",   # 导出历史到 debug_history.jsonl（调试用）
    "load_history",   # 从文件加载历史（调试/还原用）
}

# 命令路由（CoPawAgent.reply() 短路处理）
if self.command_handler.is_command(query):
    msg = await self.command_handler.handle_command(query)
    await self.print(msg)
    return msg         # ← 短路！不进入 ReAct 循环，零 Token 消耗

设计精髓：系统命令通过短路机制绕过 ReAct 循环，不消耗任何 LLM Token，响应速度极快。灵工平台可以利用这个机制实现「零成本快速指令」，如 /report 直接触发 Cron 任务生成报告，完全不走 LLM 推理。

灵工场景改造建议

MCP 协议规范

优先级：★★★
源码位置：src/copaw/app/mcp/manager.py、src/copaw/agents/react_agent.py、src/copaw/config/config.py
关键机制：四种 transport + 热重载 + v0.1.0 自动重连 + rebuild_info 重建 + 异步锁原子替换

MCP 在 CoPaw 中的定位

MCP（Model Context Protocol）是 CoPaw 与外部系统交互的标准化接口协议。所有业务逻辑（查订单、查结算、调用业务 API）都应通过 MCP Server 暴露，Agent 通过 JSON-RPC 调用。

# 核心理念：LLM 不直接访问数据库
用户问：「查一下本月结算总额」
  ↓
LLM 决策：调用 tool "get_settlement_summary"
  ↓
MCP Client → JSON-RPC → MCP Server（灵工业务侧）
  ↓
MCP Server 从 Header 获取 tenant_id，强制注入查询条件
  ↓
SELECT SUM(amount) FROM settlements WHERE tenant_id=? AND month=?
  ↓
返回结果 → tool_result → LLM 生成自然语言回复

四种 Transport 对比

MCP 热重载（MCPClientManager）—— 原子替换

# MCPClientManager.replace_client() — 零停机替换（connect new (outside lock) → swap inside lock）
async def replace_client(self, key, client_config, timeout=60.0):
    # ① 锁外创建并连接新 client（可能耗时，不阻塞其他请求）
    new_client = self._build_client(client_config)
    await asyncio.wait_for(new_client.connect(), timeout=timeout)

    # ② 锁内原子替换（最小化锁持有时间）
    async with self._lock:
        old_client = self._clients.get(key)
        self._clients[key] = new_client
        if old_client:
            await old_client.close()   # 优雅关闭旧连接

# get_clients() 每次调用都加锁刷新（保证拿到最新 client 列表）
async def get_clients(self) -> list:
    async with self._lock:
        return [c for c in self._clients.values() if c is not None]

v0.1.0 MCP 自动重连机制

# react_agent.py — _recover_mcp_client() 两阶段恢复
async def _recover_mcp_client(self, client):
    # 阶段1：尝试直接重连
    if await self._reconnect_mcp_client(client):
        return client

    # 阶段2：重连失败 → 从 _copaw_rebuild_info 重建 client
    rebuilt = self._rebuild_mcp_client(client)
    if rebuilt and await self._reconnect_mcp_client(rebuilt):
        # 复用原有 client 引用（保持 manager 内部状态稳定）
        return self._reuse_shared_client_reference(client, rebuilt)

    return None   # 恢复失败，跳过此 client（不影响其他 client）

# _rebuild_mcp_client() — 从 rebuild_info 完整重建
def _rebuild_mcp_client(self, client):
    info = getattr(client, "_copaw_rebuild_info", None)
    if not info: return None

    if info["transport"] == "stdio":
        return StdIOStatefulClient(
            name=info["name"], command=info["command"],
            args=info["args"], env=info["env"],
        )
    else:   # sse / streamable_http
        # ★ 环境变量展开（${ENTERPRISE_ID} 在重建时重新展开）
        headers = {k: os.path.expandvars(v) for k, v in info["headers"].items()}
        return HttpStatefulClient(
            name=info["name"], transport=info["transport"],
            url=info["url"], headers=headers,
        )

区分 MCP 内部 Cancel 和 Task 真实 Cancel

@staticmethod
def _should_propagate_cancelled_error(error) -> bool:
    """精确判断 CancelledError 是否需要传播
    
    MCP 内部超时产生的 CancelledError → 应吞掉（不传播）
    用户主动取消任务产生的 CancelledError → 必须传播
    """
    if not isinstance(error, asyncio.CancelledError):
        return False
    task = asyncio.current_task()
    if task is None: return False

    # Python 3.11+：cancelling() > 0 表示任务被外部真正取消
    cancelling = getattr(task, "cancelling", None)
    if callable(cancelling):
        return cancelling() > 0

    # Python < 3.11：保守策略，总是传播
    return True

设计精髓：MCP 内部的 CancelledError（如工具调用超时）应被吞掉继续运行，但用户主动取消任务的 CancelledError 必须传播，否则会造成任务假死。这个细节在生产环境中非常重要。

灵工 MCP Server 设计规范

# 推荐架构：灵工业务 MCP Server（Python FastMCP）
from mcp.server.fastmcp import FastMCP

app = FastMCP("linggong-business")

# ★ 安全原则：tenant_id 从 Header 获取，不允许 LLM 传参
def get_tenant_id() -> str:
    return current_request_context.headers["X-Enterprise-Id"]

@app.tool()
async def query_worker_orders(
    date_from: str,       # ✅ LLM 传参：查询条件
    date_to: str,
    status: str = "all",
    # ❌ 绝不允许：tenant_id: str  ← 不能作为参数！
) -> str:
    """查询本企业用工订单列表"""
    tenant_id = get_tenant_id()  # ← 服务端强制获取，不可被 LLM 覆盖
    rows = await db.execute(
        "SELECT * FROM orders WHERE tenant_id=? AND date BETWEEN ? AND ? AND status=?",
        [tenant_id, date_from, date_to, status]
    )
    return format_markdown_table(rows)

# mcp.json 配置：
# headers: {"X-Enterprise-Id": "${ENTERPRISE_ID}"}  ← 环境变量展开

会话上下文管理

优先级：★★★
源码位置：src/copaw/agents/react_agent.py、src/copaw/agents/memory/memory_manager.py、src/copaw/agents/hooks/memory_compaction.py
关键数字：memory_compact_ratio=0.75（默认，可配置 0.30–0.90）、max_input_length=131072（128K，按模型设置）

完整 15 步处理时序

Memory 内容结构与生命周期

# InMemoryMemory.content 内部结构
# 每条记录是 (Msg, marks[]) 二元组
memory.content = [
    # 步骤5 写入，每次对话重建
    (Msg(role="system",    content="# AGENTS.md
..."),           marks=[]),

    # 步骤4 load，或新一轮对话追加
    (Msg(role="user",      content="查一下本月结算总额"),           marks=[]),

    # 步骤8 _reasoning 写入（LLM 决定调用工具）
    (Msg(role="assistant", content=[{"type":"tool_use","name":"query_settlement",...}]), marks=[]),

    # 步骤10 工具结果追加
    (Msg(role="user",      content=[{"type":"tool_result","output":"合计 ¥ 128,000"}]),  marks=[]),

    # 步骤8 最终回复
    (Msg(role="assistant", content="本月结算总额为 128,000 元..."),  marks=[]),

    # 步骤9 ToolGuard 拦截（如果发生）
    (Msg(role="system",    content="⛔ 工具 execute_shell_command 已被拦截"),
                                                                    marks=["tool_guard_denied"]),
]

# marks 当前值：
# []                    = 正常消息
# ["tool_guard_denied"] = 被 ToolGuard 拦截的消息（cleanup 时会移除）
# 压缩后的消息：被 mark_messages_compressed() 标记，不再传给 LLM

# 持久化生命周期：
# LOAD:  query 开始时从 sessions/{uid}_{sid}.json 完整恢复
# SAVE:  query 结束时 finally 块完整落盘（覆盖写入）
# ROLLBACK: approval 拒绝时 _cleanup_denied → 重新 save

# compressed_summary 独立存储（叠加在 memory 之上）
compressed_summary = memory.get_compressed_summary()  # → str 或 ""

# 完整上下文 = system_prompt + compressed_summary + memory.content（未被压缩的部分）
# LLM 实际接收：
messages_to_llm = [
    {"role": "system",    "content": sys_prompt + "

" + compressed_summary},
    # ... 未压缩的历史消息 ...
    {"role": "user",      "content": "当前用户输入"},
]

Token 计数与压缩参数（AgentsRunningConfig）

# 压缩触发计算（MemoryCompactionHook）
memory_compact_threshold = max_input_length * memory_compact_ratio
# 例：128K × 0.75 = 98,304 tokens → 超过此值触发压缩

# 实际触发逻辑：
str_tokens = count(sys_prompt + compressed_summary)  # 固定占用
left_threshold = memory_compact_threshold - str_tokens
# left_threshold = 剩余给历史消息的 token 预算
# 若历史消息超过 left_threshold → 触发 compact_memory()

# 压缩后保留：
# - 最新 MEMORY_COMPACT_KEEP_RECENT=3 条消息（不压缩）
# - 更早的消息 → 用 LLM 生成摘要，存入 compressed_summary
# - 摘要前置于下一次对话的 system_prompt 之后

v0.1.0 关键变化对比（v0.0.6 → v0.1.0）

MemoryCompactionHook 源码关键逻辑

class MemoryCompactionHook:
    """pre_reasoning hook：每轮对话调用，检查并在需要时触发压缩"""

    async def __call__(self, agent, kwargs):
        # 1. 热读 Agent Config（无缓存，支持配置热更新）
        agent_config = load_agent_config(self.memory_manager.agent_id)
        running_config = agent_config.running
        token_counter = get_copaw_token_counter(agent_config)

        # 2. 计算固定占用（system_prompt + compressed_summary）
        str_token_count = await token_counter.count(
            text=(agent.sys_prompt or "") + (memory.get_compressed_summary() or "")
        )
        left_threshold = running_config.memory_compact_threshold - str_token_count

        # 3. tool_result 三维压缩（先做，减少主压缩触发频率）
        await self.memory_manager.compact_tool_result(
            messages=messages,
            recent_n=running_config.tool_result_compact_recent_n,          # 默认 1
            old_threshold=running_config.tool_result_compact_old_threshold, # 默认 1000 chars
            recent_threshold=running_config.tool_result_compact_recent_threshold, # 默认 30000
            retention_days=running_config.tool_result_compact_retention_days,     # 默认 3天
        )

        # 4. 检查是否触发主压缩
        (messages_to_compact, _, is_valid) = await self.memory_manager.check_context(
            messages=messages,
            memory_compact_threshold=left_threshold,
            memory_compact_reserve=running_config.memory_compact_reserve,   # 默认 0.10
            as_token_counter=token_counter,
        )

        if not messages_to_compact:
            return None  # 未触发

        # 5. 触发主压缩
        self.memory_manager.add_async_summary_task(messages=messages_to_compact)
        await self._print_status_message(agent, "🔄 Context compaction started...")

        compact_content = await self.memory_manager.compact_memory(
            messages=messages_to_compact,
            previous_summary=memory.get_compressed_summary(),  # 累积摘要传入
        )

        if compact_content:
            await memory.mark_messages_compressed(messages_to_compact)  # 标记已压缩
            await memory.update_compressed_summary(compact_content)      # 更新摘要
            await self._print_status_message(agent, "✅ Context compaction completed")

RoutingChatModel — LLM 智能路由（v0.1.0 新增）

# src/copaw/agents/routing_chat_model.py
class RoutingChatModel(ChatModelBase):
    """在 local / cloud 两个 endpoint 之间按策略路由的 ChatModel 代理"""

    def __init__(self, *, local_endpoint, cloud_endpoint, routing_cfg):
        self.local_endpoint = local_endpoint   # Ollama / llama.cpp
        self.cloud_endpoint = cloud_endpoint   # OpenAI / Anthropic / etc.
        self.policy = RoutingPolicy(routing_cfg)

    async def __call__(self, messages, tools=None, tool_choice=None, **kwargs):
        # 提取 user 文本用于决策
        text = " ".join(m["content"] for m in messages
                        if m["role"] == "user" and isinstance(m.get("content"), str))

        decision = self.policy.decide(text=text, tools_available=tools is not None)

        endpoint = (self.local_endpoint if decision.route == "local"
                    else self.cloud_endpoint)

        logger.debug("LLM routing: route=%s provider=%s model=%s reasons=%s",
                     decision.route, endpoint.provider_id, endpoint.model_name,
                     ",".join(decision.reasons))

        return await endpoint.model(messages, tools, tool_choice, **kwargs)

# RoutingPolicy.decide()（v0.1.0 当前实现）
class RoutingPolicy:
    def decide(self, *, text="", channel="", tools_available=True):
        if getattr(self.cfg, "mode", "local_first") == "cloud_first":
            return RoutingDecision(route="cloud", reasons=["mode:cloud_first"])
        return RoutingDecision(route="local", reasons=["mode:local_first"])
        # ↑ 当前按 mode 静态决策；未来可扩展为按文本长度/工具数量动态路由

灵工场景映射：日常查询（"查上周订单"）走 local_first → 本地 Ollama 小模型（成本近零）；月度分析报告走 cloud_first → claude-sonnet（高质量）。仅需改 config.json 的 llm_routing.mode，无需重启（ConfigWatcher 热更新）。

Zero-downtime 热重载（MultiAgentManager.reload_agent）

# src/copaw/app/multi_agent_manager.py
async def reload_agent(self, agent_id: str) -> bool:
    """零停机热重载：新实例完全启动后再原子替换，正在进行的 SSE 流不中断"""

    # Step 1：检查是否在运行（加锁，快速）
    async with self._lock:
        old_instance = self.agents[agent_id]

    # Step 2：读取最新配置（锁外，慢操作）
    config = load_config()
    agent_ref = config.agents.profiles[agent_id]

    # Step 3：创建并启动新 Workspace 实例（锁外，慢操作，不阻塞其他 Agent）
    new_instance = Workspace(agent_id, workspace_dir=agent_ref.workspace_dir)

    # Step 3.5：复用旧实例的可重用组件（关键设计！）
    async with self._lock:
        old_instance = self.agents.get(agent_id)
    if old_instance:
        reusable = old_instance._service_manager.get_reusable_services()
        # reusable 包含：memory_manager（记忆不丢失）、chat_manager（索引不重建）
        # 以及已建立的 MCP client 连接（不重连）
        if reusable:
            await new_instance.set_reusable_components(reusable)

    await new_instance.start()  # 使用复用组件启动，比全新启动快很多

    # Step 4：原子替换（加锁，极短，最小化阻塞时间）
    async with self._lock:
        old_instance = self.agents[agent_id]
        self.agents[agent_id] = new_instance  # 新请求立即路由到新实例

    # Step 5：优雅停止旧实例（锁外，不影响其他操作）
    await self._graceful_stop_old_instance(old_instance, agent_id)
    # 若有活跃 SSE 流 → 延迟到流结束后再 stop（用户无感知）
    # 若无活跃任务   → 立即 stop

    return True

⚠️ 单用户设计限制 → 多租户改造建议

功能研究

聊天模块

状态：✅ 分析完成
源码位置：src/copaw/agents/react_agent.py、src/copaw/app/routers/messages.py
更新时间：2026-03-24（v0.1.0 更新）

功能描述

CoPaw 聊天的核心是一个 ReAct 循环（Reason-Act）：用户发消息 → Agent 思考 → 决定是否调工具 → 工具结果再推理 → 最终输出文本。最大迭代 max_iters=50，流式输出。

消息完整生命周期

用户发消息
      ↓
Channel Adapter 封装 → { from, content, channel, ts }
      ↓
Message Queue（去重 + FIFO 排序）
      ↓
HTTP POST /api/v1/messages → FastAPI :8088
      ↓
pre_reasoning Hook
  ├── 读取 PROFILE.md + session history（ReMeLight 提供）
  ├── compact_tool_result（压缩过长工具输出）
  └── check_context → 超 80% token 则 compact_memory
      ↓
rebuild_sys_prompt()
  ├── Agent Identity（agent_id 标识）
  ├── AGENTS.md（基础人设，必须）
  ├── SOUL.md（性格设定，必须）
  ├── PROFILE.md（用户画像，可选）
  └── 激活的 SKILL.md 内容（注入 <skill> 标签）
      ↓
CoPawAgent.reply(msg)  [ToolGuardMixin + ReActAgent]
  ├── LLM 调用（POST /v1/messages，stream=True）
  ├── 返回 type:tool_use → ToolGuard 检查 → 执行工具 → 追加 tool_result → 再次 LLM
  └── 返回 type:text → 过滤幽灵 tool_use → 输出最终回复
      ↓
post_reasoning Hook
  ├── 追加写入 session history
  └── 判断是否更新 PROFILE.md
      ↓
finally: save_session_state() → JSON 落盘
      ↓
Channel Adapter 推送回复 → 用户

Network 抓包重点

核心逻辑

Prompt 拼装（prompt.py）

# v0.1.0 PromptBuilder 构建顺序
def build_system_prompt_from_working_dir(working_dir, agent_id, heartbeat_enabled):
    parts = []

    # 1. Agent Identity（v0.1.0 新增，多 Agent 隔离）
    parts.append(f"# Agent Identity
Your agent id is `{agent_id}`.")

    # 2. 按 system_prompt_files 列表加载（可自定义）
    for filename in ["AGENTS.md", "SOUL.md", "PROFILE.md"]:
        path = working_dir / filename
        if path.exists():
            content = path.read_text()
            # 去除 YAML frontmatter
            if content.startswith("---"):
                content = content.split("---", 2)[2].strip()
            parts.append(f"# {filename}

{content}")

    # 3. env_context 追加（可注入租户权限上下文）
    if env_context:
        parts.append(env_context)

    return "

".join(parts)

ReAct 循环（react_agent.py）

class CoPawAgent(ToolGuardMixin, ReActAgent):
    # max_iters 可通过 AgentProfileConfig.running.max_iters 配置（默认 50）

    async def reply(self, msg):
        # 1. 处理文件/媒体块
        await process_file_and_media_blocks_in_message(msg)

        # 2. /command 路由（/compact、/new 等系统命令短路处理）
        if self.command_handler.is_command(query):
            return await self.command_handler.handle_command(query)

        # 3. 正常 ReAct 循环（由父类 ReActAgent 驱动）
        return await super().reply(msg=msg)

    async def _reasoning(self, tool_choice=None):
        try:
            return await super()._reasoning(tool_choice=tool_choice)
        except Exception as e:
            # 媒体块降级：模型不支持图片/音频时自动剥离并重试
            if self._is_bad_request_or_media_error(e):
                self._strip_media_blocks_from_memory()
                return await super()._reasoning(tool_choice=tool_choice)

提炼规则

• 规则1：Prompt 拼装顺序固定：Agent Identity → AGENTS.md → SOUL.md → PROFILE.md → env_context（这个顺序影响 LLM 对各部分内容的权重）
• 规则2：工具调用是纯 LLM 决策，不做硬编码路由——LLM 返回 tool_use 才调工具
• 规则3：max_iters=50 是安全上限，正常对话 1-3 轮，复杂任务 5-10 轮，超限后追加双语提示
• 规则4：流式输出从 LLM 第一个 token 开始推，_in_summarizing 状态防止幽灵 tool_use 块闪烁
• 规则5：每次 reply 都重新 rebuild_sys_prompt()，不缓存——保证每次对话都拿到最新 PROFILE 和权限上下文

公司改造点

频道路由（Channel）

状态：✅ 分析完成
源码位置：src/copaw/app/channels/、src/copaw/app/channels/manager.py
更新时间：2026-03-24（v0.1.0 新增 Voice/小艺/Matrix/Mattermost/MQTT）

功能描述

Channel 系统是 CoPaw 的多渠道统一接入层。通过 BaseChannel 抽象基类 + ChannelManager 注册表，把钉钉、飞书、QQ、Discord、iMessage 等平台封装成可插拔的渠道插件，统一经由消息队列对接 Agent 核心。

架构设计

BaseChannel 接口（v0.1.0 增强）

class BaseChannel(ABC):
    channel: ChannelType    # 渠道唯一标识

    def __init__(self, process, on_reply_sent=None,
                 show_tool_details=True,    # 是否向用户展示工具调用细节
                 filter_tool_messages=False, # 过滤工具消息（用户侧简洁）
                 filter_thinking=False,      # 过滤思维链（不暴露推理过程）
                 dm_policy="open",           # 私聊策略：open/allowed_only/closed
                 group_policy="open",        # 群聊策略：open/allowed_only/closed
                 allow_from=[],              # 用户/群组白名单
                 deny_message="",            # 拒绝时的提示语
                 require_mention=False):     # 群聊是否需要 @机器人
        ...

渠道详细对比（v0.1.0 完整版）

消息队列机制

# channel_manager.py 核心逻辑（v0.1.0）
class ChannelManager:
    def on_message(self, channel_type, raw_msg):
        # 1. 消息策略检查（dm/group policy、白名单、@require）
        if not self._check_policy(channel_type, raw_msg):
            channel.send_deny_message(raw_msg.from_user)
            return

        # 2. 构建 AgentRequest（统一格式）
        request = AgentRequest(
            session_id  = f"{channel_type}:{raw_msg.from_user}",
            user_id     = raw_msg.from_user,
            channel     = channel_type,
            messages    = [raw_msg.to_msg()],
        )

        # 3. 分发给对应 Workspace 的 Runner
        workspace = await multi_agent_manager.get_agent(agent_id)
        await workspace.runner.handle_request(request)

Tunnel（外网穿透）配置

# v0.1.0 使用 Cloudflare Tunnel（替代 ngrok）
# src/copaw/tunnel/cloudflare.py

# config.json
{
  "tunnel": {
    "provider": "cloudflare",  # cloudflare / ngrok / none
    "token": "${CF_TUNNEL_TOKEN}"
  }
}

# 启动后自动将 public_url 写入各渠道 Webhook 配置
# 本地 :8088 → https://xxx.trycloudflare.com

提炼规则

• 规则1：渠道插件化——BaseChannel ABC 保证接口一致，新增渠道不改 Agent 代码
• 规则2：统一消息格式——所有渠道消息封装为 AgentRequest，Agent 不感知渠道差异
• 规则3：消息策略前置——dm_policy/group_policy 在进入队列前就过滤，Agent 不处理无权限消息
• 规则4：filter_thinking=True 过滤思维链——用户侧不暴露内部推理过程，更专业
• 规则5：Tunnel 和 Channel 解耦——Tunnel 只负责暴露端口，Channel 负责解析消息

公司改造点

工作区（Working Directory）

状态：✅ 分析完成
源码位置：src/copaw/config/config.py、src/copaw/app/workspace/workspace.py、src/copaw/cli/init_cmd.py
更新时间：2026-03-24（v0.1.0 引入 MultiAgentManager，每 agent_id 独立 workspace_dir）

功能描述

Working Directory（工作区）是 CoPaw 的数据根目录，所有持久化数据（记忆、配置、Skill、Cron 任务）都以文件形式存放在这里。v0.1.0 起每个 Agent 拥有独立工作区，默认路径 ~/.copaw/，支持自定义。

完整目录结构

~/.copaw/{agent_id}/          ← v0.1.0：每个 Agent 独立目录（多租户隔离基础）
│
├── AGENTS.md                 ← ★ 基础人设（必须，Agent 的"职责说明书"）
├── SOUL.md                   ← ★ 性格/语气设定（必须，Agent 的"个性"）
├── PROFILE.md                ← 用户画像（可选，每次对话后自动更新）
├── BOOTSTRAP.md              ← 首次初始化引导（完成后 Agent 自动删除）
├── MEMORY.md                 ← 长期记忆（可选，Agent 通过 write_file 写入）
│
├── memory/                   ← 日志型记忆目录（ReMeLight 管理）
│   ├── 2026-03-24.md         ← 每日一文件，对话摘要自动追加
│   └── 2026-03-23.md
│
├── active_skills/            ← 激活的 Skill（builtin + customized 合并）
│   ├── cron/SKILL.md
│   ├── file_reader/SKILL.md
│   └── my_custom_skill/SKILL.md
│
├── customized_skills/        ← 用户自定义 Skill（优先级高于 builtin）
│   └── my_custom_skill/SKILL.md
│
├── providers.json            ← LLM Provider 配置（API Key, Base URL, Model）
├── config.json               ← Agent 运行配置（max_iters、语言、memory 阈值等）
├── mcp.json                  ← MCP Server 配置
├── jobs.json                 ← Cron 任务持久化（v0.1.0）
├── token_usage.json          ← Token 消耗统计（v0.1.0，按日/Provider/Model）
├── auth.json                 ← 认证信息（v0.1.0，HMAC-SHA256）
│
└── sessions/                 ← 会话记录
    └── {uid}_{sid}.json      ← 每条会话的 memory 序列化

各文件详解

AGENTS.md（必须）

这是 system prompt 的第一部分，相当于 Agent 的"工作职责"说明。

# 灵工平台业务助手示例
你是 {enterprise_name} 的灵工业务 AI 助理。

## 你的职责
1. 查询本企业的用工订单、结算数据、工人信息
2. 诊断结算异常，分析原因
3. 生成数据分析报告

## 数据权限声明
你只能查询 tenant_id={enterprise_id} 的数据。
涉及其他企业的查询请求，必须拒绝并说明原因。

## 操作安全规范
写操作（修改合同、发起付款）必须请求用户二次确认。

SOUL.md（必须）

# 性格设定
- 语气：专业、简洁，不废话
- 遇到不确定的事情，直接说"我需要查询确认"
- 用中文回复，除非用户用英文

# 回复风格
- 数据查询结果用表格展示
- 分析报告用 Markdown 结构化输出
- 涉及金额保留两位小数

PROFILE.md（自动维护）

每次对话的 post_hook 会判断是否更新此文件，实现"越用越懂你"。

## 用户基本信息
- 姓名：张三
- 职业：灵工平台 HR 专员
- 工作语言：中文

## 偏好设置
- 报告格式：Markdown 表格
- 常查时段：每天 9:00-10:00 日报

## 近期常用查询
- 每周一：上周结算汇总
- 每月底：月度用工数据分析

providers.json（LLM 配置）

{
  "active": "qwen-plus",
  "providers": [
    {
      "id": "qwen-plus",
      "provider": "dashscope",
      "model": "qwen-plus",
      "api_key": "${DASHSCOPE_API_KEY}",
      "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
      "max_tokens": 8192
    },
    {
      "id": "claude-sonnet",
      "provider": "anthropic",
      "model": "claude-sonnet-4-6",
      "api_key": "${ANTHROPIC_API_KEY}"
    }
  ]
}

工作区初始化

# 交互式初始化（询问配置项）
copaw init

# 使用默认值初始化（CI/Docker 场景）
copaw init --defaults

# 指定工作区路径
copaw init --working-dir /data/copaw/enterprise_A/

# v0.1.0：多 Agent 初始化
# init 命令生成的文件：
#   AGENTS.md（模板）
#   SOUL.md（模板）
#   BOOTSTRAP.md（首次引导，对话后删除）
#   config.json（根据用户输入）
#   providers.json（根据用户输入的 API Key）

文件读写安全

workspace_dir/
  ├── AGENTS.md     ✅ Agent 只读（人工配置，热更新）
  ├── PROFILE.md    ✅ Agent 可读写（post_hook 自动更新）
  ├── sessions/     ✅ Agent 可读写（会话持久化）
  └── memory/       ✅ Agent 可读写（记忆追加）

.secrets/ 或 envs.json（加密）
  └── env           ❌ Agent 不可访问，只通过 ${ENV_VAR} 引用

安全原则：敏感配置（API Key、数据库密码）通过 envs.json 加密存储，不进入任何 Prompt，只通过环境变量访问。

提炼规则

• 规则1：所有持久化数据都是 Markdown 文件，人可读、可用编辑器直接修改——这是 CoPaw"透明化"设计的核心
• 规则2：active_skills/ 是 builtin 和 customized Skill 的合并结果，customized 优先，builtin 不覆盖 customized
• 规则3：providers.json 支持多 Provider，active 字段指定当前生效的，切换模型不重启
• 规则4：工作区 = 数据隔离单元——一个 agent_id 对应一个独立目录，天然实现企业隔离
• 规则5：Secret（API Key）通过 ${ENV_VAR} 引用，不明文写入配置文件

公司改造点

定时任务（Cron + Heartbeat）

状态：✅ 分析完成
源码位置：src/copaw/app/crons/、src/copaw/agents/hooks/bootstrap.py
更新时间：2026-03-24（v0.1.0 重构为独立 CronManager，支持 text/agent 双类型）

功能描述

CoPaw 有两套定时机制，职责不同：

Cron 系统工作原理

copaw cron create → 写入 jobs.json（workspace_dir）
      ↓
CronManager（后台 asyncio 任务）
      ↓
按 cron 表达式触发 → 构造虚拟消息给 AgentRunner
      ↓
AgentRunner 按正常 ReAct 循环处理
      ↓
结果推送到指定 channel（钉钉/飞书/企微等）

CLI 操作（v0.1.0）

# 创建任务（type=text：定时发送固定消息）
copaw cron create   --agent-id enterprise_A \          # ★ 必须指定，多租户关键
  --type text   --name "工作日早安提醒"   --cron "0 9 * * 1-5"   --channel dingtalk   --target-user "manager_id"   --text "早上好！今日有 N 条待审批订单。"

# 创建任务（type=agent：定时向 Agent 提问并推送回答）
copaw cron create   --agent-id enterprise_A   --type agent   --name "日报生成"   --cron "0 18 * * 1-5"   --channel dingtalk   --target-user "manager_id"   --text "请生成今日灵工业务数据摘要"

# 管理操作
copaw cron list   --agent-id enterprise_A  # 查看所有任务
copaw cron state  <job_id>                 # 查看任务状态
copaw cron pause  <job_id>                 # 暂停任务
copaw cron resume <job_id>                 # 恢复任务
copaw cron delete <job_id>                 # 删除任务
copaw cron run    <job_id>                 # 立即执行（测试用）

数据结构（jobs.json）

// workspace_dir/jobs.json
{
  "jobs": [
    {
      "id": "job_abc123",
      "name": "daily_report",
      "type": "agent",          // "text" | "agent"
      "cron": "0 18 * * 1-5",
      "prompt": "请生成今日灵工业务数据摘要",
      "channel": "dingtalk",
      "target_user": "manager_id",
      "target_session": "session_xxx",
      "agent_id": "enterprise_A",  // v0.1.0：明确绑定到工作区
      "status": "active",
      "created_at": "2026-03-01T10:00:00Z",
      "last_run": "2026-03-24T18:00:00Z",
      "next_run": "2026-03-25T18:00:00Z"
    }
  ]
}

Heartbeat 系统（AGENTS.md 配置）

v0.1.0 中 Heartbeat 通过 AGENTS.md 的 heartbeat 段配置，由 BootstrapHook 管理。

# AGENTS.md 中的 heartbeat 段（带 HTML 注释开关）
<!-- heartbeat:start -->
## 主动推送配置
- 每天早 8 点，主动发送今日日程提醒到钉钉
- 每周一早 9 点，汇总上周灵工数据发送给管理员
<!-- heartbeat:end -->

# config.json 中的 heartbeat 开关
{
  "heartbeat": {
    "enabled": true,    // false 时，heartbeat 段从 system prompt 中剔除
    "interval": 60      // 检查间隔（秒）
  }
}

Cron 表达式参考

┌──分钟 (0-59)
│  ┌──小时 (0-23)
│  │  ┌──日 (1-31)
│  │  │  ┌──月 (1-12)
│  │  │  │  ┌──周 (0-7, 0和7都是周日)
│  │  │  │  │
*  *  *  *  *

常用示例：
"0 9 * * *"      每天早 9 点
"0 9 * * 1-5"    工作日早 9 点
"0 9,18 * * *"   每天 9 点和 18 点
"*/30 * * * *"   每 30 分钟
"0 9 * * 1"      每周一早 9 点
"0 0 1 * *"      每月 1 日零点（月报）

提炼规则

• 规则1：Cron 任务通过"注入虚拟消息"接入 Agent，复用完整 ReAct 流程，不需要单独执行路径
• 规则2：agent_id 必须显式指定——多租户场景下，错误的 agent_id 会把任务创建到错误企业的工作区
• 规则3：Heartbeat 通过 AGENTS.md heartbeat 段配置，heartbeat.enabled=false 时段内容自动从 Prompt 剔除，避免干扰
• 规则4：Cron 调度基于系统时间，Agent 重启后从 jobs.json 恢复任务，不丢失
• 规则5：copaw cron run <job_id> 可立即触发，方便调试，不等 cron 时间

公司改造点

运行配置（config.json）

状态：✅ 分析完成
源码位置：src/copaw/config/config.py、src/copaw/config/utils.py
更新时间：2026-03-24（v0.1.0 配置结构重构，新增 security/embedding/heartbeat 等配置块）

功能描述

CoPaw 的所有运行参数通过 config.json 配置，涵盖 Agent 行为、渠道设置、记忆阈值、安全规则等核心参数。v0.1.0 中每个 Agent 工作区有独立配置，支持热更新。

完整配置结构（v0.1.0）

// workspace_dir/config.json（单个 Agent 配置）
{
  "agent": {
    "language": "zh",               // 记忆摘要和提示语语言（zh/en/ru）
    "max_iters": 50,                // ReAct 最大迭代次数
    "working_dir": "~/.copaw/",     // 工作区路径
    "timezone": "Asia/Shanghai",    // 影响 Cron 时间、日志时间戳
    "system_prompt_files": [        // 自定义加载哪些 Prompt 文件
      "AGENTS.md", "SOUL.md", "PROFILE.md"
    ]
  },
  "running": {
    "max_input_length": 131072,       // 最大输入 token 数
    "memory_compact_threshold": 0,    // 0 = 使用 compact_ratio 计算
    "memory_compact_ratio": 0.75,     // token 占比超此值触发压缩
    "compact_with_thinking_block": false,
    "embedding_config": {
      "backend": "openai",
      "api_key": "",                  // 优先级：config > EMBEDDING_API_KEY
      "base_url": "",                 // EMBEDDING_BASE_URL
      "model_name": "",               // EMBEDDING_MODEL_NAME
      "dimensions": 1024,
      "enable_cache": true
    }
  },
  "heartbeat": {
    "enabled": false,               // true 时激活 AGENTS.md heartbeat 段
    "interval": 60                  // 检查间隔（秒）
  },
  "tools": {
    "builtin_tools": {
      "execute_shell_command": {"enabled": true},
      "read_file":             {"enabled": true},
      "write_file":            {"enabled": true},
      "edit_file":             {"enabled": true},
      "grep_search":           {"enabled": true},
      "glob_search":           {"enabled": true},
      "browser_use":           {"enabled": false},  // 多租户建议关闭
      "desktop_screenshot":    {"enabled": false},
      "view_image":            {"enabled": true},
      "send_file_to_user":     {"enabled": true},
      "get_current_time":      {"enabled": true},
      "set_user_timezone":     {"enabled": true},
      "get_token_usage":       {"enabled": true}
    }
  },
  "security": {
    "tool_guard": {
      "enabled": true,
      "guarded_tools": null,          // null = 所有工具都在守卫范围
      "denied_tools": [               // 黑名单，直接拒绝
        "execute_shell_command"        // 多租户场景强烈建议加入
      ]
    },
    "skill_scanner": {
      "mode": "block"                 // block/warn：block 时扫描失败拒绝激活
    }
  }
}

关键配置项说明

Agent 语言（agent.language）

影响 ReMeLight 生成记忆摘要时使用的语言，以及 BootstrapHook 引导文本语言。

"language": "zh"  // 记忆摘要、引导文本用中文（中文用户必须设置）
"language": "en"  // 英文（默认）
"language": "ru"  // 俄文（支持国际化）

记忆压缩阈值（running.memory_compact_ratio）

# 压缩逻辑
max_input = model.max_input_length   # 如 128K
threshold = max_input * compact_ratio  # 128K × 0.75 = 96K

当前 token 数 > threshold → 触发 MemoryCompactionHook

调参建议：
  0.75（默认）：75% 时压缩，比较保守，适合长对话
  0.65：更激进，更频繁压缩，节省 token，适合简单查询场景
  0.85：更充分利用 context，但有溢出风险

工具按 Agent 独立开关（tools.builtin_tools）

环境变量（envs.json 加密存储）

# src/copaw/envs/store.py — 加密存储，不明文暴露
# 环境变量通过 ${ENV_VAR} 在配置文件中引用

常用变量：
  DASHSCOPE_API_KEY=sk-xxx       # 阿里云模型
  OPENAI_API_KEY=sk-xxx          # OpenAI
  ANTHROPIC_API_KEY=sk-xxx       # Anthropic
  DINGTALK_CLIENT_ID=xxx         # 钉钉
  DINGTALK_CLIENT_SECRET=xxx
  FEISHU_APP_ID=xxx              # 飞书
  EMBEDDING_API_KEY=sk-xxx       # 向量化（可选，向量搜索功能）
  EMBEDDING_BASE_URL=https://...
  EMBEDDING_MODEL_NAME=text-embedding-v4
  FTS_ENABLED=true               # BM25 全文搜索
  MEMORY_STORE_BACKEND=auto      # chroma / local
  ENABLE_MEMORY_MANAGER=true     # 是否启用记忆管理器
  COPAW_AUTH_ENABLED=true        # v0.1.0：启用 Web 认证
  COPAW_AUTH_USERNAME=admin
  COPAW_AUTH_PASSWORD=xxx

CLI 完整启动流程

copaw app   # 启动 CoPaw 服务

# 启动时的初始化顺序：
# 1. 加载 config.json（所有 Agent 配置）
# 2. 解析 envs.json（解密环境变量）
# 3. 初始化 LLM Providers（按 providers.json）
# 4. 启动 MultiAgentManager（懒加载，首次请求时创建 Workspace）
# 5. 每个 Workspace 按优先级初始化：
#    P10: AgentRunner
#    P20: MemoryManager + MCPManager + ChatManager（并发）
#    P25: runner.start()
#    P30: ChannelManager.start_all()
#    P40: CronManager.start()
#    P50: ConfigWatcher + MCPConfigWatcher
# 6. 启动 Cloudflare Tunnel（如配置）
# 7. 启动 FastAPI :8088
# 8. 启动 Console 前端（:3000）

提炼规则

• 规则1：所有敏感信息（API Key）通过 ${ENV_VAR} 引用，不明文写入 config.json，envs.json 加密存储
• 规则2：memory_compact_ratio 和 max_iters 是核心调参点——影响对话质量和 Token 成本
• 规则3：工具通过 builtin_tools.enabled 按 Agent 独立开关，不同企业可开放不同工具集
• 规则4：language: zh 必须设置，否则记忆摘要默认英文，中文用户体验差
• 规则5：ConfigWatcher 热重载——修改 config.json 后立即生效，无需重启（envs 变更例外）

公司改造点

数据结构

基于 v0.1.0.post1 源码 Pydantic 模型完整提取 · 所有 Schema 均有源码出处

会话对象（ChatSpec / ChatsFile / SafeJSONSession）

源码位置：src/copaw/app/runner/models.py、src/copaw/app/runner/session.py

ChatSpec — 单条会话规格

class ChatSpec(BaseModel):
    id: str          # UUID，唯一标识（default_factory=uuid4）
    name: str        # 会话名称（默认 "New Chat"）
    session_id: str  # 格式："{channel}:{user_id}"（如 "dingtalk:user_123"）
    user_id: str     # 用户标识
    channel: str     # 渠道类型（DEFAULT_CHANNEL = "console"）
    created_at: datetime  # UTC 创建时间
    updated_at: datetime  # UTC 最后更新时间
    meta: dict       # 扩展元数据（可存 tenant_id 等自定义字段）
    status: str      # "idle" | "running"

ChatsFile — 会话索引文件

class ChatsFile(BaseModel):
    version: int = 1
    chats: list[ChatSpec] = []   # 所有会话列表

# 文件路径：workspace_dir/chats.json
# 作用：会话元数据索引（不含消息内容，消息内容在 sessions/ 目录）
# session_id 生成规则：
session_id = f"{channel}:{user_id}"  # 例："dingtalk:user_123"
# 文件名生成规则（SafeJSONSession）：
filename = sanitize_filename(f"{user_id}_{session_id}.json")
# sanitize：把 Windows 非法字符（: * ? " < > | \ /）替换为 "--"
# 例："discord--user_123_discord--dm--456.json"

ChatHistory — 完整会话视图

class ChatHistory(BaseModel):
    messages: list[Message] = []   # 完整消息列表（AgentScope Message）
    status: str = "idle"           # "idle" | "running"

Memory 序列化格式（sessions/*.json）

# 文件内容结构（InMemoryMemory.state_dict()）
{
  "agent": {
    "memory": {
      "content": [
        [{"role": "system",    "content": "...system prompt..."},    []],
        [{"role": "user",      "content": "用户发的消息"},            []],
        [{"role": "assistant", "content": [...tool_use blocks...]},  []],
        [{"role": "user",      "content": [...tool_result blocks...]}, []],
        [{"role": "assistant", "content": "最终回复文本"},   ["tool_guard_denied"]]
        # 二元组：[消息体 dict, 标记列表 marks]
        # marks = ["tool_guard_denied"] 表示该条消息被 ToolGuard 拦截
      ]
    }
  }
}

ChannelAddress — 渠道路由地址

@dataclass
class ChannelAddress:
    kind: str   # "dm" | "channel" | "webhook" | "console" | ...
    id: str     # 目标 ID（用户ID、频道ID、群组ID等）
    extra: dict | None = None   # 扩展字段（如 to_handle、session_webhook）

    def to_handle(self) -> str:
        # 返回格式如："discord:ch:123456"
        return f"{self.kind}:{self.id}"

# 内置渠道类型（BUILTIN_CHANNEL_TYPES）：
# "imessage", "discord", "dingtalk", "feishu", "qq",
# "telegram", "mqtt", "console", "voice", "xiaoyi"

灵工改造建议

消息格式（Message / Content Block）

源码位置：agentscope.message（框架层）、src/copaw/agents/schema.py

Msg 核心结构

# AgentScope Msg 对象（序列化为 role+content 格式）
{
  "role": "system" | "user" | "assistant",
  "content": str | list[ContentBlock],  # 纯文本 或 多模态 Block 列表
  "name": str | None,                   # 可选，发送方名称
  "url": str | None                     # 可选，附件 URL
}

Content Block 类型（多模态）

# 文本块
{"type": "text", "text": "回复内容..."}

# 工具调用块（LLM 决定调用工具时返回）
{
  "type": "tool_use",
  "id": "toolu_xxx",
  "name": "query_business_data",
  "input": {"sql": "SELECT ...", "tenant_id": "enterprise_A"}
}

# 工具结果块（工具执行后追加）
{
  "type": "tool_result",
  "tool_use_id": "toolu_xxx",
  "output": "查询结果：共 42 条记录..."
}

# 图片块
{
  "type": "image",
  "source": {
    "type": "base64",
    "media_type": "image/png",
    "data": "iVBORw0KGgo..."
  }
}

# 文件块（CoPaw 扩展）
{
  "type": "file",
  "source": {"type": "base64" | "url", ...},
  "filename": "report.xlsx"
}

流式响应格式（SSE）

# POST /api/v1/messages 流式响应（Server-Sent Events）
data: {"type": "text_delta", "delta": "正在"}
data: {"type": "text_delta", "delta": "查询"}
data: {"type": "text_delta", "delta": "数据..."}
data: {"type": "tool_use_start", "name": "query_db", "id": "toolu_xxx"}
data: {"type": "tool_use_input_delta", "delta": "{"sql":"}
data: {"type": "tool_use_done"}
data: {"type": "message_done"}

# 关键识别点：
# - type:tool_use → LLM 决定调用工具
# - type:text_delta → 最终文字回复流式输出
# - _in_summarizing 状态下幽灵 tool_use 块被过滤

灵工改造建议

Skill 配置（SkillInfo / SKILL.md Schema）

源码位置：src/copaw/agents/skills_manager.py

SkillInfo Pydantic 模型

class SkillInfo(BaseModel):
    name: str          # 技能名（= 目录名，作为 toolkit 注册标识）
    description: str   # 触发描述（来自 frontmatter，LLM 根据此字段决策何时调用）
    content: str       # SKILL.md 全文内容（包含 frontmatter）
    source: str        # "builtin" | "customized" | "active"
    path: str          # 磁盘绝对路径
    references: dict[str, Any]  # references/ 目录树（见下方说明）
    scripts: dict[str, Any]     # scripts/ 目录树
    # 注：builtin_skill_version 在 frontmatter metadata 中，通过 frontmatter.loads() 解析

SKILL.md frontmatter 完整 Schema

---
name: my_skill                       # 必填：技能标识（同目录名）
description: "触发条件的自然语言描述"  # 必填：LLM 用这个字段决定何时调用此 Skill
homepage: https://github.com/...     # 可选：来源链接
metadata:
  builtin_skill_version: "1.0"       # 内置技能版本（用于升级比对）
  copaw:                             # CoPaw 特有元数据
    emoji: "🔧"
    requires:
      bins: ["my_binary"]            # 依赖的外部二进制
    install:                         # 安装方式声明
      - id: brew
        kind: brew
        formula: my_tool
        bins: ["my_tool"]
        label: "Install via Homebrew"
---
# Skill 正文（Markdown，注入 system prompt）
当用户询问 XXX 时，按以下步骤执行：
1. 调用 references/doc.md 中的规范
2. 执行 scripts/run.sh

目录树格式说明

# 读取时（source="builtin"/"active"）：value = None 表示文件存在
references = {
    "doc.md": None,               # 文件
    "examples": {                 # 目录
        "example1.py": None,
        "data": {"sample.json": None}
    }
}

# 创建时（SkillService.create_skill()）：value = 文件内容字符串
references = {
    "doc.md": "# 规范文档
内容...",
    "examples": {
        "example1.py": "print('hello')"
    }
}

Skill 三层目录状态机

┌─────────────────┐      enable_skill()     ┌──────────────┐
│ builtin_skills/ │ ──────────────────────► │              │
│   (代码包内置)   │                         │ active_skills│
│   只读，v0.1.0  │   create_skill()        │  (运行时激活) │
└─────────────────┘ ──► customized ─────►   │              │
                        _skills/         │   └──────────────┘
┌──────────────────┐   (用户自定义)      │
│  Hub/GitHub/ZIP  │ ──────────────────► │  disable_skill()
│  (在线安装来源)   │                     └─► 从 active 移除
└──────────────────┘

灵工改造建议

MCP 配置（MCPClientConfig / MCPConfig）

源码位置：src/copaw/config/config.py、src/copaw/app/mcp/manager.py

MCPClientConfig — 单个 MCP Server 配置

class MCPClientConfig(BaseModel):
    name: str           # 唯一标识，用作工具名前缀
    description: str = ""
    enabled: bool = True
    transport: Literal["stdio", "streamable_http", "sse"] = "stdio"
    url: str = ""       # HTTP/SSE 时必填（支持 ${ENV_VAR} 引用）
    headers: dict[str, str] = {}   # 请求头（如 Authorization）
    command: str = ""   # stdio 时必填（可执行文件路径）
    args: list[str] = []           # stdio 命令参数
    env: dict[str, str] = {}       # 环境变量（子进程继承）
    cwd: str = ""       # 工作目录（stdio）
    # 别名兼容：isActive → enabled，baseUrl → url

完整 mcp.json 示例（灵工业务场景）

// workspace_dir/mcp.json
{
  "clients": {
    "linggong-query": {
      "name": "linggong-query",
      "description": "灵工平台只读查询接口",
      "enabled": true,
      "transport": "sse",
      "url": "https://mcp.linggong.com/query/sse",
      "headers": {
        "Authorization": "Bearer ${LINGGONG_API_TOKEN}",
        "X-Enterprise-Id": "${ENTERPRISE_ID}",   // 租户上下文注入
        "X-User-Role": "${USER_ROLE}"
      }
    },
    "linggong-analysis": {
      "name": "linggong-analysis",
      "description": "灵工平台数据分析接口（需高级权限）",
      "enabled": true,
      "transport": "streamable_http",
      "url": "https://mcp.linggong.com/analysis/mcp",
      "headers": {
        "Authorization": "Bearer ${LINGGONG_ADMIN_TOKEN}",
        "X-Enterprise-Id": "${ENTERPRISE_ID}"
      }
    },
    "local-tools": {
      "name": "local-tools",
      "description": "本地脚本工具集",
      "enabled": false,
      "transport": "stdio",
      "command": "python",
      "args": ["-m", "my_mcp_server"],
      "env": {"PYTHONPATH": "/app/tools"},
      "cwd": "/app"
    }
  }
}

MCP rebuild_info（v0.1.0 自动重连元数据）

# Client 对象上附加的重建信息（用于连接断开后自动重建）
client._copaw_rebuild_info = {
    "transport": "sse",           # 传输协议
    "name": "linggong-query",     # Server 名称
    "url": "https://...",         # HTTP/SSE 地址
    "headers": {...},             # 请求头
    # stdio 额外字段：
    "command": "python",
    "args": ["-m", "server"],
    "env": {...},
    "cwd": "/app"
}
# 重连顺序：reconnect() → rebuild_info 重建 client → 再次 connect()

MCP Server 工具设计规范（灵工业务场景）

# 正确写法：enterprise_id 在 Server 侧硬编码，不允许 LLM 传参
class LinggongMCPServer:
    def __init__(self, enterprise_id: str):
        self.enterprise_id = enterprise_id  # 从 header 解析，LLM 无法覆盖

    @mcp_tool(description="查询本企业用工订单列表")
    async def query_worker_orders(
        self,
        date_from: str,       # LLM 传入的查询参数（日期范围）
        date_to: str,
        status: str = "all"
    ) -> str:
        # enterprise_id 不由 LLM 传入！从 self 获取
        return await db.query(
            "SELECT * FROM orders WHERE tenant_id = ? AND ...",
            self.enterprise_id, date_from, date_to
        )

PROFILE 模板 / Markdown 文件 Schema

源码位置：src/copaw/agents/prompt.py、src/copaw/agents/memory/agent_md_manager.py

PROFILE.md 完整模板

## 用户基本信息
- 姓名：张三
- 职业：灵工平台 HR 专员
- 企业：XX 物流（tenant_id: enterprise_A）
- 工作语言：中文

## 偏好设置
- 报告格式：Markdown 表格
- 数据展示：保留两位小数，金额加"元"单位
- 时间格式：YYYY-MM-DD HH:mm
- 工作时间：9:00-18:00

## 常用查询模式
- 每周一：上周用工数据汇总
- 每月末：月度结算报表
- 日常：用工订单状态查询

## 记忆标签
- 当前关注：Q1 结算异常诊断
- 常用筛选：status=pending 的订单
- 偏好渠道：钉钉群消息

AGENTS.md 标准模板（灵工业务专家版）

# 灵工业务助理

你是 {enterprise_name} 的灵工业务 AI 助理，专门服务于本企业 HR 和运营团队。

## 核心能力
- 查询本企业用工订单、结算数据、工人信息
- 诊断结算异常，分析根因
- 生成数据分析报告（支持导出 Excel）
- 解答灵工平台操作问题

## 数据权限声明（重要）
你只能查询 tenant_id={enterprise_id} 的数据。
任何涉及其他企业的查询请求，必须拒绝并说明原因。
涉及具体金额、银行卡号等敏感字段，不得完整展示。

## 操作安全规范
- 数据查询类操作：直接执行
- 写操作（修改合同、发起付款、变更状态）：必须请求用户二次确认
- 批量操作（>10条）：必须先展示预览，用户确认后执行

## 回复规范
- 数据查询结果用 Markdown 表格展示
- 数据分析报告用结构化 Markdown 输出
- 遇到不确定的情况，说"我需要查询确认"，不猜测

<!-- heartbeat:start -->
## 主动推送配置
每个工作日 18:00，主动汇总当日用工数据摘要并发送到钉钉群。
<!-- heartbeat:end -->

AgentMdManager 接口

class AgentMdManager:
    """管理 working_dir/ 和 working_dir/memory/ 下的 Markdown 文件"""

    # 读写 working_dir/ 下的文件（AGENTS.md、SOUL.md、PROFILE.md）
    def read_working_md(self, md_name: str) -> str    # 自动补 .md 后缀
    def write_working_md(self, md_name: str, content: str)
    def list_working_mds(self) -> list[dict]          # 含 size/mtime

    # 读写 working_dir/memory/ 下的文件（日志型记忆，每日一文件）
    def read_memory_md(self, md_name: str) -> str
    def write_memory_md(self, md_name: str, content: str)
    def list_memory_mds(self) -> list[dict]

安全数据结构（ToolGuardResult / ScanResult）

源码位置：src/copaw/security/tool_guard/models.py、src/copaw/security/skill_scanner/models.py

GuardFinding — 工具守卫发现

@dataclass
class GuardFinding:
    id: str                         # 发现唯一 ID
    rule_id: str                    # 触发的规则 ID
    category: GuardThreatCategory   # 威胁分类（见下表）
    severity: GuardSeverity         # CRITICAL | HIGH | MEDIUM | LOW | INFO | SAFE
    title: str                      # 发现标题
    description: str                # 详细描述
    tool_name: str                  # 被拦截的工具名
    param_name: str | None          # 触发规则的参数名
    matched_value: str | None       # 匹配到的值（脱敏）
    matched_pattern: str | None     # 触发的正则模式
    snippet: str | None             # 上下文片段
    remediation: str | None         # 修复建议
    guardian: str | None            # 产生此发现的 Guardian 名称

GuardThreatCategory 威胁分类

ToolGuardResult — 单次工具调用守卫结果

@dataclass
class ToolGuardResult:
    tool_name: str                  # 被守卫的工具名
    params: dict                    # 工具调用参数
    findings: list[GuardFinding]    # 所有发现列表
    guard_duration_seconds: float   # 守卫耗时
    guardians_used: list[str]       # 参与的 Guardian 名称列表
    guardians_failed: list[dict]    # 失败的 Guardian（{name, error}）
    timestamp: datetime             # UTC 时间戳

    @property
    def is_safe(self) -> bool:
        # True = 无 CRITICAL 或 HIGH 级别发现
        return not any(f.severity in (CRITICAL, HIGH) for f in self.findings)

    @property
    def max_severity(self) -> GuardSeverity:
        # 返回最高严重等级，无发现时返回 SAFE

ScanResult — Skill 扫描结果

@dataclass
class ScanResult:
    skill_name: str
    skill_directory: str
    findings: list[Finding]         # 扫描发现列表（同 GuardFinding 结构）
    scan_duration_seconds: float
    analyzers_used: list[str]       # 使用的分析器（PatternAnalyzer 等）
    analyzers_failed: list[dict]
    timestamp: datetime

    @property
    def is_safe(self) -> bool:
        # True = 无 CRITICAL 或 HIGH 级别发现（可安全激活）

SkillScanner ThreatCategory（技能扫描威胁分类）

完整配置 Schema（AgentProfileConfig）

源码位置：src/copaw/config/config.py（Pydantic BaseModel，强类型）

AgentProfileConfig — 完整结构

class AgentProfileConfig(BaseModel):
    id: str               # 唯一 Agent ID（对应 workspace 目录名）
    name: str             # 人类可读名称
    description: str = ""
    workspace_dir: str = ""   # 工作区路径

    # 渠道配置（各渠道独立配置）
    channels: ChannelConfig | None = None
    mcp: MCPConfig | None = None              # MCP Server 列表
    heartbeat: HeartbeatConfig | None = None  # 主动推送配置
    last_dispatch: LastDispatchConfig | None = None

    running: AgentsRunningConfig   # 运行时行为配置（见下）
    llm_routing: AgentsLLMRoutingConfig    # LLM 路由策略
    active_model: ModelSlotConfig | None   # 当前激活模型
    language: str = "zh"           # 记忆摘要语言（zh/en/ru）
    system_prompt_files: list[str] = ["AGENTS.md", "SOUL.md", "PROFILE.md"]
    tools: ToolsConfig | None = None       # 内置工具开关
    security: SecurityConfig | None = None # 安全配置

AgentsRunningConfig — 运行时行为配置

class AgentsRunningConfig(BaseModel):
    max_iters: int = 50              # ReAct 最大迭代次数（防死循环）
    llm_retry_enabled: bool = True   # LLM 调用失败时自动重试
    llm_max_retries: int = 3         # 最大重试次数
    llm_backoff_base: float = 1.0    # 退避基础延迟（秒）
    llm_backoff_cap: float = 10.0    # 退避最大延迟（秒）
    token_count_model: str = "default"
    token_count_estimate_divisor: float = 3.75  # 字符→token 估算除数

    # 记忆压缩（来自 config.json running 节）
    max_input_length: int            # 最大输入 token 数（按模型设置）
    memory_compact_ratio: float = 0.75   # token 达到此比例触发压缩
    compact_with_thinking_block: bool = False
    embedding_config: EmbeddingConfig    # 向量化配置（可选）

BuiltinToolConfig — 内置工具配置

class BuiltinToolConfig(BaseModel):
    name: str          # 工具函数名
    enabled: bool = True           # 是否启用（False = 完全不加载）
    description: str = ""
    display_to_user: bool = True   # 工具输出是否渲染给用户

# 13 种内置工具默认配置（_default_builtin_tools()）：
{
    "execute_shell_command": BuiltinToolConfig(name=..., enabled=True),
    "read_file":             BuiltinToolConfig(name=..., enabled=True),
    "write_file":            BuiltinToolConfig(name=..., enabled=True),
    "edit_file":             BuiltinToolConfig(name=..., enabled=True),
    "grep_search":           BuiltinToolConfig(name=..., enabled=True),
    "glob_search":           BuiltinToolConfig(name=..., enabled=True),
    "browser_use":           BuiltinToolConfig(name=..., enabled=True),
    "desktop_screenshot":    BuiltinToolConfig(name=..., enabled=True),
    "view_image":            BuiltinToolConfig(name=..., enabled=True),
    "send_file_to_user":     BuiltinToolConfig(name=..., enabled=True),
    "get_current_time":      BuiltinToolConfig(name=..., enabled=True),
    "set_user_timezone":     BuiltinToolConfig(name=..., enabled=True),
    "get_token_usage":       BuiltinToolConfig(name=..., enabled=True),
}
# 注：memory_search 不在此列，由 MemoryManager 动态注册

CronJobSpec — 定时任务完整 Schema

class CronJobSpec(BaseModel):
    id: str                      # 任务唯一 ID
    name: str                    # 任务名称
    enabled: bool = True

    schedule: ScheduleSpec       # 调度配置
    task_type: "text" | "agent"  # 任务类型
    text: str | None             # task_type="text" 时的固定消息内容
    request: CronJobRequest | None  # task_type="agent" 时的 Agent 请求

    dispatch: DispatchSpec       # 推送目标配置
    runtime: JobRuntimeSpec      # 运行时参数
    meta: dict = {}

class ScheduleSpec(BaseModel):
    type: "cron" = "cron"
    cron: str         # 标准 5 字段 cron 表达式（自动 normalize 周字段）
    timezone: str = "UTC"

class DispatchSpec(BaseModel):
    type: "channel" = "channel"
    channel: str              # 目标渠道（如 "dingtalk"）
    target: DispatchTarget    # {user_id, session_id}
    mode: "stream" | "final" = "stream"

class JobRuntimeSpec(BaseModel):
    max_concurrency: int = 1       # 最大并发执行数
    timeout_seconds: int = 120     # 任务超时（秒）
    misfire_grace_seconds: int = 60  # 错过触发的容忍窗口（秒）

分析结论

基于对 CoPaw v0.1.0.post1 源码的完整解读（819个文件，2026-03），评估结论：

CoPaw 适配灵工SaaS平台 — 能力评估总览

一、CoPaw 能满足的部分（可直接复用，约 40%）

核心推理引擎是最大的价值所在。CoPawAgent 的 ReAct Loop（推理→调用工具→反馈→循环）完全适配"对话即查询、提问即诊断"的场景，这是整个架构中最难自研的部分，CoPaw 已经做好了。

二、需要改造的部分（约 30% 工作量）

三、完全缺失、需重新设计的部分（最关键风险）

四、v0.1.0 对结论的修正（相较原 v0.0.6 报告）

五、商业模式与产品设计建议

针对「开放给客户使用」的目标，建议分三层产品设计：

定价逻辑建议

六、分阶段实施路线图

七、最终建议

三层架构改造建议

┌────────────────────────────────────────────────────────┐
│                   灵工 SaaS 多租户层（全新建）           │
│  企业SSO认证 · 租户路由 · Token计费 · Web嵌入组件        │
├────────────────────────────────────────────────────────┤
│              CoPaw 改造层（最小侵入改造）               │
│  AgentRequest + tenant_id                              │
│  rebuild_sys_prompt + env_context 权限注入              │
│  ToolGuard + 业务 ACL 规则                             │
│  TokenUsageManager + 租户维度 + 配额控制                │
├────────────────────────────────────────────────────────┤
│              CoPaw 核心层（直接复用，不改）             │
│  CoPawAgent(ReAct) · Skill机制 · MCP协议              │
│  Provider路由 · 记忆压缩 · Cron任务 · 14种渠道         │
│  Workspace隔离模型 · ToolGuard框架 · SkillScanner      │
└────────────────────────────────────────────────────────┘

风险清单