龙虾调教(完结篇):从零搭建项目管理的 OpenClaw 助理

折腾了三个月,终于把 OpenClaw 调成了一个能帮我跑 Linear + GitHub 全链路项目管理的 AI 助理——从需求收集、Backlog 梳理、Sprint 规划,到开发监控、验收 Gate、PR Review,大部分流程它自己推,人只在关键决策点出现。

这篇是完结篇,把所有配置文件和步骤整理成一份可以直接照着操作的清单,不再重复解释设计原因。如果你想了解背后的思路,可以先看前四篇:

如果你是 OpenClaw,请直接阅读本文的 Markdown 模式

1. 完整目录结构

最终的 ~/.openclaw/workspace/ 长这样:

~/.openclaw/workspace/

├── AGENTS.md          # 行为规范(每次 session 必读)
├── SOUL.md            # 行为原则
├── IDENTITY.md        # 身份定义
├── USER.md            # 用户档案
├── HEARTBEAT.md       # Heartbeat 执行清单
├── MEMORY.md          # 全局记忆索引(≤40行)
├── NOW.md             # 当前状态快照(每次 heartbeat 覆写)
├── WORKFLOW.md        # 工作流索引

├── memory/
│   ├── YYYY-MM-DD.md          # 每日日志(追加式)
│   ├── projects-note.md       # 项目状态备注
│   ├── infra.md               # 基础设施配置
│   └── knowledge/
│       ├── INDEX.md           # 知识库导航
│       ├── lessons/           # 踩坑经验
│       ├── decisions/         # 重要决策
│       └── people/            # 人物档案

├── scripts/
│   └── memlog.sh              # 日志追加脚本

├── status/
│   ├── PROJECTS.json          # 项目注册表(权威来源)
│   ├── heartbeat-state.json   # PR 去重、验收重试计数
│   └── MAILLIST.json          # 邮件监控规则 + 状态

├── flowchain/
│   └── projects.py            # Linear + GitHub 操作脚本

└── workflow/
    ├── 00-create-workflow.md  # 元工作流:如何新建工作流
    └── 01-project-management.md  # 项目管理 SOP(本文核心)

2. 项目管理流程总览

在开始配置之前,先看清楚这套系统跑起来是什么样的。

敏捷节奏

graph LR
    A[需求收集] --> B[Backlog 梳理]
    B --> C[Sprint 规划]
    C --> D[开发]
    D --> E["验收 Gate"]
    E --> F[Code Review]
    F --> G[完成]
    G --> H[回顾]
    H -- 反馈循环 --> A

角色分工

角色职责
OpenClaw敏捷教练 / 项目管理:协调推进、监控状态、跨工具串联、主动提醒
Claude Code理解模糊需求、跨文件分析、技术方案规划
Codex任务明确时快速实现、生成测试、写 PR 描述
GitHub Copilot AgentGitHub 原生全流程自主执行(issue → PR)
Cursor开发者本地深度编码,OpenClaw 不干预

状态流转

graph LR
    Backlog --> Todo
    Todo --> InProgress[In Progress]
    InProgress -- 手动任务,PR open --> InReview[In Review]
    InReview -- PR merged --> Done
    InProgress -- AI 自动任务 --> Gate["Phase 4.5 验收 Gate"]
    Gate -- 通过 --> Done
    Gate -- 取消 --> Canceled
状态变更触发方时机
Backlog → TodoOpenClaw(自动任务)/ 用户确认(手动任务)排入 Sprint
Todo → In ProgressOpenClaw明确开始动工
In Progress → In ReviewOpenClaw 自动检测到关联分支有 PR open
In Progress → DoneOpenClaw 自动Phase 4.5 验收 Gate 通过(AI 自动任务)
In Review → DoneOpenClaw 自动检测到 PR merged
任意 → Canceled用户发起,OpenClaw 验收检查需求取消

Heartbeat 自动化链路

每 60 分钟,三条线并行:

Step A:Linear Auto 任务检查
        ├─ 超时告警(In Progress > 3 天无 PR)
        ├─ 验收兜底(AI 完成但 Gate 未触发)
        └─ 并发控制(Auto 任务 ≤ 3 个)

Step B:GitHub PR 扫描(flowchain 脚本)
        ├─ 新 PR open → 触发 Code Review
        ├─ PR merged → Linear issue 标 Done
        └─ CI 失败 → 立即推送告警

Step B2:邮件检查
        └─ Copilot Agent 开 PR → 邮件通知 → 触发验收 Gate

Step C:状态持久化 + NOW.md 覆写

关键约定

  • Branch 命名feature/{ISSUE_ID}-描述fix/{ISSUE_ID}-描述{ISSUE_ID} 是 OpenClaw 关联 PR 和 issue 的唯一依据
  • 验收标准:每个 issue 的 description 必须包含 ## 验收标准 区块,AI 自动任务以此为验收依据
  • Auto label:打了 Auto 的 issue 表示可由 AI 工具独立完成,OpenClaw 会主动调度并纳入 Heartbeat 监控
  • 代码不自动 push:OpenClaw 不会自动将代码推送到远程仓库,这一步永远需要人确认

3. 基础配置:身份与行为

SOUL.md

# SOUL.md - Who You Are

_You're not a chatbot. You're becoming someone._

## Core Truths

**Be genuinely helpful, not performatively helpful.** Skip the "Great question!" and "I'd be happy to help!" — just help.

**Have opinions.** You're allowed to disagree, prefer things, find stuff amusing or boring.

**Be resourceful before asking.** Try to figure it out. Read the file. Check the context. Search for it. _Then_ ask if you're stuck.

**Earn trust through competence.** Be careful with external actions (emails, public posts). Be bold with internal ones (reading, organizing, learning).

**Remember you're a guest.** You have access to someone's life. Treat it with respect.

## Boundaries

- Private things stay private.
- When in doubt, ask before acting externally.
- Never send half-baked replies to messaging surfaces.

## Continuity

Each session, you wake up fresh. These files _are_ your memory. Read them. Update them.

IDENTITY.md

# IDENTITY.md - Who Am I?

- **Name:** [给你的 AI 起个名字]
- **Creature:** AI assistant
- **Vibe:** Resourceful, direct, genuine.
- **Emoji:** [一个代表性 emoji]

USER.md

# USER.md - About Your Human

- **Name:** [用户名]
- **Timezone:** Asia/Shanghai (GMT+8)
- **Notes:** Prefers Chinese.

## Work

[职业背景]

## Interests

[关注领域]

## Contact Preference

Telegram 或 Dashboard(web chat)。

## Reply Style

[偏好风格]

AGENTS.md(核心部分)

# AGENTS.md - Your Workspace

## Every Session

Before doing anything else:

1. Read `SOUL.md` — this is who you are
2. Read `USER.md` — this is who you're helping
3. Read `memory/YYYY-MM-DD.md` (today + yesterday) for recent context
4. **If in MAIN SESSION**: Also read `MEMORY.md`
5. Read `WORKFLOW.md` — workflow index, know what SOPs exist

## Permission Levels

**Free to do without asking:**
- Read files, browse directories
- Search the web
- Check calendar and email
- Work within the workspace

**Must confirm with user first:**
- Sending emails, tweets, or public posts
- Any operation that sends data externally
- Deleting or modifying important files

## Heartbeats

When you receive a heartbeat poll, check `HEARTBEAT.md` and follow it.
If nothing needs attention, reply `HEARTBEAT_OK`.

## Workflow Playbooks

工作流 SOP 存放在 `workflow/` 目录,`WORKFLOW.md` 是索引(session 启动时已加载)。

收到任务时,先检查 WORKFLOW.md 是否有匹配的触发词:
- **命中** → read 对应 playbook 文件,按 SOP 执行
- **未命中** → 自由推断

执行逻辑(按 status + 保鲜期):

```

├─ status = draft      → 提醒用户这是未验证工作流,谨慎执行
├─ status = deprecated → 拒绝执行,告知已废弃
└─ status = active
   ├─ last_verified 距今 < ttl → 直接执行
   └─ last_verified 距今 ≥ ttl → 边执行边验证
      ├─ 有改进点 → 执行完后等用户确认,再更新 playbook
      └─ 无改进点 → 仅更新 last_verified

```

每次执行 workflow 后:
1. 在当天 `memory/YYYY-MM-DD.md` 追加执行记录
2. 更新 workflow 文件的 `last_run` 字段

4. 基础配置:记忆系统

三层架构

对话内容
    │ 实时写入

memory/YYYY-MM-DD.md        ← 每日日志(追加式,原始记录)
    │ 每晚 23:45 cron 提炼

memory/knowledge/            ← 结构化知识库
    ├── lessons/             ← 踩坑经验
    ├── decisions/           ← 重要决策
    └── people/              ← 人物档案
    │ 每周日 GC 归档

memory/.archive/             ← 冷存储(不主动加载)

两个特殊文件:

  • **MEMORY.md**:全局索引,硬性限制 40 行以内,每次主会话必读
  • **NOW.md**:当前状态快照,每次 heartbeat 覆写(不追加)

MEMORY.md 模板

# MEMORY.md — [AI名字]'s Index

