dingtalk-workspace-cli/internal/helpers/connect_daemon.go

// Copyright 2026 Alibaba Group
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package helpers

import (
	"context"
	"encoding/json"
	"fmt"
	"io"
	"os"
	"os/exec"
	"os/signal"
	"path/filepath"
	"strings"
	"syscall"
	"time"

	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/cobracmd"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/corecmd/contract"
	apperrors "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/errors"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/executor"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/logging"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/output"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/tui"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
	"github.com/spf13/cobra"
)

// connect_daemon turns the foreground `robot connect` connector into a 7x24
// background service. Three responsibilities live here and nowhere else (the
// forwarding/session/knowledge logic in devapp_connect.go and connect_stream.go
// is untouched):
//
//  1. detach: `connect --daemon` re-execs dws in supervisor mode in a new
//     session (POSIX setsid), prints pid + log path, and exits.
//  2. supervise: the supervisor process runs the real connector as a worker
//     child and restarts it with exponential backoff when it crashes.
//  3. status/stop: read the daemon pid file, probe liveness (reusing
//     processAlive from connect_lock.go), and signal a graceful stop.
//
// Two hidden internal flags select the mode of a re-exec:
//
//	--daemon-supervise : run the supervisor loop (set by the --daemon parent)
//	--daemon-worker    : run a single foreground connector (set by the supervisor)
const (
	daemonSuperviseFlag = "daemon-supervise"
	daemonWorkerFlag    = "daemon-worker"
	daemonFlag          = "daemon"
)

// daemonState is the JSON persisted to the daemon pid file. It records the
// supervisor pid plus enough context for `status` to report without re-deriving
// it (start time for uptime, log path, the dir key it was filed under).
type daemonState struct {
	Pid           int    `json:"pid"`
	StartUnix     int64  `json:"startUnix"`
	LogPath       string `json:"logPath"`
	DirKey        string `json:"dirKey"`
	ClientID      string `json:"clientId,omitempty"`
	UnifiedAppID  string `json:"unifiedAppId,omitempty"`
	Channel       string `json:"channel,omitempty"`
	NotifyStaffID string `json:"notifyStaffId,omitempty"`
	// Profile records the --profile selector the connector was started with,
	// so `restart` re-fetches credentials against the same org instead of the
	// default profile (which may not know the unifiedAppId at all).
	Profile string `json:"profile,omitempty"`
	// AlwaysOn controls whether the supervisor auto-restarts the worker on
	// crash. Without it the supervisor exits after the first worker exit.
	AlwaysOn bool `json:"alwaysOn,omitempty"`
}

// connectDaemonDirOverride lets tests redirect the per-client daemon directory
// away from the real ~/.dws tree. Empty means use config.DefaultConfigDir.
var connectDaemonDirOverride string

// Process and clock hooks keep daemon lifecycle tests deterministic while the
// production defaults remain the operating-system implementations.
var (
	daemonDetachEnabled = daemonDetachSupported
	daemonExecutable    = os.Executable
	daemonCommand       = exec.Command
	daemonNow           = time.Now
	daemonCreateTemp    = os.CreateTemp
	daemonFileChmod     = func(file *os.File, mode os.FileMode) error { return file.Chmod(mode) }
	daemonCopy          = io.Copy
	daemonFileSync      = func(file *os.File) error { return file.Sync() }
	daemonFileClose     = func(file *os.File) error { return file.Close() }
	daemonRename        = os.Rename
	daemonFindProcess   = os.FindProcess
	daemonProcessAlive  = processAlive
	daemonSignalProcess = func(process *os.Process, signal os.Signal) error {
		return process.Signal(signal)
	}
	daemonSignalContext = func() (context.Context, context.CancelFunc) {
		return signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
	}
)

// daemonDirKey derives a filesystem-safe directory key identifying a connector.
// Priority: clientId (the robot's AppKey, the natural identity) > unifiedAppID.
// Reuses sanitizeLockID (connect_lock.go) so the key matches the lock naming
// convention. Returns "" when neither is available.
func daemonDirKey(clientID, unifiedAppID string) string {
	if v := strings.TrimSpace(clientID); v != "" {
		return sanitizeLockID(v)
	}
	if v := strings.TrimSpace(unifiedAppID); v != "" {
		return "app-" + sanitizeLockID(v)
	}
	return ""
}

// connectDaemonDir returns <configDir>/connect/<dirKey>, creating it. This holds
// daemon.pid and daemon.log for one connector.
func connectDaemonDir(dirKey string) (string, error) {
	base := connectDaemonDirOverride
	if base == "" {
		base = config.DefaultConfigDir()
	}
	dir := filepath.Join(base, "connect", dirKey)
	if err := os.MkdirAll(dir, config.DirPerm); err != nil {
		return "", err
	}
	return dir, nil
}

func daemonPidPath(dir string) string   { return filepath.Join(dir, "daemon.pid") }
func daemonStatePath(dir string) string { return filepath.Join(dir, "daemon-state.json") }
func daemonLogPath(dir string) string   { return filepath.Join(dir, "daemon.log") }
func daemonExecutablePath(dir string) string {
	return filepath.Join(dir, "dws-daemon")
}

// stageDaemonExecutable snapshots the current dws binary into the connector's
// persistent directory. One-click launchers may execute a temporary download
// (for example dws-latest) and remove it after connect --daemon returns; the
// supervisor must not depend on that transient path for future worker restarts.
func stageDaemonExecutable(src, dir string) (string, error) {
	in, err := os.Open(src)
	if err != nil {
		return "", err
	}
	defer in.Close()

	tmp, err := daemonCreateTemp(dir, ".dws-daemon-*")
	if err != nil {
		return "", err
	}
	tmpPath := tmp.Name()
	defer os.Remove(tmpPath)

	if err := daemonFileChmod(tmp, daemonExecutablePerm); err != nil {
		tmp.Close()
		return "", err
	}
	if _, err := daemonCopy(tmp, in); err != nil {
		tmp.Close()
		return "", err
	}
	if err := daemonFileSync(tmp); err != nil {
		tmp.Close()
		return "", err
	}
	if err := daemonFileClose(tmp); err != nil {
		return "", err
	}

	dst := daemonExecutablePath(dir)
	if err := daemonRename(tmpPath, dst); err != nil {
		return "", err
	}
	return dst, nil
}

// writeDaemonState atomically persists the daemon state to daemon-state.json
// (persistent, survives supervisor exit) so restart/list can recover config.
func writeDaemonState(dir string, st daemonState) error {
	data, _ := json.MarshalIndent(st, "", "  ")
	tmp := daemonStatePath(dir) + ".tmp"
	if err := os.WriteFile(tmp, data, config.FilePerm); err != nil {
		return err
	}
	return daemonRename(tmp, daemonStatePath(dir))
}

// readDaemonState loads the daemon state. Reads daemon-state.json (persistent)
// first, falls back to daemon.pid for backward compat with old connectors.
func readDaemonState(dir string) (*daemonState, error) {
	for _, p := range []string{daemonStatePath(dir), daemonPidPath(dir)} {
		data, err := os.ReadFile(p)
		if err != nil {
			if os.IsNotExist(err) {
				continue
			}
			return nil, err
		}
		var st daemonState
		if err := json.Unmarshal(data, &st); err != nil {
			return nil, fmt.Errorf("daemon state file %s is corrupt: %w", p, err)
		}
		return &st, nil
	}
	return nil, nil
}

