Hermes = 执行层
daemon = 底层协调控制层
Manas  = 高层全局监督 / 反思 / 裁决层

Hermes 做事。
daemon 管进程、日志、cron、控制指令。
Manas 理解意义、判断方向、决定是否干预。

                         用户 / Discord
                              │
                              ▼
                    ┌──────────────────┐
                    │   Hermes Main     │
                    │ Discord 对话执行层 │
                    └────────┬─────────┘
                             │
                 gateway hook / plugin hook
                             │
                             ▼
┌──────────────────────────────────────────────────────┐
│                    Event Bus                          │
│              ~/.hermes/manas/events.sqlite            │
│                                                      │
│  记录：Discord 消息、Hermes 回复、工具调用、cron 状态、错误 │
└───────────────┬──────────────────────────────┬───────┘
                │                              │
                ▼                              ▼
      ┌──────────────────┐          ┌──────────────────┐
      │      daemon       │          │      Manas        │
      │ 底层 watcher/control │          │ 高层反思/裁决/策略 │
      └────────┬─────────┘          └────────┬─────────┘
               │                             │
               │  执行控制指令                 │  写入决策
               ▼                             ▼
┌──────────────────────────────────────────────────────┐
│                 Control Plane                         │
│             ~/.hermes/manas/control.sqlite            │
│                                                      │
│  pause_cron / resume_cron / block_tool / inject_context │
│  stop_gateway / kill_process / restart_profile          │
└──────────────────────────────────────────────────────┘

Manas 不直接替 Hermes 聊天。
Manas 观察 Hermes。
Manas 指挥 daemon。
daemon 控制 Hermes。
Hermes 服务用户。

1. 用户发给 Discord bot 的消息
2. Hermes 对用户的回复
3. Hermes 的工具调用
4. Hermes 的工具结果
5. cron 即将运行
6. cron 已运行 / 失败 / 输出
7. gateway 启动 / 停止 / 重启
8. CLI agent / 子 agent / delegate_task 状态
9. 危险命令审批事件
10. 进程卡死、循环、长时间无响应

~/.hermes/
  manas/
    events.sqlite
    control.sqlite
    reflections.md
    state.json
    policies.yaml
    lib/
      manas_bus.py
    bin/
      init_manas.py
      manas_supervisor.py
      daemon_cron_watcher.py
      daemon_actuator.py
      daemon_process_watcher.py
 
  hooks/
    manas-gateway-feed/
      HOOK.yaml
      handler.py
 
  plugins/
    manas-control/
      plugin.yaml
      __init__.py

mkdir -p ~/.hermes/manas/{bin,lib}
mkdir -p ~/.hermes/hooks/manas-gateway-feed
mkdir -p ~/.hermes/plugins/manas-control

#!/usr/bin/env python3
import sqlite3
from pathlib import Path
 
BASE = Path.home() / ".hermes" / "manas"
EVENT_DB = BASE / "events.sqlite"
CONTROL_DB = BASE / "control.sqlite"
 
BASE.mkdir(parents=True, exist_ok=True)
 
with sqlite3.connect(EVENT_DB) as con:
    con.execute("""
    CREATE TABLE IF NOT EXISTS events (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        ts TEXT NOT NULL,
        source TEXT NOT NULL,
        type TEXT NOT NULL,
        severity TEXT DEFAULT 'info',
        session_id TEXT,
        task_id TEXT,
        payload TEXT NOT NULL
    )
    """)
    con.execute("""
    CREATE INDEX IF NOT EXISTS idx_events_ts ON events(ts)
    """)
    con.execute("""
    CREATE INDEX IF NOT EXISTS idx_events_type ON events(type)
    """)
 