## User
- Name: [用户名] | Timezone: Asia/Shanghai | Lang: Chinese preferred
- Role: [职业背景]
- Contact: Telegram / webchat

## Memory Layers
| Layer | File | Purpose |
|-------|------|---------|
| Index | MEMORY.md | This file. Keep <40 lines |
| Short-term | NOW.md | Current status, overwritten each heartbeat |
| Daily log | memory/YYYY-MM-DD.md | 时间戳事件流,append-only |
| Knowledge | memory/knowledge/INDEX.md | 提炼后的可复用知识导航 |
| Project index | status/PROJECTS.json | 项目注册表 |
| Status | status/ | 运行时状态 |
| Cold | memory/.archive/ | 归档冷数据,不主动加载 |

## Daily Log Rules
- Format: `### HH:MM — Title` + 细节,append-only,禁止覆写
-`scripts/memlog.sh "Title" "Body"` 写入,自动加时间戳
- 每晚 23:45 cron 提炼 → knowledge vault

memlog.sh

#!/usr/bin/env bash
# memlog.sh — 自动时间戳的日志追加工具
# 用法: memlog.sh "Title" "Content body"

set -euo pipefail

WORKSPACE_DIR="${WORKSPACE_DIR:-~/.openclaw/workspace}"
DAILY_DIR="$WORKSPACE_DIR/memory"
TODAY=$(TZ=Asia/Shanghai date +%Y-%m-%d)
WEEKDAY=$(TZ=Asia/Shanghai date +%A)
NOW=$(TZ=Asia/Shanghai date +%H:%M)
FILE="$DAILY_DIR/$TODAY.md"
TITLE="${1:?Usage: memlog.sh \"Title\" \"Body\"}"
BODY="${2:-}"

mkdir -p "$DAILY_DIR"

if [[ ! -f "$FILE" ]]; then
    cat > "$FILE" << EOF
# $TODAY · $WEEKDAY

## 今日一句

> 

---

## 事件流

---

## 收获 & 反思

> 

## 明天 / 待处理

- [ ] 
EOF
fi

ENTRY=$(printf "\n### %s — %s\n\n%s\n" "$NOW" "$TITLE" "$BODY")

python3 - "$FILE" "$ENTRY" << 'PYEOF'
import sys
filepath = sys.argv[1]
entry = sys.argv[2]
with open(filepath, 'r') as f:
    content = f.read()
marker = "\n---\n\n## 收获"
idx = content.find(marker)
if idx == -1:
    content += entry
else:
    content = content[:idx] + entry + content[idx:]
with open(filepath, 'w') as f:
    f.write(content)
PYEOF

Knowledge Vault CRUD 校验规则

写入 lessons/decisions/people/ 之前,必须先读再写:

准备写入知识文件

├─ Step 1: 读取目标文件当前内容(文件不存在则创建)
├─ Step 2: 比较新知识与已有内容
│  ├─ 已有内容完全覆盖 → NOOP(不写)
│  ├─ 新知识是对旧内容的更新 → UPDATE(旧版加 ~~Superseded~~ 标记)
│  ├─ 新知识与旧内容矛盾 → CONFLICT(两版保留,加 ⚠️ CONFLICT 标记)
│  └─ 全新知识 → ADD(追加新段落)
└─ Step 3: 更新 frontmatter 中的 last_verified 日期

知识文件 frontmatter 规范:

---
title: "标题"
date: YYYY-MM-DD
category: lessons | decisions | people
priority: 🔴 | 🟡 | ⚪
status: active | superseded | conflict
last_verified: YYYY-MM-DD
tags: [tag1, tag2]
---

优先级标记:🔴 核心知识永不归档,🟡 一般重要,⚪ 低优先级。超过 30 天未验证的条目加 ⚠️ stale 标记。

写入禁忌:

禁忌原因
❌ 用 write 覆写 memory/ 文件覆写 = 数据丢失(NOW.md 是唯一例外)
❌ 不读就写知识文件导致重复条目和冲突
❌ 硬编码时间戳用系统时间(date 命令或 memlog.sh)
❌ 写无实质内容的噪音浪费检索精度

5. 基础配置:Heartbeat 与 Cron

HEARTBEAT.md 基础模板

# HEARTBEAT.md
# Heartbeat runs every 60 minutes.

## 0. 推送提醒到 Telegram(有提醒事项时必做)

凡是需要提醒用户的事,必须用 `message` 工具主动推送到 Telegram。

**触发条件(满足任一即推送):**
- 日历事件距现在 < 2 小时
- 邮件命中 `status/MAILLIST.json` 🔴 立即推送规则
- 距上次主动推送 > 8 小时且有值得说的内容

**安静时段(23:00–08:00)豁免条件:**
- CI 失败
- 日历事件距现在 < 30 分钟
- 邮件命中 🔴 立即推送规则

**不推送的情况:**
- 23:00–08:00 安静时段(豁免条件除外)
- 没有新事项,仅例行 heartbeat
- 上次推送 < 60 分钟前

项目监控部分在第 11 节单独给出,追加到此文件末尾。

Cron 定时任务

{
  "crons": [
    {
      "name": "daily-reflection",
      "schedule": "45 23 * * *",
      "task": "执行每日反思流程:读取今日 memory/YYYY-MM-DD.md,提炼有价值的内容写入对应知识库文件(lessons/decisions/people),同步更新 MEMORY.md 中的核心信息,清理噪声记录。在反思推送中包含当天执行过的 workflow 报告,等用户确认后回写 playbook。"
    },
    {
      "name": "weekly-knowledge-distill",
      "schedule": "0 0 * * 0",
      "task": "扫描最近7天的日志,检查 knowledge/INDEX.md 中 stale 标记(>30天未验证),对过期条目标注 ⚠️,将超过阈值的旧日志移入 memory/.archive/(保留🔴优先级的知识文件)。"
    },
    {
      "name": "weekly-security-check",
      "schedule": "0 10 * * 1",
      "task": "执行安全检查:运行 openclaw security audit,检查监听端口变化(与上次结果对比),如有新增问题或未知端口,通过 Telegram 推送告警。结果记录到 memory/YYYY-MM-DD.md。"
    }
  ]
}

6. 接入渠道:Telegram

配置文件路径:~/.openclaw/config.json(OpenClaw 主配置文件)。

通过 BotFather 创建 Bot 并获取 token,然后写入以下配置:

{
  channels: {
    telegram: {
      enabled: true,
      botToken: "YOUR_BOT_TOKEN",

      // 私聊权限策略:pairing(推荐)| allowlist | open | disabled
      // pairing:首次配对后自动加入白名单,不需要手动维护 allowFrom
      dmPolicy: "pairing",
      allowFrom: [],   // allowlist 模式下填数字 ID,不支持 @username

      // 群组权限策略:allowlist | open | disabled
      groupPolicy: "allowlist",
      groups: {
        "-1001234567890": {        // 群组数字 ID(负数),获取方式:转发消息给 @getidsbot
          requireMention: true,    // 需要 @bot 才响应
          groupPolicy: "open",     // 该群组允许所有成员
        },
        "*": {
          requireMention: true,    // 全局默认:需要 @mention
        },
      },

      // 流式输出:partial(推荐)| block | progress | off
      // partial:DM 中原地更新草稿消息,生成完成后无第二条消息,体验最干净
      streaming: "partial",

      // Inline Buttons:off | dm | group | all | allowlist
      capabilities: {
        inlineButtons: "allowlist",
      },

      // 消息处理中显示 Ack 表情(👀 表示"收到,处理中")
      ackReaction: "👀",
      reactionLevel: "minimal",        // off | ack | minimal | extensive
      reactionNotifications: "own",    // 用户对 Bot 消息点表情时是否触发通知

      // 自定义命令菜单(Telegram 左下角 / 列表)
      // 命令只是菜单入口,行为由 AI 根据 skill 和上下文决定
      customCommands: [
        { command: "brief", description: "今日简报" },
        { command: "sprint", description: "项目进度" },
        { command: "issue", description: "新建 issue" },
      ],
    },
  },
}

配对流程(首次使用):

openclaw gateway                        # 启动
openclaw pairing list telegram          # 查看待配对请求
openclaw pairing approve telegram <CODE>  # 批准配对

7. 模型配置

配置文件路径:~/.openclaw/config.json,与 Telegram 配置在同一文件中。

{
  agents: {
    defaults: {
      // 主模型 + Fallback 链
      // 主模型失败(限速/超时)时依次尝试 fallbacks,全部失败才报错
      // 推荐用 OpenRouter 作为统一入口:一个 Key 接入几十个模型
      model: {
        primary: "openrouter/anthropic/claude-sonnet-4-6",
        fallbacks: [
          "openrouter/google/gemini-2.5-pro",
          "openrouter/deepseek/deepseek-chat",
        ],
      },

      // 模型 Alias:配置后切换时不需要输入完整路径
      // 在 Telegram 中发 /model sonnet 即可切换
      models: {
        "openrouter/anthropic/claude-sonnet-4-6": { alias: "sonnet" },
        "openrouter/google/gemini-2.5-pro":       { alias: "gemini" },
        "openrouter/deepseek/deepseek-chat":       { alias: "deepseek" },
        "openrouter/deepseek/deepseek-reasoner":   { alias: "r1" },
      },
    },
  },
}

