返回 DeepSeek-Reasonix
extensions.go
根目录 / internal / agent / extensions.go
1 package agent
2
3 import (
4 "bytes"
5 "context"
6 "encoding/json"
7 "errors"
8 "fmt"
9 "reflect"
10 "strings"
11
12 "reasonix/internal/event"
13 "reasonix/internal/extension"
14 "reasonix/internal/extension/dispatch"
15 "reasonix/internal/extension/providerconv"
16 "reasonix/internal/provider"
17 )
18
19 // Extension Protocol v2 agent-side wiring. The agent consults the
20 // frozen dispatcher at the nine agent-loop intercept points:
21 //
22 // agent.before_start Run, before the turn is appended (block aborts the run)
23 // context.prepare stream, on the request message copy (never the session)
24 // provider.request stream, after the request is fully assembled
25 // provider.response stream, after a successful stream, before persisting
26 // tool.before executeOne, right after the call parses
27 // permission.decision executeOne, at the permission gate (allow/deny rulings)
28 // tool.after executeOne, after Execute returns (success or error)
29 // compaction.prepare compact, before the fold is archived and summarized
30 // compaction.complete compact, after the summary is produced, before persist
31 //
32 // Decision semantics are uniform: continue passes the payload through;
33 // replace substitutes it (strictly re-decoded and revalidated by the
34 // dispatcher, then checked against host invariants here); block fails the
35 // local operation — the run (before_start), the provider request (context/
36 // provider.request), the turn (provider.response), the tool call
37 // (tool.before/after, permission.decision), or the compaction pass — with the
38 // redacted reason. allow/deny is terminal at permission.decision only, where
39 // the host verdict is computed FIRST and combined: an extension allow
40 // overrides a host deny (full-trust contract, audited), an extension deny
41 // overrides a host allow, and continue leaves the host decision standing.
42 //
43 // Error policy follows the dispatcher: a required extension's failure fails
44 // the local operation; an optional extension's failure is warned about once
45 // and skipped. A nil dispatcher (no runtime packages installed) passes
46 // every point through untouched, so behavior stays byte-identical to the
47 // pre-dispatch path.
48 //
49 // Two-phase ruling at slot-mapped points: points that map to a replacement
50 // slot (context.prepare → context, provider.request → provider_request,
51 // provider.response → provider_response, compaction.prepare/complete →
52 // compaction, permission.decision → permission) first walk the intercept
53 // chain, then give the slot's OWNER the final say through RunStrategy over
54 // the (possibly interceptor-modified) payload. An owner that also declared
55 // the point under intercepts participates in both phases, in exactly this
56 // order — chain interceptor first, slot strategy last — so its strategy
57 // ruling is always the final replacement phase. A chain block short-circuits
58 // the strategy phase (the operation is already stopped). The owner is
59 // required-class by definition: its block, timeout, error, or contract
60 // violation is fatal to the local operation. Replaced values are adopted only
61 // when a replace ruling actually changed the payload, so a no-replacement
62 // walk (including an unowned slot) keeps the original values byte-identically.
63 //
64 // Ephemerality is the cache contract: context.prepare and provider.request
65 // replacements shape only the request being assembled — a.session.Messages is
66 // never mutated — while a provider.response replacement is persisted as the
67 // visible assistant turn (that IS the user's transcript), and tool.after
68 // replacements become the tool result the model reads.
69 //
70 // Observer model: after every completed intercept walk (blocked or not) the
71 // agent fires the point's fire-and-forget Event with the final payload, so
72 // observation-only extensions see exactly what the host acted on. A
73 // required-extension failure skips the event — the operation itself failed.
74
75 // extensionBlockedError reports an extension's block ruling as an operation
76 // failure. The reason is already credential-redacted by the dispatcher.
77 func extensionBlockedError(point extension.InterceptorPoint, reason string) error {
78 reason = strings.TrimSpace(reason)
79 if reason == "" {
80 reason = "no reason given"
81 }
82 return fmt.Errorf("extension blocked %s: %s", point, reason)
83 }
84
85 // extensionBlockReason normalizes a block reason for tool-result surfaces.
86 func extensionBlockReason(reason string) string {
87 reason = strings.TrimSpace(reason)
88 if reason == "" {
89 return "blocked by extension"
90 }
91 return reason
92 }
93
94 // strategyReplaced runs the replacement-slot owner's strategy for the point
95 // and reports whether a replace ruling actually changed the payload (the
96 // adoption signal for the caller's converted values). An unowned slot no-ops
97 // inside RunStrategy, so the fast path costs one comparison. The owner is
98 // required-class: a block, timeout, error, or contract violation is returned
99 // as a fatal error for the local operation.
100 func strategyReplaced(ctx context.Context, d *dispatch.Dispatcher, slot extension.Slot, point extension.InterceptorPoint, payloadPtr any) (bool, error) {
101 before := reflect.ValueOf(payloadPtr).Elem().Interface()
102 if err := d.RunStrategy(ctx, slot, point, payloadPtr); err != nil {
103 return false, err
104 }
105 return !reflect.DeepEqual(before, reflect.ValueOf(payloadPtr).Elem().Interface()), nil
106 }
107
108 // interceptAgentStart runs agent.before_start at the top of Run. A block (or
109 // a required extension's failure) aborts the run before the user turn is
110 // appended; the error surfaces like a normal run error.
111 func (a *Agent) interceptAgentStart(ctx context.Context) error {
112 d := a.svc.extensions
113 if d == nil {
114 return nil
115 }
116 providerCtx := a.withAgentContext(ctx)
117 payload := dispatch.AgentStartPayload{
118 Model: a.svc.prov.Name(),
119 ToolCount: len(a.svc.tools.SchemasForContext(providerCtx)),
120 SessionID: ParentSession(ctx),
121 }
122 result, err := d.Intercept(ctx, extension.PointAgentBeforeStart, &payload)
123 if err != nil {
124 return err
125 }
126 d.Event(extension.PointAgentBeforeStart, payload)
127 if result.Blocked {
128 return extensionBlockedError(extension.PointAgentBeforeStart, result.BlockReason)
129 }
130 return nil
131 }
132
133 // interceptContextPrepare runs context.prepare on the request message copy.
134 // The returned slice feeds only this provider request: the session log is
135 // never touched, so a replacement is invisible to the next turn (and to the
136 // prompt-cache prefix) — ephemerality is the cache contract.
137 func (a *Agent) interceptContextPrepare(ctx context.Context, messages []provider.Message) ([]provider.Message, error) {
138 d := a.svc.extensions
139 if d == nil {
140 return messages, nil
141 }
142 payload := dispatch.ContextPayload{Messages: providerconv.MessagesToProtocol(messages)}
143 result, err := d.Intercept(ctx, extension.PointContextPrepare, &payload)
144 if err != nil {
145 return nil, err
146 }
147 if result.Blocked {
148 d.Event(extension.PointContextPrepare, payload)
149 return nil, extensionBlockedError(extension.PointContextPrepare, result.BlockReason)
150 }
151 // The context slot owner gets the final say over the chain-walked payload.
152 replaced, err := strategyReplaced(ctx, d, extension.SlotContext, extension.PointContextPrepare, &payload)
153 if err != nil {
154 return nil, err
155 }
156 d.Event(extension.PointContextPrepare, payload)
157 if len(result.Applied) > 0 || replaced {
158 return providerconv.MessagesFromProtocol(payload.Messages), nil
159 }
160 return messages, nil
161 }
162
163 // interceptProviderRequest runs provider.request on the fully assembled
164 // request (post CreatedAt-strip). A replacement is revalidated by the payload
165 // registry (tool parameter schemas must be JSON objects, messages/tools must
166 // be arrays) before it may substitute the request being sent.
167 func (a *Agent) interceptProviderRequest(ctx context.Context, req provider.Request) (provider.Request, error) {
168 d := a.svc.extensions
169 if d == nil {
170 return req, nil
171 }
172 payload := dispatch.ProviderRequestPayload{Request: providerconv.RequestToProtocol(req)}
173 result, err := d.Intercept(ctx, extension.PointProviderRequest, &payload)
174 if err != nil {
175 return provider.Request{}, err
176 }
177 if result.Blocked {
178 d.Event(extension.PointProviderRequest, payload)
179 return provider.Request{}, extensionBlockedError(extension.PointProviderRequest, result.BlockReason)
180 }
181 // The provider_request slot owner gets the final say over the
182 // chain-walked payload.
183 replaced, err := strategyReplaced(ctx, d, extension.SlotProviderRequest, extension.PointProviderRequest, &payload)
184 if err != nil {
185 return provider.Request{}, err
186 }
187 d.Event(extension.PointProviderRequest, payload)
188 if len(result.Applied) > 0 || replaced {
189 return providerconv.RequestFromProtocol(payload.Request), nil
190 }
191 return req, nil
192 }
193
194 // interceptProviderResponse runs provider.response after the stream completed
195 // successfully, before the assistant turn is persisted. A replacement is
196 // persisted as the visible turn — the user's transcript and the model's own
197 // history on the next request. The live text/reasoning deltas already
198 // streamed to the frontend are not retroactively changed; the closing Message
199 // event and the session carry the replaced values. Session-level cache
200 // counters keep the provider's real usage (they were accumulated while
201 // streaming); a replaced Usage drives only this turn's Usage event and
202 // compaction decision. A block fails the turn with the redacted reason.
203 func (a *Agent) interceptProviderResponse(ctx context.Context, text, reasoning, signature string, calls []provider.ToolCall, usage *provider.Usage) (string, string, string, []provider.ToolCall, *provider.Usage, error) {
204 d := a.svc.extensions
205 if d == nil {
206 return text, reasoning, signature, calls, usage, nil
207 }
208 payload := dispatch.ProviderResponsePayload{
209 Text: text,
210 Reasoning: reasoning,
211 Signature: signature,
212 Calls: providerconv.ToolCallsToProtocol(calls),
213 Usage: providerconv.UsageToProtocol(usage),
214 }
215 result, err := d.Intercept(ctx, extension.PointProviderResponse, &payload)
216 if err != nil {
217 return "", "", "", nil, nil, err
218 }
219 if result.Blocked {
220 d.Event(extension.PointProviderResponse, payload)
221 return "", "", "", nil, nil, extensionBlockedError(extension.PointProviderResponse, result.BlockReason)
222 }
223 // The provider_response slot owner gets the final say over the
224 // chain-walked payload.
225 replaced, err := strategyReplaced(ctx, d, extension.SlotProviderResponse, extension.PointProviderResponse, &payload)
226 if err != nil {
227 return "", "", "", nil, nil, err
228 }
229 d.Event(extension.PointProviderResponse, payload)
230 if len(result.Applied) > 0 || replaced {
231 return payload.Text, payload.Reasoning, payload.Signature,
232 providerconv.ToolCallsFromProtocol(payload.Calls), providerconv.UsageFromProtocol(payload.Usage), nil
233 }
234 return text, reasoning, signature, calls, usage, nil
235 }
236
237 // interceptToolBefore runs tool.before after the host resolved and validated
238 // the concrete target. A block
239 // fails the call with the reason as the tool-result error (mirroring a
240 // PreToolUse hook block). A replacement substitutes the provider-visible name
241 // and arguments, but only after host revalidation — the arguments must decode
242 // as a JSON object and the name must still resolve in the registry — and the
243 // substituted call is then re-parsed so policy, permission, and evidence all
244 // see the call that will actually execute. An invalid replacement fails the
245 // call with a contract-violation error result.
246 func (a *Agent) interceptToolBefore(ctx context.Context, plan *toolCallPlan) (toolOutcome, bool) {
247 d := a.svc.extensions
248 if d == nil {
249 return toolOutcome{}, false
250 }
251 payload := dispatch.ToolBeforePayload{Name: plan.call.Name, Arguments: plan.call.Arguments}
252 result, err := d.Intercept(ctx, extension.PointToolBefore, &payload)
253 if err != nil {
254 msg := fmt.Sprintf("error: %v", err)
255 return toolOutcome{output: msg, errMsg: firstLine(err.Error())}, true
256 }
257 d.Event(extension.PointToolBefore, payload)
258 if result.Blocked {
259 reason := extensionBlockReason(result.BlockReason)
260 return toolOutcome{output: "blocked: " + reason, blocked: true, errMsg: "blocked by extension"}, true
261 }
262 if len(result.Applied) == 0 {
263 return toolOutcome{}, false
264 }
265 plugin := result.Applied[len(result.Applied)-1]
266 violation := func(detail string) (toolOutcome, bool) {
267 msg := fmt.Sprintf("extension %s violated the intercept contract at %s: %s", plugin, extension.PointToolBefore, detail)
268 return toolOutcome{output: "error: " + msg, errMsg: msg}, true
269 }
270 trimmed := strings.TrimSpace(payload.Arguments)
271 if trimmed == "" || trimmed[0] != '{' {
272 return violation("arguments must decode as a JSON object")
273 }
274 t, _, ambiguous := a.svc.tools.ResolveCall(payload.Name)
275 if t == nil || len(ambiguous) > 0 {
276 return violation(fmt.Sprintf("substituted tool name %q does not resolve in the registry", payload.Name))
277 }
278 plan.call.Name = payload.Name
279 plan.call.Arguments = payload.Arguments
280 return toolOutcome{}, false
281 }
282
283 // interceptExtensionPermission runs permission.decision at the gate point.
284 // The host decision is computed first and rides the payload; the extension
285 // ruling combines with it: allow overrides a host deny (the full-trust
286 // contract — the dispatcher records the audit note, surfaced here as a
287 // warning notice), deny or block overrides a host allow, continue leaves the
288 // host decision standing. allow is updated in place; early=true carries the
289 // blocked outcome.
290 func (a *Agent) interceptExtensionPermission(ctx context.Context, plan *toolCallPlan, allow *bool) (toolOutcome, bool) {
291 d := a.svc.extensions
292 if d == nil {
293 return toolOutcome{}, false
294 }
295 hostDecision := "deny"
296 if *allow {
297 hostDecision = "allow"
298 }
299 payload := dispatch.PermissionPayload{
300 Name: plan.permName,
301 Arguments: string(plan.permArgs),
302 ReadOnly: plan.readOnly,
303 HostDecision: hostDecision,
304 }
305 result, err := d.Intercept(ctx, extension.PointPermissionDecision, &payload)
306 if err != nil {
307 return toolOutcome{
308 output: fmt.Sprintf("blocked: %v", err),
309 blocked: true,
310 errMsg: "blocked by extension permission policy",
311 }, true
312 }
313 // The permission slot owner gets the final say after the chain walk. Its
314 // effective rulings here are continue (the chain/host combination stands)
315 // and block (veto); a replace adjusts only the payload observers see —
316 // allow/deny remains the chain's terminal mechanism.
317 if !result.Blocked {
318 if serr := d.RunStrategy(ctx, extension.SlotPermission, extension.PointPermissionDecision, &payload); serr != nil {
319 reason := serr.Error()
320 var blockErr *dispatch.BlockError
321 if errors.As(serr, &blockErr) {
322 reason = extensionBlockReason(blockErr.Reason)
323 }
324 return toolOutcome{
325 output: "blocked: " + reason,
326 blocked: true,
327 errMsg: "blocked by extension permission policy",
328 }, true
329 }
330 }
331 d.Event(extension.PointPermissionDecision, payload)
332 for _, note := range result.Audit {
333 a.svc.sink.Emit(event.Event{Kind: event.Notice, Level: event.LevelWarn, Text: note})
334 }
335 switch {
336 case result.Blocked:
337 reason := extensionBlockReason(result.BlockReason)
338 return toolOutcome{
339 output: "blocked: " + reason,
340 blocked: true,
341 errMsg: "blocked by extension permission policy",
342 }, true
343 case result.Permission != nil && !*result.Permission:
344 return toolOutcome{
345 output: "blocked: denied by extension permission policy",
346 blocked: true,
347 errMsg: "blocked by extension permission policy",
348 }, true
349 case result.Permission != nil && *result.Permission:
350 *allow = true
351 }
352 return toolOutcome{}, false
353 }
354
355 // interceptToolAfter runs tool.after on the executed result. A replacement
356 // substitutes the visible result string and the error flag — clearing IsError
357 // converts a failed call into a success with the replaced text, setting it
358 // converts a success into an error result carrying the replaced text. A block
359 // (or a required extension's failure) converts the call to an error tool
360 // result with the reason; the tool itself already ran.
361 func (a *Agent) interceptToolAfter(ctx context.Context, call provider.ToolCall, result string, err error) (string, error) {
362 d := a.svc.extensions
363 if d == nil {
364 return result, err
365 }
366 payload := dispatch.ToolAfterPayload{
367 Name: call.Name,
368 Arguments: call.Arguments,
369 Result: result,
370 IsError: err != nil,
371 }
372 res, ierr := d.Intercept(ctx, extension.PointToolAfter, &payload)
373 if ierr != nil {
374 return "", ierr
375 }
376 d.Event(extension.PointToolAfter, payload)
377 if res.Blocked {
378 return "", errors.New(extensionBlockReason(res.BlockReason))
379 }
380 if len(res.Applied) > 0 {
381 result = payload.Result
382 switch {
383 case payload.IsError && err == nil:
384 err = errors.New("extension replaced this tool result with an error")
385 case !payload.IsError:
386 err = nil
387 }
388 }
389 return result, err
390 }
391
392 // interceptCompactionPrepare runs compaction.prepare before the fold is
393 // archived and summarized, colocated with the PreCompact hook so the payload's
394 // Guidance is the hook-contributed guidance (plus any /compact focus text). A
395 // replacement's messages and guidance drive only this compaction pass; a
396 // block skips the pass with the reason surfaced through the caller's notice.
397 func (a *Agent) interceptCompactionPrepare(ctx context.Context, fold []provider.Message, guidance string) ([]provider.Message, string, error) {
398 d := a.svc.extensions
399 if d == nil {
400 return fold, guidance, nil
401 }
402 payload := dispatch.CompactionPreparePayload{
403 Messages: providerconv.MessagesToProtocol(fold),
404 Guidance: guidance,
405 }
406 originalMessages, err := json.Marshal(payload.Messages)
407 if err != nil {
408 return nil, "", err
409 }
410 result, err := d.Intercept(ctx, extension.PointCompactionPrepare, &payload)
411 if err != nil {
412 return nil, "", err
413 }
414 if result.Blocked {
415 d.Event(extension.PointCompactionPrepare, payload)
416 return nil, "", extensionBlockedError(extension.PointCompactionPrepare, result.BlockReason)
417 }
418 // The compaction slot owner gets the final say over the chain-walked fold
419 // and guidance.
420 replaced, err := strategyReplaced(ctx, d, extension.SlotCompaction, extension.PointCompactionPrepare, &payload)
421 if err != nil {
422 return nil, "", err
423 }
424 d.Event(extension.PointCompactionPrepare, payload)
425 if len(result.Applied) > 0 || replaced {
426 preparedMessages, marshalErr := json.Marshal(payload.Messages)
427 if marshalErr != nil {
428 return nil, "", marshalErr
429 }
430 if bytes.Equal(preparedMessages, originalMessages) {
431 return fold, payload.Guidance, nil
432 }
433 return providerconv.MessagesFromProtocol(payload.Messages), payload.Guidance, nil
434 }
435 return fold, guidance, nil
436 }
437
438 // interceptCompactionComplete runs compaction.complete after the summary is
439 // produced (including the mechanical-fold fallback), before it is written
440 // into the session. A replacement is persisted as the summary; a block skips
441 // the pass.
442 func (a *Agent) interceptCompactionComplete(ctx context.Context, summary string) (string, error) {
443 d := a.svc.extensions
444 if d == nil {
445 return summary, nil
446 }
447 payload := dispatch.CompactionCompletePayload{Summary: summary}
448 result, err := d.Intercept(ctx, extension.PointCompactionComplete, &payload)
449 if err != nil {
450 return "", err
451 }
452 if result.Blocked {
453 d.Event(extension.PointCompactionComplete, payload)
454 return "", extensionBlockedError(extension.PointCompactionComplete, result.BlockReason)
455 }
456 // The compaction slot owner gets the final say over the chain-walked
457 // summary.
458 replaced, err := strategyReplaced(ctx, d, extension.SlotCompaction, extension.PointCompactionComplete, &payload)
459 if err != nil {
460 return "", err
461 }
462 d.Event(extension.PointCompactionComplete, payload)
463 if len(result.Applied) > 0 || replaced {
464 return payload.Summary, nil
465 }
466 return summary, nil
467 }
468
468 lines GO