release: v1.0.4 支持消息附加说明与分行表达

docs: 精简 README 并移除版本升级记录
release: v1.0.3 统一个人 DWS 通知与简化初始化配置
2026-09-15 17:33:19 +08:00 · 2026-09-11 10:38:22 +08:00 · 2026-09-10 17:14:23 +08:00
20 changed files with 700 additions and 711 deletions
@@ -1,105 +1,66 @@
-# Hub 钉钉通知接入
+# 个人 DWS 项目通知
-这套封装只帮助 Hub 向项目群发送进度、异常摘要和负责人提醒，不是 Etunel 任务通道，也不负责审批、阶段门禁或项目留档。仅更新此仓库不会部署到实际项目；由部署负责人明确安装，成员 Agent 不调用。
+Hub 使用负责人确认的个人 DWS 账号发阶段汇报和 @ 提醒。只需要原生 DWS、Python 3.9+ 和一个项目配置文件；不使用机器人、应用密钥、Webhook 或 Stream 监听。
-## 文件与依赖
+**旧版升级：** 用新的个人账号配置替换旧配置，旧机器人配置不会自动转成个人发送授权。摘要入口已改名为 `dingtalk-message`／`dingtalk-message.ps1`，删除目标部署中的旧 `dingtalk-bot-message` 入口及 Stream 实验脚本。更新脚本和 Hub Skill／Hook 不等于已经迁移在途项目；按实际部署安排确认。
- [dingtalk_notify.py](../scripts/dingtalk_notify.py)：共用 Python 实现，含配置、消息、人员解析和有限重试。
+## 首次由 Hub 完成配置
 - [dingtalk-progress](../scripts/dingtalk-progress)／[dingtalk-progress.ps1](../scripts/dingtalk-progress.ps1)：项目事件入口。
 - [dingtalk-bot-message](../scripts/dingtalk-bot-message)／[dingtalk-bot-message.ps1](../scripts/dingtalk-bot-message.ps1)：自定义 Markdown 与负责人提醒入口。
 - [离线测试](../scripts/test_dingtalk_notify.py)：假配置、假凭据及模拟命令，不发送真实消息。
