feat(mcp): session-scoped notification delivery and wire-format negotiation

cmd/gortex/daemon_mcp.go

@@ -97,6 +97,11 @@ func (d *mcpDispatcher) Dispatch(ctx context.Context, sess *daemon.Session, fram
 	// every query runs against the whole multi-workspace graph and a
 	// session in workspace A can see workspace B's nodes.
 	ctx = gortexmcp.WithSessionCWD(ctx, sess.CWD)
+	// Run session-aware: attaching the connected client session (wired
+	// by SessionStarted) lets mcp-go's initialize handler mark it
+	// initialized — the gate SendNotificationToAllClients applies
+	// before delivering tools/list_changed and other pushes.
+	ctx = d.srv.WithClientSession(ctx, sess.ID)
 
 	// Relay the client-forwarded tool-surface preference (GORTEX_TOOLS /
 	// --tools of the proxy) so the MCP server resolves THIS session's
@@ -224,11 +229,29 @@ func rewriteUntrackedResponse(method string, out []byte, cwd string) []byte {
 	return out
 }
 
+// SessionStarted implements daemon.SessionStartedHook: register the
+// connection as a live mcp-go client session so server-initiated
+// notifications (tools/list_changed after a tools_search promotion,
+// graph_invalidated, diagnostics pushes, ...) reach this client over
+// the socket. Best-effort — on failure the session still works
+// request/response, it just misses pushes.
+func (d *mcpDispatcher) SessionStarted(sess *daemon.Session, write func([]byte) error) {
+	if d.srv == nil || sess == nil {
+		return
+	}
+	if err := d.srv.ConnectSession(sess.ID, write); err != nil {
+		d.logger.Debug("daemon: notification session unavailable",
+			zap.String("session_id", sess.ID), zap.Error(err))
+	}
+}
+
 // SessionEnded implements daemon.SessionEndedHook. When a proxy
-// disconnects, drop its entry from the MCP server's session map so idle
-// per-session state doesn't accumulate for the daemon's lifetime.
+// disconnects, unregister its notification session and drop its entry
+// from the MCP server's session map so idle per-session state doesn't
+// accumulate for the daemon's lifetime.
 func (d *mcpDispatcher) SessionEnded(sess *daemon.Session) {
 	if d.srv != nil && sess != nil {
+		d.srv.DisconnectSession(sess.ID)
 		d.srv.ReleaseSession(sess.ID)
 	}
 }
@@ -389,6 +412,9 @@ func (d *mcpDispatcher) maybeSnoopInitialize(sess *daemon.Session, frame []byte)
 				Name    string `json:"name"`
 				Version string `json:"version"`
 			} `json:"clientInfo"`
+			Capabilities struct {
+				WireFormats []string `json:"gortex/wire"`
+			} `json:"capabilities"`
 		} `json:"params"`
 	}
 	if err := json.Unmarshal(frame, &peek); err != nil {
@@ -397,6 +423,12 @@ func (d *mcpDispatcher) maybeSnoopInitialize(sess *daemon.Session, frame []byte)
 	if peek.Method != "initialize" {
 		return
 	}
+	// A gortex/wire capability self-declares the client's compact-format
+	// decoders; session format resolution prefers it over the built-in
+	// client-name allowlist, so new clients need no server-side entry.
+	if formats := peek.Params.Capabilities.WireFormats; len(formats) > 0 && d.srv != nil {
+		d.srv.NoteSessionWireFormats(sess.ID, formats)
+	}
 	if peek.Params.ClientInfo.Name == "" && peek.Params.ClientInfo.Version == "" {
 		return
 	}

internal/daemon/mcp_roundtrip_test.go

@@ -186,3 +186,75 @@ func TestDaemon_SessionEndedHook_FiresOnDisconnect(t *testing.T) {
 		t.Fatal("SessionEnded hook never fired after client close")
 	}
 }
