Merge pull request 'Rename a resumed session's husk with an 'X ' prefix' (#2) from rename-husk into main

Reviewed-on: #1

README.md

@@ -62,6 +62,26 @@ runs `claude stop <old id>` after a successful resume. Without that the husk
 would be picked up again on a later tick and resumed into a *second* session
 working the same conversation — two agents racing on the same git worktree.
 
+A stopped husk is still listed, under the same name as the successor that
+replaced it, which is confusing to work next to. So the watchdog also prefixes
+it — `X rewrite larvitsmpp typescript esm` — by rewriting `name` in the husk's
+`state.json`, which is where `claude agents` reads it and which no command sets.
+`--superseded-prefix` changes it.
+
+### Host headroom
+
+A resume starts a fresh process that reloads a whole conversation, so before
+listing sessions — and again before each resume — the watchdog reads
+`/proc/meminfo` and holds off while the host is short: less than
+`--min-available-mb` of `MemAvailable`, or swap fuller than
+`--max-swap-used-pct`. Each resume of the same run is charged for its own
+headroom, because a session launched seconds ago has not grown into its memory
+yet.
+
+Holding off costs nothing: the block is still there on the next tick. On a host
+that reports no `MemAvailable` the guard stays inactive rather than blocking
+every resume.
+
 ### Circuit breakers
 
 Three limits stop a broken session, or a still-ongoing outage, from spawning an
@@ -77,7 +97,9 @@ endless chain of blocked jobs:
 again it is a *new* session id and qualifies on its own, so an outage spanning
 several reset windows still gets retried. What the cap prevents is one
 conversation being resumed into several concurrent copies. A resume that fails
-to launch at all records no attempt and is retried on the next tick.
+to launch at all records no attempt and is retried on the next tick; one that
+times out records an attempt, because the launch may have registered after the
+watchdog stopped waiting for it.
 
 `--max-per-day` is the one guard that cannot be derived from anything else. A
 usage-limit block is self-pacing, because the stated reset time is respected —
@@ -164,15 +186,18 @@ A stalled session that has been resumed shows up as a new job in
 Every option is a CLI flag with an environment-variable fallback — nothing is
 hardcoded. To make one permanent, add it to the crontab line.
 
-| Flag              | Env var                   | Default              |
+| Flag                  | Env var                      | Default      |
-| ----------------- | ------------------------- | -------------------- |
+| --------------------- | ---------------------------- | ------------ |
-| `--claude-bin`    | `CLAUDE_BIN`              | `claude`             |
+| `--claude-bin`        | `CLAUDE_BIN`                 | `claude`     |
-| `--claude-home`   | `CLAUDE_HOME`             | `~/.claude`          |
+| `--claude-home`       | `CLAUDE_HOME`                | `~/.claude`  |
-| `--max-attempts`  | `WATCHDOG_MAX_ATTEMPTS`   | `1`                  |
+| `--max-attempts`      | `WATCHDOG_MAX_ATTEMPTS`      | `1`          |
-| `--max-per-day`   | `WATCHDOG_MAX_PER_DAY`    | `50`                 |
+| `--max-per-day`       | `WATCHDOG_MAX_PER_DAY`       | `50`         |
-| `--max-per-run`   | `WATCHDOG_MAX_PER_RUN`    | `0` (all)            |
+| `--max-per-run`       | `WATCHDOG_MAX_PER_RUN`       | `0` (all)    |
-| `--prompt`        | `WATCHDOG_PROMPT`         | see `--help`         |
+| `--max-swap-used-pct` | `WATCHDOG_MAX_SWAP_USED_PCT` | `50`         |
-| `--dry-run`       | —                         | off                  |
+| `--min-available-mb`  | `WATCHDOG_MIN_AVAILABLE_MB`  | `2048`       |
+| `--prompt`            | `WATCHDOG_PROMPT`            | see `--help` |
+| `--superseded-prefix` | `WATCHDOG_SUPERSEDED_PREFIX` | `X `         |
+| `--dry-run`           | —                            | off          |
 
 ## Files
 

