| 1 | // Package providerext adapts extension-hosted sidecar providers into the |
| 2 | // host's provider.Resolver surface (Extension Protocol v1, stage 7). Each |
| 3 | // started sidecar holds its own provider credentials and runs streams; the |
| 4 | // host only ever sees the credential-free wire DTOs. The Resolver merges the |
| 5 | // base resolver's catalog with every sidecar's declared catalog and routes |
| 6 | // plugin-namespaced refs (plugin/<plugin>/<provider>/<model>) to the owning |
| 7 | // sidecar's Provider. It mirrors the Remote broker's host-side semantics |
| 8 | // (internal/remote/broker) exactly: 1-based contiguous seq buffering, a |
| 9 | // bounded delivery queue, a gap timer on stream end, cancellation through |
| 10 | // stream/cancel, and interruption via StreamInterruptedError. A selected |
| 11 | // sidecar's crash fails its streams — the adapter never falls back to a |
| 12 | // different provider for plugin refs. |
| 13 | package providerext |
| 14 | |
| 15 | import ( |
| 16 | "context" |
| 17 | "fmt" |
| 18 | "log/slog" |
| 19 | "strings" |
| 20 | "sync" |
| 21 | "time" |
| 22 | |
| 23 | "reasonix/internal/extension" |
| 24 | "reasonix/internal/extension/protocol" |
| 25 | "reasonix/internal/extension/providerconv" |
| 26 | "reasonix/internal/extension/sidecar" |
| 27 | "reasonix/internal/provider" |
| 28 | ) |
| 29 | |
| 30 | // ProviderClient is the slice of a live sidecar connection the adapter needs. |
| 31 | // *sidecar.Client satisfies it; tests substitute fakes. |
| 32 | type ProviderClient interface { |
| 33 | // PluginID returns the installed plugin package name this client serves. |
| 34 | PluginID() string |
| 35 | // Crashed reports whether the connection ended unexpectedly. |
| 36 | Crashed() bool |
| 37 | // Disconnected returns a channel closed when the connection's serve loop |
| 38 | // ends for any reason — crash, orderly shutdown, or transport failure. |
| 39 | Disconnected() <-chan struct{} |
| 40 | // Handshake returns the sidecar's validated initialize result; its |
| 41 | // Providers are the declaration of record for routing and conflicts. |
| 42 | Handshake() protocol.InitializeResult |
| 43 | // ProviderCatalog fetches the sidecar's full provider catalog. |
| 44 | ProviderCatalog(ctx context.Context) ([]protocol.ProviderDescriptor, error) |
| 45 | // ProviderStreamOpen opens one stream; chunks arrive as notifications. |
| 46 | ProviderStreamOpen(ctx context.Context, params protocol.StreamOpenParams) (protocol.StreamOpenResult, error) |
| 47 | // ProviderStreamCancel cancels one in-flight stream, best effort. |
| 48 | ProviderStreamCancel(streamID string) |
| 49 | } |
| 50 | |
| 51 | // ProviderConflict records one sidecar provider ref that collides with a base |
| 52 | // catalog entry without the plugin owning the provider:<ref> replacement slot. |
| 53 | type ProviderConflict struct { |
| 54 | // Ref is the colliding provider ref (identical in both catalogs). |
| 55 | Ref string |
| 56 | // PluginID is the extension declaring Ref. |
| 57 | PluginID string |
| 58 | // Slot is the replacement slot the plugin must claim to override legally. |
| 59 | Slot extension.Slot |
| 60 | } |
| 61 | |
| 62 | // ConflictError fails a build whose sidecar providers collide with the base |
| 63 | // catalog without a manifest replacement claim. Boot treats it as fatal, the |
| 64 | // same class as sidecar.RequiredStartError. |
| 65 | type ConflictError struct { |
| 66 | Conflicts []ProviderConflict |
| 67 | } |
| 68 | |
| 69 | func (e *ConflictError) Error() string { |
| 70 | lines := make([]string, 0, len(e.Conflicts)) |
| 71 | for _, c := range e.Conflicts { |
| 72 | lines = append(lines, fmt.Sprintf( |
| 73 | "plugin %q declares provider ref %q that the host provider catalog already serves; declare %q in the plugin manifest's runtime.replaces to override it", |
| 74 | c.PluginID, c.Ref, string(c.Slot))) |
| 75 | } |
| 76 | return "extension provider conflict: " + strings.Join(lines, "; ") |
| 77 | } |
| 78 | |
| 79 | // Resolver merges the base provider.Resolver with the extension sidecars' |
| 80 | // provider catalogs. It also implements sidecar.StreamRouter: the sidecar |
| 81 | // clients deliver inbound stream/chunk and stream/end notifications here, and |
| 82 | // the Resolver routes them by stream ID to the owning buffered stream. |
| 83 | type Resolver struct { |
| 84 | base provider.Resolver |
| 85 | clients func() []ProviderClient |
| 86 | |
| 87 | // replaced maps a base catalog ref to the plugin ID whose claimed |
| 88 | // provider:<ref> slot lets its descriptor substitute the base entry. |
| 89 | replaced map[string]string |
| 90 | |
| 91 | mu sync.Mutex |
| 92 | streams map[string]*extensionStream |
| 93 | catalogCache map[string][]provider.Descriptor |
| 94 | catalogCalls map[string]*catalogCall |
| 95 | } |
| 96 | |
| 97 | // catalogCall is one in-flight catalog fetch shared by every concurrent |
| 98 | // Catalog caller for the same plugin. The first caller owns the sidecar RPC; |
| 99 | // followers wait for done and then receive defensive copies of the same |
| 100 | // result. This prevents duplicate first-use RPCs and last-completion-wins |
| 101 | // cache contents when a sidecar catalog is dynamic. |
| 102 | type catalogCall struct { |
| 103 | done chan struct{} |
| 104 | descriptors []provider.Descriptor |
| 105 | ok bool |
| 106 | } |
| 107 | |
| 108 | var ( |
| 109 | _ provider.Resolver = (*Resolver)(nil) |
| 110 | _ sidecar.StreamRouter = (*Resolver)(nil) |
| 111 | _ ProviderClient = (*sidecar.Client)(nil) |
| 112 | ) |
| 113 | |
| 114 | // catalogFetchTimeout bounds one sidecar's extension/provider/catalog call, |
| 115 | // mirroring the broker host's catalog budget. |
| 116 | const catalogFetchTimeout = 10 * time.Second |
| 117 | |
| 118 | // New builds the merged resolver. clients supplies the live sidecar |
| 119 | // connections (typically the sidecar Manager's client list); claims is the |
| 120 | // build's frozen replacement-slot ownership table (the kernel snapshot's |
| 121 | // Replacements). A sidecar ref colliding exactly with a base catalog ref is |
| 122 | // legal only when the plugin owns the provider:<ref> slot — then the sidecar |
| 123 | // descriptor replaces the base entry — and is a *ConflictError otherwise. |
| 124 | func New(base provider.Resolver, clients func() []ProviderClient, claims map[extension.Slot]extension.ContributionSource) (*Resolver, error) { |
| 125 | if base == nil { |
| 126 | base = &provider.StaticResolver{} |
| 127 | } |
| 128 | if clients == nil { |
| 129 | clients = func() []ProviderClient { return nil } |
| 130 | } |
| 131 | r := &Resolver{ |
| 132 | base: base, |
| 133 | clients: clients, |
| 134 | replaced: map[string]string{}, |
| 135 | streams: make(map[string]*extensionStream), |
| 136 | catalogCache: make(map[string][]provider.Descriptor), |
| 137 | catalogCalls: make(map[string]*catalogCall), |
| 138 | } |
| 139 | baseRefs := map[string]bool{} |
| 140 | for _, d := range base.Catalog() { |
| 141 | baseRefs[d.Ref] = true |
| 142 | } |
| 143 | var conflicts []ProviderConflict |
| 144 | for _, client := range clients() { |
| 145 | prefix := "plugin/" + client.PluginID() + "/" |
| 146 | for _, decl := range client.Handshake().Providers { |
| 147 | ref := strings.TrimSpace(decl.Ref) |
| 148 | if !strings.HasPrefix(ref, prefix) { |
| 149 | // The handshake validation already enforces the namespace; |
| 150 | // skip defensively rather than failing the build twice. |
| 151 | continue |
| 152 | } |
| 153 | if !baseRefs[ref] { |
| 154 | continue |
| 155 | } |
| 156 | slot := extension.SlotProviderRef(ref) |
| 157 | owner, claimed := claims[slot] |
| 158 | if !claimed || owner.PluginID != client.PluginID() { |
| 159 | conflicts = append(conflicts, ProviderConflict{Ref: ref, PluginID: client.PluginID(), Slot: slot}) |
| 160 | continue |
| 161 | } |
| 162 | r.replaced[ref] = client.PluginID() |
| 163 | } |
| 164 | } |
| 165 | if len(conflicts) > 0 { |
| 166 | return nil, &ConflictError{Conflicts: conflicts} |
| 167 | } |
| 168 | return r, nil |
| 169 | } |
| 170 | |
| 171 | // Catalog returns the base catalog minus claim-replaced entries plus every |
| 172 | // reachable sidecar's catalog. Sidecar catalogs are cached per client for the |
| 173 | // client's lifetime (its generation); a crashed sidecar's cache is dropped |
| 174 | // and its entries stop being offered. Fetch failures skip that sidecar for |
| 175 | // this call, mirroring the broker's best-effort catalog. |
| 176 | func (r *Resolver) Catalog() []provider.Descriptor { |
| 177 | base := r.base.Catalog() |
| 178 | out := make([]provider.Descriptor, 0, len(base)) |
| 179 | for _, d := range base { |
| 180 | if _, replaced := r.replaced[d.Ref]; replaced { |
| 181 | continue |
| 182 | } |
| 183 | out = append(out, d) |
| 184 | } |
| 185 | for _, client := range r.clients() { |
| 186 | descriptors, ok := r.catalogFor(client) |
| 187 | if !ok { |
| 188 | continue |
| 189 | } |
| 190 | out = append(out, descriptors...) |
| 191 | } |
| 192 | return out |
| 193 | } |
| 194 | |
| 195 | // catalogFor returns one sidecar's converted catalog, serving the per-client |
| 196 | // cache when warm. Entries outside the plugin's own namespace are skipped: |
| 197 | // the catalog RPC result must honor the same contract the handshake enforced. |
| 198 | func (r *Resolver) catalogFor(client ProviderClient) ([]provider.Descriptor, bool) { |
| 199 | pluginID := client.PluginID() |
| 200 | if client.Crashed() { |
| 201 | r.mu.Lock() |
| 202 | delete(r.catalogCache, pluginID) |
| 203 | r.mu.Unlock() |
| 204 | return nil, false |
| 205 | } |
| 206 | r.mu.Lock() |
| 207 | if cached, ok := r.catalogCache[pluginID]; ok { |
| 208 | out := append([]provider.Descriptor(nil), cached...) |
| 209 | r.mu.Unlock() |
| 210 | return out, true |
| 211 | } |
| 212 | if call := r.catalogCalls[pluginID]; call != nil { |
| 213 | r.mu.Unlock() |
| 214 | <-call.done |
| 215 | return append([]provider.Descriptor(nil), call.descriptors...), call.ok |
| 216 | } |
| 217 | call := &catalogCall{done: make(chan struct{})} |
| 218 | r.catalogCalls[pluginID] = call |
| 219 | r.mu.Unlock() |
| 220 | |
| 221 | ctx, cancel := context.WithTimeout(context.Background(), catalogFetchTimeout) |
| 222 | defer cancel() |
| 223 | declared, err := client.ProviderCatalog(ctx) |
| 224 | if err != nil { |
| 225 | slog.Debug("providerext: sidecar catalog fetch failed", "plugin", pluginID, "err", err) |
| 226 | r.finishCatalogCall(pluginID, call, nil, false) |
| 227 | return nil, false |
| 228 | } |
| 229 | prefix := "plugin/" + pluginID + "/" |
| 230 | out := make([]provider.Descriptor, 0, len(declared)) |
| 231 | for _, d := range declared { |
| 232 | if !strings.HasPrefix(d.Ref, prefix) { |
| 233 | slog.Debug("providerext: skipping catalog ref outside the plugin namespace", "plugin", pluginID, "ref", d.Ref) |
| 234 | continue |
| 235 | } |
| 236 | out = append(out, providerconv.DescriptorFromProtocol(d)) |
| 237 | } |
| 238 | ok := !client.Crashed() |
| 239 | r.finishCatalogCall(pluginID, call, out, ok) |
| 240 | return append([]provider.Descriptor(nil), out...), ok |
| 241 | } |
| 242 | |
| 243 | // finishCatalogCall publishes one fetch atomically before waking followers. |
| 244 | // Only a live client's successful result enters the generation-local cache. |
| 245 | func (r *Resolver) finishCatalogCall(pluginID string, call *catalogCall, descriptors []provider.Descriptor, ok bool) { |
| 246 | r.mu.Lock() |
| 247 | call.descriptors = append([]provider.Descriptor(nil), descriptors...) |
| 248 | call.ok = ok |
| 249 | if ok { |
| 250 | r.catalogCache[pluginID] = append([]provider.Descriptor(nil), descriptors...) |
| 251 | } |
| 252 | delete(r.catalogCalls, pluginID) |
| 253 | close(call.done) |
| 254 | r.mu.Unlock() |
| 255 | } |
| 256 | |
| 257 | // Resolve routes a plugin-namespaced ref to the owning sidecar's Provider and |
| 258 | // everything else to the base resolver. A plugin ref whose plugin is not |
| 259 | // running, or that the plugin never declared, is an unknown-model-style |
| 260 | // error: the adapter NEVER falls back to a different provider for it. |
| 261 | func (r *Resolver) Resolve(selection provider.Selection) (provider.Provider, error) { |
| 262 | ref := strings.TrimSpace(selection.Ref) |
| 263 | if ref == "" { |
| 264 | return nil, fmt.Errorf("provider selection ref is required") |
| 265 | } |
| 266 | pluginID := PluginRefOwner(ref) |
| 267 | if pluginID == "" { |
| 268 | return r.base.Resolve(selection) |
| 269 | } |
| 270 | for _, client := range r.clients() { |
| 271 | if client.PluginID() != pluginID { |
| 272 | continue |
| 273 | } |
| 274 | descriptor, ok := declaredDescriptor(client, ref) |
| 275 | if !ok { |
| 276 | return nil, fmt.Errorf("unknown provider ref %q: extension plugin %q does not declare it", ref, pluginID) |
| 277 | } |
| 278 | return &Provider{ |
| 279 | resolver: r, |
| 280 | client: client, |
| 281 | ref: descriptor.Ref, |
| 282 | effort: selection.Effort, |
| 283 | descriptor: descriptor, |
| 284 | }, nil |
| 285 | } |
| 286 | return nil, fmt.Errorf("unknown provider ref %q: extension plugin %q is not running", ref, pluginID) |
| 287 | } |
| 288 | |
| 289 | // PluginRefOwner extracts the plugin ID from a plugin-namespaced ref |
| 290 | // (plugin/<pluginID>/<rest...>). It re-exports the protocol package's |
| 291 | // canonical namespace helper; see protocol.PluginRefOwner. |
| 292 | func PluginRefOwner(ref string) string { |
| 293 | return protocol.PluginRefOwner(ref) |
| 294 | } |
| 295 | |
| 296 | // declaredDescriptor finds the plugin's handshake declaration for ref: an |
| 297 | // exact match, or the broker-style prefix form where ref names a provider and |
| 298 | // the declaration adds the model segment. The returned descriptor carries the |
| 299 | // full declared ref. |
| 300 | func declaredDescriptor(client ProviderClient, ref string) (provider.Descriptor, bool) { |
| 301 | for _, decl := range client.Handshake().Providers { |
| 302 | if decl.Ref == ref || strings.HasPrefix(decl.Ref, ref+"/") { |
| 303 | return providerconv.DescriptorFromProtocol(decl), true |
| 304 | } |
| 305 | } |
| 306 | return provider.Descriptor{}, false |
| 307 | } |
| 308 |