+
+// pushDispatcher satisfies MCPDispatcher and SessionStartedHook so we
+// can prove a dispatcher-initiated frame (an MCP notification) reaches
+// the client over the socket, interleaved with the reply path.
+type pushDispatcher struct {
+	echoDispatcher
+	started chan func([]byte) error
+}
+
+func (p *pushDispatcher) SessionStarted(_ *Session, write func([]byte) error) {
+	p.started <- write
+}
+
+// TestDaemon_SessionStartedHook_PushesFrames pins the server-initiated
+// notification path: the daemon hands dispatchers a write callback on
+// connect, and frames written through it arrive at the client without a
+// pending request — the transport leg that lets tools/list_changed
+// (and every other MCP push) reach unix-socket proxy clients.
+func TestDaemon_SessionStartedHook_PushesFrames(t *testing.T) {
+	dir, err := os.MkdirTemp("/tmp", "gx")
+	require.NoError(t, err)
+	defer func() { _ = os.RemoveAll(dir) }()
+	socket := filepath.Join(dir, "s")
+	t.Setenv("GORTEX_DAEMON_SOCKET", socket)
+	t.Setenv("GORTEX_DAEMON_PIDFILE", filepath.Join(dir, "p"))
+
+	disp := &pushDispatcher{
+		echoDispatcher: echoDispatcher{seen: make(chan echoFrame, 4)},
+		started:        make(chan func([]byte) error, 1),
+	}
+	srv := New(socket, "test", zap.NewNop())
+	srv.MCPDispatcher = disp
+	srv.Controller = &fakeController{}
+	require.NoError(t, srv.Listen())
+	go func() { _ = srv.Serve() }()
+	defer func() { _ = srv.Shutdown() }()
+
+	require.Eventually(t, func() bool { return IsRunningAt(socket) },
+		2*time.Second, 10*time.Millisecond)
+
+	client, err := DialTo(socket, Handshake{Mode: ModeMCP, CWD: "/tmp/fake-repo"})
+	require.NoError(t, err)
+	defer client.Close()
+
+	var write func([]byte) error
+	select {
+	case write = <-disp.started:
+	case <-time.After(2 * time.Second):
+		t.Fatal("SessionStarted hook never fired after connect")
+	}
+
+	// Push an unprompted notification; the client must read it even
+	// though it never sent a request.
+	notif := `{"jsonrpc":"2.0","method":"notifications/tools/list_changed"}`
+	require.NoError(t, write([]byte(notif)))
+
+	frame, err := client.ReadMCPFrame()
+	require.NoError(t, err)
+	var got map[string]any
+	require.NoError(t, json.Unmarshal(frame, &got))
+	assert.Equal(t, "notifications/tools/list_changed", got["method"])
+
+	// The request/reply path still works alongside the push path.
+	rpc := `{"jsonrpc":"2.0","id":1,"method":"tools/list","params":{}}`
+	require.NoError(t, client.WriteMCPFrame([]byte(rpc)))
+	reply, err := client.ReadMCPFrame()
+	require.NoError(t, err)
+	var replyObj map[string]any
+	require.NoError(t, json.Unmarshal(reply, &replyObj))
+	_, hasResult := replyObj["result"]
+	assert.True(t, hasResult, "reply must carry a result: %s", reply)
+}

internal/daemon/server.go

@@ -94,6 +94,20 @@ type SessionEndedHook interface {
 	SessionEnded(sess *Session)
 }
 
+// SessionStartedHook is an optional extension that MCPDispatcher
+// implementations can satisfy to receive a connect callback for each
+// ModeMCP session, fired after the handshake ack and before the first
+// frame is dispatched. write delivers one server-initiated JSON-RPC
+// frame (without trailing newline) to the client; it is safe for
+// concurrent use with the reply path — the daemon serialises all
+// writes to the connection — and returns an error once the connection
+// is gone. This is how server-initiated MCP notifications
+// (tools/list_changed, graph_invalidated, ...) reach socket clients;
+// the matching SessionEnded fires on teardown.
+type SessionStartedHook interface {
+	SessionStarted(sess *Session, write func([]byte) error)
+}
+
 // Controller implements the daemon's control surface. Separated from
 // MCPDispatcher so the two can evolve independently and so control-only
 // tests don't need a full MCP stack.
@@ -477,6 +491,21 @@ func (s *Server) serveMCP(conn net.Conn, reader *bufio.Reader, sess *Session) {
 		})
 		return
 	}
