| 1 | package providerext |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | "time" |
| 9 | |
| 10 | "reasonix/internal/extension/protocol" |
| 11 | "reasonix/internal/extension/providerconv" |
| 12 | "reasonix/internal/provider" |
| 13 | ) |
| 14 | |
| 15 | // Provider is the host-side handle for one extension-hosted provider ref. |
| 16 | // Streams run on the owning sidecar, which holds the credentials: the open |
| 17 | // params carry only the request, the ref, the model, and the effort — the |
| 18 | // host never sends another provider's keys across the extension boundary. |
| 19 | // The effort from the resolving Selection is baked in at construction, the |
| 20 | // way boot bakes effort into local providers. |
| 21 | type Provider struct { |
| 22 | resolver *Resolver |
| 23 | client ProviderClient |
| 24 | owner string // plugin package id; used to re-resolve live backends |
| 25 | ref string |
| 26 | effort *string |
| 27 | descriptor provider.Descriptor |
| 28 | } |
| 29 | |
| 30 | var _ provider.Provider = (*Provider)(nil) |
| 31 | |
| 32 | // Name returns the provider instance name: the ref's first segment, mirroring |
| 33 | // the broker's hostProvider ("plugin" for extension refs). |
| 34 | func (p *Provider) Name() string { |
| 35 | if p == nil { |
| 36 | return "extension" |
| 37 | } |
| 38 | if i := strings.IndexByte(p.ref, '/'); i > 0 { |
| 39 | return p.ref[:i] |
| 40 | } |
| 41 | return p.ref |
| 42 | } |
| 43 | |
| 44 | // ModelInfo exposes the sidecar adapter's exact model capability metadata. |
| 45 | // Older sidecars only provide the legacy vision bit, which is projected into |
| 46 | // the canonical modality list for compatibility. |
| 47 | func (p *Provider) ModelInfo() provider.ModelInfo { |
| 48 | if p == nil { |
| 49 | return provider.ModelInfo{} |
| 50 | } |
| 51 | info := provider.ModelInfo{ID: p.descriptor.Model, InputModalities: append([]provider.ModelModality(nil), p.descriptor.InputModalities...)} |
| 52 | if info.InputModalities == nil { |
| 53 | if p.descriptor.Vision { |
| 54 | info.InputModalities = []provider.ModelModality{provider.ModalityText, provider.ModalityImage} |
| 55 | } else { |
| 56 | info.InputModalities = []provider.ModelModality{provider.ModalityText} |
| 57 | } |
| 58 | } |
| 59 | return info |
| 60 | } |
| 61 | |
| 62 | // SupportsTools binds the agent's structured-tool path to the extension's |
| 63 | // declared provider capability. A text-only extension receives no schemas and |
| 64 | // keeps the legacy visible-text completion contract. |
| 65 | func (p *Provider) SupportsTools() bool { |
| 66 | return p != nil && p.descriptor.Tools |
| 67 | } |
| 68 | |
| 69 | // RequiresToolCallReasoning reports the descriptor's replay policy, mirroring |
| 70 | // the broker's hostProvider. |
| 71 | func (p *Provider) RequiresToolCallReasoning() bool { |
| 72 | return p != nil && p.descriptor.ToolCallReasoning |
| 73 | } |
| 74 | |
| 75 | // RequiresReasoningRoundTrip reports the descriptor's round-trip policy. |
| 76 | func (p *Provider) RequiresReasoningRoundTrip() bool { |
| 77 | return p != nil && p.descriptor.ReasoningRoundTrip |
| 78 | } |
| 79 | |
| 80 | // WarnOnMissingToolCallReasoning reports the descriptor's warning policy. |
| 81 | func (p *Provider) WarnOnMissingToolCallReasoning() bool { |
| 82 | return p != nil && p.descriptor.WarnOnMissingToolCallReasoning |
| 83 | } |
| 84 | |
| 85 | // MissingToolCallReasoningWarningIdentity supplies the stable, non-credential |
| 86 | // configuration identity used to rate-limit missing-reasoning diagnostics. |
| 87 | func (p *Provider) MissingToolCallReasoningWarningIdentity() string { |
| 88 | if p == nil { |
| 89 | return "" |
| 90 | } |
| 91 | effort := "" |
| 92 | if p.effort != nil { |
| 93 | effort = strings.TrimSpace(*p.effort) |
| 94 | } |
| 95 | return strings.Join([]string{ |
| 96 | "extension-sidecar", strings.TrimSpace(p.client.PluginID()), strings.TrimSpace(p.ref), |
| 97 | strings.TrimSpace(p.descriptor.Model), effort, |
| 98 | }, "\x00") |
| 99 | } |
| 100 | |
| 101 | // Stream opens one sidecar stream. Each call re-resolves the live backend so |
| 102 | // rolling replacement keeps the same provider-visible ref (cache-stable). |
| 103 | func (p *Provider) ReasoningCapability() provider.ReasoningCapability { |
| 104 | return provider.ReasoningOptions(p.descriptor.DefaultEffort, p.descriptor.Efforts...) |
| 105 | } |
| 106 | |
| 107 | func (p *Provider) Stream(ctx context.Context, request provider.Request) (<-chan provider.Chunk, error) { |
| 108 | if err := p.ReasoningCapability().Validate(p.descriptor.Model, request.EffortOverride); err != nil { |
| 109 | return nil, err |
| 110 | } |
| 111 | if p == nil || p.resolver == nil { |
| 112 | return nil, fmt.Errorf("extension provider is unavailable") |
| 113 | } |
| 114 | client := p.client |
| 115 | if live := p.resolver.liveClient(p.owner); live != nil { |
| 116 | client = live |
| 117 | } |
| 118 | if client == nil { |
| 119 | return nil, fmt.Errorf("extension provider is unavailable") |
| 120 | } |
| 121 | return p.resolver.open(ctx, p, client, request) |
| 122 | } |
| 123 | |
| 124 | // open registers the buffered stream, asks the sidecar to start it, and arms |
| 125 | // the cancellation/disconnect watcher. The seq buffering, delivery, and gap |
| 126 | // semantics mirror the broker's Host.open exactly. |
| 127 | func (r *Resolver) open(ctx context.Context, p *Provider, client ProviderClient, request provider.Request) (<-chan provider.Chunk, error) { |
| 128 | if client.Crashed() { |
| 129 | return nil, &provider.StreamInterruptedError{Err: fmt.Errorf("extension sidecar %s crashed", client.PluginID())} |
| 130 | } |
| 131 | id := "es_" + randomID(12) |
| 132 | stream := &extensionStream{ |
| 133 | client: client, |
| 134 | out: make(chan provider.Chunk, 64), |
| 135 | done: make(chan struct{}), |
| 136 | abortDelivery: make(chan struct{}), |
| 137 | deliveryWake: make(chan struct{}, 1), |
| 138 | nextSeq: 1, |
| 139 | pending: make(map[int64]provider.Chunk), |
| 140 | activity: make(chan struct{}, 1), |
| 141 | } |
| 142 | r.mu.Lock() |
| 143 | r.streams[id] = stream |
| 144 | r.mu.Unlock() |
| 145 | go r.deliverStream(stream) |
| 146 | |
| 147 | gen := r.owner.Gate.Published() |
| 148 | streamID := id |
| 149 | streamRef := stream |
| 150 | // Register before opening: a sidecar may emit stream/end or overflow while |
| 151 | // ProviderStreamOpen is still returning, and those paths must unregister. |
| 152 | unregisterDrainCancel := r.owner.Gate.RegisterDrainCancel(gen, func() { |
| 153 | r.mu.Lock() |
| 154 | if r.streams[streamID] == streamRef { |
| 155 | r.abortDeliveryLocked(streamRef) |
| 156 | r.finishLocked(streamID, streamRef, provider.Chunk{Type: provider.ChunkError, Err: &provider.StreamInterruptedError{ |
| 157 | Err: fmt.Errorf("extension stream %s: generation %d drain timed out", streamID, gen), |
| 158 | }}) |
| 159 | } |
| 160 | r.mu.Unlock() |
| 161 | go client.ProviderStreamCancel(streamID) |
| 162 | }) |
| 163 | r.installDrainCancel(id, stream, unregisterDrainCancel) |
| 164 | |
| 165 | effort := p.descriptor.DefaultEffort |
| 166 | if p.effort != nil { |
| 167 | effort = *p.effort |
| 168 | } |
| 169 | if request.EffortOverride != "" { |
| 170 | effort = request.EffortOverride |
| 171 | } |
| 172 | idleTimeout := r.idleTimeout |
| 173 | if idleTimeout <= 0 { |
| 174 | idleTimeout = defaultStreamIdleTimeout |
| 175 | } |
| 176 | openCtx, cancelOpen := context.WithTimeout(ctx, idleTimeout) |
| 177 | opened, err := client.ProviderStreamOpen(openCtx, protocol.StreamOpenParams{ |
| 178 | StreamID: id, |
| 179 | ProviderRef: p.ref, |
| 180 | Model: p.descriptor.Model, |
| 181 | Effort: effort, |
| 182 | Request: providerconv.RequestToProtocol(request), |
| 183 | SeqBase: 1, |
| 184 | }) |
| 185 | cancelOpen() |
| 186 | if err != nil { |
| 187 | r.removeStream(id, stream) |
| 188 | go client.ProviderStreamCancel(id) |
| 189 | return nil, mapStreamOpenError(client, err) |
| 190 | } |
| 191 | if !opened.Accepted { |
| 192 | r.removeStream(id, stream) |
| 193 | go client.ProviderStreamCancel(id) |
| 194 | return nil, fmt.Errorf("extension %s declined provider stream %q", client.PluginID(), p.ref) |
| 195 | } |
| 196 | // Provider request is already submitted to the sidecar — irreversible for |
| 197 | // recovery (never claim rollback of an in-flight provider call). |
| 198 | r.owner.RecordProviderSubmit(gen, id, client.PluginID()) |
| 199 | go r.watchStream(ctx, id, stream) |
| 200 | return stream.out, nil |
| 201 | } |
| 202 | |
| 203 | // installDrainCancel publishes the gate unregister callback under Resolver.mu. |
| 204 | // If expiration already finished the stream, unregister immediately instead of |
| 205 | // attaching cleanup state to a completed stream. |
| 206 | func (r *Resolver) installDrainCancel(id string, stream *extensionStream, unregister func()) { |
| 207 | if unregister == nil { |
| 208 | return |
| 209 | } |
| 210 | r.mu.Lock() |
| 211 | if r.streams[id] == stream { |
| 212 | stream.unregisterDrainCancel = unregister |
| 213 | unregister = nil |
| 214 | } |
| 215 | r.mu.Unlock() |
| 216 | if unregister != nil { |
| 217 | unregister() |
| 218 | } |
| 219 | } |
| 220 | |
| 221 | // mapStreamOpenError lifts the sidecar's provider_interrupted family into |
| 222 | // StreamInterruptedError so the agent's interruption recovery applies; every |
| 223 | // other failure passes through with its frozen protocol reason intact. |
| 224 | func mapStreamOpenError(client ProviderClient, err error) error { |
| 225 | var protocolErr *protocol.ProtocolError |
| 226 | if errors.As(err, &protocolErr) && protocolErr.Reason == protocol.ErrProviderInterrupted { |
| 227 | return &provider.StreamInterruptedError{Err: errors.New(protocolErr.Message)} |
| 228 | } |
| 229 | return fmt.Errorf("extension %s provider stream open: %w", client.PluginID(), err) |
| 230 | } |
| 231 | |
| 232 | // watchStream finishes the stream on caller cancellation or sidecar loss. A |
| 233 | // cancel aborts delivery (the consumer is gone) and notifies the sidecar; a |
| 234 | // disconnect keeps draining buffered chunks before the terminal interruption, |
| 235 | // mirroring the broker's detach semantics. |
| 236 | func (r *Resolver) watchStream(ctx context.Context, id string, stream *extensionStream) { |
| 237 | idleTimeout := r.idleTimeout |
| 238 | if idleTimeout <= 0 { |
| 239 | idleTimeout = defaultStreamIdleTimeout |
| 240 | } |
| 241 | timer := time.NewTimer(idleTimeout) |
| 242 | defer timer.Stop() |
| 243 | for { |
| 244 | select { |
| 245 | case <-stream.done: |
| 246 | return |
| 247 | case <-stream.activity: |
| 248 | if !timer.Stop() { |
| 249 | select { |
| 250 | case <-timer.C: |
| 251 | default: |
| 252 | } |
| 253 | } |
| 254 | timer.Reset(idleTimeout) |
| 255 | continue |
| 256 | case <-stream.client.Disconnected(): |
| 257 | r.mu.Lock() |
| 258 | if r.streams[id] == stream { |
| 259 | r.finishLocked(id, stream, provider.Chunk{Type: provider.ChunkError, Err: &provider.StreamInterruptedError{ |
| 260 | Err: fmt.Errorf("extension sidecar %s disconnected", stream.client.PluginID()), |
| 261 | }}) |
| 262 | } |
| 263 | r.mu.Unlock() |
| 264 | return |
| 265 | case <-ctx.Done(): |
| 266 | case <-timer.C: |
| 267 | r.mu.Lock() |
| 268 | if r.streams[id] == stream { |
| 269 | r.finishLocked(id, stream, provider.Chunk{Type: provider.ChunkError, Err: &provider.StreamInterruptedError{ |
| 270 | Err: fmt.Errorf("extension provider stream stalled: no activity for %s", idleTimeout), |
| 271 | }}) |
| 272 | } |
| 273 | r.mu.Unlock() |
| 274 | go stream.client.ProviderStreamCancel(id) |
| 275 | return |
| 276 | } |
| 277 | break |
| 278 | } |
| 279 | r.mu.Lock() |
| 280 | if r.streams[id] == stream { |
| 281 | r.abortDeliveryLocked(stream) |
| 282 | r.finishLocked(id, stream, provider.Chunk{Type: provider.ChunkError, Err: &provider.StreamInterruptedError{Err: ctx.Err()}}) |
| 283 | } |
| 284 | r.mu.Unlock() |
| 285 | go stream.client.ProviderStreamCancel(id) |
| 286 | } |
| 287 |