修复 kanban 原子更新后的二次写回风险并减少冗余 I/O

scripts/kanban_update.py

@@ -32,7 +32,7 @@
 logging.basicConfig(level=logging.INFO, format='%(asctime)s [%(name)s] %(message)s', datefmt='%H:%M:%S')
 
 # 文件锁 —— 防止多 Agent 同时读写 tasks_source.json
-from file_lock import atomic_json_read, atomic_json_update, atomic_json_write  # noqa: E402
+from file_lock import atomic_json_read, atomic_json_update  # noqa: E402
 
 STATE_ORG_MAP = {
     'Taizi': '太子', 'Zhongshu': '中书省', 'Menxia': '门下省', 'Assigned': '尚书省',
@@ -66,8 +66,8 @@
 def load():
     return atomic_json_read(TASKS_FILE, [])
 
-def save(tasks):
-    atomic_json_write(TASKS_FILE, tasks)
+def trigger_refresh():
+    """异步触发 REFRESH_SCRIPT 更新 live_status，避免阻塞主更新流程。"""
     # 异步触发刷新，不阻塞调用方
     try:
         subprocess.Popen(['python3', str(REFRESH_SCRIPT)],
@@ -201,7 +201,7 @@ def modifier(tasks):
         })
         return tasks
     atomic_json_update(TASKS_FILE, modifier, [])
-    save(load())  # trigger refresh
+    trigger_refresh()
     log.info(f'✅ 创建 {task_id} | {title[:30]} | state={state}')
 
 
@@ -222,7 +222,7 @@ def modifier(tasks):
         t['updatedAt'] = now_iso()
         return tasks
     atomic_json_update(TASKS_FILE, modifier, [])
-    save(load())  # trigger refresh
+    trigger_refresh()
     log.info(f'✅ {task_id} 状态更新: {old_state[0]} → {new_state}')
 
 
@@ -240,7 +240,7 @@ def modifier(tasks):
         t['updatedAt'] = now_iso()
         return tasks
     atomic_json_update(TASKS_FILE, modifier, [])
-    save(load())  # trigger refresh
+    trigger_refresh()
     log.info(f'✅ {task_id} 流转记录: {from_dept} → {to_dept}')
 
 
@@ -261,7 +261,7 @@ def modifier(tasks):
         t['updatedAt'] = now_iso()
         return tasks
     atomic_json_update(TASKS_FILE, modifier, [])
-    save(load())  # trigger refresh
+    trigger_refresh()
     log.info(f'✅ {task_id} 已完成')
 
 
@@ -277,7 +277,7 @@ def modifier(tasks):
         t['updatedAt'] = now_iso()
         return tasks
     atomic_json_update(TASKS_FILE, modifier, [])
-    save(load())  # trigger refresh
+    trigger_refresh()
     log.warning(f'⚠️ {task_id} 已阻塞: {reason}')
 
 
@@ -366,7 +366,7 @@ def modifier(tasks):
         total_cnt[0] = len(t.get('todos', []))
         return tasks
     atomic_json_update(TASKS_FILE, modifier, [])
-    save(load())  # trigger refresh
+    trigger_refresh()
     res_info = ''
     if tokens or cost or elapsed:
         res_info = f' [res: {tokens}tok/${cost:.4f}/{elapsed}s]'
@@ -406,7 +406,7 @@ def modifier(tasks):
         result_info[1] = len(t['todos'])
         return tasks
     atomic_json_update(TASKS_FILE, modifier, [])
-    save(load())  # trigger refresh
+    trigger_refresh()
     log.info(f'✅ {task_id} todo [{result_info[0]}/{result_info[1]}]: {todo_id} → {status}')
 
 _CMD_MIN_ARGS = {

tests/test_kanban.py

@@ -61,3 +61,26 @@ def test_block_and_unblock(tmp_path):
         assert tasks[0]['block'] == '等待依赖'
     finally:
         kb.TASKS_FILE = original
+
+
+def test_state_update_does_not_reload_file_again(tmp_path, monkeypatch):
+    """state 更新后不应再次 load/save 同一文件。"""
+    tasks_file = tmp_path / 'tasks_source.json'
+    tasks_file.write_text(json.dumps([
+        {'id': 'T-3', 'title': 'perf', 'state': 'Inbox'}
+    ]))
+
+    original = kb.TASKS_FILE
+    kb.TASKS_FILE = tasks_file
+    popen_calls = []
+    def fail_on_reload(*_args, **_kwargs):
+        raise AssertionError('unexpected reload')
+    monkeypatch.setattr(kb.subprocess, 'Popen', lambda *args, **kwargs: popen_calls.append((args, kwargs)))
+    monkeypatch.setattr(kb, 'atomic_json_read', fail_on_reload)
+    try:
+        kb.cmd_state('T-3', 'Doing')
+        tasks = json.loads(tasks_file.read_text())
+        assert tasks[0]['state'] == 'Doing'
+        assert len(popen_calls) == 1
+    finally:
+        kb.TASKS_FILE = original