resume-stalled-sessions.py

@@ -22,11 +22,15 @@ from zoneinfo import ZoneInfo
 # Where `claude` installs itself, and what a scheduler's minimal PATH omits.
 CLAUDE_FALLBACK_DIRS = ("~/.local/bin", "/usr/local/bin", "/opt/homebrew/bin")
 
+MEMINFO = Path("/proc/meminfo")
+
 DEFAULT_PROMPT = (
     "Continue where you left off. Your previous run was interrupted by a "
     "transient failure (usage limit, API outage, or network), which has cleared."
 )
 
+DEFAULT_PREFIX = "X "
+
 # Every quota Claude Code can report as "You've hit your <limit> · <how to clear
 # it>", including the ones that cost money rather than time.
 QUOTA_BLOCK = re.compile(
@@ -86,6 +90,37 @@ def log(message):
     print(f"{datetime.now().astimezone().isoformat(timespec='seconds')} {message}", flush=True)
 
 
+def memory_strain(min_available_mb, max_swap_used_pct, meminfo=MEMINFO):
+    """Why the host cannot take another session, or None when it can.
+
+    Also None where the host does not report memory this way: a guard that
+    cannot read the numbers must not be the thing that stops recovery.
+    """
+    try:
+        lines = meminfo.read_text().splitlines()
+    except OSError:
+        return None
+    fields = {}
+    for line in lines:
+        key, _, rest = line.partition(":")
+        amount = rest.split()
+        if amount and amount[0].isdigit():
+            fields[key] = int(amount[0])
+
+    if "MemAvailable" not in fields:
+        return None
+    available_mb = fields["MemAvailable"] // 1024
+    if available_mb < min_available_mb:
+        return f"{available_mb} MB available, need {min_available_mb}"
+
+    swap_total = fields.get("SwapTotal", 0)
+    if swap_total:
+        used_pct = round(100 * (swap_total - fields.get("SwapFree", 0)) / swap_total)
+        if used_pct > max_swap_used_pct:
+            return f"swap {used_pct}% used, max {max_swap_used_pct}%"
+    return None
+
+
 def resolve_claude(name):
     """Absolute path to the Claude Code binary, or None if it cannot be found."""
     found = shutil.which(name)
@@ -254,11 +289,42 @@ def candidates(sessions, jobs_dir, now):
             "blocked_at": blocked_at,
             "name": session.get("name"),
             "session_id": session_id,
+            "state_file": state_file,
         })
     return sorted(found, key=lambda job: job["blocked_at"])
 
 
-def resume(job, claude_bin, prompt, dry_run):
+def mark_superseded(state_file, prefix):
+    """Prefix the husk's name so it reads apart from its successor, or say why not.
+
+    `claude agents` takes the name from state.json and offers no command to set
+    it, so the file is the only seam.
+    """
+    try:
+        state = json.loads(state_file.read_text())
+    except (OSError, json.JSONDecodeError) as err:
+        return str(err)
+    name = state.get("name")
+    if not isinstance(name, str) or name.startswith(prefix):
+        return None
+
+    state["name"] = prefix + name
+    flags = state.get("respawnFlags")
+    if isinstance(flags, list) and "--name" in flags:
+        at = flags.index("--name") + 1
+        if at < len(flags):
+            flags[at] = state["name"]
+
+    staged = state_file.with_name(state_file.name + ".new")
+    try:
+        staged.write_text(json.dumps(state, indent=2))
+        staged.replace(state_file)
+    except OSError as err:
+        return str(err)
+    return None
+
+
+def resume(job, claude_bin, prompt, prefix, dry_run):
     cwd = job["cwd"]
     if not cwd or not Path(cwd).is_dir():
         log(f"{job['id']}: skipped, working directory is gone ({cwd})")
@@ -274,6 +340,11 @@ def resume(job, claude_bin, prompt, dry_run):
             command + [prompt],
             capture_output=True, cwd=cwd, text=True, timeout=180, check=True,
         )
