Add endpoint aliases for reranking models (#201)

* Add endpoint aliases for reranking models
* Add MetricsMiddleware to the previous reranking endpoint
* Fix the embeddings endpoint not having model set

README.md

@@ -18,7 +18,7 @@ Written in golang, it is very easy to install (single binary with no dependencie
   - `v1/completions`
   - `v1/chat/completions`
   - `v1/embeddings`
-  - `v1/rerank`
+  - `v1/rerank`, `v1/reranking`, `rerank`
   - `v1/audio/speech` ([#36](https://github.com/mostlygeek/llama-swap/issues/36))
   - `v1/audio/transcriptions` ([docs](https://github.com/mostlygeek/llama-swap/issues/41#issuecomment-2722637867))
 - ✅ llama-swap custom API endpoints
@@ -70,9 +70,10 @@ See the [configuration documentation](https://github.com/mostlygeek/llama-swap/w
 
 ## Web UI
 
-llama-swap ships with a web based interface to make it easier to monitor logs and check the status of models. 
+llama-swap ships with a real time web interface to monitor logs and status of models:
+
+<img width="1786" height="1334" alt="image" src="https://github.com/user-attachments/assets/d6258cb9-1dad-40db-828f-2be860aec8fe" />
 
-<img width="1758" alt="image" src="https://github.com/user-attachments/assets/31ae5bcd-5efd-46b0-b64b-6db9e60196d3" />
 
 ## Docker Install ([download images](https://github.com/mostlygeek/llama-swap/pkgs/container/llama-swap))
 
@@ -173,6 +174,13 @@ Any OpenAI compatible server would work. llama-swap was originally designed for
 
 For Python based inference servers like vllm or tabbyAPI it is recommended to run them via podman or docker. This provides clean environment isolation as well as responding correctly to `SIGTERM` signals to shutdown.
 
+## Contributors 
+<a href="https://github.com/mostlygeek/llama-swap/graphs/contributors">
+  <img src="https://contrib.rocks/image?repo=mostlygeek/llama-swap" />
+</a>
+
+Made with [contrib.rocks](https://contrib.rocks).
+
 ## Star History
 
 [![Star History Chart](https://api.star-history.com/svg?repos=mostlygeek/llama-swap&type=Date)](https://www.star-history.com/#mostlygeek/llama-swap&Date)

config.example.yaml

@@ -15,6 +15,12 @@ healthCheckTimeout: 500
 # - Valid log levels: debug, info, warn, error
 logLevel: info
 
+# metricsMaxInMemory: maximum number of metrics to keep in memory
+# - optional, default: 1000
+# - controls how many metrics are stored in memory before older ones are discarded
+# - useful for limiting memory usage when processing large volumes of metrics
+metricsMaxInMemory: 1000
+
 # startPort: sets the starting port number for the automatic ${PORT} macro.
 # - optional, default: 5800
 # - the ${PORT} macro can be used in model.cmd and model.proxy settings
@@ -200,4 +206,4 @@ groups:
     members:
       - "forever-modelA"
       - "forever-modelB"
-      - "forever-modelc"
+      - "forever-modelc"

docker/config.example.yaml

@@ -1,5 +1,6 @@
 healthCheckTimeout: 300
 logRequests: true
+metricsMaxInMemory: 1000
 
 models:
   "qwen2.5":

event/README.md

@@ -0,0 +1,3 @@
+The code in `event` was originally a part of https://github.com/kelindar/event (v1.5.2)
+
+The original code uses a `time.Ticker` to process the event queue which caused a large increase in CPU usage ([#189](https://github.com/mostlygeek/llama-swap/issues/189)). This code was ported to remove the ticker and instead be more event driven.

event/default.go

@@ -0,0 +1,30 @@
+// Copyright (c) Roman Atachiants and contributore. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for detaile.
+
+package event
+
+import (
+	"context"
+)
+
+// Default initializes a default in-process dispatcher
+var Default = NewDispatcherConfig(25000)
+
+// On subscribes to an event, the type of the event will be automatically
+// inferred from the provided type. Must be constant for this to work. This
+// functions same way as Subscribe() but uses the default dispatcher instead.
+func On[T Event](handler func(T)) context.CancelFunc {
+	return Subscribe(Default, handler)
+}
+
+// OnType subscribes to an event with the specified event type. This functions
+// same way as SubscribeTo() but uses the default dispatcher instead.
+func OnType[T Event](eventType uint32, handler func(T)) context.CancelFunc {
+	return SubscribeTo(Default, eventType, handler)
+}
+
+// Emit writes an event into the dispatcher. This functions same way as
+// Publish() but uses the default dispatcher instead.
+func Emit[T Event](ev T) {
+	Publish(Default, ev)
+}

event/default_test.go

@@ -0,0 +1,54 @@
+// Copyright (c) Roman Atachiants and contributore. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for detaile.
+
+package event
+
+import (
+	"sync"
+	"sync/atomic"
+	"testing"
+
+	"github.com/stretchr/testify/assert"
+)
+
+/*
+cpu: 13th Gen Intel(R) Core(TM) i7-13700K
+BenchmarkSubcribeConcurrent-24    	 1826686	       606.3 ns/op	    1648 B/op	       5 allocs/op
+*/
+func BenchmarkSubscribeConcurrent(b *testing.B) {
+	d := NewDispatcher()
+	b.ReportAllocs()
+	b.ResetTimer()
+
+	b.RunParallel(func(pb *testing.PB) {
+		for pb.Next() {
+			unsub := Subscribe(d, func(ev MyEvent1) {})
+			unsub()
+		}
+	})
+}
+
+func TestDefaultPublish(t *testing.T) {
+	var wg sync.WaitGroup
+
+	// Subscribe
+	var count int64
+	defer On(func(ev MyEvent1) {
+		atomic.AddInt64(&count, 1)
+		wg.Done()
+	})()
+
+	defer OnType(TypeEvent1, func(ev MyEvent1) {
+		atomic.AddInt64(&count, 1)
+		wg.Done()
+	})()
+
+	// Publish
+	wg.Add(4)
+	Emit(MyEvent1{})
+	Emit(MyEvent1{})
+
+	// Wait and check
+	wg.Wait()
+	assert.Equal(t, int64(4), count)
+}

event/event.go

@@ -0,0 +1,324 @@
+// Copyright (c) Roman Atachiants and contributors. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for details.
+
+package event
+
+import (
+	"context"
+	"fmt"
+	"reflect"
+	"sort"
+	"strings"
+	"sync"
+	"sync/atomic"
+)
+
+// Event represents an event contract
+type Event interface {
+	Type() uint32
+}
+
+// registry holds an immutable sorted array of event mappings
+type registry struct {
+	keys []uint32 // Event types (sorted)
+	grps []any    // Corresponding subscribers
+}
+
+// ------------------------------------- Dispatcher -------------------------------------
+
+// Dispatcher represents an event dispatcher.
+type Dispatcher struct {
+	subs     atomic.Pointer[registry] // Atomic pointer to immutable array
+	done     chan struct{}            // Cancellation
+	maxQueue int                      // Maximum queue size per consumer
+	mu       sync.Mutex               // Only for writes (subscribe/unsubscribe)
+}
+
+// NewDispatcher creates a new dispatcher of events.
+func NewDispatcher() *Dispatcher {
+	return NewDispatcherConfig(50000)
+}
+
+// NewDispatcherConfig creates a new dispatcher with configurable max queue size
+func NewDispatcherConfig(maxQueue int) *Dispatcher {
+	d := &Dispatcher{
+		done:     make(chan struct{}),
+		maxQueue: maxQueue,
+	}
+
+	d.subs.Store(&registry{
+		keys: make([]uint32, 0, 16),
+		grps: make([]any, 0, 16),
+	})
+	return d
+}
+
+// Close closes the dispatcher
+func (d *Dispatcher) Close() error {
+	close(d.done)
+	return nil
+}
+
+// isClosed returns whether the dispatcher is closed or not
+func (d *Dispatcher) isClosed() bool {
+	select {
+	case <-d.done:
+		return true
+	default:
+		return false
+	}
+}
+
+// findGroup performs a lock-free binary search for the event type
+func (d *Dispatcher) findGroup(eventType uint32) any {
+	reg := d.subs.Load()
+	keys := reg.keys
+
+	// Inlined binary search for better cache locality
+	left, right := 0, len(keys)
+	for left < right {
+		mid := left + (right-left)/2
+		if keys[mid] < eventType {
+			left = mid + 1
+		} else {
+			right = mid
+		}
+	}
+
+	if left < len(keys) && keys[left] == eventType {
+		return reg.grps[left]
+	}
+	return nil
+}
+
+// Subscribe subscribes to an event, the type of the event will be automatically
+// inferred from the provided type. Must be constant for this to work.
+func Subscribe[T Event](broker *Dispatcher, handler func(T)) context.CancelFunc {
+	var event T
+	return SubscribeTo(broker, event.Type(), handler)
+}
+
+// SubscribeTo subscribes to an event with the specified event type.
+func SubscribeTo[T Event](broker *Dispatcher, eventType uint32, handler func(T)) context.CancelFunc {
+	if broker.isClosed() {
+		panic(errClosed)
+	}
+
+	broker.mu.Lock()
+	defer broker.mu.Unlock()
+
+	// Check if group already exists
+	if existing := broker.findGroup(eventType); existing != nil {
+		grp := groupOf[T](eventType, existing)
+		sub := grp.Add(handler)
+		return func() {
+			grp.Del(sub)
+		}
+	}
+
+	// Create new group
+	grp := &group[T]{cond: sync.NewCond(new(sync.Mutex)), maxQueue: broker.maxQueue}
+	sub := grp.Add(handler)
+
+	// Copy-on-write: insert new entry in sorted position
+	old := broker.subs.Load()
+	idx := sort.Search(len(old.keys), func(i int) bool {
+		return old.keys[i] >= eventType
+	})
+
+	// Create new arrays with space for one more element
+	newKeys := make([]uint32, len(old.keys)+1)
+	newGrps := make([]any, len(old.grps)+1)
+
+	// Copy elements before insertion point
+	copy(newKeys[:idx], old.keys[:idx])
+	copy(newGrps[:idx], old.grps[:idx])
+
+	// Insert new element
+	newKeys[idx] = eventType
+	newGrps[idx] = grp
+
+	// Copy elements after insertion point
+	copy(newKeys[idx+1:], old.keys[idx:])
+	copy(newGrps[idx+1:], old.grps[idx:])
+
+	// Atomically store the new registry (mutex ensures no concurrent writers)
+	newReg := &registry{keys: newKeys, grps: newGrps}
+	broker.subs.Store(newReg)
+
+	return func() {
+		grp.Del(sub)
+	}
+}
+
+// Publish writes an event into the dispatcher
+func Publish[T Event](broker *Dispatcher, ev T) {
+	eventType := ev.Type()
+	if sub := broker.findGroup(eventType); sub != nil {
+		group := groupOf[T](eventType, sub)
+		group.Broadcast(ev)
+	}
+}
+
+// Count counts the number of subscribers, this is for testing only.
+func (d *Dispatcher) count(eventType uint32) int {
+	if group := d.findGroup(eventType); group != nil {
+		return group.(interface{ Count() int }).Count()
+	}
+	return 0
+}
+
+// groupOf casts the subscriber group to the specified generic type
+func groupOf[T Event](eventType uint32, subs any) *group[T] {
+	if group, ok := subs.(*group[T]); ok {
+		return group
+	}
+
+	panic(errConflict[T](eventType, subs))
+}
+
+// ------------------------------------- Subscriber -------------------------------------
+
+// consumer represents a consumer with a message queue
+type consumer[T Event] struct {
+	queue []T  // Current work queue
+	stop  bool // Stop signal
+}
+
+// Listen listens to the event queue and processes events
+func (s *consumer[T]) Listen(c *sync.Cond, fn func(T)) {
+	pending := make([]T, 0, 128)
+
+	for {
+		c.L.Lock()
+		for len(s.queue) == 0 {
+			switch {
+			case s.stop:
+				c.L.Unlock()
+				return
+			default:
+				c.Wait()
+			}
+		}
+
+		// Swap buffers and reset the current queue
+		temp := s.queue
+		s.queue = pending[:0]
+		pending = temp
+		c.L.Unlock()
+
+		// Outside of the critical section, process the work
+		for _, event := range pending {
+			fn(event)
+		}
+
+		// Notify potential publishers waiting due to backpressure
+		c.Broadcast()
+	}
+}
+
+// ------------------------------------- Subscriber Group -------------------------------------
+
+// group represents a consumer group
+type group[T Event] struct {
+	cond     *sync.Cond
+	subs     []*consumer[T]
+	maxQueue int // Maximum queue size per consumer
+	maxLen   int // Current maximum queue length across all consumers
+}
+
+// Broadcast sends an event to all consumers
+func (s *group[T]) Broadcast(ev T) {
+	s.cond.L.Lock()
+	defer s.cond.L.Unlock()
+
+	// Calculate current maximum queue length
+	s.maxLen = 0
+	for _, sub := range s.subs {
+		if len(sub.queue) > s.maxLen {
+			s.maxLen = len(sub.queue)
+		}
+	}
+
+	// Backpressure: wait if queues are full
+	for s.maxLen >= s.maxQueue {
+		s.cond.Wait()
+
+		// Recalculate after wakeup
+		s.maxLen = 0
+		for _, sub := range s.subs {
+			if len(sub.queue) > s.maxLen {
+				s.maxLen = len(sub.queue)
+			}
+		}
+	}
+
+	// Add event to all queues and track new maximum
+	newMax := 0
+	for _, sub := range s.subs {
+		sub.queue = append(sub.queue, ev)
+		if len(sub.queue) > newMax {
+			newMax = len(sub.queue)
+		}
+	}
+	s.maxLen = newMax
+	s.cond.Broadcast() // Wake consumers
+}
+
+// Add adds a subscriber to the list
+func (s *group[T]) Add(handler func(T)) *consumer[T] {
+	sub := &consumer[T]{
+		queue: make([]T, 0, 64),
+	}
+
+	// Add the consumer to the list of active consumers
+	s.cond.L.Lock()
+	s.subs = append(s.subs, sub)
+	s.cond.L.Unlock()
+
+	// Start listening
+	go sub.Listen(s.cond, handler)
+	return sub
+}
+
+// Del removes a subscriber from the list
+func (s *group[T]) Del(sub *consumer[T]) {
+	s.cond.L.Lock()
+	defer s.cond.L.Unlock()
+
+	// Search and remove the subscriber
+	sub.stop = true
+	for i, v := range s.subs {
+		if v == sub {
+			copy(s.subs[i:], s.subs[i+1:])
+			s.subs = s.subs[:len(s.subs)-1]
+			break
+		}
+	}
+}
+
+// ------------------------------------- Debugging -------------------------------------
+
+var errClosed = fmt.Errorf("event dispatcher is closed")
+
+// Count returns the number of subscribers in this group
+func (s *group[T]) Count() int {
+	return len(s.subs)
+}
+
+// String returns string representation of the type
+func (s *group[T]) String() string {
+	typ := reflect.TypeOf(s).String()
+	idx := strings.LastIndex(typ, "/")
+	typ = typ[idx+1 : len(typ)-1]
+	return typ
+}
+
+// errConflict returns a conflict message
+func errConflict[T any](eventType uint32, existing any) string {
+	var want T
+	return fmt.Sprintf(
+		"conflicting event type, want=<%T>, registered=<%s>, event=0x%v",
+		want, existing, eventType,
+	)
+}

event/event_test.go

@@ -0,0 +1,324 @@
+// Copyright (c) Roman Atachiants and contributore. All rights reserved.
+// Licensed under the MIT license. See LICENSE file in the project root for detaile.
+
+package event
+
+import (
+	"fmt"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/stretchr/testify/assert"
+)
+
+func TestPublish(t *testing.T) {
+	d := NewDispatcher()
+	var wg sync.WaitGroup
+
+	// Subscribe, must be received in order
+	var count int64
+	defer Subscribe(d, func(ev MyEvent1) {
+		assert.Equal(t, int(atomic.AddInt64(&count, 1)), ev.Number)
+		wg.Done()
+	})()
+
+	// Publish
+	wg.Add(3)
+	Publish(d, MyEvent1{Number: 1})
+	Publish(d, MyEvent1{Number: 2})
+	Publish(d, MyEvent1{Number: 3})
+
+	// Wait and check
+	wg.Wait()
+	assert.Equal(t, int64(3), count)
+}
+
+func TestUnsubscribe(t *testing.T) {
+	d := NewDispatcher()
+	assert.Equal(t, 0, d.count(TypeEvent1))
+	unsubscribe := Subscribe(d, func(ev MyEvent1) {
+		// Nothing
+	})
+
+	assert.Equal(t, 1, d.count(TypeEvent1))
+	unsubscribe()
+	assert.Equal(t, 0, d.count(TypeEvent1))
+}
+
+func TestConcurrent(t *testing.T) {
+	const max = 1000000
+	var count int64
+	var wg sync.WaitGroup
+	wg.Add(1)
+
+	d := NewDispatcher()
+	defer Subscribe(d, func(ev MyEvent1) {
+		if current := atomic.AddInt64(&count, 1); current == max {
+			wg.Done()
+		}
+	})()
+
+	// Asynchronously publish
+	go func() {
+		for i := 0; i < max; i++ {
+			Publish(d, MyEvent1{})
+		}
+	}()
+
+	defer Subscribe(d, func(ev MyEvent1) {
+		// Subscriber that does nothing
+	})()
+
+	wg.Wait()
+	assert.Equal(t, max, int(count))
+}
+
+func TestSubscribeDifferentType(t *testing.T) {
+	d := NewDispatcher()
+	assert.Panics(t, func() {
+		SubscribeTo(d, TypeEvent1, func(ev MyEvent1) {})
+		SubscribeTo(d, TypeEvent1, func(ev MyEvent2) {})
+	})
+}
+
+func TestPublishDifferentType(t *testing.T) {
+	d := NewDispatcher()
+	assert.Panics(t, func() {
+		SubscribeTo(d, TypeEvent1, func(ev MyEvent2) {})
+		Publish(d, MyEvent1{})
+	})
+}
+
+func TestCloseDispatcher(t *testing.T) {
+	d := NewDispatcher()
+	defer SubscribeTo(d, TypeEvent1, func(ev MyEvent2) {})()
+
+	assert.NoError(t, d.Close())
+	assert.Panics(t, func() {
+		SubscribeTo(d, TypeEvent1, func(ev MyEvent2) {})
+	})
+}
+
+func TestMatrix(t *testing.T) {
+	const amount = 1000
+	for _, subs := range []int{1, 10, 100} {
+		for _, topics := range []int{1, 10} {
+			expected := subs * topics * amount
+			t.Run(fmt.Sprintf("%dx%d", topics, subs), func(t *testing.T) {
+				var count atomic.Int64
+				var wg sync.WaitGroup
+				wg.Add(expected)
+
+				d := NewDispatcher()
+				for i := 0; i < subs; i++ {
+					for id := 0; id < topics; id++ {
+						defer SubscribeTo(d, uint32(id), func(ev MyEvent3) {
+							count.Add(1)
+							wg.Done()
+						})()
+					}
+				}
+
+				for n := 0; n < amount; n++ {
+					for id := 0; id < topics; id++ {
+						go Publish(d, MyEvent3{ID: id})
+					}
+				}
+
+				wg.Wait()
+				assert.Equal(t, expected, int(count.Load()))
+			})
+		}
+	}
+}
+
+func TestConcurrentSubscriptionRace(t *testing.T) {
+	// This test specifically targets the race condition that occurs when multiple
+	// goroutines try to subscribe to different event types simultaneously.
+	// Without the CAS loop, subscriptions could be lost due to registry corruption.
+
+	const numGoroutines = 100
+	const numEventTypes = 50
+
+	d := NewDispatcher()
+	defer d.Close()
+
+	var wg sync.WaitGroup
+	var receivedCount int64
+	var subscribedTypes sync.Map // Thread-safe map
+
+	wg.Add(numGoroutines)
+
+	// Start multiple goroutines that subscribe to different event types concurrently
+	for i := 0; i < numGoroutines; i++ {
+		go func(goroutineID int) {
+			defer wg.Done()
+
+			// Each goroutine subscribes to a unique event type
+			eventType := uint32(goroutineID%numEventTypes + 1000) // Offset to avoid collision with other tests
+
+			// Subscribe to the event type
+			SubscribeTo(d, eventType, func(ev MyEvent3) {
+				atomic.AddInt64(&receivedCount, 1)
+			})
+
+			// Record that this type was subscribed
+			subscribedTypes.Store(eventType, true)
+		}(i)
+	}
+
+	// Wait for all subscriptions to complete
+	wg.Wait()
+
+	// Count the number of unique event types subscribed
+	expectedTypes := 0
+	subscribedTypes.Range(func(key, value interface{}) bool {
+		expectedTypes++
+		return true
+	})
+
+	// Small delay to ensure all subscriptions are fully processed
+	time.Sleep(10 * time.Millisecond)
+
+	// Publish events to each subscribed type
+	subscribedTypes.Range(func(key, value interface{}) bool {
+		eventType := key.(uint32)
+		Publish(d, MyEvent3{ID: int(eventType)})
+		return true
+	})
+
+	// Wait for all events to be processed
+	time.Sleep(50 * time.Millisecond)
+
+	// Verify that we received at least the expected number of events
+	// (there might be more if multiple goroutines subscribed to the same event type)
+	received := atomic.LoadInt64(&receivedCount)
+	assert.GreaterOrEqual(t, int(received), expectedTypes,
+		"Should have received at least %d events, got %d", expectedTypes, received)
+
+	// Verify that we have the expected number of unique event types
+	assert.Equal(t, numEventTypes, expectedTypes,
+		"Should have exactly %d unique event types", numEventTypes)
+}
+
+func TestConcurrentHandlerRegistration(t *testing.T) {
+	const numGoroutines = 100
+
+	// Test concurrent subscriptions to the same event type
+	t.Run("SameEventType", func(t *testing.T) {
+		d := NewDispatcher()
+		var handlerCount int64
+		var wg sync.WaitGroup
+
+		// Start multiple goroutines subscribing to the same event type (0x1)
+		for i := 0; i < numGoroutines; i++ {
+			wg.Add(1)
+			go func() {
+				defer wg.Done()
+				SubscribeTo(d, uint32(0x1), func(ev MyEvent1) {
+					atomic.AddInt64(&handlerCount, 1)
+				})
+			}()
+		}
+
+		wg.Wait()
+
+		// Verify all handlers were registered by publishing an event
+		atomic.StoreInt64(&handlerCount, 0)
+		Publish(d, MyEvent1{})
+
+		// Small delay to ensure all handlers have executed
+		time.Sleep(10 * time.Millisecond)
+
+		assert.Equal(t, int64(numGoroutines), atomic.LoadInt64(&handlerCount),
+			"Not all handlers were registered due to race condition")
+	})
+
+	// Test concurrent subscriptions to different event types
+	t.Run("DifferentEventTypes", func(t *testing.T) {
+		d := NewDispatcher()
+		var wg sync.WaitGroup
+		receivedEvents := make(map[uint32]*int64)
+
+		// Create multiple event types and subscribe concurrently
+		for i := 0; i < numGoroutines; i++ {
+			eventType := uint32(100 + i)
+			counter := new(int64)
+			receivedEvents[eventType] = counter
+
+			wg.Add(1)
+			go func(et uint32, cnt *int64) {
+				defer wg.Done()
+				SubscribeTo(d, et, func(ev MyEvent3) {
+					atomic.AddInt64(cnt, 1)
+				})
+			}(eventType, counter)
+		}
+
+		wg.Wait()
+
+		// Publish events to all types
+		for eventType := uint32(100); eventType < uint32(100+numGoroutines); eventType++ {
+			Publish(d, MyEvent3{ID: int(eventType)})
+		}
+
+		// Small delay to ensure all handlers have executed
+		time.Sleep(10 * time.Millisecond)
+
+		// Verify all event types received their events
+		for eventType, counter := range receivedEvents {
+			assert.Equal(t, int64(1), atomic.LoadInt64(counter),
+				"Event type %d did not receive its event", eventType)
+		}
+	})
+}
+
+func TestBackpressure(t *testing.T) {
+	d := NewDispatcher()
+	d.maxQueue = 10
+
+	var processedCount int64
+	unsub := SubscribeTo(d, uint32(0x200), func(ev MyEvent3) {
+		atomic.AddInt64(&processedCount, 1)
+	})
+	defer unsub()
+
+	const eventsToPublish = 1000
+	for i := 0; i < eventsToPublish; i++ {
+		Publish(d, MyEvent3{ID: 0x200})
+	}
+
+	time.Sleep(100 * time.Millisecond)
+
+	// Verify all events were eventually processed
+	finalProcessed := atomic.LoadInt64(&processedCount)
+	assert.Equal(t, int64(eventsToPublish), finalProcessed)
+	t.Logf("Events processed: %d/%d", finalProcessed, eventsToPublish)
+}
+
+// ------------------------------------- Test Events -------------------------------------
+
+const (
+	TypeEvent1 = 0x1
+	TypeEvent2 = 0x2
+)
+
+type MyEvent1 struct {
+	Number int
+}
+
+func (t MyEvent1) Type() uint32 { return TypeEvent1 }
+
+type MyEvent2 struct {
+	Text string
+}
+
+func (t MyEvent2) Type() uint32 { return TypeEvent2 }
+
+type MyEvent3 struct {
+	ID int
+}
+
+func (t MyEvent3) Type() uint32 { return uint32(t.ID) }

go.mod

@@ -3,17 +3,16 @@ module github.com/mostlygeek/llama-swap
 go 1.23.0
 
 require (
+	github.com/billziss-gh/golib v0.2.0
 	github.com/fsnotify/fsnotify v1.9.0
 	github.com/gin-gonic/gin v1.10.0
 	github.com/stretchr/testify v1.9.0
 	github.com/tidwall/gjson v1.18.0
 	github.com/tidwall/sjson v1.2.5
 	gopkg.in/yaml.v3 v3.0.1
-	github.com/kelindar/event v1.5.2
 )
 
 require (
-	github.com/billziss-gh/golib v0.2.0 // indirect
 	github.com/bytedance/sonic v1.11.6 // indirect
 	github.com/bytedance/sonic/loader v0.1.1 // indirect
 	github.com/cloudwego/base64x v0.1.4 // indirect

go.sum

@@ -32,12 +32,8 @@ github.com/goccy/go-json v0.10.2/go.mod h1:6MelG93GURQebXPDq3khkgXZkazVtN9CRI+MG
 github.com/google/go-cmp v0.5.5 h1:Khx7svrCpmxxtHBq5j2mp/xVjsi8hQMfNLvJFAlrGgU=
 github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
 github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
-github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510 h1:El6M4kTTCOh6aBiKaUGG7oYTSPP8MxqL4YI3kZKwcP4=
-github.com/google/shlex v0.0.0-20191202100458-e7afc7fbc510/go.mod h1:pupxD2MaaD3pAXIBCelhxNneeOaAeabZDe5s4K6zSpQ=
 github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
 github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
-github.com/kelindar/event v1.5.2 h1:qtgssZqMh/QQMCIxlbx4wU3DoMHOrJXKdiZhphJ4YbY=
-github.com/kelindar/event v1.5.2/go.mod h1:UxWPQjWK8u0o9Z3ponm2mgREimM95hm26/M9z8F488Q=
 github.com/klauspost/cpuid/v2 v2.0.9/go.mod h1:FInQzS24/EEf25PyTYn52gqo7WaD8xa0213Md/qVLRg=
 github.com/klauspost/cpuid/v2 v2.2.7 h1:ZWSB3igEs+d0qvnxR/ZBzXVmxkgt8DdzP6m9pfuVLDM=
 github.com/klauspost/cpuid/v2 v2.2.7/go.mod h1:Lcz8mBdAVJIBVzewtcLocK12l3Y+JytZYpaMropDUws=

llama-swap.go

@@ -14,7 +14,7 @@ import (
 
 	"github.com/fsnotify/fsnotify"
 	"github.com/gin-gonic/gin"
-	"github.com/kelindar/event"
+	"github.com/mostlygeek/llama-swap/event"
 	"github.com/mostlygeek/llama-swap/proxy"
 )
 

misc/simple-responder/simple-responder.go

@@ -35,20 +35,90 @@ func main() {
 
 	// Set up the handler function using the provided response message
 	r.POST("/v1/chat/completions", func(c *gin.Context) {
-		c.Header("Content-Type", "application/json")
-
-		// add a wait to simulate a slow query
-		if wait, err := time.ParseDuration(c.Query("wait")); err == nil {
-			time.Sleep(wait)
-		}
-
 		bodyBytes, _ := io.ReadAll(c.Request.Body)
 
-		c.JSON(http.StatusOK, gin.H{
-			"responseMessage":  *responseMessage,
-			"h_content_length": c.Request.Header.Get("Content-Length"),
-			"request_body":     string(bodyBytes),
-		})
+		// Check if streaming is requested
+		// Query is checked instead of JSON body since that event stream conflicts with other tests
+		isStreaming := c.Query("stream") == "true"
+
+		if isStreaming {
+			// Set headers for streaming
+			c.Header("Content-Type", "text/event-stream")
+			c.Header("Cache-Control", "no-cache")
+			c.Header("Connection", "keep-alive")
+			c.Header("Transfer-Encoding", "chunked")
+
+			// add a wait to simulate a slow query
+			if wait, err := time.ParseDuration(c.Query("wait")); err == nil {
+				time.Sleep(wait)
+			}
+
+			// Send 10 "asdf" tokens
+			for i := 0; i < 10; i++ {
+				data := gin.H{
+					"created": time.Now().Unix(),
+					"choices": []gin.H{
+						{
+							"index": 0,
+							"delta": gin.H{
+								"content": "asdf",
+							},
+							"finish_reason": nil,
+						},
+					},
+				}
+				c.SSEvent("message", data)
+				c.Writer.Flush()
+			}
+
+			// Send final data with usage info
+			finalData := gin.H{
+				"usage": gin.H{
+					"completion_tokens": 10,
+					"prompt_tokens":     25,
+					"total_tokens":      35,
+				},
+				// add timings to simulate llama.cpp
+				"timings": gin.H{
+					"prompt_n":             25,
+					"prompt_ms":            13,
+					"predicted_n":          10,
+					"predicted_ms":         17,
+					"predicted_per_second": 10,
+				},
+			}
+			c.SSEvent("message", finalData)
+			c.Writer.Flush()
+
+			// Send [DONE]
+			c.SSEvent("message", "[DONE]")
+			c.Writer.Flush()
+		} else {
+			c.Header("Content-Type", "application/json")
+
+			// add a wait to simulate a slow query
+			if wait, err := time.ParseDuration(c.Query("wait")); err == nil {
+				time.Sleep(wait)
+			}
+
+			c.JSON(http.StatusOK, gin.H{
+				"responseMessage":  *responseMessage,
+				"h_content_length": c.Request.Header.Get("Content-Length"),
+				"request_body":     string(bodyBytes),
+				"usage": gin.H{
+					"completion_tokens": 10,
+					"prompt_tokens":     25,
+					"total_tokens":      35,
+				},
+				"timings": gin.H{
+					"prompt_n":             25,
+					"prompt_ms":            13,
+					"predicted_n":          10,
+					"predicted_ms":         17,
+					"predicted_per_second": 10,
+				},
+			})
+		}
 	})
 
 	// for issue #62 to check model name strips profile slug
@@ -74,6 +144,11 @@ func main() {
 		c.Header("Content-Type", "application/json")
 		c.JSON(http.StatusOK, gin.H{
 			"responseMessage": *responseMessage,
+			"usage": gin.H{
+				"completion_tokens": 10,
+				"prompt_tokens":     25,
+				"total_tokens":      35,
+			},
 		})
 
 	})

proxy/config.go

@@ -142,6 +142,7 @@ type Config struct {
 	HealthCheckTimeout int                    `yaml:"healthCheckTimeout"`
 	LogRequests        bool                   `yaml:"logRequests"`
 	LogLevel           string                 `yaml:"logLevel"`
+	MetricsMaxInMemory int                    `yaml:"metricsMaxInMemory"`
 	Models             map[string]ModelConfig `yaml:"models"` /* key is model ID */
 	Profiles           map[string][]string    `yaml:"profiles"`
 	Groups             map[string]GroupConfig `yaml:"groups"` /* key is group ID */
@@ -194,6 +195,7 @@ func LoadConfigFromReader(r io.Reader) (Config, error) {
 		HealthCheckTimeout: 120,
 		StartPort:          5800,
 		LogLevel:           "info",
+		MetricsMaxInMemory: 1000,
 	}
 	err = yaml.Unmarshal(data, &config)
 	if err != nil {

proxy/config_posix_test.go

@@ -196,6 +196,7 @@ groups:
 			},
 		},
 		HealthCheckTimeout: 15,
+		MetricsMaxInMemory: 1000,
 		Profiles: map[string][]string{
 			"test": {"model1", "model2"},
 		},

proxy/config_windows_test.go

@@ -193,6 +193,7 @@ groups:
 			},
 		},
 		HealthCheckTimeout: 15,
+		MetricsMaxInMemory: 1000,
 		Profiles: map[string][]string{
 			"test": {"model1", "model2"},
 		},

proxy/events.go

@@ -6,6 +6,7 @@ const ProcessStateChangeEventID = 0x01
 const ChatCompletionStatsEventID = 0x02
 const ConfigFileChangedEventID = 0x03
 const LogDataEventID = 0x04
+const TokenMetricsEventID = 0x05
 
 type ProcessStateChangeEvent struct {
 	ProcessName string

proxy/logMonitor.go

@@ -8,7 +8,7 @@ import (
 	"os"
 	"sync"
 
-	"github.com/kelindar/event"
+	"github.com/mostlygeek/llama-swap/event"
 )
 
 type LogLevel int
@@ -40,7 +40,7 @@ func NewLogMonitor() *LogMonitor {
 
 func NewLogMonitorWriter(stdout io.Writer) *LogMonitor {
 	return &LogMonitor{
-		eventbus: event.NewDispatcher(),
+		eventbus: event.NewDispatcherConfig(1000),
 		buffer:   ring.New(10 * 1024), // keep 10KB of buffered logs
 		stdout:   stdout,
 		level:    LevelInfo,

proxy/metrics_middleware.go

@@ -0,0 +1,168 @@
+package proxy
+
+import (
+	"bytes"
+	"fmt"
+	"io"
+	"net/http"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/tidwall/gjson"
+)
+
+// MetricsMiddleware sets up the MetricsResponseWriter for capturing upstream requests
+func MetricsMiddleware(pm *ProxyManager) gin.HandlerFunc {
+	return func(c *gin.Context) {
+		bodyBytes, err := io.ReadAll(c.Request.Body)
+		if err != nil {
+			pm.sendErrorResponse(c, http.StatusBadRequest, "could not ready request body")
+			return
+		}
+		c.Request.Body = io.NopCloser(bytes.NewBuffer(bodyBytes))
+
+		requestedModel := gjson.GetBytes(bodyBytes, "model").String()
+		if requestedModel == "" {
+			pm.sendErrorResponse(c, http.StatusBadRequest, "missing or invalid 'model' key")
+			return
+		}
+
+		realModelName, found := pm.config.RealModelName(requestedModel)
+		if !found {
+			pm.sendErrorResponse(c, http.StatusBadRequest, fmt.Sprintf("could not find real modelID for %s", requestedModel))
+			return
+		}
+		c.Set("ls-real-model-name", realModelName)
+
+		writer := &MetricsResponseWriter{
+			ResponseWriter: c.Writer,
+			metricsRecorder: &MetricsRecorder{
+				metricsMonitor: pm.metricsMonitor,
+				realModelName:  realModelName,
+				isStreaming:    gjson.GetBytes(bodyBytes, "stream").Bool(),
+				startTime:      time.Now(),
+			},
+		}
+		c.Writer = writer
+		c.Next()
+
+		rec := writer.metricsRecorder
+		rec.processBody(writer.body)
+	}
+}
+
+type MetricsRecorder struct {
+	metricsMonitor *MetricsMonitor
+	realModelName  string
+	isStreaming    bool
+	startTime      time.Time
+}
+
+// processBody handles response processing after request completes
+func (rec *MetricsRecorder) processBody(body []byte) {
+	if rec.isStreaming {
+		rec.processStreamingResponse(body)
+	} else {
+		rec.processNonStreamingResponse(body)
+	}
+}
+
+func (rec *MetricsRecorder) parseAndRecordMetrics(jsonData gjson.Result) bool {
+	usage := jsonData.Get("usage")
+	if !usage.Exists() {
+		return false
+	}
+
+	// default values
+	outputTokens := int(jsonData.Get("usage.completion_tokens").Int())
+	inputTokens := int(jsonData.Get("usage.prompt_tokens").Int())
+	tokensPerSecond := -1.0
+	durationMs := int(time.Since(rec.startTime).Milliseconds())
+
+	// use llama-server's timing data for tok/sec and duration as it is more accurate
+	if timings := jsonData.Get("timings"); timings.Exists() {
+		tokensPerSecond = jsonData.Get("timings.predicted_per_second").Float()
+		durationMs = int(jsonData.Get("timings.prompt_ms").Float() + jsonData.Get("timings.predicted_ms").Float())
+	}
+
+	rec.metricsMonitor.addMetrics(TokenMetrics{
+		Timestamp:       time.Now(),
+		Model:           rec.realModelName,
+		InputTokens:     inputTokens,
+		OutputTokens:    outputTokens,
+		TokensPerSecond: tokensPerSecond,
+		DurationMs:      durationMs,
+	})
+
+	return true
+}
+
+func (rec *MetricsRecorder) processStreamingResponse(body []byte) {
+	// Iterate **backwards** through the lines looking for the data payload with
+	// usage data
+	lines := bytes.Split(body, []byte("\n"))
+
+	for i := len(lines) - 1; i >= 0; i-- {
+		line := bytes.TrimSpace(lines[i])
+		if len(line) == 0 {
+			continue
+		}
+
+		// SSE payload always follows "data:"
+		prefix := []byte("data:")
+		if !bytes.HasPrefix(line, prefix) {
+			continue
+		}
+		data := bytes.TrimSpace(line[len(prefix):])
+
+		if len(data) == 0 {
+			continue
+		}
+
+		if bytes.Equal(data, []byte("[DONE]")) {
+			// [DONE] line itself contains nothing of interest.
+			continue
+		}
+
+		if gjson.ValidBytes(data) {
+			if rec.parseAndRecordMetrics(gjson.ParseBytes(data)) {
+				return // short circuit if a metric was recorded
+			}
+		}
+	}
+}
+
+func (rec *MetricsRecorder) processNonStreamingResponse(body []byte) {
+	if len(body) == 0 {
+		return
+	}
+
+	// Parse JSON to extract usage information
+	if gjson.ValidBytes(body) {
+		rec.parseAndRecordMetrics(gjson.ParseBytes(body))
+	}
+}
+
+// MetricsResponseWriter captures the entire response for non-streaming
+type MetricsResponseWriter struct {
+	gin.ResponseWriter
+	body            []byte
+	metricsRecorder *MetricsRecorder
+}
+
+func (w *MetricsResponseWriter) Write(b []byte) (int, error) {
+	n, err := w.ResponseWriter.Write(b)
+	if err != nil {
+		return n, err
+	}
+	w.body = append(w.body, b...)
+	return n, nil
+}
+
+func (w *MetricsResponseWriter) WriteHeader(statusCode int) {
+	w.ResponseWriter.WriteHeader(statusCode)
+}
+
+func (w *MetricsResponseWriter) Header() http.Header {
+	return w.ResponseWriter.Header()
+}

proxy/metrics_monitor.go

@@ -0,0 +1,82 @@
+package proxy
+
+import (
+	"encoding/json"
+	"sync"
+	"time"
+
+	"github.com/mostlygeek/llama-swap/event"
+)
+
+// TokenMetrics represents parsed token statistics from llama-server logs
+type TokenMetrics struct {
+	ID              int       `json:"id"`
+	Timestamp       time.Time `json:"timestamp"`
+	Model           string    `json:"model"`
+	InputTokens     int       `json:"input_tokens"`
+	OutputTokens    int       `json:"output_tokens"`
+	TokensPerSecond float64   `json:"tokens_per_second"`
+	DurationMs      int       `json:"duration_ms"`
+}
+
+// TokenMetricsEvent represents a token metrics event
+type TokenMetricsEvent struct {
+	Metrics TokenMetrics
+}
+
+func (e TokenMetricsEvent) Type() uint32 {
+	return TokenMetricsEventID // defined in events.go
+}
+
+// MetricsMonitor parses llama-server output for token statistics
+type MetricsMonitor struct {
+	mu         sync.RWMutex
+	metrics    []TokenMetrics
+	maxMetrics int
+	nextID     int
+}
+
+func NewMetricsMonitor(config *Config) *MetricsMonitor {
+	maxMetrics := config.MetricsMaxInMemory
+	if maxMetrics <= 0 {
+		maxMetrics = 1000 // Default fallback
+	}
+
+	mp := &MetricsMonitor{
+		maxMetrics: maxMetrics,
+	}
+
+	return mp
+}
+
+// addMetrics adds a new metric to the collection and publishes an event
+func (mp *MetricsMonitor) addMetrics(metric TokenMetrics) {
+	mp.mu.Lock()
+	defer mp.mu.Unlock()
+
+	metric.ID = mp.nextID
+	mp.nextID++
+	mp.metrics = append(mp.metrics, metric)
+	if len(mp.metrics) > mp.maxMetrics {
+		mp.metrics = mp.metrics[len(mp.metrics)-mp.maxMetrics:]
+	}
+
+	event.Emit(TokenMetricsEvent{Metrics: metric})
+}
+
+// GetMetrics returns a copy of the current metrics
+func (mp *MetricsMonitor) GetMetrics() []TokenMetrics {
+	mp.mu.RLock()
+	defer mp.mu.RUnlock()
+
+	result := make([]TokenMetrics, len(mp.metrics))
+	copy(result, mp.metrics)
+	return result
+}
+
+// GetMetricsJSON returns metrics as JSON
+func (mp *MetricsMonitor) GetMetricsJSON() ([]byte, error) {
+	mp.mu.RLock()
+	defer mp.mu.RUnlock()
+	return json.Marshal(mp.metrics)
+}

proxy/process.go

@@ -14,7 +14,7 @@ import (
 	"syscall"
 	"time"
 
-	"github.com/kelindar/event"
+	"github.com/mostlygeek/llama-swap/event"
 )
 
 type ProcessState string

proxy/proxymanager.go

@@ -14,7 +14,6 @@ import (
 	"time"
 
 	"github.com/gin-gonic/gin"
-	"github.com/tidwall/gjson"
 	"github.com/tidwall/sjson"
 )
 
@@ -33,6 +32,8 @@ type ProxyManager struct {
 	upstreamLogger *LogMonitor
 	muxLogger      *LogMonitor
 
+	metricsMonitor *MetricsMonitor
+
 	processGroups map[string]*ProcessGroup
 
 	// shutdown signaling
@@ -78,6 +79,8 @@ func New(config Config) *ProxyManager {
 		muxLogger:      stdoutLogger,
 		upstreamLogger: upstreamLogger,
 
+		metricsMonitor: NewMetricsMonitor(&config),
+
 		processGroups: make(map[string]*ProcessGroup),
 
 		shutdownCtx:    shutdownCtx,
@@ -149,14 +152,18 @@ func (pm *ProxyManager) setupGinEngine() {
 		c.Next()
 	})
 
+	mm := MetricsMiddleware(pm)
+
 	// Set up routes using the Gin engine
-	pm.ginEngine.POST("/v1/chat/completions", pm.proxyOAIHandler)
+	pm.ginEngine.POST("/v1/chat/completions", mm, pm.proxyOAIHandler)
 	// Support legacy /v1/completions api, see issue #12
-	pm.ginEngine.POST("/v1/completions", pm.proxyOAIHandler)
+	pm.ginEngine.POST("/v1/completions", mm, pm.proxyOAIHandler)
 
 	// Support embeddings
-	pm.ginEngine.POST("/v1/embeddings", pm.proxyOAIHandler)
-	pm.ginEngine.POST("/v1/rerank", pm.proxyOAIHandler)
+	pm.ginEngine.POST("/v1/embeddings", mm, pm.proxyOAIHandler)
+	pm.ginEngine.POST("/v1/rerank", mm, pm.proxyOAIHandler)
+	pm.ginEngine.POST("/v1/reranking", mm, pm.proxyOAIHandler)
+	pm.ginEngine.POST("/rerank", mm, pm.proxyOAIHandler)
 
 	// Support audio/speech endpoint
 	pm.ginEngine.POST("/v1/audio/speech", pm.proxyOAIHandler)
@@ -360,13 +367,13 @@ func (pm *ProxyManager) proxyOAIHandler(c *gin.Context) {
 		return
 	}
 
-	requestedModel := gjson.GetBytes(bodyBytes, "model").String()
-	if requestedModel == "" {
-		pm.sendErrorResponse(c, http.StatusBadRequest, "missing or invalid 'model' key")
+	realModelName := c.GetString("ls-real-model-name") // Should be set in MetricsMiddleware
+	if realModelName == "" {
+		pm.sendErrorResponse(c, http.StatusInternalServerError, "ls-real-model-name not set")
 		return
 	}
 
-	processGroup, realModelName, err := pm.swapProcessGroup(requestedModel)
+	processGroup, _, err := pm.swapProcessGroup(realModelName)
 	if err != nil {
 		pm.sendErrorResponse(c, http.StatusInternalServerError, fmt.Sprintf("error swapping process group: %s", err.Error()))
 		return

proxy/proxymanager_api.go

@@ -7,7 +7,7 @@ import (
 	"sort"
 
 	"github.com/gin-gonic/gin"
-	"github.com/kelindar/event"
+	"github.com/mostlygeek/llama-swap/event"
 )
 
 type Model struct {
@@ -24,6 +24,7 @@ func addApiHandlers(pm *ProxyManager) {
 	{
 		apiGroup.POST("/models/unload", pm.apiUnloadAllModels)
 		apiGroup.GET("/events", pm.apiSendEvents)
+		apiGroup.GET("/metrics", pm.apiGetMetrics)
 	}
 }
 
@@ -85,6 +86,7 @@ type messageType string
 const (
 	msgTypeModelStatus messageType = "modelStatus"
 	msgTypeLogData     messageType = "logData"
+	msgTypeMetrics     messageType = "metrics"
 )
 
 type messageEnvelope struct {
@@ -130,6 +132,18 @@ func (pm *ProxyManager) apiSendEvents(c *gin.Context) {
 		}
 	}
 
+	sendMetrics := func(metrics TokenMetrics) {
+		jsonData, err := json.Marshal(metrics)
+		if err == nil {
+			select {
+			case sendBuffer <- messageEnvelope{Type: msgTypeMetrics, Data: string(jsonData)}:
+			case <-ctx.Done():
+				return
+			default:
+			}
+		}
+	}
+
 	/**
 	 * Send updated models list
 	 */
@@ -150,10 +164,20 @@ func (pm *ProxyManager) apiSendEvents(c *gin.Context) {
 		sendLogData("upstream", data)
 	})()
 
+	/**
+	 * Send Metrics data
+	 */
+	defer event.On(func(e TokenMetricsEvent) {
+		sendMetrics(e.Metrics)
+	})()
+
 	// send initial batch of data
 	sendLogData("proxy", pm.proxyLogger.GetHistory())
 	sendLogData("upstream", pm.upstreamLogger.GetHistory())
 	sendModels()
+	for _, metrics := range pm.metricsMonitor.GetMetrics() {
+		sendMetrics(metrics)
+	}
 
 	for {
 		select {
@@ -169,3 +193,12 @@ func (pm *ProxyManager) apiSendEvents(c *gin.Context) {
 		}
 	}
 }
+
+func (pm *ProxyManager) apiGetMetrics(c *gin.Context) {
+	jsonData, err := pm.metricsMonitor.GetMetricsJSON()
+	if err != nil {
+		c.JSON(http.StatusInternalServerError, gin.H{"error": "failed to get metrics"})
+		return
+	}
+	c.Data(http.StatusOK, "application/json", jsonData)
+}

proxy/proxymanager_test.go

@@ -165,9 +165,11 @@ func TestProxyManager_SwapMultiProcessParallelRequests(t *testing.T) {
 			}
 
 			mu.Lock()
-			var response map[string]string
+			var response map[string]interface{}
 			assert.NoError(t, json.Unmarshal(w.Body.Bytes(), &response))
-			results[key] = response["responseMessage"]
+			result, ok := response["responseMessage"].(string)
+			assert.Equal(t, ok, true)
+			results[key] = result
 			mu.Unlock()
 		}(key)
 
@@ -644,7 +646,7 @@ func TestProxyManager_ChatContentLength(t *testing.T) {
 
 	proxy.ServeHTTP(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
-	var response map[string]string
+	var response map[string]interface{}
 	assert.NoError(t, json.Unmarshal(w.Body.Bytes(), &response))
 	assert.Equal(t, "81", response["h_content_length"])
 	assert.Equal(t, "model1", response["responseMessage"])
@@ -672,7 +674,7 @@ func TestProxyManager_FiltersStripParams(t *testing.T) {
 
 	proxy.ServeHTTP(w, req)
 	assert.Equal(t, http.StatusOK, w.Code)
-	var response map[string]string
+	var response map[string]interface{}
 	assert.NoError(t, json.Unmarshal(w.Body.Bytes(), &response))
 
 	// `temperature` and `stream` are gone but model remains
@@ -683,3 +685,73 @@ func TestProxyManager_FiltersStripParams(t *testing.T) {
 	// assert.Equal(t, "abc", response["y_param"])
 	// t.Logf("%v", response)
 }
+
+func TestProxyManager_MiddlewareWritesMetrics_NonStreaming(t *testing.T) {
+	config := AddDefaultGroupToConfig(Config{
+		HealthCheckTimeout: 15,
+		Models: map[string]ModelConfig{
+			"model1": getTestSimpleResponderConfig("model1"),
+		},
+		LogLevel: "error",
+	})
+
+	proxy := New(config)
+	defer proxy.StopProcesses(StopWaitForInflightRequest)
+
+	// Make a non-streaming request
+	reqBody := `{"model":"model1", "stream": false}`
+	req := httptest.NewRequest("POST", "/v1/chat/completions", bytes.NewBufferString(reqBody))
+	w := httptest.NewRecorder()
+
+	proxy.ServeHTTP(w, req)
+	assert.Equal(t, http.StatusOK, w.Code)
+
+	// Check that metrics were recorded
+	metrics := proxy.metricsMonitor.GetMetrics()
+	if !assert.NotEmpty(t, metrics, "metrics should be recorded for non-streaming request") {
+		return
+	}
+
+	// Verify the last metric has the correct model
+	lastMetric := metrics[len(metrics)-1]
+	assert.Equal(t, "model1", lastMetric.Model)
+	assert.Equal(t, 25, lastMetric.InputTokens, "input tokens should be 25")
+	assert.Equal(t, 10, lastMetric.OutputTokens, "output tokens should be 10")
+	assert.Greater(t, lastMetric.TokensPerSecond, 0.0, "tokens per second should be greater than 0")
+	assert.Greater(t, lastMetric.DurationMs, 0, "duration should be greater than 0")
+}
+
+func TestProxyManager_MiddlewareWritesMetrics_Streaming(t *testing.T) {
+	config := AddDefaultGroupToConfig(Config{
+		HealthCheckTimeout: 15,
+		Models: map[string]ModelConfig{
+			"model1": getTestSimpleResponderConfig("model1"),
+		},
+		LogLevel: "error",
+	})
+
+	proxy := New(config)
+	defer proxy.StopProcesses(StopWaitForInflightRequest)
+
+	// Make a streaming request
+	reqBody := `{"model":"model1", "stream": true}`
+	req := httptest.NewRequest("POST", "/v1/chat/completions?stream=true", bytes.NewBufferString(reqBody))
+	w := httptest.NewRecorder()
+
+	proxy.ServeHTTP(w, req)
+	assert.Equal(t, http.StatusOK, w.Code)
+
+	// Check that metrics were recorded
+	metrics := proxy.metricsMonitor.GetMetrics()
+	if !assert.NotEmpty(t, metrics, "metrics should be recorded for streaming request") {
+		return
+	}
+
+	// Verify the last metric has the correct model
+	lastMetric := metrics[len(metrics)-1]
+	assert.Equal(t, "model1", lastMetric.Model)
+	assert.Equal(t, 25, lastMetric.InputTokens, "input tokens should be 25")
+	assert.Equal(t, 10, lastMetric.OutputTokens, "output tokens should be 10")
+	assert.Greater(t, lastMetric.TokensPerSecond, 0.0, "tokens per second should be greater than 0")
+	assert.Greater(t, lastMetric.DurationMs, 0, "duration should be greater than 0")
+}

ui/src/App.tsx

@@ -3,6 +3,7 @@ import { useTheme } from "./contexts/ThemeProvider";
 import { APIProvider } from "./contexts/APIProvider";
 import LogViewerPage from "./pages/LogViewer";
 import ModelPage from "./pages/Models";
+import ActivityPage from "./pages/Activity";
 
 function App() {
   const theme = useTheme();
@@ -21,6 +22,10 @@ function App() {
                 <NavLink to="/models" className={({ isActive }) => (isActive ? "navlink active" : "navlink")}>
                   Models
                 </NavLink>
+
+                <NavLink to="/activity" className={({ isActive }) => (isActive ? "navlink active" : "navlink")}>
+                  Activity
+                </NavLink>
                 <button className="btn btn--sm" onClick={theme.toggleTheme}>
                   {theme.isDarkMode ? "🌙" : "☀️"}
                 </button>
@@ -32,6 +37,7 @@ function App() {
             <Routes>
               <Route path="/" element={<LogViewerPage />} />
               <Route path="/models" element={<ModelPage />} />
+              <Route path="/activity" element={<ActivityPage />} />
               <Route path="*" element={<Navigate to="/" replace />} />
             </Routes>
           </main>

ui/src/contexts/APIProvider.tsx

@@ -19,30 +19,41 @@ interface APIProviderType {
   enableAPIEvents: (enabled: boolean) => void;
   proxyLogs: string;
   upstreamLogs: string;
+  metrics: Metrics[];
 }
+
+interface Metrics {
+  id: number;
+  timestamp: string;
+  model: string;
+  input_tokens: number;
+  output_tokens: number;
+  tokens_per_second: number;
+  duration_ms: number;
+}
+
 interface LogData {
   source: "upstream" | "proxy";
   data: string;
 }
 interface APIEventEnvelope {
-  type: "modelStatus" | "logData";
+  type: "modelStatus" | "logData" | "metrics";
   data: string;
 }
 
 const APIContext = createContext<APIProviderType | undefined>(undefined);
 type APIProviderProps = {
   children: ReactNode;
+  autoStartAPIEvents?: boolean;
 };
 
-export function APIProvider({ children }: APIProviderProps) {
+export function APIProvider({ children, autoStartAPIEvents = true }: APIProviderProps) {
   const [proxyLogs, setProxyLogs] = useState("");
   const [upstreamLogs, setUpstreamLogs] = useState("");
-  const proxyEventSource = useRef<EventSource | null>(null);
-  const upstreamEventSource = useRef<EventSource | null>(null);
+  const [metrics, setMetrics] = useState<Metrics[]>([]);
   const apiEventSource = useRef<EventSource | null>(null);
 
   const [models, setModels] = useState<Model[]>([]);
-  const modelStatusEventSource = useRef<EventSource | null>(null);
 
   const appendLog = useCallback((newData: string, setter: React.Dispatch<React.SetStateAction<string>>) => {
     setter((prev) => {
@@ -55,6 +66,7 @@ export function APIProvider({ children }: APIProviderProps) {
     if (!enabled) {
       apiEventSource.current?.close();
       apiEventSource.current = null;
+      setMetrics([]);
       return;
     }
 
@@ -75,7 +87,7 @@ export function APIProvider({ children }: APIProviderProps) {
               }
               break;
 
-            case "logData": {
+            case "logData":
               const logData = JSON.parse(message.data) as LogData;
               switch (logData.source) {
                 case "proxy":
@@ -85,7 +97,16 @@ export function APIProvider({ children }: APIProviderProps) {
                   appendLog(logData.data, setUpstreamLogs);
                   break;
               }
-            }
+              break;
+
+            case "metrics":
+              {
+                const newMetric = JSON.parse(message.data) as Metrics;
+                setMetrics((prevMetrics) => {
+                  return [newMetric, ...prevMetrics];
+                });
+              }
+              break;
           }
         } catch (err) {
           console.error(e.data, err);
@@ -105,12 +126,14 @@ export function APIProvider({ children }: APIProviderProps) {
   }, []);
 
   useEffect(() => {
+    if (autoStartAPIEvents) {
+      enableAPIEvents(true);
+    }
+
     return () => {
-      proxyEventSource.current?.close();
-      upstreamEventSource.current?.close();
-      modelStatusEventSource.current?.close();
+      enableAPIEvents(false);
     };
-  }, []);
+  }, [enableAPIEvents, autoStartAPIEvents]);
 
   const listModels = useCallback(async (): Promise<Model[]> => {
     try {
@@ -163,8 +186,9 @@ export function APIProvider({ children }: APIProviderProps) {
       enableAPIEvents,
       proxyLogs,
       upstreamLogs,
+      metrics,
     }),
-    [models, listModels, unloadAllModels, loadModel, enableAPIEvents, proxyLogs, upstreamLogs]
+    [models, listModels, unloadAllModels, loadModel, enableAPIEvents, proxyLogs, upstreamLogs, metrics]
   );
 
   return <APIContext.Provider value={value}>{children}</APIContext.Provider>;

ui/src/lib/Utils.ts

@@ -1,18 +0,0 @@
-export function processEvalTimes(text: string) {
-  const lines = text.match(/^ *eval time.*$/gm) || [];
-
-  let totalTokens = 0;
-  let totalTime = 0;
-
-  lines.forEach((line) => {
-    const tokensMatch = line.match(/\/\s*(\d+)\s*tokens/);
-    const timeMatch = line.match(/=\s*(\d+\.\d+)\s*ms/);
-
-    if (tokensMatch) totalTokens += parseFloat(tokensMatch[1]);
-    if (timeMatch) totalTime += parseFloat(timeMatch[1]);
-  });
-
-  const avgTokensPerSecond = totalTime > 0 ? totalTokens / (totalTime / 1000) : 0;
-
-  return [lines.length, totalTokens, Math.round(avgTokensPerSecond * 100) / 100];
-}

ui/src/pages/Activity.tsx

@@ -0,0 +1,77 @@
+import { useState, useEffect } from "react";
+import { useAPI } from "../contexts/APIProvider";
+
+const formatTimestamp = (timestamp: string): string => {
+  return new Date(timestamp).toLocaleString();
+};
+
+const formatSpeed = (speed: number): string => {
+  return speed < 0 ? "unknown" : speed.toFixed(2) + " t/s";
+};
+
+const formatDuration = (ms: number): string => {
+  return (ms / 1000).toFixed(2) + "s";
+};
+
+const ActivityPage = () => {
+  const { metrics } = useAPI();
+  const [error, setError] = useState<string | null>(null);
+
+  useEffect(() => {
+    if (metrics.length > 0) {
+      setError(null);
+    }
+  }, [metrics]);
+
+  if (error) {
+    return (
+      <div className="p-6">
+        <h1 className="text-2xl font-bold mb-4">Activity</h1>
+        <div className="bg-red-50 border border-red-200 rounded-md p-4">
+          <p className="text-red-800">{error}</p>
+        </div>
+      </div>
+    );
+  }
+
+  return (
+    <div className="p-6">
+      <h1 className="text-2xl font-bold mb-4">Activity</h1>
+
+      {metrics.length === 0 ? (
+        <div className="text-center py-8">
+          <p className="text-gray-600">No metrics data available</p>
+        </div>
+      ) : (
+        <div className="overflow-x-auto">
+          <table className="min-w-full divide-y">
+            <thead>
+              <tr>
+                <th className="px-6 py-3 text-left text-xs font-medium uppercase tracking-wider">Timestamp</th>
+                <th className="px-6 py-3 text-left text-xs font-medium uppercase tracking-wider">Model</th>
+                <th className="px-6 py-3 text-left text-xs font-medium uppercase tracking-wider">Input Tokens</th>
+                <th className="px-6 py-3 text-left text-xs font-medium uppercase tracking-wider">Output Tokens</th>
+                <th className="px-6 py-3 text-left text-xs font-medium uppercase tracking-wider">Generation Speed</th>
+                <th className="px-6 py-3 text-left text-xs font-medium uppercase tracking-wider">Duration</th>
+              </tr>
+            </thead>
+            <tbody className="divide-y">
+              {metrics.map((metric, index) => (
+                <tr key={`${metric.id}-${index}`}>
+                  <td className="px-6 py-4 whitespace-nowrap text-sm">{formatTimestamp(metric.timestamp)}</td>
+                  <td className="px-6 py-4 whitespace-nowrap text-sm">{metric.model}</td>
+                  <td className="px-6 py-4 whitespace-nowrap text-sm">{metric.input_tokens.toLocaleString()}</td>
+                  <td className="px-6 py-4 whitespace-nowrap text-sm">{metric.output_tokens.toLocaleString()}</td>
+                  <td className="px-6 py-4 whitespace-nowrap text-sm">{formatSpeed(metric.tokens_per_second)}</td>
+                  <td className="px-6 py-4 whitespace-nowrap text-sm">{formatDuration(metric.duration_ms)}</td>
+                </tr>
+              ))}
+            </tbody>
+          </table>
+        </div>
+      )}
+    </div>
+  );
+};
+
+export default ActivityPage;

ui/src/pages/LogViewer.tsx

@@ -3,14 +3,7 @@ import { useAPI } from "../contexts/APIProvider";
 import { usePersistentState } from "../hooks/usePersistentState";
 
 const LogViewer = () => {
-  const { proxyLogs, upstreamLogs, enableAPIEvents } = useAPI();
-
-  useEffect(() => {
-    enableAPIEvents(true);
-    return () => {
-      enableAPIEvents(false);
-    };
-  }, []);
+  const { proxyLogs, upstreamLogs } = useAPI();
 
   return (
     <div className="flex flex-col gap-5" style={{ height: "calc(100vh - 125px)" }}>

ui/src/pages/Models.tsx

@@ -1,11 +1,10 @@
-import { useState, useEffect, useCallback, useMemo } from "react";
+import { useState, useCallback, useMemo } from "react";
 import { useAPI } from "../contexts/APIProvider";
 import { LogPanel } from "./LogViewer";
-import { processEvalTimes } from "../lib/Utils";
 import { usePersistentState } from "../hooks/usePersistentState";
 
 export default function ModelsPage() {
-  const { models, unloadAllModels, loadModel, upstreamLogs, enableAPIEvents } = useAPI();
+  const { models, unloadAllModels, loadModel, upstreamLogs, metrics } = useAPI();
   const [isUnloading, setIsUnloading] = useState(false);
   const [showUnlisted, setShowUnlisted] = usePersistentState("showUnlisted", true);
 
@@ -13,13 +12,6 @@ export default function ModelsPage() {
     return models.filter((model) => showUnlisted || !model.unlisted);
   }, [models, showUnlisted]);
 
-  useEffect(() => {
-    enableAPIEvents(true);
-    return () => {
-      enableAPIEvents(false);
-    };
-  }, []);
-
   const handleUnloadAllModels = useCallback(async () => {
     setIsUnloading(true);
     try {
@@ -34,9 +26,15 @@ export default function ModelsPage() {
     }
   }, []);
 
-  const [totalLines, totalTokens, avgTokensPerSecond] = useMemo(() => {
-    return processEvalTimes(upstreamLogs);
-  }, [upstreamLogs]);
+  const [totalRequests, totalTokens, avgTokensPerSecond] = useMemo(() => {
+    const totalRequests = metrics.length;
+    if (totalRequests === 0) {
+      return [0, 0, 0];
+    }
+    const totalTokens = metrics.reduce((sum, m) => sum + m.output_tokens, 0);
+    const avgTokensPerSecond = (metrics.reduce((sum, m) => sum + m.tokens_per_second, 0) / totalRequests).toFixed(2);
+    return [totalRequests, totalTokens, avgTokensPerSecond];
+  }, [metrics]);
 
   return (
     <div>
@@ -96,14 +94,13 @@ export default function ModelsPage() {
 
         {/* Right Column */}
         <div className="w-full md:w-1/2 flex flex-col" style={{ height: "calc(100vh - 125px)" }}>
-          <div className="card mb-4 min-h-[250px]">
-            <h2>Log Stats</h2>
-            <p className="italic my-2">note: eval logs from llama-server</p>
+          <div className="card mb-4 min-h-[225px]">
+            <h2>Chat Activity</h2>
             <table className="w-full border border-gray-200">
               <tbody>
                 <tr className="border-b border-gray-200">
                   <td className="py-2 px-4 font-medium border-r border-gray-200">Requests</td>
-                  <td className="py-2 px-4 text-right">{totalLines}</td>
+                  <td className="py-2 px-4 text-right">{totalRequests}</td>
                 </tr>
                 <tr className="border-b border-gray-200">
                   <td className="py-2 px-4 font-medium border-r border-gray-200">Total Tokens Generated</td>