proxy: fix race conditions during swap (#667)

I pointed Opus 4.7 (high effort) at proxy.ProcessGroup to identify any
race conditions in the swapping code. It found a race condition where
there is a small window in the fast path for routing a request to a
loaded model. There is a very small window where:

- model M1 is loaded and ready for requests
- a request, R1, for M1 comes in 
- a request, R2, for M2 comes in almost immediately after
- R1 acquires the lock, sees M1 is loaded (fast path), releases the lock
`[race window]` and the request is ready to be forwarded
- the race window occurs between the release of the lock and the request
being forwarded
  - the lock is released so requests can be handled concurrently 
- R2 comes in within the `[race window]`, acquires the lock, triggers a
model swap to M2. stopping M1
- R1 is forwarded to a model that is unloaded or in the process of
shutting down creating an error response

In deployed systems the race window is very small and doesn't happen
often. However with #635 and PR #656 I though this deserved a bit more
attention. It is not concluded that this race is the cause of #635 but
the race is likely to happen more often under sustained or high load.

AI Note: Opus 4.7 x-high effort took about an hour to write the original
patch. With the pattern discovered the fix to matrix.go was very quick.
GLM 5.1 using the previous established patterns was able to easily write
the fix for ProcessGroup.StopProcesses().

Supersedes: #656
Updates: #277, #635

proxy/processgroup.go

@@ -24,6 +24,22 @@ type ProcessGroup struct {
 	// map of current processes
 	processes       map[string]*Process
 	lastUsedProcess string
+
+	// inflight tracks fast-path requests (requests for the already-selected
+	// model in a swap group). Fast-path requests Add(1) while holding pg.Lock
+	// and Done() on completion; a concurrent swap request calls inflight.Wait()
+	// under pg.Lock before stopping the current process. Without this tracking,
+	// a fast-path request that has released pg.Lock but has not yet called
+	// Process.inFlightRequests.Add(1) races with Stop()'s Wait() and can be
+	// killed mid-request.
+	inflight sync.WaitGroup
+
+	// testDelayFastPath is a test-only hook that, when non-nil, is invoked in
+	// the fast path after pg.Lock is released but before the request is
+	// dispatched to Process.ProxyRequest. Tests use it to park a fast-path
+	// request at the exact race window to deterministically reproduce the
+	// fast-path vs swap race.
+	testDelayFastPath func()
 }
 
 func NewProcessGroup(id string, config config.Config, proxyLogger *LogMonitor, upstreamLogger *LogMonitor) *ProcessGroup {
@@ -64,6 +80,13 @@ func (pg *ProcessGroup) ProxyRequest(modelID string, writer http.ResponseWriter,
 		pg.Lock()
 		if pg.lastUsedProcess != modelID {
 
+			// Wait for in-flight fast-path requests to drain before stopping
+			// the previous process. Without this, a fast-path request that has
+			// released pg.Lock but has not yet incremented
+			// Process.inFlightRequests races with Stop() and can be killed
+			// mid-request.
+			pg.inflight.Wait()
+
 			// is there something already running?
 			if pg.lastUsedProcess != "" {
 				pg.processes[pg.lastUsedProcess].Stop()
@@ -78,7 +101,16 @@ func (pg *ProcessGroup) ProxyRequest(modelID string, writer http.ResponseWriter,
 			pg.Unlock()
 			return nil
 		}
+
+		// Fast path: register this request in inflight before releasing
+		// pg.Lock so a concurrent swap will wait for it to complete.
+		pg.inflight.Add(1)
+		defer pg.inflight.Done()
 		pg.Unlock()
+
+		if pg.testDelayFastPath != nil {
+			pg.testDelayFastPath()
+		}
 	}
 
 	pg.processes[modelID].ProxyRequest(writer, request)
@@ -123,6 +155,10 @@ func (pg *ProcessGroup) StopProcesses(strategy StopStrategy) {
 	pg.Lock()
 	defer pg.Unlock()
 
+	if strategy != StopImmediately {
+		pg.inflight.Wait()
+	}
+
 	if len(pg.processes) == 0 {
 		return
 	}