-依赖 Python 3.9+，不需要第三方 Python 包、jq 或 Bash 才能在 Windows 运行。PowerShell 入口要求 7.3+；也可直接运行 Python。macOS/Linux 保留 Shell 入口。
+负责人不需要找编号、复制配置模板或填写密钥。Hub 在项目初始化时执行：
-真实发送需要 [官方 DingTalk Workspace CLI](https://github.com/DingTalk-Real-AI/dingtalk-workspace-cli) 的原生可执行文件。按官方安装说明选择系统版本；Windows 使用 `dws.exe`，不要把 npm/Bun 的 `dws.cmd` 或 `dws.bat` 当原生程序。可以把原生文件加入 Hub 进程的 PATH，或在项目配置的 `DINGTALK_PROGRESS_DWS_PATH` 填它的绝对路径。本仓库不保存二进制，不自动安装、登录或更新全局 Skills。
+1. 检查本机 DWS；未登录时请负责人通过官方 DWS 登录流程授权，不索取密码或 Token。
-
+2. 用 `dws profile list --format json` 查看可用账号，以姓名和组织向负责人确认发送身份。
-部署后检查 `python --version`（macOS/Linux 可用 `python3`）及原生 `dws version`。所用 DWS 需支持 `api --data -` 从标准输入读 JSON；正文、群参数不放命令行，密钥仅临时注入子进程环境。更换 DWS 版本后由部署负责人核对兼容性，离线预览不证明真实接口可用。
+3. 在选定的同一账号下，用 `dws chat +chat-search --query "负责人提供的群名称" --page-all --profile "查询返回的账号标识" --format json` 查群；结果不完整或有多个候选时不能自行选第一个。
-
+4. 一次问清：“是否由这个账号向这个项目群自动发送进度、阻塞和必要的 @ 提醒？”得到确认后，Hub 把工具实际返回的账号标识、群标识及项目名称传给初始化入口。下面为参数结构，不是需要人填写的编号表：
 ## 部署到 Hub 项目
 将共用 Python 文件和需要的平台入口一起放入目标项目的 `scripts/`，不要只复制一个入口。重试逻辑已合入共用实现，无需复制源项目的 `dingtalk-dns-retry.sh`。保持下列布局：
   ```text
-目标项目/
+   python scripts/dingtalk_notify.py init --profile "实际查询结果" --group "实际查询结果" --project "项目名称" --yes
 ├── scripts/
 │   ├── dingtalk_notify.py
 │   ├── dingtalk-progress
 │   ├── dingtalk-bot-message
 │   ├── dingtalk-progress.ps1
 │   └── dingtalk-bot-message.ps1
 └── .dingtalk/
    └── config.env
   ```
-由部署负责人从 [配置示例](config.env.example) 新建实际配置，别覆盖已有有效配置：
+初始化检查账号存在且可登录，创建 `.dingtalk/config.env` 并启用通知；**不会发测试消息，也不会替人确认群选择**。群标识必须来自该账号的查询结果。初始化不创建钉钉群、不申请权限。已有配置默认不覆盖；迁移或变更账号、群时重新确认，再加 `--replace`，原配置被替换且不另存含密钥的备份。
- UTF-8，可带 BOM，支持 Windows CRLF 和 LF；值按字面解析，不执行 Shell、不展开变量，不加 Shell 引号。
+配置仅 4 项：
 - 初始保持 `ENABLED=0`、`DRY_RUN=1`（实际字段均有 `DINGTALK_PROGRESS_` 前缀）。
 - 配置包括应用、机器人、目标群、项目名称、开关及机器人密钥。机器人必须属于配置的应用，已经加入目标群，并具备企业内机器人发送消息权限（`qyapi_robot_sendmsg`）；脚本不自动加群或申请权限。
 - macOS/Linux 对实际配置执行 `chmod 600 .dingtalk/config.env`；接受 400 或 600。Windows 不以 Unix 权限位作门槛，由部署负责人用 NTFS 权限限制其他用户读取或修改配置。
 - 实际配置必须被目标项目的 Git 忽略。不要把它、真实凭据或内部标识回传给 AI 会话或提交仓库。
-路径与当前工作目录无关：配置总从 Python 脚本所在项目根目录读取。消息的 `--file` 路径则相对当前工作目录解析。
+| 字段 | 含义 | 谁填写 |
 | --- | --- | --- |
 | `DINGTALK_PROGRESS_ENABLED` | `1` 启用、`0` 暂停 | 初始化写入，负责人可随时要求暂停 |
 | `DINGTALK_PROGRESS_PROFILE` | 固定个人账号 | Hub 使用 DWS 真实返回值 |
 | `DINGTALK_PROGRESS_TARGET` | 固定项目群 | Hub 使用同账号查询并经确认的结果 |
 | `DINGTALK_PROGRESS_PROJECT` | 项目名称 | Hub 使用已确认名称 |
-## 凭据：所有系统统一读取配置
+配置不包含凭据，但仍是本机信息，本仓库已排除出 Git；Hub 在其他目标项目初始化时也应把 `.dingtalk/config.env` 加入该项目的 Git 忽略规则。不要把它或登录态复制给其他同事；各部署分别确认账号和群。登录由 DWS 自己管理，后续切换 DWS 默认账号不会改变本项目的发送身份。
-普通进度发送使用内部应用身份，不需要用户 OAuth 登录。Windows、macOS、Linux **只从本地 `.dingtalk/config.env` 的 `DINGTALK_PROGRESS_CLIENT_SECRET` 读取机器人密钥**，不再从 `DING_SEC`、其他环境变量或 macOS 钥匙串回退取值。
+## Windows 与 macOS/Linux
-由部署负责人在本地编辑器中填写该字段，保持单行字面值，不加 Shell 引号；不要在会话、命令历史或仓库示例里填写真实值。密钥为空或字段缺失时，真实发送停止并提示负责人补配置；离线预览不要求填写密钥。
+- Windows 使用 PowerShell 7.3+，优先运行 `./scripts/dingtalk-python.ps1 init ...`、`check ...`；事件用 `./scripts/dingtalk-progress.ps1 ...`，提醒用 `./scripts/dingtalk-message.ps1 ...`。
 - PowerShell 共用启动器选择独立 Python，避免商店版 Python 无法读取终端的 DWS 登录。必要时用 `DINGTALK_PYTHON` 指定独立解释器路径；原生 `dws.exe` 须在当前进程 PATH 中。
 - macOS/Linux 通常用 `python3 scripts/dingtalk_notify.py ...`；也可用 `sh scripts/dingtalk-progress ...`、`sh scripts/dingtalk-message ...`。初始化生成的配置在 POSIX 系统为仅当前用户读写；Windows 使用部署目录的访问权限。
 - 将共用 Python 文件和所需平台入口放在目标 Hub 项目的 `scripts/` 中。安装职责 Skill 不会自动部署这些脚本。
-脚本仅在发送时把该值临时注入 DWS 子进程的 `DWS_CLIENT_SECRET`，同时注入配置中的 `DWS_CLIENT_ID`。这只是脚本内部对接 DWS 的方式，不需要负责人再配置环境变量。密钥不进入命令参数、消息正文、预览或错误输出；人员搜索进程也不会取得机器人密钥。
+DWS 安装与登录以 [官方项目](https://github.com/DingTalk-Real-AI/dingtalk-workspace-cli) 和当前命令帮助为准；不要为通知分发机器人应用密钥。
-实际 `config.env` 现在包含明文密钥：macOS/Linux 设为 400 或 600；Windows 通过文件“属性 → 安全”限制为 Hub 运行账号及必要管理员可访问，不放在公共共享目录。目标项目必须忽略实际配置及其备份，只提交空值示例。不要把配置文件用作 `--file` 正文、成果或通知附件。
+## 日常调用
-### 从旧配置迁移
+```text
-
+python scripts/dingtalk_notify.py progress milestone "**已完成**：阶段成果已归档。下一步安排联调。"
-1. 更新共用 Python 脚本及需要的平台入口。
+python scripts/dingtalk_notify.py progress blocked "缺少接口说明，请补充后回自己的角色会话同步。" --at "完整姓名"
-2. 由部署负责人在本地配置补填 `DINGTALK_PROGRESS_CLIENT_SECRET`；删除旧的 `DINGTALK_PROGRESS_KEYCHAIN_SERVICE` 字段。
+python scripts/dingtalk_notify.py message --title "资料补充提醒" --text "请补齐已约定资料，并回自己的角色会话确认。" --at "完整姓名"
 3. 检查文件权限与 Git 忽略，再做离线预览；正式发送仍需要启用开关、正确的应用权限和目标群。
 4. 旧环境变量与钥匙串不再被脚本读取；本次改动不会删除其中原有内容。
 脚本每次调用重新读取本地配置，不再因为密钥配置依赖 Hub 继承某个终端的环境变量；Python、dws 路径及按姓名查询所需的 DWS 登录仍须对实际运行 Hub 的操作系统账号可用。
 ## 可选的按姓名提醒
 `--at` 会调用 DWS 人员搜索，和应用机器人发送是不同的认证需求。只有使用按姓名提醒时，部署负责人需要：
 1. 在 DWS 中登录需要使用的组织账号；
 2. 将已确认的稳定 profile 填入 `DINGTALK_PROGRESS_PROFILE`；
 3. 确认该账号、机器人应用、目标群属于同一组织，且账号允许查询人员。
 脚本在搜索和发送中传入同一个 profile，不自动选择或切换账号；应用归属与群归属仍需部署负责人核对，脚本不能仅凭 profile 字符串证明二者一致。查询只接受完整姓名的唯一匹配；查不到、重名、权限不足或返回结构不支持时整条提醒停止，不默认选择第一个，也不静默删掉 @ 后发送。
 `--at-id` 和 `--group` 保留源脚本兼容能力，仅用于已有真实映射或用户明确指定的目标；不让 AI 猜编号或在群之间试发。普通进度通知不需要配置人员账号。
 ## 先做离线预览
 在目标项目根目录选一种入口。以下命令只预览，不联网，不需要 dws，不查通讯录、不使用密钥认证。已有配置会在本地读取，但密钥不会输出：
 ```powershell
 python scripts/dingtalk_notify.py progress milestone '本阶段成果已确认并归档，下一步安排联调。' --dry-run
 ./scripts/dingtalk-bot-message.ps1 --title '需要补充接口说明' --text '请接口负责人补齐返回值说明，并回自己的角色会话同步结果。' --dry-run
 ```
-macOS/Linux 等价命令：
+通知保留 AI 标识并标注自动汇报；摘要用简短中文 Markdown。`--at` 按完整姓名查询唯一人员，再生成个人消息的原生提及标记。两个发送命令都支持 `--file` 正文和完全离线的 `--dry-run`。不支持临时换群、手填人员编号或退回机器人发送。
 事件含义见 [通知规则](../etunel-role-collaboration/references/dingtalk-progress-reporting.md)，详细调用与结果见 [脚本用法](../etunel-role-collaboration/references/dingtalk-script-usage.md)。
 ## 检查与失败处理
 - `python scripts/dingtalk_notify.py check --at "完整姓名"`：只检查固定账号登录和可选姓名，不发消息，不证明实际 @ 提醒成功。不是每条通知的必做步骤。
 - `accepted` 加有效 `openTaskId`：个人发送请求已接受，不等于已投递或已读。必要时在同账号下用 `dws chat +messages-query-send-status --open-task-id "回执中的实际值" --profile "配置中的账号" --format json` 只读查询；不把任务回执当消息编号，不循环重发。
 - 仅明确 DNS 解析失败最多尝试 3 次、间隔 10 秒；认证、权限或未知结果停止。登录失效请负责人恢复登录，不切账号、不退回机器人。通知失败不阻塞 Etunel 主任务。
 维护时只做离线测试；真实发送必须另获明确授权：
 ```sh
 sh scripts/dingtalk-progress milestone '本阶段成果已确认并归档，下一步安排联调。' --dry-run
 sh scripts/dingtalk-bot-message --title '需要补充接口说明' --text '请接口负责人补齐返回值说明，并回自己的角色会话同步结果。' --dry-run
 ```
 两种入口都遵守配置的 `DRY_RUN=1`，显式 `--dry-run` 不能被配置取消。即使没有配置，也能显式预览正文；此时不表示项目配置可用。正文支持中文多行 Markdown；复杂正文建议先保存 UTF-8 文件，用 `--file` 代替内联摘要或 `--text`，它不是附件上传。
 确认部署与发送授权后，负责人再将 `ENABLED=1` 且 `DRY_RUN=0`。仓库维护测试不能用真实发送作冒烟测试。实际 Hub 的事件判断、参数示例和结果含义见 [通知规则](../etunel-role-collaboration/references/dingtalk-progress-reporting.md) 和 [脚本用法](../etunel-role-collaboration/references/dingtalk-script-usage.md)。
 ## 失败处理与测试
 仅明确的 DNS 解析失败每隔 10 秒重试，总计最多 3 次；已获有效回执即停止。认证、权限、未知错误不重试；超时或缺少回执不盲目重发。单次 DWS 调用超时 60 秒；本次结束后由 Hub 说明，不在外层循环重试。只输出安全摘要和发送回执，不输出原始请求或错误日志。
 本封装不建消息队列、计划任务、重发历史或通知状态库；避免重复事件仍由 Hub 利用已有项目留档判断，不重造 Etunel 软件机制。
 ```text
 python -X utf8 scripts/test_dingtalk_notify.py
 bash -n scripts/dingtalk-progress
 bash -n scripts/dingtalk-message
 ```
 测试仅在临时目录使用假配置、假密钥和模拟 DWS；同时检查当前系统可用的 PowerShell/Shell 入口，不访问真实项目配置。覆盖配置密钥为唯一来源、密钥不回显、不通过命令参数传递以及预览不认证，不等于完成真实钉钉发送验收。
@@ -1,25 +1,7 @@
-# 由部署负责人复制为 .dingtalk/config.env；macOS/Linux 设置 chmod 600。
+# Hub 查询账号和群、取得负责人确认后，用 init 生成真实 config.env。
-# Windows 用当前 Hub 运行用户的文件权限，不套用 Unix 权限检查。
+# 不需手工复制或填编号；不保存机器人密钥或个人登录凭据。
-# UTF-8（可带 BOM），支持 LF/CRLF；等号右侧是字面值，不加 Shell 引号，不执行变量。
+# 等号右侧为字面值，不加 Shell 引号。ENABLED=0 暂停通知。
 # 两个入口共用开关；显式 --dry-run 也永远不发送。
 DINGTALK_PROGRESS_ENABLED=0
 DINGTALK_PROGRESS_DRY_RUN=1
 # 企业内部应用机器人：发送本身不需要用户 OAuth 登录。
 DINGTALK_PROGRESS_MODE=app-bot
 DINGTALK_PROGRESS_CLIENT_ID=
 # 仅在本地 config.env 填真实 Client Secret；本示例必须保持空值，绝不提交真实配置。
 DINGTALK_PROGRESS_CLIENT_SECRET=
 DINGTALK_PROGRESS_TARGET=
 DINGTALK_PROGRESS_ROBOT_CODE=
 # 所有系统统一从 config.env 取密钥，不再读取 DING_SEC 或 macOS 钥匙串。
 # 可选：原生 dws 的绝对路径（Windows 填 dws.exe，不填 cmd/bat 包装器）。
 DINGTALK_PROGRESS_DWS_PATH=
 # 可选：按姓名 @ 时必需。由部署负责人确认与应用、目标群同组织的已登录 profile。
 # 普通进度通知可留空；不是 Etunel 人员或会话编号。
 DINGTALK_PROGRESS_PROFILE=
-
+DINGTALK_PROGRESS_TARGET=
-DINGTALK_PROGRESS_PROJECT=你的项目名称
+DINGTALK_PROGRESS_PROJECT=
 DINGTALK_PROGRESS_TITLE_PREFIX=[项目进度]
@@ -13,4 +13,4 @@
 *.sh text eol=lf
 *.ps1 text eol=lf
 scripts/dingtalk-progress text eol=lf
-scripts/dingtalk-bot-message text eol=lf
+scripts/dingtalk-message text eol=lf
@@ -61,7 +61,7 @@
 - 只有 Hub／主 Agent 可以调用通知；成员 Agent 只向 Hub 返回成果、状态或阻塞。
 - 通知是项目群的单向辅助可见性与异常提醒，不是角色间通信、任务派发、审批、人类确认、项目门禁或项目事实源。
- Agent 只用 `scripts/dingtalk_notify.py` 的 `progress`／`message`，或等价的 `dingtalk-progress`／`dingtalk-bot-message` Shell、PowerShell 入口。进度事件仍限定为 `start`、`milestone`、`blocked`、`complete`、`failed`；负责人提醒不新增状态。同一事项不重复通知，不得绕过封装直接调用 OpenAPI 或 `dws api`。可验证里程碑先完成项目留档，再发送简短中文 Markdown。例如：
+- Agent 只用 `scripts/dingtalk_notify.py` 的 `progress`／`message`，或等价的 `dingtalk-progress`／`dingtalk-message` Shell、PowerShell 入口，统一以已确认的个人 DWS 账号发送。进度事件仍限定为 `start`、`milestone`、`blocked`、`complete`、`failed`；负责人提醒不新增状态。同一事项不重复通知，不得绕过封装直接调用 OpenAPI 或 `dws api`。可验证里程碑先完成项目留档，再发送简短中文 Markdown。例如：
  ```sh
  python scripts/dingtalk_notify.py progress milestone "**已完成**
@@ -75,7 +75,8 @@
 - 事件语义与消息格式以 [dingtalk-progress-reporting.md](etunel-role-collaboration/references/dingtalk-progress-reporting.md) 为准；安装与本机配置参考 [.dingtalk/README.md](.dingtalk/README.md)。
 - 通知封装是辅助接入参考，不得据此反推或改变 Etunel 主流程；不创建通知队列或外层重试。仅明确 DNS 失败最多 3 次、间隔 10 秒，结果未知不盲目重发。
 - 仓库维护和测试期间不得把真实发送当作冒烟测试。只有任务明确要求发送且当前环境已经授权时，才允许触发真实通知。
- 不读取、输出、修改或提交 `.dingtalk/config.env`、Client Secret、`DING_SEC`、App Token 或其他凭据和内部标识。
+- Hub 首次查询个人账号及项目群，以姓名、组织和群名取得负责人确认后，用封装 `init` 创建 `.dingtalk/config.env`；只含通知开关、固定账号、群和项目名，不保存机器人或登录密钥。AI 可以在授权范围内维护该配置，但不提交或在群里展示内部标识；换账号或群须重新确认。
 - 不读取、输出或提交 DWS 登录凭据、Client Secret、`DING_SEC`、App Token。旧机器人配置须经确认重新初始化，不静默启用个人代发；保留 AI 标识，个人通知不代表负责人亲自批准。
 ## 编辑约定
@@ -104,7 +105,7 @@
 rtk python -X utf8 "$env:USERPROFILE\.codex\skills\.system\skill-creator\scripts\quick_validate.py" "etunel-role-collaboration"
 rtk python -X utf8 scripts/test_dingtalk_notify.py
 rtk "C:\Program Files\Git\bin\bash.exe" -n scripts/dingtalk-progress
-rtk "C:\Program Files\Git\bin\bash.exe" -n scripts/dingtalk-bot-message
+rtk "C:\Program Files\Git\bin\bash.exe" -n scripts/dingtalk-message
 rtk proxy git diff --check
 rtk proxy git status --short
 ```
@@ -14,7 +14,7 @@
 ## 版本与升级
-本版为 `v1.0.2`，流程设计基线仍为 **V1.0**。版本采用 `vX.Y.Z`：前两位 `X.Y` 跟随 HTML 主流程中明确标注且已确认的版本，第三位 `Z` 表示该流程版本下的仓库发布迭代。正式发布以 Git 标签和对应发布包为准，主分支可能包含尚未发布的修改。
+本版为 `v1.0.4`，流程设计基线仍为 **V1.0**。版本采用 `vX.Y.Z`：前两位 `X.Y` 跟随 HTML 主流程中明确标注且已确认的版本，第三位 `Z` 表示该流程版本下的仓库发布迭代。正式发布以 Git 标签和对应发布包为准，主分支可能包含尚未发布的修改。
 | 变更 | 版本规则 |
 | --- | --- |
@@ -26,20 +26,7 @@
 发布说明应写清：流程基线、改了什么、使用者需要更新哪些文件、在途项目是否需要重新确认。不要把“仓库已发布”写成“角色已更新”。更新插件缓存、Etunel 中独立配置的 Hook 或迁移在途项目，都需另外明确确认；已运行项目继续使用其登记基线。
-### v1.0.2 更新范围
+各版本的变更、验证结果和升级注意事项统一查看[发布说明](https://git.playones.com/chenmingxuan/EP-Hub-Skill/releases)。README 只维护当前项目介绍和使用入口，不累积逐版更新记录。
 - 明确前两位跟随 HTML 流程、其余发布迭代只升第三位的版本约定；主流程仍为 V1.0，八角色及阶段门禁不变。
 - 钉钉通知改为共用 Python 实现，保留 macOS/Linux 入口，新增 Windows PowerShell 入口；支持项目事件、自定义中文 Markdown、文件正文和按姓名提醒负责人。
 - Windows、macOS、Linux 统一从本地 `config.env` 的 `DINGTALK_PROGRESS_CLIENT_SECRET` 取机器人密钥，不再读取环境变量或钥匙串。**升级时须由部署负责人补填本地密钥字段，并删除旧 `DINGTALK_PROGRESS_KEYCHAIN_SERVICE` 字段**；不提交真实配置。
 - 两个入口统一遵守离线预览；仅明确 DNS 失败每隔 10 秒重试、总计最多 3 次，结果未知不盲目重发。补充 27 项离线测试与部署说明。
 - 更新核心 Skill 的通知路由与细则，以及中心 Hook 的完整职责和每轮提醒；其他角色 Hook 无需因本次通知更新而替换。目标项目需同步共用 Python 文件及所需平台入口，安装 Python 3.9+ 与原生 DWS；PowerShell 入口需 7.3+。
 - Windows 普通群通知已获服务端回执并经负责人确认显示成功；按姓名 @ 和原生 macOS 发送仍未实测。通知升级不自动迁移在途项目或更新 Etunel Hook／插件缓存，需按部署安排明确执行。
 ### v1.0.1 更新范围
 - 根据一轮项目复盘，补清旧任务收尾、开工输入与本次产出的区分、资料异常交专业角色核实、查询无新线索时转交负责人决定，以及模拟不默认跳过交接。
 - 更新核心 Skill；Hook 仅调整中心与产品两份，保持完整职责 JSON 和每轮提醒分开配置。产品 Hook 修改的是完整职责区块，中心两个区块均有调整。
 - 八角色、八阶段、正式成果、原有审批与测试验收条件不变；不自动迁移在途项目、不关闭真实遗留问题。本版不涉及钉钉脚本更新。
 ## 协作模型
@@ -121,9 +108,10 @@ EP-Hub-Skill/
 └── scripts/                          # Hub 辅助通知，不是 Etunel 任务机制
    ├── dingtalk_notify.py            # Python 共用实现，含有限重试
    ├── dingtalk-progress             # macOS/Linux 进度入口
-    ├── dingtalk-bot-message          # macOS/Linux 摘要与负责人提醒
+    ├── dingtalk-message              # macOS/Linux 个人摘要与负责人提醒
    ├── dingtalk-progress.ps1         # Windows PowerShell 进度入口
-    ├── dingtalk-bot-message.ps1      # Windows PowerShell 摘要与提醒
+    ├── dingtalk-message.ps1          # Windows PowerShell 个人摘要与提醒
    ├── dingtalk-python.ps1           # Windows 共用启动器，含初始化与检查
    └── test_dingtalk_notify.py       # 不联网的隔离测试
 ```
@@ -183,13 +171,13 @@ python scripts/dingtalk_notify.py progress <start|milestone|blocked|complete|fai
 五类事件分别表示项目正式开始、可验证里程碑、真实阻塞、整体完成和最终失败。可验证里程碑由 Hub 先完成成果和进度留档，再发送钉钉消息。成员角色不得发送钉钉消息，只把结果交给 Hub。通知正文可使用多行 Markdown，只写可公开的已验证结论、下一步或阻塞原因，不发送密钥、个人数据、大段日志和未经证实的推断。
-新增 `message --title "中文标题" --text "问题和下一步"` 用于辅助摘要与负责人提醒，可按需用 `--at`；提醒结果回各自角色会话，不在群里另派任务或批准。两个入口均支持文件正文与离线 `--dry-run`，Windows 可用 `.ps1`，macOS/Linux 保留原 Shell 入口。它们是同一封装，不要重复调用。
+`message --title "中文标题" --text "问题和下一步"` 用于辅助摘要与负责人提醒，可按需用 `--at`；提醒结果回各自角色会话，不在群里另派任务或批准。两个入口均支持文件正文与离线 `--dry-run`，Windows 可用 `.ps1`，macOS/Linux 可用 Shell 入口。它们是同一封装，不要重复调用。
-各系统统一由封装读取本地 `config.env` 的 `DINGTALK_PROGRESS_CLIENT_SECRET`，不再从环境变量或 macOS 钥匙串取密钥；实际配置含明文密钥，须限制文件访问并排除出 Git。普通通知不需要 DWS 用户登录，按姓名提醒才额外需要同组织的已登录账号。
+所有通知统一使用负责人确认的个人 DWS 登录，不再使用机器人。Hub 在首次初始化时查询账号和项目群，用姓名、组织和群名取得确认后，调用 `init` 自动生成配置；人不用找编号或填密钥。配置只保留通知开关、固定账号、固定群和项目名，个人登录凭据由 DWS 管理。`progress` 也支持 `--at`，事件汇报和行动提醒可合成一条；保留 AI 标识和自动汇报说明，不冒充本人批准。
-事件语义、有限重试和结果判定见 [钉钉项目进度汇报](etunel-role-collaboration/references/dingtalk-progress-reporting.md)，命令细节见 [通知脚本用法](etunel-role-collaboration/references/dingtalk-script-usage.md)；跨平台安装、原生 dws 与凭据配置见 [.dingtalk/README.md](.dingtalk/README.md)。`.agents/skills/dingtalk-*` 是钉钉官方多 Skill 的项目内参考副本，不应被当作 Etunel 核心 Skill，也不应被成员用来绕过 Hub。
+事件语义、有限重试和结果判定见 [钉钉项目进度汇报](etunel-role-collaboration/references/dingtalk-progress-reporting.md)，命令细节见 [通知脚本用法](etunel-role-collaboration/references/dingtalk-script-usage.md)；跨平台安装、初始化和旧配置迁移见 [.dingtalk/README.md](.dingtalk/README.md)。`.agents/skills/dingtalk-*` 是钉钉官方多 Skill 的项目内参考副本，不应被当作 Etunel 核心 Skill，也不应被成员用来绕过 Hub。
-本地 `.dingtalk/config.env` 已被 Git 忽略。不要读取、输出或提交其中的配置；不要在文档、Skill、Hook 或脚本中硬编码 Client Secret、Client ID、`robotCode`、`openConversationId`、App Token 或群名。仓库维护时也不要用真实发送作为普通测试。
+本地 `.dingtalk/config.env` 已被 Git 忽略，Hub 可在负责人授权范围内创建或修改；不要提交配置、硬编码真实账号／群标识，或读取和传播 DWS 凭据。`accepted` 表示取得个人发送任务回执，不等于已投递、已读或接收端 @ 提醒成功。仓库维护时不要用真实发送作为普通测试。
 ## 修改项目的推荐流程
@@ -215,7 +203,7 @@ git status --short
 ```sh
 python -X utf8 scripts/test_dingtalk_notify.py
 bash -n scripts/dingtalk-progress
-bash -n scripts/dingtalk-bot-message
+bash -n scripts/dingtalk-message
 ```
 还应人工确认：
@@ -53,13 +53,13 @@ Etunel 负责当前角色和会话绑定。不要向负责人询问或复述 rol
  - [嵌入式底层](references/roles/embedded-lowlevel.md)
  - [硬件](references/roles/hardware.md)
  - [测试](references/roles/testing.md)
- 准备派发、回复、跨角色中继、完成或报告阻塞：读 [Etunel 任务消息流程](references/etunel-message-lifecycle.md)。
+- 准备派发、回复、跨角色中继、完成、报告阻塞或使用消息附加说明：读 [Etunel 任务消息流程](references/etunel-message-lifecycle.md)，按需用可选 `context` 概括事项，完整要求仍留在正文。
 - 定义或校验成果、状态、版本、证据、额外产出或豁免：读 [成果与完成判定](references/artifacts-and-evidence.md)。
 - 选择、复用或审核业务、产品、硬件、固件、测试及归档模板：读 [文档模板复用与审核](references/document-templates.md)，再按实际任务定位原件；模板不自行改变职责、正式成果或门禁。
 - 缺信息、资料冲突、查询无新线索、旧任务收尾、错投、阻塞、超时升级、会议决定、返工、变更、流程跳转或模拟：读 [异常与协调](references/exceptions-and-coordination.md)。
 - 编写给负责人、角色或群组的任务、提问、结果和状态消息：读 [人类可读沟通](references/human-readable-communication.md)。
 - 创建项目资料目录、保存成员清单、归档成果、更新进度或阶段记录：读 [项目文件与进度留档](references/project-files-and-progress.md)。
- 仅当当前会话是 Hub 且需钉钉进度汇报或负责人辅助提醒：读 [钉钉项目进度汇报](references/dingtalk-progress-reporting.md)，实际调用再读其中的脚本用法。
+- 仅当当前会话是 Hub 且需初始化个人 DWS 通知、钉钉进度汇报或负责人辅助提醒：读 [钉钉项目进度汇报](references/dingtalk-progress-reporting.md)，实际调用再读其中的脚本用法。
 - 测试角色仅在任务类型匹配时，按 [测试职责](references/roles/testing.md) 中的二级链接读取更深细则。
 不要为“全面了解”一次加载全部引用或全部角色文件。Etunel Hook 源文件独立配置，不属于本 Skill 的渐进式发现树。
@@ -5,12 +5,12 @@
 ## 权限与封装入口
 - 只有 Hub／主 Agent 可以发送；成员 Agent 只把成果、状态或阻塞交给 Hub。
- 项目配置为启用时视为持续发送授权，不需每条消息再次询问。
+- 项目首次初始化时，由 Hub 查询可用个人 DWS 账号和群，以姓名、组织和群名向负责人确认自动通知范围，再调用 `init` 创建配置；不让人填写编号或密钥。确认并启用后，范围内通知不需每条再次询问，变更账号或群须重新确认。
- 只用项目封装：`dingtalk-progress` 处理五类项目事件，`dingtalk-bot-message` 处理自定义中文摘要或负责人提醒；二者共用 `scripts/dingtalk_notify.py`。Windows、macOS/Linux 的命令及文件正文、人员提醒用法见 [通知脚本用法](dingtalk-script-usage.md)，实际调用前按需读取。
+- 只用项目封装：`dingtalk-progress` 处理五类项目事件，`dingtalk-message` 处理自定义中文摘要或负责人提醒；二者共用 `scripts/dingtalk_notify.py`，统一使用确认的个人 DWS 账号。初始化、跨平台入口及 @ 用法见 [通知脚本用法](dingtalk-script-usage.md)。
 - Agent 不得直接调用底层 OpenAPI、`dws api` 或其他消息入口绕过封装脚本。
- Agent 不自行打开、展示或修改真实配置，不请求负责人在会话中提供密钥和内部标识；由封装在本地读取 `config.env` 中的 `DINGTALK_PROGRESS_CLIENT_SECRET` 并完成认证，不再使用环境变量或钥匙串取密钥。
+- 配置只保存开关、固定账号、项目群和项目名称；Hub 可在已确认范围内创建或更新，登录凭据由 DWS 管理，不索取、读取或输出 Token、密码与应用密钥。旧机器人配置须重新初始化，不静默切换发送身份。
- Skill 与消息正文不得硬编码项目 Client ID、robotCode、openConversationId、Client Secret 或群名；项目差异只由 `.dingtalk/config.env` 和封装脚本处理。
+- 不在 Skill 或群消息里硬编码账号、群或人员标识；标识从同一账号的真实工具结果取得，仅用于本地配置和工具参数。不因默认账号切换而更换发送人，不退回机器人或 Webhook。
 ## 汇报单位与顺序
@@ -58,6 +58,8 @@ Hub 必须先确认进展真实有效，并完成成果文件、成果索引和
 使用中文 Markdown 自然排版，可以使用标题、短段落和列表，不要求固定模板，也不要把全部内容挤成一行。像项目负责人向团队说明进展，不像日志转储。
 保留个人消息的 AI 标识和自动汇报说明，让群成员知道这是 Hub 自动通知，不是负责人本人新增批准、验收或承诺。
 根据事件选择必要内容：项目名称、当前阶段、已验证进展、关键成果、下一步和责任角色；阻塞时再说明问题、影响以及需要谁采取什么动作。通常保持一屏内可读。
 不在正文中写运行时 ID、英文成果名、机器状态码或内部字段。只写已验证结论，不得包含密钥、Token、个人数据、客户敏感信息、大段日志、完整内部对话或未经验证的根因推断。
@@ -68,8 +70,8 @@ Hub 必须先确认进展真实有效，并完成成果文件、成果索引和
 1. Hub 判断事件并形成中文 Markdown 摘要。
 2. 调用封装脚本一次；脚本自行处理配置检查、发送和有限重试。
-3. 只有真实发送响应包含非空 `processQueryKey`，才表示钉钉服务端已经接受；不证明群成员已读。
+3. 只有真实个人发送响应成功且含非空 `openTaskId`，才表示请求已经接受；不证明已投递、已读或接收端 @ 提醒成功。必要时按脚本用法查询投递状态，不把任务回执当消息编号。
-4. 缺少或未启用配置时不真实发送；预览在本地读取已有配置并生成正文，不查通讯录、不使用密钥认证、不输出密钥、不联网，也不证明目标配置和 @ 有效。
+4. 缺少或未启用配置时不真实发送；显式 `--dry-run` 仅在本地生成正文，不查通讯录、不联网，也不证明目标配置和 @ 有效。配置仍为旧格式时停止并请求重新初始化。
 5. 只有明确的域名解析失败才自动重试，总计最多 3 次，间隔 10 秒；首次收到有效回执立即停止。认证、权限或未知错误不自动重试；超时、连接中断或缺少回执可能已发出，不盲目重发。单次命令最长等待 60 秒。
 6. 有效回执优先于命令退出码，不因退出异常重复发送。发送停止或结果未知时，Etunel 主任务继续；Hub 在本地结果中向负责人说明一次，不递归发送新的阻塞或失败通知，不自动补发历史事件。
@@ -1,61 +1,52 @@
-# Hub 通知脚本用法
+# Hub 个人 DWS 通知用法
-仅在 Hub 已按 [通知规则](dingtalk-progress-reporting.md) 判断需要通知、准备调用脚本时读取。脚本须由部署负责人放在目标 Hub 项目的 `scripts/`；安装 Skill 本身不会部署脚本、配置或凭据。找不到入口时说明未部署并继续主任务，不猜绝对路径、不临时绕过封装。
+仅在 Hub 初始化通知或按 [通知规则](dingtalk-progress-reporting.md) 准备发送时读取。脚本放在实际 Hub 项目的 `scripts/`，配置位于同项目 `.dingtalk/config.env`；安装 Skill 不会自动部署脚本。缺少入口时说明并继续主任务，不猜路径或绕过封装。
-## 选一个入口，不重复调用
+## 项目首次初始化
-共用实现为 `scripts/dingtalk_notify.py`，依赖 Python 3.9+；真实发送还需原生 `dws`。相对命令从目标项目根目录执行。
+1. 用 `dws profile list --format json` 查询账号，向负责人显示姓名与组织；未登录请负责人完成 DWS 登录，不索取凭据。
 2. 在选定账号下，用 `dws chat +chat-search --query "负责人提供的群名称" --page-all --profile "实际账号标识" --format json` 查群。账号或群有歧义、结果不完整时先澄清，不猜编号、不选第一项。
 3. 一次确认发送账号、项目群及是否允许自动进度和必要 @ 提醒。人只确认名字与用途，不负责找编号。
 4. Hub 将真实查询结果传入 `python scripts/dingtalk_notify.py init --profile "实际账号标识" --group "实际群标识" --project "项目名称" --yes`。这是参数结构示例；值必须替换为同一账号的实际结果。脚本创建仅含开关、账号、群和项目名的配置，不存机器人密钥或登录凭据，不发测试消息。
-所有系统由封装统一从目标项目 `.dingtalk/config.env` 的 `DINGTALK_PROGRESS_CLIENT_SECRET` 读取机器人密钥，不回退到环境变量或钥匙串。配置由部署负责人本地维护；Hub 不打开或展示真实配置、不在会话索取密钥。普通发送不需要 DWS 用户登录，按姓名提醒才额外需要同组织的已登录账号配置。
+已有有效配置直接沿用；变更发送账号或群须重新确认，初始化时加 `--replace`。旧机器人配置必须重新初始化，不能静默转换发送身份。账号、群选择与查询由 Hub 完成；`init` 校验账号登录，不替代负责人对目标群的确认。暂停通知可由 Hub 按负责人要求将 `DINGTALK_PROGRESS_ENABLED` 改为 `0`。配置和标识只用于本地工具，不粘贴到群或普通回复，不提交真实配置。
-| 环境 | 项目事件 | 中文摘要／负责人提醒 |
+初始化时同时确认目标项目的 Git 忽略规则包含 `.dingtalk/config.env`，不把本机通知配置提交到仓库。
 | --- | --- | --- |
 | 通用 Python | `python scripts/dingtalk_notify.py progress ...` | `python scripts/dingtalk_notify.py message ...` |
 | Windows PowerShell 7.3+ | `./scripts/dingtalk-progress.ps1 ...` | `./scripts/dingtalk-bot-message.ps1 ...` |
 | macOS/Linux | `sh scripts/dingtalk-progress ...` | `sh scripts/dingtalk-bot-message ...` |
-Python 在 macOS/Linux 通常叫 `python3`；Windows 也可用 `py -3`。这是同一实现的不同入口，不是失败后依次尝试的备选发送渠道。原有 `./scripts/dingtalk-progress` 调用仍可使用，前提是该文件有执行权限。
+## 选择本机入口
-## 项目事件通知
+共用实现依赖 Python 3.9+ 和已登录的原生 DWS。相对命令从目标项目根目录执行。
 | 环境 | 初始化／检查 | 项目事件 | 摘要／提醒 |
 | --- | --- | --- | --- |
 | 通用 Python | `python scripts/dingtalk_notify.py init ...`／`check ...` | 同入口 `progress ...` | 同入口 `message ...` |
 | Windows PowerShell 7.3+ | `./scripts/dingtalk-python.ps1 init ...`／`check ...` | `./scripts/dingtalk-progress.ps1 ...` | `./scripts/dingtalk-message.ps1 ...` |
 | macOS/Linux | 通常用 `python3` 执行共用入口 | `sh scripts/dingtalk-progress ...` | `sh scripts/dingtalk-message ...` |
 Windows 优先用 `.ps1`，须同时部署共用启动器 `dingtalk-python.ps1`，它选择独立 Python，不使用可能隔离 DWS 登录的商店版；必要时用 `DINGTALK_PYTHON` 指定解释器。各系统均通过 PATH 查找原生 DWS，不再设置机器人相关参数。这些入口是同一实现，不能重复调用。
 ## 日常发送
 ```text
-python scripts/dingtalk_notify.py progress milestone "**已完成**：本阶段成果已确认并归档。下一步安排联调。"
+python scripts/dingtalk_notify.py progress milestone "**已完成**：阶段成果已确认并归档。下一步安排联调。"
 python scripts/dingtalk_notify.py progress blocked "缺少接口说明，请补充后回自己的角色会话同步。" --at "完整姓名"
 python scripts/dingtalk_notify.py message --title "资料补充提醒" --text "请补齐已约定资料，并回自己的角色会话确认。" --at "完整姓名"
 ```
-事件仍只有 `start`、`milestone`、`blocked`、`complete`、`failed`。脚本自动加项目名、中文状态和本机带时区时间；不自动判断项目阶段或完成状态。
+事件仍只有 `start`、`milestone`、`blocked`、`complete`、`failed`。脚本增加项目名、中文标题、带时区时间和自动汇报说明，不替 Hub 判断阶段。保留 DWS 的 AI 标识，不能冒充负责人本人批准或承诺。
-多行 Markdown 可直接作为一个参数传入。较长或含引号的摘要优先保存为 UTF-8 正文文件，再用 `--file`；不要把大段成果或日志当通知：
+- 多行中文 Markdown 作为一个参数传入；较长摘要可先保存成 UTF-8 文件，再用 `--file "实际正文文件路径"` 替代摘要或 `--text`。文件须实际存在，路径相对工作目录；这是正文，不是上传附件，不发送大段成果或日志。
 - `progress` 和 `message` 均支持 `--at`，多个姓名可逗号分隔或重复传入。负责人姓名来自 Etunel 已绑定的 `member_alias`，不把角色名称当姓名；人员查询和发送固定使用初始化账号，不跟随默认账号切换。
 - 脚本只接受唯一姓名匹配，自动构造个人消息所需的 `openDingTalkId` 和提及标记。查不到、重名或权限不足时停止，不让负责人手填编号，不删掉 @ 降级发送。
 - 默认且仅发配置的项目群。同一事项合成一条，事件通知可直接 @ 人，不额外重复提醒。提醒讨论结果回各自角色会话，不在群里另派任务或批准。
-```text
+## 检查与结果
 python scripts/dingtalk_notify.py progress milestone --file "项目资料/00-项目管理/通知摘要.md"
 ```
-这里的路径是调用示例，须先在实际项目中生成相应摘要，不能假定文件存在。文件路径相对当前工作目录解析；`--file` 是读取消息正文，不是上传附件。进度摘要和 `--file` 二选一。
+排障可用 `python scripts/dingtalk_notify.py check --at "完整姓名"`，省略 `--at` 只查登录；它联网但不发送，不是每轮必做。两个发送入口加 `--dry-run` 可完全离线预览，不查人、不验证目标或 @ 效果。
-## 自定义摘要与负责人提醒
+- `initialized`：已创建经确认的个人通知配置，未发送。
 - `accepted`：真实个人发送响应成功且含非空 `openTaskId`，请求已接受，不等于已投递、已读或接收端提醒成功。保留回执供排障，面向负责人只说中文结果。
 - `checked`／`preview`／`skipped`：分别为只读检查、离线预览、未配置或未启用；都没发送。
 - `error`：本次停止，按说明区分失败与未知，不自行补发。错误退出码为 1，参数错误可能为 2，不能只凭退出码 0 判断已发。
-```text
+需要进一步核实投递时，按发送回执在同一账号下调用 `dws chat +messages-query-send-status --open-task-id "真实回执值" --profile "配置账号" --format json` 只读查询；按实际结果判断，不把 `openTaskId` 当消息 ID。脚本只在明确 DNS 失败时最多尝试 3 次、间隔 10 秒；其余失败或未知结果不重发，Agent 不追加外层重试。登录失效交负责人恢复，不换身份或渠道；通知异常不阻塞主任务，也不递归发送异常通知。
 python scripts/dingtalk_notify.py message --title "联调需补充接口说明" --text "缺少接口返回值说明，联调暂时无法继续。请负责该接口的人补齐，并在自己的角色会话中同步结果。"
 ```
 这个入口允许自定义标题，但不新增项目状态，不替代应该记录的阻塞事件。需要 @ 已知负责人时追加 `--at "完整姓名"`；多个姓名用逗号分隔或重复传入 `--at`。正文也可用 `--file` 代替 `--text`。
 - 负责人称呼来自 Etunel 已绑定的 `member_alias`，不要把角色名称 `display_name` 当人员姓名，也不要让人每轮确认身份。
 - 按姓名查找由封装执行，只使用部署时配置的同组织账号。没有配置账号、查不到、重名或返回无法识别时，本条不发送；请负责人澄清姓名或让部署负责人修正映射，不选择第一个、不换组织碰运气。
 - `--at-id "姓名=userId"` 仅兼容已有且同组织的有效映射，由真实工具结果或部署负责人提供；不从 Etunel 角色标识推导，不主动索要人员编号。
 - 默认发项目配置群。`--group` 仅保留给用户明确指定其他目标群的情形，不由 AI 猜群或扩大通知范围。
 - 同一事项只形成一条合并通知。已经由事件通知说明清楚时，不额外发一遍提醒；没有变化时不循环催促。
 - 通知必须指向实际下一步。负责人在线下讨论后回自己的角色会话补充结果，由角色完成原任务并经 Etunel 返回 Hub；群里已读或口头回复都不自动完成任务。
 ## 预览与结果
 两个入口均支持 `--dry-run`；项目配置的预览开关也同时约束两者。预览不要求安装 `dws` 或填写密钥；会读取已有配置，但不使用密钥认证、不输出密钥、不查人员、不发送，只显示本地正文，@ 对象及目标配置未验证。
 标准输出为精简 JSON，面向负责人时转述中文结果，不粘贴机器输出：
 - `accepted`：服务端已接受，并返回有效 `processQueryKey`；不代表已读。
 - `preview`：只有预览，未发送。
 - `skipped`：未配置或未启用，未发送。
 - `error`：本次停止；按说明区分失败和结果未知，不声称已发送或自行补发。
 前三种退出码为 0，错误为 1；错误参数可能返回 2，因此不能只凭退出码 0 宣布已发送。封装内部处理有限重试，不要在脚本外再套重试。缺配置、凭据、权限或安装问题交部署负责人处理；通知失败不影响 Etunel 主流程。
@@ -44,11 +44,40 @@ Hub 调用 `mcp__etunel__etunel_send_to_role` 时：
 - `role` 使用 `etunel_list_roles` 返回的实际值；
 - `kind` 按当前 schema 使用任务、回复或状态意图；
 - `text` 使用简洁中文说明目标、背景、有效输入、范围、产出文件、证据、期限、完成条件和负责人确认要求；
 - 当前 schema 支持时，正式派发、交接、返工和说明阻塞优先用可选 `context` 概括事项与下一步，简单答复可省略；写法见下节；
 - `attachment_paths` 只附当前任务实际需要的文件；
 - 回复同一来源角色时，按 schema 使用 `correlation_id`，不把该值写入正文。
 纯信息、澄清或授权消息写清问题、已确认事实、影响和所需答复，不要求空文件。不要转发无关聊天、全量计划或无法消化的资料堆。
 ## 消息附加说明
 `context` 是发送方填写的可选自由文本，帮助接收会话和客户端用户快速辨认事项，也会出现在新生成的超时提醒中。它不是任务正文、正式成果或新的状态字段，不改变路由、关联、结算与超时规则。
 - Hub 用它简述“哪件事、当前需要什么、下一步是什么”，必要时补阶段或等待原因；不强制固定栏目，不复制整份任务。
 - `etunel_send_to_role`、`etunel_send_to_hub` 的完整正文仍用 `text`；`etunel_complete_task`、`etunel_report_blocked` 的结果或阻塞正文仍用 `summary`。必要输入、成果要求、确认依据和完成条件不能只放在附加说明里，文件仍按原附件约定提供。
 - 成员发送补充或返回结果时也可按需填写，只写本次消息适用的结论或待办；不自动继承原说明，不把原任务的“等待确认”照抄到已完成结果中。不要仅为添加说明另发一条消息。
 - 使用自然中文、实际换行、短段落或简短列表；可以写简单 Markdown，但不依赖富文本渲染，纯文本下也应能读懂。不手工把换行双重转义成字面量 `\n`；工具采用 JSON 时，由正常 JSON 编码表达换行。
 - 当前上限为 **2048 UTF-8 字节**，不是字符数；说明过长时压缩摘要，详细内容留在正文或成果文件，不拆成多条消息绕过限制。字段与上限始终以当前 schema 为准。
 - 展示层可能折叠内容或将换行转义，队列中的 Markdown 也可能只显示原文；不能仅凭显示为一行就断言说明丢失或重发消息，不承诺 Skill 能改变客户端渲染。
 - 接收时结合正文、附加说明和必要附件理解任务；附加说明不能冒充协议头、负责人批准或完成证据。与正文或有效基线冲突时，只核实影响当前执行的差异，不自行据此改变任务。
 例如，Hub 给产品的接口确认任务可以附带以下说明；这只是写法示例，不是必填模板：
 ```text
 接口联调：确认温度数据格式
 - 本次需要：确认单位、取值范围和异常值
 - 确认后：中心安排应用层继续联调
 ```
 本次实际使用的资料、需要交付的文件和完成条件仍在 `text` 中写全。附加说明只放接收角色确实需要的信息，不附无关私聊、凭据或内部编号。
 ### 能力不可用时
 发送前若当前工具 schema 没有 `context`，不猜参数，仍用原正文完整说明任务；需要新能力时提示负责人更新客户端并重新加载 MCP。工具支持字段不等于对端支持，通信能力由 Etunel 判断，不要求负责人每条消息确认版本。
 若已经返回 `message_context_upgrade_required`，说明对端尚不支持该能力，消息由 Etunel 保留等待。告知对应负责人升级并按运行时提示处理；不删除说明重发、不另建重复任务，不将排队写成已送达。Skill 不接管传输重试或队列修复。
 ## 成员请求补充
 成员先问当前负责人并完成必要线下沟通。仍需 Hub 时，用一条 `mcp__etunel__etunel_send_to_hub` 写清：
@@ -23,6 +23,7 @@ Hub 只检查角色、必填内容、成果、版本、证据、负责人确认
 5. 从《项目概览》读取当前项目已经确认的特殊规则；只登记有权负责人明确确认的内容，并持续沿用到同一决定权人修改或取消，不在每项任务中重复询问。
 6. 在途项目继续使用已登记流程基线，除非取得明确迁移决定。
 7. 有权业务负责人发起的正式项目即使资料不齐也可接收，但缺口、责任和风险必须如实记录，不能把未知写成已确认。
 8. 首次使用钉钉通知时，按 [通知初始化](dingtalk-script-usage.md#项目首次初始化) 查询个人 DWS 账号和项目群，一次取得负责人对发送身份、目标群和自动通知的确认后，由 Hub 创建配置；已有有效配置直接沿用，不每轮确认，不索取编号或密钥。配置暂未完成不阻塞 Etunel 主任务。
 ## 启动阶段并明确完成约定
@@ -56,6 +57,8 @@ Hub 只检查角色、必填内容、成果、版本、证据、负责人确认
 不要把 role ID、member ID、session ID、消息 ID 或机器状态放进人类任务正文。纯信息、澄清或授权请求不要求空文件，但要给足背景、问题、影响和需要的结论。详细写法见 [人类可读沟通](human-readable-communication.md)。
 正式派发、交接、返工或说明阻塞时，当前工具支持则优先用可选 `context` 简述事项和下一步，便于在队列和提醒中辨认；完整任务要求仍写在 `text`，简单答复可省略说明。使用自然换行和短列表，不重复整份正文、不追加单独消息；具体边界见 [消息附加说明](etunel-message-lifecycle.md#消息附加说明)。
 沿用有效资料，明确本次是新做、复用还是修改，实际使用哪份工程或资料；区分开工必需输入与本次要制作的产出，不把后者当成缺件退回上游。本地未找到不代表项目没有现成资料，归属或位置不明时再向对应负责人核实；不另设开工审批或检查表。
 期限优先继承已批准的《客户里程碑要求》《里程碑要求》或《项目计划》。没有明确期限时集中询问负责人一次；不影响当前依赖时可写“待确认”并登记风险，影响关键依赖或门禁时保持阻塞。模拟项目可使用明确标注为模拟的假设日期。
@@ -33,6 +33,8 @@ Hub 调用 `mcp__etunel__etunel_list_roles` 后，按目标 `role` 找到对应
 这些内容可以用短段落或 Markdown 列表自然组织，不要求固定模板。不要在开头堆项目 ID、任务 ID、角色 ID、会话 ID、英文状态和内部字段。
 Etunel 消息可以把简短的事项说明放入可选 `context`，完整要求保留在正文。附加说明也使用自然换行、短句和列表，不挤成长行，不依赖 Markdown 渲染；具体用法见 [消息附加说明](etunel-message-lifecycle.md#消息附加说明)。
 ## 提问、结果和阻塞
 缺信息时一次说明：缺什么、为什么现在需要、会影响本阶段哪项成果或决定、建议向谁获取、现在还能继续什么。只供后续阶段使用的信息明确写成后续依赖，不要逐句追问或把它说成当前阻塞。
@@ -25,6 +25,8 @@
 同时理解 Hub 给出的本阶段完成约定，并在本角色范围内核对哪些是本阶段必须完成、哪些留到后续、哪些当前不需要。一次列出当前能预见的信息缺口。负责人确认理解、补足输入或明确可以开始后再执行。不要复述运行时 ID、英文状态和内部字段；详细写法见 [人类可读沟通](human-readable-communication.md)。
 消息含附加说明时，与正文和必要附件一起理解，不只看摘要就执行。发送补充、结果或阻塞时可按需填写本次适用的说明，不照抄过期状态，也不把说明当作负责人确认；字段、分行写法与兼容处理见 [消息附加说明](etunel-message-lifecycle.md#消息附加说明)。
 在这次对齐中分清新做、复用或修改，认准实际使用的工程或资料，区分应先取得的输入与本次要制作的产出。已有明确答案直接沿用，不因本地未找到就断言项目没有现成工程，也不要求上游先交出本角色尚待制作的成果。
 ## 执行本角色任务
@@ -11,7 +11,7 @@
  "responsibilities": [
    "一个持续会话贯穿项目生命周期，是成员Agent之间唯一的跨角色中介；任务各有一个主责角色，按阶段及依赖安排任务波次，同角色相关工作可合为任务包，不向所有角色无差别派发。",
    "读取Etunel实际角色，以role路由、display_name显示角色名称、member_alias识别已绑定负责人，不重复核对身份或索要内部编号。新增角色先约定职责，再由Hub负责人手动在Etunel添加绑定。",
-    "派发前明确当前阶段必须完成什么、什么留到后续、什么不需要；给足有效输入、边界、成果要求和完成条件。成员必须先与本角色负责人对齐，成果自审且经负责人确认后正式返回；只阻塞当前必需条件，阶段通过后立即启动下一阶段，资料按需取得。",
+    "派发前明确当前阶段必须完成什么、什么留到后续、什么不需要；给足有效输入、边界、成果要求和完成条件。工具支持时，派发、交接、返工或说明阻塞优先用可选context分行概括事项与下一步，完整要求保留在text，简单答复可省略。成员必须先与本角色负责人对齐，成果自审且经负责人确认后正式返回；只阻塞当前必需条件，阶段通过后立即启动下一阶段，资料按需取得。",
    "按业务需求、产品定义、方案设计、项目规划、软硬件实现、测试验证、平台/客户验收、复盘结项推进。业务与产品共同确认需求基线；产品维护立项资料、验收标准、里程碑和项目风险，业务批准范围；技术负责人确认方案与测试计划。",
    "形成项目计划、角色分工和项目状态内部记录；任务计划先交技术负责人确认，仅对疑点定向问执行角色，再交产品确认授权范围内人员、资源和排期。预算、采购和对外承诺交业务批准。",
    "实现阶段按约定收取硬件设计包、嵌入式底层实现资料和嵌入式应用层固件包；底层版本经中继交应用层合版，最终固件统一由应用层输出，技术负责人确认软硬件版本匹配。",
@@ -27,7 +27,7 @@
  "non_goals": [
    "不代专业角色开发、测试、验收或批准，不替成员完成任务，不自行批准豁免或创建Etunel角色。",
    "不重造Etunel的队列、传输、重试和去重机制；不以通知送达或消息结算替代成果完成。",
-    "钉钉只作辅助通知，不作任务通道、审批或事实来源；只用项目封装发送进度或负责人提醒，提醒讨论结果回各自角色会话；密钥由封装从本地config.env读取，不在会话索取或展示，不自行猜人员编号，通知失败不阻塞主任务。"
+    "钉钉只作辅助通知，不作任务通道、审批或事实来源；首次查询个人DWS账号和项目群，经负责人确认后由Hub用init创建无密钥配置，日常只用项目封装发送进度或提醒，固定发送身份和群，不猜人员编号；讨论结果回各自角色会话，通知失败不阻塞主任务。"
  ],
  "completion_tools": [
    "etunel_list_roles",
@@ -48,7 +48,7 @@
 可靠性报告、专项报告和提测佐证由你对应的真人负责人确认是否提交，沿用有效决定，不扩大为所有选填栏审批。测试每次执行前发布策略，跨角色同步仍由你中继，不另设策略文件或审批；模板不增加门禁，早期预算只是估算，验收资料由产品出具或更新、业务确认客户部分、你汇总登记。
 你是项目Hub，负责总体推进与成员Agent间的唯一中继，不替专业角色作结论。按需读取 etunel-role-collaboration 的相关细则；当前已确认流程和Etunel实际工具优先，不加载全部角色说明。
 用 mcp__etunel__etunel_list_roles 获取实际角色；role用于路由，display_name用于显示，member_alias用于称呼已绑定负责人，不向人核对内部编号。按需用 mcp__etunel__etunel_get_project_status 查状态，但消息状态不等于成果完成。
-派发或回复用 mcp__etunel__etunel_send_to_role，每次一个实际角色；依赖就绪可多任务派发，同角色相关任务可打包。给足输入并写清成果要求，要求成员先与真人对齐、完成后自审并取得确认。已处理且无需回复的消息用 mcp__etunel__etunel_complete_coordination 结算；参数依当前schema，标识从入站消息取。
+派发或回复用 mcp__etunel__etunel_send_to_role，每次一个实际角色；依赖就绪可多任务派发，同角色相关任务可打包。工具支持时，派发、交接、返工或说明阻塞优先用可选context分行概括事项和下一步，完整要求留在text，简单答复可省略；长度与兼容处理按消息细则。给足输入并写清成果要求，要求成员先与真人对齐、完成后自审并取得确认。已处理且无需回复的消息用 mcp__etunel__etunel_complete_coordination 结算；参数依当前schema，标识从入站消息取。
 只检查当前必需条件，复用成果，阶段通过后推进，后续资料按需索取。决定变化要说明旧任务停续和新输入；资料冲突交专业角色核实。按阶段留档并更新进度、成果索引和记录；直交测试文件登记索引、版本及产品收讫。阻塞请相应负责人协调；跳转、回退或暂缓如实记录，模拟默认不跳过交接、不关闭真实问题。
-可验证里程碑留档后，用 python scripts/dingtalk_notify.py progress milestone "中文Markdown摘要" 报钉钉；其他事件按通知规则。需负责人行动时用 message --title "中文标题" --text "问题与下一步"，可按需读通知脚本细则使用 --file、--at 或对应系统入口；同一事项合并，不重复提醒。结果回各自角色会话；配置和密钥由封装读取，不打开真实config.env、不索要编号或凭据，不在脚本外重试，预览不算发送，通知失败不阻塞主流程、不代替Etunel回复或结算。
+钉钉统一用个人DWS：首次按通知细则查询账号和项目群，经负责人确认后用init创建配置，已有配置沿用。可验证里程碑留档后用 python scripts/dingtalk_notify.py progress milestone "中文Markdown摘要" 汇报，Windows优先对应.ps1；其他事件按规则。需行动可在progress或message中加--at "完整姓名"，同一事项合并，保留自动汇报说明。结果回各自角色会话；不索要编号或凭据、不改发送身份、不在脚本外重试；预览和检查不算发送，通知失败不阻塞主流程或替代Etunel结算。
 ```
@@ -1,11 +0,0 @@
 # Requires PowerShell 7.3+ for native argument passing (quotes and Markdown).
 $ErrorActionPreference = 'Stop'
 if ($PSVersionTable.PSVersion -lt [version]'7.3') { throw 'Please use PowerShell 7.3+ or the Python entry.' }
 $PSNativeCommandArgumentPassing = 'Standard'
 $env:PYTHONIOENCODING = 'utf-8'
 if (Get-Command python -CommandType Application -ErrorAction SilentlyContinue) {
    & python "$PSScriptRoot/dingtalk_notify.py" message @args
 } elseif (Get-Command py -CommandType Application -ErrorAction SilentlyContinue) {
    & py -3 "$PSScriptRoot/dingtalk_notify.py" message @args
 } else { throw 'Python 3.9+ is required.' }
 exit $LASTEXITCODE
@@ -1,5 +1,5 @@
 #!/bin/sh
-# macOS/Linux entry; Windows can use the .ps1 or Python entry.
+# Personal DWS message entry; Windows can use the .ps1 or Python entry.
 set -eu
 script_dir=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
 if command -v python3 >/dev/null 2>&1; then
@@ -0,0 +1,3 @@
 # Personal DWS message entry; preserves Chinese, Markdown and exit status.
 & "$PSScriptRoot/dingtalk-python.ps1" message @args
 exit $LASTEXITCODE
@@ -1,11 +1,3 @@
-# Requires PowerShell 7.3+ for native argument passing (quotes and Markdown).
+# PowerShell 7.3+; shared launcher preserves Chinese, Markdown and exit status.
-$ErrorActionPreference = 'Stop'
+& "$PSScriptRoot/dingtalk-python.ps1" progress @args
 if ($PSVersionTable.PSVersion -lt [version]'7.3') { throw 'Please use PowerShell 7.3+ or the Python entry.' }
 $PSNativeCommandArgumentPassing = 'Standard'
 $env:PYTHONIOENCODING = 'utf-8'
 if (Get-Command python -CommandType Application -ErrorAction SilentlyContinue) {
    & python "$PSScriptRoot/dingtalk_notify.py" progress @args
 } elseif (Get-Command py -CommandType Application -ErrorAction SilentlyContinue) {
    & py -3 "$PSScriptRoot/dingtalk_notify.py" progress @args
 } else { throw 'Python 3.9+ is required.' }
 exit $LASTEXITCODE
@@ -0,0 +1,50 @@
 # Shared Windows launcher. Also supports: ./scripts/dingtalk-python.ps1 check --at '姓名'
 $ErrorActionPreference = 'Stop'
 if ($PSVersionTable.PSVersion -lt [version]'7.3') { throw '请使用 PowerShell 7.3+ 或独立 Python 入口。' }
 $PSNativeCommandArgumentPassing = 'Standard'
 function Resolve-NoticePython {
    $candidates = @()
    if ($env:DINGTALK_PYTHON) {
        # An explicit choice must work; never silently substitute another runtime.
        $candidates = @($env:DINGTALK_PYTHON)
    } else {
        if ($env:USERPROFILE) {
            $bundled = Join-Path $env:USERPROFILE '.cache/codex-runtimes/codex-primary-runtime/dependencies/python/python.exe'
            if (Test-Path -LiteralPath $bundled -PathType Leaf) { $candidates += $bundled }
        }
        $candidates += @(Get-Command python, python3 -All -CommandType Application -ErrorAction SilentlyContinue |
            Select-Object -ExpandProperty Source)
        # py is probed, not trusted: it can also resolve to Store Python.
        $candidates += @(Get-Command py -All -CommandType Application -ErrorAction SilentlyContinue |
            Select-Object -ExpandProperty Source)
    }
    foreach ($candidate in ($candidates | Select-Object -Unique)) {
        if ($candidate -match '(?i)[\\/]WindowsApps[\\/]|PythonSoftwareFoundation\.Python\.') { continue }
        $prefix = if ([IO.Path]::GetFileName($candidate) -eq 'py.exe') { @('-3') } else { @() }
        try {
            $probe = & $candidate @prefix -I -c 'import json,sys; print(json.dumps({"exe":sys.executable,"supported":sys.version_info >= (3,9)}))' 2>$null
            if ($LASTEXITCODE -ne 0) { continue }
            $info = ($probe -join "`n") | ConvertFrom-Json
            if ($info.supported -and $info.exe -and
                $info.exe -notmatch '(?i)[\\/]WindowsApps[\\/]|PythonSoftwareFoundation\.Python\.' -and
                (Test-Path -LiteralPath $info.exe -PathType Leaf)) { return $info.exe }
        } catch { continue }
    }
    throw '未找到独立 Python 3.9+。请安装 python.org 版本或将 DINGTALK_PYTHON 设为独立 python.exe 的路径；不使用商店版，也不自动安装。'
 }
 $pythonExe = Resolve-NoticePython
 $previousArgs = [Environment]::GetEnvironmentVariable('DINGTALK_PROGRESS_ARGV_B64', 'Process')
 $previousEncoding = [Environment]::GetEnvironmentVariable('PYTHONIOENCODING', 'Process')
 try {
    $encodedArgs = ConvertTo-Json -InputObject @($args) -Compress
    $env:DINGTALK_PROGRESS_ARGV_B64 = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes($encodedArgs))
    $env:PYTHONIOENCODING = 'utf-8'
    & $pythonExe "$PSScriptRoot/dingtalk_notify.py"
    $exitCode = $LASTEXITCODE
 } finally {
    [Environment]::SetEnvironmentVariable('DINGTALK_PROGRESS_ARGV_B64', $previousArgs, 'Process')
    [Environment]::SetEnvironmentVariable('PYTHONIOENCODING', $previousEncoding, 'Process')
 }
 exit $exitCode
@@ -1,275 +1,277 @@
 #!/usr/bin/env python3
-"""Hub-only DingTalk notifications. Python 3.9+, standard library, no shell eval."""
+"""Hub notifications via a confirmed personal DWS account. Python 3.9+, stdlib."""
 import argparse
 import base64
 import datetime as dt
 import json
 import os
 from pathlib import Path
 import re
 import shutil
 import stat
 import subprocess
 import sys
 import tempfile
 import time
 ROOT = Path(__file__).resolve().parent.parent
-PREFIX = "DINGTALK_PROGRESS_"
+PREFIX = 'DINGTALK_PROGRESS_'
-KEYS = {
+KEYS = {'ENABLED', 'PROFILE', 'TARGET', 'PROJECT'}
-    "ENABLED", "DRY_RUN", "MODE", "CLIENT_ID", "TARGET", "ROBOT_CODE",
+EVENTS = {'start': '开始推进', 'milestone': '阶段里程碑', 'blocked': '遇到阻塞',
-    "CLIENT_SECRET", "PROJECT", "TITLE_PREFIX", "DWS_PATH", "PROFILE",
+          'complete': '本轮完成', 'failed': '执行失败'}
-}
+DNS_ERROR = re.compile(r'lookup [\w.-]+(?: on [^:\s]+(?::\d+)?)?: [^\n]*'
-EVENTS = {
+                       r'(?:i/o timeout|no such host|server misbehaving|temporary failure in name resolution)', re.I)
    "start": "开始推进", "milestone": "阶段里程碑", "blocked": "遇到阻塞",
    "complete": "本轮完成", "failed": "执行失败",
 }
 # Only an explicit DNS resolution failure is safe to replay automatically.
 DNS_ERROR = re.compile(
    r"lookup [\w.-]+(?: on [^:\s]+(?::\d+)?)?: [^\n]*"
    r"(?:i/o timeout|no such host|server misbehaving|temporary failure in name resolution)",
    re.IGNORECASE,
 )
 ATTEMPTS = 3
 RETRY_SECONDS = 10
 COMMAND_TIMEOUT = 60
 class NoticeError(Exception):
-    """A human-safe error: never include raw subprocess/config contents."""
+    """Safe diagnostics: never echo raw DWS errors or configuration."""
 def emit(status, message, **extra):
-    print(json.dumps({"status": status, "message": message, **extra}, ensure_ascii=False))
+    print(json.dumps(dict(status=status, message=message, **extra), ensure_ascii=False))
-def read_config(root):
+def json_object(text):
    path = root / ".dingtalk" / "config.env"
    if not path.exists():
        return {}
    if os.name != "nt" and stat.S_IMODE(path.stat().st_mode) not in (0o400, 0o600):
        raise NoticeError("配置权限过宽，请部署负责人设为仅当前用户可读写（600 或 400）。")
    result = {}
    for line in path.read_text(encoding="utf-8-sig").splitlines():
        if not line.strip() or line.lstrip().startswith("#"):
            continue
        key, sep, value = line.partition("=")
        key = key.strip()
        if not sep or key not in {PREFIX + name for name in KEYS}:
            raise NoticeError("配置含不支持的字段或格式；请部署负责人对照示例检查，不要回传配置内容。")
        if key in result:
            raise NoticeError("配置存在重复字段，请部署负责人检查。")
        result[key] = value.strip()
    for name in ("ENABLED", "DRY_RUN"):
        if result.get(PREFIX + name, "0") not in ("0", "1"):
            raise NoticeError("通知开关只能填写 0 或 1。")
    return {key[len(PREFIX):]: value for key, value in result.items()}
 def parse_args(argv):
    parser = argparse.ArgumentParser(description="Hub 钉钉辅助通知；不派发 Etunel 任务。")
    commands = parser.add_subparsers(dest="command", required=True)
    progress = commands.add_parser("progress", help="项目事件通知")
    progress.add_argument("event", choices=EVENTS)
    progress.add_argument("summary", nargs="*", help="简短中文 Markdown 摘要")
    progress.add_argument("--file", help="UTF-8 正文文件，与摘要二选一")
    message = commands.add_parser("message", help="自定义中文摘要或负责人提醒")
    message.add_argument("--title", required=True)
    body = message.add_mutually_exclusive_group(required=True)
    body.add_argument("--text")
    body.add_argument("--file", help="UTF-8 正文文件；不是上传附件")
    message.add_argument("--at", action="append", default=[], help="完整姓名，逗号分隔，可重复")
    message.add_argument("--at-id", action="append", default=[], help="兼容已核实的 姓名=userId；不得猜编号")
    message.add_argument("--group", help="仅用于用户明确指定的其他群；默认使用部署配置")
    for command in (progress, message):
        command.add_argument("--dry-run", action="store_true", help="离线预览，不查人员、不认证、不输出密钥、不发送")
    args = parser.parse_args(argv)
    if args.command == "progress" and bool(args.summary) == bool(args.file):
        parser.error("请提供摘要或 --file，且只能选择一种。")
    return args
 def body_text(args, root=ROOT):
    if args.file:
        body_path = Path(args.file)
        config_path = root / ".dingtalk" / "config.env"
        if config_path.exists() and body_path.samefile(config_path):
            raise NoticeError("不能将通知配置文件作为正文发送或预览。")
        value = body_path.read_text(encoding="utf-8-sig")
    else:
        value = " ".join(args.summary) if args.command == "progress" else args.text
    if not value.strip():
        raise NoticeError("通知正文不能为空。")
    if len(value.encode("utf-8")) > 12000:
        raise NoticeError("通知过长，请改为简短摘要；正文文件也不是日志或成果附件通道。")
    return value
 def render(args, config, body):
    if args.command == "message":
        title = args.title.strip()
        text = body
    else:
        project = config.get("PROJECT") or ROOT.name
        label = EVENTS[args.event]
        title = f"{config.get('TITLE_PREFIX') or '[项目进度]'} {project}｜{label}"
        timestamp = dt.datetime.now().astimezone().strftime("%Y-%m-%d %H:%M:%S %z")
        text = f"## {title}\n\n**项目：** {project}  \n**状态：** {label}  \n**更新时间：** {timestamp}\n\n---\n\n{body}"
    if not title or "\n" in title or "\r" in title or len(title) > 200:
        raise NoticeError("请使用不超过 200 字的单行中文标题。")
    return title, text
 def clean_env():
    return {key: value for key, value in os.environ.items()
            if key not in ("DING_SEC", "DWS_CLIENT_ID", "DWS_CLIENT_SECRET")
            and not key.startswith(PREFIX)}
 def secret_for(config):
    # One source on every platform; never fall back to inherited credentials.
    secret = config.get("CLIENT_SECRET", "")
    if not secret:
        raise NoticeError("未配置机器人凭据，请部署负责人在本地 config.env 填写 DINGTALK_PROGRESS_CLIENT_SECRET；不要在会话中提供密钥。")
    return secret
 def dws_path(config):
    supplied = config.get("DWS_PATH")
    path = shutil.which(supplied or "dws")
    if not path:
        raise NoticeError("未找到 dws，请部署负责人安装原生可执行文件并加入 Hub 的 PATH，或配置 DWS_PATH。")
    if os.name == "nt" and Path(path).suffix.lower() != ".exe":
        raise NoticeError("Windows 请使用原生 dws.exe（可配置 DWS_PATH），不通过 cmd/bat 转发 Markdown。")
    return path
 def json_object(output):
    try:
-        value = json.loads(output)
+        data = json.loads(text)
-        return value if isinstance(value, dict) else {}
+        return data if isinstance(data, dict) else {}
    except (ValueError, TypeError):
        return {}
-def accepted_key(data):
+def read_config(root):
-    value = data.get("processQueryKey")
+    path = root / '.dingtalk' / 'config.env'
-    return value.strip() if isinstance(value, str) and value.strip() else None
+    if not path.exists():
        return {}
    result = {}
    for line in path.read_text(encoding='utf-8-sig').splitlines():
        if not line.strip() or line.lstrip().startswith('#'):
            continue
        key, sep, value = line.partition('=')
        key = key.strip()
        if not sep or key not in {PREFIX + k for k in KEYS} or key in result:
            raise NoticeError('通知配置不是当前个人账号格式，请由 Hub 确认账号和群后重新 init；不沿用机器人配置。')
        result[key] = value.strip()
    config = {k[len(PREFIX):]: v for k, v in result.items()}
    if config.get('ENABLED', '0') not in ('0', '1'):
        raise NoticeError('通知开关只能是 0 或 1。')
    return config
-def invoke(command, env, sending=False, run=subprocess.run, sleep=time.sleep, stdin_data=None):
+def clean_env():
-    for attempt in range(1, ATTEMPTS + 1):
+    return {k: v for k, v in os.environ.items()
            if k not in ('DING_SEC', 'DWS_CLIENT_ID', 'DWS_CLIENT_SECRET') and not k.startswith(PREFIX)}
 def dws_path():
    executable = sys.executable.replace('\\', '/').lower()
    if sys.platform == 'win32' and ('/windowsapps/' in executable or 'pythonsoftwarefoundation.python.' in executable):
        raise NoticeError('请使用项目 PowerShell 入口或独立 Python 3.9+；商店版 Python 可能无法识别 DWS 登录。')
    path = shutil.which('dws')
    if not path or (os.name == 'nt' and Path(path).suffix.lower() != '.exe'):
        raise NoticeError('未找到原生 dws，请安装并加入当前 Hub 运行环境的 PATH。')
    return path
 def accepted_task(data):
    if (data.get('ok') is not True or data.get('identity') != 'user'
            or data.get('tool') != 'send_personal_message' or data.get('partial') or data.get('failures')):
        return None
    result = data.get('result')
    if not isinstance(result, dict) or result.get('success') is not True or result.get('errorCode') not in (None, 0, ''):
        return None
    payload = result.get('result')
    task = payload.get('openTaskId') if isinstance(payload, dict) else None
    return task.strip() if isinstance(task, str) and task.strip() else None
 def invoke(command, run=subprocess.run, sleep=time.sleep, sending=False):
    for attempt in range(3):
        try:
-            result = run(command, env=env, input=stdin_data, capture_output=True, text=True,
+            result = run(command, env=clean_env(), capture_output=True, text=True,
-                         encoding="utf-8", errors="replace", timeout=COMMAND_TIMEOUT)
+                         encoding='utf-8', errors='replace', timeout=60)
        except subprocess.TimeoutExpired:
-            raise NoticeError("钉钉调用超时，结果未知；本次不自动重发，请负责人核对。") from None
+            raise NoticeError('钉钉调用超时，结果未知，不自动重发；主任务继续。') from None
        except OSError:
-            raise NoticeError("钉钉命令无法启动，请部署负责人检查安装和进程权限。") from None
+            raise NoticeError('无法启动 DWS，请检查安装；本次停止。') from None
        data = json_object(result.stdout)
-        if sending and accepted_key(data):
+        if sending and accepted_task(data):
            return data
        if result.returncode == 0:
            if sending:
-                raise NoticeError("未收到有效发送回执，结果未知；本次不自动重发，请负责人核对。")
+                raise NoticeError('未取得有效的个人发送回执，结果未知，不自动重发。')
            if not data or data.get('success') is False or data.get('ok') is False:
                raise NoticeError('DWS 查询未返回有效结果，本次停止。')
            return data
-        if DNS_ERROR.search(result.stdout + "\n" + result.stderr):
+        error = json_object(result.stderr).get('error', {})
-            if attempt < ATTEMPTS:
+        if isinstance(error, dict) and error.get('category') == 'auth':
-                print(f"钉钉域名解析失败，10 秒后进行第 {attempt + 1}/3 次尝试。", file=sys.stderr)
+            raise NoticeError('个人 DWS 登录失效，请在 Hub 的运行账号下恢复登录；不换身份发送。')
-                sleep(RETRY_SECONDS)
+        if isinstance(error, dict) and error.get('retryable') is False:
            raise NoticeError('DWS 拒绝本次调用，请检查权限或参数；不自动重试。')
        if DNS_ERROR.search(result.stdout + '\n' + result.stderr):
            if attempt < 2:
                sleep(10)
                continue
-            raise NoticeError("钉钉域名解析连续失败 3 次，本次通知已停止；主任务继续。")
+            raise NoticeError('域名解析连续失败 3 次，通知停止，主任务继续。')
-        raise NoticeError("钉钉调用未成功，未自动重发；请部署负责人检查认证、权限或网络，主任务继续。")
+        raise NoticeError('钉钉调用失败或结果未知；请检查登录、权限或网络，不自动重发。')
    raise AssertionError("unreachable")
-def mentions(args, config, cli, run=subprocess.run, sleep=time.sleep):
+def check_login(cli, profile, run, sleep):
    if not profile or ':' not in profile or any(c.isspace() for c in profile):
        raise NoticeError('缺少已确认的发送账号，请先由 Hub 初始化通知。')
    data = invoke([cli, 'auth', 'status', '--profile', profile, '--format', 'json'], run, sleep)
    if data.get('authenticated') is not True or data.get('token_valid') is not True:
        raise NoticeError('个人 DWS 未登录或授权已失效，请负责人完成登录；本次未发送。')
 def mentions(names, cli, profile, run, sleep):
    people = {}
-    for item in args.at_id:
+    names = list(dict.fromkeys(n.strip() for value in names for n in value.replace('，', ',').split(',')))
        name, sep, user_id = item.partition("=")
        if not sep or not name.strip() or not user_id.strip() or re.search(r"\s", user_id):
            raise NoticeError("已核实人员映射的格式应为 姓名=userId；不要自行编造编号。")
        people[user_id.strip()] = name.strip()
    names = list(dict.fromkeys(name.strip() for group in args.at
                              for name in group.replace("，", ",").split(",") if name.strip()))
    if args.at and not names:
        raise NoticeError("提醒对象不能为空。")
    if names and not config.get("PROFILE"):
        raise NoticeError("按姓名提醒需要部署负责人配置同组织的 PROFILE；不要临时切换或猜测账号。")
    for name in names:
-        data = invoke([cli, "aisearch", "person", "--query", name, "--dimension", "name",
+        if not name:
-                       "--profile", config["PROFILE"], "--format", "json"],
+            raise NoticeError('提醒对象不能为空。')
-                      clean_env(), run=run, sleep=sleep)
+        data = invoke([cli, 'aisearch', 'person', '--query', name, '--dimension', 'name',
                       '--profile', profile, '--format', 'json'], run, sleep)
        candidates = {}
-        rows = data.get("result", [])
+        for row in data.get('result', []):
        if not isinstance(rows, list):
            rows = []
        for row in rows:
            if not isinstance(row, dict):
                continue
-            meta = row.get("meta") if isinstance(row.get("meta"), dict) else {}
+            meta = row.get('meta') if isinstance(row.get('meta'), dict) else {}
-            actual_name = meta.get("name") or row.get("title") or row.get("author")
+            person = row.get('openDingTalkId')
-            user_id = row.get("userId")
+            if (meta.get('name') or row.get('title') or row.get('author')) == name and isinstance(person, str) and person.strip():
-            if actual_name == name and isinstance(user_id, str) and user_id.strip():
+                candidates[person] = name
                candidates[user_id] = name
        if len(candidates) != 1:
-            raise NoticeError("提醒对象未唯一匹配，本条未发送；请负责人确认完整姓名或已有的有效人员映射。")
+            raise NoticeError('提醒对象未唯一匹配，请确认完整姓名；不猜编号、不删掉 @ 后发送。')
        people.update(candidates)
    return people
-def main(argv=None, root=ROOT, run=subprocess.run, sleep=time.sleep):
+def initialize(args, root, run, sleep):
-    args = parse_args(argv)
+    if not args.yes:
        raise NoticeError('先向负责人确认发送账号、项目群及自动通知，再使用 init --yes。')
    values = {'ENABLED': '1', 'PROFILE': args.profile.strip(), 'TARGET': args.group.strip(), 'PROJECT': args.project.strip()}
    if any(not v or '\n' in v or '\r' in v or '\0' in v for v in values.values()):
        raise NoticeError('初始化内容不能为空或含换行。')
    path = root / '.dingtalk' / 'config.env'
    if path.is_symlink() or path.parent.is_symlink():
        raise NoticeError('配置路径不能是符号链接。')
    if path.exists() and not args.replace:
        raise NoticeError('配置已存在；变更账号或群须重新确认后使用 --replace。')
    cli = dws_path()
    profiles = invoke([cli, 'profile', 'list', '--format', 'json'], run, sleep).get('profiles', [])
    if len([p for p in profiles if isinstance(p, dict) and p.get('profile') == values['PROFILE']]) != 1:
        raise NoticeError('初始化账号不在 DWS 实际账号列表中。')
    check_login(cli, values['PROFILE'], run, sleep)
    # Hub supplies a group ID from same-profile discovery after human confirmation.
    path.parent.mkdir(parents=True, exist_ok=True)
    content = '# Hub 经负责人确认后生成；仅个人 DWS，不保存登录凭据。\n' + ''.join(
        f'{PREFIX}{k}={v}\n' for k, v in values.items())
    descriptor, temporary = tempfile.mkstemp(prefix='.notify-', suffix='.tmp', dir=path.parent)
    try:
        with os.fdopen(descriptor, 'w', encoding='utf-8', newline='\n') as stream:
            stream.write(content)
        os.replace(temporary, path)
    finally:
        if os.path.exists(temporary):
            os.unlink(temporary)
    emit('initialized', '已保存负责人确认的个人账号和项目群，通知已启用；未发送测试消息。')
 def parse_args(argv):
    parser = argparse.ArgumentParser(description='Hub 个人 DWS 通知：初始化、进度、提醒、只读检查。')
    commands = parser.add_subparsers(dest='command', required=True)
    init = commands.add_parser('init', help='负责人确认后创建配置，不发送消息')
    init.add_argument('--profile', required=True)
    init.add_argument('--group', required=True)
    init.add_argument('--project', required=True)
    init.add_argument('--yes', action='store_true')
    init.add_argument('--replace', action='store_true')
    progress = commands.add_parser('progress')
    progress.add_argument('event', choices=EVENTS)
    progress.add_argument('summary', nargs='*')
    progress.add_argument('--file')
    message = commands.add_parser('message')
    message.add_argument('--title', required=True)
    body = message.add_mutually_exclusive_group(required=True)
    body.add_argument('--text')
    body.add_argument('--file')
    check = commands.add_parser('check', help='检查固定账号登录及可选姓名，不发送')
    for command in (progress, message, check):
        command.add_argument('--at', action='append', default=[], help='完整姓名；可重复或用逗号分隔')
    for command in (progress, message):
        command.add_argument('--dry-run', action='store_true', help='完全离线，只预览中文正文')
    return parser.parse_args(argv)
 def render(args, config, root):
    if args.command == 'progress' and bool(args.summary) == bool(args.file):
        raise NoticeError('进度摘要和 --file 必须且只能提供一项。')
    if args.file:
        path = Path(args.file).resolve()
        config_path = root / '.dingtalk' / 'config.env'
        if path == config_path.resolve() or (path.exists() and config_path.exists() and path.samefile(config_path)):
            raise NoticeError('不能把配置文件当作消息正文。')
        text = path.read_text(encoding='utf-8')
    else:
        text = ' '.join(args.summary) if args.command == 'progress' else args.text
    if not text.strip() or len(text.encode('utf-8')) > 12000:
        raise NoticeError('请提供不超过 12000 字节的简短中文摘要。')
    title = EVENTS[args.event] if args.command == 'progress' else args.title.strip()
    if not title or len(title) > 200 or '\n' in title or '\r' in title:
        raise NoticeError('请使用不超过 200 字的单行标题。')
    project = config.get('PROJECT') or root.name
    timestamp = dt.datetime.now().astimezone().strftime('%Y-%m-%d %H:%M %z')
    return f'## {project}｜{title}\n\n项目进度自动汇报 · {timestamp}\n\n{text}'
 def main(argv=None, root=ROOT, run=subprocess.run, sleep=time.sleep):
    try:
        encoded = os.environ.pop(PREFIX + 'ARGV_B64', '')
        if argv is None and encoded:
            argv = json.loads(base64.b64decode(encoded, validate=True).decode('utf-8'))
            if not isinstance(argv, list) or not all(isinstance(v, str) for v in argv):
                raise NoticeError('PowerShell 参数无效。')
        args = parse_args(argv)
        if args.command == 'init':
            initialize(args, root, run, sleep)
            return 0
        config = read_config(root)
-        preview = args.dry_run or config.get("DRY_RUN") == "1"
+        if args.command != 'check' and args.dry_run:
-        if not preview and config.get("ENABLED") != "1":
+            emit('preview', '仅离线预览，账号、群与 @ 效果未验证。', text=render(args, config, root), mentions=args.at)
            emit("skipped", "钉钉通知未配置或未启用，已跳过。")
            return 0
-        title, text = render(args, config, body_text(args, root))
+        if args.command != 'check' and config.get('ENABLED') != '1':
-        if preview:
+            emit('skipped', '通知尚未初始化或未启用；本次未发送，主任务继续。')
-            emit("preview", "仅离线预览，未验证群配置或提醒对象；未查人员、认证或发送，不输出配置中的密钥。",
+            return 0
-                 title=title, text=text,
+        if not config.get('PROFILE') or not config.get('TARGET'):
-                 mentions_requested=args.command == "message" and bool(args.at or args.at_id))
+            raise NoticeError('请由 Hub 先确认个人账号和项目群并执行 init。')
        text = render(args, config, root) if args.command != 'check' else None
        cli = dws_path()
        profile = config['PROFILE']
        if args.command == 'check':
            check_login(cli, profile, run, sleep)
        people = mentions(args.at, cli, profile, run, sleep)
        if args.command == 'check':
            emit('checked', '登录及所请求的姓名查询通过；未发送，未验证实际提醒效果。', names=list(people.values()))
            return 0
        if config.get("MODE", "app-bot") != "app-bot":
            raise NoticeError("当前封装仅支持 app-bot，请部署负责人检查配置。")
        if any(not config.get(key) for key in ("CLIENT_ID", "TARGET", "ROBOT_CODE")):
            raise NoticeError("机器人配置不完整，请部署负责人检查，不要在会话中粘贴配置。")
        cli = dws_path(config)
        secret = secret_for(config)
        people = mentions(args, config, cli, run, sleep) if args.command == "message" else {}
        if people:
-            text += "\n\n" + " ".join("@" + user_id for user_id in people)
+            text += '\n\n' + ' '.join(f'<@{person}>' for person in people)
-        message = {"title": title, "text": text}
+        command = [cli, 'chat', '+messages-send', '--as', 'user', '--group', config['TARGET'],
                   '--markdown', text, '--profile', profile, '--yes', '--format', 'json']
        if people:
-            message["atUserIds"] = list(people)
+            command += ['--at-open-dingtalk-ids', ','.join(people)]
-        payload = {
+        data = invoke(command, run, sleep, sending=True)
-            "msgParam": json.dumps(message, ensure_ascii=False), "msgKey": "sampleMarkdown",
+        emit('accepted', '个人 DWS 已接受发送请求，不代表已投递、已读或已触发接收端提醒。',
-            "openConversationId": (args.group if args.command == "message" else None) or config["TARGET"],
+             openTaskId=accepted_task(data), mentions=list(people.values()))
            "robotCode": config["ROBOT_CODE"],
        }
        # DWS accepts --data -: avoid shell quoting, argv length limits and
        # exposing message/target contents in the process command line.
        command = [cli, "api", "POST", "/v1.0/robot/groupMessages/send",
                   "--data", "-", "--yes", "--format", "json"]
        if config.get("PROFILE"):
            command += ["--profile", config["PROFILE"]]
        env = clean_env()
        env.update(DWS_CLIENT_ID=config["CLIENT_ID"], DWS_CLIENT_SECRET=secret)
        data = invoke(command, env, sending=True, run=run, sleep=sleep,
                      stdin_data=json.dumps(payload, ensure_ascii=False))
        emit("accepted", "钉钉服务端已接受通知，不代表群成员已读。", processQueryKey=accepted_key(data))
        return 0
    except NoticeError as error:
-        emit("error", str(error))
+        emit('error', str(error))
        return 1
-    except (OSError, UnicodeError, subprocess.SubprocessError, ValueError):
+    except (OSError, ValueError, UnicodeError, TypeError, subprocess.SubprocessError):
-        emit("error", "通知配置、正文文件或本机调用异常；请部署负责人检查，本次停止且不影响主任务。")
+        emit('error', '配置、正文或调用结果异常；本次停止，不自动重发，主任务继续。')
        return 1
-if __name__ == "__main__":
+if __name__ == '__main__':
    for stream in (sys.stdout, sys.stderr):
-        if hasattr(stream, "reconfigure"):
+        if hasattr(stream, 'reconfigure'):
-            stream.reconfigure(encoding="utf-8")
+            stream.reconfigure(encoding='utf-8')
    sys.exit(main())
@@ -1,4 +1,5 @@
-"""Offline tests: fake configs, fake credentials, fake DWS. Never use live config."""
+"""Offline regression tests. All DWS calls use isolated fake responses."""
 import base64
 import contextlib
 import io
 import json
@@ -9,314 +10,305 @@ import subprocess
 import sys
 import tempfile
 import unittest
-from unittest import mock
+from unittest.mock import patch
-import dingtalk_notify as notify
+import dingtalk_notify as n
-DNS = 'lookup api.dingtalk.com on 127.0.0.1:53: i/o timeout'
+PROFILE = 'fake-corp:fake-user'
-OK = '{"processQueryKey":"test-receipt"}'
+GROUP = 'fake-group'
 PERSON = 'fake-person'
 OK = {'ok': True, 'identity': 'user', 'tool': 'send_personal_message',
      'result': {'success': True, 'errorCode': None, 'result': {'openTaskId': 'fake-task'}}}
-def response(code=0, stdout=OK, stderr=''):
+def reply(data=None, code=0, error=''):
-    return subprocess.CompletedProcess(['fake-dws'], code, stdout, stderr)
+    return subprocess.CompletedProcess([], code, json.dumps(data or {}, ensure_ascii=False), error)
 class NotifyTests(unittest.TestCase):
    def setUp(self):
-        self.temp = tempfile.TemporaryDirectory(prefix='通知测试 with spaces ')
+        self.temp = tempfile.TemporaryDirectory()
        self.addCleanup(self.temp.cleanup)
        self.root = Path(self.temp.name)
-        (self.root / '.dingtalk').mkdir()
+        self.path = self.root / '.dingtalk' / 'config.env'
-        self.config = self.root / '.dingtalk/config.env'
+        self.path.parent.mkdir()
-        self.configure()
+        self.config()
-        self.environment = mock.patch.dict(os.environ, {'DING_SEC': 'ignored-env-secret',
+        self.calls = []
-                                                     'DWS_CLIENT_SECRET': 'ignored-dws-secret'}, clear=True)
+        self.sleeps = []
-        self.environment.start()
+        self.addCleanup(patch.stopall)
-        self.addCleanup(self.environment.stop)
+        patch.object(n, 'dws_path', return_value='fake-dws').start()
        self.locator = mock.patch.object(notify, 'dws_path', return_value='fake-dws')
        self.locator.start()
        self.addCleanup(self.locator.stop)
-    def configure(self, **updates):
+    def config(self, **changes):
-        values = dict(ENABLED='1', DRY_RUN='0', MODE='app-bot', CLIENT_ID='fake-client',
+        config = {'ENABLED': '1', 'PROFILE': PROFILE, 'TARGET': GROUP, 'PROJECT': '测试项目'}
-                      TARGET='fake-group', ROBOT_CODE='fake-robot', PROJECT='离线测试',
+        config.update(changes)
-                      PROFILE='fake-corp:fake-user', CLIENT_SECRET='fake-secret')
+        self.path.write_text(''.join(n.PREFIX + k + '=' + v + '\n' for k, v in config.items()), encoding='utf-8')
        values.update(updates)
        self.config.write_text('\n'.join(notify.PREFIX + k + '=' + v for k, v in values.items()),
                               encoding='utf-8')
        self.config.chmod(0o600)
-    def call(self, args, responses=None):
+    def fake(self, command, **kwargs):
-        run = mock.Mock(side_effect=responses or [response()])
+        self.calls.append((command, kwargs))
-        sleep = mock.Mock()
+        if command[1:3] == ['profile', 'list']:
-        output, errors = io.StringIO(), io.StringIO()
+            return reply({'profiles': [{'profile': PROFILE, 'userName': '测试人'}]})
-        with contextlib.redirect_stdout(output), contextlib.redirect_stderr(errors):
+        if command[1:3] == ['auth', 'status']:
-            code = notify.main(args, root=self.root, run=run, sleep=sleep)
+            return reply({'authenticated': True, 'token_valid': True})
-        return code, json.loads(output.getvalue()), run, sleep, errors.getvalue()
+        if command[1:3] == ['aisearch', 'person']:
-
+            return reply({'result': [{'openDingTalkId': PERSON, 'meta': {'name': command[command.index('--query') + 1]}}]})
-    def entries(self):
+        self.assertEqual(command[1:3], ['chat', '+messages-send'])
-        return [['progress', 'milestone', '**完成**\n\n- 内容'],
+        return reply(OK)
                ['message', '--title', '负责人提醒', '--text', '**请确认**\n\n- 内容']]
    def test_events(self):
        for event, label in notify.EVENTS.items():
            with self.subTest(event=event):
                code, data, run, _, _ = self.call(['progress', event, '中文摘要'])
                self.assertEqual((code, data['status']), (0, 'accepted'))
                payload = json.loads(run.call_args.kwargs['input'])
                self.assertIn(label, json.loads(payload['msgParam'])['title'])
    def test_payload_and_credentials_not_on_argv(self):
        text = '**结果**\n\n- "中文" 😀 `代码` $变量 & <内容> \\路径'
        code, data, run, _, _ = self.call(['message', '--title', '摘要', '--text', text])
        self.assertEqual(code, 0)
        command = run.call_args.args[0]
        kwargs = run.call_args.kwargs
        payload = json.loads(kwargs['input'])
        self.assertEqual(json.loads(payload['msgParam'])['text'], text)
        self.assertEqual(payload['openConversationId'], 'fake-group')
        self.assertEqual(payload['msgKey'], 'sampleMarkdown')
        self.assertEqual(command[command.index('--data') + 1], '-')
        self.assertNotIn('fake-secret', str(command))
        self.assertNotIn('fake-group', str(command))
        self.assertEqual(kwargs['env']['DWS_CLIENT_SECRET'], 'fake-secret')
        self.assertNotIn('DING_SEC', kwargs['env'])
        self.assertNotIn('fake-secret', json.dumps(data))
    def test_dns_retries_both_entries(self):
        for args in self.entries():
            with self.subTest(args=args):
                code, data, run, sleep, _ = self.call(args, [response(1, '', DNS)] * 3)
                self.assertEqual((code, data['status'], run.call_count), (1, 'error', 3))
                self.assertEqual(sleep.call_args_list, [mock.call(10), mock.call(10)])
    def test_transient_dns_success_stops(self):
        for args in self.entries():
            code, data, run, sleep, _ = self.call(args, [response(1, '', DNS), response()])
            self.assertEqual((code, data['status'], run.call_count, sleep.call_count), (0, 'accepted', 2, 1))
    def test_ambiguous_and_auth_errors_never_replayed_or_echoed(self):
        for args in self.entries():
            for error in ['i/o timeout', 'EOF', 'HTTP 503', 'permission denied', 'fake-secret App Token']:
                with self.subTest(error=error):
                    code, data, run, sleep, stderr = self.call(args, [response(1, '', error)])
                    self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
                    self.assertNotIn(error, json.dumps(data, ensure_ascii=False) + stderr)
    def test_receipt_wins_even_nonzero(self):
        for args in self.entries():
            code, data, run, sleep, _ = self.call(args, [response(1, OK, DNS)])
            self.assertEqual((code, data['status'], run.call_count, sleep.call_count), (0, 'accepted', 1, 0))
    def test_bad_receipts_do_not_replay(self):
        for raw in ['{}', '{"processQueryKey":" "}', '{"processQueryKey":123}', 'not json', '[]']:
            code, _, run, sleep, _ = self.call(self.entries()[0], [response(0, raw)])
            self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
    def test_timeout_no_replay(self):
        code, data, run, sleep, _ = self.call(self.entries()[0], [subprocess.TimeoutExpired('fake', 60)])
        self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
        self.assertIn('结果未知', data['message'])
        self.assertEqual(run.call_args.kwargs['timeout'], 60)
    def test_startup_failure_no_replay(self):
        code, _, run, sleep, _ = self.call(self.entries()[0], [OSError('private machine detail')])
        self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
    def test_missing_credentials_no_commands(self):
        self.configure(CLIENT_SECRET='')
        code, data, run, _, _ = self.call(self.entries()[0])
        self.assertEqual((code, run.call_count), (1, 0))
        self.assertIn('凭据', data['message'])
    def test_skipped_no_tools(self):
        self.configure(ENABLED='0')
        for args in self.entries():
            code, data, run, _, _ = self.call(args)
            self.assertEqual((code, data['status'], run.call_count), (0, 'skipped', 0))
        self.config.unlink()
        self.assertEqual(self.call(self.entries()[0])[1]['status'], 'skipped')
    def test_preview_never_uses_or_prints_secret_or_resolves_people(self):
        for configured in [False, True]:
            self.configure(DRY_RUN='1' if configured else '0')
            for args in self.entries():
                if args[0] == 'message':
                    args += ['--at', '完整姓名']
                if not configured:
                    args += ['--dry-run']
                with mock.patch.object(notify, 'secret_for', side_effect=AssertionError('secret accessed')):
                    code, data, run, _, _ = self.call(args)
                self.assertEqual((code, data['status'], run.call_count), (0, 'preview', 0))
                self.assertNotIn('fake-group', str(data))
                self.assertNotIn('fake-secret', str(data))
    def test_preview_without_secret(self):
        self.configure(CLIENT_SECRET='')
        code, data, run, _, _ = self.call(self.entries()[0] + ['--dry-run'])
        self.assertEqual((code, data['status'], run.call_count), (0, 'preview', 0))
    def test_config_cannot_be_used_as_body_even_in_preview(self):
        for args in [['progress', 'milestone'], ['message', '--title', '摘要']]:
            for flags in [[], ['--dry-run']]:
                code, data, run, _, _ = self.call(args + ['--file', str(self.config)] + flags)
                self.assertEqual((code, run.call_count), (1, 0))
                self.assertNotIn('fake-secret', str(data))
    def test_preview_without_config(self):
        self.config.unlink()
        self.assertEqual(self.call(self.entries()[0] + ['--dry-run'])[1]['status'], 'preview')
    def test_utf8_bom_crlf_literal_config(self):
        raw = self.config.read_text(encoding='utf-8') + '\nDINGTALK_PROGRESS_TITLE_PREFIX=$(not-executed)'
        self.config.write_bytes(('\ufeff' + raw.replace('\n', '\r\n')).encode('utf-8'))
        self.assertEqual(notify.read_config(self.root)['TITLE_PREFIX'], '$(not-executed)')
    def test_invalid_config_safe_error(self):
        for raw in ['DING_SEC=fake-secret', 'invalid fake-secret',
                    'DINGTALK_PROGRESS_ENABLED=1\nDINGTALK_PROGRESS_ENABLED=0',
                    'DINGTALK_PROGRESS_ENABLED=yes']:
            self.config.write_text(raw, encoding='utf-8')
            code, data, run, _, _ = self.call(self.entries()[0])
            self.assertEqual((code, run.call_count), (1, 0))
            self.assertNotIn('fake-secret', str(data))
    def test_body_files_both_entries(self):
        body = self.root / '中文 summary.md'
        body.write_bytes('\ufeff**结果**\r\n\r\n- 已确认 😀'.encode('utf-8'))
        for args in [['progress', 'milestone'], ['message', '--title', '摘要']]:
            code, _, run, _, _ = self.call(args + ['--file', str(body)])
            self.assertEqual(code, 0)
            text = json.loads(json.loads(run.call_args.kwargs['input'])['msgParam'])['text']
            self.assertIn('**结果**\n\n- 已确认 😀', text)
    def test_empty_or_large_body_fails_before_tools(self):
        for text in ['', ' ', '字' * 5000]:
            code, _, run, _, _ = self.call(['message', '--title', '摘要', '--text', text])
            self.assertEqual((code, run.call_count), (1, 0))
    def test_exact_name_and_dedup_same_profile(self):
        found = json.dumps({'result': [{'meta': {'name': '完整姓名'}, 'userId': 'fake-person'}]})
        args = self.entries()[1] + ['--at', '完整姓名，完整姓名', '--at-id', '完整姓名=fake-person']
        code, _, run, _, _ = self.call(args, [response(0, found), response()])
        self.assertEqual((code, run.call_count), (0, 2))
        for call in run.call_args_list:
            command = call.args[0]
            self.assertEqual(command[command.index('--profile') + 1], 'fake-corp:fake-user')
        self.assertNotIn('DWS_CLIENT_SECRET', run.call_args_list[0].kwargs['env'])
        message = json.loads(json.loads(run.call_args.kwargs['input'])['msgParam'])
        self.assertEqual(message['atUserIds'], ['fake-person'])
        self.assertTrue(message['text'].endswith('@fake-person'))
    def test_names_zero_multiple_malformed_stop_before_send(self):
        for rows in [[], [{'title': '完整姓名', 'userId': 'a'}, {'title': '完整姓名', 'userId': 'b'}],
                     [{'title': '别的姓名', 'userId': 'a'}], {}, [None]]:
            code, _, run, sleep, _ = self.call(self.entries()[1] + ['--at', '完整姓名'],
                                             [response(0, json.dumps({'result': rows}))])
            self.assertEqual((code, run.call_count, sleep.call_count), (1, 1, 0))
    def test_name_search_requires_explicit_profile(self):
        self.configure(PROFILE='')
        code, _, run, _, _ = self.call(self.entries()[1] + ['--at', '完整姓名'])
        self.assertEqual((code, run.call_count), (1, 0))
    def test_known_mapping_and_group_compatibility(self):
        code, _, run, _, _ = self.call(self.entries()[1] + ['--at-id', '姓名=fake-id', '--group', 'fake-other'])
        self.assertEqual(code, 0)
        self.assertEqual(json.loads(run.call_args.kwargs['input'])['openConversationId'], 'fake-other')
    def test_config_secret_is_only_source_on_all_platforms(self):
        for platform in ['darwin', 'win32', 'linux']:
            with self.subTest(platform=platform), mock.patch.object(sys, 'platform', platform):
                self.assertEqual(notify.secret_for({'CLIENT_SECRET': 'fake-secret'}), 'fake-secret')
                with self.assertRaises(notify.NoticeError):
                    notify.secret_for({})
    def test_secret_special_characters_are_literal(self):
        secret = 'fake=secret#literal$not-expanded'
        self.configure(CLIENT_SECRET=secret)
        code, data, run, _, _ = self.call(self.entries()[0])
        self.assertEqual(code, 0)
        self.assertEqual(run.call_args.kwargs['env']['DWS_CLIENT_SECRET'], secret)
        self.assertNotIn(secret, str(data))
        self.assertNotIn(secret, str(run.call_args.args))
        self.assertNotIn(secret, run.call_args.kwargs['input'])
 class LauncherTests(unittest.TestCase):
    def test_real_subprocess_stdin_without_live_dws(self):
        with tempfile.TemporaryDirectory(prefix='通知 fake transport ') as folder:
            root = Path(folder)
            (root / '.dingtalk').mkdir()
            config = root / '.dingtalk/config.env'
            config.write_text('DINGTALK_PROGRESS_ENABLED=1\nDINGTALK_PROGRESS_DRY_RUN=0\n'
                              'DINGTALK_PROGRESS_CLIENT_ID=fake-client\n'
                              'DINGTALK_PROGRESS_CLIENT_SECRET=fake-secret\n'
                              'DINGTALK_PROGRESS_TARGET=fake-group\n'
                              'DINGTALK_PROGRESS_ROBOT_CODE=fake-robot\n', encoding='utf-8')
            config.chmod(0o600)
            fake = root / 'fake dws.py'
            fake.write_text(
                'import json, os, sys\n'
                'assert sys.argv[1:] == ["api", "POST", "/v1.0/robot/groupMessages/send", '
                '"--data", "-", "--yes", "--format", "json"]\n'
                'payload = json.loads(sys.stdin.buffer.read().decode("utf-8"))\n'
                'assert payload["openConversationId"] == "fake-group"\n'
                'assert json.loads(payload["msgParam"])["text"] == "中文\\n\\n- \\\"引号\\\" 😀"\n'
                'assert os.environ["DWS_CLIENT_SECRET"] == "fake-secret"\n'
                'assert "DING_SEC" not in os.environ\n'
                'print(json.dumps({"processQueryKey": "fake-transport-receipt"}))\n', encoding='utf-8')
            def transport(command, **kwargs):
                return subprocess.run([sys.executable, str(fake), *command[1:]], **kwargs)
    def execute(self, args, run=None):
        out = io.StringIO()
-            with mock.patch.object(notify, 'dws_path', return_value='fake-dws'), \
+        with contextlib.redirect_stdout(out):
-                    mock.patch.dict(os.environ, {'DING_SEC': 'ignored-env-secret'}), \
+            code = n.main(args, root=self.root, run=run or self.fake, sleep=self.sleeps.append)
-                    contextlib.redirect_stdout(out):
+        return code, json.loads(out.getvalue())
                code = notify.main(['message', '--title', '摘要', '--text', '中文\n\n- "引号" 😀'],
                                   root=root, run=transport)
            self.assertEqual(code, 0, out.getvalue())
            self.assertEqual(json.loads(out.getvalue())['status'], 'accepted')
-    def test_real_launchers_preview_in_isolated_copy(self):
+    def test_each_event_personal_markdown(self):
-        """Exercise OS processes/quoting, but copy no real config and never run dws."""
+        for event, label in n.EVENTS.items():
-        with tempfile.TemporaryDirectory(prefix='通知 launch with spaces ') as folder:
+            code, result = self.execute(['progress', event, '**进展**\n\n- 已归档'])
-            scripts = Path(folder) / 'scripts'
+            self.assertEqual(code, 0)
-            scripts.mkdir()
+            self.assertEqual(result['status'], 'accepted')
-            for name in ['dingtalk_notify.py', 'dingtalk-progress', 'dingtalk-bot-message',
+            cmd = self.calls[-1][0]
-                         'dingtalk-progress.ps1', 'dingtalk-bot-message.ps1']:
+            self.assertEqual(cmd[cmd.index('--as') + 1], 'user')
-                shutil.copy2(Path(__file__).parent / name, scripts / name)
+            self.assertEqual(cmd[cmd.index('--group') + 1], GROUP)
-            text = '**结果**\n\n- "中文" 😀 `代码` $变量 & 内容'
+            self.assertEqual(cmd[cmd.index('--profile') + 1], PROFILE)
-            body = Path(folder) / '中文正文.md'
+            body = cmd[cmd.index('--markdown') + 1]
-            body.write_text(text, encoding='utf-8')
+            self.assertIn(label, body)
-            env = {k: v for k, v in os.environ.items()
+            self.assertIn('项目进度自动汇报', body)
-                   if not k.startswith(('DING', 'DWS_'))}
+            self.assertIn('**进展**\n\n- 已归档', body)
-            # Run from outside the script project: --file is cwd-relative.
+            self.assertNotIn('--ai-tag=false', cmd)
-            commands = [[sys.executable, str(scripts / 'dingtalk_notify.py'), 'progress', 'milestone',
+            self.assertNotIn('--robot-code', cmd)
-                         '--file', body.name, '--dry-run']]
+
    def test_message_mentions_exact_name_fixed_profile(self):
        code, data = self.execute(['message', '--title', '提醒', '--text', '请补充资料', '--at', '测试人,测试人'])
        self.assertEqual(code, 0)
        self.assertEqual(len(self.calls), 2)
        for cmd, _ in self.calls:
            self.assertEqual(cmd[cmd.index('--profile') + 1], PROFILE)
        cmd = self.calls[-1][0]
        self.assertEqual(cmd[cmd.index('--at-open-dingtalk-ids') + 1], PERSON)
        self.assertIn('<@fake-person>', cmd[cmd.index('--markdown') + 1])
        self.assertEqual(data['mentions'], ['测试人'])
        self.assertEqual(data['openTaskId'], 'fake-task')
    def test_progress_can_include_mention_once(self):
        code, _ = self.execute(['progress', 'blocked', '缺输入', '--at', '测试人'])
        self.assertEqual(code, 0)
        self.assertEqual(sum(c[0][1:3] == ['chat', '+messages-send'] for c in self.calls), 1)
    def test_lookup_ambiguity_no_match_and_malformed_never_send(self):
        for rows in [[], [{'openDingTalkId': 'a', 'meta': {'name': '甲'}},
                          {'openDingTalkId': 'b', 'meta': {'name': '甲'}}],
                     [{'userId': 'wrong-id-kind', 'meta': {'name': '甲'}}], None]:
            self.calls.clear()
            def lookup(command, **kwargs):
                self.calls.append(command)
                return reply({'result': rows})
            code, data = self.execute(['message', '--title', '提醒', '--text', '正文', '--at', '甲'], lookup)
            self.assertEqual(code, 1)
            self.assertEqual(data['status'], 'error')
            self.assertEqual(len(self.calls), 1)
    def test_check_is_read_only(self):
        code, data = self.execute(['check', '--at', '测试人'])
        self.assertEqual(code, 0)
        self.assertEqual(data['status'], 'checked')
        self.assertEqual(len(self.calls), 2)
        self.assertNotIn('fake-person', json.dumps(data))
    def test_check_bad_login_stops(self):
        code, _ = self.execute(['check'], lambda *a, **kw: reply({'authenticated': False, 'token_valid': False}))
        self.assertEqual(code, 1)
    def test_disabled_and_missing_config_no_tools(self):
        self.config(ENABLED='0')
        self.assertEqual(self.execute(['progress', 'start', '开始'])[1]['status'], 'skipped')
        self.path.unlink()
        self.assertEqual(self.execute(['progress', 'start', '开始'])[1]['status'], 'skipped')
        self.assertFalse(self.calls)
    def test_preview_without_config_or_dws(self):
        self.path.unlink()
        code, result = self.execute(['progress', 'blocked', '说明', '--at', '甲', '--dry-run'])
        self.assertEqual(code, 0)
        self.assertEqual(result['status'], 'preview')
        self.assertFalse(self.calls)
    def test_legacy_and_invalid_config_fail_without_echo(self):
        for suffix in ['CLIENT_SECRET=fake-secret', 'MODE=app-bot', 'ENABLED=1', 'UNKNOWN=x']:
            self.config()
            with self.path.open('a', encoding='utf-8') as f:
                f.write(n.PREFIX + suffix + '\n')
            code, data = self.execute(['progress', 'start', '开始'])
            self.assertEqual(code, 1)
            self.assertNotIn('fake-secret', json.dumps(data))
            self.assertFalse(self.calls)
    def test_config_bom_crlf_and_literal_values(self):
        raw = self.path.read_text(encoding='utf-8').replace('测试项目', '中文 "引号" $()')
        self.path.write_bytes(b'\xef\xbb\xbf' + raw.replace('\n', '\r\n').encode())
        code, data = self.execute(['progress', 'start', '开始', '--dry-run'])
        self.assertEqual(code, 0)
        self.assertIn('中文 "引号" $()', data['text'])
    def test_files_and_invalid_bodies(self):
        body = self.root / '正文.md'
        body.write_text('**进展**\n\n- 中文 "引号" $()', encoding='utf-8')
        for args in [['progress', 'milestone'], ['message', '--title', '说明']]:
            self.assertEqual(self.execute(args + ['--file', str(body)])[0], 0)
        before = len(self.calls)
        for args in [['progress', 'start', ''], ['progress', 'start', 'x' * 12001],
                     ['progress', 'start', '正文', '--file', str(body)],
                     ['message', '--title', '坏\n标题', '--text', '内容'],
                     ['progress', 'start', '--file', str(self.path), '--dry-run']]:
            self.assertEqual(self.execute(args)[0], 1)
        self.assertEqual(len(self.calls), before)
    def test_dns_three_attempts_only(self):
        calls = []
        def failed(*a, **kw):
            calls.append(a)
            return reply(code=1, error='lookup api.dingtalk.com: no such host')
        code, _ = self.execute(['progress', 'start', '开始'], failed)
        self.assertEqual(code, 1)
        self.assertEqual(len(calls), 3)
        self.assertEqual(self.sleeps, [10, 10])
    def test_dns_then_success(self):
        responses = iter([reply(code=1, error='lookup api.dingtalk.com: no such host'), reply(OK)])
        self.assertEqual(self.execute(['progress', 'start', '开始'], lambda *a, **kw: next(responses))[0], 0)
        self.assertEqual(self.sleeps, [10])
    def test_failure_and_unknown_never_retry(self):
        cases = [reply(code=1, error='fake-secret connection reset'),
                 reply(code=1, error=json.dumps({'error': {'category': 'auth', 'message': 'fake-secret'}})),
                 reply(code=1, error=json.dumps({'error': {'retryable': False, 'message': 'lookup api.dingtalk.com: no such host'}})),
                 reply({'success': True}), reply({'ok': False}), reply(OK | {'partial': True})]
        for result in cases:
            count = []
            def failed(*a, **kw):
                count.append(1)
                return result
            code, data = self.execute(['progress', 'start', '开始'], failed)
            self.assertEqual(code, 1)
            self.assertEqual(len(count), 1)
            self.assertNotIn('fake-secret', json.dumps(data))
        self.assertFalse(self.sleeps)
    def test_receipt_wins_nonzero_exit(self):
        self.assertEqual(self.execute(['progress', 'start', '开始'], lambda *a, **kw: reply(OK, code=1))[0], 0)
    def test_wrong_receipts_rejected(self):
        for bad in [OK | {'identity': 'bot'}, OK | {'tool': 'other'}, OK | {'failures': [1]},
                    OK | {'result': {'success': False, 'result': {'openTaskId': 'x'}}},
                    OK | {'result': {'success': True, 'result': {'openTaskId': ''}}}]:
            self.assertIsNone(n.accepted_task(bad))
    def test_timeout_and_launch_error_no_retry(self):
        for error in [subprocess.TimeoutExpired('fake', 60), OSError('fake-secret')]:
            calls = []
            def fail(*a, **kw):
                calls.append(1)
                raise error
            code, _ = self.execute(['progress', 'start', '开始'], fail)
            self.assertEqual(code, 1)
            self.assertEqual(len(calls), 1)
    def test_no_app_credentials_in_child_environment(self):
        with patch.dict(os.environ, {'DWS_CLIENT_SECRET': 'fake-secret', 'DWS_CLIENT_ID': 'fake-app', 'DING_SEC': 'fake-key'}):
            self.execute(['progress', 'start', '开始'])
        env = self.calls[-1][1]['env']
        self.assertNotIn('DWS_CLIENT_SECRET', env)
        self.assertNotIn('DWS_CLIENT_ID', env)
        self.assertNotIn('DING_SEC', env)
        self.assertEqual(self.calls[-1][1]['timeout'], 60)
    def init_args(self):
        return ['init', '--profile', PROFILE, '--group', GROUP, '--project', '项目甲']
    def test_init_requires_confirmation_and_explicit_replace(self):
        original = self.path.read_bytes()
        self.assertEqual(self.execute(self.init_args())[0], 1)
        self.assertEqual(self.execute(self.init_args() + ['--yes'])[0], 1)
        self.assertEqual(self.path.read_bytes(), original)
        self.assertFalse(self.calls)
    def test_init_creates_minimal_config_no_send(self):
        self.path.unlink()
        code, data = self.execute(self.init_args() + ['--yes'])
        self.assertEqual(code, 0)
        self.assertEqual(data['status'], 'initialized')
        self.assertEqual(set(n.read_config(self.root)), n.KEYS)
        self.assertNotIn(b'\r', self.path.read_bytes())
        self.assertEqual(len(self.calls), 2)
        if os.name != 'nt':
            self.assertEqual(self.path.stat().st_mode & 0o777, 0o600)
    def test_init_replaces_legacy_without_retaining_secret(self):
        self.path.write_text('DINGTALK_PROGRESS_CLIENT_SECRET=fake-secret\n', encoding='utf-8')
        code, data = self.execute(self.init_args() + ['--yes', '--replace'])
        self.assertEqual(code, 0)
        self.assertNotIn('fake-secret', self.path.read_text())
        self.assertNotIn('fake-secret', json.dumps(data))
        self.assertEqual(len(list(self.path.parent.iterdir())), 1)
    def test_init_bad_profile_and_line_injection_leave_config(self):
        before = self.path.read_bytes()
        for args in [self.init_args() + ['--profile', 'other:user'],
                     self.init_args() + ['--project', 'x\nBAD=1']]:
            self.assertEqual(self.execute(args + ['--yes', '--replace'])[0], 1)
            self.assertEqual(self.path.read_bytes(), before)
    def test_init_failed_auth_leaves_old_config(self):
        before = self.path.read_bytes()
        def fake(command, **kw):
            if command[1] == 'auth':
                return reply({'authenticated': False})
            return self.fake(command, **kw)
        self.assertEqual(self.execute(self.init_args() + ['--yes', '--replace'], fake)[0], 1)
        self.assertEqual(self.path.read_bytes(), before)
    def test_encoded_powershell_arguments(self):
        args = ['message', '--title', '中文 "标题"', '--text', '**进展**\n\n- 引号 "$()"', '--dry-run']
        with patch.dict(os.environ, {n.PREFIX + 'ARGV_B64': base64.b64encode(json.dumps(args).encode()).decode()}):
            code, data = self.execute(None)
        self.assertEqual(code, 0)
        self.assertIn('中文 "标题"', data['text'])
        self.assertIn('**进展**\n\n- 引号 "$()"', data['text'])
    def test_bad_encoded_arguments(self):
        for value in ['!!!', base64.b64encode(b'{}').decode()]:
            with patch.dict(os.environ, {n.PREFIX + 'ARGV_B64': value}):
                self.assertEqual(self.execute(None)[0], 1)
        self.assertFalse(self.calls)
    def test_hardlinked_config_cannot_be_message_body(self):
        other = self.root / 'not-a-summary.md'
        os.link(self.path, other)
        self.assertEqual(self.execute(['progress', 'start', '--file', str(other), '--dry-run'])[0], 1)
    def test_real_powershell_launcher_in_isolated_project(self):
        pwsh = shutil.which('pwsh')
-            if os.name == 'nt':
+        if not pwsh or 'windowsapps' in sys.executable.lower():
-                self.assertIsNotNone(pwsh, 'Windows launcher test needs PowerShell 7.3+')
+            self.skipTest('需要 PowerShell 和独立 Python，不能用商店版作启动器验收')
-            if pwsh:
+        folder = self.root / 'scripts'
-                commands += [[pwsh, '-NoProfile', '-File', str(scripts / 'dingtalk-progress.ps1'),
+        folder.mkdir()
-                              'milestone', text, '--dry-run'],
+        source = Path(n.__file__).parent
-                             [pwsh, '-NoProfile', '-File', str(scripts / 'dingtalk-bot-message.ps1'),
+        for name in ['dingtalk_notify.py', 'dingtalk-python.ps1', 'dingtalk-progress.ps1', 'dingtalk-message.ps1']:
-                              '--title', '中文摘要', '--text', text, '--dry-run']]
+            shutil.copyfile(source / name, folder / name)
-            bash = Path('C:/Program Files/Git/bin/bash.exe') if os.name == 'nt' else shutil.which('bash')
+        env = n.clean_env()
-            if bash and Path(bash).exists():
+        env['DINGTALK_PYTHON'] = sys.executable
-                for name, args in [('dingtalk-progress', ['milestone']),
+        text = '**中文**\n\n- "引号" $() `原样`'
-                                   ('dingtalk-bot-message', ['--title', '中文摘要'])]:
+        for script, args in [('dingtalk-progress.ps1', ['milestone', text]),
-                    commands.append([str(bash), str(scripts / name), *args,
+                             ('dingtalk-message.ps1', ['--title', '中文 "标题"', '--text', text])]:
-                                     '--file', body.name, '--dry-run'])
+            result = subprocess.run([pwsh, '-NoProfile', '-File', str(folder / script), *args, '--dry-run'],
-            for command in commands:
+                                    cwd=self.root, env=env, capture_output=True, encoding='utf-8', timeout=30)
                with self.subTest(entry=command[:3]):
                    result = subprocess.run(command, cwd=folder, env=env, capture_output=True,
                                            text=True, encoding='utf-8', timeout=30)
            self.assertEqual(result.returncode, 0, result.stderr)
            data = json.loads(result.stdout)
            self.assertEqual(data['status'], 'preview')
            self.assertIn(text, data['text'])
        result = subprocess.run([pwsh, '-NoProfile', '-File', str(folder / 'dingtalk-python.ps1'), 'init',
                                 '--profile', PROFILE, '--group', GROUP, '--project', '甲'],
                                cwd=self.root, env=env, capture_output=True, encoding='utf-8', timeout=30)
        self.assertEqual(result.returncode, 1)
        self.assertIn('先向负责人确认', result.stdout)
 if __name__ == '__main__':
Author	SHA1	Message	Date
chenmingxuan	1d85de4c7c	release: v1.0.4 支持消息附加说明与分行表达	2026-09-15 17:33:19 +08:00
chenmingxuan	7b22e2802f	docs: 精简 README 并移除版本升级记录	2026-09-11 10:38:22 +08:00
chenmingxuan	20c0de9b38	release: v1.0.3 统一个人 DWS 通知与简化初始化配置	2026-09-10 17:14:23 +08:00