Note

Agent-task

7 min read

atask

AI Agent 任务管理 CLI 工具。为多 Agent 并发协作设计，通过文件锁保证原子操作，JSONL 日志支持跨会话上下文恢复。

安装

npm install
npm run build

快速开始

# 初始化项目
atask init
 
# 添加任务
atask add "实现用户认证" --pri=high --tag=backend
atask add "设计数据库" --pri=medium --dep=1 --desc="设计 schema"
 
# 查看任务
atask list
atask list --status=pending --tag=backend
atask show 1
 
# Agent 自动领取最高优先级可用任务
atask next
 
# 记录进度
atask log 1 "完成了 JWT 模块"
 
# 完成任务
atask done 1
 
# 查看依赖图和统计
atask deps
atask stats

工作机制

整体架构

graph TB
    subgraph CLI["CLI 层"]
        Entry["cli.ts<br/>minimist 解析 + 命令路由"]
    end

    subgraph Commands["命令层"]
        direction LR
        Data["数据操作<br/>init / add / list<br/>show / edit / rm"]
        Status["状态流转<br/>start / done<br/>block / reset"]
        Agent["Agent 命令<br/>next / log"]
        Analysis["分析命令<br/>deps / stats / export"]
    end

    subgraph Core["核心层"]
        Store["store.ts<br/>读写 index + tasks + logs"]
        Lock["lock.ts<br/>proper-lockfile"]
        Scheduler["scheduler.ts<br/>优先级排序 + DAG 解析"]
    end

    subgraph Storage["文件存储 (.atask/)"]
        Index["index.json<br/>任务索引 🔒"]
        Tasks["tasks/*.json<br/>任务详情"]
        Logs["logs/*.jsonl<br/>追加日志"]
    end

    Entry --> Data & Status & Agent & Analysis
    Data & Status & Agent --> Store
    Analysis --> Store & Scheduler
    Agent --> Scheduler
    Store --> Lock
    Lock --> Index
    Store --> Tasks
    Store --> Logs

任务状态机

stateDiagram-v2
    [*] --> pending : add

    pending --> in_progress : start / next
    pending --> blocked : block

    in_progress --> done : done
    in_progress --> blocked : block
    in_progress --> pending : reset

    blocked --> pending : reset

    done --> pending : reset
    done --> [*]

next 命令：原子领取流程

sequenceDiagram
    participant A as Agent A
    participant B as Agent B
    participant L as 文件锁 (index.json)
    participant S as Store
    participant Sch as Scheduler

    A->>L: 请求锁
    B->>L: 请求锁
    L-->>A: ✅ 获得锁
    L-->>B: ⏳ 等待重试 (50-500ms)

    A->>S: 读取 index.json
    S->>Sch: getAvailableTasks()
    Note over Sch: 筛选 pending 且<br/>blockedBy 全部 done
    Sch->>Sch: sortByPriority()
    Note over Sch: critical > high ><br/>medium > low<br/>同优先级按创建时间
    Sch-->>A: 返回 Task #1

    A->>S: 更新 #1 → in_progress<br/>assignee = Agent A UUID
    S->>S: 写 index.json + tasks/1.json
    A->>L: 释放锁

    L-->>B: ✅ 获得锁
    B->>S: 读取 index.json
    S->>Sch: getAvailableTasks()
    Note over Sch: #1 已是 in_progress<br/>不再可用
    Sch-->>B: 返回 Task #2
    B->>S: 更新 #2 → in_progress
    B->>L: 释放锁

依赖解析与任务调度

graph LR
    subgraph 任务池
        T1["#1 认证模块<br/>🔴 HIGH / pending"]
        T2["#2 数据库设计<br/>🟡 MED / pending<br/>dep: #1"]
        T3["#3 API 网关<br/>🔴 HIGH / pending<br/>dep: #1, #2"]
        T4["#4 编写文档<br/>🟢 LOW / pending"]
    end

    T1 -->|blocks| T2
    T1 -->|blocks| T3
    T2 -->|blocks| T3

    subgraph Scheduler["调度器判定"]
        direction TB
        S1["getAvailableTasks()"]
        S2["#1 ✅ 无依赖"]
        S3["#2 ❌ 等待 #1"]
        S4["#3 ❌ 等待 #1, #2"]
        S5["#4 ✅ 无依赖"]
        S1 --> S2 & S3 & S4 & S5

        Sort["sortByPriority()"]
        S2 --> Sort
        S5 --> Sort
        Result["next 返回 #1<br/>(HIGH > LOW)"]
        Sort --> Result
    end

