Struct CodexProxy 

pub struct CodexProxy {
    pub router: Arc<BackendRouter>,
    pub providers: HashMap<String, Arc<dyn Provider>>,
    pub log_body: bool,
    pub log_dir: PathBuf,
    pub conversation_store: Arc<ConversationStore>,
}

Codex proxy handler implementing ProxyHttp trait.

Fields

router: Arc<BackendRouter>

Backend router for multi-backend routing.

providers: HashMap<String, Arc<dyn Provider>>

Providers for each backend (shared, stateless).

log_body: bool

Whether to log request/response bodies.

log_dir: PathBuf

Directory for debug log files.

conversation_store: Arc<ConversationStore>

In-memory conversation store for previous_response_id expansion.

Implementations

impl CodexProxy

pub fn new( router: Arc<BackendRouter>, providers: HashMap<String, Arc<dyn Provider>>, log_body: bool, log_dir: PathBuf, ) -> Self

Create a new CodexProxy instance.

pub fn get_provider(&self, name: &str) -> Option<Arc<dyn Provider>>

Get a shared handle to the provider for a backend.

pub fn get_conversation( &self, response_id: &str, ) -> Option<ConversationSnapshot>

Lookup conversation snapshot by response id.

pub fn store_conversation( &self, response_id: String, snapshot: ConversationSnapshot, )

Persist conversation snapshot for follow-up turns.

Trait Implementations

impl ProxyHttp for CodexProxy

fn request_filter<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, session: &'life1 mut Session, ctx: &'life2 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = PingoraResult<bool>> + Send + 'async_trait>>

where Self::CTX: Send + Sync, Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Request filter - called for each request to select backend and prepare context.

fn upstream_peer<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _session: &'life1 mut Session, ctx: &'life2 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = PingoraResult<Box<HttpPeer>>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait,

Select upstream peer for proxying.

fn upstream_request_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, session: &'life1 mut Session, upstream_request: &'life2 mut RequestHeader, ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = PingoraResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Filter and modify the upstream request.

fn request_body_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, body: &'life2 mut Option<Bytes>, end_of_stream: bool, ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = PingoraResult<()>> + Send + 'async_trait>>

where Self::CTX: Send + Sync, Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Capture and transform request body (Responses API → Chat API).

For conversion requests, we need to buffer ALL chunks and send the converted body at end_of_stream. This is because HTTP/2 DATA frames arrive incrementally and we can’t convert partial bodies.

fn response_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, upstream_response: &'life2 mut ResponseHeader, ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = PingoraResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Handle upstream response headers.

fn response_body_filter( &self, _session: &mut Session, body: &mut Option<Bytes>, end_of_body: bool, ctx: &mut Self::CTX, ) -> PingoraResult<Option<Duration>>

where Self::CTX: Send + Sync,

Transform response body (Chat API → Responses API for streaming).

fn connected_to_upstream<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, _session: &'life1 mut Session, reused: bool, peer: &'life2 HttpPeer, _fd: RawFd, digest: Option<&'life3 Digest>, ctx: &'life4 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = PingoraResult<()>> + Send + 'async_trait>>

where Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait,

Called when connected to upstream.

fn error_while_proxy( &self, peer: &HttpPeer, _session: &mut Session, e: Box<Error>, ctx: &mut Self::CTX, _client_reused: bool, ) -> Box<Error>

Handle proxy errors.

fn fail_to_proxy<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, session: &'life1 mut Session, e: &'life2 Error, ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = FailToProxy> + Send + 'async_trait>>

where Self::CTX: Send + Sync, Self: 'async_trait, 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait,

Handle fatal errors when proxy cannot be established.

type CTX = ProxyContext

The per request object to share state across the different filters

fn new_ctx(&self) -> Self::CTX

Define how the ctx should be created.

fn init_downstream_modules(&self, modules: &mut HttpModules)

Set up downstream modules.

fn early_request_filter<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _ctx: &'life2 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<(), Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

Handle the incoming request before any downstream module is executed.

fn allow_spawning_subrequest( &self, _session: &Session, _ctx: &Self::CTX, ) -> bool

where Self::CTX: Send + Sync,

Returns whether this session is allowed to spawn subrequests.

fn request_cache_filter( &self, _session: &mut Session, _ctx: &mut Self::CTX, ) -> Result<(), Box<Error>>

where Self::CTX: Send + Sync,

This filter decides if the request is cacheable and what cache backend to use

fn cache_key_callback( &self, _session: &Session, _ctx: &mut Self::CTX, ) -> Result<CacheKey, Box<Error>>

This callback generates the cache key.

