fix(connect): harden daemon restart lifecycle (#599)

go.mod

@@ -13,7 +13,7 @@ require (
 	github.com/gorilla/websocket v1.5.0
 	github.com/itchyny/gojq v0.12.18
 	github.com/muesli/termenv v0.16.0
-	github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.1
+	github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.2-0.20260705041131-325e7c1049ad
 	github.com/spf13/cobra v1.10.2
 	github.com/zalando/go-keyring v0.2.8
 	golang.org/x/crypto v0.49.0

go.sum

@@ -88,8 +88,8 @@ github.com/muesli/cancelreader v0.2.2 h1:3I4Kt4BQjOR54NavqnDogx/MIoWBFa0StPA8ELU
 github.com/muesli/cancelreader v0.2.2/go.mod h1:3XuTXfFS2VjM+HTLZY9Ak0l6eUKfijIfMUZ4EgX0QYo=
 github.com/muesli/termenv v0.16.0 h1:S5AlUN9dENB57rsbnkPyfdGuWIlkmzJjbFf0Tf5FWUc=
 github.com/muesli/termenv v0.16.0/go.mod h1:ZRfOIKPFDYQoDFF4Olj7/QJbW60Ol/kL1pU3VfY/Cnk=
-github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.1 h1:Lb/Uzkiw2Ugt2Xf03J5wmv81PdkYOiWbI8CNBi1boC8=
-github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.1/go.mod h1:ln3IqPYYocZbYvl9TAOrG/cxGR9xcn4pnZRLdCTEGEU=
+github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.2-0.20260705041131-325e7c1049ad h1:Bb4I+suYd+ehQ8e22aimLLze+5XTN3+WTc/x2LafmH8=
+github.com/open-dingtalk/dingtalk-stream-sdk-go v0.9.2-0.20260705041131-325e7c1049ad/go.mod h1:ln3IqPYYocZbYvl9TAOrG/cxGR9xcn4pnZRLdCTEGEU=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
 github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
 github.com/rivo/uniseg v0.4.7 h1:WUdvkW8uEhrYfLC4ZzdpI2ztxP1I582+49Oc5Mq64VQ=

internal/helpers/connect_daemon.go

@@ -113,6 +113,50 @@ func connectDaemonDir(dirKey string) (string, error) {
 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 := os.CreateTemp(dir, ".dws-daemon-*")
+	if err != nil {
+		return "", err
+	}
+	tmpPath := tmp.Name()
+	defer os.Remove(tmpPath)
+
+	if err := tmp.Chmod(daemonExecutablePerm); err != nil {
+		tmp.Close()
+		return "", err
+	}
+	if _, err := io.Copy(tmp, in); err != nil {
+		tmp.Close()
+		return "", err
+	}
+	if err := tmp.Sync(); err != nil {
+		tmp.Close()
+		return "", err
+	}
+	if err := tmp.Close(); err != nil {
+		return "", err
+	}
+
+	dst := daemonExecutablePath(dir)
+	if err := os.Rename(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.
@@ -169,8 +213,9 @@ func backoffDelay(consecutiveFailures int, base, maxDelay time.Duration) time.Du
 }
 
 const (
-	daemonBackoffBase = time.Second
-	daemonBackoffCap  = 60 * time.Second
+	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).
@@ -226,6 +271,10 @@ func startDaemon(cmd *cobra.Command, dirKey, clientID, unifiedAppID, channel, no
 	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)
@@ -386,6 +435,7 @@ func runSupervisor(cmd *cobra.Command) error {
 		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++
@@ -398,6 +448,7 @@ func runSupervisor(cmd *cobra.Command) error {
 		fmt.Fprintf(out, "[daemon] worker started (pid %d)\n", worker.Process.Pid)
 
 		waitErr := superviseWait(ctx, worker)
+		cleanupWorkerProcessGroup(worker.Process.Pid)
 		ran := time.Since(started)
 
 		if ctx.Err() != nil {

internal/helpers/connect_daemon_test.go

@@ -68,6 +68,36 @@ func TestDaemonDirKey(t *testing.T) {
 	}
 }
 
+func TestStageDaemonExecutable(t *testing.T) {
+	dir := t.TempDir()
+	src := dir + "/source"
+	if err := os.WriteFile(src, []byte("test-binary"), 0o700); err != nil {
+		t.Fatal(err)
+	}
+
+	dst, err := stageDaemonExecutable(src, dir)
+	if err != nil {
+		t.Fatalf("stageDaemonExecutable: %v", err)
+	}
+	if dst != daemonExecutablePath(dir) {
+		t.Fatalf("path = %q, want %q", dst, daemonExecutablePath(dir))
+	}
+	got, err := os.ReadFile(dst)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if string(got) != "test-binary" {
+		t.Fatalf("content = %q, want test-binary", got)
+	}
+	info, err := os.Stat(dst)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if info.Mode().Perm() != daemonExecutablePerm {
+		t.Fatalf("mode = %o, want %o", info.Mode().Perm(), daemonExecutablePerm)
+	}
+}
+
 func TestBuildWorkerArgs(t *testing.T) {
 	in := []string{"devapp", "robot", "connect", "--daemon", "--robot-client-id", "abc", "--channel=claudecode"}
 	got := buildWorkerArgs(in)

internal/helpers/connect_daemon_unix.go

@@ -36,3 +36,17 @@ func detachSysProcAttr() *syscall.SysProcAttr {
 func applyDetach(cmd *exec.Cmd) {
 	cmd.SysProcAttr = detachSysProcAttr()
 }
+
+// configureWorkerProcessGroup isolates the connector worker and every local
+// agent process it starts (opencode serve, codex app-server, etc.) in one group.
+func configureWorkerProcessGroup(cmd *exec.Cmd) {
+	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
+}
+
+// cleanupWorkerProcessGroup removes descendants left behind when the worker
+// exits abruptly and cannot run its normal deferred cleanup.
+func cleanupWorkerProcessGroup(pid int) {
+	if pid > 0 {
+		_ = syscall.Kill(-pid, syscall.SIGKILL)
+	}
+}

internal/helpers/connect_daemon_unix_test.go

@@ -0,0 +1,29 @@
+// 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.
+
+//go:build !windows
+
+package helpers
+
+import (
+	"os/exec"
+	"testing"
+)
+
+func TestConfigureWorkerProcessGroup(t *testing.T) {
+	cmd := exec.Command("true")
+	configureWorkerProcessGroup(cmd)
+	if cmd.SysProcAttr == nil || !cmd.SysProcAttr.Setpgid {
+		t.Fatal("worker must start in its own process group")
+	}
+}

internal/helpers/connect_daemon_windows.go

@@ -25,3 +25,7 @@ const daemonDetachSupported = false
 
 // applyDetach is a no-op on Windows; daemon mode is rejected earlier.
 func applyDetach(_ *exec.Cmd) {}
+
+func configureWorkerProcessGroup(_ *exec.Cmd) {}
+
+func cleanupWorkerProcessGroup(_ int) {}

internal/helpers/connect_stream.go

@@ -1117,7 +1117,7 @@ func runStreamConnector(ctx context.Context, channel, clientID, clientSecret str
 		mediaCli = newAICardClient(clientID, clientSecret, "")
 	}
 
-	keepAlive := envDurationMS("DWS_CONNECT_KEEPALIVE_MS", 30000)
+	keepAlive := envDurationMS("DWS_CONNECT_KEEPALIVE_MS", 30*time.Second)
 	fmt.Fprintf(os.Stderr, "[connect] keepAlive=%s autoReconnect=true\n", keepAlive)
 	cli := client.NewStreamClient(
 		client.WithAppCredential(client.NewAppCredentialConfig(clientID, clientSecret)),