Hermes Agent Kanban 深度解析：原理、实现与竞品分析 #

-- 任务表：状态机 + 声明锁 + 工作区
CREATE TABLE tasks (
    id                   TEXT PRIMARY KEY,       -- t_ + 4 hex bytes (~4.3B 空间)
    title                TEXT NOT NULL,
    body                 TEXT,
    assignee             TEXT,                   -- Profile 名称
    status               TEXT NOT NULL,           -- triage|todo|ready|running|blocked|done|archived
    priority             INTEGER DEFAULT 0,
    workspace_kind       TEXT DEFAULT 'scratch',  -- scratch | worktree | dir
    workspace_path       TEXT,
    claim_lock           TEXT,                    -- host:pid 格式的声明锁
    claim_expires        INTEGER,                 -- 过期时间戳 (默认 15 分钟)
    tenant               TEXT,                    -- 租户隔离
    current_run_id       INTEGER,                 -- 指向 task_runs 的当前活跃行
    max_runtime_seconds  INTEGER,                 -- 每任务运行时限
    spawn_failures       INTEGER DEFAULT 0,       -- 连续派生失败计数
    last_spawn_error     TEXT,
    worker_pid           INTEGER,                 -- 当前 Worker 进程 PID
    last_heartbeat_at    INTEGER,                 -- 最后心跳时间
    skills               TEXT                     -- JSON: 强制加载的技能列表
);

-- 依赖图：DAG 边
CREATE TABLE task_links (
    parent_id  TEXT NOT NULL,
    child_id   TEXT NOT NULL,
    PRIMARY KEY (parent_id, child_id)
);

-- 尝试历史：每次执行都是独立的 run
CREATE TABLE task_runs (
    id                  INTEGER PRIMARY KEY AUTOINCREMENT,
    task_id             TEXT NOT NULL,
    profile             TEXT,
    status              TEXT NOT NULL,     -- running|done|blocked|crashed|timed_out|failed|reclaimed
    claim_lock          TEXT,
    claim_expires       INTEGER,
    worker_pid          INTEGER,
    max_runtime_seconds INTEGER,
    last_heartbeat_at   INTEGER,
    started_at          INTEGER NOT NULL,
    ended_at            INTEGER,
    outcome             TEXT,              -- completed|blocked|crashed|timed_out|spawn_failed|gave_up|reclaimed
    summary             TEXT,              -- 人类可读的交接摘要
    metadata            TEXT,              -- JSON: 结构化事实
    error               TEXT
);

-- 评论线程：跨 Agent 通信
CREATE TABLE task_comments (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    task_id TEXT NOT NULL, author TEXT NOT NULL,
    body TEXT NOT NULL, created_at INTEGER NOT NULL
);

-- 事件日志：状态变迁的审计追踪
CREATE TABLE task_events (
    id INTEGER PRIMARY KEY AUTOINCREMENT,
    task_id TEXT NOT NULL, run_id INTEGER,
    kind TEXT NOT NULL, payload TEXT, created_at INTEGER NOT NULL
);

# kanban_db.py: write_txn()
@contextlib.contextmanager
def write_txn(conn: sqlite3.Connection):
    conn.execute("BEGIN IMMEDIATE")  # 立即获取写锁
    try:
        yield conn
    except Exception:
        conn.execute("ROLLBACK")
        raise
    else:
        conn.execute("COMMIT")

# kanban_db.py: create_task()
if idempotency_key:
    row = conn.execute(
        "SELECT id FROM tasks WHERE idempotency_key = ? AND status != 'archived' "
        "ORDER BY created_at DESC LIMIT 1",
        (idempotency_key,),
    ).fetchone()
    if row:
        return row["id"]  # 返回已存在的任务 ID

# kanban_db.py: claim_task()
cur = conn.execute(
    """
    UPDATE tasks
       SET status        = 'running',
           claim_lock    = ?,       -- "hostname:pid"
           claim_expires = ?,       -- now + 900 (15分钟)
           started_at    = COALESCE(started_at, ?)
     WHERE id = ?
       AND status = 'ready'         -- 前置条件：必须是 ready
       AND claim_lock IS NULL       -- 前置条件：没有其他人在 claim
    """,
    (lock, expires, now, task_id),
)
if cur.rowcount != 1:
    return None  # 被其他人抢先了，返回 None

