Drive a single persistent claude per worker via bidirectional stream-json (closes #455)

CLAUDE.md

@@ -16,6 +16,8 @@ Multi-repo: one kennel process handles multiple repos. Each repo has its own tas
 
 **Concurrency model**: one fido per repo, one issue per fido, one PR per issue. Fido finishes the current issue (PR merged or closed) before picking up the next. Two repos = two fidos max, running in parallel, each on their own issue.
 
+**ClaudeSession persistence**: the persistent `ClaudeSession` (bidirectional stream-json subprocess) is held on `WorkerThread._session` and survives individual `Worker` crashes — the watchdog restarts the thread and the next `Worker` inherits the same session. It does *not* survive a kennel/home restart: `os.execvp` replaces the process entirely, so the new kennel starts with `_session = None` and creates a fresh session on its first iteration.
+
 ## Runner vs workspace clones
 
 Kennel runs from a dedicated **runner clone** at `/home/rhencke/home-runner/`, separate from the **workspace clone** at `/home/rhencke/workspace/home/`.

kennel/registry.py

@@ -69,7 +69,7 @@ class WorkerRegistry:
         registry.stop_all()          # clean shutdown
     """
 
-    def __init__(self, thread_factory: Callable[[RepoConfig], WorkerThread]) -> None:
+    def __init__(self, thread_factory: Callable[..., WorkerThread]) -> None:
         self._threads: dict[str, WorkerThread] = {}
         self._factory = thread_factory
         self._activities: dict[str, WorkerActivity] = {}
@@ -83,8 +83,24 @@ def __init__(self, thread_factory: Callable[[RepoConfig], WorkerThread]) -> None
         self._webhook_lock = threading.Lock()
 
     def start(self, repo_cfg: RepoConfig) -> None:
-        """Create and start a WorkerThread for *repo_cfg*."""
-        thread = self._factory(repo_cfg)
+        """Create and start a WorkerThread for *repo_cfg*.
+
+        If a previous thread for this repo crashed (dead but not stopped
+        orderly), its live session is rescued and handed to the replacement so
+        the persistent :class:`~kennel.claude.ClaudeSession` survives the crash.
+        """
+        session = None
+        session_issue = None
+        old_thread = self._threads.get(repo_cfg.name)
+        if (
+            old_thread is not None
+            and not old_thread.is_alive()
+            and not old_thread._stop
+        ):
+            # Crashed thread — rescue the live session before replacing it
+            session, old_thread._session = old_thread._session, None
+            session_issue, old_thread._session_issue = old_thread._session_issue, None
+        thread = self._factory(repo_cfg, session=session, session_issue=session_issue)
         self._threads[repo_cfg.name] = thread
         with self._started_at_lock:
             self._started_at[repo_cfg.name] = _utcnow()
@@ -248,12 +264,24 @@ def get_thread_crash_error(self, repo_name: str) -> str | None:
         thread = self._threads.get(repo_name)
         return thread.crash_error if thread is not None else None
 
+    def get_session_owner(self, repo_name: str) -> str | None:
+        """Return the name of the thread currently holding the ClaudeSession lock.
+
+        Delegates to :attr:`~kennel.worker.WorkerThread.session_owner` on the
+        registered thread.  Returns ``None`` when no thread is registered for
+        the repo, no session exists, or the lock is currently free.
+        """
+        thread = self._threads.get(repo_name)
+        return thread.session_owner if thread is not None else None
+
 
 def _make_thread(
     repo_cfg: RepoConfig,
     registry: WorkerRegistry,
     *,
     gh: GitHub,
+    session=None,
+    session_issue=None,
     _WorkerThread=WorkerThread,
 ) -> WorkerThread:
     """Default factory: create a WorkerThread with the provided GitHub client."""
@@ -263,6 +291,8 @@ def _make_thread(
         gh,
         registry,
         repo_cfg.membership,
+        session=session,
+        session_issue=session_issue,
     )
 
 
@@ -279,8 +309,10 @@ def make_registry(
     (with a mock factory) in tests instead of calling this.
     """
 
-    def factory(cfg: RepoConfig) -> WorkerThread:
-        return _thread_factory(cfg, registry, gh=gh)
+    def factory(cfg: RepoConfig, *, session=None, session_issue=None) -> WorkerThread:
+        return _thread_factory(
+            cfg, registry, gh=gh, session=session, session_issue=session_issue
+        )
 
     registry = WorkerRegistry(factory)
     for repo_cfg in repos.values():

kennel/server.py

@@ -547,6 +547,7 @@ def do_GET(self) -> None:
                         ),
                         "worker_uptime_seconds": worker_uptime,
                         "webhook_activities": webhooks,
+                        "session_owner": self.registry.get_session_owner(a.repo_name),
                     }
                 )
             body = json.dumps(activities).encode()

kennel/status.py

@@ -47,6 +47,7 @@ class RepoStatus:
     task_total: int | None = None
     worker_uptime: int | None = None
     webhook_activities: list[WebhookActivityInfo] = field(default_factory=list)
+    session_owner: str | None = None
 
 
 @dataclass
@@ -196,6 +197,7 @@ def _fetch_activities(
                 "is_stuck": item.get("is_stuck", False),
                 "worker_uptime_seconds": item.get("worker_uptime_seconds"),
                 "webhook_activities": item.get("webhook_activities", []),
+                "session_owner": item.get("session_owner"),
             }
             for item in data
             if "repo_name" in item and "what" in item
@@ -317,6 +319,7 @@ def repo_status(
     worker_stuck: bool = False,
     worker_uptime: int | None = None,
     webhook_activities: list[WebhookActivityInfo] | None = None,
+    session_owner: str | None = None,
 ) -> RepoStatus:
     """Collect status for a single repo."""
     webhook_activities = list(webhook_activities or [])
@@ -343,6 +346,7 @@ def repo_status(
             last_crash_error=last_crash_error,
             worker_stuck=worker_stuck,
             webhook_activities=webhook_activities,
+            session_owner=session_owner,
         )
 
     fido_dir = git_dir / "fido"
@@ -388,6 +392,7 @@ def repo_status(
         last_crash_error=last_crash_error,
         worker_stuck=worker_stuck,
         webhook_activities=webhook_activities,
+        session_owner=session_owner,
     )
 
 
@@ -427,6 +432,7 @@ def collect() -> KennelStatus:
                 worker_stuck=info["is_stuck"] if info else False,
                 worker_uptime=worker_uptime_val,
                 webhook_activities=webhook_list,
+                session_owner=info.get("session_owner") if info else None,
             )
         )
     return KennelStatus(kennel_pid=pid, kennel_uptime=uptime, repos=repos)
@@ -497,8 +503,13 @@ def _format_repo_body(repo: RepoStatus) -> list[str]:
 
     if repo.claude_pid is not None:
         claude_str = f"  └─ claude pid {repo.claude_pid}"
+        parts: list[str] = []
         if repo.claude_uptime is not None:
-            claude_str += f" (running {_format_uptime(repo.claude_uptime)})"
+            parts.append(f"running {_format_uptime(repo.claude_uptime)}")
+        if repo.session_owner is not None:
+            parts.append(f"held by {repo.session_owner}")
+        if parts:
+            claude_str += f" ({', '.join(parts)})"
         body.append(claude_str)
 
     for w in repo.webhook_activities: