feat(mcpx): sampling.answerLocally — answer upstream sampling from the configured provider layer (#1925 slice 3)

pkgs/mcpx/docs/configuration.md

@@ -993,6 +993,12 @@ skipped launcher phases).
 | `retention.maxTools` | int | `24` | daemon | no | `--retention-max-tools` | `MCPX_RETENTION_MAX_TOOLS` | maximum active tools in reactive working set |
 | `retention.stickyWindow` | duration | `2h` | daemon | no | `--retention-sticky-window` | `MCPX_RETENTION_STICKY_WINDOW` | duration explicitly requested tools remain retained |
 
+### sampling
+
+| setting | kind | default | scope | hot | flag | variable | governs |
+| --- | --- | --- | --- | --- | --- | --- | --- |
+| `sampling.answerLocally` | bool | `false` | daemon | no | `--sampling-answer-locally` | `MCPX_SAMPLING_ANSWER_LOCALLY` | answer upstream sampling requests with mcpx's configured model provider instead of relaying them to the caller |
+
 ### script
 
 | setting | kind | default | scope | hot | flag | variable | governs |

pkgs/mcpx/internal/daemon/consumer.go

@@ -58,6 +58,10 @@ type ConsumerPolicy struct {
 	ModelDefault        string
 	ModelBudget         int
 	ModelTimeout        time.Duration
+
+	// SamplingAnswerLocally is sampling.answerLocally: answer upstream
+	// sampling requests from the configured model layer (#1925 slice 3).
+	SamplingAnswerLocally bool
 }
 
 // consumerState is everything the consumer routes need that the daemon did
@@ -163,6 +167,7 @@ func consumerPolicyFrom(set *settings.Set) ConsumerPolicy {
 	p.ModelDefault = set.String("model.default")
 	p.ModelBudget = set.Int("model.budget")
 	p.ModelTimeout = set.Duration("model.timeout")
+	p.SamplingAnswerLocally = set.Bool("sampling.answerLocally")
 	return p
 }
 

pkgs/mcpx/internal/daemon/hooks.go

