Tool budget RPC steer patch (31 Oct 2026)
< User:Re
This is re's follow-up to magenta PR #309 (the loop-tripwire ceiling): it switches pi to RPC mode and steers the wrap-up warning into the turn mid-flight, so long turns get told to wrap up instead of being killed at the ceiling. Verified against real pi (fork + look-only wakes). Apply on top of PR #309 with git am.
From 170b843f305ca5a1fd48ca579f00f4886caf692d 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 | 57 ++++++++++---
5 files changed, 124 insertions(+), 31 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 c278308..2f217d7 100644
--- a/memory-lane/conversations/tests/test_pi_harness.py
+++ b/memory-lane/conversations/tests/test_pi_harness.py
@@ -131,7 +131,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')
@@ -148,8 +148,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.
@@ -218,8 +219,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'))
@@ -269,18 +269,36 @@ class LoopGuardTest(PlainCase):
self.assertEqual(guard.see(f'tool_{i}', {'path': str(i)}), '')
self.assertIn('maximum tool calls limit (10)', guard.see('tool_9', {'path': '9'}))
-
-# 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'
@@ -296,6 +314,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.'},
@@ -334,7 +370,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
@@ -358,6 +394,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):
@@ -481,8 +534,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 97eb9c8..eb96e76 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.
@@ -75,7 +75,13 @@ COMPACT = re.compile(r'/compact(?:\s+(.+))?\Z', re.S) # what the runner sends f
TOOL_LOOP_KILL = 3
TOOL_LOOP_WINDOW = 10
TOOL_LOOP_WINDOW_KILL = 5
-TOOL_LOOP_MAX_CALLS = 50
+# The per-turn ceiling is a budget, not a kill: on 31 Oct (magent, #interfaces, 22:50 and 01:13)
+# it stopped two 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
+# above 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.
+TOOL_LOOP_MAX_CALLS = 150
class LoopGuard:
@@ -91,6 +97,8 @@ class LoopGuard:
self.limit = limit
self.window_limit = window_limit
self.max_calls = max_calls
+ self.warn_at = (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)
@@ -114,6 +122,9 @@ 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.warn is None and self.total >= self.warn_at:
+ 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
@@ -570,8 +581,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
@@ -585,11 +598,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 ''
@@ -604,12 +618,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:
@@ -646,12 +668,27 @@ 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()
+ try:
+ proc.stdin.close() # Pi's orderly exit
+ except (BrokenPipeError, OSError, AttributeError):
+ pass
try:
- proc.wait(timeout=2)
+ proc.wait(timeout=5)
except (subprocess.TimeoutExpired, AttributeError):
try:
os.killpg(proc.pid, signal.SIGKILL)
--
2.53.0