环境变量(写入 ~/.openclaw/.env 或系统环境):

Provider环境变量说明
OpenRouterOPENROUTER_API_KEY统一入口,推荐
AnthropicANTHROPIC_API_KEY直连
GoogleGEMINI_API_KEYGemini 系列
DeepSeekDEEPSEEK_API_KEY直连,价格低

多 Key 轮换:配置 OPENROUTER_API_KEYS(逗号分隔),限速时自动轮换。

运行时切换模型(在 Telegram 对话中):

/model              → 打开模型选择器
/model sonnet       → 切换到 claude-sonnet-4-6
/model r1           → 切换到 DeepSeek R1
/model status       → 查看当前模型状态

切换只影响当前 session,不修改配置文件。/new 开启新 session 后恢复默认模型。


8. 项目管理核心:status/ 与 flowchain/

status/PROJECTS.json

项目注册表,所有被监控的项目都在这里维护:

{
  "projects": [
    {
      "name": "ProjectA",
      "description": "项目描述",
      "local_path": "~/Documents/GitHub/ProjectA",
      "linear_id": "****-****-****-****",
      "github": "your-org/ProjectA",
      "status": "进行中",
      "heartbeat": true
    },
    {
      "name": "ProjectB",
      "description": "项目描述",
      "local_path": "~/Documents/GitHub/ProjectB",
      "linear_id": "****-****-****-****",
      "github": "your-org/ProjectB",
      "status": "进行中",
      "heartbeat": false
    }
  ],
  "last_updated": "YYYY-MM-DD"
}

heartbeat: true 的项目会被纳入每次 heartbeat 的 PR 扫描范围。

status/heartbeat-state.json

运行时状态,由 flowchain 脚本自动维护:

{
  "validation_retries": {},
  "_updated": "YYYY-MM-DDTHH:MM:SS",
  "last_email_check": "YYYY-MM-DDTHH:MM:SS+08:00",
  "seen_merged": {},
  "seen_ci": {},
  "seen_open_stale": {},
  "seen_stale_issues": {},
  "last_heartbeat": "YYYY-MM-DDTHH:MM:SS+08:00"
}

去重规则:

类别去重字段说明
CI 失败seen_ci同一 PR 的 CI 失败只推一次
Stale open PRseen_open_stale同一 PR 的 stale 提醒只推一次
Stale Linear issueseen_stale_issues + updatedAt 对比有新动弹才重报
Merged PR → Doneseen_merged同一 PR 只推一次

status/MAILLIST.json

邮件监控规则与运行状态:

{
  "last_summary_date": "YYYY-MM-DD",
  "last_urgent_ids": [],
  "last_run": "YYYY-MM-DDTHH:MM:SS+08:00",
  "config": {
    "immediate": {
      "sender_whitelist": [
        "*@github.com",
        "[email protected]"
      ],
      "subject_keywords": [
        "安全",
        "security alert",
        "unauthorized",
        "invoice",
        "payment",
        "账单",
        "offer",
        "录用"
      ]
    },
    "summary": {
      "label_include": ["IMPORTANT", "INBOX", "CATEGORY_UPDATES"],
      "label_exclude": ["CATEGORY_PROMOTIONS", "CATEGORY_SOCIAL"],
      "sender_watchlist": ["*@linear.app", "*@github.com"]
    },
    "ignore": {
      "label_blacklist": ["CATEGORY_PROMOTIONS", "CATEGORY_SOCIAL"],
      "sender_blacklist": ["*@linkedin.com"]
    }
  }
}

flowchain/projects.py 接口

所有 Linear 和 GitHub 操作统一通过此脚本执行,输出协议为 [ok] / [warn] / [error] 前缀:

# Heartbeat 全量扫描(输出结构化 JSON)
python3 flowchain/projects.py heartbeat

# Sprint 报告
python3 flowchain/projects.py sprint
python3 flowchain/projects.py sprint ProjectA

# Issue 操作
python3 flowchain/projects.py issue create "标题" --project ProjectA
python3 flowchain/projects.py issue view GEO-123
python3 flowchain/projects.py issue move GEO-123 "In Progress"
python3 flowchain/projects.py issue label GEO-123 Feature
python3 flowchain/projects.py issue priority GEO-123 high
python3 flowchain/projects.py issue start GEO-123
python3 flowchain/projects.py issue cancel GEO-123
python3 flowchain/projects.py issue report GEO-123 \
  --passed "验收项A:通过" \
  --failed "验收项B:未通过(原因)" \
  --conclusion "⚠️ 需修复"

heartbeat 命令输出 JSON 格式:

{
  "ci_failures":        [{"repo": "org/Repo", "pr_number": 5, "pr_title": "fix: crash", "identifier": "GEO-12"}],
  "pr_open_stale":      [{"repo": "org/Repo", "pr_number": 12, "pr_title": "feat: ...", "identifier": "GEO-45", "hours_open": 25}],
  "pr_merged_to_done":  [{"repo": "org/Repo", "pr_number": 8, "pr_title": "feat: ...", "identifier": "GEO-45"}],
  "stale_in_progress":  [{"identifier": "GEO-30", "title": "...", "days_in_progress": 4}],
  "pr_moved_to_review": [{"identifier": "GEO-45", "pr_number": 12}],
  "validation_retries": [{"identifier": "GEO-20", "retries": 3}]
}

flowchain/projects.py 源码

展开查看完整源码
#!/usr/bin/env python3
"""
flowchain/projects.py — Linear + GitHub 操作统一执行器

输出协议:
  stdout: [ok] / [warn] / [error] 前缀 + 内容
  stderr: 调试信息(不影响 OpenClaw 解析)
  heartbeat 子命令:stdout 输出纯 JSON

依赖:
  pip install linear-sdk PyGithub python-dateutil
  环境变量:LINEAR_API_KEY, GITHUB_TOKEN
"""

import argparse
import json
import os
import re
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path

# ── 路径配置 ──────────────────────────────────────────────────────────────────

WORKSPACE = Path(os.environ.get("OPENCLAW_WORKSPACE", Path.home() / ".openclaw" / "workspace"))
PROJECTS_JSON   = WORKSPACE / "status" / "PROJECTS.json"
HB_STATE_JSON   = WORKSPACE / "status" / "heartbeat-state.json"

LINEAR_API_KEY  = os.environ.get("LINEAR_API_KEY", "")
GITHUB_TOKEN    = os.environ.get("GITHUB_TOKEN", "")

STALE_PR_HOURS       = int(os.environ.get("STALE_PR_HOURS", "24"))
STALE_ISSUE_DAYS     = int(os.environ.get("STALE_ISSUE_DAYS", "3"))
MAX_VALIDATION_RETRY = int(os.environ.get("MAX_VALIDATION_RETRY", "3"))

# ── 工具函数 ──────────────────────────────────────────────────────────────────

def ok(msg: str):
    print(f"[ok] {msg}")

def warn(msg: str):
    print(f"[warn] {msg}")

def err(msg: str, exit_code: int = 1):
    print(f"[error] {msg}", file=sys.stderr)
    sys.exit(exit_code)

def now_iso() -> str:
    return datetime.now(timezone.utc).astimezone().isoformat(timespec="seconds")

def load_json(path: Path) -> dict:
    if not path.exists():
        return {}
    with open(path, encoding="utf-8") as f:
        return json.load(f)

def save_json(path: Path, data: dict):
    path.parent.mkdir(parents=True, exist_ok=True)
    with open(path, "w", encoding="utf-8") as f:
        json.dump(data, f, ensure_ascii=False, indent=2)

def load_projects() -> list[dict]:
    data = load_json(PROJECTS_JSON)
    return data.get("projects", [])

def find_project(name: str) -> dict | None:
    for p in load_projects():
        if p["name"].lower() == name.lower():
            return p
    return None

def gh(*args) -> str:
    """调用 gh CLI,返回 stdout;失败时抛出 RuntimeError。"""
    result = subprocess.run(
        ["gh", *args],
        capture_output=True, text=True
    )
    if result.returncode != 0:
        raise RuntimeError(result.stderr.strip())
    return result.stdout.strip()

def linear_query(query: str, variables: dict | None = None) -> dict:
    """执行 Linear GraphQL 查询,返回 data 字段。"""
    import urllib.request
    payload = json.dumps({"query": query, "variables": variables or {}}).encode()
    req = urllib.request.Request(
        "https://api.linear.app/graphql",
        data=payload,
        headers={
            "Content-Type": "application/json",
            "Authorization": LINEAR_API_KEY,
        },
    )
    with urllib.request.urlopen(req) as resp:
        body = json.loads(resp.read())
    if "errors" in body:
        raise RuntimeError(body["errors"][0]["message"])
    return body["data"]

def extract_identifier(text: str) -> str | None:
    """从 PR 标题或 body 中提取 Linear identifier,如 GEO-123。"""
    m = re.search(r"\b([A-Z]{2,6}-\d+)\b", text or "")
    return m.group(1) if m else None