with sqlite3.connect(CONTROL_DB) as con:
    con.execute("""
    CREATE TABLE IF NOT EXISTS decisions (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        ts TEXT NOT NULL,
        actor TEXT NOT NULL,
        action TEXT NOT NULL,
        target_type TEXT,
        target_id TEXT,
        priority INTEGER DEFAULT 50,
        status TEXT DEFAULT 'pending',
        reason TEXT,
        params TEXT DEFAULT '{}',
        result TEXT
    )
    """)
    con.execute("""
    CREATE INDEX IF NOT EXISTS idx_decisions_status ON decisions(status)
    """)
    con.execute("""
    CREATE INDEX IF NOT EXISTS idx_decisions_action ON decisions(action)
    """)
 
print("Manas databases initialized.")

python3 ~/.hermes/manas/bin/init_manas.py

import json
import sqlite3
from pathlib import Path
from datetime import datetime, timezone
 
BASE = Path.home() / ".hermes" / "manas"
EVENT_DB = BASE / "events.sqlite"
CONTROL_DB = BASE / "control.sqlite"
 
def now():
    return datetime.now(timezone.utc).isoformat()
 
def safe_json(obj):
    return json.dumps(obj, ensure_ascii=False, default=str)
 
def emit_event(
    source: str,
    type: str,
    payload: dict,
    severity: str = "info",
    session_id: str | None = None,
    task_id: str | None = None,
):
    BASE.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(EVENT_DB) as con:
        con.execute("""
        INSERT INTO events (ts, source, type, severity, session_id, task_id, payload)
        VALUES (?, ?, ?, ?, ?, ?, ?)
        """, (
            now(),
            source,
            type,
            severity,
            session_id,
            task_id,
            safe_json(payload),
        ))
 
def add_decision(
    action: str,
    target_type: str | None = None,
    target_id: str | None = None,
    priority: int = 50,
    reason: str = "",
    params: dict | None = None,
    actor: str = "manas",
):
    BASE.mkdir(parents=True, exist_ok=True)
    with sqlite3.connect(CONTROL_DB) as con:
        con.execute("""
        INSERT INTO decisions
        (ts, actor, action, target_type, target_id, priority, reason, params)
        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
        """, (
            now(),
            actor,
            action,
            target_type,
            target_id,
            priority,
            reason,
            safe_json(params or {}),
        ))
 
def fetch_pending_decision(action: str | None = None, target_id: str | None = None):
    query = """
    SELECT * FROM decisions
    WHERE status = 'pending'
    """
    args = []
 
    if action:
        query += " AND action = ?"
        args.append(action)
 
    if target_id:
        query += " AND target_id = ?"
        args.append(target_id)
 
    query += " ORDER BY priority DESC, id ASC LIMIT 1"
 
    with sqlite3.connect(CONTROL_DB) as con:
        con.row_factory = sqlite3.Row
        row = con.execute(query, args).fetchone()
        return dict(row) if row else None
 
def mark_decision(decision_id: int, status: str, result: str = ""):
    with sqlite3.connect(CONTROL_DB) as con:
        con.execute("""
        UPDATE decisions
        SET status = ?, result = ?
        WHERE id = ?
        """, (status, result, decision_id))

name: manas-gateway-feed
description: Feed gateway lifecycle and Discord events into Manas event bus
events:
  - gateway:startup
  - session:start
  - session:end
  - session:reset
  - agent:start
  - agent:step
  - agent:end
  - command:*

import sys
from pathlib import Path
 
LIB = Path.home() / ".hermes" / "manas" / "lib"
sys.path.insert(0, str(LIB))
 
from manas_bus import emit_event
 
def handle(event_type: str, context: dict):
    emit_event(
        source="hermes-gateway",
        type=event_type,
        payload=context,
        severity="info",
        session_id=context.get("session_id") or context.get("session_key"),
        task_id=context.get("task_id"),
    )

Discord 触发的 Hermes 会话
用户消息上下文
Hermes agent start/end
slash command
session reset/new

name: manas-control
version: "0.1.0"
description: Manas control plugin for Hermes supervision

import sys
import json
from pathlib import Path
 