// backoffDelay computes the restart delay for the Nth consecutive worker
// failure: base * 2^(n-1) capped at cap. Pure and unit-tested. n<=0 returns 0
// (first start is immediate). Caller resets n when a worker stays healthy.
func backoffDelay(consecutiveFailures int, base, maxDelay time.Duration) time.Duration {
	if consecutiveFailures <= 0 {
		return 0
	}
	d := base
	for i := 1; i < consecutiveFailures; i++ {
		d *= 2
		if d >= maxDelay {
			return maxDelay
		}
	}
	if d > maxDelay {
		return maxDelay
	}
	return d
}

const (
	daemonBackoffBase                = time.Second
	daemonBackoffCap                 = 60 * time.Second
	daemonExecutablePerm os.FileMode = 0o700
	// daemonMaxFastFailures is the consecutive fast-failure ceiling: after this
	// many crashes that each happened within daemonHealthyAfter, the supervisor
	// gives up rather than spin forever (e.g. bad credentials).
	daemonMaxFastFailures = 10
	// daemonHealthyAfter is how long a worker must run before the supervisor
	// considers it healthy and resets the failure counter.
	daemonHealthyAfter = 60 * time.Second
	// daemonStopTimeout bounds the graceful wait in `stop` before SIGKILL.
	daemonStopTimeout = 10 * time.Second
)

// buildWorkerArgs rewrites the supervisor's own argv into a worker argv: it
// strips the daemon-control flags (--daemon / --daemon-supervise) and appends
// --daemon-worker, preserving every other flag (credentials, channel, knowledge,
// etc.) so the worker connects exactly as the foreground command would. Pure for
// unit testing.
func buildWorkerArgs(args []string) []string {
	out := make([]string, 0, len(args)+1)
	for _, a := range args {
		switch {
		case a == "--"+daemonFlag, a == "--"+daemonSuperviseFlag, a == "--"+daemonWorkerFlag:
			continue
		case strings.HasPrefix(a, "--"+daemonFlag+"="),
			strings.HasPrefix(a, "--"+daemonSuperviseFlag+"="),
			strings.HasPrefix(a, "--"+daemonWorkerFlag+"="):
			continue
		}
		out = append(out, a)
	}
	out = append(out, "--"+daemonWorkerFlag)
	return out
}

// startDaemon implements `connect --daemon`: it re-execs dws in supervisor mode
// detached from the terminal, writes nothing itself to the worker log (the
// supervisor does), prints the pid + log path, and returns so the parent exits.
func startDaemon(cmd *cobra.Command, dirKey, clientID, unifiedAppID, channel, notifyStaffID, profile string, alwaysOn bool) error {
	if !daemonDetachEnabled {
		return apperrors.NewValidation("--daemon is not supported on this OS; run the foreground connector under a service manager instead")
	}
	dir, err := connectDaemonDir(dirKey)
	if err != nil {
		return apperrors.NewInternal("create daemon dir: " + err.Error())
	}
	// Refuse to start a second supervisor for the same connector. The Stream
	// single-instance lock would also catch this at the worker layer, but a
	// pre-flight check gives a clearer message and avoids an orphaned supervisor.
	if st, _ := readDaemonState(dir); st != nil && st.Pid > 0 && daemonProcessAlive(st.Pid) {
		return apperrors.NewValidation(fmt.Sprintf("a connect daemon is already running for %s (pid %d); use `robot connect status`/`stop`", dirKey, st.Pid))
	}

	exe, err := daemonExecutable()
	if err != nil {
		return apperrors.NewInternal("resolve executable: " + err.Error())
	}
	exe, err = stageDaemonExecutable(exe, dir)
	if err != nil {
		return apperrors.NewInternal("stage daemon executable: " + err.Error())
	}
	superviseArgs := buildSuperviseArgs(os.Args[1:])

	logPath := daemonLogPath(dir)
	logFile, err := os.OpenFile(logPath, os.O_CREATE|os.O_WRONLY|os.O_APPEND, config.FilePerm)
	if err != nil {
		return apperrors.NewInternal("open daemon log: " + err.Error())
	}
	defer logFile.Close()

	child := daemonCommand(exe, superviseArgs...)
	child.Stdout = logFile
	child.Stderr = logFile
	child.Stdin = nil
	// Pass the resolved dir key + clientId through the environment so the
	// supervisor files its pid under the same key the parent computed (rather
	// than re-deriving and risking a mismatch).
	child.Env = append(os.Environ(),
		"DWS_CONNECT_DAEMON_DIRKEY="+dirKey,
		"DWS_CONNECT_DAEMON_CLIENTID="+clientID,
		"DWS_CONNECT_DAEMON_UNIFIEDAPPID="+unifiedAppID,
		"DWS_CONNECT_DAEMON_CHANNEL="+channel,
		"DWS_CONNECT_DAEMON_NOTIFY_STAFF_ID="+notifyStaffID,
		"DWS_CONNECT_DAEMON_PROFILE="+profile,
	)
	if alwaysOn {
		child.Env = append(child.Env, "DWS_CONNECT_DAEMON_ALWAYSON=true")
	}
	if connectDaemonDirOverride != "" {
		child.Env = append(child.Env, "DWS_CONNECT_DAEMON_DIR="+connectDaemonDirOverride)
	}
	applyDetach(child)

	if err := child.Start(); err != nil {
		return apperrors.NewInternal("start daemon: " + err.Error())
	}
	// Release the child so the parent can exit without leaving a zombie.
	pid := child.Process.Pid
	_ = child.Process.Release()
	// The foreground/daemon root remains one streaming legacy command in this
	// release. It cannot adopt a terminal result until the modes are split.
	writeConnectDaemonStarted(cmd.OutOrStdout(), pid, logPath, clientID, dirKey)
	return nil
}

// connectDaemonStartedResult 是 `dev connect --daemon` 父进程的结果 DTO
// （队列 B110）：pid/日志路径等机器事实进成功信封的 data 层。
type connectDaemonStartedResult struct {
	Status   string `json:"status"`
	Pid      int    `json:"pid"`
	LogPath  string `json:"logPath,omitempty"`
	DirKey   string `json:"dirKey,omitempty"`
	ClientID string `json:"clientId,omitempty"`
}

func writeConnectDaemonStarted(w io.Writer, pid int, logPath, clientID, dirKey string) {
	fmt.Fprintf(w, "connect daemon started (pid %d)\n", pid)
	fmt.Fprintf(w, "  logs:   %s\n", logPath)
	fmt.Fprintf(w, "  status: dws dev connect status%s\n", statusHintArgs(clientID, dirKey))
	fmt.Fprintf(w, "  stop:   dws dev connect stop%s\n", statusHintArgs(clientID, dirKey))
	fmt.Fprint(w, connectLocalDebugNotice())
}

// buildSuperviseArgs rewrites argv to run the supervisor: strip --daemon, append
// --daemon-supervise. Pure for testing.
func buildSuperviseArgs(args []string) []string {
	out := make([]string, 0, len(args)+1)
	for _, a := range args {
		if a == "--"+daemonFlag || strings.HasPrefix(a, "--"+daemonFlag+"=") {
			continue
		}
		out = append(out, a)
	}
	out = append(out, "--"+daemonSuperviseFlag)
	return out
}

