control

A local socket a detached job runner listens on, so a host’s page can prompt, steer, stop and answer its agent from another process.

The socket

A program that hosts agent jobs, such as a web app, often runs each job in a process of its own, so that restarting the web server does not kill a turn halfway through. The page that started the job then needs a way to reach it: to send a prompt, steer the running turn, stop it, and answer the approval requests it raises. Control gives it that way in. It opens a Unix socket, an address in the file system that only processes on this machine can connect to, in the session’s run folder, and only the current user may connect.

The protocol is one JSON object per line in each direction. A request is {"id", "method", "params"}. The reply is {"id", "result"} on success and {"id", "error"} on failure. METHODS lists the methods:

  • prompt(text) starts a turn on a worker thread. If a turn is already running, text is queued to run after it.
  • steer(text) hands text to the running turn along with its next tool result, which redirects the model without stopping it.
  • cancel() stops the running turn and drops the prompts queued behind it.
  • asks() lists the approval requests waiting for an answer, oldest first. answer(id, ok, note) answers one of them.
  • status() says whether a turn is running, how many prompts are queued, and how the last turn ended.
  • events(offset) returns the records in the session’s event file from byte offset on, and the offset to pass next time. It works only after the runner has called agent.events_to(path).

control_call(path, method, **params) is the other end. It connects, sends one request, and returns the result, or raises RuntimeError with the error’s message. The page uses it, and so can any script that needs to reach a running job.

start refuses a path where another runner is already listening, and a path that holds something other than a socket. It removes a socket left behind by a runner that crashed. A Unix socket path may be at most about a hundred bytes long (SOCKET_MAX), so when the session’s run folder is deeper than that, the socket goes in the temporary folder as ramabana-<session>.sock.

source

control_call

def control_call(
    path, method, timeout:int=30, **params
):

Send one request to the Control socket at path and return its result; an error reply raises RuntimeError.

source

Control

def Control(
    agent, path:NoneType=None, # the socket; None puts it in the session's run folder as `CONTROL`
):

A Unix socket through which another process prompts, steers, stops and answers an agent running in this one.

The example plays both sides in one process. fake_agent from ramabana.testing supplies an agent whose model replies with a scripted line. The first cell sends a prompt through the socket, waits for the turn to finish, checks how it ended, and reads the turn’s events.

The second cell raises an approval request for edit_file from another thread, as a tool would, finds it through asks, and approves it through the socket. The agent was built with no_listener='wait', which keeps the request waiting for an answer rather than refusing it because no terminal is attached. An unknown method comes back as an error, which control_call raises as RuntimeError.

import tempfile, time
from ramabana import agent
from ramabana.testing import fake_agent

d = Path(tempfile.mkdtemp(dir='/tmp'))   # a Unix socket path must stay short
a, be = fake_agent(cfg=d, replies=['the first reply'], approvals=agent.Approvals(tools={'edit_file'}, timeout=5, no_listener='wait'))
a.events_to(d/'job.events')
ctl = Control(a).start()
call = lambda method, **kw: control_call(ctl.path, method, **kw)

test_eq(call('prompt', text='hello'), {'started': True})
for _ in range(200):
    if not call('status')['running']: break
    time.sleep(.01)
st = call('status')
test_eq((st['running'], st['last']['state'], st['last']['reply']), (False, 'complete', 'the first reply'))
ev = call('events')
assert {'turn_start', 'turn_end'} <= {e['kind'] for e in ev['events']} and ev['offset'] > 0
test_eq(call('events', offset=ev['offset'])['events'], [])
out = {}
threading.Thread(target=lambda: out.setdefault('r', a.approvals.request('edit_file', {'path': 'a.py'})), daemon=True).start()
for _ in range(200):
    if (waiting := call('asks')): break
    time.sleep(.01)
test_eq(waiting[0]['tool'], 'edit_file')
assert call('answer', id=waiting[0]['id'], ok=True)
for _ in range(200):
    if 'r' in out: break
    time.sleep(.01)
assert out['r'], 'answered through the socket'
try: call('fly')
except RuntimeError as e: assert 'no method' in str(e)
ctl.close(); a.close()