多 Agent 协作流程

sequenceDiagram
    participant H as 人类 / 主 Agent
    participant A as Agent A
    participant B as Agent B
    participant FS as .atask/ 文件系统

    H->>FS: atask init
    H->>FS: atask add "认证模块" --pri=high
    H->>FS: atask add "数据库" --dep=1
    H->>FS: atask add "测试" --pri=low

    par Agent A 工作循环
        A->>FS: atask next
        Note over A,FS: 领取 #1 (high, 无依赖)
        A->>FS: atask log 1 "开始 JWT 实现"
        Note over A: ... 工作中 ...
        A->>FS: atask log 1 "JWT 完成"
        A->>FS: atask done 1
        A->>FS: atask next
        Note over A,FS: 领取 #2 (#1 已完成,解除阻塞)
    and Agent B 工作循环
        B->>FS: atask next
        Note over B,FS: 领取 #3 (low, 无依赖)
        B->>FS: atask log 3 "编写单元测试"
        B->>FS: atask done 3
        B->>FS: atask next
        Note over B,FS: 无可用任务
    end

    H->>FS: atask stats
    Note over H: Total: 3  Done: 2 (66%)

存储层读写与锁策略

graph TB
    subgraph 需要锁["🔒 需要文件锁 (withLock)"]
        add["add — 分配 ID + 写索引"]
        update["updateTask — 更新索引 + 详情"]
        remove["removeTask — 删索引 + 清理依赖"]
        next["next — 读 → 筛选 → 写回"]
    end

    subgraph 无锁["🔓 无锁操作"]
        read_idx["readIndex — 只读索引"]
        read_task["readTask — 只读详情"]
        append_log["appendLog — OS 原子追加"]
        read_logs["readLogs — 只读日志"]
        list_cmd["list / show / deps / stats"]
    end

    subgraph Files["文件系统"]
        index["index.json"]
        tasks["tasks/*.json"]
        logs["logs/*.jsonl"]
    end

    add & update & remove & next -->|"proper-lockfile<br/>retries: 5<br/>stale: 10s"| index
    add & update & remove --> tasks
    append_log --> logs
    read_idx --> index
    read_task --> tasks
    read_logs --> logs
    list_cmd --> read_idx & read_task & read_logs

命令

命令说明
atask init初始化 .atask/ 目录
atask add <title> [options]添加任务。--pri=high --tag=x --dep=1,2 --desc="..."
atask list [--status=x] [--tag=x]列出任务
atask show <id>查看任务详情 + 最近日志
atask edit <id> [--title=x] [--pri=x]编辑任务
atask rm <id>删除任务
atask start <id>pending → in_progress
atask done <id>in_progress → done
atask block <id>→ blocked
atask reset <id>→ pending
atask next原子领取最高优先级可用任务
atask log <id> <message>追加进度日志
atask deps [<id>]依赖关系图
atask stats统计摘要
atask export [--format=json]导出所有任务

所有命令支持 --json 标志输出机器可读 JSON。

存储结构

.atask/
  config.json           # 项目配置
  index.json            # 任务索引（轻量，锁保护）
  tasks/
    1.json, 2.json      # 任务详情
  logs/
    1.jsonl, 2.jsonl    # JSONL 追加日志（无锁）

多 Agent 并发

  • atask next 通过文件锁实现原子领取，多 Agent 同时调用不会领取同一任务
  • 每个 Agent 进程自动生成 UUID 作为 assignee
  • 日志追加使用 OS 原子写入，无需锁

开发

npm run dev            # watch 模式构建
npm test               # 运行测试
npm run test:watch     # watch 模式测试

Graph View

Backlinks