Module: Pikuri::Testing

Defined in:
lib/pikuri/testing.rb

Overview

Test support: run the real Agent (real run_loop, real listeners, real tools/extensions, real Event stream, real ruby_llm loop) against a scripted HTTP backend — no server, no network, no mock agent.

Not loaded by default: require 'pikuri/testing' from a spec. It lives off the production load path (the core Zeitwerk loader ignores it).

How it works

Testing.fake_transport returns an ordinary Agent::ChatTransport whose faraday_adapter is an anonymous Faraday::Adapter serving the responses you scripted. Because Agent#build_context installs that adapter on the agent's per-chat RubyLLM::Context (never the global RubyLLM.config), ruby_llm's own loop runs for real — it parses tool calls, invokes your real registered tools, appends observations, and re-requests — and each scripted entry is the model's next turn. No hand-rolled agent loop, nothing to drift against a ruby_llm upgrade.

require 'pikuri/testing'

transport = Pikuri::Testing.fake_transport do |llm|
llm.tool_call('calculator', expression: '2 + 2') # turn 1: call a real tool
llm.reply('The answer is 4.')                     # turn 2: final answer
end

agent = Pikuri::Agent.new(transport:, system_prompt: 'test') do |c|
c.add_listener capture
c.add_tool Pikuri::Tool::CALCULATOR
end
agent.run_loop(user_message: 'What is 2 + 2?')
# the real loop ran the real calculator; capture saw ToolCall/ToolResult/Assistant

Limitations

A fake transport feeds one ordered response list. Propagation to sub-agents / the synthesizer works (they derive their transport via parent.transport.with(...), and Data#with preserves the adapter), but they share that one list — fine for a single linear script, not for targeting a specific sub-agent's replies. Per-sub-agent scripting is a future upgrade (dispatch by model id / distinct fake api_base). Tool-call streaming (fragmented tool_calls deltas) is not modelled; Script#stream streams text only.

Defined Under Namespace

Classes: Gate, Script

Class Method Summary collapse

Class Method Details

.build_adapter(responses, gate = nil) ⇒ Class

An anonymous Faraday::Adapter subclass that serves responses in order, one per HTTP request. Closes over the (mutable) list, so the queue drains across the several requests of one conversation regardless of how Faraday reuses adapter instances.

Parameters:

  • responses (Array<Hash>) —

    entries shaped {json: Hash} (non-stream) or {sse: Array<String>} (stream), as Script produces

  • gate (Gate, nil) (defaults to: nil) —

    awaited before each response when non-nil

Returns:

  • (Class) —

    a Faraday::Adapter subclass



100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
 
# File 'lib/pikuri/testing.rb', line 100

def self.build_adapter(responses, gate = nil)
  queue = responses # captured by reference; #call shifts from it
  Class.new(Faraday::Adapter) do
    # +define_method+ (not +def+) so the body closes over +queue+/+gate+;
    # +super+ must be explicit here (implicit super is illegal in a
    # define_method).
    define_method(:call) do |env|
      super(env) # base builds env.response; skipping it → "finish for nil"
      gate&.await # park here while the test holds the turn open
      entry = queue.shift or raise "Pikuri::Testing: no scripted response left for #{env.url}"

      if env.request.stream_response? # ruby_llm set req.options.on_data
        # v2 on_data routes to the chunk handler only at status 200; set it
        # before feeding, mirroring how a real adapter reads the status line
        # ahead of the body.
        env.status = 200
        on_data = env.request.on_data
        size = 0
        entry.fetch(:sse).each do |chunk|
          size += chunk.bytesize
          on_data.call(chunk, size, env)
        end
        save_response(env, 200, '', { 'Content-Type' => 'text/event-stream' }, nil)
      else
        save_response(env, 200, JSON.generate(entry.fetch(:json)),
                      { 'Content-Type' => 'application/json' }, nil)
      end
      @app.call(env)
    end
  end
end

.fake_transport(model: 'fake-model', context_window: 8192, api_base: 'http://fake.invalid/v1', gate: nil) {|script| ... } ⇒ Agent::ChatTransport

Build a fake transport from a scripted conversation. Pass a block that calls Script verbs in the order the model should "reply"; each verb queues one HTTP response.

Examples:

a two-turn tool loop

Pikuri::Testing.fake_transport do |llm|
  llm.tool_call('calculator', expression: '2 + 2')
  llm.reply('The answer is 4.')
end

Parameters:

  • model (String) (defaults to: 'fake-model') —

    model id the fake reports (any string; the fake server "knows" it)

  • context_window (Integer) (defaults to: 8192) —

    baked onto the transport so Agent#detect_and_emit_context_cap! skips the real /props HTTP probe

  • api_base (String) (defaults to: 'http://fake.invalid/v1') —

    the transport's base URL. Never dialed (the adapter answers), but a host may key display/config off it — pass the base of the connection this fake impersonates so a UI resolves its name.

  • gate (Gate, nil) (defaults to: nil) —

    a latch to hold turns open mid-flight. When engaged (Pikuri::Testing::Gate#block!), the adapter parks before serving each response until Pikuri::Testing::Gate#release! — so a test can observe spinner / queued-injection / Ctrl+C paths through the real turn lifecycle. Default nil (no gating).

Yield Parameters:

  • script (Script) —

    the conversation builder

Returns:



74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
 
# File 'lib/pikuri/testing.rb', line 74

def self.fake_transport(model: 'fake-model', context_window: 8192,
                        api_base: 'http://fake.invalid/v1', gate: nil)
  script = Script.new
  yield script if block_given?
  Agent::ChatTransport.new(
    model: model,
    provider: :openai,
    assume_model_exists: true,
    # An api_base forces the connection-override path (Context#chat), the
    # only branch that honours faraday_adapter; the host is never dialed.
    api_base: api_base,
    api_key: 'not-needed',
    context_window: context_window,
    faraday_adapter: build_adapter(script.responses, gate)
  )
end