kin: auto-commit after pipeline

core/watchdog.py

@@ -6,30 +6,30 @@ If the PID is dead (ProcessLookupError / ESRCH), mark pipeline as failed
 and task as blocked with a descriptive reason.
 """
 
+import errno
 import logging
 import os
-import sqlite3
 import threading
 import time
 from pathlib import Path
 
 from core import models
-from core.db import init_db
+from core.db import get_connection, init_db
 
 _logger = logging.getLogger("kin.watchdog")
 
+_watchdog_started = False
+
 
 def _check_dead_pipelines(db_path: Path) -> None:
     """Single watchdog pass: open a fresh connection, scan running pipelines."""
+    conn = get_connection(db_path)
     try:
-        conn = sqlite3.connect(str(db_path), check_same_thread=False)
-        conn.row_factory = sqlite3.Row
         try:
             running = models.get_running_pipelines_with_pid(conn)
         except Exception as exc:
             # Table may not exist yet on very first startup
             _logger.debug("Watchdog: could not query pipelines (%s)", exc)
-            conn.close()
             return
 
         for row in running:
@@ -38,23 +38,23 @@ def _check_dead_pipelines(db_path: Path) -> None:
             task_id = row["task_id"]
             try:
                 os.kill(pid, 0)  # signal 0 = existence check
-            except ProcessLookupError:
-                reason = f"Process died unexpectedly (PID {pid})"
-                _logger.warning(
-                    "Watchdog: pipeline %s PID %s is dead — marking blocked (%s)",
-                    pipeline_id, pid, task_id,
-                )
-                try:
-                    models.update_pipeline(conn, pipeline_id, status="failed")
-                    models.update_task(conn, task_id, status="blocked", blocked_reason=reason)
-                except Exception as upd_exc:
-                    _logger.error("Watchdog: failed to update pipeline/task: %s", upd_exc)
-            except PermissionError:
-                # Process exists but we can't signal it (e.g. different user) — skip
-                pass
-        conn.close()
+            except OSError as e:
+                if e.errno == errno.ESRCH:
+                    reason = f"Process died unexpectedly (PID {pid})"
+                    _logger.warning(
+                        "Watchdog: pipeline %s PID %s is dead — marking blocked (%s)",
+                        pipeline_id, pid, task_id,
+                    )
+                    try:
+                        models.update_pipeline(conn, pipeline_id, status="failed")
+                        models.update_task(conn, task_id, status="blocked", blocked_reason=reason)
+                    except Exception as upd_exc:
+                        _logger.error("Watchdog: failed to update pipeline/task: %s", upd_exc)
+                # else: PermissionError (EACCES) — process exists but we can't signal it, skip
     except Exception as exc:
         _logger.error("Watchdog pass failed: %s", exc)
+    finally:
+        conn.close()
 
 
 def _watchdog_loop(db_path: Path, interval: int) -> None:
@@ -67,6 +67,10 @@ def _watchdog_loop(db_path: Path, interval: int) -> None:
 
 def start_watchdog(db_path: Path, interval: int = 30) -> None:
     """Start the background watchdog thread (daemon=True, so it dies with the process)."""
+    global _watchdog_started
+    if _watchdog_started:
+        return
+    _watchdog_started = True
     t = threading.Thread(
         target=_watchdog_loop,
         args=(db_path, interval),