func statusHintArgs(clientID, dirKey string) string {
	if strings.TrimSpace(clientID) != "" {
		return " --robot-client-id " + clientID
	}
	if strings.HasPrefix(dirKey, "app-") {
		return " --unified-app-id " + strings.TrimPrefix(dirKey, "app-")
	}
	return ""
}

// runSupervisor is the entry point when dws is started with --daemon-supervise.
// It writes the daemon pid file, then loops launching the worker child and
// restarting it with backoff until told to stop (SIGTERM/SIGINT) or it exhausts
// the fast-failure budget. On stop it forwards the signal to the worker, waits,
// and removes its pid file.
func runSupervisor(cmd *cobra.Command) error {
	dirKey := strings.TrimSpace(os.Getenv("DWS_CONNECT_DAEMON_DIRKEY"))
	if dirKey == "" {
		return apperrors.NewInternal("supervisor started without DWS_CONNECT_DAEMON_DIRKEY")
	}
	if v := strings.TrimSpace(os.Getenv("DWS_CONNECT_DAEMON_DIR")); v != "" {
		connectDaemonDirOverride = v
	}
	clientID := strings.TrimSpace(os.Getenv("DWS_CONNECT_DAEMON_CLIENTID"))
	unifiedAppID := strings.TrimSpace(os.Getenv("DWS_CONNECT_DAEMON_UNIFIEDAPPID"))
	channel := strings.TrimSpace(os.Getenv("DWS_CONNECT_DAEMON_CHANNEL"))
	notifyStaffID := strings.TrimSpace(os.Getenv("DWS_CONNECT_DAEMON_NOTIFY_STAFF_ID"))
	profile := strings.TrimSpace(os.Getenv("DWS_CONNECT_DAEMON_PROFILE"))
	alwaysOn := strings.TrimSpace(os.Getenv("DWS_CONNECT_DAEMON_ALWAYSON")) == "true"
	dir, err := connectDaemonDir(dirKey)
	if err != nil {
		return apperrors.NewInternal("create daemon dir: " + err.Error())
	}
	st := daemonState{
		Pid:           os.Getpid(),
		StartUnix:     daemonNow().Unix(),
		LogPath:       daemonLogPath(dir),
		DirKey:        dirKey,
		ClientID:      clientID,
		UnifiedAppID:  unifiedAppID,
		Channel:       channel,
		NotifyStaffID: notifyStaffID,
		Profile:       profile,
		AlwaysOn:      alwaysOn,
	}
	if err := writeDaemonState(dir, st); err != nil {
		return apperrors.NewInternal("write daemon pid file: " + err.Error())
	}
	defer os.Remove(daemonPidPath(dir))

	// Route worker stdout/stderr through the shared size-rotating writer
	// (logging.NewRotatingFile) so a long-running connector's logs don't grow
	// unbounded. The detached parent already redirected our own fds to the same
	// daemon.log; from here we append through the rotator. Fall back to stderr if
	// the rotator can't be opened.
	var out io.Writer = cmd.ErrOrStderr()
	if rot, rerr := logging.NewRotatingFile(daemonLogPath(dir)); rerr == nil {
		defer rot.Close()
		out = rot
	}

	// The supervisor reacts to SIGTERM/SIGINT itself (cmd.Context from root is
	// already wired to these). We use an independent NotifyContext so cancelling
	// it stops the loop and lets us forward the signal to the worker explicitly.
	ctx, stop := daemonSignalContext()
	defer stop()

	workerArgs := buildWorkerArgs(os.Args[1:])
	exe, err := daemonExecutable()
	if err != nil {
		return apperrors.NewInternal("resolve executable: " + err.Error())
	}

	failures := 0
	daemonNotifyStateChange(notifyStaffID, channel, clientID, "started", "")
	for {
		if delay := backoffDelay(failures, daemonBackoffBase, daemonBackoffCap); delay > 0 {
			fmt.Fprintf(out, "[daemon] restarting worker in %s (consecutive failures: %d)\n", delay, failures)
			select {
			case <-ctx.Done():
				daemonNotifyStateChange(notifyStaffID, channel, clientID, "stopped", "")
				return nil
			case <-helperAfter(delay):
			}
		}
		if ctx.Err() != nil {
			daemonNotifyStateChange(notifyStaffID, channel, clientID, "stopped", "")
			return nil
		}

		started := daemonNow()
		worker := daemonCommand(exe, workerArgs...)
		worker.Stdout = out
		worker.Stderr = out
		worker.Env = os.Environ()
		configureWorkerProcessGroup(worker)
		if err := worker.Start(); err != nil {
			fmt.Fprintf(out, "[daemon] failed to start worker: %v\n", err)
			failures++
			if failures >= daemonMaxFastFailures {
				daemonNotifyStateChange(notifyStaffID, channel, clientID, "gave_up", fmt.Sprintf("worker 启动失败 %d 次", failures))
				return apperrors.NewInternal("daemon worker failed to start too many times; giving up")
			}
			continue
		}
		fmt.Fprintf(out, "[daemon] worker started (pid %d)\n", worker.Process.Pid)

		waitErr := superviseWait(ctx, worker)
		cleanupWorkerProcessGroup(worker.Process.Pid)
		ran := daemonNow().Sub(started)

		if ctx.Err() != nil {
			// We were asked to stop; the worker has been (or is being) signalled.
			fmt.Fprintln(out, "[daemon] stop requested, worker shut down; exiting supervisor")
			daemonNotifyStateChange(notifyStaffID, channel, clientID, "stopped", "")
			return nil
		}

		if !alwaysOn {
			fmt.Fprintln(out, "[daemon] worker exited; not restarting (--alwayson not set)")
			daemonNotifyStateChange(notifyStaffID, channel, clientID, "stopped", "worker 退出，未启用 --alwayson")
			return nil
		}

		if ran >= daemonHealthyAfter {
			failures = 0
		} else {
			failures++
		}
		fmt.Fprintf(out, "[daemon] worker exited after %s (err=%v); consecutive failures: %d\n", ran.Round(time.Second), waitErr, failures)
		if failures >= daemonMaxFastFailures {
			daemonNotifyStateChange(notifyStaffID, channel, clientID, "gave_up", fmt.Sprintf("连续崩溃 %d 次", failures))
			return apperrors.NewInternal(fmt.Sprintf("daemon worker crashed %d times in a row; giving up (check %s)", failures, daemonLogPath(dir)))
		}
		daemonNotifyStateChange(notifyStaffID, channel, clientID, "crashed", fmt.Sprintf("worker 退出 (%s)，正在重启", ran.Round(time.Second)))
	}
}

// superviseWait waits for the worker to exit, but if the supervisor's ctx is
// cancelled first it forwards SIGTERM to the worker for a graceful shutdown
// (releasing the Stream single-instance lock) and then waits.
func superviseWait(ctx context.Context, worker *exec.Cmd) error {
	done := make(chan error, 1)
	go func() { done <- worker.Wait() }()
	select {
	case err := <-done:
		return err
	case <-ctx.Done():
		_ = daemonSignalProcess(worker.Process, syscall.SIGTERM)
		select {
		case err := <-done:
			return err
		case <-helperAfter(daemonStopTimeout):
			_ = worker.Process.Kill()
			return <-done
		}
	}
}