# ── heartbeat ─────────────────────────────────────────────────────────────────

def cmd_heartbeat(_args):
    state = load_json(HB_STATE_JSON)
    seen_ci          = state.get("seen_ci", {})
    seen_open_stale  = state.get("seen_open_stale", {})
    seen_merged      = state.get("seen_merged", {})
    seen_stale_issues = state.get("seen_stale_issues", {})
    validation_retries = state.get("validation_retries", {})

    result = {
        "ci_failures":        [],
        "pr_open_stale":      [],
        "pr_merged_to_done":  [],
        "stale_in_progress":  [],
        "pr_moved_to_review": [],
        "validation_retries": [],
    }

    projects = [p for p in load_projects() if p.get("heartbeat")]

    for proj in projects:
        repo = proj.get("github", "")
        if not repo:
            continue

        # ── Open PR 扫描 ──────────────────────────────────────────────────────
        try:
            raw = gh("pr", "list", "--repo", repo,
                     "--state", "open", "--json",
                     "number,title,body,statusCheckRollup,createdAt")
            prs = json.loads(raw)
        except Exception as e:
            print(f"[warn] {repo} open PR 拉取失败: {e}", file=sys.stderr)
            prs = []

        for pr in prs:
            num   = pr["number"]
            title = pr.get("title", "")
            body  = pr.get("body", "")
            ident = extract_identifier(title) or extract_identifier(body)
            key   = f"{repo}#{num}"

            # CI 失败检测
            checks = pr.get("statusCheckRollup") or []
            failed = [c for c in checks if c.get("conclusion") == "FAILURE"]
            if failed and key not in seen_ci:
                result["ci_failures"].append({
                    "repo": repo, "pr_number": num,
                    "pr_title": title, "identifier": ident,
                })
                seen_ci[key] = now_iso()

            # Stale open PR(超过阈值小时未合并)
            created = datetime.fromisoformat(pr["createdAt"].replace("Z", "+00:00"))
            hours_open = (datetime.now(timezone.utc) - created).total_seconds() / 3600
            if hours_open >= STALE_PR_HOURS and key not in seen_open_stale:
                result["pr_open_stale"].append({
                    "repo": repo, "pr_number": num,
                    "pr_title": title, "identifier": ident,
                    "hours_open": round(hours_open, 1),
                })
                seen_open_stale[key] = now_iso()

            # PR 已开启 → 自动移入 In Review
            if ident and key not in seen_open_stale:
                result["pr_moved_to_review"].append({
                    "identifier": ident, "pr_number": num,
                })

        # ── Merged PR 扫描 ────────────────────────────────────────────────────
        try:
            raw = gh("pr", "list", "--repo", repo,
                     "--state", "merged", "--limit", "20", "--json",
                     "number,title,body,mergedAt")
            merged_prs = json.loads(raw)
        except Exception as e:
            print(f"[warn] {repo} merged PR 拉取失败: {e}", file=sys.stderr)
            merged_prs = []

        for pr in merged_prs:
            num   = pr["number"]
            title = pr.get("title", "")
            body  = pr.get("body", "")
            ident = extract_identifier(title) or extract_identifier(body)
            key   = f"{repo}#{num}"
            if ident and key not in seen_merged:
                result["pr_merged_to_done"].append({
                    "repo": repo, "pr_number": num,
                    "pr_title": title, "identifier": ident,
                })
                seen_merged[key] = now_iso()

    # ── Linear stale In Progress issues ──────────────────────────────────────
    try:
        data = linear_query("""
            query {
              issues(filter: {
                state: { name: { eq: "In Progress" } }
              }, first: 50) {
                nodes {
                  identifier title updatedAt
                  state { name }
                }
              }
            }
        """)
        for issue in data["issues"]["nodes"]:
            ident = issue["identifier"]
            updated = datetime.fromisoformat(issue["updatedAt"].replace("Z", "+00:00"))
            days = (datetime.now(timezone.utc) - updated).total_seconds() / 86400
            prev = seen_stale_issues.get(ident)
            if days >= STALE_ISSUE_DAYS and (not prev or prev != issue["updatedAt"]):
                result["stale_in_progress"].append({
                    "identifier": ident,
                    "title": issue["title"],
                    "days_in_progress": round(days, 1),
                })
                seen_stale_issues[ident] = issue["updatedAt"]
    except Exception as e:
        print(f"[warn] Linear stale issue 查询失败: {e}", file=sys.stderr)

    # ── 验收重试计数上报 ──────────────────────────────────────────────────────
    for ident, count in validation_retries.items():
        if count >= MAX_VALIDATION_RETRY:
            result["validation_retries"].append({
                "identifier": ident, "retries": count,
            })

    # ── 写回状态 ──────────────────────────────────────────────────────────────
    state.update({
        "seen_ci":           seen_ci,
        "seen_open_stale":   seen_open_stale,
        "seen_merged":       seen_merged,
        "seen_stale_issues": seen_stale_issues,
        "validation_retries": validation_retries,
        "last_heartbeat":    now_iso(),
        "_updated":          now_iso(),
    })
    save_json(HB_STATE_JSON, state)

    print(json.dumps(result, ensure_ascii=False, indent=2))

# ── sprint ────────────────────────────────────────────────────────────────────

def cmd_sprint(args):
    project_name = args.project_name

    query = """
        query($filter: IssueFilter) {
          issues(filter: $filter, first: 100) {
            nodes {
              identifier title priority
              state { name }
              assignee { name }
              labels { nodes { name } }
              updatedAt
            }
          }
        }
    """
    variables: dict = {}
    if project_name:
        proj = find_project(project_name)
        if not proj:
            err(f"项目 {project_name!r} 不在 PROJECTS.json 中")
        variables["filter"] = {"project": {"id": {"eq": proj["linear_id"]}}}

    try:
        data = linear_query(query, variables)
    except Exception as e:
        err(f"Linear 查询失败: {e}")

    issues = data["issues"]["nodes"]
    by_state: dict[str, list] = {}
    for issue in issues:
        state = issue["state"]["name"]
        by_state.setdefault(state, []).append({
            "identifier": issue["identifier"],
            "title":      issue["title"],
            "priority":   issue["priority"],
            "assignee":   issue.get("assignee", {}).get("name") if issue.get("assignee") else None,
            "labels":     [l["name"] for l in issue["labels"]["nodes"]],
            "updatedAt":  issue["updatedAt"],
        })

    print(json.dumps({"project": project_name, "by_state": by_state}, ensure_ascii=False, indent=2))

# ── issue ─────────────────────────────────────────────────────────────────────

def _linear_issue_id(identifier: str) -> str:
    """将 identifier(如 GEO-123)解析为 Linear 内部 UUID。"""
    data = linear_query(
        'query($id: String!) { issue(id: $id) { id } }',
        {"id": identifier}
    )
    return data["issue"]["id"]

def _linear_state_id(state_name: str) -> str:
    data = linear_query(
        'query($name: String!) { workflowStates(filter: { name: { eq: $name } }) { nodes { id } } }',
        {"name": state_name}
    )
    nodes = data["workflowStates"]["nodes"]
    if not nodes:
        raise RuntimeError(f"状态 {state_name!r} 不存在")
    return nodes[0]["id"]

def _linear_label_id(label_name: str) -> str:
    data = linear_query(
        'query($name: String!) { issueLabels(filter: { name: { eq: $name } }) { nodes { id } } }',
        {"name": label_name}
    )
    nodes = data["issueLabels"]["nodes"]
    if not nodes:
        raise RuntimeError(f"标签 {label_name!r} 不存在")
    return nodes[0]["id"]

PRIORITY_MAP = {"urgent": 1, "high": 2, "medium": 3, "low": 4, "no priority": 0}

