fix(kanban): add inter-process flock to prevent SQLite WAL corruption

hermes_cli/kanban_db.py

@@ -85,7 +85,7 @@
 import time
 from dataclasses import dataclass, field
 from pathlib import Path
-from typing import Any, Iterable, Optional
+from typing import Any, Iterable, Optional, TextIO
 
 from toolsets import get_toolset_names
 
@@ -102,6 +102,19 @@
 KNOWN_TOOLSET_NAMES = frozenset(name.casefold() for name in get_toolset_names())
 _IS_WINDOWS = sys.platform == "win32"
 
+# Optional fcntl import — Windows does not ship fcntl.
+# On non-Windows platforms, used for per-DB inter-process write locking.
+# On Windows, _KANBAN_WRITE_LOCKS entries are None and flock is a no-op.
+try:
+    import fcntl  # type: ignore[import-not-found,import-untyped]
+except ImportError:
+    fcntl = None  # type: ignore[assignment]  # Windows / other platforms without fcntl
+
+# Maps id(conn) → open file handle for per-DB write locking.
+# sqlite3.Connection is a C-extension type: no attributes, no weakref.
+# We use id(conn) as key and accept the dict grows lazily.
+_KANBAN_WRITE_LOCKS: dict[int, "TextIO | None"] = {}
+
 # A running task's claim is valid for 15 minutes by default; after that the
 # next dispatcher tick reclaims it. Workers that outlive this window should
 # call ``heartbeat_claim(task_id)`` periodically. In practice most kanban
@@ -1393,6 +1406,19 @@ def connect(
         _guard_existing_db_is_healthy(path)
         resolved = str(path.resolve())
         conn = _sqlite_connect(path)
+        # Per-DB file-level write lock — prevents multiple worker processes from
+        # racing through BEGIN IMMEDIATE transactions simultaneously, which can
+        # corrupt the WAL under heavy concurrent write pressure (task_events +
+        # task_runs + status transitions all within the same tick).  Windows
+        # (no fcntl) and network filesystems gracefully fall back to no-op.
+        if fcntl is not None:
+            lock_path = path.with_suffix('.db.lock')
+            try:
+                _KANBAN_WRITE_LOCKS[id(conn)] = open(lock_path, 'w')
+            except OSError:
+                _KANBAN_WRITE_LOCKS[id(conn)] = None  # best-effort: don't fail
+        else:
+            _KANBAN_WRITE_LOCKS[id(conn)] = None  # Windows / no-fcntl: skip file lock entirely
         try:
             conn.row_factory = sqlite3.Row
             with _INIT_LOCK:
@@ -1423,8 +1449,22 @@ def connect(
                     # process are cheap. The lock prevents same-process dispatcher
                     # threads from racing through the additive ALTER TABLE pass with
                     # stale PRAGMA snapshots during gateway startup.
-                    conn.executescript(SCHEMA_SQL)
-                    _migrate_add_optional_columns(conn)
+                    #
+                    # CRITICAL: DDL writes (executescript + migrations) must be under
+                    # the per-DB .db.lock to prevent racing with other processes'
+                    # write_txn() calls. Without this, a new worker's schema init can
+                    # collide with a dispatching gateway's DML writes, corrupting
+                    # the WAL. .init.lock alone is NOT sufficient because it is a
+                    # different file from .db.lock.
+                    lock_handle = _KANBAN_WRITE_LOCKS.get(id(conn))
+                    if lock_handle is not None:
+                        fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX)
+                    try:
+                        conn.executescript(SCHEMA_SQL)
+                        _migrate_add_optional_columns(conn)
+                    finally:
+                        if lock_handle is not None:
+                            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
                     _INITIALIZED_PATHS.add(resolved)
         except Exception:
             conn.close()
@@ -1929,10 +1969,20 @@ def write_txn(conn: sqlite3.Connection):
     task + recording an event, etc.).  A claim CAS inside this context is
     atomic -- at most one concurrent writer can succeed.
 
+    Additionally, if conn's corresponding lock fd is set (by connect()),
+    an inter-process exclusive file lock is acquired around the entire
+    transaction.  This prevents multiple worker processes from racing
+    through BEGIN IMMEDIATE simultaneously, which can corrupt the WAL
+    under heavy concurrent write pressure.  On platforms without fcntl
+    or when the lock fd is unavailable, this is a graceful no-op.
+
     The explicit ROLLBACK on exception is wrapped in try/except so that
     a SQLite auto-rollback (which leaves no active transaction) does not
     shadow the original exception with a spurious rollback error.
     """
+    lock_handle = _KANBAN_WRITE_LOCKS.get(id(conn))
+    if lock_handle is not None:
+        fcntl.flock(lock_handle.fileno(), fcntl.LOCK_EX)
     conn.execute("BEGIN IMMEDIATE")
     try:
         yield conn
@@ -1950,6 +2000,9 @@ def write_txn(conn: sqlite3.Connection):
         # Post-commit file-length check: header page_count must match actual file pages.
         # A discrepancy means a torn-extend — raise now rather than silently corrupt.
         _check_file_length_invariant(conn)
+    finally:
+        if lock_handle is not None:
+            fcntl.flock(lock_handle.fileno(), fcntl.LOCK_UN)
 
 
 # ---------------------------------------------------------------------------

tools/send_message_tool.py

@@ -1505,6 +1505,9 @@ async def _send_dingtalk(extra, chat_id, message):
     delivery we use a static robot webhook URL instead, which must be
     configured via ``DINGTALK_WEBHOOK_URL`` env var or ``webhook_url`` in the
     platform's extra config.
+
+    Sends as markdown (matching the live adapter) with fallback to plain text
+    if the API rejects the markdown payload.
     """
     try:
         import httpx
@@ -1514,15 +1517,36 @@ async def _send_dingtalk(extra, chat_id, message):
         webhook_url = extra.get("webhook_url") or os.getenv("DINGTALK_WEBHOOK_URL", "")
         if not webhook_url:
             return {"error": "DingTalk not configured. Set DINGTALK_WEBHOOK_URL env var or webhook_url in dingtalk platform extra config."}
+
+        # Truncate to 20K characters to match the live adapter's limit
+        max_len = 20000
+        if len(message) > max_len:
+            message = message[:max_len]
+
         async with httpx.AsyncClient(timeout=30.0) as client:
-            resp = await client.post(
-                webhook_url,
-                json={"msgtype": "text", "text": {"content": message}},
-            )
+            try:
+                # Send as markdown (same format as live DingTalk adapter)
+                resp = await client.post(
+                    webhook_url,
+                    json={"msgtype": "markdown", "markdown": {"title": "Hermes", "text": message}},
+                )
+            except Exception as e:
+                return _error(f"DingTalk send failed: {e}")
+
             resp.raise_for_status()
             data = resp.json()
             if data.get("errcode", 0) != 0:
-                return _error(f"DingTalk API error: {data.get('errmsg', 'unknown')}")
+                # Fallback to plain text if markdown is rejected
+                errmsg = data.get("errmsg", "")
+                logger.warning("DingTalk markdown rejected (%s), falling back to text", errmsg)
+                resp = await client.post(
+                    webhook_url,
+                    json={"msgtype": "text", "text": {"content": message}},
+                )
+                resp.raise_for_status()
+                data = resp.json()
+                if data.get("errcode", 0) != 0:
+                    return _error(f"DingTalk API error: {data.get('errmsg', 'unknown')}")
         return {"success": True, "platform": "dingtalk", "chat_id": chat_id}
     except Exception as e:
         return _error(f"DingTalk send failed: {e}")