fix(mcp): serialize write tools

Author: Alan-TheGentleman

internal/mcp/mcp.go

@@ -237,6 +237,8 @@ func shouldRegister(name string, allowlist map[string]bool) bool {
 }
 
 func registerTools(srv *server.MCPServer, s *store.Store, cfg MCPConfig, allowlist map[string]bool, activity *SessionActivity) {
+	writeQueue := newWriteQueue(defaultMCPWriteQueueSize)
+
 	// ─── mem_search (profile: agent, core — always in context) ─────────
 	if shouldRegister("mem_search", allowlist) {
 		srv.AddTool(
@@ -324,7 +326,7 @@ Examples:
 					mcp.Description("Optional topic identifier for upserts (e.g. architecture/auth-model). Reuses and updates the latest observation in same project+scope."),
 				),
 			),
-			handleSave(s, cfg, activity),
+			queuedWriteHandler(writeQueue, handleSave(s, cfg, activity)),
 		)
 	}
 
@@ -359,7 +361,7 @@ Examples:
 					mcp.Description("New topic key (normalized internally)"),
 				),
 			),
-			handleUpdate(s),
+			queuedWriteHandler(writeQueue, handleUpdate(s)),
 		)
 	}
 
@@ -407,7 +409,7 @@ Examples:
 					mcp.Description("If true, permanently deletes the observation"),
 				),
 			),
-			handleDelete(s),
+			queuedWriteHandler(writeQueue, handleDelete(s)),
 		)
 	}
 
@@ -429,7 +431,7 @@ Examples:
 					mcp.Description("Session ID to associate with (default: manual-save-{project})"),
 				),
 			),
-			handleSavePrompt(s, cfg),
+			queuedWriteHandler(writeQueue, handleSavePrompt(s, cfg)),
 		)
 	}
 
@@ -570,7 +572,7 @@ GUIDELINES:
 				),
 				// project field intentionally omitted — auto-detect only (REQ-308 write-tool contract)
 			),
-			handleSessionSummary(s, cfg, activity),
+			queuedWriteHandler(writeQueue, handleSessionSummary(s, cfg, activity)),
 		)
 	}
 
@@ -593,7 +595,7 @@ GUIDELINES:
 					mcp.Description("Working directory"),
 				),
 			),
-			handleSessionStart(s, cfg, activity),
+			queuedWriteHandler(writeQueue, handleSessionStart(s, cfg, activity)),
 		)
 	}
 
@@ -616,7 +618,7 @@ GUIDELINES:
 					mcp.Description("Summary of what was accomplished"),
 				),
 			),
-			handleSessionEnd(s, cfg, activity),
+			queuedWriteHandler(writeQueue, handleSessionEnd(s, cfg, activity)),
 		)
 	}
 
@@ -646,7 +648,7 @@ Duplicates are automatically detected and skipped — safe to call multiple time
 					mcp.Description("Source identifier (e.g. 'subagent-stop', 'session-end')"),
 				),
 			),
-			handleCapturePassive(s, cfg, activity),
+			queuedWriteHandler(writeQueue, handleCapturePassive(s, cfg, activity)),
 		)
 	}
 
@@ -670,7 +672,7 @@ Duplicates are automatically detected and skipped — safe to call multiple time
 					mcp.Description("The canonical project name to merge INTO (e.g. 'engram')"),
 				),
 			),
-			handleMergeProjects(s),
+			queuedWriteHandler(writeQueue, handleMergeProjects(s)),
 		)
 	}
 
@@ -739,7 +741,7 @@ Re-judging an already-judged ID overwrites the verdict (deliberate revision).`),
 					mcp.Description("Session ID for provenance (default: auto)"),
 				),
 			),
-			handleJudge(s, activity),
+			queuedWriteHandler(writeQueue, handleJudge(s, activity)),
 		)
 	}
 }

internal/mcp/write_queue.go

@@ -0,0 +1,104 @@
+package mcp
+
+import (
+	"context"
+	"errors"
+	"fmt"
+
+	mcppkg "github.com/mark3labs/mcp-go/mcp"
+	"github.com/mark3labs/mcp-go/server"
+)
+
+const defaultMCPWriteQueueSize = 32
+
+var errMCPWriteQueueFull = errors.New("engram: mcp write queue is full; retry shortly")
+
+type writeQueue struct {
+	jobs chan writeJob
+}
+
+type writeJob struct {
+	ctx    context.Context
+	run    func(context.Context) (*mcppkg.CallToolResult, error)
+	result chan writeResult
+}
+
+type writeResult struct {
+	result *mcppkg.CallToolResult
+	err    error
+}
+
+func newWriteQueue(size int) *writeQueue {
+	if size <= 0 {
+		size = defaultMCPWriteQueueSize
+	}
+	q := &writeQueue{jobs: make(chan writeJob, size)}
+	go q.run()
+	return q
+}
+
+func (q *writeQueue) run() {
+	for job := range q.jobs {
+		if err := job.ctx.Err(); err != nil {
+			job.result <- writeResult{err: err}
+			continue
+		}
+
+		result, err := runWriteJob(job)
+		job.result <- writeResult{result: result, err: err}
+	}
+}
+
+func runWriteJob(job writeJob) (result *mcppkg.CallToolResult, err error) {
+	defer func() {
+		if recovered := recover(); recovered != nil {
+			result = nil
+			err = fmt.Errorf("mcp write handler panic: %v", recovered)
+		}
+	}()
+
+	return job.run(job.ctx)
+}
+
+func (q *writeQueue) Do(ctx context.Context, run func(context.Context) (*mcppkg.CallToolResult, error)) (*mcppkg.CallToolResult, error) {
+	if err := ctx.Err(); err != nil {
+		return nil, err
+	}
+
+	job := writeJob{
+		ctx:    ctx,
+		run:    run,
+		result: make(chan writeResult, 1),
+	}
+
+	select {
+	case q.jobs <- job:
+		// Enqueued.
+	default:
+		return nil, errMCPWriteQueueFull
+	}
+
+	// The worker owns the post-enqueue cancellation decision. Returning directly
+	// on ctx.Done() here can race with the worker starting the job: the caller may
+	// see cancellation while the handler is about to mutate SQLite. Waiting for the
+	// worker's result makes the outcome deterministic: queued canceled jobs are
+	// skipped by the worker before start, while started jobs finish and return the
+	// handler result.
+	res := <-job.result
+	return res.result, res.err
+}
+
+func queuedWriteHandler(q *writeQueue, h server.ToolHandlerFunc) server.ToolHandlerFunc {
+	return func(ctx context.Context, req mcppkg.CallToolRequest) (*mcppkg.CallToolResult, error) {
+		result, err := q.Do(ctx, func(runCtx context.Context) (*mcppkg.CallToolResult, error) {
+			return h(runCtx, req)
+		})
+		if err == nil {
+			return result, nil
+		}
+		if errors.Is(err, errMCPWriteQueueFull) || errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded) {
+			return mcppkg.NewToolResultError(fmt.Sprintf("MCP write queue error: %s", err)), nil
+		}
+		return nil, err
+	}
+}