refactor(hub): dispatch 家族收敛为 dispatchsvc 包并去 Setter 装配

hub-server/internal/middleware/auth_test.go

@@ -109,12 +109,12 @@ func TestAuthMiddlewareInvalidToken(t *testing.T) {
 }
 
 func TestAuthMiddlewareRejectsTokenDanceTokenWithoutExpectedAudience(t *testing.T) {
-	token := "not-a-valid-local-token"
+	authHeaderValue := "not-a-valid-local-token"
 	cfg := testConfig()
 	cfg.TokenDanceID.IssuerURL = "https://id.example"
 	cfg.TokenDanceID.ClientID = ""
 
-	c, w := ginRequest(http.MethodGet, "/client/users/me", "Bearer "+token)
+	c, w := ginRequest(http.MethodGet, "/client/users/me", "Bearer "+authHeaderValue)
 	newTestAuthMW(cfg, AuthDependencies{}, nil).Handler()(c)
 
 	if !c.IsAborted() {

hub-server/internal/service/agent.go

@@ -14,6 +14,7 @@ import (
 	"github.com/agenthub/hub-server/internal/errcode"
 	"github.com/agenthub/hub-server/internal/model"
 	"github.com/agenthub/hub-server/internal/repository"
+	"github.com/agenthub/hub-server/internal/service/dispatchsvc"
 	"github.com/agenthub/hub-server/internal/ws"
 )
 
@@ -54,7 +55,7 @@ type AgentService struct {
 	// Constructed in NewAgentService; tests using struct literals fall back to
 	// a lazy facade via dispatchService(). DeliveryOutbox retries call into
 	// DispatchService through Redispatcher (dispatchPayload stays private).
-	dispatch *DispatchService
+	dispatch *dispatchsvc.DispatchService
 	// edgeCfg/jwtSecret are the Hub→Edge dispatch configuration (#1549),
 	// injected by the composition root and forwarded to DispatchService.
 	edgeCfg config.EdgeDispatchConfig
@@ -79,7 +80,7 @@ func NewAgentService(db *gorm.DB, bus *bus.Bus, mgr *ws.Manager, cacheClient *ca
 		s.deliveryOutbox, // DeliveryOutbox implements edgeCallbackOutbox via autoAckDeliveriesForTask
 	)
 	// Dispatch after outbox so RecordDelivery/MarkDeliverySent ports are ready.
-	s.dispatch = NewDispatchService(db, bus, mgr, s.cacheClient, relay, s.deliveryOutbox, edgeCfg, edgeClient, jwtSecret)
+	s.dispatch = dispatchsvc.NewDispatchService(db, bus, wsManagerAdapter{manager: mgr}, s.cacheClient, relayServiceAdapter{relay: relay}, s.deliveryOutbox, edgeCfg, edgeClient, jwtSecret)
 	s.deliveryOutbox.SetRedispatcher(dispatchRedispatcher{s.dispatch})
 	return s
 }

hub-server/internal/service/agent_dispatch_facade.go

@@ -5,6 +5,7 @@ import (
 
 	"github.com/agenthub/hub-server/internal/model"
 	"github.com/agenthub/hub-server/internal/service/dispatch"
+	"github.com/agenthub/hub-server/internal/service/dispatchsvc"
 )
 
 // ── AgentService facade (wiring/handler stability) ───────────────────────────
@@ -15,11 +16,11 @@ import (
 
 // dispatchService returns the composed DispatchService, lazily constructing one
 // from AgentService deps when tests use struct literals without NewAgentService.
-func (s *AgentService) dispatchService() *DispatchService {
+func (s *AgentService) dispatchService() *dispatchsvc.DispatchService {
 	if dispatch.ComposedDispatchReady(s.dispatch != nil) {
 		return s.dispatch
 	}
-	return NewDispatchService(s.db, s.bus, s.mgr, s.cacheClient, s.relay, s.deliveryOutboxService(), s.edgeCfg, s.edgeClient, s.jwtSecret)
+	return dispatchsvc.NewDispatchService(s.db, s.bus, wsManagerAdapter{manager: s.mgr}, s.cacheClient, relayServiceAdapter{relay: s.relay}, s.deliveryOutboxService(), s.edgeCfg, s.edgeClient, s.jwtSecret)
 }
 
 // TriggerAgentTask creates a pending task for an agent and dispatches it to the inviter's edge.

hub-server/internal/service/agent_logic_test.go

@@ -1,20 +1,15 @@
 package service
 
 import (
-	"context"
 	"encoding/json"
 	"testing"
 
 	"github.com/stretchr/testify/assert"
-	"github.com/stretchr/testify/require"
 
-	"github.com/agenthub/hub-server/internal/bus"
-	"github.com/agenthub/hub-server/internal/config"
 	"github.com/agenthub/hub-server/internal/errcode"
 	"github.com/agenthub/hub-server/internal/model"
 	"github.com/agenthub/hub-server/internal/service/agentevent"
 	"github.com/agenthub/hub-server/internal/service/dispatch"
-	"github.com/agenthub/hub-server/internal/ws"
 )
 
 // --- normalizeRuntimeAgentType ---
@@ -265,138 +260,6 @@ func TestValidateAgentCallbackEdgeRunID(t *testing.T) {
 	})
 }
 
-// --- DispatchService residual ports (#617) ---
-
-type recordingDispatchBus struct {
-	events []bus.Event
-}
-
-func (b *recordingDispatchBus) Publish(ctx context.Context, event bus.Event) error {
-	b.events = append(b.events, event)
-	return nil
-}
-
-type recordingDispatchCache struct {
-	routes map[string]string
-	pushed []string
-}
-
-func (c *recordingDispatchCache) GetRoute(ctx context.Context, userID, deviceType string) (string, error) {
-	if c.routes == nil {
-		return "", nil
-	}
-	return c.routes[userID+":"+deviceType], nil
-}
-
-func (c *recordingDispatchCache) GetRouteForDevice(ctx context.Context, userID, deviceType, deviceID string) (string, error) {
-	if c.routes == nil {
-		return "", nil
-	}
-	return c.routes[userID+":"+deviceType+":"+deviceID], nil
-}
-
-func (c *recordingDispatchCache) PushPendingTask(ctx context.Context, userID, taskJSON string) error {
-	c.pushed = append(c.pushed, userID+":"+taskJSON)
-	return nil
-}
-
-func (c *recordingDispatchCache) PushPendingTargetTask(ctx context.Context, userID, targetID, deviceID, taskJSON string) error {
-	c.pushed = append(c.pushed, userID+":"+targetID+":"+deviceID+":"+taskJSON)
-	return nil
-}
-
-type recordingDispatchWS struct {
-	conn   *ws.Conn
-	pushed int
-}
-
-func (m *recordingDispatchWS) FindByConnID(connID string) *ws.Conn {
-	if m.conn == nil || m.conn.ID != connID {
-		return nil
-	}
-	return m.conn
-}
-
-func (m *recordingDispatchWS) PushToConn(connID string, frame ws.Frame) ws.DeliveryResult {
-	m.pushed++
-	return ws.DeliveryResult{Queued: true, Status: ws.DeliveryStatusQueued}
-}
-
-type recordingDispatchOutbox struct {
-	recorded  int
-	marked    int
-	dead      int
-	lastError string
-}
-
-func (o *recordingDispatchOutbox) RecordDelivery(ctx context.Context, taskID, payload, edgeDeviceID string) (string, error) {
-	o.recorded++
-	return "deliv-1", nil
-}
-
-func (o *recordingDispatchOutbox) MarkDeliverySent(ctx context.Context, deliveryID string) error {
-	o.marked++
-	return nil
-}
-
-func (o *recordingDispatchOutbox) MoveDeliveryToDeadLetter(ctx context.Context, deliveryID string, lastError string) error {
-	o.dead++
-	o.lastError = lastError
-	return nil
-}
-
-func TestDispatchService_NilBusPublishIsNoop(t *testing.T) {
-	svc := &DispatchService{}
-	// Must not panic when b port is unset (partial construction).
-	svc.publish(context.Background(), bus.Event{Type: "agent.cancel", Payload: "x"})
-}
-
-func TestDispatchService_NilOutboxWrappers(t *testing.T) {
-	svc := &DispatchService{}
-	_, err := svc.recordDelivery(context.Background(), "t1", "{}", "")
-	require.Error(t, err)
-	require.Contains(t, err.Error(), "dispatch outbox unavailable")
-	require.Error(t, svc.markDeliverySent(context.Background(), "d1"))
-	// dead-letter is a no-op when outbox is unset
-	svc.moveDeliveryToDeadLetter(context.Background(), "d1", "boom")
-}
-
-func TestDispatchService_SetPortsComposition(t *testing.T) {
-	b := &recordingDispatchBus{}
-	cachePort := &recordingDispatchCache{routes: map[string]string{"u1:desktop": "conn-1"}}
-	wsPort := &recordingDispatchWS{conn: &ws.Conn{ID: "conn-1", UserID: "u1", DeviceType: "desktop", DeviceID: "dev-1"}}
-	outbox := &recordingDispatchOutbox{}
-
-	svc := NewDispatchService(nil, nil, nil, nil, nil, nil, config.EdgeDispatchConfig{}, nil, "")
-	require.NotNil(t, svc)
-
-	svc.SetBus(b)
-	svc.SetCache(cachePort)
-	svc.SetManager(wsPort)
-	svc.SetOutbox(outbox)
-	svc.SetRelay(nil)
-
-	svc.publish(context.Background(), bus.Event{Type: "agent.regenerate", Payload: map[string]string{"k": "v"}})
-	require.Len(t, b.events, 1)
-	assert.Equal(t, "agent.regenerate", b.events[0].Type)
-
-	id, err := svc.recordDelivery(context.Background(), "task-1", `{"task_id":"task-1"}`, "dev-1")
-	require.NoError(t, err)
-	assert.Equal(t, "deliv-1", id)
-	require.NoError(t, svc.markDeliverySent(context.Background(), id))
-	svc.moveDeliveryToDeadLetter(context.Background(), id, "hard-fail")
-	assert.Equal(t, 1, outbox.recorded)
-	assert.Equal(t, 1, outbox.marked)
-	assert.Equal(t, 1, outbox.dead)
-	assert.Equal(t, "hard-fail", outbox.lastError)
-
-	got := svc.cachePort()
-	route, err := got.GetRoute(context.Background(), "u1", "desktop")
-	require.NoError(t, err)
-	assert.Equal(t, "conn-1", route)
-	assert.Same(t, wsPort.conn, svc.mgr.FindByConnID("conn-1"))
-}
-
 func TestIsLoopback(t *testing.T) {
 	assert.True(t, dispatch.IsLoopback("http://127.0.0.1:3210"))
 	assert.True(t, dispatch.IsLoopback("http://localhost:3210"))