Class: LLM::Function::Thread::Task

Inherits:
LLM::Function::Task show all
Defined in:
lib/llm/function/thread/task.rb

Overview

LLM::Function::Thread::Task wraps a function call in a background thread for concurrent tool execution. The thread is created lazily when #wait is called, not when the task is constructed — so you can build a task, pass it around, and decide when to run it.

Interrupting a running task raises Interrupt inside the thread, which stops the tool call mid-flight. A cancel that arrives before the thread exists is held rather than dropped, and spent inside the call: the tool is entered, so a tool whose own rescue LLM::Interrupt cleans up is cleaned up by a cancel that arrived before it started, the way it is on :fork and :ractor. The thread is created with report_on_exception disabled so unhandled exceptions propagate through #wait instead of to stderr.

A tool that implements on_interrupt is told on that thread, once the call has ended, rather than on the thread that cancelled it.

Instance Attribute Summary

Attributes inherited from LLM::Function::Task

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

Instance Method Summary collapse

Methods inherited from LLM::Function::Task

#intercepted?

Constructor Details

#initialize(fn, options = {}) ⇒ Task

Returns a new instance of Task.

Parameters:



27
28
29
 
# File 'lib/llm/function/thread/task.rb', line 27

def initialize(fn, options = {})
  super
end

Instance Method Details

#spawn ⇒ nil

Returns:

  • (nil) 


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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
 
# File 'lib/llm/function/thread/task.rb', line 33

def spawn
  @span = @tracer&.on_tool_start(
    id: @function.id,
    name: @function.name,
    arguments: @function.arguments,
    model: @function.model
  )
  return if intercepted?
  @ready = Queue.new
  @thread = ::Thread.new do
    ##
    # The window opens on this thread, before the call, and it is
    # published on the queue a canceller may already be waiting on.
    #
    # It is what makes a cancel a delivery rather than a race. The
    # window is idle until `running!` below and `#interrupt!` waits on
    # it, so a cancel that arrives before this thread has reached the
    # call is held rather than dropped - and the raise it issues is not
    # issued before the call, which is what puts it at the tool in
    # practice.
    @window = LLM::Function::Window.new(thread: ::Thread.current)
    @ready << @window
    ##
    # **A cancel that arrived before this thread existed is delivered
    # by a watcher of its own.**
    #
    # `#interrupt!` was answered before there was a thread to raise on,
    # so this is the only place left that can hold the cancel - and
    # raising it here is raising it before the call, which is what this
    # task used to do: the tool was never entered, so a tool whose own
    # rescue cleans up was never cleaned up. The window holds it
    # instead, and opens on the call.
    ::Thread.new { @window.interrupt! } if @cancelled
    @window.running!
    @started_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    function.call
  ensure
    @finished_at = Process.clock_gettime(Process::CLOCK_MONOTONIC)
    ##
    # The window closes before the hook runs, so a cancel that arrives
    # once the call has returned finds nothing to interrupt - which is
    # the no-op {LLM::Function::Return#interrupt!} already says one is.
    @window&.finished!
    ##
    # The hook runs on this thread rather than on the one that
    # cancelled, because a tool's state belongs to the thread its call
    # runs on, and the caller's thread is not that thread.
    #
    # It runs where an ask was taken up for this call, and not for a
    # cancel the window never asked about: the window is the frame that
    # decides whether a raise is issued, so it is the frame that knows,
    # and `Window#interrupted?` is true where one was issued.
    #
    # **Issued is not delivered, and this strategy is the one where the
    # two come apart least.** The raise here is a `Thread#raise`, and a
    # thread can be interrupted where it stands, so a cancel that the
    # window asks about lands on the call. `:fiber` and `:async` ask a
    # scheduler instead, which places the raise at the tool's next
    # suspension - so there a tool that never suspends is asked about
    # and never reached, and the hook runs for it all the same.
    #
    # That is the side to be on, and a tool whose own rescue answers the
    # interrupt rather than raising is why: it was interrupted, and its
    # hook is the only thing left that can say so on its own thread.
    # What does not tell the tool is a cancel the window never asked
    # about - one that arrives once the call has returned - which is the
    # no-op {LLM::Function::Return#interrupt!} already says one is.
    #
    # On the job's own thread the hook can only run after the call's
    # frame has ended - you cannot run code on a thread blocked inside
    # a method it owns except by raising into it - so it runs here
    # rather than before the raise, as `:fork` runs it.
    #
    # The hook runs before this thread ends, so it has run before
    # `#wait` can return - which is the other half of telling the tool
    # before the caller. A hook that raises becomes what this thread
    # returns instead, so the error reaches the caller in place of the
    # call's result.
    function.interrupt! if @window&.interrupted?
  end
  @thread.report_on_exception = false
  nil
end

#alive? ⇒ Boolean

Returns:

  • (Boolean) 


119
120
121
 
# File 'lib/llm/function/thread/task.rb', line 119

def alive?
  @thread&.alive? || false
end

#interrupt! ⇒ nil Also known as: cancel!

Asks the tool to stop, and waits for it to be possible.

It can wait, and the waits are bounded by the body's next instruction. One is for the window itself - the body publishes it on @ready before it opens the call - and one is inside the window, which holds the raise until the call is running. A task this method was called on before it had a thread is the exception: the cancel is recorded and the body delivers it, because there is nothing here to raise on yet.

A group and Context#interrupt! call this in a loop, which is why the wait is worth a line: a loop's cost is the longest wait in it, and every one of those is a dispatch wide.

Returns:

  • (nil) 


138
139
140
141
142
143
144
 
# File 'lib/llm/function/thread/task.rb', line 138

def interrupt!
  @cancelled = true
  if @thread&.alive?
    window.interrupt!
  end
  nil
end

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

Returns:



149
150
151
152
153
154
155
156
157
158
159
 
# File 'lib/llm/function/thread/task.rb', line 149

def wait
  return @intercepted if intercepted?
  return @result.unbox if @result
  spawn unless @thread
  begin
    @result = LLM::Object.from(unbox: @thread.value)
  rescue LLM::Interrupt => ex
    @result = LLM::Object.from(unbox: ex)
  end
  @result.unbox
end

#group_class ⇒ Class

Returns:

  • (Class) 


164
165
166
 
# File 'lib/llm/function/thread/task.rb', line 164

def group_class
  LLM::Function::Thread::Group
end