def cmd_issue(args):
    sub = args.issue_cmd

    if sub == "create":
        proj = find_project(args.project)
        if not proj:
            err(f"项目 {args.project!r} 不在 PROJECTS.json 中")
        try:
            data = linear_query(
                """
                mutation($title: String!, $projectId: String!) {
                  issueCreate(input: { title: $title, projectId: $projectId }) {
                    issue { identifier title }
                  }
                }
                """,
                {"title": args.title, "projectId": proj["linear_id"]},
            )
            issue = data["issueCreate"]["issue"]
            ok(f"{issue['identifier']}{issue['title']}")
        except Exception as e:
            err(f"创建 issue 失败: {e}")

    elif sub == "view":
        try:
            data = linear_query(
                """
                query($id: String!) {
                  issue(id: $id) {
                    identifier title description priority
                    state { name }
                    assignee { name }
                    labels { nodes { name } }
                    comments { nodes { body createdAt user { name } } }
                  }
                }
                """,
                {"id": args.identifier},
            )
            print(json.dumps(data["issue"], ensure_ascii=False, indent=2))
        except Exception as e:
            err(f"查询 issue 失败: {e}")

    elif sub == "move":
        try:
            issue_id = _linear_issue_id(args.identifier)
            state_id = _linear_state_id(args.state)
            linear_query(
                "mutation($id: String!, $stateId: String!) { issueUpdate(id: $id, input: { stateId: $stateId }) { success } }",
                {"id": issue_id, "stateId": state_id},
            )
            ok(f"{args.identifier}{args.state}")
        except Exception as e:
            err(f"移动 issue 失败: {e}")

    elif sub == "label":
        try:
            issue_id = _linear_issue_id(args.identifier)
            label_id = _linear_label_id(args.label)
            linear_query(
                "mutation($id: String!, $labelIds: [String!]!) { issueAddLabel(id: $id, labelId: $labelIds) { success } }",
                {"id": issue_id, "labelIds": [label_id]},
            )
            ok(f"{args.identifier} 标签 → {args.label}")
        except Exception as e:
            err(f"打标签失败: {e}")

    elif sub == "priority":
        prio_val = PRIORITY_MAP.get(args.priority.lower())
        if prio_val is None:
            err(f"优先级无效,可选:{', '.join(PRIORITY_MAP.keys())}")
        try:
            issue_id = _linear_issue_id(args.identifier)
            linear_query(
                "mutation($id: String!, $priority: Int!) { issueUpdate(id: $id, input: { priority: $priority }) { success } }",
                {"id": issue_id, "priority": prio_val},
            )
            ok(f"{args.identifier} 优先级 → {args.priority}")
        except Exception as e:
            err(f"设置优先级失败: {e}")

    elif sub == "start":
        try:
            issue_id = _linear_issue_id(args.identifier)
            state_id = _linear_state_id("In Progress")
            linear_query(
                "mutation($id: String!, $stateId: String!) { issueUpdate(id: $id, input: { stateId: $stateId }) { success } }",
                {"id": issue_id, "stateId": state_id},
            )
            ok(f"{args.identifier} → In Progress")
        except Exception as e:
            err(f"开始 issue 失败: {e}")

    elif sub == "cancel":
        try:
            issue_id = _linear_issue_id(args.identifier)
            state_id = _linear_state_id("Cancelled")
            linear_query(
                "mutation($id: String!, $stateId: String!) { issueUpdate(id: $id, input: { stateId: $stateId }) { success } }",
                {"id": issue_id, "stateId": state_id},
            )
            ok(f"{args.identifier} → Cancelled")
        except Exception as e:
            err(f"取消 issue 失败: {e}")

    elif sub == "report":
        passed   = args.passed or []
        failed   = args.failed or []
        conclusion = args.conclusion or ""
        lines = ["## 验收报告\n"]
        if passed:
            lines.append("**通过项**")
            for p in passed:
                lines.append(f"- ✅ {p}")
        if failed:
            lines.append("\n**未通过项**")
            for f_ in failed:
                lines.append(f"- ❌ {f_}")
        if conclusion:
            lines.append(f"\n**结论:** {conclusion}")
        comment_body = "\n".join(lines)

        # 更新验收重试计数
        state = load_json(HB_STATE_JSON)
        retries = state.get("validation_retries", {})
        if failed:
            retries[args.identifier] = retries.get(args.identifier, 0) + 1
        else:
            retries.pop(args.identifier, None)
        state["validation_retries"] = retries
        save_json(HB_STATE_JSON, state)

        try:
            issue_id = _linear_issue_id(args.identifier)
            linear_query(
                "mutation($id: String!, $body: String!) { commentCreate(input: { issueId: $id, body: $body }) { success } }",
                {"id": issue_id, "body": comment_body},
            )
            if failed:
                warn(f"{args.identifier} 验收未通过(重试次数: {retries.get(args.identifier, 1)})")
            else:
                ok(f"{args.identifier} 验收通过,评论已写入")
        except Exception as e:
            err(f"写入验收报告失败: {e}")

    else:
        err(f"未知 issue 子命令: {sub}")

# ── CLI 入口 ──────────────────────────────────────────────────────────────────

def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(
        prog="projects.py",
        description="flowchain — Linear + GitHub 操作统一执行器",
    )
    sub = parser.add_subparsers(dest="cmd", required=True)

    # heartbeat
    sub.add_parser("heartbeat", help="全量扫描,输出结构化 JSON")

    # sprint
    sp = sub.add_parser("sprint", help="Sprint 报告")
    sp.add_argument("project_name", nargs="?", default=None, help="项目名(可选)")

    # issue
    ip = sub.add_parser("issue", help="Issue 操作")
    isub = ip.add_subparsers(dest="issue_cmd", required=True)

    ic = isub.add_parser("create", help="创建 issue")
    ic.add_argument("title")
    ic.add_argument("--project", required=True)

    iv = isub.add_parser("view", help="查看 issue 详情")
    iv.add_argument("identifier")

    im = isub.add_parser("move", help="移动 issue 状态")
    im.add_argument("identifier")
    im.add_argument("state")

    il = isub.add_parser("label", help="给 issue 打标签")
    il.add_argument("identifier")
    il.add_argument("label")

    ipr = isub.add_parser("priority", help="设置优先级")
    ipr.add_argument("identifier")
    ipr.add_argument("priority", choices=list(PRIORITY_MAP.keys()))

    ist = isub.add_parser("start", help="开始 issue(→ In Progress)")
    ist.add_argument("identifier")

    ica = isub.add_parser("cancel", help="取消 issue(→ Cancelled)")
    ica.add_argument("identifier")

    irp = isub.add_parser("report", help="写入验收报告评论")
    irp.add_argument("identifier")
    irp.add_argument("--passed",     action="append", metavar="TEXT", help="通过项(可多次)")
    irp.add_argument("--failed",     action="append", metavar="TEXT", help="未通过项(可多次)")
    irp.add_argument("--conclusion", metavar="TEXT",  help="总结结论")

    return parser

def main():
    if not LINEAR_API_KEY:
        err("环境变量 LINEAR_API_KEY 未设置")

    parser = build_parser()
    args = parser.parse_args()

    dispatch = {
        "heartbeat": cmd_heartbeat,
        "sprint":    cmd_sprint,
        "issue":     cmd_issue,
    }
    dispatch[args.cmd](args)

if __name__ == "__main__":
    main()

9. 项目管理核心:Workflow 系统

WORKFLOW.md 索引

# WORKFLOW.md — 工作流索引

_Last updated: YYYY-MM-DD_

> 收到任务时先匹配触发词;命中则 read 对应文件按 SOP 执行;未命中则自由推断。

## 工作流列表

| # | 工作流 | 触发词 | 文件 | Status | TTL | 简述 |
|---|--------|--------|------|--------|-----|------|
| 00 | 创建工作流 | 新建工作流、创建 workflow、建个 SOP | workflow/00-create-workflow.md | active | 180d | 标准化新建任何 workflow 的元流程 |
| 01 | 项目管理与自动化开发 | 项目进度、有个bug、新建issue、sprint报告、帮我建issue、开始开发 | workflow/01-project-management.md | active | 90d | Linear+GitHub全链路项目管理,AI工具分工,自动化推送 |
| 02 | 邮件监控 | 检查邮件、查邮件、邮件汇总、有没有重要邮件 | workflow/02-email-check.md | active | 30d | heartbeat 邮件检查,Copilot PR 通知自动触发验收 Gate |

## 执行规则(速查)

- **draft** → 提醒用户未验证,谨慎执行
- **active + 在保鲜期内** → 直接执行
- **active + 超过 TTL** → 边执行边验证,完成后更新 last_verified
- **deprecated** → 拒绝执行,告知已废弃
- **skill 缺失** → 告知,建议安装或新建,不主动安装

00-create-workflow.md

---
title: "创建工作流"
created: YYYY-MM-DD
last_run: ~
last_verified: YYYY-MM-DD
ttl: 180d
status: active
skills: []
tags: [meta, workflow]
---

# 创建工作流 Playbook

## 触发条件
说:新建工作流、创建 workflow、建个 SOP、帮我把 X 做成工作流

## 前置检查
- [ ] 确认新工作流的名称和用途
- [ ] 确认触发词(2-4 个关键词)
- [ ] 确认所需 skills
- [ ] 确认 TTL(参考类型建议表)

## 执行步骤

### Step 1 — 确定编号
查看 workflow/ 目录,取当前最大数字前缀 +1 作为新文件编号。

### Step 2 — 创建 Playbook 文件
文件名格式:workflow/NN-slug.md(slug 用英文小写 + 连字符)

Frontmatter 模板:
---
title: "工作流名称"
created: YYYY-MM-DD
last_run: ~
last_verified: YYYY-MM-DD
ttl: 30d
status: draft
skills: []
tags: []
---

新建时默认 status: draft,首次执行验证后改 active。

### Step 3 — 填写 Playbook 内容
必须包含以下章节:
- 触发条件
- 前置检查(含参数/条件)
- 执行步骤(Step by step,含具体命令)
- 权限边界(AI 可以自主决定什么,什么必须询问)
- 输出/交付物
- 异常处理

### Step 4 — 更新 WORKFLOW.md 索引
在表格追加一行:
| NN | 工作流名 | 触发词 | workflow/NN-slug.md | draft | TTL | 简述 |

### Step 5 — 通知用户
新工作流已创建,状态 draft,首次执行后可升 active。

## 主动识别机会
发现某件事反复出现 ≥ 3 次且没有对应 workflow 时,主动建议:
> "我注意到 [X] 已经出现了几次,要不要建个工作流?"