+    except subprocess.TimeoutExpired:
+        # The launch may have registered before we gave up, and a second try
+        # would put two agents on one conversation.
+        log(f"{job['id']}: resume timed out, counting the attempt")
+        return True
     except (OSError, subprocess.SubprocessError) as err:
         log(f"{job['id']}: resume failed: {err}")
         return False
@@ -288,6 +359,10 @@ def resume(job, claude_bin, prompt, dry_run):
         )
     except (OSError, subprocess.SubprocessError) as err:
         log(f"{job['id']}: could not retire stale entry: {err}")
+
+    problem = mark_superseded(job["state_file"], prefix)
+    if problem:
+        log(f"{job['id']}: could not rename stale entry: {problem}")
     return True
 
 
@@ -318,7 +393,23 @@ def main():
         default=int(os.environ.get("WATCHDOG_MAX_PER_RUN", "0")),
         help="resumes per invocation, 0 for every eligible session (default: 0)",
     )
+    parser.add_argument(
+        "--max-swap-used-pct", type=int,
+        default=int(os.environ.get("WATCHDOG_MAX_SWAP_USED_PCT", "50")),
+        help="swap fill above which the host is already trading, so no resume (default: 50)",
+    )
+    parser.add_argument(
+        "--min-available-mb", type=int,
+        default=int(os.environ.get("WATCHDOG_MIN_AVAILABLE_MB", "2048")),
+        help="memory a resume needs the host to have spare, in MB (default: 2048)",
+    )
     parser.add_argument("--prompt", default=os.environ.get("WATCHDOG_PROMPT", DEFAULT_PROMPT))
+    parser.add_argument(
+        "--superseded-prefix",
+        default=os.environ.get("WATCHDOG_SUPERSEDED_PREFIX", DEFAULT_PREFIX),
+        help="prepended to the name of a husk once its successor runs "
+             f"(default: {DEFAULT_PREFIX!r})",
+    )
     args = parser.parse_args()
 
     claude_bin = resolve_claude(args.claude_bin)
@@ -347,6 +438,11 @@ def main():
         log(f"circuit breaker: {len(recent)} resumes in the last 24h (max {args.max_per_day})")
         return 0
 
+    strain = memory_strain(args.min_available_mb, args.max_swap_used_pct)
+    if strain:
+        log(f"holding off: {strain}")
+        return 0
+
     sessions = active_sessions(claude_bin)
     if sessions is None:
         return 1
@@ -358,7 +454,13 @@ def main():
         tried = ledger["attempts"].get(job["session_id"], 0)
         if tried >= args.max_attempts:
             continue
-        if not resume(job, claude_bin, args.prompt, args.dry_run):
+        # A session launched seconds ago has not grown into its memory yet, so
+        # charge every resume of this run for the headroom it is about to take.
+        strain = memory_strain(args.min_available_mb * (resumed + 1), args.max_swap_used_pct)
+        if strain:
+            log(f"holding off: {strain}")
+            break
+        if not resume(job, claude_bin, args.prompt, args.superseded_prefix, args.dry_run):
             continue
         ledger["attempts"][job["session_id"]] = tried + 1
         ledger["resumes"] = recent + [now.timestamp()]

test-resume-stalled-sessions.py

@@ -109,6 +109,45 @@ class ResetMoment(unittest.TestCase):
         self.assertIsNone(watchdog.reset_moment("API error — see detail", self.blocked_at))
 
 
