Merge upstream/main (auto-sync feat/copilot)

- 55440f0 feat(auth): add runtime auth removal and unscheduling logic
- fd30944 feat(auth): add error event publishing and Redis queue integration

Author: github-actions[bot]

internal/api/handlers/management/auth_files.go

@@ -770,7 +770,7 @@ func (h *Handler) DeleteAuthFile(c *gin.Context) {
 					return
 				}
 				deleted++
-				h.disableAuth(ctx, full)
+				h.removeAuth(ctx, full)
 			}
 		}
 		c.JSON(200, gin.H{"status": "ok", "deleted": deleted})
@@ -976,9 +976,9 @@ func (h *Handler) deleteAuthFileByName(ctx context.Context, name string) (string
 		return filepath.Base(name), http.StatusInternalServerError, errDeleteRecord
 	}
 	if targetID != "" {
-		h.disableAuth(ctx, targetID)
+		h.removeAuth(ctx, targetID)
 	} else {
-		h.disableAuth(ctx, targetPath)
+		h.removeAuth(ctx, targetPath)
 	}
 	return filepath.Base(name), http.StatusOK, nil
 }
@@ -1558,33 +1558,23 @@ func syncAuthFileDisabledState(auth *coreauth.Auth) {
 	auth.StatusMessage = ""
 }
 