## TTL 建议表

| 类型 | 建议 TTL |
|------|---------|
| 依赖 CLI/API 工具 | 30d |
| 依赖外部平台(爬虫、网页) | 14d |
| 纯流程(招聘、调研) | 90d |
| 元工作流 | 180d |

10. 项目管理核心:01-project-management.md

这是整套系统的核心 Playbook,直接复制到 workflow/01-project-management.md,替换占位符后即可使用。

---
title: "项目管理与自动化开发"
created: YYYY-MM-DD
last_run: ~
last_verified: YYYY-MM-DD
ttl: 90d
status: active
skills: [linear-cli, github, coding-agent]
tags: [project-management, linear, github, automation, ai-coding, agile]
---

# 项目管理与自动化开发

> 基于敏捷开发节奏,Linear + GitHub 为底座,OpenClaw 作为自动化中间层。

⚠️ **执行前必读 `status/PROJECTS.json`**:获取项目 ID、GitHub Repo 对应关系、监控范围。

---

## 触发条件

项目进度、查一下XX、XX做完了、有个bug、新建issue、sprint报告、帮我建issue、开始开发XX、预定级、拆分任务

---

## 配置(单一 source of truth)

> ⚠️ API key、Team ID、Project IDs、State IDs 统一存放在 `credentials/linear.json`
> 格式参考:`credentials/linear.example.json`

### Label 规范

| Label | 含义 |
|-------|------|
| `Bug` | Bug 修复 |
| `Feature` | 新功能 |
| `Chore` | 杂项任务 |
| `Docs` | 文档更新 |
| `Auto` | AI 自动执行任务(Heartbeat 监控依据,排期为自动任务时必打)|

### Issue 命名规范

- Feature:`[Feature] 简短描述`
- Bug:`[Bug] 简短描述`

### Branch 命名规范

```
feature/{ISSUE_ID}-简短描述
fix/{ISSUE_ID}-简短描述
```

> `{ISSUE_ID}` 即 Linear 返回的 identifier(如 `GEO-123`)。
> Branch 名包含 `{ISSUE_ID}` 是 OpenClaw 判断 PR 关联 issue 的唯一依据。

### AI 工具分工

| 阶段 | 工具 | 触发方 | 用途 |
|------|------|--------|------|
| 规划 | Claude Code | OpenClaw 后台调用 | 技术方案、架构、难点分析 |
| 实现(手动) | Cursor | 用户明确说"我自己来" | 本地 IDE 编码,OpenClaw 不干预 |
| 实现(AI 驱动) | Claude Code / Codex / Copilot Agent | **OpenClaw 决策调度** | 见下方工具决策规则 |
| Review | Claude Code | OpenClaw 自动触发(PR open)| diff 分析、问题标注 |

### OpenClaw 工具决策规则

| 场景 | Claude Code | Codex | Copilot Agent |
|------|:-----------:|:-----:|:-------------:|
| 任务描述模糊,需理解上下文 | ✅ | | |
| 跨文件分析 / 架构判断 | ✅ | | |
| Bug 原因不明,需推理 | ✅ | | |
| 收尾文档整理 | ✅ | | |
| 任务清晰、范围小、步骤明确 | | ✅ | |
| 生成测试用例 | | ✅ | |
| 生成 PR 描述 | | ✅ | |
| Bug 修复(定位已明确)| | ✅ | |
| GitHub issue 直接指派 AI 全流程执行 | | | ✅ |

默认优先 Claude Code;任务极清晰时选 Codex;需要 GitHub 原生全流程时选 Copilot Agent。

### AI 调用规范

```bash
# Claude Code(规划/Review)
cd /path/to/{repo} && claude --permission-mode bypassPermissions --print '任务描述'

# Codex(测试/PR 描述,需要 PTY)
# exec(pty=true, workdir=/path/to/{repo}, command="codex exec --full-auto '任务描述'")

# GitHub Copilot Agent(issue 指派后在 GitHub 上自主执行)
# gh issue edit {number} --repo your-org/{repo} --add-assignee @copilot
```

---

## 敏捷流程总览

```
需求收集 → Backlog 梳理 → Sprint 规划 → 开发 → Code Review → 验收完成 → 回顾
    ↑_________________________反馈循环________________________________|
```

---

## Phase 1 · 需求收集

**入口:** 随时触发(Telegram / Linear 直接建)

### 三种来源

**A. 告知 OpenClaw(最常见)**
- 说出需求 → OpenClaw 运行 `python3 flowchain/projects.py issue create "标题" --project <项目名>` → 推送确认(含 identifier)
- ⚠️ 建 issue 时必须在 description 中包含 `## 验收标准` 区块(格式:`- [ ] 验收项`
- 若未明确提供验收标准,OpenClaw 根据需求描述自行推断后写入,建完连同验收标准推送确认

**B. 直接在 Linear 建**
- OpenClaw 在下次 heartbeat 时检测到新 Backlog issue → 进入 Phase 2
- ⚠️ 若缺少 `## 验收标准` 区块,OpenClaw 主动补写,推送确认

**C. 先写代码后补记录**
- 说"xxx 做完了,帮我记一下" → OpenClaw 建 issue 并立即标记 Done

**出口:** Issue 存在于 Backlog,且 description 包含 `## 验收标准` 区块

---

## Phase 2 · Backlog 梳理(OpenClaw 主动,不等触发)

**入口:** Issue 进入 Backlog

### 2.1 打 Label

```bash
python3 flowchain/projects.py issue label {ISSUE_ID} <Bug|Feature|Chore|Docs|Auto>
```

不确定时默认 `Feature`

### 2.2 评估优先级


| 优先级      | 判断标准           |
| -------- | -------------- |
| `Urgent` | 影响主流程 / 线上故障   |
| `High`   | 重要功能 / 计划内核心工作 |
| `Medium` | 一般改进           |
| `Low`    | 可延期优化          |


```bash
python3 flowchain/projects.py issue priority {ISSUE_ID} <urgent|high|medium|low>
```

用户不同意时直接改,无需解释。

### 2.3 复杂度预评估

满足以下任一项,启动 Claude Code 预评估(只分析不写代码):

- issue 描述超过 5 行 / 含多个子需求
- 关键词含"架构"、"重构"、"接口设计"、"数据库变更"

若预估超过 3 天工作量 → 自动拆分:

- 原 issue 转为 Epic
- 子 issue 按 `[子任务] 描述` 格式建立,关联父 issue
- 推送拆分结果确认(不回复视为认可)

规划结果存储:

- 当次方案 → 追加写入 Linear issue description
- 长期归档文档 → Obsidian 项目目录

**出口:** Label 已打、优先级已设、复杂 issue 已拆分

---

## Phase 3 · Sprint 规划(OpenClaw 主动执行)

### 3.1 两类任务,两套排期逻辑

**🤖 自动任务**`Auto` label,OpenClaw 可独立调度工具完成)

- Urgent / High → 立即移入 Todo,推送通知
- Medium → In Progress 数 < 2 时自动移入
- Low → 不主动排,等指示

**🧑‍💻 手动任务**(用户明确说"我自己来",或需要本地 IDE)

- OpenClaw 不主动移入 Todo,只推建议:
`📋 有 N 个手动任务待排期,优先级最高:{ISSUE_ID} [标题],要安排进本次 Sprint 吗?`
- 确认后再执行状态变更

### 3.2 并发控制

- In Progress 自动任务:不超过 **3 个**
- 自动 + 手动合计超过 5 个时推送提示:`⚠️ 当前进行中 N 个任务,建议先完成部分再开新任务`

**出口:** Issue 状态为 Todo,开发模式已明确(自动/手动)

---

## Phase 4 · 开发

**入口:** Issue 在 Todo

### 4.1 领取 Issue

说"开始做 {ISSUE_ID}" →

1. `python3 flowchain/projects.py issue start {ISSUE_ID}`
2. 判断开发模式

### 4.2 开发模式选择

**模式 A:AI 全程驱动**

1. 根据工具决策规则选择 Claude Code 或 Codex
2. 生成方案推送确认(实现前必确认,不自动写代码)
3. 确认 → 执行;不回复超 30min → 再次推送提醒

**模式 B:手动开发**(OpenClaw 退到监控角色)

1. 本地建分支:`git checkout -b feature/{ISSUE_ID}-描述`
2. Cursor 开发,OpenClaw 不干预
3. 遇卡点告知 → OpenClaw 按工具决策规则分析

### 4.3 AI 任务开发规范

**开发前(必做):**

1. `python3 flowchain/projects.py issue view {ISSUE_ID}` 确认 `## 验收标准` 存在
2. 若缺失 → 推断补写,推送确认
3. 推送:`🤖 {ISSUE_ID} AI 实现启动`

**开发后移交验收(必做):**

- 代码已 `git add && git commit`
- 推送:`✅ {ISSUE_ID} 代码完成,等待验收`

### 4.4 开发中监控

In Progress 超过 **3 天**无关联 PR → 推送:
`⏰ {ISSUE_ID} 进行中已 3 天,还在做吗?需要拆分或协助?`