// daemonStatus reports connector health to w. It combines two independent
// signals: the daemon supervisor pid file (is a supervisor alive) and the
// connector heartbeat (is the connection live and receiving — see
// connect_health.go). jsonOut emits the machine-readable health report an
// external supervisor (launchd/systemd/pm2/cron) consumes to decide restarts.
func daemonStatus(w io.Writer, dirKey string, jsonOut bool) error {
	dir, err := connectDaemonDir(dirKey)
	if err != nil {
		return apperrors.NewInternal("resolve daemon dir: " + err.Error())
	}
	st, err := readDaemonState(dir)
	if err != nil {
		return apperrors.NewInternal(err.Error())
	}
	supervised := st != nil && st.Pid > 0 && daemonProcessAlive(st.Pid)

	hb, err := readConnectHeartbeat(dir)
	if err != nil {
		return apperrors.NewInternal("read connector heartbeat: " + err.Error())
	}
	report := deriveConnectHealth(hb, supervised, daemonNow())

	if jsonOut {
		data, _ := json.MarshalIndent(report, "", "  ")
		fmt.Fprintln(w, string(data))
		return nil
	}

	var lines []string
	lines = append(lines, fmt.Sprintf("%s  %s", tui.Key("state"), colorConnectState(report.State)))
	if report.Detail != "" {
		lines = append(lines, fmt.Sprintf("%s %s", tui.Key("detail"), report.Detail))
	}
	if report.Pid > 0 {
		lines = append(lines, fmt.Sprintf("%s    %s", tui.Key("pid"), tui.White(fmt.Sprintf("%d", report.Pid))))
	}
	if report.Channel != "" {
		lines = append(lines, fmt.Sprintf("%s %s", tui.Key("channel"), tui.White(report.Channel)))
	}
	if report.ClientID != "" {
		lines = append(lines, fmt.Sprintf("%s %s", tui.Key("client"), tui.White(report.ClientID)))
	}
	if report.UptimeSec > 0 {
		lines = append(lines, fmt.Sprintf("%s %s", tui.Key("uptime"), tui.White((time.Duration(report.UptimeSec)*time.Second).Round(time.Second).String())))
	}
	lines = append(lines, fmt.Sprintf("%s  %s", tui.Key("super"), supervisedLabel(supervised)))
	if hb != nil {
		if report.LastPushAgoSec > 0 {
			lines = append(lines, fmt.Sprintf("%s  %s ago", tui.Key("recv"), tui.White((time.Duration(report.LastPushAgoSec)*time.Second).Round(time.Second).String())))
		} else {
			lines = append(lines, fmt.Sprintf("%s  %s", tui.Key("recv"), tui.Dim("(none since start)")))
		}
		if report.LastError != "" {
			lines = append(lines, fmt.Sprintf("%s  %s", tui.Key("error"), tui.Danger(report.LastError)))
		}
		lines = append(lines, fmt.Sprintf("%s   %s", tui.Key("logs"), tui.Dim(daemonLogPath(dir))))
	}
	return tui.Panel(w, tui.Bold("connect status"), lines)
}

func supervisedLabel(supervised bool) string {
	if supervised {
		return "running (--daemon)"
	}
	return "none (foreground or external)"
}

// connectStopResult 是 `dev connect stop` 的结果 DTO（统一输出 dev 域试点，
// 队列 B112）：进成功信封的 data 层。status 是机器可分支的终态枚举。
type connectStopResult struct {
	Status      string `json:"status"`
	PreviewKind string `json:"preview_kind,omitempty"`
	Pid         int    `json:"pid,omitempty"`
	Detail      string `json:"detail,omitempty"`
}

// daemonStop gracefully stops the connector daemon: SIGTERM the supervisor (it
// forwards to the worker, which releases the lock and Stream connection), poll
// until it exits, escalate to SIGKILL on timeout, and clean up the pid file.
// 人读进度行写 w（调用方传 stderr）；机器可消费的结果作为返回值由 RunE 层
// 包进统一信封写 stdout（契约规范 §5.1：stdout 只承载数据）。
func daemonStopResult(w io.Writer, dirKey string) (*connectStopResult, error) {
	dir, err := connectDaemonDir(dirKey)
	if err != nil {
		return nil, apperrors.NewInternal("resolve daemon dir: " + err.Error())
	}
	st, err := readDaemonState(dir)
	if err != nil {
		return nil, apperrors.NewInternal(err.Error())
	}
	if st == nil || st.Pid <= 0 {
		fmt.Fprintf(w, "connect daemon: not running (nothing to stop)\n")
		return &connectStopResult{Status: "not_running", Detail: "no daemon record found"}, nil
	}
	if !daemonProcessAlive(st.Pid) {
		_ = os.Remove(daemonPidPath(dir))
		// The supervisor is dead, but its worker may still be alive (e.g. the
		// supervisor was kill -9'd). Check the heartbeat for the worker pid and
		// stop it so we don't leave an orphan.
		if hb, _ := readConnectHeartbeat(dir); hb != nil && hb.Pid > 0 && daemonProcessAlive(hb.Pid) {
			fmt.Fprintf(w, "connect daemon: supervisor (pid %d) was dead, stopping orphan worker (pid %d)...\n", st.Pid, hb.Pid)
			if proc, perr := daemonFindProcess(hb.Pid); perr == nil {
				_ = daemonSignalProcess(proc, syscall.SIGTERM)
				deadline := daemonNow().Add(daemonStopTimeout)
				for daemonNow().Before(deadline) {
					if !daemonProcessAlive(hb.Pid) {
						break
					}
					helperSleep(200 * time.Millisecond)
				}
				if daemonProcessAlive(hb.Pid) {
					_ = daemonSignalProcess(proc, syscall.SIGKILL)
					helperSleep(200 * time.Millisecond)
				}
			}
			fmt.Fprintf(w, "connect daemon: orphan worker stopped (pid %d)\n", hb.Pid)
			return &connectStopResult{Status: "orphan_worker_stopped", Pid: hb.Pid, Detail: "supervisor was dead; orphan worker stopped"}, nil
		}
		fmt.Fprintf(w, "connect daemon: was not running (cleaned up stale pid file for pid %d)\n", st.Pid)
		return &connectStopResult{Status: "cleaned_stale", Pid: st.Pid, Detail: "cleaned up stale pid file"}, nil
	}
	proc, err := daemonFindProcess(st.Pid)
	if err != nil {
		return nil, apperrors.NewInternal(fmt.Sprintf("find daemon process %d: %v", st.Pid, err))
	}
	if err := daemonSignalProcess(proc, syscall.SIGTERM); err != nil {
		return nil, apperrors.NewInternal(fmt.Sprintf("signal daemon %d: %v", st.Pid, err))
	}
	fmt.Fprintf(w, "sent SIGTERM to connect daemon (pid %d), waiting for graceful stop...\n", st.Pid)

	deadline := daemonNow().Add(daemonStopTimeout)
	for daemonNow().Before(deadline) {
		if !daemonProcessAlive(st.Pid) {
			_ = os.Remove(daemonPidPath(dir))
			fmt.Fprintf(w, "connect daemon stopped (pid %d)\n", st.Pid)
			return &connectStopResult{Status: "stopped", Pid: st.Pid}, nil
		}
		helperSleep(200 * time.Millisecond)
	}
	// Graceful window elapsed; force kill.
	_ = daemonSignalProcess(proc, syscall.SIGKILL)
	helperSleep(200 * time.Millisecond)
	_ = os.Remove(daemonPidPath(dir))
	fmt.Fprintf(w, "connect daemon did not stop in %s; sent SIGKILL (pid %d)\n", daemonStopTimeout, st.Pid)
	return &connectStopResult{Status: "force_killed", Pid: st.Pid, Detail: "graceful stop timed out; SIGKILL sent"}, nil
}

