Class: Rubino::Tools::ShellRegistry

Inherits:
Object
  • Object
show all
Defined in:
lib/rubino/tools/shell_registry.rb

Overview

Process-wide registry for shell commands started with run_in_background. Each entry owns a pgid (process group), a reader thread that drains stdout+stderr into an in-memory ring buffer, and the wait_thr for exit.

The registry survives a single CLI/server process — it is intentionally NOT persisted to disk. Background shells die with the agent process.

Defined Under Namespace

Classes: Entry

Constant Summary collapse

RING_BYTES =

cap per run; older bytes are dropped

256 * 1024
RETIRED_TTL =

A backgrounded command that FINISHES before the next turn used to be dropped from the registry the moment a reader (shell_output/tail/kill) saw it non-running — which also collapsed any? to false, so the shell-management tools vanished from the schema next turn and the model could never fetch a short bg command's captured output (#78). Instead a finished entry is RETIRED: it stays in the registry (its buffer + exit status intact, retrievable by shell_output) and any? keeps the tools exposed, until it is read again OR these bounds reap it. RETIRED_TTL caps how long a finished-but-unread entry lingers; MAX_RETIRED caps how many we keep at once (oldest-retired evicted first) so the registry stays bounded across a long session.

300
MAX_RETIRED =

seconds a finished, unread bg shell stays retrievable

16

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize ⇒ ShellRegistry

Returns a new instance of ShellRegistry.



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
 
# File 'lib/rubino/tools/shell_registry.rb', line 74

def initialize
  @entries = {}
  @mutex   = Mutex.new
  # Live FOREGROUND shell process groups, keyed by pgid. A foreground
  # shell's pgid otherwise lives only in the ShellTool#execute_foreground
  # stack frame of the (sub)agent thread that started it — so on
  # parent-death there is nothing process-wide to reap it and it
  # reparents to init as an orphan (MED-2). Tracking it here lets
  # #kill_all_groups SIGTERM/SIGKILL it synchronously on teardown.
  @fg_pgids = {}
  # Lock-free, atomically-swapped snapshot of every live shell pgid
  # (background entries + tracked foreground pgids). The SIGTERM/SIGHUP
  # teardown trap (#478) reaps the child groups, but it CANNOT take the
  # mutex above — Ruby forbids Mutex#synchronize from a trap context
  # (ThreadError). The writers always rebuild this frozen Array UNDER the
  # mutex; the trap reads it with a single, lock-free ivar read (an atomic
  # reference load in MRI) and never iterates a structure another thread
  # is mutating. See #kill_all_groups_trap_safe.
  @pgid_snapshot = [].freeze
end

Class Method Details

.instance ⇒ Object 



62
63
64
 
# File 'lib/rubino/tools/shell_registry.rb', line 62

def instance
  @instance ||= new
end

.reset! ⇒ Object

