Hold off resuming while the host is short on memory or swapping
This commit is contained in:
@@ -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
|
||||||
@@ -165,12 +179,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 |
|
||||||
|
|
||||||
|
|||||||
@@ -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)
|
||||||
@@ -318,6 +351,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 +390,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 +406,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
|
||||||
|
|||||||
@@ -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."""
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user