LIB = Path.home() / ".hermes" / "manas" / "lib"
sys.path.insert(0, str(LIB))
 
from manas_bus import emit_event, fetch_pending_decision, mark_decision
 
def pre_tool_guard(tool_name: str, args: dict, task_id: str = "", **kwargs):
    emit_event(
        source="hermes-plugin",
        type="tool:pre",
        payload={
            "tool_name": tool_name,
            "args": args,
            "kwargs": kwargs,
        },
        severity="info",
        task_id=task_id,
    )
 
    # Manas 可以要求阻断某个 task 的下一次工具调用
    decision = fetch_pending_decision(
        action="block_tool",
        target_id=task_id or None,
    )
 
    if decision:
        mark_decision(decision["id"], "applied", "blocked by pre_tool_guard")
        return {
            "action": "block",
            "message": f"Blocked by Manas: {decision.get('reason') or 'no reason'}"
        }
 
    # 全局危险命令保护，可按需扩展
    if tool_name in {"terminal", "write_file", "patch"}:
        global_decision = fetch_pending_decision(action="block_dangerous_tools")
        if global_decision:
            mark_decision(global_decision["id"], "applied", "blocked dangerous tool")
            return {
                "action": "block",
                "message": f"Blocked by Manas global policy: {global_decision.get('reason')}"
            }
 
    return None
 
def post_tool_logger(tool_name: str, result=None, task_id: str = "", duration_ms: int = 0, **kwargs):
    emit_event(
        source="hermes-plugin",
        type="tool:post",
        payload={
            "tool_name": tool_name,
            "result_preview": str(result)[:4000],
            "duration_ms": duration_ms,
            "kwargs": kwargs,
        },
        severity="info",
        task_id=task_id,
    )
 
def pre_llm_injector(
    session_id: str = "",
    user_message: str = "",
    conversation_history: list | None = None,
    model: str = "",
    platform: str = "",
    **kwargs
):
    emit_event(
        source="hermes-plugin",
        type="llm:pre",
        payload={
            "session_id": session_id,
            "user_message": user_message,
            "model": model,
            "platform": platform,
        },
        severity="info",
        session_id=session_id,
    )
 
    decision = fetch_pending_decision(
        action="inject_context",
        target_id=session_id or None,
    )
 
    if decision:
        params = json.loads(decision.get("params") or "{}")
        context = params.get("context") or decision.get("reason") or ""
        mark_decision(decision["id"], "applied", "context injected")
        if context:
            return {
                "context": f"[Manas Supervisor Injection]\n{context}"
            }
 
    return None
 
def transform_llm_output(response_text: str, session_id: str = "", model: str = "", platform: str = "", **kwargs):
    emit_event(
        source="hermes-plugin",
        type="llm:final",
        payload={
            "response_preview": response_text[:4000],
            "model": model,
            "platform": platform,
        },
        severity="info",
        session_id=session_id,
    )
    return None
 
def approval_logger(command: str, description: str, pattern_key: str = "", pattern_keys=None,
                    session_key: str = "", surface: str = "", **kwargs):
    emit_event(
        source="hermes-plugin",
        type="approval:pre",
        payload={
            "command": command,
            "description": description,
            "pattern_key": pattern_key,
            "pattern_keys": pattern_keys or [],
            "session_key": session_key,
            "surface": surface,
        },
        severity="warning",
        session_id=session_key,
    )
 
def register(ctx):
    ctx.register_hook("pre_tool_call", pre_tool_guard)
    ctx.register_hook("post_tool_call", post_tool_logger)
    ctx.register_hook("pre_llm_call", pre_llm_injector)
    ctx.register_hook("transform_llm_output", transform_llm_output)
    ctx.register_hook("pre_approval_request", approval_logger)

cron:soon
cron:running
cron:done
cron:error

#!/usr/bin/env python3
import sys
import json
import time
from pathlib import Path
from datetime import datetime, timezone
 
