Compare commits

...

2 Commits

3 changed files with 126 additions and 10 deletions
+20 -2
View File
@@ -62,6 +62,20 @@ 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 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. working the same conversation — two agents racing on the same git worktree.
### 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 ### Circuit breakers
Three limits stop a broken session, or a still-ongoing outage, from spawning an Three limits stop a broken session, or a still-ongoing outage, from spawning an
@@ -77,7 +91,9 @@ endless chain of blocked jobs:
again it is a *new* session id and qualifies on its own, so an outage spanning 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 several reset windows still gets retried. What the cap prevents is one
conversation being resumed into several concurrent copies. A resume that fails 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 `--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 — usage-limit block is self-pacing, because the stated reset time is respected —
@@ -165,12 +181,14 @@ Every option is a CLI flag with an environment-variable fallback — nothing is
hardcoded. To make one permanent, add it to the crontab line. 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) |
| `--max-swap-used-pct` | `WATCHDOG_MAX_SWAP_USED_PCT` | `50` |
| `--min-available-mb` | `WATCHDOG_MIN_AVAILABLE_MB` | `2048` |
| `--prompt` | `WATCHDOG_PROMPT` | see `--help` | | `--prompt` | `WATCHDOG_PROMPT` | see `--help` |
| `--dry-run` | — | off | | `--dry-run` | — | off |
+59
View File
@@ -22,6 +22,8 @@ from zoneinfo import ZoneInfo
# Where `claude` installs itself, and what a scheduler's minimal PATH omits. # Where `claude` installs itself, and what a scheduler's minimal PATH omits.
CLAUDE_FALLBACK_DIRS = ("~/.local/bin", "/usr/local/bin", "/opt/homebrew/bin") CLAUDE_FALLBACK_DIRS = ("~/.local/bin", "/usr/local/bin", "/opt/homebrew/bin")
MEMINFO = Path("/proc/meminfo")
DEFAULT_PROMPT = ( DEFAULT_PROMPT = (
"Continue where you left off. Your previous run was interrupted by a " "Continue where you left off. Your previous run was interrupted by a "
"transient failure (usage limit, API outage, or network), which has cleared." "transient failure (usage limit, API outage, or network), which has cleared."
@@ -86,6 +88,37 @@ def log(message):
print(f"{datetime.now().astimezone().isoformat(timespec='seconds')} {message}", flush=True) 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): def resolve_claude(name):
"""Absolute path to the Claude Code binary, or None if it cannot be found.""" """Absolute path to the Claude Code binary, or None if it cannot be found."""
found = shutil.which(name) found = shutil.which(name)
@@ -274,6 +307,11 @@ def resume(job, claude_bin, prompt, dry_run):
command + [prompt], command + [prompt],
capture_output=True, cwd=cwd, text=True, timeout=180, check=True, 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: except (OSError, subprocess.SubprocessError) as err:
log(f"{job['id']}: resume failed: {err}") log(f"{job['id']}: resume failed: {err}")
return False return False
@@ -318,6 +356,16 @@ def main():
default=int(os.environ.get("WATCHDOG_MAX_PER_RUN", "0")), default=int(os.environ.get("WATCHDOG_MAX_PER_RUN", "0")),
help="resumes per invocation, 0 for every eligible session (default: 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("--prompt", default=os.environ.get("WATCHDOG_PROMPT", DEFAULT_PROMPT))
args = parser.parse_args() args = parser.parse_args()
@@ -347,6 +395,11 @@ def main():
log(f"circuit breaker: {len(recent)} resumes in the last 24h (max {args.max_per_day})") log(f"circuit breaker: {len(recent)} resumes in the last 24h (max {args.max_per_day})")
return 0 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) sessions = active_sessions(claude_bin)
if sessions is None: if sessions is None:
return 1 return 1
@@ -358,6 +411,12 @@ def main():
tried = ledger["attempts"].get(job["session_id"], 0) tried = ledger["attempts"].get(job["session_id"], 0)
if tried >= args.max_attempts: if tried >= args.max_attempts:
continue continue
# 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.dry_run): if not resume(job, claude_bin, args.prompt, args.dry_run):
continue continue
ledger["attempts"][job["session_id"]] = tried + 1 ledger["attempts"][job["session_id"]] = tried + 1
+39
View File
@@ -109,6 +109,45 @@ class ResetMoment(unittest.TestCase):
self.assertIsNone(watchdog.reset_moment("API error — see detail", self.blocked_at)) 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): class Candidates(unittest.TestCase):
"""The whole path: a spend-limited background job with no process left.""" """The whole path: a spend-limited background job with no process left."""