test(agentcompat): add dashboard runtime harness

Co-authored-by: naiba/CloudCode <hi+cloudcode@nai.ba>

integration/agentcompat/internal/dashboard/dashboard.go

@@ -0,0 +1,233 @@
+//go:build linux
+
+package dashboard
+
+import (
+	"bufio"
+	"context"
+	"errors"
+	"fmt"
+	"net"
+	"net/http"
+	"path/filepath"
+	"sync"
+	"time"
+
+	"github.com/nezhahq/nezha/integration/agentcompat/internal/client"
+	"github.com/nezhahq/nezha/integration/agentcompat/internal/fixture"
+	processharness "github.com/nezhahq/nezha/integration/agentcompat/internal/process"
+	"github.com/nezhahq/nezha/integration/agentcompat/internal/workspace"
+)
+
+const (
+	agentSecret             = "0123456789abcdef0123456789abcdef"
+	jwtSecret               = "agentcompat-dashboard-jwt-secret"
+	defaultReadinessTimeout = 60 * time.Second
+	// Active Agent streams need the Dashboard graceful shutdown window before
+	// the harness is allowed to escalate process-group cleanup to SIGKILL.
+	defaultProcessStopTimeout   = 15 * time.Second
+	defaultProcessKillTimeout   = 5 * time.Second
+	failedStartCleanupTimeout   = 15 * time.Second
+	dashboardMaxLogBytes        = 1 << 20
+	dashboardHTTPClientTimeout  = 5 * time.Second
+	dashboardRequestRetryPeriod = 25 * time.Millisecond
+)
+
+type StartConfig struct {
+	SourceDir        string
+	EnableTLS        bool
+	ReceiptGate      bool
+	ReadinessTimeout time.Duration
+}
+
+type Clients struct {
+	REST      *client.Client
+	MCP       *client.Client
+	WebSocket *client.Client
+}
+
+type FixtureIdentity struct {
+	WorkspaceRoot string
+	ConfigPath    string
+	DatabasePath  string
+	BinaryPath    string
+	HTTP          workspace.ListenerIdentity
+	Receipt       workspace.ListenerIdentity
+	HTTPS         workspace.ListenerIdentity
+}
+
+type RuntimeIdentity struct {
+	Generation     uint64
+	PID            int
+	ProcessGroupID int
+}
+
+type dashboardGeneration struct {
+	supervisor    dashboardSupervisor
+	identity      RuntimeIdentity
+	record        processharness.CleanupRecord
+	receiptConn   net.Conn
+	httpTransport *http.Transport
+	tlsTransport  *http.Transport
+}
+
+type BootstrapResult struct {
+	LoginAuthenticated bool
+	CSRFCookiePresent  bool
+	PATID              uint64
+	PATScopes          []string
+	MCPProtocolVersion string
+	MCPServerName      string
+	MCPToolCount       int
+	TLSAuthenticated   bool
+}
+
+type dashboardSupervisor interface {
+	Start() error
+	Stop(context.Context) error
+	Exited() <-chan struct{}
+	PID() int
+	ProcessGroupID() int
+	CleanupRecord() processharness.CleanupRecord
+}
+
+type Dashboard struct {
+	workspace        *workspace.Workspace
+	supervisor       dashboardSupervisor
+	clients          Clients
+	restHTTPClient   *http.Client
+	httpTransport    *http.Transport
+	tlsTransport     *http.Transport
+	tlsFixture       fixture.LocalTLSFixture
+	httpAddress      string
+	httpsAddress     string
+	receiptAddress   string
+	receiptConn      net.Conn
+	receiptReader    *bufio.Reader
+	receiptEvents    chan string
+	eventNotify      chan struct{}
+	eventMu          sync.RWMutex
+	eventClosed      bool
+	configPath       string
+	databasePath     string
+	logPath          string
+	bootstrap        BootstrapResult
+	readinessTimeout time.Duration
+	startConfig      StartConfig
+	binaryPath       string
+	generation       uint64
+	currentProcess   *dashboardGeneration
+	processes        []*dashboardGeneration
+	httpListener     *workspace.OwnedListener
+	receiptListener  *workspace.OwnedListener
+	httpsListener    *workspace.OwnedListener
+
+	cleanupOnce          sync.Once
+	cleanupDone          chan struct{}
+	cleanupMu            sync.Mutex
+	cleanupError         error
+	cleanupReceipt       processharness.CleanupReceipt
+	receiptMu            sync.RWMutex
+	receiptAccepted      bool
+	receiptAcceptedCount uint64
+	receiptGeneration    uint64
+	info2Mu              sync.Mutex
+	info2Events          map[string]struct{}
+	stateMu              sync.Mutex
+	stateEvents          map[stateEventIdentity]struct{}
+	mcpReceiptEvents     []MCPReceiptEvent
+	mcpReceiptSequence   uint64
+	eventGeneration      uint64
+	lifecycleMu          sync.Mutex
+}
+
+type stateEventIdentity struct {
+	ServerID   uint64
+	UUID       string
+	Generation uint64
+	Count      uint64
+}
+
+type MCPReceiptKind string
+
+const (
+	MCPReceiptTask   MCPReceiptKind = "task"
+	MCPReceiptResult MCPReceiptKind = "result"
+)
+
+type MCPReceiptCursor struct {
+	Sequence uint64
+}
+
+type MCPReceiptEvent struct {
+	Sequence            uint64         `json:"sequence"`
+	DashboardGeneration uint64         `json:"dashboard_generation"`
+	GateGeneration      uint64         `json:"gate_generation"`
+	ServerID            uint64         `json:"server_id"`
+	TaskID              uint64         `json:"task_id"`
+	TaskType            uint64         `json:"task_type"`
+	Kind                MCPReceiptKind `json:"kind"`
+}
+
+type MCPReceiptExpectation struct {
+	DashboardGeneration uint64
+	GateGeneration      uint64
+	ServerID            uint64
+	TaskID              uint64
+	TaskType            uint64
+}
+
+type MCPReceiptPair struct {
+	Task   MCPReceiptEvent `json:"task"`
+	Result MCPReceiptEvent `json:"result"`
+}
+
+var ErrReceiptGateClosed = errors.New("receipt gate closed")
+
+func Start(ctx context.Context, config StartConfig) (*Dashboard, error) {
+	if config.SourceDir == "" || !filepath.IsAbs(config.SourceDir) {
+		return nil, errors.New("dashboard source directory must be absolute")
+	}
+	// Dashboard owns cancellation order so the process group is gone before the
+	// workspace verifies listeners, PIDs, and temporary files are absent.
+	workspaceRoot, err := workspace.New(context.WithoutCancel(ctx))
+	if err != nil {
+		return nil, fmt.Errorf("create dashboard workspace: %w", err)
+	}
+	dashboard := &Dashboard{workspace: workspaceRoot, cleanupDone: make(chan struct{}), startConfig: config}
+	dashboard.readinessTimeout = config.ReadinessTimeout
+	if dashboard.readinessTimeout <= 0 {
+		dashboard.readinessTimeout = defaultReadinessTimeout
+	}
+	if err := dashboard.prepare(ctx, config); err != nil {
+		cleanupContext, cancel := context.WithTimeout(context.WithoutCancel(ctx), failedStartCleanupTimeout)
+		defer cancel()
+		return nil, errors.Join(err, dashboard.Stop(cleanupContext))
+	}
+	go dashboard.cleanupOnCancellation(ctx)
+	return dashboard, nil
+}
+
+func (dashboard *Dashboard) Stop(ctx context.Context) error {
+	dashboard.cleanupOnce.Do(func() { go dashboard.cleanup(context.WithoutCancel(ctx)) })
+	select {
+	case <-dashboard.cleanupDone:
+		dashboard.cleanupMu.Lock()
+		defer dashboard.cleanupMu.Unlock()
+		return dashboard.cleanupError
+	case <-ctx.Done():
+		return ctx.Err()
+	}
+}
+
+func (dashboard *Dashboard) Close(ctx context.Context) error { return dashboard.Stop(ctx) }
+
+func (dashboard *Dashboard) URL() string { return "http://" + dashboard.httpAddress }
+
+func (dashboard *Dashboard) Endpoint() string { return dashboard.httpAddress }
+
+func (dashboard *Dashboard) TLSEndpoint() string { return dashboard.httpsAddress }
+
+func (dashboard *Dashboard) ReceiptGateEnabled() bool { return dashboard.receiptAddress != "" }
+
+func (dashboard *Dashboard) ReceiptGateEndpoint() string { return dashboard.receiptAddress }