jianf/oniagent
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat: priority system, batch tasks, action feedback, event daemon

Browse Source 
- Priority system: GET /api/state/priorities, POST /api/action/priority_global, POST /api/action/priority_type
- Batch system: POST /api/action/batch with per-action result tracking
- Enhanced action feedback with error codes and AI suggestions:
  cell_occupied, cell_solid, cell_occupied_by_dupe, material_shortage, etc.
- Event stream: GET /api/state/events?since=&limit= for incremental polling
- Event daemon: scripts/event_daemon.py continuously polls events,
  classifies by severity, auto-triggers analysis on critical events
- Games state now reports suffocating/starving/stressed counts
- Cell data includes isDiggable and isSafeForDupe flags
- Added 'events', 'queue', 'batch', 'priority_global', 'priority_type' CLI commands
- Added sample batch plan file
This commit is contained in:
root
2026-05-22 09:01:12 +08:00
parent 4c6a9b882f
commit a4f2062c1a
6 changed files with 1350 additions and 133 deletions
Download Patch File Download Diff File Expand all files Collapse all files

42
README.md
View File

@ -99,6 +99,13 @@ python3 tools/oni_api.py prioritize 15 12 9 # 优先
python3 tools/oni_api.py research_select ImprovedOxygen # 科研
python3 tools/oni_api.py mop 10 10 # 清理液体
python3 tools/oni_api.py harvest 20 15 # 收获植物
# 批量任务
python3 tools/oni_api.py batch docs/batch_example.json # 执行批量建造计划
# 优先级管理
python3 tools/oni_api.py priority_global dig 7 # 全局挖掘优先级设为 7
python3 tools/oni_api.py priority_type Electrolyzer 9 # 电解器建造优先级设为 9
```
### `tools/oni_analyzer.py` — 智能分析器
@ -138,6 +145,27 @@ python3 tools/oni_builder.py build spom 42 42 # 建造 SPOM
| `cooling` | 蒸汽涡轮冷却模块 | 8×6 |
| `bedroom` | 标准卧室（床+梯子床+装饰） | 8×4 |
### `scripts/event_daemon.py` — 事件守护进程（AI 输入源）
持续轮询游戏事件并将其注入 AI 输入流。这是 AI 感知游戏状态变化的实时通道。
```bash
python3 scripts/event_daemon.py
```
工作原理：
1. 每 5 秒轮询 `GET /api/state/events?since=<seq>` 获取新事件
2. 对事件分类（critical / warning / info）
3. critical 事件 → 红色告警 + **自动触发全量游戏快照**（周期/窒息/饥饿/压力）
4. warning 事件 → 结构化输出给 AI
5. 维护滚动事件历史（最多 200 条），AI 可随时查询摘要
事件类型：
- `critical` — 复制人窒息、建筑损坏、电力中断 → 立即触发分析
- `warning` — 低氧、食物短缺、高温 → 主动通知给 AI
- `action_feedback` — 建造/挖掘等操作的结果反馈
- `info` — 常规游戏状态变化
### 辅助脚本
```bash
@ -145,6 +173,7 @@ bash scripts/auto_repair.sh # 诊断 Mod 连接
bash scripts/auto_analyze.sh # 一键健康检查+状态+分析
bash scripts/watch.sh 60 # 每 60 秒持续监控
bash scripts/setup.sh # 环境初始化
python3 scripts/event_daemon.py # 事件守护进程（AI 输入源）
```
## Mod API 完整端点
@ -164,6 +193,9 @@ bash scripts/setup.sh # 环境初始化
| `/api/state/critters` | 小动物（位置/种类/幸福度） |
| `/api/state/plants` | 植物（位置/生长进度/是否枯萎） |
| `/api/state/rooms` | 房间（类型/格数/建筑数） |
| `/api/state/queue` | 任务队列（查看待处理任务） |
| `/api/state/events?since=&limit=` | 事件流（AI 轮询增量事件） |
| `/api/state/priorities` | 优先级配置（全局/建筑/复制人） |
### 地图/格子数据 (GET)
@ -181,6 +213,7 @@ bash scripts/setup.sh # 环境初始化
| `/api/registry/buildings` | 全部建筑定义（尺寸/功耗/材料） |
| `/api/registry/elements` | 全部元素定义（比热容/熔沸点/导热） |
| `/api/registry/techs` | 全部科技定义（前置/解锁） |
| `/api/registry/priorities` | 优先级级别含义对照表 |
### 操作 (POST)
@ -195,6 +228,9 @@ bash scripts/setup.sh # 环境初始化
| `/api/action/harvest` | `{x, y}` |
| `/api/action/schedule` | `{duplicantId, schedule}` |
| `/api/action/wardrobe` | `{duplicantId, equipment}` |
| `/api/action/batch` | `{actions: [{type, ...}]}` — 批量执行 |
| `/api/action/priority_global` | `{target, priority}` — 全局默认优先级 |
| `/api/action/priority_type` | `{buildingType, priority}` — 按建筑类型设优先级 |
## 项目结构
@ -217,11 +253,13 @@ oni-agent/
│ ├── setup.sh # 环境初始化
│ ├── auto_repair.sh # 连接诊断
│ ├── auto_analyze.sh # 一键分析
│ └── watch.sh # 持续监控模式
│ ├── watch.sh # 持续监控模式
│ └── event_daemon.py # 事件守护进程（AI 实时输入源）
│
├── docs/
│ ├── MOD_DEV_GUIDE.md # Mod 开发规范与约束
│ └── AI_KNOWLEDGE_BASE.md # 200+ 建筑/元素/科技 ID 知识库
│ ├── AI_KNOWLEDGE_BASE.md # 200+ 建筑/元素/科技 ID 知识库
│ └── batch_example.json # 批量任务示例文件
│
└── skills/
└── oni_agent.md # Agent skill 定义

108
SKILL.md
View File

@ -234,6 +234,110 @@ python3 tools/oni_builder.py build spom 42 42
---
## AI 事件驱动工作流
AI 应持续运行事件守护进程，形成"事件 → 分析 → 操作 → 反馈"的闭环：
```
┌───────────────────────────────────┐
│ Event Daemon │
│ (scripts/event_daemon.py) │
│ polls every 5 seconds │
└──────────┬────────────────────────┘
│ 新事件
▼
┌───────────────────────────────────┐
│ AI Decision Loop │
│ │
│ 1. 收到事件 → 分类严重程度 │
│ 2. 严重 → 立即用 tools 调查状态 │
│ 3. 分析根本原因 │
│ 4. 执行操作（dig/build/batch） │
│ 5. 检查操作反馈（success/fail） │
│ 6. 失败 → 读取错误原因 + 建议 │
│ 7. 调整方案后重试 │
└───────────────────────────────────┘
```
### 事件驱动示例：复制人窒息
```
[EVENT CRITICAL] Cycle 42 @ 14:32:15
Title: suffocating
Entity: Dup1
→ AI 收到此事件后自动执行：
1. python3 tools/oni_api.py duplicants # 查看所有复制人氧气值
2. python3 tools/oni_api.py cell 23 45 # 查看 Dup1 所在格子
3. python3 tools/oni_api.py resources # 检查 O2 + Algae 存量
4. python3 tools/oni_api.py buildings # 是否有电解器/扩散器
5. 根据分析结果：
- 如果无电解器且 Algae < 1t → 紧急建造 SPOM
- 如果有扩散器但无 Algae → 改用电解器
- 如果 Dup1 在 CO2 里 → 挖掘排气通道
6. python3 tools/oni_api.py build Electrolyzer 42 42
7. 读取反馈：成功？材料不足？格子被占？
```
### 操作反馈处理
每次操作后 AI **必须**检查反馈中的 `success` 字段：
```json
// 成功
{ "success": true, "result": "build_queued", "buildingId": "Electrolyzer" }
// 失败 — AI 必须读取 error, errorMessage, suggestion
{
"success": false,
"result": "failed",
"error": "cell_occupied",
"errorMessage": "Cell (42,42) already has building 'GasPump'",
"suggestion": "Choose a different location, or deconstruct the existing building first"
}
```
常见错误码：
| 错误 | 含义 | AI 应如何处理 |
|------|------|-------------|
| `cell_occupied` | 格子已被建筑占据 | 换位置或先拆除 |
| `cell_solid` | 格子是固体方块（未挖掘） | 先 dig 再 build |
| `cell_occupied_by_dupe` | 复制人站在那 | 等待或取消其任务 |
| `material_shortage` | 建造材料不足 | 检查资源并安排生产 |
| `cell_out_of_bounds` | 超出地图范围 | 调整坐标 |
| `unknown_building` | buildingId 错误 | 查询 registry buildings |
| `missing_prerequisites` | 科技未研究 | 先研究前置科技 |
| `no_liquid_at_cell` | 没有液体可清理 | 用 cell 命令检查 |
| `invalid_priority` | 优先级必须是 1-9 | 调整数字 |
### 批量任务示例
AI 可以通过批处理一次性执行一个复杂的建造计划：
```bash
# 1. 查看批量计划内容
cat docs/batch_example.json
# 2. 执行批量计划
python3 tools/oni_api.py batch docs/batch_example.json
```
批量反馈会逐个报告每个动作的结果，AI 应遍历并处理失败项。
### 优先级系统
ONI 优先级范围 1（最低）~ 9（紧急/黄 alert）：
```bash
# 设置全局默认
python3 tools/oni_api.py priority_global dig 9
# 查看优先级含义
python3 tools/oni_api.py registry priorities
```
---
## AI 如何表达"在哪个格子做什么"
### 定位语法
@ -270,14 +374,16 @@ AI 在描述操作时应使用以下格式：
| 工具 | 用途 |
|------|------|
| `tools/oni_api.py` | 与 Mod HTTP API 通信（所有查询/操作） |
| `tools/oni_api.py` | Mod API 客户端（状态/格子/注册表/操作/批量/优先级/事件） |
| `tools/oni_analyzer.py` | 自动分析游戏状态、生成预警和建议 |
| `tools/oni_builder.py` | 预置蓝图建造（SPOM/农场/养殖等） |
| `scripts/auto_repair.sh` | 诊断 Mod 连接问题 |
| `scripts/auto_analyze.sh` | 一键健康检查+状态+分析 |
| `scripts/watch.sh [秒]` | 循环监控模式 |
| `scripts/setup.sh` | 环境初始化与检查 |
| `scripts/event_daemon.py` | **事件守护进程** — 持续轮询事件 → AI 输入流 |
| `docs/AI_KNOWLEDGE_BASE.md` | 建筑/元素/科技 ID 注册表和游戏机制参考 |
| `docs/batch_example.json` | 批量任务示例文件 |
---

40
docs/batch_example.json Normal file
View File

@ -0,0 +1,40 @@
{
"name": "Build SPOM - Step 1: Dig and Electrolyzer",
"actions": [
{
"type": "dig",
"x": 42,
"y": 40,
"width": 8,
"height": 6
},
{
"type": "build",
"buildingId": "Electrolyzer",
"x": 45,
"y": 42
},
{
"type": "build",
"buildingId": "GasPump",
"x": 43,
"y": 42
},
{
"type": "build",
"buildingId": "HydrogenGenerator",
"x": 45,
"y": 40
},
{
"type": "wait",
"delayMs": 100
},
{
"type": "priority",
"x": 45,
"y": 42,
"priority": 9
}
]
}

863
mod/ONIAgentBridge.cs
View File

File diff suppressed because it is too large Load Diff

273
scripts/event_daemon.py Executable file
View File

@ -0,0 +1,273 @@
#!/usr/bin/env python3
"""
ONI Agent Event Daemon
======================
Continuous event poller that feeds game events to the AI's input stream.
Architecture:
Game Mod --> Event Queue (via HTTP) --> Event Daemon --> AI Input Stream
The daemon:
1. Polls GET /api/state/events?since=<seq> every N seconds
2. Classifies events by severity (critical/warning/info)
3. For critical events: immediately triggers full analysis + prints alert
4. For warning events: logs and optionally triggers targeted checks
5. For info events: accumulates and reports periodically
6. Maintains a compact event log for AI context
"""
import json
import os
import sys
import time
import datetime
# Add tools to path
TOOLS_DIR = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'tools')
sys.path.insert(0, TOOLS_DIR)
from oni_api import api_get, api_post, api_url
# ── Configuration ──────────────────────────────────────────────────────────
POLL_INTERVAL = 5 # seconds between event polls
CRITICAL_POLL_INTERVAL = 2 # poll faster when critical events detected
MAX_EVENT_HISTORY = 200 # events kept in rolling buffer
CRITICAL_SEVERITIES = {'critical', 'duplicantdeath', 'buildingdamage', 'poweroutage'}
WARNING_SEVERITIES = {'warning', 'duplicantstress', 'lowoxygen', 'foodshortage'}
# ── Event History ──────────────────────────────────────────────────────────
class EventHistory:
"""Rolling buffer of events + statistics for AI context."""
def __init__(self, maxlen=MAX_EVENT_HISTORY):
self.events = []
self.maxlen = maxlen
self.stats = {
'total': 0,
'critical': 0,
'warning': 0,
'info': 0,
'by_category': {},
'by_type': {},
'last_poll_cycle': 0,
}
def push(self, events):
for e in events:
self.events.append(e)
self.stats['total'] += 1
sev = (e.get('severity') or 'info').lower()
cat = e.get('category', 'unknown')
etype = e.get('type', 'unknown')
if sev in ('critical', 'duplicantdeath', 'buildingdamage'):
self.stats['critical'] += 1
elif sev in ('warning',):
self.stats['warning'] += 1
else:
self.stats['info'] += 1
self.stats['by_category'][cat] = self.stats['by_category'].get(cat, 0) + 1
self.stats['by_type'][etype] = self.stats['by_type'].get(etype, 0) + 1
self.stats['last_poll_cycle'] = e.get('cycle', 0)
# Trim
if len(self.events) > self.maxlen:
self.events = self.events[-self.maxlen:]
def get_summary(self):
return {
'total_events': self.stats['total'],
'critical_count': self.stats['critical'],
'warning_count': self.stats['warning'],
'info_count': self.stats['info'],
'categories': dict(sorted(self.stats['by_category'].items(),
key=lambda x: -x[1])[:10]),
'last_cycle': self.stats['last_poll_cycle'],
'recent_critical': [e for e in self.events[-20:]
if (e.get('severity') or '').lower() in CRITICAL_SEVERITIES][-5:],
}
# ── Event Classifier ──────────────────────────────────────────────────────
def classify_event(e):
"""Return the action type for a given event."""
sev = (e.get('severity') or '').lower()
title = (e.get('title') or '').lower()
msg = (e.get('message') or '').lower()
cat = (e.get('category') or '').lower()
if sev in CRITICAL_SEVERITIES:
return 'critical'
if sev in WARNING_SEVERITIES:
return 'warning'
if cat == 'action':
return 'action_feedback'
# Content-based classification
combined = title + ' ' + msg
if any(w in combined for w in ['suffocat', 'choking', 'no oxygen', 'out of air']):
return 'critical'
if any(w in combined for w in ['starving', 'food', 'hungry']):
return 'warning'
if any(w in combined for w in ['heat', 'overheat', 'temperature', 'melt']):
return 'warning'
if any(w in combined for w in ['power', 'wattage', 'shutoff']):
return 'warning'
if any(w in combined for w in ['duplicant', 'stress', 'break']):
return 'warning'
return 'info'
def format_event_for_ai(e):
"""Format an event as a structured string for AI input."""
ts = datetime.datetime.fromtimestamp(e.get('timestamp', time.time())).strftime('%H:%M:%S')
cycle = e.get('cycle', '?')
severity = e.get('severity', 'info').upper()
title = e.get('title', '?')
message = e.get('message', '')
lines = [f"[EVENT {severity}] Cycle {cycle} @ {ts}"]
lines.append(f" Title: {title}")
if message:
lines.append(f" Message: {message}")
entity = e.get('entity')
if entity:
lines.append(f" Entity: {entity}")
cell = e.get('cell')
if isinstance(cell, int) and cell >= 0:
lines.append(f" Cell: {cell}")
return '\n'.join(lines)
# ── Polling Loop ──────────────────────────────────────────────────────────
def poll_loop(event_history):
seq = 0
consecutive_errors = 0
print("[ONI Event Daemon] Starting event poll...")
print(f"[ONI Event Daemon] Poll interval: {POLL_INTERVAL}s")
print()
while True:
try:
data = api_get(f"/api/state/events?since={seq}&limit=50")
if 'error' in data:
consecutive_errors += 1
if consecutive_errors == 1:
print(f"[!] Cannot reach game: {data['error']}")
print(" Waiting for game connection...")
time.sleep(POLL_INTERVAL * 2)
continue
consecutive_errors = 0
events = data.get('events', [])
next_seq = data.get('next_seq', seq)
if events:
event_history.push(events)
# Classify and report
critical_events = []
for e in events:
cls = classify_event(e)
if cls == 'critical':
critical_events.append(e)
# Print alert with clear marker
print("=" * 56)
print(" *** CRITICAL EVENT ***")
print(format_event_for_ai(e))
print("=" * 56)
print()
# Auto-trigger full analysis on critical events
_trigger_emergency_analysis(e)
elif cls == 'warning':
print(format_event_for_ai(e))
print()
else:
# Only print non-info events or batch feedback
cat = e.get('category', '')
if cat != 'general' or cls != 'info':
print(format_event_for_ai(e))
print()
# If critical events happened, poll faster for a bit
if critical_events:
seq = next_seq
time.sleep(CRITICAL_POLL_INTERVAL)
continue
seq = next_seq
time.sleep(POLL_INTERVAL)
except KeyboardInterrupt:
print("\n[ONI Event Daemon] Shutting down.")
summary = event_history.get_summary()
print(f" Total events seen: {summary['total_events']}")
print(f" Critical: {summary['critical_count']}, Warning: {summary['warning_count']}")
break
except Exception as e:
consecutive_errors += 1
if consecutive_errors <= 2:
print(f"[!] Poll error: {e}")
time.sleep(POLL_INTERVAL)
def _trigger_emergency_analysis(event):
"""On critical events, pull game state snapshot for AI context."""
try:
print(" -> Triggering emergency snapshot...")
game = api_get('/api/state/game')
alerts = api_get('/api/state/alert')
dups = api_get('/api/state/duplicants')
if 'error' not in game:
print(f" [SNAPSHOT] Cycle {game.get('cycle', '?')}, "
f"{game.get('duplicantCount', '?')} dupes, "
f"{game.get('suffocating', 0)} suffocating, "
f"{game.get('starving', 0)} starving, "
f"{game.get('stressed', 0)} stressed")
if isinstance(alerts, list) and alerts:
print(f" [ALERTS] {len(alerts)} active:")
for a in alerts[:3]:
print(f" - [{a.get('severity', '?')}] {a.get('title', '?')}")
print()
except:
pass
# ── Main ──────────────────────────────────────────────────────────────────
def main():
history = EventHistory()
try:
poll_loop(history)
except KeyboardInterrupt:
pass
# Print final summary
summary = history.get_summary()
print()
print("=" * 56)
print(" Event Daemon Session Summary")
print("=" * 56)
print(f" Total events: {summary['total_events']}")
print(f" Critical: {summary['critical_count']}")
print(f" Warning: {summary['warning_count']}")
print(f" Info: {summary['info_count']}")
print(f" Top categories: {', '.join(summary['categories'].keys())}")
print("=" * 56)
if __name__ == '__main__':
main()

149
tools/oni_api.py
View File

@ -353,6 +353,140 @@ def cmd_harvest(args):
result = api_post('/api/action/harvest', {"x": int(args[0]), "y": int(args[1])})
print(json.dumps(result, indent=2, ensure_ascii=False))
# ---------------------------------------------------------------------------
# Event / Queue / Batch / Priority
# ---------------------------------------------------------------------------
def cmd_events(args):
"""Poll game events. Usage: events [since] [limit]"""
since = args[0] if len(args) > 0 else "0"
limit = args[1] if len(args) > 1 else "50"
data = api_get(f"/api/state/events?since={since}&limit={limit}")
if 'error' in data:
print(f"Error: {data['error']}")
return
events = data.get('events', [])
next_seq = data.get('next_seq', 0)
has_more = data.get('has_more', False)
if not events:
print("No new events.")
print(f"Next sequence: {next_seq}")
return
print(f"Events ({len(events)} new, next_seq={next_seq}, has_more={has_more}):")
print()
for e in events:
severity = e.get('severity', '?')
sev_mark = {'critical': '!!!', 'warning': '!!', 'info': 'i'}.get(severity.lower(), '?')
cat = e.get('category', '?')
title = e.get('title', '?')
msg = e.get('message', '')
cycle = e.get('cycle', '?')
entity = e.get('entity', '')
cell = e.get('cell', '')
print(f" [{sev_mark}] ({cycle}) {title}")
if msg:
print(f" {msg}")
if entity:
print(f" entity: {entity}")
if isinstance(cell, int) and cell >= 0:
print(f" cell index: {cell}")
print()
def cmd_queue(args):
"""View pending task queue. Usage: queue [batch_id]"""
params = ""
if args:
params = f"?batch_id={args[0]}"
data = api_get(f"/api/state/queue{params}")
if 'error' in data:
print(f"Error: {data['error']}")
return
print(f"Task Queue:")
print(f" Length: {data.get('queue_length', '?')}")
print(f" Batch ID: {data.get('batch_id', 'none')}")
for t in data.get('tasks', []):
print(f" - {t.get('type', '?')}: {t.get('value', '?')}")
def cmd_batch(args):
"""Execute a batch of actions. Usage: batch <json_file>"""
if not args:
print("Usage: batch <json_file>")
print(" JSON format: { \"actions\": [ { \"type\": \"build|dig|...\", ... } ] }")
return
try:
with open(args[0]) as f:
plan = json.load(f)
except Exception as e:
print(f"Error reading file: {e}")
return
result = api_post('/api/action/batch', plan)
if 'error' in result:
print(f"Error: {result['error']}")
return
print(f"Batch: {result.get('batchId', '?')}")
print(f" Total: {result.get('total', 0)}")
print(f" OK: {result.get('successCount', 0)}")
print(f" Failed: {result.get('failCount', 0)}")
print(f" Summary: {result.get('summary', '?')}")
print()
for action in result.get('actions', []):
status = 'OK' if action.get('success') else 'FAIL'
result_type = action.get('result', '?')
error = action.get('error', '')
err_msg = action.get('errorMessage', '')
suggestion = action.get('suggestion', '')
print(f" [{status}] {result_type}")
if error:
print(f" error: {error}")
if err_msg:
print(f" msg: {err_msg}")
if suggestion:
print(f" -> {suggestion}")
print()
def cmd_priority_global(args):
"""Set global priority. Usage: priority_global <target> <priority>"""
if len(args) < 2:
print("Usage: priority_global <target> <priority>")
print(" target: 'dig', 'build', 'clear', or 'all'")
print(" priority: 1 (lowest) to 9 (emergency)")
return
result = api_post('/api/action/priority_global', {
"target": args[0],
"priority": int(args[1])
})
_print_feedback(result)
def cmd_priority_type(args):
"""Set priority for a building type. Usage: priority_type <buildingType> <priority>"""
if len(args) < 2:
print("Usage: priority_type <buildingType> <priority>")
return
result = api_post('/api/action/priority_type', {
"buildingType": args[0],
"priority": int(args[1])
})
_print_feedback(result)
def _print_feedback(result):
"""Pretty-print action feedback."""
if result.get('success'):
print(f"OK: {result.get('result', 'done')}")
for k, v in result.get('data', {}).items():
print(f" {k}: {v}")
else:
print(f"FAIL: {result.get('error', 'unknown_error')}")
print(f" {result.get('errorMessage', '')}")
sug = result.get('suggestion')
if sug:
print(f" -> {sug}")
def cmd_explore(args):
"""AI-friendly exploration: reads a region and returns structured text summary."""
x = int(args[0]) if len(args) > 0 else 0
@ -432,6 +566,8 @@ COMMANDS = {
'critters': cmd_critters,
'plants': cmd_plants,
'rooms': cmd_rooms,
'queue': cmd_queue,
'events': cmd_events,
'cell': cmd_cell,
'cells': cmd_cells,
'slice': cmd_cell_slice,
@ -445,6 +581,9 @@ COMMANDS = {
'mop': cmd_mop,
'harvest': cmd_harvest,
'explore': cmd_explore,
'batch': cmd_batch,
'priority_global': cmd_priority_global,
'priority_type': cmd_priority_type,
}
if __name__ == '__main__':
@ -473,10 +612,15 @@ if __name__ == '__main__':
print(" gas <x> <y> [r] Gas analysis in radius r")
print(" explore <x> <y> <w> <h> AI-friendly region summary")
print("")
print("=== Event / Queue ===")
print(" events [since] [limit] Poll new game events")
print(" queue [batch_id] View pending task queue")
print("")
print("=== Registries (AI Reference) ===")
print(" registry buildings [f] List all building IDs with metadata")
print(" registry elements [f] List all element IDs with properties")
print(" registry techs [f] List all tech IDs with unlocks")
print(" registry priorities Show priority level meanings")
print("")
print("=== Actions ===")
print(" dig <x> <y> <w> <h> Dig area")
@ -486,5 +630,10 @@ if __name__ == '__main__':
print(" research_select <id> Select tech to research")
print(" mop <x> <y> Mop liquid")
print(" harvest <x> <y> Harvest plant")
print("")
print("=== Batch / Priority (Advanced) ===")
print(" batch <json_file> Execute batch plan")
print(" priority_global <t> <p> Set global default priority")
print(" priority_type <type> <p> Set per-building-type priority")
else:
COMMANDS[cmd](sys.argv[2:])