-func (h *Handler) disableAuth(ctx context.Context, id string) {
+func (h *Handler) removeAuth(ctx context.Context, id string) {
 	if h == nil || h.authManager == nil {
 		return
 	}
 	id = strings.TrimSpace(id)
 	if id == "" {
 		return
 	}
-	if auth, ok := h.authManager.GetByID(id); ok {
-		auth.Disabled = true
-		auth.Status = coreauth.StatusDisabled
-		auth.StatusMessage = "removed via management API"
-		auth.UpdatedAt = time.Now()
-		_, _ = h.authManager.Update(ctx, auth)
+	if _, ok := h.authManager.GetByID(id); ok {
+		h.authManager.Remove(ctx, id)
 		return
 	}
 	authID := h.authIDForPath(id)
 	if authID == "" {
 		return
 	}
-	if auth, ok := h.authManager.GetByID(authID); ok {
-		auth.Disabled = true
-		auth.Status = coreauth.StatusDisabled
-		auth.StatusMessage = "removed via management API"
-		auth.UpdatedAt = time.Now()
-		_, _ = h.authManager.Update(ctx, auth)
-	}
+	h.authManager.Remove(ctx, authID)
 }
 
 func (h *Handler) deleteTokenRecord(ctx context.Context, path string) error {

internal/api/handlers/management/auth_files_delete_test.go

@@ -127,3 +127,49 @@ func TestDeleteAuthFile_FallbackToAuthDirPath(t *testing.T) {
 		t.Fatalf("expected auth file to be removed from auth dir, stat err: %v", errStat)
 	}
 }
+
+func TestDeleteAuthFile_RemovesRuntimeAuth(t *testing.T) {
+	t.Setenv("MANAGEMENT_PASSWORD", "")
+	gin.SetMode(gin.TestMode)
+
+	authDir := t.TempDir()
+	fileName := "runtime-remove-user.json"
+	filePath := filepath.Join(authDir, fileName)
+	if errWrite := os.WriteFile(filePath, []byte(`{"type":"codex","email":"runtime@example.com"}`), 0o600); errWrite != nil {
+		t.Fatalf("failed to write auth file: %v", errWrite)
+	}
+
+	manager := coreauth.NewManager(nil, nil, nil)
+	record := &coreauth.Auth{
+		ID:       "runtime-remove-auth",
+		FileName: fileName,
+		Provider: "codex",
+		Status:   coreauth.StatusActive,
+		Attributes: map[string]string{
+			"path": filePath,
+		},
+		Metadata: map[string]any{
+			"type":  "codex",
+			"email": "runtime@example.com",
+		},
+	}
+	if _, errRegister := manager.Register(context.Background(), record); errRegister != nil {
+		t.Fatalf("failed to register auth record: %v", errRegister)
+	}
+
+	h := NewHandlerWithoutConfigFilePath(&config.Config{AuthDir: authDir}, manager)
+	h.tokenStore = &memoryAuthStore{}
+
+	deleteRec := httptest.NewRecorder()
+	deleteCtx, _ := gin.CreateTestContext(deleteRec)
+	deleteReq := httptest.NewRequest(http.MethodDelete, "/v0/management/auth-files?name="+url.QueryEscape(fileName), nil)
+	deleteCtx.Request = deleteReq
+	h.DeleteAuthFile(deleteCtx)
+
+	if deleteRec.Code != http.StatusOK {
+		t.Fatalf("expected delete status %d, got %d with body %s", http.StatusOK, deleteRec.Code, deleteRec.Body.String())
+	}
+	if _, ok := manager.GetByID(record.ID); ok {
+		t.Fatalf("expected runtime auth %q to be removed", record.ID)
+	}
+}

internal/api/redis_queue_protocol.go

@@ -14,7 +14,10 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-const redisUsageChannel = "usage"
+const (
+	redisUsageChannel  = "usage"
+	redisErrorsChannel = "errors"
+)
 
 type redisSubscriptionCommand struct {
 	args []string
@@ -150,15 +153,15 @@ func (s *Server) handleRedisConnection(conn net.Conn, reader *bufio.Reader) {
 				}
 				continue
 			}
-			if !strings.EqualFold(channel, redisUsageChannel) {
+			messages, unsubscribe, ok := subscribeRedisChannel(channel)
+			if !ok {
 				_ = writeRedisError(writer, fmt.Sprintf("ERR unsupported channel '%s'", channel))
 				if !flush() {
 					return
 				}
 				continue
 			}
-			messages, unsubscribe := redisqueue.SubscribeUsage()
-			if errWrite := writeRedisPubSubSubscribe(writer, redisUsageChannel, 1); errWrite != nil {
+			if errWrite := writeRedisPubSubSubscribe(writer, channel, 1); errWrite != nil {
 				unsubscribe()
 				log.Errorf("redis protocol subscribe response error: %v", errWrite)
 				return
@@ -167,7 +170,7 @@ func (s *Server) handleRedisConnection(conn net.Conn, reader *bufio.Reader) {
 				unsubscribe()
 				return
 			}
-			s.streamRedisUsageSubscription(reader, writer, messages, unsubscribe)
+			s.streamRedisSubscription(reader, writer, channel, messages, unsubscribe)
 			return
 		case "LPOP", "RPOP":
 			count, hasCount, ok := parsePopCount(args)
@@ -185,7 +188,14 @@ func (s *Server) handleRedisConnection(conn net.Conn, reader *bufio.Reader) {
 				}
 				continue
 			}
-			items := redisqueue.PopOldest(count)
+			items, ok := popRedisQueueItems(args[1], count)
+			if !ok {
+				_ = writeRedisError(writer, fmt.Sprintf("ERR unsupported channel '%s'", strings.TrimSpace(args[1])))
+				if !flush() {
+					return
+				}
+				continue
+			}
 			if hasCount {
 				_ = writeRedisArrayOfBulkStrings(writer, items)
 				if !flush() {
@@ -213,7 +223,29 @@ func (s *Server) handleRedisConnection(conn net.Conn, reader *bufio.Reader) {
 	}
 }
 
-func (s *Server) streamRedisUsageSubscription(reader *bufio.Reader, writer *bufio.Writer, messages <-chan []byte, unsubscribe func()) {
+func subscribeRedisChannel(channel string) (<-chan []byte, func(), bool) {
+	switch strings.ToLower(strings.TrimSpace(channel)) {
+	case redisUsageChannel:
+		messages, unsubscribe := redisqueue.SubscribeUsage()
+		return messages, unsubscribe, true
+	case redisErrorsChannel:
+		messages, unsubscribe := redisqueue.SubscribeErrors()
+		return messages, unsubscribe, true
+	default:
+		return nil, nil, false
+	}
+}
+
+func popRedisQueueItems(channel string, count int) ([][]byte, bool) {
+	switch strings.ToLower(strings.TrimSpace(channel)) {
+	case redisUsageChannel:
+		return redisqueue.PopOldest(count), true
+	default:
+		return nil, false
+	}
+}
+
+func (s *Server) streamRedisSubscription(reader *bufio.Reader, writer *bufio.Writer, channel string, messages <-chan []byte, unsubscribe func()) {
 	if unsubscribe == nil {
 		return
 	}
@@ -231,7 +263,7 @@ func (s *Server) streamRedisUsageSubscription(reader *bufio.Reader, writer *bufi
 			if !ok {
 				return
 			}
-			if errWrite := writeRedisPubSubMessage(writer, redisUsageChannel, msg); errWrite != nil {
+			if errWrite := writeRedisPubSubMessage(writer, channel, msg); errWrite != nil {
 				log.Errorf("redis protocol publish message error: %v", errWrite)
 				return
 			}
@@ -243,7 +275,7 @@ func (s *Server) streamRedisUsageSubscription(reader *bufio.Reader, writer *bufi
 			if !ok {
 				return
 			}
-			keepOpen := handleRedisSubscriptionCommand(writer, command)
+			keepOpen := handleRedisSubscriptionCommand(writer, channel, command)
 			if errFlush := writer.Flush(); errFlush != nil {
 				log.Errorf("redis protocol flush error: %v", errFlush)
 				return
@@ -277,7 +309,7 @@ func readRedisSubscriptionCommands(reader *bufio.Reader, commands chan<- redisSu
 	}
 }
 
-func handleRedisSubscriptionCommand(writer *bufio.Writer, command redisSubscriptionCommand) bool {
+func handleRedisSubscriptionCommand(writer *bufio.Writer, channel string, command redisSubscriptionCommand) bool {
 	if command.err != nil {
 		_ = writeRedisError(writer, "ERR "+command.err.Error())
 		return false
@@ -297,7 +329,7 @@ func handleRedisSubscriptionCommand(writer *bufio.Writer, command redisSubscript
 		_ = writeRedisPubSubPong(writer, payload)
 		return true
 	case "UNSUBSCRIBE":
-		_ = writeRedisPubSubUnsubscribe(writer, redisUsageChannel, 0)
+		_ = writeRedisPubSubUnsubscribe(writer, channel, 0)
 		return false
 	case "QUIT":
 		_ = writeRedisSimpleString(writer, "OK")

internal/api/redis_queue_protocol_integration_test.go

@@ -359,6 +359,60 @@ func TestRedisProtocol_SUBSCRIBE_UsageSendsSupportRefresh(t *testing.T) {
 	}
 }
 
+func TestRedisProtocol_SUBSCRIBE_ErrorsReceivesErrorEvents(t *testing.T) {
+	const managementPassword = "test-management-password"
+
+	t.Setenv("MANAGEMENT_PASSWORD", managementPassword)
+	redisqueue.SetEnabled(false)
+	t.Cleanup(func() { redisqueue.SetEnabled(false) })
+
+	server := newTestServer(t)
+	if !server.managementRoutesEnabled.Load() {
+		t.Fatalf("expected managementRoutesEnabled to be true")
+	}
+
+	addr, stop := startRedisMuxListener(t, server)
+	t.Cleanup(stop)
+
+	conn, errDial := net.DialTimeout("tcp", addr, time.Second)
+	if errDial != nil {
+		t.Fatalf("failed to dial redis listener: %v", errDial)
+	}
+	t.Cleanup(func() { _ = conn.Close() })
+
+	reader := bufio.NewReader(conn)
+	_ = conn.SetDeadline(time.Now().Add(5 * time.Second))
+
+	if errWrite := writeTestRESPCommand(conn, "AUTH", managementPassword); errWrite != nil {
+		t.Fatalf("failed to write AUTH command: %v", errWrite)
+	}
+	if msg, errRead := readTestRESPSimpleString(reader); errRead != nil {
+		t.Fatalf("failed to read AUTH response: %v", errRead)
+	} else if msg != "OK" {
+		t.Fatalf("unexpected AUTH response: %q", msg)
+	}
+
+	if errWrite := writeTestRESPCommand(conn, "SUBSCRIBE", "errors"); errWrite != nil {
+		t.Fatalf("failed to write SUBSCRIBE command: %v", errWrite)
+	}
+	channel, subscriptions, errSubscribe := readTestRESPPubSubSubscribe(reader)
+	if errSubscribe != nil {
+		t.Fatalf("failed to read subscribe response: %v", errSubscribe)
+	}
+	if channel != "errors" || subscriptions != 1 {
+		t.Fatalf("unexpected subscribe response channel=%q subscriptions=%d", channel, subscriptions)
+	}
+
+	redisqueue.EnqueueError([]byte(`{"auth_index":"auth-1","status_code":401}`))
+	channel, payload, errMessage := readTestRESPPubSubMessage(reader)
+	if errMessage != nil {
+		t.Fatalf("failed to read error message: %v", errMessage)
+	}
+	if channel != "errors" || string(payload) != `{"auth_index":"auth-1","status_code":401}` {
+		t.Fatalf("unexpected error message channel=%q payload=%q", channel, string(payload))
+	}
+}
+
 func TestRedisProtocol_AUTH_And_PopContracts(t *testing.T) {
 	const managementPassword = "test-management-password"
 
@@ -450,4 +504,13 @@ func TestRedisProtocol_AUTH_And_PopContracts(t *testing.T) {
 	if len(emptyItems) != 0 {
 		t.Fatalf("expected empty array for empty queue with count, got %#v", emptyItems)
 	}
+
+	if errWrite := writeTestRESPCommand(conn, "RPOP", "errors", "2"); errWrite != nil {
+		t.Fatalf("failed to write RPOP errors count command: %v", errWrite)
+	}
+	if msg, errRead := readTestRESPError(reader); errRead != nil {
+		t.Fatalf("failed to read RPOP errors response: %v", errRead)
+	} else if msg != "ERR unsupported channel 'errors'" {
+		t.Fatalf("unexpected RPOP errors response: %q", msg)
+	}
 }

internal/redisqueue/queue.go

@@ -10,6 +10,7 @@ const (
 	defaultRetentionSeconds int64 = 60
 	maxRetentionSeconds     int64 = 3600
 	usageSubscriberBuffer         = 256
+	errorSubscriberBuffer         = 256
 
 	usageSupportRefreshPayload = `{"support_refresh":true}`
 	usageRefreshPayload        = `{"refresh":true}`
@@ -32,6 +33,7 @@ var (
 	enabled          atomic.Bool
 	retentionSeconds atomic.Int64
 	global           queue
+	errorGlobal      queue
 )
 
 func init() {
@@ -42,6 +44,7 @@ func SetEnabled(value bool) {
 	enabled.Store(value)
 	if !value {
 		global.clear()
+		errorGlobal.clear()
 	}
 }
 
@@ -72,6 +75,16 @@ func Enqueue(payload []byte) {
 	global.enqueue(payload)
 }
 
+func EnqueueError(payload []byte) {
+	if !Enabled() {
+		return
+	}
+	if len(payload) == 0 {
+		return
+	}
+	errorGlobal.publishToSubscribers(payload)
+}
+
 func PopOldest(count int) [][]byte {
 	if !Enabled() {
 		return nil
@@ -83,7 +96,11 @@ func PopOldest(count int) [][]byte {
 }
 
 func SubscribeUsage() (<-chan []byte, func()) {
-	return global.subscribeUsage()
+	return global.subscribe(usageSubscriberBuffer, []byte(usageSupportRefreshPayload))
+}
+
+func SubscribeErrors() (<-chan []byte, func()) {
+	return errorGlobal.subscribe(errorSubscriberBuffer, nil)
 }
 
 func NotifyUsageRefresh() {
@@ -142,9 +159,11 @@ func (q *queue) publishToSubscribers(payload []byte) bool {
 	return true
 }
 
-func (q *queue) subscribeUsage() (<-chan []byte, func()) {
-	subscriber := make(chan []byte, usageSubscriberBuffer)
-	subscriber <- []byte(usageSupportRefreshPayload)
+func (q *queue) subscribe(buffer int, initialPayload []byte) (<-chan []byte, func()) {
+	subscriber := make(chan []byte, buffer)
+	if len(initialPayload) > 0 {
+		subscriber <- append([]byte(nil), initialPayload...)
+	}
 
 	q.mu.Lock()
 	if q.subscribers == nil {
@@ -158,13 +177,13 @@ func (q *queue) subscribeUsage() (<-chan []byte, func()) {
 	var once sync.Once
 	unsubscribe := func() {
 		once.Do(func() {
-			q.unsubscribeUsage(id)
+			q.unsubscribe(id)
 		})
 	}
 	return subscriber, unsubscribe
 }
 
-func (q *queue) unsubscribeUsage(id uint64) {
+func (q *queue) unsubscribe(id uint64) {
 	q.mu.Lock()
 	subscriber, ok := q.subscribers[id]
 	if ok {