+class MemoryStrain(unittest.TestCase):
+    """A host with no room left must not be given another session to run."""
+
+    def meminfo(self, available_kb, swap_total_kb=4194300, swap_free_kb=4194300):
+        temp = tempfile.NamedTemporaryFile("w", suffix=".meminfo", delete=False)
+        self.addCleanup(os.unlink, temp.name)
+        temp.write(
+            "MemTotal:       32877496 kB\n"
+            "MemFree:         1048576 kB\n"
+            f"MemAvailable:   {available_kb} kB\n"
+            f"SwapTotal:      {swap_total_kb} kB\n"
+            f"SwapFree:       {swap_free_kb} kB\n"
+        )
+        temp.close()
+        return Path(temp.name)
+
+    def test_healthy_host_takes_another_session(self):
+        self.assertIsNone(watchdog.memory_strain(2048, 50, self.meminfo(30083284)))
+
+    def test_low_memory_holds_off(self):
+        strain = watchdog.memory_strain(2048, 50, self.meminfo(1048576))
+        self.assertIn("1024 MB available", strain)
+
+    def test_swap_in_heavy_use_holds_off(self):
+        strain = watchdog.memory_strain(2048, 50, self.meminfo(30083284, swap_free_kb=1048575))
+        self.assertIn("swap 75% used", strain)
+
+    def test_swapless_host_is_judged_on_memory_alone(self):
+        self.assertIsNone(
+            watchdog.memory_strain(2048, 50, self.meminfo(30083284, swap_total_kb=0, swap_free_kb=0))
+        )
+
+    def test_host_that_does_not_report_memory_is_not_blocked(self):
+        self.assertIsNone(watchdog.memory_strain(2048, 50, Path("/nonexistent/meminfo")))
+        kernel_without_memavailable = self.meminfo(0)
+        kernel_without_memavailable.write_text("MemTotal:       32877496 kB\n")
+        self.assertIsNone(watchdog.memory_strain(2048, 50, kernel_without_memavailable))
+
+
 class Candidates(unittest.TestCase):
     """The whole path: a spend-limited background job with no process left."""
 
@@ -133,11 +172,50 @@ class Candidates(unittest.TestCase):
         session = self.blocked_job(self.SPEND, timedelta(hours=2))
         found = watchdog.candidates([session], self.jobs, self.now)
         self.assertEqual([job["id"] for job in found], [session["id"]])
+        self.assertEqual(found[0]["state_file"], self.jobs / session["id"] / "state.json")
 
     def test_fresh_spend_block_waits(self):
         session = self.blocked_job(self.SPEND, timedelta(minutes=10))
         self.assertEqual(watchdog.candidates([session], self.jobs, self.now), [])
 
 
+class MarkSuperseded(unittest.TestCase):
+    """The husk keeps its conversation, so its name must not read as the live one."""
+
+    def setUp(self):
+        temp = tempfile.TemporaryDirectory()
+        self.addCleanup(temp.cleanup)
+        self.state = Path(temp.name) / "state.json"
+
+    def write(self, **fields):
+        self.state.write_text(json.dumps(fields))
+
+    def read(self):
+        return json.loads(self.state.read_text())
+
+    def test_name_gains_the_prefix(self):
+        self.write(name="rewrite larvitsmpp typescript esm", state="blocked")
+        self.assertIsNone(watchdog.mark_superseded(self.state, "X "))
+        self.assertEqual(self.read()["name"], "X rewrite larvitsmpp typescript esm")
+
+    def test_the_respawn_name_stays_in_step(self):
+        self.write(name="fejkdata", respawnFlags=["--agent", "claude", "--name", "fejkdata"])
+        watchdog.mark_superseded(self.state, "X ")
+        self.assertEqual(self.read()["respawnFlags"], ["--agent", "claude", "--name", "X fejkdata"])
+
+    def test_a_husk_is_prefixed_only_once(self):
+        self.write(name="X fejkdata")
+        self.assertIsNone(watchdog.mark_superseded(self.state, "X "))
+        self.assertEqual(self.read()["name"], "X fejkdata")
+
+    def test_an_unnamed_job_is_left_alone(self):
+        self.write(state="blocked")
+        self.assertIsNone(watchdog.mark_superseded(self.state, "X "))
+        self.assertNotIn("name", self.read())
+
+    def test_unwritable_state_is_reported_rather_than_raised(self):
+        self.assertIsNotNone(watchdog.mark_superseded(self.state.with_name("gone.json"), "X "))
+
+
 if __name__ == "__main__":
     unittest.main()