class Riffer::Rig::Runtime::Turn

Constants

MAX_JOB_CONTINUATIONS: Per turn, so a job that starts jobs cannot keep a turn alive forever.

Public Class Methods

new (yielder, host:, abort:, agent:, hooks:, continuation:, deliver:)

(Enumerator::Yielder yielder, host: Riffer::Rig::Hosts::Mirror, abort: Riffer::Rig::Runtime::Abort, agent: ^() → Riffer::Agent, hooks: ^() → Riffer::Rig::Runtime::Hooks, continuation: ^() → Enumerator[::Riffer::StreamEvents::Base, Riffer::Agent::Response]?, deliver: ^(Riffer::Messages::Base) → void) → void

Source

# File lib/riffer/rig/runtime/turn.rb, line 24
def initialize(yielder, host:, abort:, agent:, hooks:, continuation:, deliver:)
  @yielder = yielder
  @host = host
  @abort = abort
  @agent = agent
  @hooks = hooks
  @continuation = continuation
  @deliver = deliver
  @draft = []
end

@rbs yielder: Enumerator::Yielder @rbs host: Riffer::Rig::Hosts::Mirror @rbs abort: Riffer::Rig::Runtime::Abort @rbs agent: ^() -> Riffer::Agent @rbs hooks: ^() -> Riffer::Rig::Runtime::Hooks @rbs continuation: ^() -> Enumerator[::Riffer::StreamEvents::Base, Riffer::Agent::Response]? @rbs deliver: ^(Riffer::Messages::Base) -> void @rbs return: void

Public Instance Methods

clear_draft ()

() → void

Source

# File lib/riffer/rig/runtime/turn.rb, line 67
def clear_draft
  @draft = []
end

@rbs return: void

forward (event)

(::Riffer::StreamEvents::Base | Riffer::Rig::Events::_Event event) → void

Source

# File lib/riffer/rig/runtime/turn.rb, line 59
def forward(event)
  @yielder << event
  # Mid-turn notifies (a failing hook, a blocked request) reach the stream
  # next to the event that raised them, not at the next turn.
  drain
end

@rbs event: ::Riffer::StreamEvents::Base | Riffer::Rig::Events::_Event @rbs return: void

run (session_start, &)

(Riffer::Rig::Events::SessionStart? session_start) { () → Enumerator[::Riffer::StreamEvents::Base, Riffer::Agent::Response] } → Riffer::Agent::Response

Source

# File lib/riffer/rig/runtime/turn.rb, line 38
def run(session_start, &)
  if session_start
    @hooks.call.observe(:session_start, session_start)
    @yielder << session_start
  end
  drain
  response = @abort.window do
    stream = Thread.handle_interrupt(Riffer::Rig::Runtime::Abort::SHIELD, &)
    continued(relay(stream), MAX_JOB_CONTINUATIONS)
  end
  response ||= aborted
  stop_reason = Riffer::Rig::Runtime::StopReason.of(response.outcome)
  turn_end = Riffer::Rig::Events::TurnEnd.new(stop_reason, response.token_usage)
  @hooks.call.observe(:turn_end, turn_end)
  drain
  @yielder << turn_end
  response
end

@rbs session_start: Riffer::Rig::Events::SessionStart? @rbs &: () -> Enumerator[::Riffer::StreamEvents::Base, Riffer::Agent::Response] @rbs return: Riffer::Agent::Response