LIB = Path.home() / ".hermes" / "manas" / "lib"
sys.path.insert(0, str(LIB))
 
from manas_bus import emit_event
 
JOBS = Path.home() / ".hermes" / "cron" / "jobs.json"
STATE = Path.home() / ".hermes" / "manas" / "cron_watcher_state.json"
 
def parse_ts(s):
    if not s:
        return None
    return datetime.fromisoformat(s.replace("Z", "+00:00"))
 
def load_json(path, default):
    if not path.exists():
        return default
    try:
        return json.loads(path.read_text(encoding="utf-8"))
    except Exception:
        return default
 
def save_json(path, data):
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
 
def main():
    seen = load_json(STATE, {})
    while True:
        jobs = load_json(JOBS, [])
        now = datetime.now(timezone.utc)
 
        if isinstance(jobs, dict):
            jobs_iter = jobs.get("jobs", [])
        else:
            jobs_iter = jobs
 
        for job in jobs_iter:
            job_id = job.get("id")
            if not job_id:
                continue
 
            state = job.get("state")
            enabled = job.get("enabled")
            next_run_at = job.get("next_run_at")
            last_run_at = job.get("last_run_at")
            last_status = job.get("last_status")
 
            key_state = f"{job_id}:state"
            key_last = f"{job_id}:last_run_at"
            key_soon = f"{job_id}:soon:{next_run_at}"
 
            if state != seen.get(key_state):
                emit_event(
                    source="daemon-cron-watcher",
                    type=f"cron:state:{state}",
                    payload=job,
                    severity="info" if state != "running" else "warning",
                    task_id=job_id,
                )
                seen[key_state] = state
 
            if last_run_at and last_run_at != seen.get(key_last):
                emit_event(
                    source="daemon-cron-watcher",
                    type="cron:finished",
                    payload=job,
                    severity="info" if last_status == "ok" else "error",
                    task_id=job_id,
                )
                seen[key_last] = last_run_at
 
            dt = parse_ts(next_run_at)
            if enabled and state == "scheduled" and dt:
                seconds = (dt - now).total_seconds()
                if 0 <= seconds <= 120 and not seen.get(key_soon):
                    emit_event(
                        source="daemon-cron-watcher",
                        type="cron:soon",
                        payload={
                            "seconds_until_run": seconds,
                            "job": job,
                        },
                        severity="warning",
                        task_id=job_id,
                    )
                    seen[key_soon] = True
 
        save_json(STATE, seen)
        time.sleep(15)
 
if __name__ == "__main__":
    main()

hermes profile create manas --clone

nano ~/.hermes/profiles/manas/SOUL.md

你是 Manas，Hermes 系统的高层监督、反思、裁决 agent。
 
你的定位：
- Hermes 负责行动。
- daemon 负责底层协调控制。
- Manas 负责高层意义判断、全局反思、任务优先级和干预决策。
 
你能看到：
- Discord 用户与 Hermes 的对话事件
- Hermes 工具调用和结果
- cron 即将运行、运行中、完成、失败
- gateway/session/agent 生命周期
- daemon 监控到的进程状态
 
你的任务：
1. 判断用户真实意图。
2. 判断 Hermes 是否跑偏。
3. 判断 cron 是否重复、过时、危险、浪费。
4. 判断工具调用是否应被阻断。
5. 必要时下发 control decision。
6. 所有决策必须说明事实、推断、风险、理由。
7. 能注入上下文就不阻断；能阻断工具就不 kill；能 pause cron 就不删除。
8. 高危动作只在高置信度时执行。
 
你输出 JSON，不要输出散文。

#!/usr/bin/env python3
import sys
import json
import time
import sqlite3
import subprocess
from pathlib import Path
 
LIB = Path.home() / ".hermes" / "manas" / "lib"
sys.path.insert(0, str(LIB))
 
from manas_bus import EVENT_DB, add_decision
 
