Class: Mistri::Transport

Inherits:
Object
  • Object
show all
Defined in:
lib/mistri/transport.rb

Overview

HTTP for one provider origin, held open across the turns of a run: a multi-turn agent pays the TCP and TLS handshake once, not per turn. Not shareable across threads; a mutex serializes accidental concurrent use onto the single socket, and re-entering from a streaming callback raises ThreadError.

Streaming reads abort two ways: cooperatively between fragments, and hard, by closing the socket from the abort signal's callback, so a stalled read stops immediately instead of waiting out the read timeout.

Constant Summary collapse

KEEP_ALIVE_SECONDS = 
30

Instance Method Summary collapse

Constructor Details

#initialize(origin:, open_timeout: 15, read_timeout: 300, write_timeout: 60) ⇒ Transport

Returns a new instance of Transport.



20
21
22
23
24
25
26
27
28
 
# File 'lib/mistri/transport.rb', line 20

def initialize(origin:, open_timeout: 15, read_timeout: 300, write_timeout: 60)
  @origin = origin.to_s.chomp("/")
  @uri = URI(@origin)
  @open_timeout = open_timeout
  @read_timeout = read_timeout
  @write_timeout = write_timeout
  @mutex = Mutex.new
  @connection = nil
end

Instance Method Details

#close ⇒ Object 



78
79
80
 
# File 'lib/mistri/transport.rb', line 78

def close
  @mutex.synchronize { teardown }
end

#post(path, body:, headers: {}) ⇒ Object

POST and decode a JSON response body. Retries once on a dead idle socket, so it suits idempotent endpoints.



32
33
34
35
36
37
38
 
# File 'lib/mistri/transport.rb', line 32

def post(path, body:, headers: {})
  response = @mutex.synchronize do
    with_retry { connection.request(build_request(path, body, headers)) }
  end
  raise_for_status(response)
  JSON.parse(response.body)
end

#post_either(path, body:, headers: {}, &block) ⇒ Object

POST for Streamable-HTTP endpoints (the MCP shape) that answer either a JSON body or an SSE stream: yields each JSON record either way and returns the response headers, downcased. Retries only a dead idle socket that failed before any response started, so a side-effecting call can never run twice.



53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/mistri/transport.rb', line 53

def post_either(path, body:, headers: {}, &block)
  @mutex.synchronize do
    retried = false
    begin
      started = false
      response_headers = nil
      connection.request(build_request(path, body, headers, streaming: true)) do |response|
        started = true
        raise_for_status(response)
        response_headers = response.to_hash.transform_values(&:first)
        read_either(response, &block)
      end
      response_headers
    rescue IOError, SocketError, SystemCallError, Timeout::Error => e
      teardown
      if started || retried || e.is_a?(Timeout::Error)
        raise ProviderError, "connection failed: #{e.message}"
      end

      retried = true
      retry
    end
  end
end

#stream_post(path, body:, headers: {}, signal: nil, &block) ⇒ Object

POST and stream the SSE response, yielding each decoded data record. Returns :aborted when the signal cancelled the stream, else nil.



42
43
44
45
46
 
# File 'lib/mistri/transport.rb', line 42

def stream_post(path, body:, headers: {}, signal: nil, &block)
  return :aborted if signal&.aborted?

  @mutex.synchronize { stream_locked(path, body, headers, signal, &block) }
end