@@ -180,6 +180,15 @@ func mustJSON(v any) json.RawMessage {
 // become a stored question that whatever drives mcpx can answer -- which is
 // exactly what "passthrough" means.
 func (r *Registry) answerServer(ctx context.Context, server, key, method string, params json.RawMessage) (any, error) {
+	if method == "sampling/createMessage" && r.consumer.SamplingAnswerLocally {
+		// sampling.answerLocally (#1925 slice 3): mcpx answers from its
+		// own configured provider layer instead of relaying the request
+		// to a caller. This precedes the broker check on purpose: a
+		// local answer needs no broker, and a daemon with the setting
+		// on but nothing configured hears the clean failure from
+		// sampleLocally rather than a missing-broker error.
+		return r.sampleLocally(ctx, server, key, params)
+	}
 	if r.broker == nil {
 		// Nothing can answer, so say so rather than hang. Cancel is honest:
 		// nobody was asked, so nobody chose.

pkgs/mcpx/internal/daemon/sampling_local.go

@@ -0,0 +1,203 @@
+package daemon
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+
+	"github.com/developing-today/code/pkgs/mcpx/internal/elicit"
+	"github.com/developing-today/code/pkgs/mcpx/internal/mcpclient"
+	"github.com/developing-today/code/pkgs/mcpx/internal/model"
+)
+
+// sampling.answerLocally (#1925, slice 3): answer an upstream server's
+// sampling/createMessage request from mcpx's own configured provider
+// layer instead of relaying it through the broker to the caller.
+//
+// The relay stays the default (the setting is off) and is untouched when
+// it is off: answerServer only diverts here when the setting is on. When
+// it is on, the request is answered by internal/model Resolve over the
+// daemon's model.* policy -- provider or profile -- with two deliberate
+// differences from the consumer generation path:
+//
+//   - Sampler is nil. Resolving the sampling backend here would relay
+//     the request to the caller, which is the path this setting exists
+//     to replace; a configuration that names only sampling therefore
+//     resolves to unavailable, honestly.
+//   - There is no broker fallback. Enabled with no provider configured,
+//     or with a provider that cannot answer, the request fails cleanly
+//     and deterministically (an internal error naming the cause) rather
+//     than hanging on a broker question nobody will answer or silently
+//     succeeding with an invented answer.
+//
+// Bounds: the request's own maxTokens, capped by model.budget, and
+// model.timeout as the deadline -- the same bounds the model layer's
+// other callers get (modelMaxTokens, Server.modelLayer).
+func (r *Registry) sampleLocally(ctx context.Context, server, key string, params json.RawMessage) (any, error) {
+	pol := r.consumer
+
+	// The request is checked against the era the upstream negotiated
+	// before anything is spent on it, exactly as the broker path checks
+	// it: a malformed sampling request is the server's mistake (-32602),
+	// not a generation to attempt.
+	era, known := mcpclient.SamplingEraFrom(ctx)
+	arrayOK := !known || era.Negotiated == "" || era.Negotiated >= "2025-11-25"
+	if err := elicit.ValidateSampleRequest(params, arrayOK); err != nil {
+		return nil, &elicit.SamplingInvalidParams{Err: err}
+	}
+
+	if !modelLayerConfigured(pol) {
+		return nil, mcpclient.InternalError("sampling.answerLocally is set but no model provider is configured " +
+			"(set model.provider, model.default or model.profiles); the request was not relayed")
+	}
+	prov, err := model.Resolve(model.Config{
+		Provider:             pol.ModelProvider,
+		BaseURL:              pol.ModelBaseURL,
+		APIKeyEnv:            pol.ModelAPIKeyEnv,
+		ProviderSettingsJSON: pol.ModelProviders,
+		ProfilesJSON:         pol.ModelProfiles,
+		DefaultModel:         pol.ModelDefault,
+		Timeout:              pol.ModelTimeout,
+		// Sampler deliberately nil: see the package comment above.
+	})
+	if err != nil {
+		return nil, mcpclient.InternalError("sampling.answerLocally: " + err.Error())
+	}
+	if prov == nil {
+		return nil, mcpclient.InternalError("sampling.answerLocally is set but model.provider resolved to no provider; " +
+			"the request was not relayed")
+	}
+
+	req, err := samplingLocalRequest(params, pol)
+	if err != nil {
+		return nil, &elicit.SamplingInvalidParams{Err: err}
+	}
+
+	if pol.ModelTimeout > 0 {
+		var cancel context.CancelFunc
+		ctx, cancel = context.WithTimeout(ctx, pol.ModelTimeout)
+		defer cancel()
+	}
+	resp, err := prov.Generate(ctx, req)
+	if err != nil {
+		return nil, mcpclient.InternalError("sampling.answerLocally: provider " + prov.Name() + " did not answer: " + err.Error())
+	}
+	if strings.TrimSpace(resp.Text) == "" {
+		return nil, mcpclient.InternalError("sampling.answerLocally: provider " + prov.Name() + " answered with no text")
+	}
+
+	modelName := resp.Model
+	if modelName == "" {
+		modelName = model.ModelID(req.Model)
+	}
+	if modelName == "" {
+		modelName = model.ModelID(pol.ModelDefault)
+	}
+	result := map[string]any{
+		"role":       "assistant",
+		"model":      modelName,
+		"content":    map[string]any{"type": "text", "text": resp.Text},
+		"stopReason": "endTurn",
+	}
+	// The answer is held to the same shape rules as a brokered one: a
+	// local answer that is not a CreateMessageResult is refused upstream
+	// as -32602 rather than relayed in a shape nobody sent.
+	raw, err := json.Marshal(result)
+	if err != nil {
+		return nil, err
+	}
+	if err := elicit.ValidateSampleResult(raw, arrayOK, false); err != nil {
+		return nil, &elicit.SamplingInvalidParams{Err: err}
+	}
+	return result, nil
+}
+
+// samplingLocalRequest converts a sampling CreateMessageRequest into a
+// model.Request for the provider layer.
+//
+// The layer is text-only (model.Message carries a string), so the
+// conversion is honest about what it cannot carry: a request with tools
+// (whose answer would be tool_use blocks a text generation cannot
+// produce) or with non-text content blocks fails as invalid params for
+// this path rather than being silently flattened. Model preference
+// hints are honoured the way the specification allows them to be used:
+// the first hint names the model when one is given; otherwise
+// model.default supplies it through Resolve/Generate as usual.
+func samplingLocalRequest(params json.RawMessage, pol ConsumerPolicy) (model.Request, error) {
+	var p struct {
+		Messages []struct {
+			Role    string          `json:"role"`
+			Content json.RawMessage `json:"content"`
+		} `json:"messages"`
+		SystemPrompt     string `json:"systemPrompt"`
+		MaxTokens        int    `json:"maxTokens"`
+		ModelPreferences struct {
+			Hints []struct {
+				Name string `json:"name"`
+			} `json:"hints"`
+		} `json:"modelPreferences"`
+		Tools json.RawMessage `json:"tools"`
+	}
+	if err := json.Unmarshal(params, &p); err != nil {
+		return model.Request{}, fmt.Errorf("a CreateMessageRequest must be an object: %w", err)
+	}
+	if len(p.Tools) > 0 && string(p.Tools) != "null" && string(p.Tools) != "[]" {
+		return model.Request{}, errors.New("the request carries tools, which a local text generation cannot answer")
+	}
+
+	req := model.Request{MaxTokens: modelMaxTokens(p.MaxTokens, pol.ModelBudget)}
+	if p.SystemPrompt != "" {
+		req.Messages = append(req.Messages, model.Message{Role: "system", Content: p.SystemPrompt})
+	}
+	for i, m := range p.Messages {
+		text, err := samplingTextContent(m.Content)
+		if err != nil {
+			return model.Request{}, fmt.Errorf("message %d: %w", i, err)
+		}
+		role := m.Role
+		if role == "" {
+			role = "user"
+		}
+		req.Messages = append(req.Messages, model.Message{Role: role, Content: text})
+	}
+	if len(p.ModelPreferences.Hints) > 0 && p.ModelPreferences.Hints[0].Name != "" {
+		req.Model = p.ModelPreferences.Hints[0].Name
+	}
+	return req, nil
+}
+
+// samplingTextContent extracts the text of one sampling message's
+// content: a single block or a list of blocks (both eras' shapes are
+// accepted here; the era's own rule was already enforced by
+// ValidateSampleRequest). Every block must be a text block.
+func samplingTextContent(raw json.RawMessage) (string, error) {
+	b := strings.TrimSpace(string(raw))
+	if b == "" || b == "null" {
+		return "", errors.New("content is missing")
+	}
+	var blocks []json.RawMessage
+	if b[0] == '[' {
+		if err := json.Unmarshal(raw, &blocks); err != nil {
+			return "", fmt.Errorf("content is not a list of content blocks: %w", err)
+		}
+	} else {
+		blocks = []json.RawMessage{raw}
+	}
+	var texts []string
+	for _, blk := range blocks {
+		var block struct {
+			Type string `json:"type"`
+			Text string `json:"text"`
+		}
+		if err := json.Unmarshal(blk, &block); err != nil {
+			return "", fmt.Errorf("content is not a content block: %w", err)
+		}
+		if block.Type != "text" {
+			return "", fmt.Errorf("a %q content block cannot be answered by a local text generation", block.Type)
+		}
+		texts = append(texts, block.Text)
+	}
+	return strings.Join(texts, "\n"), nil
+}

pkgs/mcpx/internal/daemon/sampling_local_test.go

@@ -0,0 +1,133 @@
+package daemon
+
+import (
+	"context"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"testing"
+	"time"
+)
+
+// sampling.answerLocally (#1925 slice 3): with the setting on, an
+// upstream sampling request is answered by the configured model layer;
+// with it off, the broker relay path is untouched; with it on but no
+// provider configured, the failure is clean and deterministic.
+
+func TestSampleLocallyAnswersViaConfiguredProvider(t *testing.T) {
+	var gotModel string
+	var gotMessages []map[string]any
+	srv := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.URL.Path != "/chat/completions" {
+			t.Errorf("path = %s, want /chat/completions", r.URL.Path)
+		}
+		body, _ := io.ReadAll(r.Body)
+		var payload struct {
+			Model    string           `json:"model"`
+			Messages []map[string]any `json:"messages"`
+		}
+		if err := json.Unmarshal(body, &payload); err != nil {
+			t.Errorf("payload: %v", err)
+		}
+		gotModel = payload.Model
+		gotMessages = payload.Messages
+		w.Header().Set("Content-Type", "application/json")
+		_, _ = io.WriteString(w, `{"model":"test-model","choices":[{"message":{"role":"assistant","content":"local answer"},"finish_reason":"stop"}]}`)
+	}))
+	defer srv.Close()
+
+	r := &Registry{consumer: ConsumerPolicy{
+		SamplingAnswerLocally: true,
+		ModelProvider:         "openai",
+		ModelBaseURL:          srv.URL,
+		ModelDefault:          "openai/test-model",
+		ModelTimeout:          5 * time.Second,
+	}}
+	out, err := r.answerServer(context.Background(), "demo", "key", "sampling/createMessage",
+		json.RawMessage(`{"messages":[{"role":"user","content":{"type":"text","text":"hi"}}],"systemPrompt":"be terse","maxTokens":7}`))
+	if err != nil {
+		t.Fatalf("answerServer: %v", err)
+	}
+	result, ok := out.(map[string]any)
+	if !ok {
+		t.Fatalf("result type %T, want map", out)
+	}
+	if result["role"] != "assistant" || result["model"] != "test-model" || result["stopReason"] != "endTurn" {
+		t.Errorf("result = %v", result)
+	}
+	content, ok := result["content"].(map[string]any)
+	if !ok || content["type"] != "text" || content["text"] != "local answer" {
+		t.Errorf("content = %v", result["content"])
+	}
+	if gotModel != "test-model" {
+		t.Errorf("provider was asked for model %q, want test-model (model.default, prefix stripped)", gotModel)
+	}
+	if len(gotMessages) != 2 || gotMessages[0]["content"] != "be terse" || gotMessages[1]["content"] != "hi" {
+		t.Errorf("provider messages = %v, want system prompt then the user message", gotMessages)
+	}
+}
+
+func TestSampleLocallyEnabledWithoutProviderFailsCleanly(t *testing.T) {
+	r := &Registry{consumer: ConsumerPolicy{SamplingAnswerLocally: true}}
+	_, err := r.answerServer(context.Background(), "demo", "key", "sampling/createMessage",
+		json.RawMessage(`{"messages":[{"role":"user","content":{"type":"text","text":"hi"}}],"maxTokens":7}`))
+	if err == nil || !strings.Contains(err.Error(), "no model provider is configured") {
+		t.Fatalf("err = %v, want the clean no-provider failure", err)
+	}
+}
+
+func TestSampleLocallySamplingOnlyProviderFailsCleanly(t *testing.T) {
+	// model.provider=sampling would relay to the caller -- the very path
+	// answerLocally replaces -- so with no other provider configured the
+	// local answer is unavailable, not a silent relay.
+	r := &Registry{consumer: ConsumerPolicy{
+		SamplingAnswerLocally: true,
+		ModelProvider:         "sampling",
+		ModelTimeout:          5 * time.Second,
+	}}
+	_, err := r.answerServer(context.Background(), "demo", "key", "sampling/createMessage",
+		json.RawMessage(`{"messages":[{"role":"user","content":{"type":"text","text":"hi"}}],"maxTokens":7}`))
+	if err == nil || !strings.Contains(err.Error(), "sampling.answerLocally") {
+		t.Fatalf("err = %v, want a clean sampling.answerLocally failure", err)
+	}
+}
+
+func TestSamplingRelayUnchangedWhenAnswerLocallyOff(t *testing.T) {
+	// Setting off (the default): answerServer keeps the relay behaviour,
+	// which with no broker is the long-standing no-broker error.
+	r := &Registry{}
+	_, err := r.answerServer(context.Background(), "demo", "key", "sampling/createMessage",
+		json.RawMessage(`{"messages":[{"role":"user","content":{"type":"text","text":"hi"}}],"maxTokens":7}`))
+	if err == nil || !strings.Contains(err.Error(), "no broker") {
+		t.Fatalf("err = %v, want the pre-existing no-broker error", err)
+	}
+}
+
+func TestSamplingLocalRequestConversion(t *testing.T) {
+	pol := ConsumerPolicy{ModelBudget: 10}
+	req, err := samplingLocalRequest(json.RawMessage(
+		`{"messages":[{"role":"user","content":[{"type":"text","text":"a"},{"type":"text","text":"b"}]}],"maxTokens":100,"modelPreferences":{"hints":[{"name":"openai/hinted"}]}}`), pol)
+	if err != nil {
+		t.Fatalf("samplingLocalRequest: %v", err)
+	}
+	if req.Model != "openai/hinted" {
+		t.Errorf("model = %q, want the first preference hint", req.Model)
+	}
+	if req.MaxTokens != 10 {
+		t.Errorf("maxTokens = %d, want model.budget's cap of 10", req.MaxTokens)
+	}
+	if len(req.Messages) != 1 || req.Messages[0].Content != "a\nb" {
+		t.Errorf("messages = %+v, want the two text blocks joined", req.Messages)
+	}
+
+	if _, err := samplingLocalRequest(json.RawMessage(
+		`{"messages":[{"role":"user","content":{"type":"text","text":"hi"}}],"tools":[{"name":"x"}],"maxTokens":7}`), pol); err == nil {
+		t.Error("a request carrying tools must fail rather than be flattened to text")
+	}
+	if _, err := samplingLocalRequest(json.RawMessage(
+		`{"messages":[{"role":"user","content":{"type":"image","data":"AA==","mimeType":"image/png"}}],"maxTokens":7}`), pol); err == nil {
+		t.Error("a non-text content block must fail rather than be dropped silently")
+	}
+}