STATE = Path.home() / ".hermes" / "manas" / "supervisor_state.json"
REFLECTIONS = Path.home() / ".hermes" / "manas" / "reflections.md"
 
def load_state():
    if STATE.exists():
        return json.loads(STATE.read_text(encoding="utf-8"))
    return {"last_event_id": 0}
 
def save_state(s):
    STATE.write_text(json.dumps(s, ensure_ascii=False, indent=2), encoding="utf-8")
 
def fetch_events(after_id: int, limit: int = 80):
    if not EVENT_DB.exists():
        return []
 
    with sqlite3.connect(EVENT_DB) as con:
        con.row_factory = sqlite3.Row
        rows = con.execute("""
        SELECT * FROM events
        WHERE id > ?
        ORDER BY id ASC
        LIMIT ?
        """, (after_id, limit)).fetchall()
        return [dict(r) for r in rows]
 
def call_manas(events):
    prompt = f"""
你是 Manas Supervisor。下面是 Hermes/daemon 的最新事件流。
 
请你判断：
1. 是否需要反思记录
2. 是否需要注入上下文
3. 是否需要阻断工具
4. 是否需要暂停/重跑 cron
5. 是否需要让 daemon kill/restart 某个进程
6. 是否只是观察即可
 
只输出 JSON，格式如下：
 
{{
  "summary": "简短总结",
  "reflection": "可写入 reflections.md 的反思",
  "intervention_needed": true,
  "decisions": [
    {{
      "action": "inject_context | block_tool | pause_cron | resume_cron | run_cron | stop_gateway | restart_gateway | kill_process | observe",
      "target_type": "session | task | cron | gateway | process | global",
      "target_id": "目标 id，没有则为空",
      "priority": 50,
      "reason": "理由",
      "params": {{}}
    }}
  ]
}}
 
事件流 JSON：
{json.dumps(events, ensure_ascii=False, default=str)}
"""
 
    proc = subprocess.run(
        ["hermes", "-p", "manas", "chat", "-q", prompt],
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=180,
    )
 
    text = proc.stdout.strip() or proc.stderr.strip()
 
    # 尽量从输出中提取 JSON
    start = text.find("{")
    end = text.rfind("}")
    if start >= 0 and end > start:
        text = text[start:end+1]
 
    return json.loads(text)
 
def main():
    state = load_state()
 
    while True:
        events = fetch_events(state.get("last_event_id", 0), limit=80)
 
        if not events:
            time.sleep(10)
            continue
 
        state["last_event_id"] = events[-1]["id"]
        save_state(state)
 
        # 小事件攒着，但遇到 warning/error/cron/tool/approval 立即跑
        important = any(
            e["severity"] in {"warning", "error"}
            or e["type"].startswith(("cron:", "tool:", "approval:"))
            for e in events
        )
 
        if not important and len(events) < 20:
            time.sleep(5)
            continue
 
        try:
            result = call_manas(events)
        except Exception as e:
            REFLECTIONS.parent.mkdir(parents=True, exist_ok=True)
            with REFLECTIONS.open("a", encoding="utf-8") as f:
                f.write(f"\n\n## Manas error\n\n{e}\n")
            continue
 
        reflection = result.get("reflection") or result.get("summary") or ""
        if reflection:
            with REFLECTIONS.open("a", encoding="utf-8") as f:
                f.write("\n\n## Manas Reflection\n\n")
                f.write(reflection)
                f.write("\n")
 
        for d in result.get("decisions", []):
            action = d.get("action")
            if not action or action == "observe":
                continue
 
            add_decision(
                action=action,
                target_type=d.get("target_type"),
                target_id=d.get("target_id"),
                priority=int(d.get("priority", 50)),
                reason=d.get("reason", ""),
                params=d.get("params") or {},
                actor="manas",
            )
 
        time.sleep(10)
 
if __name__ == "__main__":
    main()

#!/usr/bin/env python3
import sys
import json
import time
import signal
import sqlite3
import subprocess
from pathlib import Path
 