+
+	// Server-initiated frames (notifications pushed by the dispatcher)
+	// share the connection with request replies; serialise every write
+	// under one lock so frames never interleave mid-line.
+	var writeMu sync.Mutex
+	writeFrame := func(frame []byte) error {
+		writeMu.Lock()
+		defer writeMu.Unlock()
+		_, err := conn.Write(append(frame, '\n'))
+		return err
+	}
+	if hook, ok := s.MCPDispatcher.(SessionStartedHook); ok && hook != nil {
+		hook.SessionStarted(sess, writeFrame)
+	}
+
 	for {
 		line, err := reader.ReadBytes('\n')
 		if err != nil {
@@ -506,7 +535,7 @@ func (s *Server) serveMCP(conn net.Conn, reader *bufio.Reader, sess *Session) {
 			continue
 		}
 		// The dispatcher returns a full JSON-RPC frame; re-append newline.
-		if _, werr := conn.Write(append(reply, '\n')); werr != nil {
+		if werr := writeFrame(reply); werr != nil {
 			s.Logger.Debug("daemon: mcp write failed",
 				zap.String("session_id", sess.ID), zap.Error(werr))
 			return

internal/mcp/conn_session.go

@@ -0,0 +1,114 @@
+package mcp
+
+import (
+	"context"
+	"encoding/json"
+	"fmt"
+	"sync/atomic"
+
+	"github.com/mark3labs/mcp-go/mcp"
+	"github.com/mark3labs/mcp-go/server"
+)
+
+// connClientSession adapts one daemon-socket connection to mcp-go's
+// server.ClientSession so server-initiated notifications
+// (notifications/tools/list_changed after a tools_search promotion,
+// graph_invalidated, diagnostics pushes, ...) reach stdio-proxy
+// clients. Without this registration, HandleMessage-dispatched
+// sessions are invisible to SendNotificationToAllClients: the
+// broadcast iterates registered sessions only, so unix-socket clients
+// silently miss every push — the streamable HTTP transport and the
+// embedded stdio server were the only paths that ever delivered one.
+//
+// The pump goroutine serialises each notification and hands it to the
+// transport-owned write callback; the daemon interleaves those frames
+// with request replies under its own connection write lock.
+type connClientSession struct {
+	id          string
+	notifCh     chan mcp.JSONRPCNotification
+	done        chan struct{}
+	initialized atomic.Bool
+}
+
+func (c *connClientSession) SessionID() string { return c.id }
+func (c *connClientSession) NotificationChannel() chan<- mcp.JSONRPCNotification {
+	return c.notifCh
+}
+func (c *connClientSession) Initialize()       { c.initialized.Store(true) }
+func (c *connClientSession) Initialized() bool { return c.initialized.Load() }
+
+// connNotifBuffer sizes the per-session notification channel. mcp-go
+// delivers non-blocking and drops on overflow, so the buffer only has
+// to absorb short bursts (a promotion sweep fires a single frame).
+const connNotifBuffer = 16
+
+// ConnectSession registers a daemon-socket connection as a live MCP
+// client session on the underlying mcp-go server. write must deliver
+// one complete JSON-RPC frame to the client (no trailing newline) and
+// be safe to call from the pump goroutine concurrently with the
+// request/reply path; once it errors the pump stops (the daemon's
+// teardown calls DisconnectSession shortly after).
+//
+// Pair every ConnectSession with a DisconnectSession — the mcp-go
+// session registry and the pump goroutine live until then.
+func (s *Server) ConnectSession(sessionID string, write func([]byte) error) error {
+	if s.mcpServer == nil {
+		return fmt.Errorf("connect session %s: mcp server not initialised", sessionID)
+	}
+	cs := &connClientSession{
+		id:      sessionID,
+		notifCh: make(chan mcp.JSONRPCNotification, connNotifBuffer),
+		done:    make(chan struct{}),
+	}
+	if _, loaded := s.connSessions.LoadOrStore(sessionID, cs); loaded {
+		return fmt.Errorf("connect session %s: already connected", sessionID)
+	}
+	if err := s.mcpServer.RegisterSession(context.Background(), cs); err != nil {
+		s.connSessions.Delete(sessionID)
+		return fmt.Errorf("connect session %s: %w", sessionID, err)
+	}
+	go func() {
+		for {
+			select {
+			case n := <-cs.notifCh:
+				frame, err := json.Marshal(n)
+				if err != nil {
+					continue
+				}
+				if write(frame) != nil {
+					return
+				}
+			case <-cs.done:
+				return
+			}
+		}
+	}()
+	return nil
+}
+
+// DisconnectSession unregisters a session connected via ConnectSession
+// and stops its notification pump. Unknown ids are a no-op, so callers
+// can invoke it unconditionally on connection teardown.
+func (s *Server) DisconnectSession(sessionID string) {
+	v, ok := s.connSessions.LoadAndDelete(sessionID)
+	if !ok {
+		return
+	}
+	if s.mcpServer != nil {
+		s.mcpServer.UnregisterSession(context.Background(), sessionID)
+	}
+	close(v.(*connClientSession).done)
+}
+
+// WithClientSession attaches the connected session for sessionID to
+// ctx so HandleMessage runs session-aware. This is what lets mcp-go's
+// initialize handler mark the session initialized — the gate
+// sendNotificationToAllClients applies before delivering a broadcast —
+// and what enables session-targeted APIs generally. A sessionID
+// without a connected session returns ctx unchanged.
+func (s *Server) WithClientSession(ctx context.Context, sessionID string) context.Context {
+	if v, ok := s.connSessions.Load(sessionID); ok {
+		return s.mcpServer.WithContext(ctx, v.(server.ClientSession))
+	}
+	return ctx
+}