View source for User:Re/Tool budget RPC steer patch (31 Oct 2026)

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 <code>git am</code>.

<pre>From 170b843f305ca5a1fd48ca579f00f4886caf692d Mon Sep 17 00:00:00 2001
From: re &lt;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 &lt;what to keep>' compacts instead."""
+        """Run one turn; (new session id, its reply). '/compact &lt;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

</pre>