fn cache_miss(&self, session: &mut Session, _ctx: &mut Self::CTX)

This callback is invoked when a cacheable response is ready to be admitted to cache.

fn cache_hit_filter<'life0, 'life1, 'life2, 'life3, 'life4, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _meta: &'life2 CacheMeta, _hit_handler: &'life3 mut Box<dyn HandleHit + Send + Sync>, _is_fresh: bool, _ctx: &'life4 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<Option<ForcedFreshness>, Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, 'life4: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

This filter is called after a successful cache lookup and before the cache asset is ready to be used.

fn proxy_upstream_filter<'life0, 'life1, 'life2, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _ctx: &'life2 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<bool, Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

Decide if a request should continue to upstream after not being served from cache.

fn response_cache_filter( &self, _session: &Session, _resp: &ResponseHeader, _ctx: &mut Self::CTX, ) -> Result<RespCacheable, Box<Error>>

Decide if the response is cacheable

fn cache_vary_filter( &self, _meta: &CacheMeta, _ctx: &mut Self::CTX, _req: &RequestHeader, ) -> Option<[u8; 16]>

Decide how to generate cache vary key from both request and response

fn cache_not_modified_filter( &self, session: &Session, resp: &ResponseHeader, _ctx: &mut Self::CTX, ) -> Result<bool, Box<Error>>

Decide if the incoming request’s condition fails against the cached response.

fn range_header_filter( &self, session: &mut Session, resp: &mut ResponseHeader, _ctx: &mut Self::CTX, ) -> RangeType

This filter is called when cache is enabled to determine what byte range to return (in both cache hit and miss cases) from the response body. It is only used when caching is enabled, otherwise the upstream is responsible for any filtering. It allows users to define the range this request is for via its return type range_filter::RangeType.

fn upstream_response_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _upstream_response: &'life2 mut ResponseHeader, _ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<(), Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

Modify the response header from the upstream

fn upstream_response_body_filter( &self, _session: &mut Session, _body: &mut Option<Bytes>, _end_of_stream: bool, _ctx: &mut Self::CTX, ) -> Result<Option<Duration>, Box<Error>>

Similar to Self::upstream_response_filter() but for response body

fn upstream_response_trailer_filter( &self, _session: &mut Session, _upstream_trailers: &mut HeaderMap, _ctx: &mut Self::CTX, ) -> Result<(), Box<Error>>

Similar to Self::upstream_response_filter() but for response trailers

fn response_trailer_filter<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _upstream_trailers: &'life2 mut HeaderMap, _ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = Result<Option<Bytes>, Box<Error>>> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

Similar to Self::response_filter() but for response trailers. Note, returning an Ok(Some(Bytes)) will result in the downstream response trailers being written to the response body.

fn logging<'life0, 'life1, 'life2, 'life3, 'async_trait>( &'life0 self, _session: &'life1 mut Session, _e: Option<&'life2 Error>, _ctx: &'life3 mut Self::CTX, ) -> Pin<Box<dyn Future<Output = ()> + Send + 'async_trait>>

where 'life0: 'async_trait, 'life1: 'async_trait, 'life2: 'async_trait, 'life3: 'async_trait, Self::CTX: Send + Sync, Self: Sync + 'async_trait,

This filter is called when the entire response is sent to the downstream successfully or there is a fatal error that terminate the request.

fn suppress_error_log( &self, _session: &Session, _ctx: &Self::CTX, _error: &Error, ) -> bool

A value of true means that the log message will be suppressed. The default value is false.

fn fail_to_connect( &self, _session: &mut Session, _peer: &HttpPeer, _ctx: &mut Self::CTX, e: Box<Error>, ) -> Box<Error>

This filter is called when there is an error in the process of establishing a connection to the upstream.

fn should_serve_stale( &self, _session: &mut Session, _ctx: &mut Self::CTX, error: Option<&Error>, ) -> bool

Decide whether should serve stale when encountering an error or during revalidation

fn request_summary(&self, session: &Session, _ctx: &Self::CTX) -> String

This callback is invoked every time request related error log needs to be generated

fn is_purge(&self, _session: &Session, _ctx: &Self::CTX) -> bool

Whether the request should be used to invalidate(delete) the HTTP cache

fn purge_response_filter( &self, _session: &Session, _ctx: &mut Self::CTX, _purge_status: PurgeStatus, _purge_response: &mut Cow<'static, ResponseHeader>, ) -> Result<(), Box<Error>>

This filter is called after the proxy cache generates the downstream response to the purge request (to invalidate or delete from the HTTP cache), based on the purge status, which indicates whether the request succeeded or failed.