# kanban_db.py: dispatch_once()
def dispatch_once(conn, *, spawn_fn=None, ...) -> DispatchResult:
    result = DispatchResult()

    # Step 1: 回收过期的声明锁（TTL 超时）
    result.reclaimed = release_stale_claims(conn)

    # Step 2: 检测崩溃的 Worker（PID 不再存活）
    result.crashed = detect_crashed_workers(conn)

    # Step 3: 提升依赖已满足的任务 (todo → ready)
    result.promoted = recompute_ready(conn)

    # Step 4: 派生 ready 任务的 Worker 进程
    for row in ready_rows:
        claimed = claim_task(conn, row["id"], ttl_seconds=ttl_seconds)
        workspace = resolve_workspace(claimed)
        pid = spawn_fn(claimed, str(workspace))  # → hermes -p <profile> chat -q
        _set_worker_pid(conn, claimed.id, pid)

# 找到所有 claim_expires < now 的 running 任务
stale = conn.execute(
    "SELECT id, claim_lock FROM tasks "
    "WHERE status = 'running' AND claim_expires < ?",
    (now,),
).fetchall()
for row in stale:
    # 回滚到 ready，关闭当前 run
    conn.execute(
        "UPDATE tasks SET status = 'ready', claim_lock = NULL, "
        "claim_expires = NULL, worker_pid = NULL WHERE id = ?",
        (row["id"],),
    )
    _end_run(conn, row["id"], outcome="reclaimed", ...)

def _pid_alive(pid: Optional[int]) -> bool:
    try:
        os.kill(int(pid), 0)  # signal 0 = 检查进程是否存在
    except ProcessLookupError:
        return False
    except PermissionError:
        return True  # 存在但无权 signal
    # Linux 额外检查 zombie 状态
    if sys.platform == "linux":
        with open(f"/proc/{pid}/status", "r") as f:
            for line in f:
                if line.startswith("State:") and "Z" in line:
                    return False  # zombie = 已退出但未 reap
    return True

def recompute_ready(conn: sqlite3.Connection) -> int:
    promoted = 0
    with write_txn(conn):
        for row in conn.execute("SELECT id FROM tasks WHERE status = 'todo'"):
            task_id = row["id"]
            parents = conn.execute(
                "SELECT t.status FROM tasks t JOIN task_links l "
                "ON l.parent_id = t.id WHERE l.child_id = ?",
                (task_id,),
            ).fetchall()
            if all(p["status"] == "done" for p in parents):
                conn.execute(
                    "UPDATE tasks SET status = 'ready' "
                    "WHERE id = ? AND status = 'todo'",
                    (task_id,),
                )
                promoted += 1
    return promoted

def _default_spawn(task: Task, workspace: str) -> Optional[int]:
    cmd = [
        "hermes",
        "-p", task.assignee,
        "--skills", "kanban-worker",  # 自动加载 worker 技能
        "chat",
        "-q", f"work kanban task {task.id}",
    ]
    env = {
        "HERMES_KANBAN_TASK": task.id,
        "HERMES_KANBAN_WORKSPACE": workspace,
        "HERMES_PROFILE": task.assignee,
    }
    proc = subprocess.Popen(
        cmd,
        cwd=workspace,
        stdin=subprocess.DEVNULL,
        stdout=log_f,          # 输出重定向到日志
        stderr=subprocess.STDOUT,
        env=env,
        start_new_session=True, # 独立进程组
    )
    return proc.pid

def _record_spawn_failure(conn, task_id, error, failure_limit=5):
    failures = row["spawn_failures"] + 1
    if failures >= failure_limit:
        # 超过限制 → 自动 block
        conn.execute(
            "UPDATE tasks SET status = 'blocked', "
            "spawn_failures = ?, last_spawn_error = ? WHERE id = ?",
            (failures, error[:500], task_id),
        )
        _append_event(conn, task_id, "gave_up", {...})
    else:
        # 重置为 ready，下次 tick 重试
        conn.execute(
            "UPDATE tasks SET status = 'ready', spawn_failures = ?",
            (failures, task_id),
        )

