Class: LLM::Function::Fork::Task

Inherits:
Task
  • Object
show all
Defined in:
lib/llm/function/fork/task.rb

Overview

The Task class wraps a fork-backed function call and exchanges control and result messages with the child process.

Instance Attribute Summary

Attributes inherited from Task

#finished_at, #function, #span, #started_at, #tracer

Instance Method Summary collapse

Methods inherited from Task

#intercepted?

Constructor Details

#initialize(fn, options = {}) ⇒ LLM::Function::Fork::Task

Parameters:

Options Hash (options):



13
14
15
16
17
18
 
# File 'lib/llm/function/fork/task.rb', line 13

def initialize(fn, options = {})
  super
  @spawned = false
  @waited = false
  @cancelled = false
end

Instance Method Details

#spawn ⇒ LLM::Function::Fork::Task

Returns:



22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/llm/function/fork/task.rb', line 22

def spawn
  @span = @tracer&.on_tool_start(
    id: @function.id,
    name: @function.name,
    arguments: @function.arguments,
    model: @function.model
  )
  return if intercepted?
  @ch = channels
  ##
  # **A cancel that arrived before this is written now, before the fork.**
  # The child is the reader, and it inherits both ends, so the message is
  # waiting in the channel by the time its watcher looks - which is the
  # same way a cancel that arrives a moment later is delivered, and the
  # same shape `:ractor`'s task answers with. Written earlier, it would
  # have opened a socketpair for a task that may never fork.
  if @cancelled
    @cancelled = false
    @ch.control.write(:interrupt)
  end
  @started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
  @pid = Kernel.fork do
    ##
    # The child inherits the parent's terminal. When
    # the runtime runs under a curses REPL, a forked
    # tool reading or writing the tty would steal the
    # user's input or clobber the parent's display.
    # Point all three standard streams at null so the
    # child keeps off the user's terminal entirely. A
    # tool that genuinely needs the terminal can reopen
    # it via /dev/tty; the tty fd stays available to
    # the child.
    $stdin.reopen(File::NULL)
    $stdout.reopen(File::NULL)
    $stderr.reopen(File::NULL)
    ##
    # The child's half of the note below this block: it reads control and
    # writes result, and holds no other end.
    @ch.control.w.close
    @ch.result.r.close
    Fork::Job.new(@function, @ch).call
  end
  ##
  # Each side keeps the end it uses and closes the one it does not. Both
  # channels are socketpairs, so an end closed here is still open in the
  # child, and what the rest of them cost is a read that can never be
  # woken: a child that dies before it writes leaves this side with an
  # empty socket and - while this side is holding the write end of that
  # same pair - no end of file to say so. Closed, that is an `EOFError`,
  # and {#wait} answers in band rather than waiting for a write that will
  # never come.
  @ch.control.r.close
  @ch.result.w.close
  @spawned = true
  self
end

#alive? ⇒ Boolean

Returns:

  • (Boolean) 


81
82
83
84
85
86
87
88
89
 
# File 'lib/llm/function/fork/task.rb', line 81

def alive?
  return false if @waited || !@pid
  result = ::Process.waitpid(@pid, ::Process::WNOHANG)
  @waited = !result.nil?
  !@waited
rescue Errno::ECHILD
  @waited = true
  false
end

#interrupt! ⇒ nil Also known as: cancel!

Tells the child to stop, and is a no-op once it has answered.

A cancel that arrives before spawn is not lost. There is no channel to write to yet, so the cancel is recorded and written by spawn, before the fork - the child is the reader, and the message waits in the channel until its watcher looks, which is how a cancel that arrives a moment later is delivered too.

A task the intercept blocked never forks, and one that has answered has nothing left to tell: both are a no-op, the way Return#interrupt! says one is.

Returns:

  • (nil) 


104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/llm/function/fork/task.rb', line 104

def interrupt!
  return nil if @waited || @intercepted
  if @ch
    @ch.control.write(:interrupt)
  else
    @cancelled = true
  end
  nil
rescue Errno::ESRCH, IOError
  nil
end

#wait ⇒ LLM::Function::Return Also known as: value

Wait for the child to write its result.

A second wait is answered from what the first one took. The channels are closed below, so there is nothing left to read from on a second call - and a task that has answered has answered for good, which is what Thread#value does one strategy over and what the ractor's task is asserted to do. The interrupt is held the same way, as the exception the first wait raised, so a second wait raises the same one rather than reading a channel that has gone.

A child that ended without writing is answered in band. The ends are closed above, so a channel with no writer left is an EOFError here rather than a wait nothing can wake - and it is translated into an error return the model is told about, the way the runtime answers a tool that raised, rather than raised into the turn. A child that died is a call that failed and the call is what should report it; ending the turn for it would be an exception to the rule that a tool's failure is in band.

Returns:



136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
 
# File 'lib/llm/function/fork/task.rb', line 136

def wait
  return @intercepted if intercepted?
  return @result.unbox if @result
  spawn unless @spawned
  kind, data, @finished_at = @ch.result.recv
  @result = case kind
            when :interrupt then LLM::Object.from(unbox: LLM::Interrupt.new)
            when :result then LLM::Object.from(unbox: Return.new(data[:id], data[:name], data[:value]))
            else raise ArgumentError, "unknown fork message: #{kind.inspect}"
            end
  reap
  @result.unbox
rescue EOFError
  ##
  # Stored in `@result` so a second wait is
  # given the same return as the first call.
  reap
  val = @function.return(error: true, type: EOFError.name, message: "the tool exited unexpectedly")
  @result = LLM::Object.from(unbox: val)
  @result.unbox
ensure
  if @intercepted.nil?
    reap
    [@ch.control, @ch.result].each { _1.close unless _1.closed? } if @ch
  end
end

#group_class ⇒ Class

Returns:

  • (Class) 


166
167
168
 
# File 'lib/llm/function/fork/task.rb', line 166

def group_class
  LLM::Function::Fork::Group
end