// daemonStop keeps the established helper contract for legacy callers and
// platform coverage. Unified-result commands use daemonStopResult when they
// need the typed terminal payload.
func daemonStop(w io.Writer, dirKey string) error {
	_, err := daemonStopResult(w, dirKey)
	return err
}

// newDevAppRobotConnectStatusCommand implements `dws devapp robot connect
// status`: report whether a background connector daemon is running for the
// robot identified by --robot-client-id (or --unified-app-id).
func newDevAppRobotConnectStatusCommand() *cobra.Command {
	cmd := &cobra.Command{
		Use:               "status",
		Short:             "查看连接器健康状态（healthy/degraded/down，pid、收发活动、日志路径；--json 供外部托管消费）",
		Args:              cobra.NoArgs,
		DisableAutoGenTag: true,
		RunE: func(cmd *cobra.Command, _ []string) error {
			dirKey, err := connectDaemonDirKeyFromFlags(cmd)
			if err != nil {
				return err
			}
			jsonOut, _ := cmd.Flags().GetBool("json")
			return daemonStatus(cmd.OutOrStdout(), dirKey, jsonOut)
		},
	}
	preferLegacyLeaf(cmd)
	cmd.Flags().String("robot-client-id", "", "机器人 clientId（定位守护进程）")
	cmd.Flags().String("unified-app-id", "", "统一应用 ID（当未用 clientId 起守护进程时定位）")
	cmd.Flags().Bool("json", false, "以 JSON 输出健康报告（供 launchd/systemd/pm2/cron 判断是否重启）")
	DeclareLeafMetadata(cmd, LeafSpec{
		Safety: contract.SafetySpec{
			// Preserve the published Schema safety tuple during the unified-result
			// pilot. Correcting this historical classification is a separate,
			// explicitly reviewed interface change.
			Effect: "write", Risk: "medium",
			Confirmation: "not_required", Idempotency: "unknown",
		},
		Contract: LeafContract{
			Identity: contract.ToolIdentitySpec{
				ProductID:      "dev",
				Name:           "connect_status",
				CanonicalPath:  "dev.connect_status",
				CLIPath:        "dev connect status",
				PrimaryCLIPath: "dev connect status",
			},
			Description: "查看后台连接器守护进程状态",
			Result: &contract.ResultSpec{
				Outcomes:   []contract.ResultOutcome{contract.ResultOutcomeSuccess, contract.ResultOutcomeFailure},
				DataSchema: json.RawMessage(`{"type":"object","description":"本地连接器守护进程状态","properties":{"state":{"type":"string","description":"归一化健康状态","enum":["healthy","degraded","down","not_running"]},"pid":{"type":"integer","description":"守护进程 PID；未运行时可能缺省"},"channel":{"type":"string","description":"连接器使用的通道"},"clientId":{"type":"string","description":"连接器客户端 ID"},"unifiedAppId":{"type":"string","description":"关联的统一应用 ID"},"supervised":{"type":"boolean","description":"是否由本地 supervisor 管理"},"lastError":{"type":"string","description":"最近一次连接错误；健康时可能缺省"}},"required":["state","supervised"],"additionalProperties":true}`),
			},
			Interface: &contract.InterfaceSpec{
				Mode:         "local",
				Availability: "available",
				Reason:       "命令仅操作本地进程或策略文件，不调用 MCP 接口",
			},
			Selection: contract.SelectionSpec{
				AgentSummary: "查看后台连接器守护进程状态",
				UseWhen:      []string{"需要确认本地 connect 守护进程是否在跑、pid/日志路径"},
				AvoidWhen:    []string{"要停止守护进程时用 dev connect stop"},
				Examples:     []string{"dws dev connect status --unified-app-id <unifiedAppId> --json"},
			},
		},
	})
	return cmd
}

// newDevAppRobotConnectStopCommand implements `dws devapp robot connect stop`:
// gracefully stop the background connector daemon (SIGTERM, escalate to SIGKILL
// on timeout) and clean up its pid file.
func newDevAppRobotConnectStopCommand() *cobra.Command {
	cmd := &cobra.Command{
		Use:               "stop",
		Short:             "优雅停止后台连接器守护进程（释放单实例锁与 Stream 连接）",
		Args:              cobra.NoArgs,
		DisableAutoGenTag: true,
		RunE: func(cmd *cobra.Command, _ []string) error {
			dirKey, err := connectDaemonDirKeyFromFlags(cmd)
			if err != nil {
				return err
			}
			if commandDryRun(cmd) {
				preview := &connectStopResult{
					Status:      "preview",
					PreviewKind: contract.DryRunPreviewPlan,
					Detail:      "would send SIGTERM to the selected daemon and escalate to SIGKILL only after the graceful timeout",
				}
				env := &output.Envelope{OK: true, Outcome: output.OutcomeSuccess, DryRun: true, Data: preview}
				return writeDevRolloutResult(cmd, output.Success(preview, output.WithDryRun()), env, output.FormatJSON)
			}
			result, err := daemonStopResult(cmd.ErrOrStderr(), dirKey)
			if err != nil {
				return err
			}
			// 统一输出 dev 域试点（队列 B112）：结果信封化（ok:true + data）。
			env := &output.Envelope{OK: true, Outcome: output.OutcomeSuccess, Data: result}
			return writeDevRolloutResult(cmd, output.Success(result), env, output.FormatJSON)
		},
	}
	preferLegacyLeaf(cmd)
	cmd.Flags().String("robot-client-id", "", "机器人 clientId（定位守护进程）")
	cmd.Flags().String("unified-app-id", "", "统一应用 ID（当未用 clientId 起守护进程时定位）")
	DeclareLeafMetadata(cmd, LeafSpec{
		OutputRollout: output.RolloutUnifiedActive,
		Safety: contract.SafetySpec{
			// Preserve the published Schema safety tuple during the unified-result
			// pilot. The stricter classification belongs in a separate interface
			// migration rather than this framework-only compatibility rollout.
			Effect: "write", Risk: "medium",
			Confirmation: "not_required", Idempotency: "unknown",
		},
		Validate: func(c *cobra.Command, _ []string) error {
			_, err := connectDaemonDirKeyFromFlags(c)
			return err
		},
		Contract: LeafContract{
			Identity: contract.ToolIdentitySpec{
				ProductID:      "dev",
				Name:           "connect_stop",
				CanonicalPath:  "dev.connect_stop",
				CLIPath:        "dev connect stop",
				PrimaryCLIPath: "dev connect stop",
			},
			Description: "优雅停止后台连接器守护进程",
			DryRun:      &contract.DryRunSpec{PreviewKind: contract.DryRunPreviewPlan, RemoteReads: false},
			Interface: &contract.InterfaceSpec{
				Mode:         "local",
				Availability: "available",
				Reason:       "命令仅操作本地进程或策略文件，不调用 MCP 接口",
			},
			Selection: contract.SelectionSpec{
				AgentSummary: "优雅停止后台连接器守护进程",
				UseWhen:      []string{"用户明确要求停止本地 Stream/连接器守护进程"},
				AvoidWhen:    []string{"只想查看状态时用 dev connect status"},
				Examples:     []string{"dws dev connect stop --unified-app-id <unifiedAppId>", "dws dev connect stop --unified-app-id <unifiedAppId> --dry-run --format json"},
			},
		},
	})
	return cmd
}

