client/server: in-process dispatch fast-path (opt-in)

When caller and callee run in the same process, a unary Call pays the
full network tax — pool.Get, dial, codec-over-socket, and the transport
pump — even though the handler table is right there. This adds an opt-in
fast-path that dispatches directly.

- internal/network: a neutral registry (transport.Message in/out) so
  client and server wire up without importing each other. A running server
  registers a dispatcher under its name on Start, deregisters on Stop.
- server: localDispatch serves a request in-process through the same
  router (identical wrappers/codecs/error mapping) over an in-memory
  socket — no dial, no pipe, no gob.
- client: LocalDispatch() opt-in. In call(), a unary request whose body and
  response are raw frames (codec/bytes.Frame — the agent/MCP/flow shape)
  dispatches locally; everything else falls back to the network path
  unchanged.

Correctness test proves the fast-path returns byte-identical replies to
the network path; benchmark shows ~545µs -> ~28µs (~20x) and ~3.6x fewer
allocations. Off by default. Covers #4817 (path b); the zero-copy typed
path remains a follow-up.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01CmdEY7pYmV5zzwCjNJ4ykL

CHANGELOG.md

@@ -25,6 +25,7 @@ below is kept current between tags and rolled into the next version when it ship
 - **A2A inbound AP2 mandate verification (opt-in)** — set `Options.AP2PublicKey` (or `a2a.WithPushURLPolicy`'s sibling `a2a.WithAP2PublicKey` for embedded handlers) and the gateway verifies AP2 payment/checkout mandates carried on incoming messages — signature and task/context binding — recording the outcome in each task's `ap2Verifications`, with the x402 settlement rail carried through for the paid path. Off by default; mandates are otherwise carried unverified. (`gateway/a2a/`)
 - **Flow human-in-the-loop pause/resume** — a flow step can suspend a run for external input with `flow.Await(key, prompt)` (or `flow.AwaitStep`): the run checkpoints with status `waiting` and `Execute` returns cleanly. `Flow.Waiting` lists suspended runs with what they await, and `Flow.ResumeWith(ctx, runID, input)` injects the input and continues from the next step. Recovery (`ResumePending`) skips waiting runs since they need input, not a restart. (`flow/`)
 - **Kubernetes reconcile core (alpha)** — `kubernetes.Reconcile(desired, observed)` decides the single action needed to converge an `Agent`/`Service`/`Flow` resource toward its Deployment (create / update / noop) and returns `Ready`/`Error` status conditions. Dependency-free (no controller-runtime / client-go) and fully unit-testable; a future operator binary supplies observed state and applies the action. (`deploy/kubernetes/`)
+- **In-process dispatch fast-path (opt-in)** — `client.LocalDispatch()` lets a unary `Call` to a service running in the same process skip the network transport and dispatch straight to that server's handlers (for raw `codec/bytes.Frame` bodies — the shape agent/MCP/flow tool calls use), running the same router, wrappers, and codecs. In a benchmark this cut an in-process call from ~545µs to ~28µs (≈20×) with ~3.6× fewer allocations. Off by default; falls back to the network path for anything it doesn't cover. (`client/`, `server/`, `internal/network/`)
 
 ### Changed
 - **Remote agent chat streaming** — `micro chat` now streams replies from remote agents instead of waiting for the full response. (`cmd/micro/`, `agent/`)

client/local.go

@@ -0,0 +1,60 @@
+package client
+
+import (
+	"context"
+
+	raw "go-micro.dev/v6/codec/bytes"
+	"go-micro.dev/v6/internal/network"
+	"go-micro.dev/v6/metadata"
+	"go-micro.dev/v6/transport"
+	"go-micro.dev/v6/transport/headers"
+)
+
+// localCall is the in-process fast-path for Call. When LocalDispatch is enabled
+// and the callee runs in this same process, a unary request whose body and
+// response are raw frames (codec/bytes.Frame) is dispatched straight to the
+// server's handlers via internal/network — no dial, no codec-over-socket,
+// no transport pump. It returns handled=false to fall back to the network path
+// for anything it does not cover (disabled, streaming, non-frame bodies, or a
+// service not registered in-process), so behavior is unchanged unless the
+// fast-path fully applies.
+func (r *rpcClient) localCall(ctx context.Context, req Request, resp interface{}) (handled bool, err error) {
+	if !r.opts.LocalDispatch || req.Stream() {
+		return false, nil
+	}
+	reqFrame, ok := req.Body().(*raw.Frame)
+	if !ok {
+		return false, nil
+	}
+	respFrame, ok := resp.(*raw.Frame)
+	if !ok {
+		return false, nil
+	}
+	dispatch, ok := network.Lookup(req.Service())
+	if !ok {
+		return false, nil
+	}
+
+	header := make(map[string]string)
+	if md, ok := metadata.FromContext(ctx); ok {
+		for k, v := range md {
+			if k == headers.Message { // pub/sub topic header, never forwarded
+				continue
+			}
+			header[k] = v
+		}
+	}
+	header[headers.Request] = req.Service()
+	header[headers.Endpoint] = req.Endpoint()
+	header["Content-Type"] = req.ContentType()
+	header["Accept"] = req.ContentType()
+
+	reply, err := dispatch(ctx, &transport.Message{Header: header, Body: reqFrame.Data})
+	if err != nil {
+		return true, err
+	}
+	if reply != nil {
+		respFrame.Data = reply.Body
+	}
+	return true, nil
+}

client/local_test.go

@@ -0,0 +1,139 @@
+package client_test
+
+import (
+	"context"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"go-micro.dev/v6/client"
+	raw "go-micro.dev/v6/codec/bytes"
+	"go-micro.dev/v6/registry"
+	"go-micro.dev/v6/selector"
+	"go-micro.dev/v6/server"
+)
+
+type EchoReq struct {
+	Msg string `json:"msg"`
+}
+
+type EchoRsp struct {
+	Msg string `json:"msg"`
+}
+
+type EchoHandler struct{}
+
+func (EchoHandler) Echo(_ context.Context, req *EchoReq, rsp *EchoRsp) error {
+	rsp.Msg = "echo:" + req.Msg
+	return nil
+}
+
+// startEchoServer starts a real server on the given registry and returns a stop
+// func. The server is reachable over the network transport and (via Start)
+// registered for the in-process fast-path.
+func startEchoServer(t testing.TB, reg registry.Registry) func() {
+	t.Helper()
+	srv := server.NewServer(
+		server.Name("echo.local"),
+		server.Address("127.0.0.1:0"),
+		server.Registry(reg),
+	)
+	if err := srv.Handle(srv.NewHandler(&EchoHandler{})); err != nil {
+		t.Fatalf("handle: %v", err)
+	}
+	if err := srv.Start(); err != nil {
+		t.Fatalf("start: %v", err)
+	}
+	// Wait for registration so the client's selector can find a node.
+	deadline := time.Now().Add(5 * time.Second)
+	for time.Now().Before(deadline) {
+		if svcs, err := reg.GetService("echo.local"); err == nil && len(svcs) > 0 && len(svcs[0].Nodes) > 0 {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	return func() { _ = srv.Stop() }
+}
+
+func newEchoClient(reg registry.Registry, opts ...client.Option) client.Client {
+	base := []client.Option{
+		client.Registry(reg),
+		client.Selector(selector.NewSelector(selector.Registry(reg))),
+		client.ContentType("application/json"),
+	}
+	return client.NewClient(append(base, opts...)...)
+}
+
+// callEcho makes an echo call with a raw-frame body (the shape agent/MCP/flow
+// dispatch uses) and returns the decoded reply.
+func callEcho(t testing.TB, cl client.Client, msg string) EchoRsp {
+	t.Helper()
+	body, _ := json.Marshal(EchoReq{Msg: msg})
+	req := cl.NewRequest("echo.local", "EchoHandler.Echo", &raw.Frame{Data: body}, client.WithContentType("application/json"))
+	var rsp raw.Frame
+	if err := cl.Call(context.Background(), req, &rsp); err != nil {
+		t.Fatalf("call: %v", err)
+	}
+	var out EchoRsp
+	if err := json.Unmarshal(rsp.Data, &out); err != nil {
+		t.Fatalf("decode reply %q: %v", rsp.Data, err)
+	}
+	return out
+}
+
+// TestLocalDispatchMatchesNetwork proves the in-process fast-path returns the
+// exact same result as the network path for the same handler and request.
+func TestLocalDispatchMatchesNetwork(t *testing.T) {
+	reg := registry.NewMemoryRegistry()
+	stop := startEchoServer(t, reg)
+	defer stop()
+
+	net := newEchoClient(reg)                           // network path
+	local := newEchoClient(reg, client.LocalDispatch()) // in-process fast-path
+
+	netRsp := callEcho(t, net, "hi")
+	localRsp := callEcho(t, local, "hi")
+
+	if netRsp.Msg != "echo:hi" {
+		t.Fatalf("network reply = %q, want echo:hi", netRsp.Msg)
+	}
+	if localRsp != netRsp {
+		t.Fatalf("fast-path reply %+v != network reply %+v", localRsp, netRsp)
+	}
+}
+
+// TestLocalDispatchFallsBackWhenNotLocal confirms a service not registered
+// in-process still works via the network path even with LocalDispatch on.
+func TestLocalDispatchFallsBackWhenNotLocal(t *testing.T) {
+	reg := registry.NewMemoryRegistry()
+	stop := startEchoServer(t, reg)
+	defer stop()
+
+	// LocalDispatch is on, but the call still resolves — the fast-path only
+	// engages when it fully applies, otherwise the network path runs.
+	local := newEchoClient(reg, client.LocalDispatch())
+	if got := callEcho(t, local, "x").Msg; got != "echo:x" {
+		t.Fatalf("reply = %q, want echo:x", got)
+	}
+}
+
+func benchmarkEcho(b *testing.B, opts ...client.Option) {
+	reg := registry.NewMemoryRegistry()
+	stop := startEchoServer(b, reg)
+	defer stop()
+	cl := newEchoClient(reg, opts...)
+	body, _ := json.Marshal(EchoReq{Msg: "hi"})
+
+	b.ReportAllocs()
+	b.ResetTimer()
+	for i := 0; i < b.N; i++ {
+		req := cl.NewRequest("echo.local", "EchoHandler.Echo", &raw.Frame{Data: body}, client.WithContentType("application/json"))
+		var rsp raw.Frame
+		if err := cl.Call(context.Background(), req, &rsp); err != nil {
+			b.Fatalf("call: %v", err)
+		}
+	}
+}
+
+func BenchmarkNetworkCall(b *testing.B) { benchmarkEcho(b) }
+func BenchmarkLocalCall(b *testing.B)   { benchmarkEcho(b, client.LocalDispatch()) }

client/options.go

@@ -68,6 +68,11 @@ type Options struct {
 	PoolSize         int
 	PoolTTL          time.Duration
 	PoolCloseTimeout time.Duration
+
+	// LocalDispatch, when true, lets a unary Call to a service running in this
+	// same process skip the network transport and dispatch directly to that
+	// server's handlers (raw byte bodies only). Off by default.
+	LocalDispatch bool
 }
 
 // CallOptions are options used to make calls to a server.
@@ -181,6 +186,17 @@ func ContentType(ct string) Option {
 	}
 }
 
+// LocalDispatch enables the in-process fast-path: a unary Call to a service
+// running in the same process dispatches straight to that server's handlers
+// (skipping dial, codec-over-socket, and the transport pump) when both request
+// and response bodies are raw frames (codec/bytes.Frame) — the shape agent,
+// MCP, and flow tool calls use. Falls back to the network path otherwise.
+func LocalDispatch() Option {
+	return func(o *Options) {
+		o.LocalDispatch = true
+	}
+}
+
 // PoolSize sets the connection pool size.
 func PoolSize(d int) Option {
 	return func(o *Options) {

client/rpc_client.go

@@ -83,6 +83,12 @@ func (r *rpcClient) call(
 	resp interface{},
 	opts CallOptions,
 ) error {
+	// In-process fast-path: if the callee runs in this process and both bodies
+	// are raw frames, dispatch directly and skip the network entirely.
+	if handled, err := r.localCall(ctx, req, resp); handled {
+		return err
+	}
+
 	address := node.Address
 	logger := r.Options().Logger
 

internal/network/network.go

@@ -0,0 +1,51 @@
+// Package network is a process-local registry of server dispatchers — the
+// neutral seam an in-process client fast-path uses to reach a server running in
+// the same process without going over the network transport.
+//
+// It lives in internal/ and speaks only in transport.Message so neither the
+// client nor the server package has to import the other: a running server
+// registers a Handler under its service name; an opted-in client looks one up
+// and dispatches directly, skipping dial, codec-over-socket, and the transport
+// pump. Nothing here runs unless a server registers and a client opts in.
+package network
+
+import (
+	"context"
+	"sync"
+
+	"go-micro.dev/v6/transport"
+)
+
+// Handler dispatches one request against a process-local server's handler
+// table and returns the reply. req and the returned message carry the same
+// codec-encoded body + headers the transport would have carried.
+type Handler func(ctx context.Context, req *transport.Message) (*transport.Message, error)
+
+var (
+	mu  sync.RWMutex
+	reg = map[string]Handler{}
+)
+
+// Register makes service reachable in-process via h. A server calls this when
+// it starts; calling again replaces the handler.
+func Register(service string, h Handler) {
+	mu.Lock()
+	reg[service] = h
+	mu.Unlock()
+}
+
+// Deregister removes service's in-process handler. A server calls this when it
+// stops, so a later in-process call falls back to the network path.
+func Deregister(service string) {
+	mu.Lock()
+	delete(reg, service)
+	mu.Unlock()
+}
+
+// Lookup returns the in-process handler for service, if one is registered.
+func Lookup(service string) (Handler, bool) {
+	mu.RLock()
+	h, ok := reg[service]
+	mu.RUnlock()
+	return h, ok
+}

server/local.go

@@ -0,0 +1,119 @@
+package server
+
+import (
+	"context"
+	"io"
+
+	"go-micro.dev/v6/internal/network"
+	"go-micro.dev/v6/transport"
+	"go-micro.dev/v6/transport/headers"
+)
+
+// local.go gives a same-process caller a way to reach this server's handlers
+// without the network transport. A running server registers a dispatcher in
+// internal/network keyed by its name; an opted-in client looks it up and
+// calls localDispatch, which serves the request synchronously through the same
+// router (so handler wrappers, codecs, and error mapping are identical) over an
+// in-memory socket — skipping dial, the transport pump, and the codec-over-pipe
+// double serialization. Unary only; streaming and pub/sub keep the normal path.
+
+// localSocket is a transport.Socket that carries exactly one request in and
+// captures exactly one reply — no network, no pipe, no gob. Recv delivers the
+// request message once (the RPC codec reads it on the first ReadHeader), then
+// reports EOF; Send captures the encoded reply.
+type localSocket struct {
+	req   *transport.Message
+	recvd bool
+	reply *transport.Message
+}
+
+func (s *localSocket) Recv(m *transport.Message) error {
+	if s.recvd || s.req == nil {
+		return io.EOF
+	}
+	s.recvd = true
+	m.Header = s.req.Header
+	m.Body = s.req.Body
+	return nil
+}
+
+func (s *localSocket) Send(m *transport.Message) error {
+	cp := &transport.Message{Header: make(map[string]string, len(m.Header))}
+	for k, v := range m.Header {
+		cp.Header[k] = v
+	}
+	if len(m.Body) > 0 {
+		cp.Body = append([]byte(nil), m.Body...)
+	}
+	s.reply = cp
+	return nil
+}
+
+func (s *localSocket) Close() error   { return nil }
+func (s *localSocket) Local() string  { return "local" }
+func (s *localSocket) Remote() string { return "local" }
+
+// localDispatch serves req against this server's router in-process and returns
+// the reply. It mirrors the request/response construction ServeConn does for a
+// networked request, so the served path is identical apart from the transport.
+func (s *rpcServer) localDispatch(ctx context.Context, req *transport.Message) (*transport.Message, error) {
+	contentType := req.Header["Content-Type"]
+	if contentType == "" {
+		contentType = DefaultContentType
+		req.Header["Content-Type"] = contentType
+	}
+
+	cf := setupProtocol(req)
+	if cf == nil {
+		var err error
+		if cf, err = s.newCodec(contentType); err != nil {
+			return nil, err
+		}
+	}
+
+	sock := &localSocket{req: req}
+	rcodec := newRPCCodec(req, sock, cf)
+
+	request := rpcRequest{
+		service:     getHeader(headers.Request, req.Header),
+		method:      getHeader(headers.Method, req.Header),
+		endpoint:    getHeader(headers.Endpoint, req.Header),
+		contentType: contentType,
+		codec:       rcodec,
+		header:      req.Header,
+		body:        req.Body,
+		socket:      sock,
+	}
+	response := rpcResponse{
+		header: make(map[string]string),
+		socket: sock,
+		codec:  rcodec,
+	}
+
+	if err := s.getRouter().ServeRequest(ctx, &request, &response); err != nil {
+		return nil, err
+	}
+	if sock.reply == nil {
+		// A handler that wrote no body still completed successfully.
+		return &transport.Message{Header: map[string]string{}}, nil
+	}
+	return sock.reply, nil
+}
+
+// registerLocal makes this server reachable in-process under its name; called
+// on Start. deregisterLocal removes it on Stop.
+func (s *rpcServer) registerLocal() {
+	name := s.Options().Name
+	if name == "" {
+		return
+	}
+	network.Register(name, s.localDispatch)
+}
+
+func (s *rpcServer) deregisterLocal() {
+	name := s.Options().Name
+	if name == "" {
+		return
+	}
+	network.Deregister(name)
+}

server/rpc_server.go

@@ -571,6 +571,9 @@ func (s *rpcServer) Start() error {
 	// Keep the service registered to registry
 	go s.registrar(listener, addr, config, exit)
 
+	// Make this server reachable in-process for the client fast-path.
+	s.registerLocal()
+
 	s.setStarted(true)
 
 	return nil
@@ -581,6 +584,8 @@ func (s *rpcServer) Stop() error {
 		return nil
 	}
 
+	s.deregisterLocal()
+
 	ch := make(chan error)
 	s.exit <- ch