# kanban_tools.py
def _check_kanban_mode() -> bool:
    return bool(os.environ.get("HERMES_KANBAN_TASK"))

kanban_complete(
    summary="shipped rate limiter — token bucket, keys on user_id with IP fallback, 14 tests pass",
    metadata={
        "changed_files": ["rate_limiter.py", "tests/test_rate_limiter.py"],
        "tests_run": 14,
        "tests_passed": 14,
        "decisions": ["user_id primary, IP fallback for unauthenticated requests"],
    },
)

# 父任务的 run.summary 和 run.metadata 会被注入到子任务的 prompt 中
## Parent task results
### t_a3f2b1c7
shipped rate limiter — token bucket...
_metadata_: `{"changed_files": [...], "tests_run": 14, ...}`

def create_task(conn, *, title, parents=(), ...):
    with write_txn(conn):
        # 检查父任务是否存在
        missing = _find_missing_parents(conn, parents)
        if missing:
            raise ValueError(f"unknown parent task(s): {', '.join(missing)}")

        # 判断初始状态
        if parents:
            rows = conn.execute(
                "SELECT status FROM tasks WHERE id IN (...) ", parents
            ).fetchall()
            if any(r["status"] != "done" for r in rows):
                initial_status = "todo"  # 有未完成的父任务
            else:
                initial_status = "ready"  # 所有父任务都已完成
        else:
            initial_status = "ready"

def _would_cycle(conn, parent_id, child_id) -> bool:
    """如果 parent_id 已经是 child_id 的后代，则形成环"""
    seen = set()
    stack = [child_id]
    while stack:
        node = stack.pop()
        if node == parent_id:
            return True
        if node in seen:
            continue
        seen.add(node)
        rows = conn.execute(
            "SELECT child_id FROM task_links WHERE parent_id = ?", (node,)
        ).fetchall()
        stack.extend(r["child_id"] for r in rows)
    return False

def link_tasks(conn, parent_id, child_id):
    # 检查循环...
    conn.execute("INSERT OR IGNORE INTO task_links ...", (parent_id, child_id))

    # 如果父任务未完成，子任务从 ready 降级到 todo
    parent_status = conn.execute(
        "SELECT status FROM tasks WHERE id = ?", (parent_id,)
    ).fetchone()["status"]
    if parent_status != "done":
        conn.execute(
            "UPDATE tasks SET status = 'todo' WHERE id = ? AND status = 'ready'",
            (child_id,),
        )

_CTX_MAX_PRIOR_ATTEMPTS = 10      # 最多显示 10 次先前尝试
_CTX_MAX_COMMENTS       = 30      # 最多显示 30 条评论
_CTX_MAX_FIELD_BYTES    = 4 * 1024   # 每个字段 4KB
_CTX_MAX_BODY_BYTES     = 8 * 1024   # task.body 8KB
_CTX_MAX_COMMENT_BYTES  = 2 * 1024   # 每条评论 2KB

# 跨任务角色历史：给 Worker 隐式的连续性
## Recent work by @reviewer
- t_x1y2z3 — Review PR #456 (2026-06-08): found 2 SQL injection issues
- t_a4b5c6 — Review PR #789 (2026-06-07): approved with minor suggestions
- t_d7e8f9 — Review auth module (2026-06-06): blocked on missing CSRF

CREATE TABLE kanban_notify_subs (
    task_id       TEXT NOT NULL,
    platform      TEXT NOT NULL,          -- telegram, discord, dingtalk...
    chat_id       TEXT NOT NULL,
    thread_id     TEXT NOT NULL DEFAULT '',
    user_id       TEXT,
    last_event_id INTEGER NOT NULL DEFAULT 0,
    PRIMARY KEY (task_id, platform, chat_id, thread_id)
);