// connectRestartResult 是 `dev connect restart` 的结果 DTO（队列 B113）：
// 进成功信封的 data 层。pid 是重启后新守护进程的 pid（从刷新后的
// daemon-state.json 回读；子进程自身的结果信封由本命令信封取代，不转发）。
type connectRestartResult struct {
	Status       string `json:"status"`
	PreviewKind  string `json:"preview_kind,omitempty"`
	Pid          int    `json:"pid,omitempty"`
	DirKey       string `json:"dirKey,omitempty"`
	UnifiedAppID string `json:"unifiedAppId,omitempty"`
	Channel      string `json:"channel,omitempty"`
	Command      string `json:"command,omitempty"`
}

func validateConnectRestart(cmd *cobra.Command) error {
	dirKey, err := connectDaemonDirKeyFromFlags(cmd)
	if err != nil {
		return err
	}
	// An explicit unified app ID is sufficient to produce a side-effect-free
	// restart plan. Runtime execution still requires persisted daemon state.
	if commandDryRun(cmd) && devAppStringFlag(cmd, "unified-app-id") != "" {
		return nil
	}
	dir, err := connectDaemonDir(dirKey)
	if err != nil {
		return apperrors.NewInternal("resolve daemon dir: " + err.Error())
	}
	st, err := readDaemonState(dir)
	if err != nil {
		return apperrors.NewInternal(err.Error())
	}
	if st == nil {
		return apperrors.NewValidation("未找到连接器记录（没有 daemon.pid）；请用 `dws dev connect --daemon` 首次启动")
	}
	if st.UnifiedAppID == "" {
		return apperrors.NewValidation("该连接器未持久化 unifiedAppId（可能是用 --robot-client-id/--robot-client-secret 直接启动的，无法安全重启：clientSecret 不落盘）；请停掉后用 `dws dev connect --daemon --unified-app-id <uappid>` 重新启动，之后 restart 就能自动从 credentials get 拉密钥、命令行不出现 secret")
	}
	return nil
}

func writeConnectRestartPreview(cmd *cobra.Command, dirKey, unifiedAppID, channel string) error {
	preview := &connectRestartResult{
		Status:       "preview",
		PreviewKind:  contract.DryRunPreviewPlan,
		DirKey:       dirKey,
		UnifiedAppID: unifiedAppID,
		Channel:      channel,
		Command:      fmt.Sprintf("dws dev connect --daemon --unified-app-id %s", unifiedAppID),
	}
	env := &output.Envelope{OK: true, Outcome: output.OutcomeSuccess, DryRun: true, Data: preview}
	return writeDevRolloutResult(cmd, output.Success(preview, output.WithDryRun()), env, output.FormatJSON)
}

// newDevAppRobotConnectRestartCommand implements `dws devapp robot connect
// restart`: stop the running daemon (if any) then re-launch it using the
// persisted unifiedAppId so credentials are freshly fetched from the dev
// platform — no secrets stored on disk.
func newDevAppRobotConnectRestartCommand() *cobra.Command {
	cmd := &cobra.Command{
		Use:               "restart",
		Short:             "重启连接器守护进程（通过持久化的 unifiedAppId 重新拉取密钥，无需本地存密钥）",
		Args:              cobra.NoArgs,
		DisableAutoGenTag: true,
		RunE: func(cmd *cobra.Command, _ []string) error {
			dirKey, err := connectDaemonDirKeyFromFlags(cmd)
			if err != nil {
				return err
			}
			explicitUnifiedAppID := devAppStringFlag(cmd, "unified-app-id")
			dir, err := connectDaemonDir(dirKey)
			if err != nil {
				return apperrors.NewInternal("resolve daemon dir: " + err.Error())
			}
			st, err := readDaemonState(dir)
			if err != nil {
				return apperrors.NewInternal(err.Error())
			}
			if st == nil && commandDryRun(cmd) && explicitUnifiedAppID != "" {
				return writeConnectRestartPreview(cmd, dirKey, explicitUnifiedAppID, "")
			}
			if st == nil {
				return apperrors.NewValidation("未找到连接器记录（没有 daemon.pid）；请用 `dws dev connect --daemon` 首次启动")
			}
			unifiedAppID := st.UnifiedAppID
			if unifiedAppID == "" {
				return apperrors.NewValidation("该连接器未持久化 unifiedAppId（可能是用 --robot-client-id/--robot-client-secret 直接启动的，无法安全重启：clientSecret 不落盘）；请停掉后用 `dws dev connect --daemon --unified-app-id <uappid>` 重新启动，之后 restart 就能自动从 credentials get 拉密钥、命令行不出现 secret")
			}
			if commandDryRun(cmd) {
				return writeConnectRestartPreview(cmd, dirKey, unifiedAppID, st.Channel)
			}
			// Stop the running daemon first (ignore "not running" — that's fine).
			// 统一输出 dev 域试点（队列 B113）：stopping/restarting 进度行改走
			// stderr，stdout 只承载最终结果信封（契约规范 §5.1）。
			fmt.Fprintln(cmd.ErrOrStderr(), "stopping existing daemon...")
			if err := daemonStop(cmd.ErrOrStderr(), dirKey); err != nil {
				fmt.Fprintf(cmd.ErrOrStderr(), "warning: stop returned %v (continuing with restart)\n", err)
			}
			// Re-exec dws dev connect --daemon with the stored flags. An explicit
			// --profile on this invocation overrides the persisted one.
			exe, err := daemonExecutable()
			if err != nil {
				return apperrors.NewInternal("resolve executable: " + err.Error())
			}
			profile := st.Profile
			if v, _ := cmd.Root().PersistentFlags().GetString("profile"); strings.TrimSpace(v) != "" {
				profile = strings.TrimSpace(v)
			}
			args := []string{"dev", "connect", "--daemon", "--unified-app-id", unifiedAppID}
			if st.Channel != "" {
				args = append(args, "--channel", st.Channel)
			}
			if st.NotifyStaffID != "" {
				args = append(args, "--notify-staff-id", st.NotifyStaffID)
			}
			if profile != "" {
				args = append(args, "--profile", profile)
			}
			if st.AlwaysOn {
				args = append(args, "--alwayson")
			}
			fmt.Fprintf(cmd.ErrOrStderr(), "restarting connector: dws %s\n", strings.Join(args, " "))
			// Run synchronously: `--daemon` itself detaches the supervisor and
			// returns quickly, so waiting here costs nothing and lets a failed
			// relaunch (e.g. credential fetch error) surface as a non-zero exit
			// instead of a silent success. 子进程自己的输出（含它自己的结果
			// 信封）不再转发到本命令的 stdout——由本命令的信封统一承载结果，
			// 避免两个 JSON 文档拼接破坏 `| jq`（契约规范 §5.1）。
			restartCmd := daemonCommand(exe, args...)
			restartCmd.Stdout = cmd.ErrOrStderr()
			restartCmd.Stderr = cmd.ErrOrStderr()
			restartCmd.Stdin = nil
			if err := restartCmd.Run(); err != nil {
				return apperrors.NewInternal(fmt.Sprintf("重启失败（旧守护进程已停止，连接器记录已清除）；恢复请手动执行: dws %s", strings.Join(args, " ")))
			}
			// 回读刷新后的 daemon-state.json 拿新 supervisor pid（restart 是同步
			// re-exec，子进程落盘早于返回；读不到则 pid 缺席，status 仍诚实）。
			result := &connectRestartResult{
				Status:       "restarted",
				DirKey:       dirKey,
				UnifiedAppID: unifiedAppID,
				Channel:      st.Channel,
				Command:      "dws " + strings.Join(args, " "),
			}
			if fresh, rerr := readDaemonState(dir); rerr == nil && fresh != nil {
				result.Pid = fresh.Pid
			}
			env := &output.Envelope{OK: true, Outcome: output.OutcomeSuccess, Data: result}
			return writeDevRolloutResult(cmd, output.Success(result), env, output.FormatJSON)
		},
	}
	preferLegacyLeaf(cmd)
	cmd.Flags().String("robot-client-id", "", "机器人 clientId（定位守护进程）")
	cmd.Flags().String("unified-app-id", "", "统一应用 ID（当未用 clientId 起守护进程时定位）")
	DeclareLeafMetadata(cmd, LeafSpec{
		OutputRollout: output.RolloutUnifiedActive,
		Safety:        contract.SafetySpec{Effect: "destructive", Risk: "high", Confirmation: "not_required", Idempotency: "unknown"},
		Validate:      func(c *cobra.Command, _ []string) error { return validateConnectRestart(c) },
		Contract: LeafContract{
			Identity:    contract.ToolIdentitySpec{ProductID: "dev", Name: "connect_restart", CanonicalPath: "dev.connect_restart", CLIPath: "dev connect restart", PrimaryCLIPath: "dev connect restart"},
			Description: "使用持久化配置重启本地连接器守护进程",
			DryRun:      &contract.DryRunSpec{PreviewKind: contract.DryRunPreviewPlan, RemoteReads: false},
			Interface:   &contract.InterfaceSpec{Mode: "composite", Availability: "available", Reason: "命令组合本地守护进程管理与远端凭证重新获取，不对应单一 MCP 接口"},
			Selection:   contract.SelectionSpec{AgentSummary: "重启本地连接器守护进程", UseWhen: []string{"连接器需要恢复或重载配置"}, AvoidWhen: []string{"仅查看状态时使用 dev connect status"}, Examples: []string{"dws dev connect restart --unified-app-id <unifiedAppId>", "dws dev connect restart --unified-app-id <unifiedAppId> --dry-run --format json"}},
		},
	})
	return cmd
}

