| 1 | package providerext |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "errors" |
| 6 | "fmt" |
| 7 | "strings" |
| 8 | |
| 9 | "reasonix/internal/extension/protocol" |
| 10 | "reasonix/internal/extension/providerconv" |
| 11 | "reasonix/internal/provider" |
| 12 | ) |
| 13 | |
| 14 | // Provider is the host-side handle for one extension-hosted provider ref. |
| 15 | // Streams run on the owning sidecar, which holds the credentials: the open |
| 16 | // params carry only the request, the ref, the model, and the effort — the |
| 17 | // host never sends another provider's keys across the extension boundary. |
| 18 | // The effort from the resolving Selection is baked in at construction, the |
| 19 | // way boot bakes effort into local providers. |
| 20 | type Provider struct { |
| 21 | resolver *Resolver |
| 22 | client ProviderClient |
| 23 | ref string |
| 24 | effort *string |
| 25 | descriptor provider.Descriptor |
| 26 | } |
| 27 | |
| 28 | var _ provider.Provider = (*Provider)(nil) |
| 29 | |
| 30 | // Name returns the provider instance name: the ref's first segment, mirroring |
| 31 | // the broker's hostProvider ("plugin" for extension refs). |
| 32 | func (p *Provider) Name() string { |
| 33 | if p == nil { |
| 34 | return "extension" |
| 35 | } |
| 36 | if i := strings.IndexByte(p.ref, '/'); i > 0 { |
| 37 | return p.ref[:i] |
| 38 | } |
| 39 | return p.ref |
| 40 | } |
| 41 | |
| 42 | // RequiresToolCallReasoning reports the descriptor's replay policy, mirroring |
| 43 | // the broker's hostProvider. |
| 44 | func (p *Provider) RequiresToolCallReasoning() bool { |
| 45 | return p != nil && p.descriptor.ToolCallReasoning |
| 46 | } |
| 47 | |
| 48 | // RequiresReasoningRoundTrip reports the descriptor's round-trip policy. |
| 49 | func (p *Provider) RequiresReasoningRoundTrip() bool { |
| 50 | return p != nil && p.descriptor.ReasoningRoundTrip |
| 51 | } |
| 52 | |
| 53 | // WarnOnMissingToolCallReasoning reports the descriptor's warning policy. |
| 54 | func (p *Provider) WarnOnMissingToolCallReasoning() bool { |
| 55 | return p != nil && p.descriptor.WarnOnMissingToolCallReasoning |
| 56 | } |
| 57 | |
| 58 | // MissingToolCallReasoningWarningIdentity supplies the stable, non-credential |
| 59 | // configuration identity used to rate-limit missing-reasoning diagnostics. |
| 60 | func (p *Provider) MissingToolCallReasoningWarningIdentity() string { |
| 61 | if p == nil { |
| 62 | return "" |
| 63 | } |
| 64 | effort := "" |
| 65 | if p.effort != nil { |
| 66 | effort = strings.TrimSpace(*p.effort) |
| 67 | } |
| 68 | return strings.Join([]string{ |
| 69 | "extension-sidecar", strings.TrimSpace(p.client.PluginID()), strings.TrimSpace(p.ref), |
| 70 | strings.TrimSpace(p.descriptor.Model), effort, |
| 71 | }, "\x00") |
| 72 | } |
| 73 | |
| 74 | // Stream opens one sidecar stream and returns its buffered chunk channel. |
| 75 | // The channel closes on a clean end; failures arrive as a terminal ChunkError |
| 76 | // (StreamInterruptedError for interruptions). Cancelling ctx aborts the |
| 77 | // stream through extension/provider/stream/cancel. A crashed sidecar fails |
| 78 | // fast — there is no fallback to another provider. |
| 79 | func (p *Provider) Stream(ctx context.Context, request provider.Request) (<-chan provider.Chunk, error) { |
| 80 | if p == nil || p.resolver == nil || p.client == nil { |
| 81 | return nil, fmt.Errorf("extension provider is unavailable") |
| 82 | } |
| 83 | return p.resolver.open(ctx, p, request) |
| 84 | } |
| 85 | |
| 86 | // open registers the buffered stream, asks the sidecar to start it, and arms |
| 87 | // the cancellation/disconnect watcher. The seq buffering, delivery, and gap |
| 88 | // semantics mirror the broker's Host.open exactly. |
| 89 | func (r *Resolver) open(ctx context.Context, p *Provider, request provider.Request) (<-chan provider.Chunk, error) { |
| 90 | client := p.client |
| 91 | if client.Crashed() { |
| 92 | return nil, &provider.StreamInterruptedError{Err: fmt.Errorf("extension sidecar %s crashed", client.PluginID())} |
| 93 | } |
| 94 | id := "es_" + randomID(12) |
| 95 | stream := &extensionStream{ |
| 96 | client: client, |
| 97 | out: make(chan provider.Chunk, 64), |
| 98 | done: make(chan struct{}), |
| 99 | abortDelivery: make(chan struct{}), |
| 100 | deliveryWake: make(chan struct{}, 1), |
| 101 | nextSeq: 1, |
| 102 | pending: make(map[int64]provider.Chunk), |
| 103 | } |
| 104 | r.mu.Lock() |
| 105 | r.streams[id] = stream |
| 106 | r.mu.Unlock() |
| 107 | go r.deliverStream(stream) |
| 108 | |
| 109 | effort := "" |
| 110 | if p.effort != nil { |
| 111 | effort = *p.effort |
| 112 | } |
| 113 | opened, err := client.ProviderStreamOpen(ctx, protocol.StreamOpenParams{ |
| 114 | StreamID: id, |
| 115 | ProviderRef: p.ref, |
| 116 | Model: p.descriptor.Model, |
| 117 | Effort: effort, |
| 118 | Request: providerconv.RequestToProtocol(request), |
| 119 | SeqBase: 1, |
| 120 | }) |
| 121 | if err != nil { |
| 122 | r.removeStream(id, stream) |
| 123 | go client.ProviderStreamCancel(id) |
| 124 | return nil, mapStreamOpenError(client, err) |
| 125 | } |
| 126 | if !opened.Accepted { |
| 127 | r.removeStream(id, stream) |
| 128 | go client.ProviderStreamCancel(id) |
| 129 | return nil, fmt.Errorf("extension %s declined provider stream %q", client.PluginID(), p.ref) |
| 130 | } |
| 131 | go r.watchStream(ctx, id, stream) |
| 132 | return stream.out, nil |
| 133 | } |
| 134 | |
| 135 | // mapStreamOpenError lifts the sidecar's provider_interrupted family into |
| 136 | // StreamInterruptedError so the agent's interruption recovery applies; every |
| 137 | // other failure passes through with its frozen protocol reason intact. |
| 138 | func mapStreamOpenError(client ProviderClient, err error) error { |
| 139 | var protocolErr *protocol.ProtocolError |
| 140 | if errors.As(err, &protocolErr) && protocolErr.Reason == protocol.ErrProviderInterrupted { |
| 141 | return &provider.StreamInterruptedError{Err: errors.New(protocolErr.Message)} |
| 142 | } |
| 143 | return fmt.Errorf("extension %s provider stream open: %w", client.PluginID(), err) |
| 144 | } |
| 145 | |
| 146 | // watchStream finishes the stream on caller cancellation or sidecar loss. A |
| 147 | // cancel aborts delivery (the consumer is gone) and notifies the sidecar; a |
| 148 | // disconnect keeps draining buffered chunks before the terminal interruption, |
| 149 | // mirroring the broker's detach semantics. |
| 150 | func (r *Resolver) watchStream(ctx context.Context, id string, stream *extensionStream) { |
| 151 | select { |
| 152 | case <-stream.done: |
| 153 | return |
| 154 | case <-stream.client.Disconnected(): |
| 155 | r.mu.Lock() |
| 156 | if r.streams[id] == stream { |
| 157 | r.finishLocked(id, stream, provider.Chunk{Type: provider.ChunkError, Err: &provider.StreamInterruptedError{ |
| 158 | Err: fmt.Errorf("extension sidecar %s disconnected", stream.client.PluginID()), |
| 159 | }}) |
| 160 | } |
| 161 | r.mu.Unlock() |
| 162 | return |
| 163 | case <-ctx.Done(): |
| 164 | } |
| 165 | r.mu.Lock() |
| 166 | if r.streams[id] == stream { |
| 167 | r.abortDeliveryLocked(stream) |
| 168 | r.finishLocked(id, stream, provider.Chunk{Type: provider.ChunkError, Err: &provider.StreamInterruptedError{Err: ctx.Err()}}) |
| 169 | } |
| 170 | r.mu.Unlock() |
| 171 | go stream.client.ProviderStreamCancel(id) |
| 172 | } |
| 173 |