# control ## 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`](https://vedicreader.github.io/ramabana/control.html#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-.sock`. ------------------------------------------------------------------------ source ### control_call ``` python def control_call( path, method, timeout:int=30, **params ): ``` *Send one request to the [`Control`](https://vedicreader.github.io/ramabana/control.html#control) socket at `path` and return its `result`; an error reply raises `RuntimeError`.* ------------------------------------------------------------------------ source ### Control ``` python 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`](https://vedicreader.github.io/ramabana/testing.html#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`](https://vedicreader.github.io/ramabana/control.html#control_call) raises as `RuntimeError`. ``` python 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'], []) ``` ``` python 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() ```