// newDevAppRobotConnectListCommand implements `dws dev connect list`: enumerate
// every connector on this machine and its health, so a developer running
// several robots sees at a glance which are alive/degraded/down without
// querying each clientId. `--json` emits the array for scripts.
func newDevAppRobotConnectListCommand(runner executor.Runner) *cobra.Command {
	cmd := &cobra.Command{
		Use:               "list",
		Short:             "列出本机所有连接器及健康状态（healthy/degraded/down）；--json 供脚本消费",
		Args:              cobra.NoArgs,
		DisableAutoGenTag: true,
		RunE: func(cmd *cobra.Command, _ []string) error {
			reports, err := listConnectors(daemonNow())
			if err != nil {
				return apperrors.NewInternal(err.Error())
			}
			resolveAppNames(cmd, runner, reports)
			w := cmd.OutOrStdout()
			if jsonOut, _ := cmd.Flags().GetBool("json"); jsonOut {
				if reports == nil {
					reports = []connectHealthReport{}
				}
				data, _ := json.MarshalIndent(reports, "", "  ")
				fmt.Fprintln(w, string(data))
				return nil
			}
			if len(reports) == 0 {
				fmt.Fprintln(w, "no connectors found")
				return nil
			}
			return writeConnectListTable(w, reports)
		},
	}
	preferLegacyLeaf(cmd)
	cmd.Flags().Bool("json", false, "以 JSON 数组输出（供脚本消费）")
	DeclareLeafMetadata(cmd, LeafSpec{
		Safety: contract.SafetySpec{Effect: "read", Risk: "low", Confirmation: "not_required", Idempotency: "idempotent"},
		Contract: LeafContract{
			Identity:    contract.ToolIdentitySpec{ProductID: "dev", Name: "connect_list", CanonicalPath: "dev.connect_list", CLIPath: "dev connect list", PrimaryCLIPath: "dev connect list"},
			Description: "列出本机连接器及其健康状态",
			Result: &contract.ResultSpec{
				Outcomes:   []contract.ResultOutcome{contract.ResultOutcomeSuccess, contract.ResultOutcomeFailure},
				DataSchema: json.RawMessage(`{"type":"array","description":"本机连接器及其健康状态列表","items":{"type":"object","properties":{"state":{"type":"string","description":"归一化健康状态","enum":["healthy","degraded","down","not_running"]},"pid":{"type":"integer","description":"守护进程 PID；未运行时可能缺省"},"channel":{"type":"string","description":"连接器使用的通道"},"clientId":{"type":"string","description":"连接器客户端 ID"},"unifiedAppId":{"type":"string","description":"关联的统一应用 ID"},"supervised":{"type":"boolean","description":"是否由本地 supervisor 管理"}},"required":["state","supervised"],"additionalProperties":true}}`),
			},
			Interface: &contract.InterfaceSpec{Mode: "composite", Availability: "available", Reason: "命令组合本地连接器状态与可选远端应用名称解析，不对应单一 MCP 接口"},
			Selection: contract.SelectionSpec{AgentSummary: "列出本机全部连接器及健康状态", UseWhen: []string{"需要查看本机连接器清单"}, AvoidWhen: []string{"只检查一个连接器时使用 dev connect status"}, Examples: []string{"dws dev connect list --json"}},
		},
	})
	return cmd
}

// resolveAppNames calls list_dev_app once to build a unifiedAppId→name map,
// then fills in AppName on each report. Failures are silent (name stays empty)
// so the list still works offline or when the API is unreachable.
func resolveAppNames(cmd *cobra.Command, runner executor.Runner, reports []connectHealthReport) {
	need := false
	for i := range reports {
		if reports[i].UnifiedAppID != "" {
			need = true
			break
		}
	}
	if !need {
		return
	}
	nameMap, err := devAppNameMap(cmd, runner)
	if err != nil || nameMap == nil {
		return
	}
	for i := range reports {
		if reports[i].UnifiedAppID == "" {
			continue
		}
		if name, ok := nameMap[reports[i].UnifiedAppID]; ok && name != "" {
			reports[i].AppName = name
		}
	}
}

