Tool budget RPC steer, merged onto PR 309 (31 Oct 2026)
This is magent's merge of re's RPC-steer delta (the original patch) adapted onto magent's PR #309 head (147538f, with the constants living in wake_policy.py and the 10s SIGTERM grace). It switches pi to RPC mode, puts the prompt on stdin, and steers the 2/3 wrap-up warning into the turn mid-flight, so long turns get told to wrap up instead of being killed at the ceiling. Verified: 48 harness tests green, full suite 1507 tests green (one pre-existing environmental failure), plus a real-pi tool-call wake settling cleanly. Apply on top of PR #309 with git am.
From 8a17392777a4f000c926e431288eb98806aef3e1 Mon Sep 17 00:00:00 2001
From: re <re@cryptograss.live>
Date: Sat, 31 Oct 2026 02:05:21 +0000
Subject: [PATCH] Tool budget: the per-turn cap warns instead of assassinating
(150, steer at 2/3)
The LoopGuard's 50-call ceiling stopped two of magent's productive turns
cold on 31 Oct (#interfaces, 22:50 and 01:13) -- big PR work, never a loop;
the kill is silent, so the model cannot wrap up. The attractor detectors
(3x identical, 5-of-10 window) stay killing. The ceiling becomes a 150-call
budget: at 2/3 of it the pi harness steers a wrap-up warning in over RPC
(steer lands before the next model call; the record keeps it as the
poller's user message) and the turn ends when the agent wraps up, not by a
kill. A run that ignores the warning still trips the ceiling and dies.
Claude Code's -p has no stdin: there the budget only moves, no warning.
PiWaker.command is RPC mode now, with the prompt on stdin -- verified on
real pi, fork and look-only paths both.
---
.../tests/test_harness_models.py | 2 +-
.../tests/test_instructions_prompt.py | 6 +-
.../conversations/tests/test_pi_harness.py | 85 +++++++++++++++----
memory-lane/poller/mood_poller.py | 5 +-
memory-lane/poller/pi_harness.py | 61 ++++++++++---
5 files changed, 126 insertions(+), 33 deletions(-)
diff --git a/memory-lane/conversations/tests/test_harness_models.py b/memory-lane/conversations/tests/test_harness_models.py
index ca529e7..5634eca 100644
--- a/memory-lane/conversations/tests/test_harness_models.py
+++ b/memory-lane/conversations/tests/test_harness_models.py
@@ -79,7 +79,7 @@ class PiCatalogTest(TestCase):
def test_the_menus_names_mean_anthropics_models_on_pi(self):
waker = PiWaker(pi='pi')
- model_of = lambda m: (lambda c: c[c.index('--model') + 1])(waker.command(None, 'n', 'hi', model=m))
+ model_of = lambda m: (lambda c: c[c.index('--model') + 1])(waker.command(None, 'n', model=m))
self.assertEqual(model_of('opus'), 'anthropic/claude-opus-5-5') # never Bedrock's, by fuzzy match
self.assertEqual(model_of('sonnet'), 'anthropic/claude-sonnet-5-5')
self.assertEqual(model_of('openrouter/moonshotai/kimi-k3'), 'openrouter/moonshotai/kimi-k3')
diff --git a/memory-lane/conversations/tests/test_instructions_prompt.py b/memory-lane/conversations/tests/test_instructions_prompt.py
index 38ed8cb..42f4847 100644
--- a/memory-lane/conversations/tests/test_instructions_prompt.py
+++ b/memory-lane/conversations/tests/test_instructions_prompt.py
@@ -46,14 +46,14 @@ class RulesPromptTest(SimpleTestCase):
self.assertIn('Never print a credential.', waker.options['system'])
def test_both_harnesses_append_it_to_their_own(self):
- pi = PiWaker(pi='pi', model='openrouter/x', session_dir='/tmp/s').command('old', 'new', 'hi', system=RULES)
+ pi = PiWaker(pi='pi', model='openrouter/x', session_dir='/tmp/s').command('old', 'new', system=RULES)
self.assertEqual(pi[pi.index('--append-system-prompt') + 1], RULES)
- self.assertEqual(pi[-1], 'hi')
+ self.assertNotIn('hi', pi) # pi is RPC mode now: the prompt rides stdin, not argv
claude = ClaudeCodeWaker(claude='claude').command('old', 'new', 'hi', system=RULES)
self.assertEqual(claude[claude.index('--append-system-prompt') + 1], RULES)
self.assertEqual(claude[claude.index('--system-prompt-snapshot') + 1], 'off')
self.assertEqual(claude[-2:], ['--', 'hi'])
- self.assertNotIn('--append-system-prompt', PiWaker(pi='pi', session_dir='/tmp/s').command('o', 'n', 'hi'))
+ self.assertNotIn('--append-system-prompt', PiWaker(pi='pi', session_dir='/tmp/s').command('o', 'n'))
without = ClaudeCodeWaker(claude='claude').command('o', 'n', 'hi')
self.assertEqual(without[without.index('--system-prompt-snapshot') + 1], 'off')
diff --git a/memory-lane/conversations/tests/test_pi_harness.py b/memory-lane/conversations/tests/test_pi_harness.py
index 5dccd62..475a827 100644
--- a/memory-lane/conversations/tests/test_pi_harness.py
+++ b/memory-lane/conversations/tests/test_pi_harness.py
@@ -99,7 +99,7 @@ class CommandTest(TestCase):
def test_look_only_reads_and_a_fork_names_its_parent(self):
waker = PiWaker(pi='pi', model='local/qwen', session_dir='/tmp/s')
- cmd = waker.command('old', 'new', 'hi', effort='max')
+ cmd = waker.command('old', 'new', effort='max') # RPC mode: the prompt goes on stdin, not argv
self.assertEqual(cmd[:2], ['pi', '--mode'])
self.assertIn('--fork', cmd)
self.assertEqual(cmd[cmd.index('--session-id') + 1], 'new')
@@ -116,8 +116,9 @@ class CommandTest(TestCase):
self.assertFalse([t for t in mcp if any(w in t for w in
('create', 'update', 'delete', 'move', 'protect', 'upload', 'browser'))])
self.assertEqual(cmd[cmd.index('--thinking') + 1], 'xhigh')
- self.assertNotIn('--tools', waker.command(None, 'new', 'hi', full=True))
- self.assertEqual(cmd[-1], 'hi')
+ self.assertNotIn('--tools', waker.command(None, 'new', full=True))
+ self.assertEqual(cmd[1:3], ['--mode', 'rpc'])
+ self.assertNotIn('hi', cmd) # the prompt never rides the command line: stdin carries it
# A Pi in RPC mode, as Pi 1.1.0 answers `compact` (tried on hunter, 18 Oct): its events, then the response.
@@ -186,8 +187,7 @@ class CompactTest(TestCase):
self.assertEqual(self.waker.last_result['subtype'], 'error')
def test_an_ordinary_wake_that_mentions_compact_is_a_turn(self):
- self.assertEqual(self.waker.command(SESSION, 'n', 'You were woken: /compact re keep it')[-1],
- 'You were woken: /compact re keep it')
+ self.assertNotIn('/compact', self.waker.command(SESSION, 'n')[-1]) # the prompt rides stdin, not argv
self.assertIsNone(__import__('poller.pi_harness', fromlist=['COMPACT']).COMPACT.match('woken: /compact x'))
@@ -241,18 +241,36 @@ class LoopGuardTest(PlainCase):
guard = LoopGuard()
self.assertEqual(guard.max_calls, 150)
-
-# A pi stuck in a machine attractor, as re's was on 28 Oct: the same read over and over,
-# with escape text every time, until something kills it.
+ def test_the_budget_warns_at_two_thirds_without_killing(self):
+ # 31 Oct: the 50-call ceiling stopped two productive magent turns cold (#interfaces).
+ # Now the ceiling is a 150-call budget, and at 2/3 of it the guard hands the harness
+ # a wrap-up warning to steer in -- the turn goes on unless a loop trips first.
+ guard = LoopGuard()
+ for i in range(99):
+ self.assertEqual(guard.see(f'read_{i}', {'path': f'f{i}'}), '')
+ self.assertIsNone(guard.warn)
+ self.assertEqual(guard.see('read_99', {'path': 'f99'}), '') # call 100: warned, not killed
+ self.assertEqual(guard.warn.split('.')[0], 'tool budget: 100 of 150 tool calls used this turn')
+ for i in range(100, 149): # the warning fires once; the turn lives on to the ceiling
+ self.assertEqual(guard.see(f'read_{i}', {'path': f'f{i}'}), '')
+ self.assertIn('tool budget: 100 of 150', guard.warn)
+ self.assertIn('maximum tool calls limit (150)', guard.see('read_149', {'path': 'f149'}))
+
+
+# A pi in RPC mode, as the waker now runs it (pi_harness, 31 Oct): the prompt arrives on stdin.
+# Its loop mode is a machine attractor, as re's was on 28 Oct: the same read over and over, with
+# escape text every time, until something kills it.
FAKE_LOOP_PI = r'''#!PYTHON
import json, sys, time
-lines = [
- {'type': 'session', 'version': 3, 'id': 'x', 'timestamp': '2026-10-30T03:00:00.000Z'},
- {'type': 'turn_start'},
-]
+mode = json.loads(sys.stdin.readline())['message']
+
+# RPC pi echoes the prompt as a user message (the translator turns it into the wake's user_message).
+print(json.dumps({'type': 'message_end', 'message': {'role': 'user', 'timestamp': 1793519999000,
+ 'content': [{'type': 'text', 'text': mode}]}}), flush=True)
+
+lines = [{'type': 'turn_start'}]
-mode = sys.argv[-1]
if 'alternating' in mode:
for i in range(10):
tool = 'grep' if i % 2 == 0 else 'read'
@@ -268,6 +286,24 @@ elif 'normal' in mode:
for line in lines:
print(json.dumps(line), flush=True)
sys.exit(0)
+elif 'steering' in mode:
+ # 120 distinct calls: past the 2/3 warning (100), short of the 150 ceiling. Print 100
+ # first (the runner steers the warning at call 100), then read the steer -- a readline
+ # before the prints would deadlock: the runner would wait for calls never printed.
+ for i in range(100):
+ print(json.dumps({'type': 'message_end', 'message': {'role': 'assistant', 'stopReason': 'toolUse',
+ 'usage': {}, 'content': [{'type': 'toolCall', 'id': f'c{i}', 'name': 'read',
+ 'arguments': {'path': f'f{i}'}}]}}), flush=True)
+ steer = json.loads(sys.stdin.readline()) # blocks until the runner steers it in
+ open(sys.argv[0] + '.steered', 'w').write(json.dumps(steer))
+ lines = [{'type': 'message_end', 'message': {'role': 'user', 'timestamp': 1793520000000,
+ 'content': [{'type': 'text', 'text': steer['message']}]}},
+ {'type': 'message_end', 'message': {'role': 'assistant', 'stopReason': 'stop', 'usage': {},
+ 'content': [{'type': 'text', 'text': 'Wrapped up early.'}]}},
+ {'type': 'agent_settled', 'aborted': False}]
+ for line in lines:
+ print(json.dumps(line), flush=True)
+ sys.exit(0)
else:
call = {'type': 'message_end', 'message': {'role': 'assistant', 'stopReason': 'toolUse', 'usage': {},
'content': [{'type': 'text', 'text': 'Let me check that file again.'},
@@ -306,7 +342,7 @@ class LoopKillTest(PlainCase):
self.assertIn('tool loop', self.waker.last_result['error'])
# The record keeps what the loop did: every call, escape text and all, plus loop notice
self.assertEqual([e['type'] for e in events],
- ['turn_started'] + ['assistant_message', 'tool_call'] * 3 + ['assistant_message'])
+ ['turn_started', 'user_message'] + ['assistant_message', 'tool_call'] * 3 + ['assistant_message'])
self.assertIn('⚠️ Turn stopped by loop tripwire', events[-1]['text'])
self.assertLess(time.time() - start, 10) # killed, not slept out
@@ -330,6 +366,23 @@ class LoopKillTest(PlainCase):
self.assertEqual(reply, 'Done reading.')
self.assertEqual(self.waker.last_result['subtype'], 'success')
+ def test_the_tool_budget_warns_mid_turn_and_the_turn_survives(self):
+ # 31 Oct, magent in #interfaces: two productive turns died at the 50-call ceiling with no
+ # word to anyone. Now the ceiling is a 150-call budget; at call 100 the harness steers
+ # the warning in, pi delivers it as the poller's user message, and the turn ends when
+ # the agent wraps up -- not by a kill.
+ events = []
+ _, reply = self.waker.wake(SESSION, 'steering test', new_session_id='new-12', on_event=events.append)
+ self.assertEqual(reply, 'Wrapped up early.')
+ self.assertEqual(self.waker.last_result['subtype'], 'success')
+ steer = json.loads(Path(self.pi + '.steered').read_text())
+ self.assertEqual((steer.get('type'), steer.get('message').split('.')[0]),
+ ('steer', 'tool budget: 100 of 150 tool calls used this turn'))
+ warned = [e for e in events if e['type'] == 'user_message']
+ self.assertEqual([e['text'] for e in warned],
+ ['steering test', 'tool budget: 100 of 150 tool calls used this turn. '
+ 'Wrap up soon: finish what is in hand, then post your reply and stop calling tools.'])
+
@override_settings(MOOD_RUNNER_KEYS={'re': KEY}, TOOL_RESULT_CONTENT_CHARS=20000)
class EndToEndTest(TestCase):
@@ -453,8 +506,8 @@ class EndToEndTest(TestCase):
self.assertFalse(is_router(model), model)
waker = PiWaker(pi='pi', session_dir=tempfile.mkdtemp())
with self.assertRaisesRegex(RuntimeError, 'router'):
- waker.command(None, 's1', 'hi', model='openrouter/openrouter/pareto-code')
- self.assertIn('--model', waker.command(None, 's1', 'hi', model='openrouter/deepseek/deepseek-v4-pro'))
+ waker.command(None, 's1', model='openrouter/openrouter/pareto-code')
+ self.assertIn('--model', waker.command(None, 's1', model='openrouter/deepseek/deepseek-v4-pro'))
def test_a_wake_that_failed_closes_its_turn(self):
client, posted = self.client, []
diff --git a/memory-lane/poller/mood_poller.py b/memory-lane/poller/mood_poller.py
index 94f8509..e87fc0b 100644
--- a/memory-lane/poller/mood_poller.py
+++ b/memory-lane/poller/mood_poller.py
@@ -697,7 +697,10 @@ class ClaudeCodeWaker:
watchdog.start()
result = None
said = '' # the last text the agent itself wrote (not a helper's)
- guard, killed = LoopGuard(), None # the tool-loop tripwire (TOOL_LOOP_KILL in pi_harness)
+ # The tool-loop tripwire (pi_harness.LoopGuard): the attractor detectors kill, and the
+ # per-turn ceiling is a 150-call budget that still ends the turn -- Claude Code's `-p`
+ # has no stdin, so unlike pi there is no mid-turn warning here (pi_harness, 31 Oct).
+ guard, killed = LoopGuard(), None
try:
for raw in proc.stdout:
heard[0] = time.monotonic()
diff --git a/memory-lane/poller/pi_harness.py b/memory-lane/poller/pi_harness.py
index 4a9ce99..b7e9b95 100644
--- a/memory-lane/poller/pi_harness.py
+++ b/memory-lane/poller/pi_harness.py
@@ -1,7 +1,7 @@
"""Pi (https://github.com/earendil-works/pi) as a harness the runner can wake.
Pi is one open-source harness for many model providers. The runner starts a
-turn with `pi --mode json`, the way it starts Claude Code with `claude -p`,
+turn with `pi --mode rpc`, the way it starts Claude Code with `claude -p`,
and this module translates what Pi prints into record events
(docs/RECORD_EVENTS.md), memory-lane's own format, before anything is
posted. Nothing of Pi's format reaches memory-lane.
@@ -77,6 +77,12 @@ LOOK_ONLY_TOOLS = _look_only_tools()
COMPACT = re.compile(r'/compact(?:\s+(.+))?\Z', re.S) # what the runner sends for "/compact re [what to keep]"
# Seconds to let a stopped turn wind down, its in-flight tool included, before SIGKILL.
TURN_STOP_GRACE = 10
+# The per-turn ceiling (TOOL_LOOP_MAX_CALLS, in wake_policy.py) is a budget, not a loop: on
+# 31 Oct it stopped two of magent's productive turns cold -- big PR work, never a loop. So it
+# sits high, and at 2/3 of it the turn gets a warning, steered in over RPC (pi's `steer` lands
+# before the next model call, and the record keeps it as the poller's user message). Only the
+# attractor detectors kill; a run that ignores the warning still trips the ceiling and dies.
+# Claude Code's `-p` has no stdin: there the budget only moves, no warning.
class LoopGuard:
@@ -92,6 +98,8 @@ class LoopGuard:
self.limit = limit
self.window_limit = window_limit
self.max_calls = TOOL_LOOP_MAX_CALLS if max_calls is None else max_calls
+ self.warn_at = (self.max_calls * 2) // 3 # where the wrap-up warning goes out (2/3 of the budget)
+ self.warn = None # set once, at warn_at: the pi harness delivers it as an RPC steer
self.streak, self.last = 0, None
self.total = 0
self.recent = deque(maxlen=window_size)
@@ -115,8 +123,10 @@ class LoopGuard:
return f'tool loop: {name} called {window_count}× within last {len(self.recent)} calls'
if self.total >= self.max_calls:
return f'tool loop: turn exceeded maximum tool calls limit ({self.total})'
- if self.total == int(self.max_calls * 2 / 3):
+ if self.warn is None and self.total >= self.warn_at:
logger.warning(f'turn approaching call ceiling: {self.total}/{self.max_calls} tool calls used')
+ self.warn = (f'tool budget: {self.total} of {self.max_calls} tool calls used this turn. '
+ 'Wrap up soon: finish what is in hand, then post your reply and stop calling tools.')
return ''
EFFORTS = {'max': 'xhigh'} # our effort names that Pi spells differently
# The names the model menu has always offered, as Pi should read them. Left
@@ -573,8 +583,10 @@ class PiWaker:
threading.Thread(target=watch, daemon=True).start()
return done
- def command(self, session_id, new_session_id, prompt, model=None, effort=None, full=False, system=None):
- cmd = [self.pi, '--mode', 'json', '--session-dir', str(self.session_dir), '--session-id', new_session_id]
+ def command(self, session_id, new_session_id, model=None, effort=None, full=False, system=None):
+ """The argv for one wake's pi process: RPC mode, so the runner prompts it on stdin and
+ can steer it mid-turn (the tool budget's wrap-up warning)."""
+ cmd = [self.pi, '--mode', 'rpc', '--session-dir', str(self.session_dir), '--session-id', new_session_id]
if session_id:
cmd += ['--fork', session_id]
model = model or self.model
@@ -588,11 +600,12 @@ class PiWaker:
cmd += ['--tools', LOOK_ONLY_TOOLS] # look-only: reads and the allowlisted MCP tools (wake_policy.py)
if system: # its instructions (mood_poller.MoodPoller.instructions_for): after Pi's own, never in the session
cmd += ['--append-system-prompt', system]
- return cmd + [prompt]
+ return cmd
def wake(self, session_id, prompt, new_session_id=None, on_event=None, model=None, effort=None, full=False,
stop=None, system=None, **ignored):
- """Run one turn; (new session id, its reply). '/compact <what to keep>' compacts instead."""
+ """Run one turn; (new session id, its reply). '/compact <what to keep>' compacts instead.
+ The prompt goes in on stdin (RPC mode), and LoopGuard's tool-budget warning is steered in mid-turn."""
keep = COMPACT.match(prompt or '')
if keep and session_id:
instructions = keep.group(1).strip() if (keep and keep.group(1)) else ''
@@ -607,12 +620,20 @@ class PiWaker:
new_session_id = new_session_id or str(uuid.uuid4())
self.session_dir.mkdir(parents=True, exist_ok=True)
translator = PiTranslator(new_session_id, effort=effort)
- cmd = self.command(session_id, new_session_id, prompt, model=model, effort=effort, full=full, system=system)
- proc = subprocess.Popen(cmd, cwd=cwd, stdin=subprocess.DEVNULL, stdout=subprocess.PIPE,
+ cmd = self.command(session_id, new_session_id, model=model, effort=effort, full=full, system=system)
+ proc = subprocess.Popen(cmd, cwd=cwd, stdin=subprocess.PIPE, stdout=subprocess.PIPE,
stderr=subprocess.PIPE, text=True, start_new_session=True)
done = self._watch(proc, stop)
- guard, killed = LoopGuard(), None # the tool-loop tripwire (TOOL_LOOP_KILL)
+ guard, killed, steered = LoopGuard(), None, False # the tool-loop tripwire (TOOL_LOOP_KILL)
try:
+ proc.stdin.write(json.dumps({'id': 'turn', 'type': 'prompt', 'message': prompt}) + '\n')
+ proc.stdin.flush()
+ # RPC mode streams no session record, so the turn starts here (as compact does).
+ if on_event:
+ try:
+ on_event(translator.event('turn_started', iso(time.time() * 1000), cwd=cwd, effort=effort))
+ except Exception as e: # never let reporting stop the turn
+ logger.warning(f'on_event failed: {e}')
for line in proc.stdout: # Pi frames on LF only; one JSON object per line
done.heard = time.monotonic()
try:
@@ -649,13 +670,29 @@ class PiWaker:
proc.terminate()
except (OSError, ProcessLookupError, AttributeError):
pass
+ elif guard.warn and not steered:
+ # The tool budget's wrap-up warning: pi delivers a steer before the
+ # next model call, and the record keeps it as the poller's user message.
+ steered = True
+ try:
+ proc.stdin.write(json.dumps({'type': 'steer', 'message': guard.warn}) + '\n')
+ proc.stdin.flush()
+ except (BrokenPipeError, OSError, AttributeError):
+ pass
if killed:
break
+ if translator.settled:
+ break # RPC pi lives on after its run; closing stdin is its orderly exit
finally:
done.set()
- # SIGTERM asks; SIGKILL is the last resort. A turn stopped mid-tool leaves the tool's own
- # child running (a `git commit`, a test run), and SIGKILL to the group murders it mid-write:
- # magent's commit died 137 that way on 31 Oct. So let the polite signal land first.
+ # Closing stdin is Pi's orderly exit in RPC mode (it lives on after its run). SIGTERM asks;
+ # SIGKILL is the last resort: a turn stopped mid-tool leaves the tool's own child running
+ # (a `git commit`, a test run), and SIGKILL to the group murders it mid-write -- magent's
+ # commit died 137 that way on 31 Oct. So let the polite signal land first.
+ try:
+ proc.stdin.close()
+ except (BrokenPipeError, OSError, AttributeError):
+ pass
try:
proc.wait(timeout=TURN_STOP_GRACE)
except (subprocess.TimeoutExpired, AttributeError):
--
2.53.0