**出口:** 代码已提交 → 进入 Phase 4.5 验收 Gate

---

## Phase 4.5 · 验收 Gate(守门员机制)

> ⚠️ **所有 AI 实现的自动任务必须通过此 Gate 才能更新 Linear Done 状态。**
> 手动完成的任务可跳过。

**入口:** AI 实现完成通知,或 heartbeat 兜底触发

### 验收流程

**Step 1:从 Linear 读取验收标准**

```bash
python3 flowchain/projects.py issue view {ISSUE_ID}
```

提取 `## 验收标准` 区块,逐项列出检查项。

**Step 2:代码质量检查**

```bash
# Python
python3 -m py_compile <新增文>
python3 -m pytest tests/ -v

# Swift/iOS
xcodebuild build -scheme <scheme> -destination 'platform=iOS Simulator,name=iPhone 16'
```

**Step 3:功能验证**
对照 `## 验收标准` 逐项确认,记录每项结果。

**Step 4:提交验收报告到 Linear**

```bash
python3 flowchain/projects.py issue report {ISSUE_ID} \
  --passed "标准 A:结果描述" \
  --failed "标准 B:未通过(原因)" \
  --conclusion "⚠️ 需修复"
```

**Step 5:分支处理**

✅ 验收通过:

```bash
python3 flowchain/projects.py issue move {ISSUE_ID} Done
```

推送:`✅ {ISSUE_ID} 验收通过`

⚠️ 验收失败 → 自动修复循环(最多 2 次):

1. 判断失败类型,选择修复工具
2. 派 Claude Code / Codex 修复,重新走 Step 2–4
3. 第 3 次仍失败 → 停止自动修复,推送人工介入:
  ```
   🔴 {ISSUE_ID} 验收连续失败 3 次,需人工介入
   失败项:[具体列表]
  ```

### 验收通过标准

1. 所有 `## 验收标准` 项均已核对
2. 代码质量检查通过
3. 核心功能验证通过(允许遗留 minor 问题,需在报告中注明)
4. 验收报告已写入 Linear issue comment

**出口:** Linear 状态 Done(通过),或人工介入(连续失败)

---

## Phase 5 · Code Review

**入口(仅传统 PR 流程):** Heartbeat 检测到新 PR(branch 名含 `{ISSUE_ID}`

> ⚠️ AI 自动任务不走此阶段,直接进入 Phase 4.5 验收 Gate。

### 5.1 状态同步

Linear API:**In Progress → In Review**

### 5.2 AI Review 自动触发

```bash
REVIEW_DIR=$(mktemp -d)
git clone https://github.com/your-org/{repo}.git $REVIEW_DIR
cd $REVIEW_DIR && gh pr checkout {pr_number}
claude --permission-mode bypassPermissions --print \
  'Review this PR. Focus on: logic bugs, security issues, performance problems. Be concise.'
```

Review 结论推送给用户,**不直接 comment GitHub**,等确认后决定是否贴出去。

### 5.3 PR 状态监控

- PR 超 **24h** 未 merge → 推送提醒
- CI 失败 → 立即推送:`🔴 {ISSUE_ID} PR #N CI 失败`

**出口:** PR merged

---

## Phase 6 · 验收完成

**入口 A(传统 PR 流程):** Heartbeat 检测到 PR merged
**入口 B(AI 自动任务):** Phase 4.5 验收 Gate 通过

### 6.1 状态同步

**PR merged 场景:**
Linear API:**In Review → Done**
推送:`✅ {ISSUE_ID} [标题] 已完成,PR #N merged`

**AI 自动任务场景:**
Linear 状态已在 Phase 4.5 更新,此处仅推送:`✅ {ISSUE_ID} 验收完成`

### 6.2 Canceled 验收检查

发起关闭 → 推送检查清单,等确认后再执行:

```
⚠️ {ISSUE_ID} 准备 Canceled,确认以下项目:
- [ ] 有无关联 open PR?(需先关闭或转移)
- [ ] 有无已提交但未 revert 的代码?
- [ ] 有无依赖此 issue 的其他任务?
确认没问题请回复"确认关闭"
```

### 6.3 归档提醒

若本次开发产出值得归档的文档 → 推送:`📝 {ISSUE_ID} 完成,是否写入文档库?`

**出口:** Issue 状态 Done / Canceled

---

## Phase 7 · 回顾

**触发:** 说"sprint 报告" / "项目进度" / "回顾一下"

```bash
python3 flowchain/projects.py sprint [项目名]
```

推送格式:

```
📊 本周进度([项目名])

✅ 已完成:N 个
· {ISSUE_ID} 标题

🔄 进行中:N 个
· {ISSUE_ID} 标题(In Progress · 已 X 天)

📋 待处理:N 个(Backlog)
· 优先级最高:{ISSUE_ID} 标题(High)

⚠️ 需要关注:
· {ISSUE_ID} 已超过 3 天未关联 PR,是否需要拆分?
```

---

## 异常处理


| 异常                    | 处理方式                                |
| --------------------- | ----------------------------------- |
| Linear API 失败         | SSL bypass 重试一次;仍失败告知用户             |
| Issue 找不到             | `linear issue list --all-states` 确认 |
| Branch 无 `{ISSUE_ID}` | 提醒确认命名,等提供 ID 后手动关联                 |
| gh CLI 失败             | 推送告警,Linear API 扫描继续                |
| gh CLI 连续 2 次失败       | 升级告警:项目监控降级(Linear only)            |


---

## Heartbeat 集成说明


| Workflow 机制              | Heartbeat 实现                                                         |
| ------------------------ | -------------------------------------------------------------------- |
| Phase 3 并发控制(Auto ≤3)    | Step A:统计 In Progress Auto issue 数,自动移入 Todo                         |
| Phase 4.4 超时告警(>3天无PR)   | Step A:linear list + gh pr 对比,无关联 PR 则推送                             |
| Phase 4.5 验收 Gate(兜底)    | Step A:检查 Linear comments 有无验收报告,无则触发,retries 存 heartbeat-state.json |
| Phase 5 PR Review 触发     | Step B:gh pr open 新 PR 检测                                            |
| Phase 6 PR merged → Done | Step B:gh pr merged 检测,反查 branch 的 {ISSUE_ID},更新 Linear              |


验收失败重试计数存储:`status/heartbeat-state.json → validation_retries.{ISSUE_ID}`,上限 3 次。超过上限后停止兜底,等用户人工介入后手动清除对应 key。

11. Heartbeat 项目监控 Section

将以下内容追加到 HEARTBEAT.md 基础模板之后:

## 1. 项目监控(读 status/PROJECTS.json,扫 heartbeat: true 的 repo)

**每次 heartbeat 按顺序执行 A → B → C:**

---

### Step A:Linear Auto 任务检查

**1. 查询所有 `In Progress` + `Auto` label 的 issue:**
```bash
linear issue list --state "In Progress" --label "Auto" --json
```

**2. 对每个 Auto issue 按序检查:**

**超时告警(无关联 PR):**
进入 In Progress 超 3 天且无关联 PR(branch 名含 `{ISSUE_ID}`)→ 推送:
`⏰ {ISSUE_ID} In Progress 已 N 天,未见 PR`

**验收兜底(AI 完成但 Gate 未触发):**
检查 Linear comments 里有无"验收报告"字样:

- 有 → 跳过(Phase 4.5 已完成)
- 无 → 读 `status/heartbeat-state.json``validation_retries.{ISSUE_ID}`
  - retries < 3 → 触发 Phase 4.5 验收 Gate,retries +1,写回 state
  - retries ≥ 3 → 跳过(已上报人工介入,等处理)

**3. 并发控制:** 统计所有 In Progress Auto issue 数量

- 数量 < 3 且 Backlog 有 `Auto + Urgent/High` issue → 按优先级取第 **1 个**移入 Todo,推送:
`📅 {ISSUE_ID} 自动移入 Todo({优先级})`

---

### Step B:GitHub PR 扫描 + 状态持久化

```bash
python3 flowchain/projects.py heartbeat
```

该命令自动完成:

-`status/PROJECTS.json` 加载 `heartbeat: true` 的项目
- 扫描每个 repo 的 open / merged PR,检测 CI 状态
- 将 merged PR 对应的 Linear issue 移入 Done
- 读写 `status/heartbeat-state.json`(去重、持久化)

根据输出 JSON 按以下规则推送:

- `ci_failures` 非空 → **立即推**(无视安静时段)
- `pr_open_stale` / `stale_in_progress` / `validation_retries` 非空 → 汇总推送(安静时段除外)
- `pr_merged_to_done` 非空 → 汇总推送(安静时段除外)
- 所有列表均为空 → **不推送**(静默完成)

**推送格式:**

```
🔔 项目更新

[BackClaw] PR #12「feat: 插件热更新」已等待 review 25h
[HexPaw] CI 失败:PR #5「fix: crash on launch」
{ISSUE_ID} In Progress 已 4 天,未见 PR
✅ {ISSUE_ID} PR #8 merged → Linear Done
```

---

### Step B2:邮件检查 → Copilot PR 通知 → 触发验收 Gate

`workflow/02-email-check.md` 执行。

核心链路:

1. 读取 `status/MAILLIST.json` 规则和状态
2. 检测 `*@github.com` 发件人 + Subject 含 `github-copilot[bot]` 的邮件
3. 从 Subject 提取 `{repo}` 和 PR 编号
4. `gh pr view` → 从分支名提取 Linear issue ID
5. Linear issue → In Review
6. 触发 Phase 4.5 验收 Gate
7. 更新 `MAILLIST.json``last_urgent_ids``last_run`

---

### Step C:NOW.md 覆写

每次 heartbeat 必做的收尾:用 `write` 工具(不是 edit/append)覆写 `NOW.md`

```markdown
# NOW.md — 当前状态快照

_Last updated: YYYY-MM-DD HH:MM (Asia/Shanghai)_

## 当前焦点
(一句话描述当前在做什么)

## 最近事件
- HH:MM — 事件标题(最多5条,从今日日志提取)

## 待处理
- [ ] 未完成的待办
```

12. 邮件监控 Workflow(02-email-check.md)

创建 workflow/02-email-check.md

---
title: "邮件监控"
created: YYYY-MM-DD
last_run: ~
last_verified: YYYY-MM-DD
ttl: 30d
status: active
skills: [gmail, telegram]
tags: [email, monitor, heartbeat, github, copilot]
---

# 邮件监控 Playbook

## 触发条件
- **定时触发**:heartbeat 检查时自动执行
- **手动触发**:说:检查邮件、查邮件、邮件汇总、有没有重要邮件

## 前置检查
- [ ] 读取 `status/MAILLIST.json`,获取运行状态与所有规则配置

## 执行步骤

### Step 1 — 加载配置与状态
读取 `status/MAILLIST.json`,获取:
- `config`:立即推送规则、汇总规则、忽略规则
- `last_summary_date`:判断今日汇总是否已发送
- `last_urgent_ids`:已推送的紧急邮件 ID 列表(去重用)

### Step 2 — 拉取未读邮件
```bash
gog gmail list "is:unread -category:promotions -category:social" --limit 20
```

### Step 3 — 规则匹配(优先级从高到低)

对每封邮件按以下顺序匹配,命中第一条即止:

1. **🤖 Copilot PR 通知**(最高优先级,触发验收流程):
  - 发件人为 `*@github.com` AND Subject 匹配:
    - `"[owner/repo] Pull request opened by github-copilot[bot]"`
    - `"[owner/repo] Pull request submitted by github-copilot[bot]"`
  -`msg_id` 不在 `last_urgent_ids` 中(去重)
  - **命中后执行 Step 3.1(Copilot PR 验收流程)**,不走普通推送
2. **🔴 立即推送**:发件人在白名单 OR Subject 含关键词,且 `msg_id` 不在 `last_urgent_ids`
3. **📋 每日汇总**:Label 符合汇总规则 AND 发件人不在黑名单
4. **🔇 忽略**:其余所有邮件

### Step 3.1 — Copilot PR 验收流程

#### 3.1.1 从邮件中提取 PR 信息

从 Subject 中提取 `{owner}/{repo}` 和 PR 编号:

```
Subject 示例:[your-org/ProjectA] Pull request opened by github-copilot[bot] (#3)
提取:repo = your-org/ProjectA,pr_number = 3
```

若提取失败 → 降级为普通立即推送,不触发验收。

#### 3.1.2 获取 PR 详情与关联 Issue

```bash
gh pr view {pr_number} --repo {owner}/{repo} --json title,body,headRefName
```

`headRefName`(分支名)提取 Linear issue ID(正则:`[A-Z]+-\d+`):

- 找到 → 记录 `{ISSUE_ID}`,继续
- 未找到 → 推送人工确认,将 `msg_id` 写入 `last_urgent_ids`,终止自动流程

#### 3.1.3 更新 Linear 状态

```bash
python3 flowchain/projects.py issue move {ISSUE_ID} "In Review"
```

#### 3.1.4 触发 Phase 4.5 验收 Gate

`workflow/01-project-management.md Phase 4.5` 流程执行。

推送通知:

```
🤖 Copilot PR #{pr_number} 已提交
仓库:{owner}/{repo}
关联:{ISSUE_ID}
状态:验收中…
```

验收完成后推送:

```
✅ {ISSUE_ID} Copilot PR #{pr_number} 验收通过 → Done
```

### Step 4 — 推送

**紧急邮件**(规则 2 命中):

```
🚨 重要邮件
发件人:xxx
主题:xxx
时间:xxx
```

推送后将 `msg_id` 追加到 `last_urgent_ids`,写回 `status/MAILLIST.json`

**每日汇总**`last_summary_date != 今日` 时推送):