// devAppNameMap calls list_dev_app with pagination to build a full
// unifiedAppId→appName map. It is best-effort: any error returns an empty map.
func devAppNameMap(cmd *cobra.Command, runner executor.Runner) (map[string]string, error) {
	out := make(map[string]string)
	cursor := ""
	for page := 0; page < 20; page++ {
		params := map[string]any{"pageSize": 100}
		if cursor != "" {
			params["cursor"] = cursor
		}
		inv := executor.NewHelperInvocation(cobracmd.LegacyCommandPath(cmd), devAppProduct, devAppListTool, params)
		res, err := runner.Run(cmd.Context(), inv)
		if err != nil {
			return out, err
		}
		payload := devAppConnectUnwrap(res.Response)
		items := devAppConnectList(payload)
		for _, item := range items {
			uid := devAppConnectFirst(item, "unifiedAppId", "id")
			name := devAppConnectFirst(item, "name", "appName")
			if uid != "" && name != "" {
				out[uid] = name
			}
		}
		hasMore := false
		if v, ok := payload["hasMore"].(bool); ok {
			hasMore = v
		}
		if !hasMore {
			break
		}
		cursor = devAppConnectFirst(payload, "nextCursor", "cursor")
		if cursor == "" {
			break
		}
	}
	return out, nil
}

// devAppConnectList extracts the array of app items from a list_dev_app payload,
// tolerating various wrapper shapes.
func devAppConnectList(payload map[string]any) []map[string]any {
	if payload == nil {
		return nil
	}
	for _, key := range []string{"items", "list", "data"} {
		if arr, ok := payload[key].([]any); ok {
			out := make([]map[string]any, 0, len(arr))
			for _, e := range arr {
				if m, ok := e.(map[string]any); ok {
					out = append(out, m)
				}
			}
			return out
		}
	}
	return nil
}

// connectDaemonDirKeyFromFlags resolves the daemon directory key from the
// status/stop flags, requiring at least one identifier.
func connectDaemonDirKeyFromFlags(cmd *cobra.Command) (string, error) {
	clientID := devAppStringFlag(cmd, "robot-client-id")
	unifiedAppID := devAppStringFlag(cmd, "unified-app-id")
	dirKey := daemonDirKey(clientID, unifiedAppID)
	if dirKey == "" {
		return "", apperrors.NewValidation("需要 --robot-client-id 或 --unified-app-id 以定位守护进程")
	}
	return dirKey, nil
}

func colorConnectState(state string) string {
	switch state {
	case healthHealthy:
		return tui.Success(state)
	case healthDegraded:
		return tui.Warning(state)
	case healthDown, healthNotRunning:
		return tui.Danger(state)
	default:
		return tui.Cyan(state)
	}
}

func writeConnectListTable(w io.Writer, reports []connectHealthReport) error {
	type col struct {
		header string
		width  int
	}
	cols := []col{
		{"STATE", 11},
		{"APP NAME", 8},
		{"CLIENT", 8},
		{"PID", 6},
		{"CHANNEL", 7},
		{"UPTIME", 6},
	}
	// compute column widths from data
	for _, r := range reports {
		if w := tui.PlainRuneWidth(r.AppName); w > cols[1].width {
			cols[1].width = w
		}
		if w := tui.PlainRuneWidth(r.ClientID); w > cols[2].width {
			cols[2].width = w
		}
		if w := tui.PlainRuneWidth(r.Channel); w > cols[4].width {
			cols[4].width = w
		}
	}
	for i := range cols {
		if cols[i].width > tui.MaxTableColumnWidth {
			cols[i].width = tui.MaxTableColumnWidth
		}
	}

	writeBorder := func(left, mid, right string, colorFn func(string) string) {
		fmt.Fprint(w, colorFn(left))
		for i, c := range cols {
			if i > 0 {
				fmt.Fprint(w, colorFn(mid))
			}
			fmt.Fprint(w, colorFn(strings.Repeat("─", c.width+2)))
		}
		fmt.Fprintln(w, colorFn(right))
	}
	writeRowCells := func(cells []string) {
		fmt.Fprint(w, tui.Gray("│"))
		for i, c := range cols {
			cell := ""
			if i < len(cells) {
				cell = cells[i]
			}
			fmt.Fprintf(w, " %s ", tui.PadRightANSI(cell, c.width))
			fmt.Fprint(w, tui.Gray("│"))
		}
		fmt.Fprintln(w)
	}

	// header
	writeBorder("╭", "┬", "╮", tui.Blue)
	headers := make([]string, len(cols))
	for i, c := range cols {
		headers[i] = tui.Brand(c.header)
	}
	writeRowCells(headers)
	writeBorder("├", "┼", "┤", tui.Gray)

	// rows
	for _, r := range reports {
		uptime := tui.Dim("-")
		if r.UptimeSec > 0 {
			uptime = tui.White((time.Duration(r.UptimeSec) * time.Second).Round(time.Second).String())
		}
		pid := tui.Dim("-")
		if r.Pid > 0 {
			pid = tui.White(fmt.Sprintf("%d", r.Pid))
		}
		channel := tui.Dim("-")
		if r.Channel != "" {
			channel = tui.White(r.Channel)
		}
		appName := tui.Dim("-")
		if r.AppName != "" {
			appName = tui.White(r.AppName)
		}
		writeRowCells([]string{
			colorConnectState(r.State),
			appName,
			tui.White(r.ClientID),
			pid,
			channel,
			uptime,
		})
	}

	writeBorder("╰", "┴", "╯", tui.Blue)
	return nil
}

// daemonNotifyStateChange sends a DingTalk message to the configured staff
// when the connector state changes. No-op when notifyStaffID is empty. It
// execs `dws chat message send` as a subprocess (fire-and-forget) so the
// supervisor is never blocked by notification delivery.
func daemonNotifyStateChange(notifyStaffID, channel, clientID, event, detail string) {
	if notifyStaffID == "" {
		return
	}
	var msg string
	switch event {
	case "started":
		msg = fmt.Sprintf("机器人已启动 ✅\n渠道: %s\nclientId: %s", channel, clientID)
	case "stopped":
		msg = fmt.Sprintf("机器人已停止 ⏹️\n渠道: %s\nclientId: %s", channel, clientID)
	case "crashed":
		msg = fmt.Sprintf("机器人已崩溃 ⚠️\n渠道: %s\n%s\n正在自动重启...", channel, detail)
	case "gave_up":
		msg = fmt.Sprintf("机器人重启失败 ❌\n渠道: %s\n%s\n请检查日志后手动重启", channel, detail)
	default:
		return
	}
	exe, err := daemonExecutable()
	if err != nil {
		return
	}
	go func() {
		cmd := daemonCommand(exe, "chat", "message", "send",
			"--staff-id", notifyStaffID,
			"--text", msg,
			"--ai-tag=false",
			"--yes",
			"--format", "json",
		)
		cmd.Stdin = nil
		cmd.Stdout = nil
		cmd.Stderr = nil
		_ = cmd.Run()
	}()
}