Test seam: drop the process-wide registry between examples so the situational shell-tool gate (#313) starts each spec with no background shell. Mirrors BackgroundTasks.reset!.



69
70
71
 
# File 'lib/rubino/tools/shell_registry.rb', line 69

def reset!
  @instance = nil
end

Instance Method Details

#any? ⇒ Boolean

True when at least one background shell is RUNNING or has finished but is still retained (retired, unread, within TTL — see #retire). The session-stable signal #313 gates the shell-management tools on this: a normal turn with no background shell never ships shell_input/shell_output/shell_tail/shell_kill, but a SHORT bg command that finished before the next turn keeps shell_output exposed so the model can still fetch its captured output (#78). Prunes stale retired entries first so the gate closes once nothing is reachable.

Returns:

  • (Boolean) 


215
216
217
218
219
220
 
# File 'lib/rubino/tools/shell_registry.rb', line 215

def any?
  @mutex.synchronize do
    prune_retired
    !@entries.empty?
  end
end

#close_stdin(entry) ⇒ Object

Closes the write end of the child's stdin (sends EOF). Idempotent.



270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
 
# File 'lib/rubino/tools/shell_registry.rb', line 270

def close_stdin(entry)
  io = entry&.stdin
  return if io.nil? || io.closed?

  # On a PTY, closing the master fd SIGHUPs the child. So while the child is
  # ALIVE, signal EOF without killing by sending EOT (Ctrl-D) — a line reader
  # in canonical mode at line-start treats it as end-of-input (a raw-mode
  # child sees a literal byte; that's out of scope). Once the child is GONE,
  # there is nothing to EOF and writing the master raises Errno::EIO — so
  # close the master fd instead, which also reclaims it (it is a SEPARATE fd
  # from the reader's, otherwise leaked until GC).
  if entry&.pty && running?(entry)
    io.write("\x04")
    io.flush
  else
    io.close
  end
rescue IOError, Errno::EIO, Errno::EBADF
  # already closed / child gone — nothing to flush
end

#exit_code(entry) ⇒ Object 



341
342
343
344
345
 
# File 'lib/rubino/tools/shell_registry.rb', line 341

def exit_code(entry)
  return nil if running?(entry)

  entry.wait_thr.value.exitstatus
end

#find(id) ⇒ Object 



203
204
205
 
# File 'lib/rubino/tools/shell_registry.rb', line 203

def find(id)
  @mutex.synchronize { @entries[id] }
end

#kill_all_groups(grace: 0.5) ⇒ Object

Synchronous teardown reaper (MED-2): SIGTERM every live shell process group this session owns — the background ENTRIES and the tracked FOREGROUND pgids — give them a brief grace, then SIGKILL any straggler. Mirrors the Python Hermes _kill_process (os.killpg SIGTERM → wait → SIGKILL). Called from BackgroundTasks#cancel_all so EVERY parent-death edge (clean quit ensure, HUP/TERM trap, REPL break) reaps the child shells the cooperative cancel token alone can't reach before the process exits and the shells reparent to init. Returns the pgids it signalled.

TRAP-SAFE (#478): reads the lock-free @pgid_snapshot — never Mutex#synchronize, which Ruby forbids from a signal-trap context (ThreadError). So the SIGTERM/SIGHUP teardown trap can call this directly. Process.kill and sleep are both async-signal-safe.



404
405
406
407
408
409
410
411
412
 
# File 'lib/rubino/tools/shell_registry.rb', line 404

def kill_all_groups(grace: 0.5)
  pgids = @pgid_snapshot
  return pgids if pgids.empty?

  pgids.each { |pgid| signal_group("TERM", pgid) }
  sleep(grace) if grace.positive?
  pgids.each { |pgid| signal_group("KILL", pgid) }
  pgids
end

#listable_entries ⇒ Object

Running PLUS retired (finished-but-retained) shells — the set shown in the /agents list + /status count, mirroring how finished subagents linger.



355
356
357
 
# File 'lib/rubino/tools/shell_registry.rb', line 355

def listable_entries
  @mutex.synchronize { @entries.values.dup }
end

#read_all(entry) ⇒ Object 



301
302
303
 
# File 'lib/rubino/tools/shell_registry.rb', line 301

def read_all(entry)
  entry.mutex.synchronize { entry.buffer.dup }
end

#read_new(entry) ⇒ Object

Reads accumulated bytes since the last read_new call. Returns the full snapshot if since is nil. Thread-safe.



293
294
295
296
297
298
299
 
# File 'lib/rubino/tools/shell_registry.rb', line 293

def read_new(entry)
  entry.mutex.synchronize do
    snapshot = entry.buffer.byteslice(entry.read_offset..) || ""
    entry.read_offset = entry.buffer.bytesize
    snapshot
  end
end

#register_pgid(pgid) ⇒ Object

Track a live foreground shell process group so teardown can reap it.



96
97
98
99
100
101
102
 
# File 'lib/rubino/tools/shell_registry.rb', line 96

def register_pgid(pgid)
  @mutex.synchronize do
    @fg_pgids[pgid] = true
    refresh_pgid_snapshot
  end
  pgid
end

#remove(id) ⇒ Object 



222
223
224
225
226
227
228
229
230
 
# File 'lib/rubino/tools/shell_registry.rb', line 222

def remove(id)
  entry = @mutex.synchronize do
    e = @entries.delete(id)
    refresh_pgid_snapshot
    e
  end
  close_stdin(entry) if entry
  entry
end

#retire(id) ⇒ Object

Retires a FINISHED background shell instead of dropping it (#78): the entry stays in the registry — its captured output + exit status intact and retrievable by a later shell_output — and any? keeps the shell-management tools exposed, so a short bg command's output is still reachable on the next turn. The process is already dead, so its pgid is cleared from the teardown snapshot and its stdin closed. Bounded by RETIRED_TTL / MAX_RETIRED (pruned here and in #any?). Stamps retired_at on the first retire and is idempotent — a second read of a retired entry keeps the original timestamp so a re-read can't extend its lifetime indefinitely. No-op for an unknown or still-running id.



242
243
244
245
246
247
248
249
250
251
252
253
254
 
# File 'lib/rubino/tools/shell_registry.rb', line 242

def retire(id)
  @mutex.synchronize do
    entry = @entries[id]
    return nil unless entry
    return entry if entry.retired_at # already retired — keep original TTL clock

    entry.retired_at = Time.now
    close_stdin(entry)    # process is gone; release its stdin pipe
    refresh_pgid_snapshot # a retired (dead) shell drops out of the teardown set
    prune_retired
    entry
  end
end

#running?(entry) ⇒ Boolean

THE single liveness oracle for a background shell — "is the work this entry represents still running?". Every surface (status, the running set, the kill/input guards, the UI cards) routes through here so they can never disagree about whether a shell is alive.

A shell is alive while EITHER:

- the `bash -c` LEADER is still alive (the normal case — a server run
without a trailing `&` keeps bash in the foreground), OR
- the output READER thread is still draining — i.e. SOME descendant
still holds the merged stdout/stderr pipe open. This is what catches
a server that backgrounds ITSELF (`npm run dev &`, `cmd & echo up`)
or a launcher that exits while its child keeps serving: the leader
reaps but the child holds the pipe, so the work is plainly still
running. Keying liveness off the leader ALONE (the old behaviour)
falsely reported these :completed the instant the launcher exited,
so shell_output retired+closed them mid-flight — orphaning a live
server and driving the model into a kill/restart loop (port already
in use → crash → restart → …).

The reader-thread signal is immune to PID/PGID reuse: it tracks OUR pipe fd, not a pid, so it can never alias an unrelated later process group the way a bare kill(0, -pgid) probe could after the group is fully reaped.

Returns:

  • (Boolean) 


327
328
329
330
331
 
# File 'lib/rubino/tools/shell_registry.rb', line 327

def running?(entry)
  return false unless entry

  entry.wait_thr&.alive? || entry.reader_thr&.alive? || false
end

#running_entries ⇒ Object

The RUNNING background shells (not yet exited, not retired) — the set the picker/cards surface as live "background work" alongside subagents.



349
350
351
 
# File 'lib/rubino/tools/shell_registry.rb', line 349

def running_entries
  @mutex.synchronize { @entries.values.select { |e| e.retired_at.nil? && running?(e) } }
end

#spawn(command:, cwd:, pty: false) ⇒ Object

Spawns command detached in its own process group so a single kill takes out the whole subtree. Returns the new entry.



114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
 
# File 'lib/rubino/tools/shell_registry.rb', line 114

def spawn(command:, cwd:, pty: false)
  # Capture the parent's notification sink on the CALLING thread (the turn
  # thread). The reader thread below can't read Rubino.background_sink —
  # thread-locals don't propagate — so a finished bg shell would notify
  # nothing (US-5 lost-completion). Stash it like a subagent does. Same for
  # the owning subagent id (thread-local), so stopping that subagent can
  # cascade-kill this shell.
  sink  = Rubino.background_sink
  owner = Rubino.current_subagent_id
  # I/O setup differs by mode but the registry bookkeeping below is shared:
  # pipe (default, non-interactive) vs PTY (interactive — real terminal).
  reader_io, stdin_io, pid = pty ? spawn_pty(command, cwd) : spawn_pipe(command, cwd)

  entry = Entry.new(
    id: new_id,
    command: command,
    cwd: cwd,
    pid: pid,
    pgid: pid,
    wait_thr: Process.detach(pid),
    buffer: +"",
    mutex: Mutex.new,
    started_at: Time.now,
    read_offset: 0,
    stdin: stdin_io,
    sink: sink,
    owner_subagent_id: owner,
    notified: false,
    pty: pty
  )
  entry.reader_thr = Thread.new { drain_into(entry, reader_io) }

  @mutex.synchronize do
    @entries[entry.id] = entry
    refresh_pgid_snapshot
  end
  entry
end

#spawn_pipe(command, cwd) ⇒ Object

Pipe-backed spawn (default): a writable stdin pipe lets the agent feed answers to line-oriented prompts (Y/N, apt-style) via shell_input; stdout+stderr merge into one read pipe. Full-screen TTY programs (vim, REPLs that require [ -t 0 ], getpass on /dev/tty) are out of scope for a plain pipe — those want PTY mode (#spawn_pty). Returns [reader, stdin, pid].

pgroup: true → the child leads a new process group (pgid == child pid), so shell_kill SIGTERMs the whole tree. bash -o pipefail mirrors the foreground shell (a mid-pipeline crash surfaces as the exit status, #156). OS write-jail (#290/#544): the argv+env go through the SAME ShellTool.sandboxed_bash_argv the foreground uses, so a backgrounded write is jailed identically (the launcher exec's bash in-place, preserving pgroup/pipes/cwd). Empty prefix when the sandbox is off ⇒ unchanged.



166
167
168
169
170
171
172
173
174
 
# File 'lib/rubino/tools/shell_registry.rb', line 166

def spawn_pipe(command, cwd)
  rd, wr = IO.pipe
  in_rd, in_wr = IO.pipe
  pid = Process.spawn(*ShellTool.sandboxed_bash_argv(command, cwd: cwd),
                      chdir: cwd, pgroup: true, in: in_rd, out: wr, err: wr)
  wr.close
  in_rd.close
  [rd, in_wr, pid]
end

#spawn_pty(command, cwd) ⇒ Object

PTY-backed spawn (interactive): the child runs on a REAL pseudo-terminal, so [ -t 0 ], tty-aware tools, y/N prompts and /dev/tty password reads all work where a pipe can't. PTY.spawn sets up the controlling terminal (setsid), making the child a session leader → pgid == pid, so the same pgroup hard-kill applies. The master is FULL-DUPLEX: the same terminal is both the output reader and the stdin writer. cwd is baked into the script (PTY.spawn takes no chdir option); the sandbox argv/env still come from the shared helper. Returns [master_reader, master_writer, pid]. Mirrors Hermes' ptyprocess path (tools/process_registry.py spawn_local use_pty).



185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
 
# File 'lib/rubino/tools/shell_registry.rb', line 185

def spawn_pty(command, cwd)
  # cwd on its OWN line, NOT a `cd && (#{command})` subshell: a command
  # ending in a `#`-comment would otherwise swallow the closing paren and
  # break. `|| exit 127` still aborts before running in the wrong dir,
  # mirroring the pipe path's `chdir:` failure.
  script = "cd #{Shellwords.escape(cwd)} || exit 127\n#{command}"
  master_r, master_w, pid = PTY.spawn(*ShellTool.sandboxed_bash_argv(script, cwd: cwd))
  # A fresh PTY is 0x0; give it a sane size so `tput cols`, pagers and
  # progress bars don't misbehave (the attach view resizes to the real
  # terminal later). Best-effort — never fail a spawn over winsize.
  begin
    master_w.winsize = [40, 120]
  rescue StandardError
    nil
  end
  [master_r, master_w, pid]
end

#status(entry) ⇒ Object 



333
334
335
336
337
338
339
 
# File 'lib/rubino/tools/shell_registry.rb', line 333

def status(entry)
  return :running if running?(entry)
  return :stopped if entry.stopped

  code = entry.wait_thr.value.exitstatus
  code && ShellTool.success_exit?(code) ? :completed : :failed
end

#terminate(entry, grace: 2) ⇒ Object

SIGTERM→grace→SIGKILL the process group, then retire so the captured output stays retrievable (shares the kill contract with shell_kill). The single per-shell stop seam the UI (/stop, picker) routes through.



377
378
379
380
381
382
383
384
385
386
387
388
389
 
# File 'lib/rubino/tools/shell_registry.rb', line 377

def terminate(entry, grace: 2)
  entry.stopped = true # a UI /stop ⇒ #status reports :stopped, not :failed
  return retire(entry.id) unless running?(entry)

  signal_group("TERM", entry.pgid)
  grace.times do
    break unless running?(entry)

    sleep 1
  end
  signal_group("KILL", entry.pgid) if running?(entry)
  retire(entry.id)
end

#terminate_owned_by(subagent_id) ⇒ Object

Cascade-stop: terminate every RUNNING shell a subagent opened — its child background work, killed when the parent subagent is stopped. Mirrors Hermes' process_registry kill_all(task_id). Returns the count terminated.



362
363
364
365
366
367
368
369
370
371
372
 
# File 'lib/rubino/tools/shell_registry.rb', line 362

def terminate_owned_by(subagent_id)
  return 0 unless subagent_id

  owned = @mutex.synchronize do
    @entries.values.select do |e|
      e.owner_subagent_id == subagent_id && e.retired_at.nil? && running?(e)
    end
  end
  owned.each { |e| terminate(e) }
  owned.size
end

#unregister_pgid(pgid) ⇒ Object

Drop a foreground shell process group once its own thread has reaped it.



105
106
107
108
109
110
 
# File 'lib/rubino/tools/shell_registry.rb', line 105

def unregister_pgid(pgid)
  @mutex.synchronize do
    @fg_pgids.delete(pgid)
    refresh_pgid_snapshot
  end
end

#write_input(entry, text, enter: true) ⇒ Object

Writes text to the background process's stdin (with a trailing newline unless enter: false) — the "press Enter to answer a prompt" path. Returns the number of bytes written, or raises if stdin is gone.

Raises:

  • (IOError) 


259
260
261
262
263
264
265
266
267
 
# File 'lib/rubino/tools/shell_registry.rb', line 259

def write_input(entry, text, enter: true)
  io = entry.stdin
  raise IOError, "stdin already closed" if io.nil? || io.closed?

  payload = enter ? "#{text}\n" : text.to_s
  io.write(payload)
  io.flush
  payload.bytesize
end