| 1 | // Package dispatch is the host-side interceptor dispatcher for Extension |
| 2 | // Protocol v1 (stage 6a). It walks the kernel's frozen interceptor chain in |
| 3 | // order, applies each extension's ruling (continue / block / replace, plus |
| 4 | // allow / deny at permission.decision only), and runs the single-owner |
| 5 | // strategy replacements for the system_prompt, context, compaction, and |
| 6 | // session_policy slots. |
| 7 | // |
| 8 | // Error policy is per extension: a required runtime (manifest required:true) |
| 9 | // or any replacement-slot owner fails the current operation when its call |
| 10 | // errors or times out; an optional observation-only extension is warned about |
| 11 | // once per process and skipped for that call. A crashed sidecar fails fast at |
| 12 | // the client layer before the dispatcher ever sees the call. |
| 13 | // |
| 14 | // The dispatcher reads only the frozen chain handed to New — per-turn dynamic |
| 15 | // data (payloads, results) never flows back into it, so one Dispatcher serves |
| 16 | // concurrent turns safely for the life of its snapshot generation. |
| 17 | package dispatch |
| 18 | |
| 19 | import ( |
| 20 | "context" |
| 21 | "encoding/json" |
| 22 | "errors" |
| 23 | "fmt" |
| 24 | "maps" |
| 25 | "reflect" |
| 26 | "slices" |
| 27 | "sync" |
| 28 | "time" |
| 29 | |
| 30 | "reasonix/internal/extension" |
| 31 | "reasonix/internal/extension/protocol" |
| 32 | "reasonix/internal/secrets" |
| 33 | ) |
| 34 | |
| 35 | // Client is the subset of sidecar.Client the dispatcher needs, abstracted for |
| 36 | // testability. *sidecar.Client satisfies it directly. The timeout argument is |
| 37 | // passed through; zero lets the sidecar resolve its per-point budget |
| 38 | // (sidecar.Client.TimeoutFor). |
| 39 | type Client interface { |
| 40 | Intercept(ctx context.Context, event protocol.InterceptEvent, payload json.RawMessage, timeout time.Duration) (protocol.InterceptResult, error) |
| 41 | TryNotifyEvent(event protocol.InterceptEvent, payload json.RawMessage) error |
| 42 | } |
| 43 | |
| 44 | // Options configures a Dispatcher. |
| 45 | type Options struct { |
| 46 | // Warn receives human-readable warnings about optional-extension failures |
| 47 | // and dropped event notifications. Nil means warnings are discarded. |
| 48 | Warn func(msg string) |
| 49 | } |
| 50 | |
| 51 | func (o Options) warnFunc() func(string) { |
| 52 | if o.Warn != nil { |
| 53 | return o.Warn |
| 54 | } |
| 55 | return func(string) {} |
| 56 | } |
| 57 | |
| 58 | // Result reports the outcome of one Intercept walk. |
| 59 | type Result struct { |
| 60 | // Blocked is true when an extension stopped the operation with a reason. |
| 61 | Blocked bool |
| 62 | // BlockReason is the extension's user-visible reason, credential-redacted. |
| 63 | BlockReason string |
| 64 | // Permission carries the terminal allow/deny ruling at permission.decision; |
| 65 | // nil means every interceptor continued and the host decision stands. The |
| 66 | // CALLER combines it with the host verdict (host deny + extension allow → |
| 67 | // allow for full-trust extensions). |
| 68 | Permission *bool |
| 69 | // Applied lists the plugin IDs that replaced the payload, in chain order. |
| 70 | Applied []string |
| 71 | // Audit holds redacted audit notes, e.g. an extension allow overriding a |
| 72 | // host deny at permission.decision. |
| 73 | Audit []string |
| 74 | } |
| 75 | |
| 76 | // ViolationError reports an extension answer that breaks the dispatch rules: |
| 77 | // allow/deny outside permission.decision, an unknown decision, or a |
| 78 | // replacement that fails strict decoding against the point's DTO. The Detail |
| 79 | // is credential-redacted. |
| 80 | type ViolationError struct { |
| 81 | Plugin string |
| 82 | Point extension.InterceptorPoint |
| 83 | Detail string |
| 84 | } |
| 85 | |
| 86 | // Error returns the redacted violation description. |
| 87 | func (e *ViolationError) Error() string { |
| 88 | return fmt.Sprintf("extension %s violated the intercept contract at %s: %s", e.Plugin, e.Point, e.Detail) |
| 89 | } |
| 90 | |
| 91 | // FailureError reports a required extension's call failure (timeout, crash, |
| 92 | // transport). The message is credential-redacted; Unwrap returns the original |
| 93 | // error so errors.As still finds *protocol.ProtocolError and its frozen |
| 94 | // reason. |
| 95 | type FailureError struct { |
| 96 | Plugin string |
| 97 | Point extension.InterceptorPoint |
| 98 | Err error |
| 99 | } |
| 100 | |
| 101 | // Error returns the redacted failure description. |
| 102 | func (e *FailureError) Error() string { |
| 103 | return fmt.Sprintf("extension %s failed at %s: %s", e.Plugin, e.Point, secrets.RedactCredentials(e.Err.Error())) |
| 104 | } |
| 105 | |
| 106 | // Unwrap returns the original call error. |
| 107 | func (e *FailureError) Unwrap() error { return e.Err } |
| 108 | |
| 109 | // BlockError reports a strategy owner blocking the operation. The Reason is |
| 110 | // credential-redacted. |
| 111 | type BlockError struct { |
| 112 | Plugin string |
| 113 | Point extension.InterceptorPoint |
| 114 | Reason string |
| 115 | } |
| 116 | |
| 117 | // Error returns the redacted block description. |
| 118 | func (e *BlockError) Error() string { |
| 119 | return fmt.Sprintf("extension %s blocked %s: %s", e.Plugin, e.Point, e.Reason) |
| 120 | } |
| 121 | |
| 122 | // Dispatcher applies the frozen interceptor chain to live payloads. It is |
| 123 | // immutable after New — every map and slice is deep-copied at construction — |
| 124 | // so concurrent turns may dispatch through one Dispatcher without locking. |
| 125 | // The single exception is the warn-once dedup set, guarded by warnedMu. |
| 126 | type Dispatcher struct { |
| 127 | chain map[extension.InterceptorPoint][]extension.Contribution |
| 128 | replacements map[extension.Slot]extension.ContributionSource |
| 129 | clients func(pluginID string) Client |
| 130 | required map[string]bool |
| 131 | slotOwners map[string]bool |
| 132 | warn func(string) |
| 133 | |
| 134 | warnedMu sync.Mutex |
| 135 | warned map[string]struct{} |
| 136 | } |
| 137 | |
| 138 | // New freezes the dispatch inputs into a Dispatcher. chain is the snapshot's |
| 139 | // kernel-sorted InterceptorChain (priority ascending, plugin ID, registration |
| 140 | // order); the dispatcher walks it exactly as given. replacements is the |
| 141 | // snapshot's Replacements map. clients resolves a plugin ID to its sidecar |
| 142 | // client (nil means no live sidecar; it must return an untyped nil). required |
| 143 | // marks plugins whose manifest declared required:true. opts.Warn defaults to |
| 144 | // a no-op. |
| 145 | func New(chain map[extension.InterceptorPoint][]extension.Contribution, replacements map[extension.Slot]extension.ContributionSource, clients func(pluginID string) Client, required map[string]bool, opts Options) *Dispatcher { |
| 146 | frozenChain := make(map[extension.InterceptorPoint][]extension.Contribution, len(chain)) |
| 147 | for point, contribs := range chain { |
| 148 | frozenChain[point] = slices.Clone(contribs) |
| 149 | } |
| 150 | frozenRequired := make(map[string]bool, len(required)) |
| 151 | maps.Copy(frozenRequired, required) |
| 152 | slotOwners := make(map[string]bool, len(replacements)) |
| 153 | for _, owner := range replacements { |
| 154 | if owner.PluginID != "" { |
| 155 | slotOwners[owner.PluginID] = true |
| 156 | } |
| 157 | } |
| 158 | return &Dispatcher{ |
| 159 | chain: frozenChain, |
| 160 | replacements: maps.Clone(replacements), |
| 161 | clients: clients, |
| 162 | required: frozenRequired, |
| 163 | slotOwners: slotOwners, |
| 164 | warn: opts.warnFunc(), |
| 165 | warned: map[string]struct{}{}, |
| 166 | } |
| 167 | } |
| 168 | |
| 169 | // Intercept walks the chain for point in frozen order, calling each |
| 170 | // plugin-backed interceptor with the current (possibly already replaced) |
| 171 | // payload. payloadPtr must be a pointer to the point's registered DTO; on |
| 172 | // return it holds the final value after any replace rulings. Interceptors |
| 173 | // whose contribution has no plugin ID are not sidecar-addressable and are |
| 174 | // skipped. |
| 175 | // |
| 176 | // Rulings: continue passes the payload through; block stops the operation and |
| 177 | // reports the redacted reason; replace substitutes the payload after strict |
| 178 | // re-decoding against the point's DTO; allow/deny are terminal at |
| 179 | // permission.decision and a protocol violation anywhere else. A required |
| 180 | // extension's call failure or contract violation fails the operation; an |
| 181 | // optional extension's is warned about once and skipped. |
| 182 | func (d *Dispatcher) Intercept(ctx context.Context, point extension.InterceptorPoint, payloadPtr any) (*Result, error) { |
| 183 | if _, err := checkPayloadType(point, payloadPtr); err != nil { |
| 184 | return nil, err |
| 185 | } |
| 186 | raw, err := json.Marshal(payloadPtr) |
| 187 | if err != nil { |
| 188 | return nil, fmt.Errorf("dispatch: marshal %s payload: %w", point, err) |
| 189 | } |
| 190 | result := &Result{} |
| 191 | // Capture the host verdict before the walk: a replace ruling rewrites the |
| 192 | // payload, but the audit note must reflect the decision the HOST made. |
| 193 | hostDecision := "" |
| 194 | if permission, ok := payloadPtr.(*PermissionPayload); ok { |
| 195 | hostDecision = permission.HostDecision |
| 196 | } |
| 197 | for _, contribution := range d.chain[point] { |
| 198 | pluginID := contribution.Source.PluginID |
| 199 | if pluginID == "" { |
| 200 | continue |
| 201 | } |
| 202 | client := d.clients(pluginID) |
| 203 | if client == nil { |
| 204 | if err := d.failure(pluginID, point, errors.New("no live sidecar client")); err != nil { |
| 205 | return nil, err |
| 206 | } |
| 207 | continue |
| 208 | } |
| 209 | answer, callErr := client.Intercept(ctx, protocol.InterceptEvent(point), raw, 0) |
| 210 | if callErr != nil { |
| 211 | if err := d.failure(pluginID, point, callErr); err != nil { |
| 212 | return nil, err |
| 213 | } |
| 214 | continue |
| 215 | } |
| 216 | switch answer.Decision { |
| 217 | case protocol.DecisionContinue: |
| 218 | // Pass the current payload to the next interceptor unchanged. |
| 219 | case protocol.DecisionBlock: |
| 220 | result.Blocked = true |
| 221 | result.BlockReason = secrets.RedactCredentials(answer.Reason) |
| 222 | return result, nil |
| 223 | case protocol.DecisionReplace: |
| 224 | fresh, decodeErr := decodePayload(point, answer.Replacement) |
| 225 | if decodeErr != nil { |
| 226 | if err := d.violation(pluginID, point, decodeErr); err != nil { |
| 227 | return nil, err |
| 228 | } |
| 229 | continue |
| 230 | } |
| 231 | if err := assignPayload(payloadPtr, fresh); err != nil { |
| 232 | return nil, err |
| 233 | } |
| 234 | raw, err = json.Marshal(fresh) |
| 235 | if err != nil { |
| 236 | return nil, fmt.Errorf("dispatch: marshal %s replacement: %w", point, err) |
| 237 | } |
| 238 | result.Applied = append(result.Applied, pluginID) |
| 239 | case protocol.DecisionAllow, protocol.DecisionDeny: |
| 240 | if point != extension.PointPermissionDecision { |
| 241 | err := fmt.Errorf("decision %q is only legal at %s", answer.Decision, extension.PointPermissionDecision) |
| 242 | if err := d.violation(pluginID, point, err); err != nil { |
| 243 | return nil, err |
| 244 | } |
| 245 | continue |
| 246 | } |
| 247 | allow := answer.Decision == protocol.DecisionAllow |
| 248 | result.Permission = &allow |
| 249 | if allow && hostDecision == "deny" { |
| 250 | result.Audit = append(result.Audit, secrets.RedactCredentials(fmt.Sprintf( |
| 251 | "extension %s allowed the tool overriding the host deny", pluginID))) |
| 252 | } |
| 253 | // The first allow/deny is terminal for the extension phase. |
| 254 | return result, nil |
| 255 | default: |
| 256 | // Unreachable through a real sidecar — the protocol registry |
| 257 | // rejects unknown decisions — but fakes and future peers can send |
| 258 | // anything; treat it as a contract violation. |
| 259 | err := fmt.Errorf("unknown decision %q", answer.Decision) |
| 260 | if err := d.violation(pluginID, point, err); err != nil { |
| 261 | return nil, err |
| 262 | } |
| 263 | } |
| 264 | } |
| 265 | return result, nil |
| 266 | } |
| 267 | |
| 268 | // Strategy returns the client of the replacement slot's owner, or false when |
| 269 | // the slot is unowned (the host default stands). |
| 270 | func (d *Dispatcher) Strategy(slot extension.Slot) (Client, bool) { |
| 271 | owner, ok := d.replacements[slot] |
| 272 | if !ok { |
| 273 | return nil, false |
| 274 | } |
| 275 | return d.clients(owner.PluginID), true |
| 276 | } |
| 277 | |
| 278 | // RunStrategy asks the slot's owner to rule on the payload at point and |
| 279 | // applies a replace ruling to payloadPtr (validated against the point's DTO, |
| 280 | // exactly like Intercept). The owner is a required-class extension by |
| 281 | // definition: a timeout, error, contract violation, or block is fatal to the |
| 282 | // current operation. An unowned slot is a no-op and keeps the host default. |
| 283 | func (d *Dispatcher) RunStrategy(ctx context.Context, slot extension.Slot, point extension.InterceptorPoint, payloadPtr any) error { |
| 284 | if _, err := checkPayloadType(point, payloadPtr); err != nil { |
| 285 | return err |
| 286 | } |
| 287 | owner, ok := d.replacements[slot] |
| 288 | if !ok { |
| 289 | return nil |
| 290 | } |
| 291 | client := d.clients(owner.PluginID) |
| 292 | if client == nil { |
| 293 | return &FailureError{Plugin: owner.PluginID, Point: point, Err: errors.New("no live sidecar client")} |
| 294 | } |
| 295 | raw, err := json.Marshal(payloadPtr) |
| 296 | if err != nil { |
| 297 | return fmt.Errorf("dispatch: marshal %s strategy payload: %w", point, err) |
| 298 | } |
| 299 | answer, callErr := client.Intercept(ctx, protocol.InterceptEvent(point), raw, 0) |
| 300 | if callErr != nil { |
| 301 | return &FailureError{Plugin: owner.PluginID, Point: point, Err: callErr} |
| 302 | } |
| 303 | switch answer.Decision { |
| 304 | case protocol.DecisionContinue: |
| 305 | return nil |
| 306 | case protocol.DecisionReplace: |
| 307 | fresh, decodeErr := decodePayload(point, answer.Replacement) |
| 308 | if decodeErr != nil { |
| 309 | return &ViolationError{Plugin: owner.PluginID, Point: point, Detail: secrets.RedactCredentials(decodeErr.Error())} |
| 310 | } |
| 311 | return assignPayload(payloadPtr, fresh) |
| 312 | case protocol.DecisionBlock: |
| 313 | return &BlockError{Plugin: owner.PluginID, Point: point, Reason: secrets.RedactCredentials(answer.Reason)} |
| 314 | default: |
| 315 | return &ViolationError{Plugin: owner.PluginID, Point: point, Detail: secrets.RedactCredentials( |
| 316 | fmt.Sprintf("strategy ruling %q is not continue or replace", answer.Decision))} |
| 317 | } |
| 318 | } |
| 319 | |
| 320 | // Event broadcasts a fire-and-forget extension/event notification to every |
| 321 | // chain member at point plus the owners of the slots that observe that point |
| 322 | // (deduplicated by plugin). Delivery is a non-blocking bounded enqueue: |
| 323 | // failures and queue saturation are warned about once per plugin and never |
| 324 | // fail or stall the caller. |
| 325 | func (d *Dispatcher) Event(point extension.InterceptorPoint, payload any) { |
| 326 | raw, err := json.Marshal(payload) |
| 327 | if err != nil { |
| 328 | d.warn(fmt.Sprintf("dispatch: dropping %s event: marshal: %v", point, err)) |
| 329 | return |
| 330 | } |
| 331 | seen := map[string]bool{} |
| 332 | notify := func(pluginID string) { |
| 333 | if pluginID == "" || seen[pluginID] { |
| 334 | return |
| 335 | } |
| 336 | seen[pluginID] = true |
| 337 | client := d.clients(pluginID) |
| 338 | if client == nil { |
| 339 | return |
| 340 | } |
| 341 | if err := client.TryNotifyEvent(protocol.InterceptEvent(point), raw); err != nil { |
| 342 | d.warnOnce("event|"+pluginID, fmt.Sprintf( |
| 343 | "extension %s dropped the %s event: %s", pluginID, point, secrets.RedactCredentials(err.Error()))) |
| 344 | } |
| 345 | } |
| 346 | for _, contribution := range d.chain[point] { |
| 347 | notify(contribution.Source.PluginID) |
| 348 | } |
| 349 | for _, slot := range slotsForPoint(point) { |
| 350 | if owner, ok := d.replacements[slot]; ok { |
| 351 | notify(owner.PluginID) |
| 352 | } |
| 353 | } |
| 354 | } |
| 355 | |
| 356 | // slotsForPoint maps a point to the replacement slots whose owners observe it |
| 357 | // for event broadcasts. Per-tool ("tool:<name>") and per-provider |
| 358 | // ("provider:<ref>") slots are addressed by strategy dispatch, not broadcast, |
| 359 | // so they are not mapped here. |
| 360 | func slotsForPoint(point extension.InterceptorPoint) []extension.Slot { |
| 361 | switch point { |
| 362 | case extension.PointSystemPromptBuild: |
| 363 | return []extension.Slot{extension.SlotSystemPrompt} |
| 364 | case extension.PointContextPrepare: |
| 365 | return []extension.Slot{extension.SlotContext} |
| 366 | case extension.PointProviderRequest: |
| 367 | return []extension.Slot{extension.SlotProviderRequest} |
| 368 | case extension.PointProviderResponse: |
| 369 | return []extension.Slot{extension.SlotProviderResponse} |
| 370 | case extension.PointCompactionPrepare, extension.PointCompactionComplete: |
| 371 | return []extension.Slot{extension.SlotCompaction} |
| 372 | case extension.PointSessionStart, extension.PointSessionEnd, extension.PointSessionLoad, |
| 373 | extension.PointSessionSave, extension.PointSessionRotate: |
| 374 | return []extension.Slot{extension.SlotSessionPolicy} |
| 375 | case extension.PointPermissionDecision: |
| 376 | return []extension.Slot{extension.SlotPermission} |
| 377 | case extension.PointFrontendEvent: |
| 378 | return []extension.Slot{extension.SlotFrontendEvents} |
| 379 | default: |
| 380 | return nil |
| 381 | } |
| 382 | } |
| 383 | |
| 384 | // isRequired reports whether the plugin is required-class: manifest |
| 385 | // required:true or the owner of any replacement slot. Required-class failures |
| 386 | // fail the operation; optional failures are warned about and skipped. |
| 387 | func (d *Dispatcher) isRequired(pluginID string) bool { |
| 388 | return d.required[pluginID] || d.slotOwners[pluginID] |
| 389 | } |
| 390 | |
| 391 | // failure applies the error policy for a failed intercept call. Required |
| 392 | // extensions fail the operation; optional extensions warn once and skip. |
| 393 | func (d *Dispatcher) failure(pluginID string, point extension.InterceptorPoint, err error) error { |
| 394 | if d.isRequired(pluginID) { |
| 395 | return &FailureError{Plugin: pluginID, Point: point, Err: err} |
| 396 | } |
| 397 | d.warnOnce("error|"+pluginID, fmt.Sprintf( |
| 398 | "extension %s failed at %s; skipping this optional extension: %s", |
| 399 | pluginID, point, secrets.RedactCredentials(err.Error()))) |
| 400 | return nil |
| 401 | } |
| 402 | |
| 403 | // violation applies the error policy for an answer that breaks the dispatch |
| 404 | // rules. Required extensions fail the operation; optional extensions warn |
| 405 | // once and their ruling is skipped. |
| 406 | func (d *Dispatcher) violation(pluginID string, point extension.InterceptorPoint, detail error) error { |
| 407 | violation := &ViolationError{Plugin: pluginID, Point: point, Detail: secrets.RedactCredentials(detail.Error())} |
| 408 | if d.isRequired(pluginID) { |
| 409 | return violation |
| 410 | } |
| 411 | d.warnOnce("violation|"+pluginID, violation.Error()+"; skipping this optional extension's ruling") |
| 412 | return nil |
| 413 | } |
| 414 | |
| 415 | // warnOnce delivers msg through Options.Warn at most once per key for the |
| 416 | // life of the process. |
| 417 | func (d *Dispatcher) warnOnce(key, msg string) { |
| 418 | d.warnedMu.Lock() |
| 419 | if _, dup := d.warned[key]; dup { |
| 420 | d.warnedMu.Unlock() |
| 421 | return |
| 422 | } |
| 423 | d.warned[key] = struct{}{} |
| 424 | d.warnedMu.Unlock() |
| 425 | d.warn(msg) |
| 426 | } |
| 427 | |
| 428 | // checkPayloadType verifies payloadPtr is a pointer to exactly the DTO |
| 429 | // registered for point. A mismatch is a host programming error, not an |
| 430 | // extension failure, so it always returns an error. |
| 431 | func checkPayloadType(point extension.InterceptorPoint, payloadPtr any) (payloadFactory, error) { |
| 432 | factory, ok := payloadRegistry[point] |
| 433 | if !ok { |
| 434 | return nil, fmt.Errorf("dispatch: no payload DTO registered for %s", point) |
| 435 | } |
| 436 | want := reflect.TypeOf(factory()) |
| 437 | if got := reflect.TypeOf(payloadPtr); got != want { |
| 438 | return nil, fmt.Errorf("dispatch: %s payload must be %s, got %s", point, want, got) |
| 439 | } |
| 440 | return factory, nil |
| 441 | } |
| 442 | |
| 443 | // assignPayload replaces the value payloadPtr points to with the freshly |
| 444 | // decoded replacement. Whole-value assignment keeps omitted (zero) fields in |
| 445 | // the replacement from leaking the previous value through. |
| 446 | func assignPayload(payloadPtr, fresh any) error { |
| 447 | target := reflect.ValueOf(payloadPtr) |
| 448 | source := reflect.ValueOf(fresh) |
| 449 | if target.Kind() != reflect.Pointer || target.IsNil() || target.Type() != source.Type() { |
| 450 | return fmt.Errorf("dispatch: cannot assign %T over %T", fresh, payloadPtr) |
| 451 | } |
| 452 | target.Elem().Set(source.Elem()) |
| 453 | return nil |
| 454 | } |
| 455 |