LIB = Path.home() / ".hermes" / "manas" / "lib"
sys.path.insert(0, str(LIB))
 
from manas_bus import CONTROL_DB, emit_event, mark_decision
 
SAFE_ACTIONS = {
    "pause_cron",
    "resume_cron",
    "run_cron",
    "block_tool",
    "inject_context",
}
 
DANGEROUS_ACTIONS = {
    "kill_process",
    "stop_gateway",
    "restart_gateway",
    "remove_cron",
}
 
def fetch_next():
    if not CONTROL_DB.exists():
        return None
 
    with sqlite3.connect(CONTROL_DB) as con:
        con.row_factory = sqlite3.Row
        row = con.execute("""
        SELECT * FROM decisions
        WHERE status = 'pending'
        ORDER BY priority DESC, id ASC
        LIMIT 1
        """).fetchone()
 
        return dict(row) if row else None
 
def run_cmd(cmd):
    proc = subprocess.run(
        cmd,
        text=True,
        stdout=subprocess.PIPE,
        stderr=subprocess.PIPE,
        timeout=120,
    )
    return proc.returncode, proc.stdout, proc.stderr
 
def apply(decision):
    action = decision["action"]
    target_id = decision["target_id"]
    params = json.loads(decision.get("params") or "{}")
 
    # 这些是 plugin 消费的，不由 daemon 处理
    if action in {"block_tool", "inject_context", "block_dangerous_tools"}:
        return "left for plugin hooks"
 
    if action == "pause_cron":
        code, out, err = run_cmd(["hermes", "cron", "pause", target_id])
        return out or err
 
    if action == "resume_cron":
        code, out, err = run_cmd(["hermes", "cron", "resume", target_id])
        return out or err
 
    if action == "run_cron":
        code, out, err = run_cmd(["hermes", "cron", "run", target_id])
        return out or err
 
    if action == "remove_cron":
        if params.get("confirm") != "MANAS_CONFIRMED":
            raise RuntimeError("remove_cron requires params.confirm = MANAS_CONFIRMED")
        code, out, err = run_cmd(["hermes", "cron", "remove", target_id])
        return out or err
 
    if action == "restart_gateway":
        profile = params.get("profile", "discord-main")
        # WSL/tmux 场景建议你自己用 tmux 管，不直接 gateway restart
        code, out, err = run_cmd(["tmux", "send-keys", "-t", f"hermes-{profile}", "C-c"])
        time.sleep(2)
        run_cmd(["tmux", "send-keys", "-t", f"hermes-{profile}", f"hermes -p {profile} gateway run", "Enter"])
        return out or err or f"restart signal sent to hermes-{profile}"
 
    if action == "stop_gateway":
        profile = params.get("profile", "discord-main")
        code, out, err = run_cmd(["tmux", "send-keys", "-t", f"hermes-{profile}", "C-c"])
        return out or err or f"stop signal sent to hermes-{profile}"
 
    if action == "kill_process":
        pid = int(target_id)
        if params.get("confirm") != "MANAS_CONFIRMED":
            raise RuntimeError("kill_process requires params.confirm = MANAS_CONFIRMED")
        os.kill(pid, signal.SIGTERM)
        return f"SIGTERM sent to pid {pid}"
 
    raise RuntimeError(f"unknown action: {action}")
 
def main():
    while True:
        d = fetch_next()
        if not d:
            time.sleep(5)
            continue
 
        try:
            result = apply(d)
            mark_decision(d["id"], "applied", result)
            emit_event(
                source="daemon-actuator",
                type=f"control:applied:{d['action']}",
                payload={"decision": d, "result": result},
                severity="warning",
                task_id=d.get("target_id"),
            )
        except Exception as e:
            mark_decision(d["id"], "failed", str(e))
            emit_event(
                source="daemon-actuator",
                type=f"control:failed:{d['action']}",
                payload={"decision": d, "error": str(e)},
                severity="error",
                task_id=d.get("target_id"),
            )
 
        time.sleep(2)
 
