Similarityoung commented on code in PR #953:
URL: https://github.com/apache/dubbo-go-pixiu/pull/953#discussion_r3485579823
##########
pkg/filter/auth/mcp/filter.go:
##########
@@ -25,6 +25,10 @@ import (
"strings"
)
+import (
+ "github.com/lestrrat-go/jwx/v3/jwt"
+)
Review Comment:
filter 这地方需要引用这个依赖吗，这个我感觉应该在内部做引用的
##########
pkg/filter/mcp/mcpserver/dynamic.go:
##########
@@ -224,70 +280,108 @@ func (d *DynamicConsumer) removeServerConfig(serverId
string) error {
func (d *DynamicConsumer) calculateCurrentMergedTools() []model.ToolConfig {
var allTools []model.ToolConfig
- // Simply accumulate tools from all servers
- for _, config := range d.serverConfigs {
- allTools = append(allTools, config.Tools...)
+ serverIDs := make([]string, 0, len(d.serverConfigs))
+ for serverID := range d.serverConfigs {
+ serverIDs = append(serverIDs, serverID)
+ }
+ sort.Strings(serverIDs)
+
+ for _, serverID := range serverIDs {
+ allTools = append(allTools, d.serverConfigs[serverID].Tools...)
}
return allTools
}
-// notifyToolsListChanged sends notifications/tools/list_changed to all
connected clients
+// notifyToolsListChanged marks notifications/tools/list_changed only for
+// sessions whose visible tool set changed. Offline sessions keep one bounded
+// pending version for reconnect.
func (d *DynamicConsumer) notifyToolsListChanged() {
if d.sessionManager == nil {
logger.Debugf("[dubbo-go-pixiu] mcp server session manager not
available, skip tools list_changed notification")
return
}
- // Get all active sessions
- sessionIDs := d.sessionManager.AllSessionIDs()
- if len(sessionIDs) == 0 {
+ sessions := d.sessionManager.SnapshotSessions()
+ if len(sessions) == 0 {
logger.Debugf("[dubbo-go-pixiu] mcp server no active sessions,
skip tools list_changed notification")
return
}
- // Send notification to each session
- successCount := 0
- for _, sessionID := range sessionIDs {
- if err := d.sendToolsListChangedNotification(sessionID); err !=
nil {
- logger.Warnf("[dubbo-go-pixiu] mcp server failed to
send tools list_changed to session %s: %v", sessionID, err)
- } else {
- successCount++
+ markedCount := 0
+ flushedCount := 0
+ changed := d.sessionsWithChangedVisibleSet()
+ for _, session := range sessions {
+ if changed != nil {
+ if _, ok := changed[session.ID]; !ok {
+ continue
+ }
+ }
+ marked, flushed, err :=
markToolsListChangedAndFlush(d.sseHandler, session)
+ if !marked {
+ continue
+ }
+ markedCount++
+ if err != nil {
+ logger.Warnf("[dubbo-go-pixiu] mcp server failed to
send tools list_changed: %v", err)
+ } else if flushed {
+ flushedCount++
}
}
- logger.Infof("[dubbo-go-pixiu] mcp server sent tools/list_changed
notification to %d/%d sessions", successCount, len(sessionIDs))
-}
-
-// sendToolsListChangedNotification sends notification to a specific session
-func (d *DynamicConsumer) sendToolsListChangedNotification(sessionID string)
error {
- session, exists := d.sessionManager.Session(sessionID)
- if !exists {
- return fmt.Errorf("session not found")
+ if markedCount == 0 {
+ logger.Debugf("[dubbo-go-pixiu] mcp server no active sessions
to mark for tools list_changed notification")
+ return
}
+ logger.Infof("[dubbo-go-pixiu] mcp server marked tools/list_changed
notification for %d sessions, flushed %d online sessions", markedCount,
flushedCount)
+}
- if session.PipeWriter == nil {
- return fmt.Errorf("SSE pipe not established")
+func (d *DynamicConsumer) sessionsWithChangedVisibleSet() map[string]struct{} {
+ d.mu.RLock()
+ governance := d.governance
+ selector := d.selector
+ plans := d.plans
+ d.mu.RUnlock()
+ if !governance || selector == nil || plans == nil {
+ return nil
}
-
- // Build tools/list_changed notification (no params needed)
- notification := map[string]any{
- "jsonrpc": "2.0",
- "method": "notifications/tools/list_changed",
+ refresher, ok := selector.(router.PlanRefreshSelector)
+ if !ok {
+ return nil
}
-
- messageJSON, err := json.Marshal(notification)
- if err != nil {
- return fmt.Errorf("failed to marshal notification: %w", err)
+ snapshot := d.registry.toolCatalogSnapshotUnsafe()
+ changed := make(map[string]struct{})
+ for _, item := range plans.SessionPlanContexts() {
+ item.Context.CatalogVersion = snapshot.Version
+ plan, committed, err :=
refresher.RefreshPlan(context.Background(), item, snapshot.orderedToolsUnsafe())
+ if err != nil {
+ logger.Warnf("[dubbo-go-pixiu] mcp server failed to
refresh session plan after catalog update: %v", err)
+ continue
+ }
+ if !committed {
+ continue
+ }
+ if !sameStrings(item.Plan.VisibleNames(), plan.VisibleNames()) {
+ changed[item.Key.SessionID] = struct{}{}
Review Comment:
这里判断通知只比较 `VisibleNames()`。如果工具名不变，但 description、args、input schema 或其他 tool
definition 发生变化，客户端可能收不到 `tools/list_changed`，继续使用旧 schema。建议比较可见工具定义的
fingerprint，而不是只比较名称。
##########
pkg/filter/mcp/mcpserver/router/composite.go:
##########
@@ -0,0 +1,597 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package router
+
+import (
+ "context"
+ "encoding/json"
+ "fmt"
+ "hash/fnv"
+ "io"
+ "sort"
+ "strconv"
+ "time"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/model"
+)
+
+// Fallback strategies.
+const (
+ DefaultFallback = FallbackFailClosed
+ FallbackBundleDefault = "bundle_default"
+ FallbackFailClosed = "fail_closed"
+)
+
+// CompositeSelector runs the deterministic selection pipeline:
+//
+// policy -> workflow -> progressive
+//
+// Each stage is optional; a nil stage is skipped. Results are cached per
session
+// in the SessionPlanStore and reused while the metadata/config version is
stable.
+type CompositeSelector struct {
+ policy *PolicyFilter
+ workflow *WorkflowSelector
+ progressive *ProgressiveGate
+ bundles *WorkflowSelector
+
+ store *SessionPlanStore
+ log *DecisionLogger
+
+ routerID string
+ fallback string
+ defaultBundle string
+ configHash string
+ progressiveHash string
+}
+
+// CompositeOptions configures a CompositeSelector. The builder populates it.
+type CompositeOptions struct {
+ Policy *PolicyFilter
+ Workflow *WorkflowSelector
+ Progressive *ProgressiveGate
+ Bundles *WorkflowSelector
+ Store *SessionPlanStore
+ Log *DecisionLogger
+ Fallback string
+ DefaultBundle string
+ ConfigHash string
+ RouterID string
+}
+
+// NewCompositeSelector assembles the pipeline from validated options.
+func NewCompositeSelector(opts CompositeOptions) (*CompositeSelector, error) {
+ if opts.Store == nil {
+ return nil, fmt.Errorf("router session plan store is nil")
+ }
+ fallback := opts.Fallback
+ if fallback == "" {
+ fallback = DefaultFallback
+ }
+ if err := validateFallback(fallback); err != nil {
+ return nil, err
+ }
+ log := opts.Log
+ if log == nil {
+ log = NewDecisionLogger(0, false)
+ }
+ return &CompositeSelector{
+ policy: opts.Policy,
+ workflow: opts.Workflow,
+ progressive: opts.Progressive,
+ bundles: opts.Bundles,
+ store: opts.Store,
+ log: log,
+ routerID: opts.RouterID,
+ fallback: fallback,
+ defaultBundle: opts.DefaultBundle,
+ configHash: opts.ConfigHash,
+ progressiveHash: progressiveConfigHash(opts.Progressive),
+ }, nil
+}
+
+type selectionPipeline struct {
+ tools []model.ToolConfig
+ policyAllowed []model.ToolConfig
+ traces []DecisionTrace
+ stageCounts map[string]StageCount
+ outcome string
+ expanded bool
+}
+
+func newSelectionPipeline(candidates []model.ToolConfig) *selectionPipeline {
+ return &selectionPipeline{
+ tools: candidates,
+ policyAllowed: candidates,
+ stageCounts: make(map[string]StageCount),
+ outcome: SelectionOutcomeSelected,
+ }
+}
+
+// Select runs the pipeline, reusing a cached plan when the version is
unchanged.
+func (c *CompositeSelector) Select(_ context.Context, sc SelectionContext,
candidates []model.ToolConfig) (*SelectionPlan, error) {
+ start := time.Now()
+ identityHash, err := identityFingerprint(sc)
+ if err != nil {
+ return nil, err
+ }
+ version := c.version(candidates, identityHash, sc.SessionID,
sc.CatalogVersion)
+ key := c.planKey(sc.SessionID)
+
+ if cached, ok := c.cachedSelectionPlan(key, version, candidates,
start); ok {
+ return cached, nil
+ }
+
+ plan := c.computeSelectionPlan(key, identityHash, sc, candidates,
version)
+ plan, err = c.storeSelectionPlan(key, plan, sc)
+ if err != nil {
+ recordFallback("plan_persistence_failed")
+ return nil, err
+ }
+
+ result := "ok"
+ if outcomeAllowsFallback(plan.Outcome) {
+ result = "fallback"
+ recordFallback(plan.Outcome)
+ }
+
+ c.recordSelectionResult(result, plan, candidates, start)
+ c.log.Log(sc, plan, len(candidates))
+
+ return plan, nil
+}
+
+func (c *CompositeSelector) cachedSelectionPlan(key PlanKey, version string,
candidates []model.ToolConfig, start time.Time) (*SelectionPlan, bool) {
+ cached, ok := c.store.Get(key)
+ if !ok || cached.Version != version {
+ return nil, false
+ }
+ elapsedMS := float64(time.Since(start).Microseconds()) / 1000.0
+ recordSelection("cached", cached.Mode, len(candidates),
len(cached.ToolNames), elapsedMS)
+ return cached, true
+}
+
+func (c *CompositeSelector) computeSelectionPlan(key PlanKey, identityHash
string, sc SelectionContext, candidates []model.ToolConfig, version string)
*SelectionPlan {
+ pipeline := c.runSelectionPipeline(key, identityHash, sc, candidates)
+ if outcomeAllowsFallback(pipeline.outcome) {
+ return c.applyFallback(sc, pipeline.policyAllowed, version,
pipeline.traces, pipeline.stageCounts, pipeline.outcome)
+ }
+ return c.newSelectionPlan(sc, candidates, version, identityHash,
pipeline)
+}
+
+func (c *CompositeSelector) runSelectionPipeline(key PlanKey, identityHash
string, sc SelectionContext, candidates []model.ToolConfig) *selectionPipeline {
+ pipeline := newSelectionPipeline(candidates)
+ pipeline.applyPolicy(c.policy, sc)
+ pipeline.applyWorkflow(c.workflow, sc)
+ pipeline.applyProgressive(c.progressive, c.store, key, identityHash,
c.progressiveHash)
+ pipeline.finalizeOutcome()
+ return pipeline
+}
+
+func (p *selectionPipeline) applyPolicy(policy *PolicyFilter, sc
SelectionContext) {
+ if policy == nil {
+ return
+ }
+ input := len(p.tools)
+ var traces []DecisionTrace
+ p.tools, traces = policy.Filter(p.tools, sc)
+ recordStageCount(p.stageCounts, StagePolicy, input, len(p.tools))
+ p.policyAllowed = p.tools
+ p.traces = append(p.traces, traces...)
+ if input > 0 && len(p.tools) == 0 {
+ p.outcome = SelectionOutcomeExplicitDeny
+ }
+}
+
+func (p *selectionPipeline) applyWorkflow(workflow *WorkflowSelector, sc
SelectionContext) {
+ if !p.selected() || workflow == nil {
+ return
+ }
+ input := len(p.tools)
+ result := workflow.filter(p.tools, sc)
+ p.tools = result.tools
+ recordStageCount(p.stageCounts, StageWorkflow, input, len(p.tools))
+ p.traces = append(p.traces, result.traces...)
+ if !result.matched && workflow.hasMatchableWorkflows() {
+ p.outcome = SelectionOutcomeNoMatch
+ return
+ }
+ if input > 0 && len(p.tools) == 0 {
+ p.outcome = SelectionOutcomeEmptyByConfiguration
+ }
+}
+
+func (p *selectionPipeline) applyProgressive(progressive *ProgressiveGate,
store *SessionPlanStore, key PlanKey, identityHash, progressiveHash string) {
+ if !p.selected() || progressive == nil {
+ return
+ }
+ input := len(p.tools)
+ var traces []DecisionTrace
+ _, p.expanded = store.CallState(key, identityHash, progressiveHash)
+ p.tools, traces = progressive.Apply(p.tools, p.expanded)
+ recordStageCount(p.stageCounts, StageProgressive, input, len(p.tools))
+ p.traces = append(p.traces, traces...)
+ if input > 0 && len(p.tools) == 0 {
+ p.outcome = SelectionOutcomeEmptyByConfiguration
+ }
+}
+
+func (p *selectionPipeline) finalizeOutcome() {
+ if p.selected() && len(p.tools) == 0 {
+ p.outcome = SelectionOutcomeNoMatch
+ }
+}
+
+func (p *selectionPipeline) selected() bool {
+ return p.outcome == SelectionOutcomeSelected
+}
+
+func (c *CompositeSelector) newSelectionPlan(sc SelectionContext, candidates
[]model.ToolConfig, version, identityHash string, pipeline *selectionPipeline)
*SelectionPlan {
+ plan := &SelectionPlan{
+ SessionID: sc.SessionID,
+ ToolNames: toolNames(pipeline.tools),
+ VisibleToolNames: visibleToolNames(pipeline.tools),
+ Mode: ModeSelected,
+ Outcome: pipeline.outcome,
+ StageCounts: pipeline.stageCounts,
+ Reasons: pipeline.traces,
+ Version: version,
+ CreatedAt: time.Now().UnixNano(),
+ IdentityHash: identityHash,
+ ProgressiveHash: c.progressiveHash,
+ ConfigHash: c.configHash,
+ CatalogVersion: c.catalogVersion(candidates,
sc.CatalogVersion),
+ Expanded: pipeline.expanded,
+ }
+ plan.toolSet = toolNameSet(plan.ToolNames)
+ return plan
+}
+
+func (c *CompositeSelector) storeSelectionPlan(key PlanKey, plan
*SelectionPlan, sc SelectionContext) (*SelectionPlan, error) {
+ if err := c.store.Set(key, plan, sc); err != nil {
+ return nil, err
+ }
+ if stored, ok := c.store.Get(key); ok {
Review Comment:
这里先 `Set(plan)`，再 `Get(key)`。同一个 session 下如果有并发请求，并且请求 claims
不同，当前请求可能读到另一个请求刚写入的 plan，再继续用它签发 authorization receipt。建议让
`SessionPlanStore.Set` 在同一把锁内返回本次提交的 committed plan，或者把 recompute plan 和
receipt issuance 收敛成一个原子操作。
##########
pkg/adapter/mcpserver/registrycenter.go:
##########
@@ -167,12 +167,12 @@ func (a *Adapter) Apply() error {
}
// 1) apply tools dynamically to registry for filter
usage
- if dc := mcpserver.GetOrInitDynamicConsumer(); dc !=
nil {
+ if dc, err :=
mcpserver.DynamicConsumerForSingleRuntime(); err == nil {
if err :=
dc.ApplyMcpServerConfigByServer(serverId, cfg); err != nil {
logger.Errorf("[dubbo-go-pixiu] mcp
adapter apply server %s config error: %v", serverId, err)
}
} else {
- logger.Infof("[dubbo-go-pixiu] mcp adapter
update received from server %s: tools=%d", serverId, len(cfg.Tools))
+ logger.Infof("[dubbo-go-pixiu] mcp adapter
update received from server %s without a unique dynamic consumer: %v",
serverId, err)
Review Comment:
当存在多个 MCP runtime 时这里会跳过 dynamic catalog 更新，但下面仍会继续按 `cfg.Tools` 注册 cluster
endpoints。这样 multi-filter 配置下 catalog/authorization plan 没更新，cluster endpoint
却更新了，两个状态会分叉。建议 ambiguity 时整个 update fail/skip，或者把 registry update 明确绑定到对应
runtime。
##########
pkg/filter/mcp/mcpserver/dynamic.go:
##########
@@ -224,70 +280,108 @@ func (d *DynamicConsumer) removeServerConfig(serverId
string) error {
func (d *DynamicConsumer) calculateCurrentMergedTools() []model.ToolConfig {
var allTools []model.ToolConfig
- // Simply accumulate tools from all servers
- for _, config := range d.serverConfigs {
- allTools = append(allTools, config.Tools...)
+ serverIDs := make([]string, 0, len(d.serverConfigs))
+ for serverID := range d.serverConfigs {
+ serverIDs = append(serverIDs, serverID)
+ }
+ sort.Strings(serverIDs)
+
+ for _, serverID := range serverIDs {
+ allTools = append(allTools, d.serverConfigs[serverID].Tools...)
}
return allTools
}
-// notifyToolsListChanged sends notifications/tools/list_changed to all
connected clients
+// notifyToolsListChanged marks notifications/tools/list_changed only for
+// sessions whose visible tool set changed. Offline sessions keep one bounded
+// pending version for reconnect.
func (d *DynamicConsumer) notifyToolsListChanged() {
if d.sessionManager == nil {
logger.Debugf("[dubbo-go-pixiu] mcp server session manager not
available, skip tools list_changed notification")
return
}
- // Get all active sessions
- sessionIDs := d.sessionManager.AllSessionIDs()
- if len(sessionIDs) == 0 {
+ sessions := d.sessionManager.SnapshotSessions()
+ if len(sessions) == 0 {
logger.Debugf("[dubbo-go-pixiu] mcp server no active sessions,
skip tools list_changed notification")
return
}
- // Send notification to each session
- successCount := 0
- for _, sessionID := range sessionIDs {
- if err := d.sendToolsListChangedNotification(sessionID); err !=
nil {
- logger.Warnf("[dubbo-go-pixiu] mcp server failed to
send tools list_changed to session %s: %v", sessionID, err)
- } else {
- successCount++
+ markedCount := 0
+ flushedCount := 0
+ changed := d.sessionsWithChangedVisibleSet()
+ for _, session := range sessions {
+ if changed != nil {
+ if _, ok := changed[session.ID]; !ok {
+ continue
+ }
+ }
+ marked, flushed, err :=
markToolsListChangedAndFlush(d.sseHandler, session)
+ if !marked {
+ continue
+ }
+ markedCount++
+ if err != nil {
+ logger.Warnf("[dubbo-go-pixiu] mcp server failed to
send tools list_changed: %v", err)
+ } else if flushed {
+ flushedCount++
}
}
- logger.Infof("[dubbo-go-pixiu] mcp server sent tools/list_changed
notification to %d/%d sessions", successCount, len(sessionIDs))
-}
-
-// sendToolsListChangedNotification sends notification to a specific session
-func (d *DynamicConsumer) sendToolsListChangedNotification(sessionID string)
error {
- session, exists := d.sessionManager.Session(sessionID)
- if !exists {
- return fmt.Errorf("session not found")
+ if markedCount == 0 {
+ logger.Debugf("[dubbo-go-pixiu] mcp server no active sessions
to mark for tools list_changed notification")
+ return
}
+ logger.Infof("[dubbo-go-pixiu] mcp server marked tools/list_changed
notification for %d sessions, flushed %d online sessions", markedCount,
flushedCount)
+}
- if session.PipeWriter == nil {
- return fmt.Errorf("SSE pipe not established")
+func (d *DynamicConsumer) sessionsWithChangedVisibleSet() map[string]struct{} {
+ d.mu.RLock()
+ governance := d.governance
+ selector := d.selector
+ plans := d.plans
+ d.mu.RUnlock()
+ if !governance || selector == nil || plans == nil {
+ return nil
}
-
- // Build tools/list_changed notification (no params needed)
- notification := map[string]any{
- "jsonrpc": "2.0",
- "method": "notifications/tools/list_changed",
+ refresher, ok := selector.(router.PlanRefreshSelector)
+ if !ok {
+ return nil
}
-
- messageJSON, err := json.Marshal(notification)
- if err != nil {
- return fmt.Errorf("failed to marshal notification: %w", err)
+ snapshot := d.registry.toolCatalogSnapshotUnsafe()
+ changed := make(map[string]struct{})
+ for _, item := range plans.SessionPlanContexts() {
+ item.Context.CatalogVersion = snapshot.Version
+ plan, committed, err :=
refresher.RefreshPlan(context.Background(), item, snapshot.orderedToolsUnsafe())
Review Comment:
这个函数在动态 catalog 更新后刷新 session plan 并决定 notification，但 catalog publish、plan
refresh、session 状态和 SSE 通知分别散在多个模块里。这里是授权治理路径，建议把 catalog 发布、受影响 session plan
刷新和 notification decision 收敛成一个更清晰的 publication 流程，减少后续遗漏一致性边界的风险。
##########
pkg/adapter/mcpserver/registrycenter.go:
##########
@@ -167,12 +167,12 @@ func (a *Adapter) Apply() error {
}
// 1) apply tools dynamically to registry for filter
usage
- if dc := mcpserver.GetOrInitDynamicConsumer(); dc !=
nil {
+ if dc, err :=
mcpserver.DynamicConsumerForSingleRuntime(); err == nil {
Review Comment:
这里现在只在 `cfg` 非 nil 时才进入 dynamic consumer，但 Nacos adapter 已经用
`onChange(serverId, nil)` 表达 server
配置删除，`DynamicConsumer.ApplyMcpServerConfigByServer(serverId, nil)`
也已经实现删除。当前早退会把删除事件吞掉，registry catalog 里的旧 tools 会一直保留；reconcile 发现 server
消失时也需要走同一个 remove path。
##########
pkg/adapter/mcpserver/registrycenter.go:
##########
@@ -167,12 +167,12 @@ func (a *Adapter) Apply() error {
}
// 1) apply tools dynamically to registry for filter
usage
- if dc := mcpserver.GetOrInitDynamicConsumer(); dc !=
nil {
+ if dc, err :=
mcpserver.DynamicConsumerForSingleRuntime(); err == nil {
if err :=
dc.ApplyMcpServerConfigByServer(serverId, cfg); err != nil {
logger.Errorf("[dubbo-go-pixiu] mcp
adapter apply server %s config error: %v", serverId, err)
}
} else {
- logger.Infof("[dubbo-go-pixiu] mcp adapter
update received from server %s: tools=%d", serverId, len(cfg.Tools))
+ logger.Infof("[dubbo-go-pixiu] mcp adapter
update received from server %s without a unique dynamic consumer: %v",
serverId, err)
}
// 2) register endpoint for each tool using BackendURL
(host:port) into cluster named by tool.Name
Review Comment:
下面这段只对当前 `cfg.Tools` 调 `SetEndpoint`，没有删除上一次该 server/tool 发布过的
endpoint。`endpointID` 是 `host:port`，所以 BackendURL 从 A 改到 B 时 B 会追加进 cluster，A
不会被删；工具或 server 下线也会留下旧 endpoint。建议按 server/tool 记录已发布 endpoint IDs，并在变更/删除时调用
`DeleteEndpoint` 做 reconcile。
##########
pkg/filter/mcp/mcpserver/router/workflow.go:
##########
@@ -0,0 +1,146 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package router
+
+import (
+ "fmt"
+ "strings"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/model"
+)
+
+// compiledWorkflow is a WorkflowConfig with its When clause precompiled and
its
+// tool membership hoisted into a set.
+type compiledWorkflow struct {
+ name string
+ when *matcher
+ tools map[string]struct{}
+ hasWhen bool
+}
+
+// WorkflowSelector narrows the candidate set to a matched workflow bundle.
+//
+// The first workflow whose When clause matches the context wins; only tools in
+// that bundle (and present in the candidate set) are kept. When no workflow
+// matches, the candidates pass through unchanged. Workflows with an empty When
+// clause are treated as bundles addressable by name (e.g. fallback bundles)
and
+// are never auto-matched.
+type WorkflowSelector struct {
+ workflows []compiledWorkflow
+ byName map[string]compiledWorkflow
+}
+
+type workflowResult struct {
+ tools []model.ToolConfig
+ traces []DecisionTrace
+ matched bool
+ rule string
+}
+
+// NewWorkflowSelector compiles the workflow definitions.
+func NewWorkflowSelector(cfgs []model.WorkflowConfig) (*WorkflowSelector,
error) {
+ ws := &WorkflowSelector{
+ workflows: make([]compiledWorkflow, 0, len(cfgs)),
+ byName: make(map[string]compiledWorkflow, len(cfgs)),
+ }
+ for i, c := range cfgs {
+ name := strings.TrimSpace(c.Name)
+ if name == "" {
+ return nil, fmt.Errorf("workflow at index %d: name is
required", i)
+ }
+ if _, exists := ws.byName[name]; exists {
+ return nil, fmt.Errorf("workflow %q is defined more
than once", name)
+ }
+ m, err := newMatcher(c.When)
+ if err != nil {
+ return nil, fmt.Errorf("workflow %q: %w", name, err)
+ }
+ cw := compiledWorkflow{
+ name: name,
+ when: m,
+ tools: toSet(c.Tools),
+ hasWhen: !m.alwaysMatches(),
Review Comment:
workflow tools 直接通过 `toSet(c.Tools)` 编译，`toSet` 目前不会
trim，也不会拒绝空字符串或纯空白字符串。配置里如果出现空工具名，可能构造出逻辑无效的 bundle，并影响 progressive initial
bundle 校验。建议在 workflow 配置编译阶段 trim tool names，并对空值 fail fast。
##########
pkg/filter/mcp/mcpserver/router/decision_log.go:
##########
@@ -0,0 +1,164 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package router
+
+import (
+ cryptorand "crypto/rand"
+ "encoding/binary"
+ "encoding/json"
+)
+
+import (
+ "github.com/apache/dubbo-go-pixiu/pkg/logger"
+)
+
+// maxDeniedSamples caps dropped tool names in a decision log.
+const maxDeniedSamples = 10
+
+// DecisionLogger emits a structured, PII-safe record for each selection. It
+// never logs session IDs, identity claims, prompt text, tool arguments, or
+// request text details. Only counts, the mode, and per-stage drop tallies are
+// emitted by default. Denied tool samples require explicit decision detail
logging.
+type DecisionLogger struct {
+ sampleRate float64
+ detailLogging bool
+}
+
+// NewDecisionLogger builds a logger. A sampleRate of 0 disables decision logs;
+// values in (0,1] sample probabilistically/all. Detailed denied samples are
+// emitted only when decision detail logging is explicitly active.
+func NewDecisionLogger(sampleRate float64, detailLogging bool) *DecisionLogger
{
+ return &DecisionLogger{sampleRate: sampleRate, detailLogging:
detailLogging}
+}
+
+// decisionRecord is the JSON shape emitted to the log.
+type decisionRecord struct {
+ Event string `json:"event"`
+ Method string `json:"method"`
+ PlanVersion string `json:"plan_version"`
+ Candidates int `json:"candidates"`
+ Selected int `json:"selected"`
+ Mode string `json:"mode"`
+ Stages map[string]int `json:"stages,omitempty"`
+ DeniedSamples []string `json:"denied_samples,omitempty"`
+}
+
+// Log emits a decision record for the given selection, subject to sampling.
+func (d *DecisionLogger) Log(sc SelectionContext, plan *SelectionPlan,
candidates int) {
+ if d == nil || plan == nil {
+ return
+ }
+ if !d.shouldSample() {
+ return
+ }
+
+ rec := d.record(sc, plan, candidates)
+
+ data, err := json.Marshal(rec)
+ if err != nil {
+ return
+ }
+ logger.Infof("[dubbo-go-pixiu] %s", string(data))
+}
+
+func (d *DecisionLogger) record(sc SelectionContext, plan *SelectionPlan,
candidates int) decisionRecord {
+ rec := decisionRecord{
+ Event: "mcp_router_decision",
+ Method: sc.Method,
+ PlanVersion: plan.Version,
+ Candidates: candidates,
Review Comment:
`plan.Version` 间接包含 identity fingerprint，而当前 fingerprint 使用无密钥 FNV。虽然不是明文
claims，但低熵 claims 仍可能被关联或字典推断。建议日志不要输出包含 identity-derived hash 的
version；如果确实需要排查关联，可以使用服务端密钥 HMAC 后截断，或只记录 catalog/config version。
##########
pkg/filter/mcp/mcpserver/router/session_plan.go:
##########
@@ -0,0 +1,576 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package router
+
+import (
+ "errors"
+ "sync"
+)
+
+// PlanKey isolates cached plan state by router instance and MCP session.
+type PlanKey struct {
+ RouterID string
+ SessionID string
+}
+
+// ReceiptVersionRequest describes the live authorization inputs a stored plan
+// must still match before a receipt can be issued without recomputing
selection.
+type ReceiptVersionRequest struct {
+ Key PlanKey
+ Requested string
+ ExpectedVersion string
+ IdentityHash string
+ ConfigHash string
+ CatalogVersion string
+ ProgressiveHash string
+ RouterID string
+}
+
+// NewPlanKey constructs a key for one router instance/session pair.
+func NewPlanKey(routerID, sessionID string) PlanKey {
+ return PlanKey{RouterID: routerID, SessionID: sessionID}
+}
+
+func (k PlanKey) valid() bool {
+ return k.RouterID != "" && k.SessionID != ""
+}
+
+// sessionEntry bundles the immutable enforcement plan with the progressive
+// state that is safe to retain for the session lifetime.
+type sessionEntry struct {
+ plan *SelectionPlan
+ context SelectionContext
+ callCount int64
+ expanded bool
+ identityHash string
+ progressiveHash string
+ configHash string
+ catalogVersion string
+ generation uint64
+ nextReceiptID uint64
+ activeReceipts map[uint64]*receiptState
+}
+
+// SessionPlanStoreOptions configures a SessionPlanStore.
+type SessionPlanStoreOptions struct {
+ MaxEntries int
+}
+
+// SessionPlanStore is an in-process, concurrency-safe store of selection plans
+// owned by one MCP filter instance. Transport session removal deletes entries
+// immediately; plans do not expire independently from their sessions.
+type SessionPlanStore struct {
+ mu sync.RWMutex
+ entries map[PlanKey]*sessionEntry
+ activeSessions map[string]uint64
+ max int
+}
+
+// DefaultPlanMaxEntries caps the number of active session plans.
+const DefaultPlanMaxEntries = 10000
+
+// ErrPlanStoreFull is returned when a filter has reached its active session
+// capacity. The caller must reject new sessions instead of evicting live
plans.
+var ErrPlanStoreFull = errors.New("mcp router session capacity reached")
+
+// NewSessionPlanStore creates a store with production defaults.
+func NewSessionPlanStore() *SessionPlanStore {
+ return NewSessionPlanStoreWithOptions(SessionPlanStoreOptions{})
+}
+
+// NewSessionPlanStoreWithMaxEntries creates a store using the supplied
+// capacity. A non-positive value means the production default.
+func NewSessionPlanStoreWithMaxEntries(maxEntries int) *SessionPlanStore {
+ return
NewSessionPlanStoreWithOptions(SessionPlanStoreOptions{MaxEntries: maxEntries})
+}
+
+// NewSessionPlanStoreWithOptions creates a store with explicit options.
+func NewSessionPlanStoreWithOptions(opts SessionPlanStoreOptions)
*SessionPlanStore {
+ maxEntries := opts.MaxEntries
+ if maxEntries <= 0 {
+ maxEntries = DefaultPlanMaxEntries
+ }
+
+ s := &SessionPlanStore{
+ entries: make(map[PlanKey]*sessionEntry),
+ activeSessions: make(map[string]uint64),
+ max: maxEntries,
+ }
+ registerPlanStoreMetric(s)
+ return s
+}
+
+// ActivateSession records the transport session generation that is allowed to
+// own plans for this session ID. The map is bounded by active sessions and is
+// cleared by DeleteSession when the transport session ends.
+func (s *SessionPlanStore) ActivateSession(sessionID string, generation
uint64) error {
+ if sessionID == "" || generation == 0 {
+ return ErrSessionPlanNotActive
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ s.activeSessions[sessionID] = generation
+ return nil
+}
+
+// Get returns an immutable copy of the plan for a session, or (nil, false) if
+// absent. Callers cannot mutate store-owned slices or maps through the result.
+func (s *SessionPlanStore) Get(key PlanKey) (*SelectionPlan, bool) {
+ if !key.valid() {
+ return nil, false
+ }
+ s.mu.RLock()
+ defer s.mu.RUnlock()
+ e, ok := s.entries[key]
+ if !ok || e.plan == nil {
+ return nil, false
+ }
+ return clonePlan(e.plan, true), true
+}
+
+// Set stores or replaces the plan for its session. Successful-call state is
+// preserved only while the verified identity and progressive config hashes are
+// unchanged; identity/config changes reset progressive disclosure.
+func (s *SessionPlanStore) Set(key PlanKey, plan *SelectionPlan, sc
SelectionContext) error {
+ stored := clonePlan(plan, false)
+ if err := validatePlanWrite(key, stored, sc); err != nil {
+ return err
+ }
+
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if !s.sessionActiveLocked(sc) {
+ return ErrSessionPlanNotActive
+ }
+ _, err := s.setLocked(key, stored, sc, true)
+ return err
+}
+
+func (s *SessionPlanStore) setLocked(key PlanKey, stored *SelectionPlan, sc
SelectionContext, allowCreate bool) (*SelectionPlan, error) {
+ e, ok := s.entries[key]
+ if !ok {
+ if !allowCreate {
+ return nil, ErrSessionPlanStale
+ }
+ if len(s.entries) >= s.max {
+ return nil, ErrPlanStoreFull
+ }
+ e = &sessionEntry{}
+ s.entries[key] = e
+ } else if e.identityHash != stored.IdentityHash || e.progressiveHash !=
stored.ProgressiveHash {
+ e.callCount = 0
+ e.expanded = false
+ }
+ e.activeReceipts = nil
+
+ e.generation++
+ stored.Expanded = e.expanded || stored.Expanded
+ stored.Generation = e.generation
+ stored.Reasons = nil
+ stored.SessionID = key.SessionID
+ e.plan = stored
+ e.context = cloneSelectionContext(sc)
+ e.identityHash = stored.IdentityHash
+ e.progressiveHash = stored.ProgressiveHash
+ e.configHash = stored.ConfigHash
+ e.catalogVersion = stored.CatalogVersion
+ s.publishActiveLocked()
+ return clonePlan(e.plan, true), nil
+}
+
+// SetIfCurrent conditionally commits a recomputed plan. It is used by dynamic
+// catalog recomputation so a task that started from an older plan cannot
+// recreate a deleted session plan or overwrite a newer tools/list result.
+func (s *SessionPlanStore) SetIfCurrent(base SessionPlanContext, plan
*SelectionPlan) (*SelectionPlan, bool, error) {
+ stored := clonePlan(plan, false)
+ if err := validatePlanWrite(base.Key, stored, base.Context); err != nil
{
+ return nil, false, err
+ }
+ if base.PlanGeneration == 0 {
+ return nil, false, ErrSessionPlanStale
+ }
+
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if !s.sessionActiveLocked(base.Context) {
+ return nil, false, nil
+ }
+ e, ok := s.entries[base.Key]
+ if !ok || e.plan == nil || e.generation != base.PlanGeneration ||
+ e.context.SessionGeneration != base.Context.SessionGeneration {
+ return nil, false, nil
+ }
+ committed, err := s.setLocked(base.Key, stored, base.Context, false)
+ if err != nil {
+ return nil, false, err
+ }
+ return committed, true, nil
+}
+
+// SessionPlanContexts returns cloned plan/context pairs for management-plane
+// recomputation without touching transport session activity.
+func (s *SessionPlanStore) SessionPlanContexts() []SessionPlanContext {
+ s.mu.RLock()
+ defer s.mu.RUnlock()
+ out := make([]SessionPlanContext, 0, len(s.entries))
+ for key, entry := range s.entries {
+ if entry.plan == nil {
+ continue
+ }
+ out = append(out, SessionPlanContext{
+ Key: key,
+ Plan: clonePlan(entry.plan, false),
+ Context: cloneSelectionContext(entry.context),
+ PlanGeneration: entry.generation,
+ SessionGeneration: entry.context.SessionGeneration,
+ })
+ }
+ return out
+}
+
+// SessionPlanContext is a cloned plan and the normalized selection attributes
+// used to build it.
+type SessionPlanContext struct {
+ Key PlanKey
+ Plan *SelectionPlan
+ Context SelectionContext
+ PlanGeneration uint64
+ SessionGeneration uint64
+}
+
+// Delete removes one router instance's plan and progressive state.
+func (s *SessionPlanStore) Delete(key PlanKey) {
+ s.delete(key, "explicit")
+}
+
+func (s *SessionPlanStore) delete(key PlanKey, reason string) {
+ if !key.valid() {
+ return
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ if _, ok := s.entries[key]; !ok {
+ return
+ }
+ delete(s.entries, key)
+ recordPlanEvicted(reason)
+ s.publishActiveLocked()
+}
+
+// DeleteSession removes all router-instance state for a transport session. It
+// is the method registered with SessionManager teardown callbacks.
+func (s *SessionPlanStore) DeleteSession(sessionID string) {
+ if sessionID == "" {
+ return
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+ removed := 0
+ for key := range s.entries {
+ if key.SessionID == sessionID {
+ delete(s.entries, key)
+ removed++
+ }
+ }
+ delete(s.activeSessions, sessionID)
+ for i := 0; i < removed; i++ {
+ recordPlanEvicted("session_end")
+ }
+ if removed > 0 {
+ s.publishActiveLocked()
+ }
+}
+
+// IssueReceipt returns a non-replayable authorization receipt for the current
+// stored plan generation.
+func (s *SessionPlanStore) IssueReceipt(key PlanKey, requested string, plan
*SelectionPlan, routerID string) (AuthorizationReceipt, error) {
+ if !key.valid() || requested == "" || plan == nil {
+ return AuthorizationReceipt{}, ErrToolNotAuthorized
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
+ e, ok := s.entries[key]
+ if !ok || e.plan == nil ||
+ e.plan.Version != plan.Version ||
+ e.identityHash != plan.IdentityHash ||
+ e.configHash != plan.ConfigHash ||
+ e.catalogVersion != plan.CatalogVersion ||
+ e.progressiveHash != plan.ProgressiveHash ||
+ !e.plan.Contains(requested) {
+ return AuthorizationReceipt{}, ErrToolNotAuthorized
+ }
+ return e.issueReceipt(key, requested, routerID), nil
+}
+
+// IssueReceiptForVersion signs a tools/call authorization without cloning the
+// stored plan. It returns stale=true when the session has a plan, but its
+// authorization inputs no longer match the caller's live inputs.
+func (s *SessionPlanStore) IssueReceiptForVersion(req ReceiptVersionRequest)
(AuthorizationReceipt, bool, error) {
+ if !req.Key.valid() || req.Requested == "" || req.ExpectedVersion == ""
{
+ return AuthorizationReceipt{}, false, ErrToolNotAuthorized
+ }
+ s.mu.Lock()
+ defer s.mu.Unlock()
+
+ e, ok := s.entries[req.Key]
+ if !ok || e.plan == nil {
+ return AuthorizationReceipt{}, false, ErrToolNotAuthorized
+ }
+ if e.plan.Version != req.ExpectedVersion ||
+ e.identityHash != req.IdentityHash ||
+ e.configHash != req.ConfigHash ||
+ e.catalogVersion != req.CatalogVersion ||
+ e.progressiveHash != req.ProgressiveHash {
+ return AuthorizationReceipt{}, true, nil
+ }
+ if !e.plan.Contains(req.Requested) {
+ return AuthorizationReceipt{}, false, ErrToolNotAuthorized
+ }
+ return e.issueReceipt(req.Key, req.Requested, req.RouterID), false, nil
+}
+
+func (e *sessionEntry) issueReceipt(key PlanKey, requested, routerID string)
AuthorizationReceipt {
+ e.nextReceiptID++
+ if e.activeReceipts == nil {
+ e.activeReceipts = make(map[uint64]*receiptState)
+ }
+ state := &receiptState{}
+ e.activeReceipts[e.nextReceiptID] = state
+ return AuthorizationReceipt{
Review Comment:
`issueReceipt` 会登记 active receipt，但当前只有 `RecordCallSuccess` 会删除它。backend
error、request cancellation、构造 backend request 失败等非成功路径会让 receipt 留在 active map
里，长 session 或大量失败调用下会累积。建议增加统一 finalize 路径：成功时消费并推进 progressive，失败时只释放 receipt。
##########
pkg/filter/mcp/mcpserver/handlers.go:
##########
@@ -109,58 +114,109 @@ func (f *MCPServerFilter) handleInitialize(ctx
*MCPContext, req mcp.JSONRPCReque
// Per MCP spec: assign a session ID at initialization time for
Streamable HTTP transport
// This enables clients to use the session for SSE-based responses
- sessionIDHeader := ctx.SessionID()
- session, _ := f.sessionManager.EnsureSession(sessionIDHeader)
+ session, err := f.sessionManager.CreateSession()
+ if err != nil {
+ logger.Errorf("[dubbo-go-pixiu] mcp server failed to create
session: %v", err)
+ return f.errorHandler.SendInternalError(ctx, req.ID, "failed to
create session")
+ }
+ if f.governanceEnabled && f.plans != nil {
+ if err := f.plans.ActivateSession(session.ID,
session.Generation); err != nil {
Review Comment:
这里无条件 `CreateSession()`。如果 router disabled 时旧行为允许客户端携带已有 `Mcp-Session-Id`
并复用 session，这里会改变非治理路径的 session lifecycle。建议明确 router enabled/disabled 两种情况下
initialize 携带 session id 的语义，并补对应测试。
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
