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. An address_resolver supplies a fresh validated set for each connection cycle; candidates are tried before the request is sent while the original hostname still owns Host, SNI, and TLS validation. JSON bodies and individual SSE lines share one configurable byte ceiling; a stream may contain any number of individually safe lines.

Constant Summary collapse

KEEP_ALIVE_SECONDS = 
30
ERROR_PREVIEW_BYTES = 
500

Instance Method Summary collapse

Constructor Details

#initialize(origin:, open_timeout: 15, read_timeout: 300, write_timeout: 60, address_resolver: nil, max_record_bytes: DEFAULT_MAX_RECORD_BYTES) ⇒ Transport

Returns a new instance of Transport.



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
 
# File 'lib/mistri/transport.rb', line 65

def initialize(origin:, open_timeout: 15, read_timeout: 300, write_timeout: 60,
               address_resolver: nil, max_record_bytes: DEFAULT_MAX_RECORD_BYTES)
  unless max_record_bytes.is_a?(Integer) && max_record_bytes.positive?
    raise ConfigurationError, "max_record_bytes: must be a positive integer"
  end

  @origin = origin.to_s.chomp("/")
  @uri = URI(@origin)
  @open_timeout = open_timeout
  @read_timeout = read_timeout
  @write_timeout = write_timeout
  @address_resolver = address_resolver
  @max_record_bytes = max_record_bytes
  @mutex = Mutex.new
  @connection = nil
end

Instance Method Details

#close ⇒ Object 



116
117
118
 
# File 'lib/mistri/transport.rb', line 116

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.



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
 
# File 'lib/mistri/transport.rb', line 84

def post(path, body:, headers: {})
  @mutex.synchronize do
    with_retry do
      parsed = nil
      connection.request(build_request(path, body, headers)) do |response|
        raise_for_status(response)
        parsed = JSON.parse(read_json_body(response))
      end
      parsed
    end
  rescue ResponseTooLargeError, ResponseTooComplexError, ProviderError, JSON::ParserError
    teardown
    raise
  end
end

#post_either(path, body:, headers: {}, replayable: true, &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. A replayable request retries once when a dead idle socket fails before any response starts.



112
113
114
 
# File 'lib/mistri/transport.rb', line 112

def post_either(path, body:, headers: {}, replayable: true, &block)
  @mutex.synchronize { post_either_locked(path, body, headers, replayable, &block) }
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.



102
103
104
105
106
 
# File 'lib/mistri/transport.rb', line 102

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

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