| 1 | package sidecar |
| 2 | |
| 3 | import ( |
| 4 | "context" |
| 5 | "fmt" |
| 6 | "log/slog" |
| 7 | "sort" |
| 8 | "strings" |
| 9 | "sync" |
| 10 | |
| 11 | "reasonix/internal/extension" |
| 12 | "reasonix/internal/extension/protocol" |
| 13 | "reasonix/internal/pluginpkg" |
| 14 | "reasonix/internal/secrets" |
| 15 | ) |
| 16 | |
| 17 | const ( |
| 18 | // maxConcurrentPackageStarts bounds process creation while still preventing |
| 19 | // one slow optional runtime from serially delaying every package after it. |
| 20 | maxConcurrentPackageStarts = 4 |
| 21 | // packageStartupBudget is shared by every runtime in one generation. Without |
| 22 | // a generation-level budget, N stalled optional runtimes could delay boot by |
| 23 | // N times the per-client handshake timeout. |
| 24 | packageStartupBudget = defaultHandshakeTimeout |
| 25 | ) |
| 26 | |
| 27 | type clientStarter func(context.Context, ClientOptions) (*Client, error) |
| 28 | |
| 29 | type packageStartJob struct { |
| 30 | item pluginpkg.InstalledPackage |
| 31 | opts ClientOptions |
| 32 | } |
| 33 | |
| 34 | type packageStartResult struct { |
| 35 | client *Client |
| 36 | err error |
| 37 | } |
| 38 | |
| 39 | // RequiredStartError reports a required runtime package that failed to start |
| 40 | // or hand shake. It fails the whole build; an optional package's failure is a |
| 41 | // warning instead. errors.As distinguishes the two at the boot call site. |
| 42 | type RequiredStartError struct { |
| 43 | Plugin string |
| 44 | Err error |
| 45 | } |
| 46 | |
| 47 | func (e *RequiredStartError) Error() string { |
| 48 | return fmt.Sprintf("required extension runtime %q failed to start: %v", e.Plugin, e.Err) |
| 49 | } |
| 50 | |
| 51 | func (e *RequiredStartError) Unwrap() error { return e.Err } |
| 52 | |
| 53 | // LoadRuntimePackages returns the installed, ENABLED packages that declare a |
| 54 | // native runtime. This is the only enumeration the Manager ever launches from — |
| 55 | // see the package doc for the authorization invariant. |
| 56 | func LoadRuntimePackages(home string) ([]pluginpkg.InstalledPackage, []string) { |
| 57 | installed, warnings := pluginpkg.LoadInstalled(home) |
| 58 | var out []pluginpkg.InstalledPackage |
| 59 | for _, item := range installed { |
| 60 | if item.Package.Manifest.Runtime != nil { |
| 61 | out = append(out, item) |
| 62 | } |
| 63 | } |
| 64 | return out, warnings |
| 65 | } |
| 66 | |
| 67 | // Manager owns every sidecar started for one runtime generation. It is the |
| 68 | // ONLY sidecar launch path, and its inputs are the pluginpkg installed state |
| 69 | // alone. It implements io.Closer so the kernel's RuntimeSet can retire it |
| 70 | // with its controller generation. |
| 71 | type Manager struct { |
| 72 | mu sync.Mutex |
| 73 | clients map[string]*Client |
| 74 | closed bool |
| 75 | // planAdopted records clients moved from a previous manager during |
| 76 | // StartPackagesWithPlan (Unchanged). RollbackPlanStart reattaches only |
| 77 | // these; newly started Added/Reloaded clients are closed by m.Close(). |
| 78 | planAdopted map[string]*Client |
| 79 | } |
| 80 | |
| 81 | // StartPackages starts every installed runtime package (cold start). Prefer |
| 82 | // StartPackagesWithPlan when a RuntimePlan can adopt unchanged packages. |
| 83 | func StartPackages(ctx context.Context, home string, sessionCtx protocol.SessionContext, ui UIHandler) (*Manager, []string, error) { |
| 84 | return StartPackagesWithPlan(ctx, home, sessionCtx, ui, nil, nil) |
| 85 | } |
| 86 | |
| 87 | // StartPackagesByName starts only the named packages and their declared |
| 88 | // package-provided dependencies. It is used by bounded provider-only work so a |
| 89 | // title request for one extension model does not launch unrelated MCP/UI/tool |
| 90 | // runtimes. Host-provided requirements are intentionally not added to the |
| 91 | // package set. |
| 92 | func StartPackagesByName(ctx context.Context, home string, sessionCtx protocol.SessionContext, ui UIHandler, names ...string) (*Manager, []string, error) { |
| 93 | packages, warnings := LoadRuntimePackages(home) |
| 94 | wanted := make(map[string]bool, len(names)) |
| 95 | for _, name := range names { |
| 96 | if name = strings.TrimSpace(name); name != "" { |
| 97 | wanted[name] = true |
| 98 | } |
| 99 | } |
| 100 | if len(wanted) == 0 { |
| 101 | return &Manager{clients: make(map[string]*Client)}, warnings, nil |
| 102 | } |
| 103 | // Resolve a transitive closure by capability identity. Version/schema |
| 104 | // compatibility is validated by the normal extension handshake and claims |
| 105 | // pipeline; this pass only decides which installed processes may start. |
| 106 | for changed := true; changed; { |
| 107 | changed = false |
| 108 | for _, item := range packages { |
| 109 | if !wanted[item.Installed.Name] { |
| 110 | continue |
| 111 | } |
| 112 | for _, requirement := range item.Package.Requires() { |
| 113 | for _, candidate := range packages { |
| 114 | if wanted[candidate.Installed.Name] { |
| 115 | continue |
| 116 | } |
| 117 | for _, provided := range candidate.Package.ProvidesCapabilities() { |
| 118 | if provided.Key == requirement.Key { |
| 119 | wanted[candidate.Installed.Name] = true |
| 120 | changed = true |
| 121 | break |
| 122 | } |
| 123 | } |
| 124 | } |
| 125 | } |
| 126 | } |
| 127 | } |
| 128 | selected := make([]pluginpkg.InstalledPackage, 0, len(wanted)) |
| 129 | for _, item := range packages { |
| 130 | if wanted[item.Installed.Name] { |
| 131 | selected = append(selected, item) |
| 132 | } |
| 133 | } |
| 134 | startupCtx, cancel := context.WithTimeout(ctx, packageStartupBudget) |
| 135 | defer cancel() |
| 136 | manager, runtimeWarnings, err := startLoadedPackages(startupCtx, selected, sessionCtx, ui, StartClient) |
| 137 | warnings = append(warnings, runtimeWarnings...) |
| 138 | return manager, warnings, err |
| 139 | } |
| 140 | |
| 141 | // startLoadedPackages starts a previously discovered, deterministically ordered |
| 142 | // package set. Handler binding stays serial; process startup and handshakes use |
| 143 | // a bounded worker pool and the caller's shared generation context. Results are |
| 144 | // consumed in package order so warnings and required-failure selection do not |
| 145 | // depend on goroutine completion order. |
| 146 | func startLoadedPackages(ctx context.Context, packages []pluginpkg.InstalledPackage, sessionCtx protocol.SessionContext, ui UIHandler, start clientStarter) (*Manager, []string, error) { |
| 147 | m := &Manager{clients: make(map[string]*Client)} |
| 148 | if len(packages) == 0 { |
| 149 | return m, nil, nil |
| 150 | } |
| 151 | var binder UIBinder |
| 152 | if b, ok := ui.(UIBinder); ok { |
| 153 | binder = b |
| 154 | } |
| 155 | jobs := make([]packageStartJob, len(packages)) |
| 156 | for i, item := range packages { |
| 157 | pluginID := item.Installed.Name |
| 158 | clientUI := ui |
| 159 | if binder != nil { |
| 160 | clientUI = binder.HandlerFor(pluginID) |
| 161 | } |
| 162 | jobs[i] = packageStartJob{item: item, opts: ClientOptions{ |
| 163 | Package: item.Package, |
| 164 | Installed: item.Installed, |
| 165 | Session: sessionCtx, |
| 166 | UI: clientUI, |
| 167 | OnCrash: func(err error) { |
| 168 | slog.Warn("extension sidecar crashed", "plugin", pluginID, "err", secrets.RedactError(err)) |
| 169 | if binder != nil { |
| 170 | binder.ClientCrashed(pluginID) |
| 171 | } |
| 172 | }, |
| 173 | }} |
| 174 | } |
| 175 | |
| 176 | results := make([]packageStartResult, len(jobs)) |
| 177 | indices := make(chan int, len(jobs)) |
| 178 | for i := range jobs { |
| 179 | indices <- i |
| 180 | } |
| 181 | close(indices) |
| 182 | workers := min(maxConcurrentPackageStarts, len(jobs)) |
| 183 | var wg sync.WaitGroup |
| 184 | wg.Add(workers) |
| 185 | for range workers { |
| 186 | go func() { |
| 187 | defer wg.Done() |
| 188 | for i := range indices { |
| 189 | if err := ctx.Err(); err != nil { |
| 190 | results[i].err = fmt.Errorf("extension generation startup stopped before launch: %w", err) |
| 191 | continue |
| 192 | } |
| 193 | results[i].client, results[i].err = start(ctx, jobs[i].opts) |
| 194 | } |
| 195 | }() |
| 196 | } |
| 197 | wg.Wait() |
| 198 | |
| 199 | var warnings []string |
| 200 | var requiredErr *RequiredStartError |
| 201 | for i, result := range results { |
| 202 | item := jobs[i].item |
| 203 | pluginID := item.Installed.Name |
| 204 | if result.err != nil { |
| 205 | if item.Package.Manifest.Runtime.Required { |
| 206 | if requiredErr == nil { |
| 207 | requiredErr = &RequiredStartError{Plugin: pluginID, Err: result.err} |
| 208 | } |
| 209 | } else { |
| 210 | warnings = append(warnings, fmt.Sprintf("%s: optional extension runtime failed to start: %v", pluginID, result.err)) |
| 211 | } |
| 212 | continue |
| 213 | } |
| 214 | m.clients[pluginID] = result.client |
| 215 | } |
| 216 | if requiredErr != nil { |
| 217 | _ = m.Close() |
| 218 | return nil, warnings, requiredErr |
| 219 | } |
| 220 | return m, warnings, nil |
| 221 | } |
| 222 | |
| 223 | // Client returns the client for one plugin ID, or nil. |
| 224 | func (m *Manager) Client(pluginID string) *Client { |
| 225 | m.mu.Lock() |
| 226 | defer m.mu.Unlock() |
| 227 | return m.clients[pluginID] |
| 228 | } |
| 229 | |
| 230 | // Clients returns every live client ordered by plugin ID. |
| 231 | func (m *Manager) Clients() []*Client { |
| 232 | m.mu.Lock() |
| 233 | defer m.mu.Unlock() |
| 234 | out := make([]*Client, 0, len(m.clients)) |
| 235 | for _, client := range m.clients { |
| 236 | out = append(out, client) |
| 237 | } |
| 238 | sort.Slice(out, func(i, j int) bool { return out[i].pluginID < out[j].pluginID }) |
| 239 | return out |
| 240 | } |
| 241 | |
| 242 | // Close shuts every sidecar down in parallel; each client's own budgets |
| 243 | // bound the total. It is idempotent. |
| 244 | func (m *Manager) Close() error { |
| 245 | m.mu.Lock() |
| 246 | if m.closed { |
| 247 | m.mu.Unlock() |
| 248 | return nil |
| 249 | } |
| 250 | m.closed = true |
| 251 | clients := make([]*Client, 0, len(m.clients)) |
| 252 | for _, client := range m.clients { |
| 253 | clients = append(clients, client) |
| 254 | } |
| 255 | m.clients = nil |
| 256 | m.planAdopted = nil |
| 257 | m.mu.Unlock() |
| 258 | |
| 259 | var wg sync.WaitGroup |
| 260 | for _, client := range clients { |
| 261 | wg.Add(1) |
| 262 | go func(c *Client) { |
| 263 | defer wg.Done() |
| 264 | _ = c.Close() |
| 265 | }(client) |
| 266 | } |
| 267 | wg.Wait() |
| 268 | return nil |
| 269 | } |
| 270 | |
| 271 | // Declaration-level kernel contribution payloads. They describe what a |
| 272 | // started sidecar declared; dispatch wiring arrives with stages 6-8. |
| 273 | |
| 274 | // InterceptorDecl is the KindInterceptor payload for one manifest-declared |
| 275 | // interceptor point. |
| 276 | type InterceptorDecl struct { |
| 277 | PluginID string |
| 278 | Point string |
| 279 | Priority int |
| 280 | } |
| 281 | |
| 282 | // StrategyDecl is the KindStrategy payload claiming one replacement slot. It |
| 283 | // implements the kernel's SlotClaimer so two runtimes claiming the same slot |
| 284 | // fail the build through ReplaceClaims. |
| 285 | type StrategyDecl struct { |
| 286 | PluginID string |
| 287 | Slots []extension.Slot |
| 288 | } |
| 289 | |
| 290 | // ReplacementSlots implements extension.SlotClaimer. |
| 291 | func (d StrategyDecl) ReplacementSlots() []extension.Slot { |
| 292 | return append([]extension.Slot(nil), d.Slots...) |
| 293 | } |
| 294 | |
| 295 | // ProviderDecl is the KindProvider payload for one handshake-declared |
| 296 | // extension-hosted provider. |
| 297 | type ProviderDecl struct { |
| 298 | PluginID string |
| 299 | Descriptor protocol.ProviderDescriptor |
| 300 | } |
| 301 | |
| 302 | // UIActionDecl is the KindUIAction payload for one handshake-declared action. |
| 303 | type UIActionDecl struct { |
| 304 | PluginID string |
| 305 | Decl protocol.UIActionDecl |
| 306 | } |
| 307 | |
| 308 | // Contributions renders every started client's declarations as kernel |
| 309 | // contributions: interceptor stubs per manifest intercept, strategy claims |
| 310 | // per manifest replaces, and provider / UI-action declarations from the |
| 311 | // validated handshake. Kernel-invalid IDs (whitespace, non-ref provider IDs) |
| 312 | // are skipped with a debug log rather than failing the whole snapshot. |
| 313 | func (m *Manager) Contributions() []extension.Contribution { |
| 314 | var out []extension.Contribution |
| 315 | for _, client := range m.Clients() { |
| 316 | rt := client.rt |
| 317 | source := extension.ContributionSource{ |
| 318 | Scope: extension.ScopePlugin, |
| 319 | PluginID: client.pluginID, |
| 320 | Version: client.version, |
| 321 | Origin: "extension-runtime", |
| 322 | } |
| 323 | for _, point := range rt.Intercepts { |
| 324 | if !kernelID(point) { |
| 325 | slog.Debug("sidecar: skipping interceptor point outside the kernel ID contract", "plugin", client.pluginID, "point", point) |
| 326 | continue |
| 327 | } |
| 328 | out = append(out, extension.Contribution{ |
| 329 | Kind: extension.KindInterceptor, |
| 330 | ID: point, |
| 331 | Source: source, |
| 332 | Priority: rt.Priority, |
| 333 | Payload: InterceptorDecl{PluginID: client.pluginID, Point: point, Priority: rt.Priority}, |
| 334 | }) |
| 335 | } |
| 336 | for _, slot := range rt.Replaces { |
| 337 | if !kernelID(slot) { |
| 338 | slog.Debug("sidecar: skipping replacement slot outside the kernel ID contract", "plugin", client.pluginID, "slot", slot) |
| 339 | continue |
| 340 | } |
| 341 | out = append(out, extension.Contribution{ |
| 342 | Kind: extension.KindStrategy, |
| 343 | ID: slot, |
| 344 | Source: source, |
| 345 | Priority: rt.Priority, |
| 346 | Payload: StrategyDecl{PluginID: client.pluginID, Slots: []extension.Slot{extension.Slot(slot)}}, |
| 347 | }) |
| 348 | } |
| 349 | result := client.Handshake() |
| 350 | prefix := "plugin/" + client.pluginID + "/" |
| 351 | for _, desc := range result.Providers { |
| 352 | ref := strings.TrimPrefix(desc.Ref, prefix) |
| 353 | if !extension.IsProviderRef(ref) { |
| 354 | slog.Debug("sidecar: skipping provider ref outside the kernel ID contract", "plugin", client.pluginID, "ref", desc.Ref) |
| 355 | continue |
| 356 | } |
| 357 | out = append(out, extension.Contribution{ |
| 358 | Kind: extension.KindProvider, |
| 359 | ID: ref, |
| 360 | Source: source, |
| 361 | Payload: ProviderDecl{PluginID: client.pluginID, Descriptor: desc}, |
| 362 | }) |
| 363 | } |
| 364 | for _, decl := range result.UIActions { |
| 365 | if !kernelID(decl.ActionID) { |
| 366 | slog.Debug("sidecar: skipping UI action outside the kernel ID contract", "plugin", client.pluginID, "action", decl.ActionID) |
| 367 | continue |
| 368 | } |
| 369 | out = append(out, extension.Contribution{ |
| 370 | Kind: extension.KindUIAction, |
| 371 | ID: decl.ActionID, |
| 372 | Source: source, |
| 373 | Payload: UIActionDecl{PluginID: client.pluginID, Decl: decl}, |
| 374 | }) |
| 375 | } |
| 376 | } |
| 377 | return out |
| 378 | } |
| 379 | |
| 380 | // kernelID mirrors the kernel's generic ID hygiene (non-empty, no |
| 381 | // whitespace); boot's legacy assembly uses the same rule. |
| 382 | func kernelID(id string) bool { |
| 383 | id = strings.TrimSpace(id) |
| 384 | return id != "" && !strings.ContainsAny(id, " \t\n") |
| 385 | } |
| 386 |