if __name__ == "__main__":
    main()

{"confirm": "MANAS_CONFIRMED"}

# 1. 正常 Discord Hermes
tmux new -s hermes-discord-main 'hermes -p discord-main gateway run'
 
# 2. cron watcher
tmux new -s daemon-cron 'python3 ~/.hermes/manas/bin/daemon_cron_watcher.py'
 
# 3. Manas 高层监督
tmux new -s manas-supervisor 'python3 ~/.hermes/manas/bin/manas_supervisor.py'
 
# 4. daemon 执行控制
tmux new -s daemon-actuator 'python3 ~/.hermes/manas/bin/daemon_actuator.py'

tmux ls

tmux attach -t manas-supervisor

你 → Discord → Hermes Main

gateway hook 记录用户消息
plugin hook 记录工具调用
Hermes 正常回复你
plugin/gateway hook 记录回复
Manas 读取事件
Manas 判断：这是不是长期项目？有没有上下文缺失？Hermes 是否跑偏？

{
  "action": "inject_context",
  "target_type": "session",
  "target_id": "agent:main:discord:dm:xxxx",
  "reason": "Hermes 没有意识到这是见政项目的政策可视化模块",
  "params": {
    "context": "这是用户长期项目「见政」的一部分，核心是政策可视化节点、关系、强度、权重逻辑。"
  }
}

daemon_cron_watcher 扫 jobs.json
发现 120 秒后 daily-news 要运行
写入 cron:soon
Manas 读取
Manas 判断：这个新闻任务是否需要更新 prompt？是否和另一个 cron 重复？

{
  "action": "pause_cron",
  "target_type": "cron",
  "target_id": "daily-news-old",
  "reason": "旧新闻任务与新任务重复，且输出目标相同。"
}

hermes cron pause daily-news-old

rm -rf some_dir

{
  "action": "block_tool",
  "target_type": "task",
  "target_id": "某个 task_id",
  "reason": "当前任务目标不需要删除目录，疑似跑偏。"
}

{"action": "block", "message": "Blocked by Manas: ..."}

identity:
  hermes: "执行、传信、对话、工具调用"
  daemon: "底层协调、状态同步、进程控制、cron 监控"
  manas: "高层反思、全局监督、意图判断、战略裁决"
 
authority:
  manas_is_highest: true
  daemon_executes_low_level_actions: true
  hermes_must_be_observable: true
 
default_strategy:
  - observe_first
  - inject_before_block
  - block_before_pause
  - pause_before_kill
  - restart_only_when_stuck
  - delete_never_without_explicit_reason
 
auto_allowed:
  - write_reflection
  - inject_context
  - block_tool
  - pause_cron
  - resume_cron
  - run_cron
 
requires_strong_reason:
  - restart_gateway
  - stop_gateway
  - kill_process
  - remove_cron
 
never_auto:
  - delete_user_files
  - wipe_memory
  - expose_private_logs_to_public_channel
  - change_api_keys
  - change_discord_token
  - enable_yolo

Hermes 是手。
daemon 是神经反射和肌肉控制。
Manas 是思量、判断和全局意识。

Hermes = Worker
daemon = Runtime Controller
Manas  = Cognitive Supervisor

Hermes-Manas Architecture

manas-supervisor      高层监督进程
manas-event-bus       事件数据库
manas-control-plane   控制数据库
daemon-cron-watcher   cron 观察器
daemon-actuator       控制执行器
daemon-process-watch  进程守护器
hermes-main           Discord 主执行 agent

1. manas-gateway-feed hook
2. manas-control plugin
3. daemon_cron_watcher.py
4. manas_supervisor.py + daemon_actuator.py

Hermes 负责行动，
daemon 负责控制，
Manas 负责看见整体、理解整体、裁决整体。