Task Watcher Skill — Task Watcher 技能

task-watcher — A同步 Task 监控ing & Callback

Version: 1.2.0 Author: sly (OpenClaw) License: MIT 安装: ClawHub 安装 task-watcher

监控 long-运行ning a同步 tasks (content review, CI/CD, 部署ments) and 获取 Discord/Telegram 通知 when 状态s change.

What It Does

Register tasks → Watcher polls for 状态 changes → 发送s 通知 on change/completion.

Perfect for:

Content publishing: 监控 XHS/社媒 review 状态 after posting CI/CD: 追踪 GitHub PR merges, build completions 部署ments: Watch rollout 状态 Any a同步 工作流: Anything that takes time and you want to know when it's done Architecture 代理 registers task → tasks.jsonl (分享d-上下文) ↓ Cron (/3 ) → watcher.py → Adapter 检查s 状态 → 状态 changed? ↓ Yes Policy decides → Notifier 发送s Discord

插件 architecture:

Adapters: Pluggable 状态 检查ers. Built-in: XHS, GitHub PR, Cron Job Notifiers: Discord (via OpenClaw 代理 --deliver), Telegram, 会话, File Policies: Notification frequency, escalation, retry 记录ic Stores: JSONL-based persistence with file locking 保护ion: expires_at timeout (default 6h), 3x delivery 失败 escalation Quick 启动

• 安装

ClawHub 安装 task-watcher

• Register a 监控ing task

导入 sys, os sys.path.insert(0, os.path.expanduser("~/.OpenClaw/workspace/技能s/task-watcher/scripts/lib"))

from 模型s 导入 CallbackTask from stores 导入 JsonlTaskStore

store = JsonlTaskStore("~/.OpenClaw/分享d-上下文/监控-tasks/tasks.jsonl") task = CallbackTask( task_id="tsk_my_task_001", owner_代理="content", tar获取_系统="xiaohongshu", tar获取_object_id="note_id_here", reply_channel="discord", reply_to="channel:YOUR_CHANNEL_ID", current_状态="submitted", expires_at="2026-03-08T12:00:00", # 6h timeout recommended ) store.创建(task)

Or via 命令行工具:

cd ~/.OpenClaw/workspace/技能s/task-watcher python3 scripts/register_task.py \ --task-id tsk_my_001 \ --系统 xiaohongshu \ --object-id note_abc123 \ --reply-to channel:1234567890 \ --expires-hours 6

• 运行 watcher (cron)

# 添加 to crontab - every 3 minutes /3 cd ~/.OpenClaw/workspace/技能s/task-watcher && \ python3 scripts/watcher.py --once >> ~/.OpenClaw/记录s/watcher.记录 2>&1

• Custom adapter

from adapters 导入 状态Adapter, 状态结果

class MyAdapter(状态Adapter): @property def name(self): return "my-系统" def supports(self, task): return task.tar获取_系统 == "my-系统" def 健康_检查(self): return True def 检查(self, task): 状态 = 检查_my_系统(task.tar获取_object_id) return 状态结果(状态=状态, terminal=(状态 == "done"))

File Structure task-watcher/ ├── 技能.md # This file ├── _meta.json # ClawHub metadata ├── scripts/ │ ├── watcher.py # Cron entry point │ ├── register_task.py # 命令行工具 task registration │ └── lib/ │ ├── __init__.py # Package 导出s │ ├── 模型s.py # Data 模型s (CallbackTask, 状态结果, 发送结果) │ ├── stores.py # JSONL TaskStore with file locking │ ├── adapters.py # 状态 adapters (XHS, GitHub, Cron) │ ├── notifiers.py # Discord/Telegram/File notifiers │ ├── policies.py # Notification policies │ └── bus.py # WatcherBus 编排器 └── tests/ ├── test_模型s.py ├── test_stores.py ├── test_adapters.py ├── test_notifiers.py ├── test_policies.py └── test_bus.py

Data Paths File Purpose ~/.OpenClaw/分享d-上下文/监控-tasks/tasks.jsonl Active tasks ~/.OpenClaw/分享d-上下文/监控-tasks/审计.记录 审计 trAIl ~/.OpenClaw/分享d-上下文/监控-tasks/通知/ Notification files Built-in Adapters Adapter 系统 How It 检查s xiaohongshu-note-review XHS content review Mock file / packet file github-pr-状态 GitHub PR merge Skeleton (extend for your use) cron-job-completion Cron job 追踪ing File-based 状态 检查 Requirements Python 3.10+ OpenClaw 2026.3+ (for OpenClaw 代理 --deliver) No external dependencies (stdlib only)