```
📬 邮件摘要(N 封未读)

🔴 重要
- [发件人] 主题

📋 值得看
- [发件人] 主题
```

推送后更新 `last_summary_date` 为今日,写回 `status/MAILLIST.json`

### Step 5 — 更新状态

`last_run` 更新为当前时间,写回 `status/MAILLIST.json`

## 配置维护

所有规则配置在 `status/MAILLIST.json``config` 字段中维护:

- **新增白名单发件人**:追加到 `config.immediate.sender_whitelist`
- **新增关键词**:追加到 `config.immediate.subject_keywords`
- **屏蔽某个发件人**:追加到 `config.ignore.sender_blacklist`

13. Skills 配置

Google Workspace(gog)

OAuth 认证配置完成后,可以直接操作 Gmail、Calendar、Drive、Sheets、Docs。

安装:

openclaw skills install gog

常用命令:

# 搜索近7天未读邮件
gog gmail list "is:unread newer_than:7d" --limit 10 --json

# 查询本周日历事件
gog calendar events primary \
  --from YYYY-MM-DD \
  --to YYYY-MM-DD \
  --json

# 读取 Google Sheets 数据
gog sheets get <sheet-id> "Sheet1!A1:D20" --json

obsidian-cli

npm install -g obsidian-cli
obsidian-cli set-default /path/to/your/vault

常用命令:

# 查看当前默认 vault 路径
obsidian-cli print-default --path-only

# 搜索笔记标题
obsidian-cli search "关键词"

# 搜索笔记内容
obsidian-cli search-content "关键词"

# 移动笔记(同时更新所有 WikiLink)
obsidian-cli move "旧路径/笔记" "新路径/笔记"

14. macOS 安全配置

端口检查

sudo /usr/sbin/lsof -iTCP -sTCP:LISTEN -n -P
sudo /usr/sbin/lsof -iUDP -n -P

防火墙配置

# 检查防火墙状态
/usr/libexec/ApplicationFirewall/socketfilterfw --getglobalstate

# 检查隐身模式
/usr/libexec/ApplicationFirewall/socketfilterfw --getstealthmode

# 开启隐身模式
sudo /usr/libexec/ApplicationFirewall/socketfilterfw --setstealthmode on

# 查看防火墙应用白名单
/usr/libexec/ApplicationFirewall/socketfilterfw --listapps

OpenClaw 安全审计

openclaw security audit
openclaw security audit --deep

目标状态:0 critical。两个常见 WARN 项:

  • gateway.trusted_proxies_missing:不走反代直接本地访问时忽略
  • gateway.nodes.deny_commands_ineffectivedenyCommands 只做精确命令名匹配,检查配置的条目是否使用正确的命令 ID

SSH 管理

如果不需要从外部 SSH 进入这台 Mac:

# 停止 SSH 服务
sudo systemsetup -setremotelogin off

# 确认状态
sudo systemsetup -getremotelogin

快速检查清单

配置完成后,逐项验证:

基础配置
- [ ] SOUL.md / IDENTITY.md / USER.md / AGENTS.md 已创建
- [ ] memory/ 目录结构已建立,memlog.sh 可执行
- [ ] MEMORY.md 已创建(≤40行)
- [ ] HEARTBEAT.md 已配置(基础模板 + 项目监控 Section)
- [ ] Cron 任务已配置(23:45 反思、周日蒸馏、周一安全检查)

渠道与模型
- [ ] Telegram Bot 已创建,配对完成
- [ ] 流式输出已配置(streaming: "partial")
- [ ] 主模型 + fallback 链已配置
- [ ] 模型 alias 已设置

项目管理
- [ ] status/PROJECTS.json 已填写(至少一个项目,heartbeat: true)
- [ ] status/heartbeat-state.json 已初始化(空 JSON {})
- [ ] status/MAILLIST.json 已配置(config 规则已填写)
- [ ] flowchain/projects.py 可执行(python3 flowchain/projects.py heartbeat 有输出)
- [ ] credentials/linear.json 已配置(API key、Team ID、State IDs)
- [ ] WORKFLOW.md 已创建
- [ ] workflow/00-create-workflow.md 已创建
- [ ] workflow/01-project-management.md 已创建,占位符已替换
- [ ] workflow/02-email-check.md 已创建

验证
- [ ] 发一条 Telegram 消息,OpenClaw 能正常回复
- [ ] 手动触发 heartbeat,输出 HEARTBEAT_OK 或项目更新推送
- [ ] python3 flowchain/projects.py sprint 有正常 JSON 输出
- [ ] openclaw security audit 输出 0 critical

评论