Dev

.github/workflows/test.yml

@@ -70,6 +70,7 @@ jobs:
                  tests/db/test_pg_sections.py \
                  tests/db/test_pg_auth_queries.py \
                  tests/db/test_pg_integrations.py \
+                 tests/db/test_memory_context_schema.py \
                  tests/api/ \
                  -v
 

Dockerfile

@@ -20,7 +20,7 @@ FROM python:3.11-slim
 # cron is needed for anchor trigger scheduling in the bot container.
 # curl is needed for the NodeSource setup script.
 RUN apt-get update && apt-get install -y --no-install-recommends \
-    gcc libffi-dev cron git gosu curl \
+    gcc libffi-dev cron git gosu curl redis-server \
     && rm -rf /var/lib/apt/lists/*
 
 # Install Node.js 20 via NodeSource so npm has a properly self-contained
@@ -34,6 +34,19 @@ RUN curl -fsSL https://deb.nodesource.com/setup_20.x | bash - \
 ARG CLAUDE_CODE_VERSION=2.1.116
 RUN npm install -g @anthropic-ai/claude-code@${CLAUDE_CODE_VERSION}
 
+# Home directory template for the agent pool manager.
+# Running `claude --version` once causes the CLI to write its initial config
+# (~/.claude.json, ~/.claude/) into the temp home.  We snapshot that state so
+# each warm subprocess gets a pre-seeded home dir, eliminating first-run setup
+# time and ~/.claude.json lock contention on concurrent spawns.
+# If the CLI produces no files (e.g. newer version skips auto-init), the
+# template dir is left empty — the isolation benefit still applies.
+RUN HOME=/tmp/claude-template \
+      claude --version 2>/dev/null || true \
+    && mkdir -p /etc/claude-home-template \
+    && cp -r /tmp/claude-template/. /etc/claude-home-template/ 2>/dev/null || true \
+    && chmod -R a+rX /etc/claude-home-template
+
 # Non-root user for running services. UID 1000 matches the default Pi user
 # so bind-mounted /data files are accessible without permission issues.
 RUN useradd -m -u 1000 -s /bin/bash tether
@@ -45,6 +58,17 @@ RUN mkdir -p /run/tether/creds \
     && chown tether:tether /run/tether/creds \
     && chmod 0700 /run/tether/creds
 
+# Agent pool manager home directory base.
+# /var/lib/ is root-owned — the tether user cannot mkdir here at runtime,
+# so the pool base dir must be pre-created and handed off in the image.
+# Also hand ownership of the Claude home template to tether so that
+# initialize() can write .claude.json into it at first boot (auth seeding).
+RUN mkdir -p /var/lib/tether/claude-homes \
+    && chown tether:tether /var/lib/tether/claude-homes \
+    && chmod 755 /var/lib/tether/claude-homes \
+    && chown tether:tether /etc/claude-home-template \
+    && chmod 755 /etc/claude-home-template
+
 WORKDIR /app
 
 # External dependencies (cached layer — only reruns when requirements.txt changes)

agent_pool_manager/config.py

@@ -39,6 +39,23 @@ class AgentPoolConfig:
     control_response_timeout_seconds: float = 60.0
     """Seconds the pool waits for a control_response before denying the tool call."""
 
+    connect_timeout_seconds: float = 15.0
+    """Max seconds to wait for ClaudeSDKClient.connect() before treating the spawn
+    as failed.  Must be shorter than prime_timeout_seconds so a hung transport
+    is detected and killed before the priming phase would time out anyway."""
+
+    initialize_timeout_ms: int = 120000
+    """Value injected as CLAUDE_CODE_STREAM_CLOSE_TIMEOUT (ms) in subprocess env.
+    The SDK uses this as its internal initialize timeout (floor: 60 000 ms).
+    Default: 120 000 ms (2 min)."""
+
+    home_dir_base: str = "/var/lib/tether/claude-homes"
+    """Directory where isolated per-subprocess home dirs are created."""
+
+    home_dir_template: str = "/etc/claude-home-template"
+    """Directory copied into each home dir at initialization.
+    If it does not exist, home dirs start empty (lock-contention fix still applies)."""
+
 
 _FIELD_NAMES = {f.name for f in fields(AgentPoolConfig)}
 

agent_pool_manager/homes.py

@@ -0,0 +1,172 @@
+"""Isolated per-subprocess home directory pool.
+
+Each subprocess gets its own ``$HOME`` to eliminate ``~/.claude.json`` lock
+contention when two subprocesses spawn simultaneously.  Dirs are pre-created
+at pool startup and reset to a template state before each assignment so stale
+CLI state from a previous subprocess never bleeds into the next one.
+
+Design notes:
+- Dirs are never deleted during normal operation — only reset.
+- Reset (``shutil.rmtree`` + ``shutil.copytree``) runs in a thread-pool
+  executor so it does not block the asyncio event loop during spawn.
+- TTL sweep evicts stale assignments (e.g. a subprocess that was lost
+  without going through ``_terminate``).
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+import shutil
+import time
+from pathlib import Path
+from typing import TYPE_CHECKING
+
+if TYPE_CHECKING:
+    from agent_pool_manager.config import AgentPoolConfig
+
+log = logging.getLogger(__name__)
+
+# Path to the running container user's .claude.json.  Isolated to a constant
+# so tests can monkeypatch without touching the filesystem at /home/tether.
+_AUTH_SOURCE_PATH: Path = Path("/home/tether/.claude.json")
+
+
+def _reset_home(home: Path, template: Path | None) -> None:
+    """Synchronous reset called from a thread executor.
+
+    Removes all content in *home* then either copies *template* in (if it
+    exists) or leaves *home* as an empty directory.
+    """
+    # Wipe and recreate so the dir is empty/fresh regardless of template.
+    shutil.rmtree(home, ignore_errors=True)
+    home.mkdir(parents=True, exist_ok=True)
+    if template is not None and template.is_dir():
+        # copytree requires the destination not to exist; use dirs_exist_ok
+        # (Python 3.8+) so we can copy *into* the already-created home dir.
+        shutil.copytree(str(template), str(home), dirs_exist_ok=True)
+
+
+class HomeDirPool:
+    """Pool of isolated home directories for agent subprocesses.
+
+    Call ``await initialize()`` once before first use.
+    Call ``await acquire()`` to obtain a home dir for a new subprocess.
+    Call ``release(path)`` when the subprocess terminates.
+    Call ``await sweep()`` periodically (from RefillLoop.run_once) to evict
+    stale assignments that were never released.
+    """
+
+    def __init__(self, config: AgentPoolConfig) -> None:
+        self._config = config
+        self._base = Path(config.home_dir_base)
+        template_path = Path(config.home_dir_template)
+        self._template: Path | None = template_path if template_path.is_dir() else None
+        self._dirs: list[Path] = []
+        self._available: asyncio.Queue[Path] = asyncio.Queue()
+        # str(path) → expiry timestamp (time.monotonic)
+        self._checked_out: dict[str, float] = {}
+
+    async def initialize(self) -> None:
+        """Create home dirs and seed from template.  Idempotent on re-call."""
+        count = self._config.capacity_total
+        self._base.mkdir(parents=True, exist_ok=True)
+
+        # Re-discover template now (it may have been created since __init__)
+        template_path = Path(self._config.home_dir_template)
+        self._template = template_path if template_path.is_dir() else None
+
+        # Seed template with auth credentials at runtime if not already present.
+        # /home/tether/.claude.json is written by `claude setup-token` after
+        # the container first boots — it is never baked into the image (Fly
+        # secret).  Copying it into the template here means every warm
+        # subprocess starts with valid credentials without an extra setup step.
+        # Skips gracefully when: template dir absent, source absent, or
+        # template already has its own .claude.json.
+        if self._template is not None:
+            template_claude_json = self._template / ".claude.json"
+            if not template_claude_json.exists() and _AUTH_SOURCE_PATH.exists():
+                try:
+                    shutil.copy2(_AUTH_SOURCE_PATH, template_claude_json)
+                    log.info(
+                        "home_pool.seeded_template source=%s dest=%s",
+                        _AUTH_SOURCE_PATH,
+                        template_claude_json,
+                    )
+                except OSError:
+                    log.warning(
+                        "home_pool.seed_failed source=%s dest=%s",
+                        _AUTH_SOURCE_PATH,
+                        template_claude_json,
+                        exc_info=True,
+                    )
+
+        for i in range(count):
+            home = self._base / f"home-{i}"
+            home.mkdir(parents=True, exist_ok=True)
+            # Skip reset for dirs currently checked out by a live subprocess —
+            # wiping them would corrupt the running CLI process's HOME.
+            if str(home) not in self._checked_out:
+                # Seed from template in executor to avoid blocking the loop.
+                await asyncio.to_thread(_reset_home, home, self._template)
+            if home not in self._dirs:
+                self._dirs.append(home)
+            # Populate the available queue; clear first to avoid duplication on re-init.
+
+        # Rebuild _available from dirs not currently checked out.
+        while not self._available.empty():
+            try:
+                self._available.get_nowait()
+            except asyncio.QueueEmpty:
+                break
+        checked_out_paths = set(self._checked_out.keys())
+        for home in self._dirs:
+            if str(home) not in checked_out_paths:
+                self._available.put_nowait(home)
+
+        log.info(
+            "home_pool.initialized count=%d template=%s available=%d",
+            count,
+            self._template,
+            self._available.qsize(),
+        )
+
+    async def acquire(self) -> Path:
+        """Return a home dir, reset to template state.
+
+        Blocks (up to the caller's timeout) if the pool is exhausted.
+        Raises ``asyncio.QueueEmpty`` if called with nowait and no dirs are
+        available — callers should use ``asyncio.wait_for`` for timeouts.
+        """
+        home = await self._available.get()
+        await asyncio.to_thread(_reset_home, home, self._template)
+        # Use 2× subprocess max_age as home TTL so sweep never evicts a home
+        # whose subprocess might still be running (drain-on-touch retirement
+        # means a subprocess can live past its nominal max_age).
+        expiry = time.monotonic() + self._config.max_age_seconds * 2
+        self._checked_out[str(home)] = expiry
+        log.debug("home_pool.acquired path=%s", home)
+        return home
+
+    def release(self, home: Path) -> None:
+        """Return *home* to the available queue.  Idempotent."""
+        key = str(home)
+        if key not in self._checked_out:
+            # Already released or never acquired — silently ignore.
+            return
+        self._checked_out.pop(key)
+        self._available.put_nowait(home)
+        log.debug("home_pool.released path=%s", home)
+
+    async def sweep(self) -> None:
+        """Evict stale assignments that were never released (TTL-based safety net)."""
+        now = time.monotonic()
+        evicted = [k for k, expiry in list(self._checked_out.items()) if now > expiry]
+        for key in evicted:
+            self._checked_out.pop(key, None)
+            path = Path(key)
+            self._available.put_nowait(path)
+            log.warning("home_pool.sweep_evict path=%s", path)
+
+    def available_count(self) -> int:
+        """Number of home dirs currently available (not checked out)."""
+        return self._available.qsize()