fix(event): support Windows bus lifecycle

.changes/windows-event-bus-lifecycle.md

@@ -0,0 +1,5 @@
+---
+category: Fixed
+---
+
+- **Windows event bus lifecycle** — start event consumers without unsupported inherited file descriptors, stop buses through local IPC with a termination fallback, and preserve subscription cleanup when startup fails.

internal/app/event_command.go

@@ -1220,7 +1220,8 @@ func newEventStopCommandWithFlags(globalFlags ...*GlobalFlags) *cobra.Command {
 			editionName := editionNameOrDefault()
 			clientIDHash := dwsevent.ClientIDHash(clientID)
 			workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindAppStream, clientIDHash)
-			if err := eventStopBus(busctl.StopConfig{WorkDir: workDir}); err != nil {
+			ipcEndpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindAppStream, clientIDHash)
+			if err := eventStopBus(busctl.StopConfig{WorkDir: workDir, IPCEndpoint: ipcEndpoint}); err != nil {
 				if errors.Is(err, busctl.ErrNotRunning) {
 					fmt.Fprintln(c.OutOrStdout(), "bus is not running")
 					return nil

internal/app/event_personal_command.go

@@ -1220,7 +1220,7 @@ func runPersonalEventStop(c *cobra.Command, opts personalStopOptions) error {
 	}
 
 	busState := "personal bus stopped"
-	if err := personalStopBus(busctl.StopConfig{WorkDir: workDir}); err != nil {
+	if err := personalStopBus(busctl.StopConfig{WorkDir: workDir, IPCEndpoint: ipcEndpoint}); err != nil {
 		if errors.Is(err, busctl.ErrNotRunning) {
 			busState = "personal bus is not running"
 		} else {

internal/event/bus/coverage_edges_test.go

@@ -361,13 +361,18 @@ func TestCrossPlatformCoverageDaemonMethodEdges(t *testing.T) {
 		}
 		return nil, net.ErrClosed
 	}}
-	d := &daemon{listener: l, log: logger, hub: NewHub(1), idleStop: make(chan struct{})}
+	d := &daemon{listener: l, log: logger, hub: NewHub(1), idleStop: make(chan struct{}), stopReq: make(chan string, 1)}
 	d.acceptLoop(context.Background())
 	d.shuttingDown.Store(true)
 	d.acceptLoop(context.Background())
 	d.triggerShutdown("test")
-	if !l.closed {
-		t.Fatal("trigger did not close listener")
+	select {
+	case reason := <-d.stopReq:
+		if reason != "test" {
+			t.Fatalf("stop reason = %q", reason)
+		}
+	default:
+		t.Fatal("trigger did not notify lifecycle loop")
 	}
 
 	d = &daemon{listener: &scriptedListener{accept: func() (net.Conn, error) { return nil, net.ErrClosed }}, log: logger, hub: NewHub(1), idleStop: make(chan struct{})}

internal/event/bus/daemon.go

@@ -186,6 +186,7 @@ func Run(ctx context.Context, cfg Config) error {
 		dedup:    dd,
 		started:  time.Now().UTC(),
 		idleStop: make(chan struct{}),
+		stopReq:  make(chan string, 1),
 	}
 
 	// 4. Signal ready BEFORE accepting consumers (avoids a slow-fork
@@ -263,6 +264,9 @@ func Run(ctx context.Context, cfg Config) error {
 		}
 	case <-d.idleStop:
 		log.Info("bus: idle timeout reached, shutting down")
+	case reason := <-d.stopReq:
+		shutdownReason = reason
+		log.Info("bus: shutdown requested via IPC", "reason", reason)
 	}
 
 	// 7. Graceful shutdown — cancel runCtx first so all background
@@ -292,6 +296,7 @@ type daemon struct {
 	shutdownMu   sync.Mutex
 	shuttingDown atomic.Bool
 	idleStop     chan struct{}
+	stopReq      chan string
 
 	credentialHandoffMu sync.Mutex
 	terminalMu          sync.RWMutex
@@ -749,15 +754,17 @@ func (d *daemon) idleWatch(ctx context.Context) {
 	}
 }
 
-// triggerShutdown is called from RPC handlers that want to end the bus.
-// It works by closing the listener (which unblocks Run's select via the
-// source error path, indirectly). For v1 a full ctx-cancellation hook is
-// out of scope; busctl/stop also sends SIGTERM which is the authoritative
-// shutdown path.
+// triggerShutdown is called from RPC handlers that want to end the bus. The
+// buffered request wakes Run's lifecycle select, which then cancels the cloud
+// source and performs the same graceful shutdown used for signals and idle
+// expiry. It is intentionally non-blocking so repeated stop requests cannot
+// strand connection handlers.
 func (d *daemon) triggerShutdown(reason string) {
 	d.log.Info("bus: shutdown triggered via IPC", "reason", reason)
-	_ = d.listener.Close() // unblocks Accept(), but doesn't kill Source
-	// Best-effort: a future version wires a context.CancelFunc here.
+	select {
+	case d.stopReq <- reason:
+	default:
+	}
 }
 
 // shutdown performs the graceful tear-down sequence:

internal/event/bus/daemon_test.go

@@ -141,6 +141,67 @@ func TestDaemon_RunStartsAndShutsDownCleanly(t *testing.T) {
 	}
 }
 
+func TestCrossPlatformCoverageDaemonIPCStopShutsDownCleanly(t *testing.T) {
+	workDir := shortTempDir(t)
+	identityHash := dwsevent.IdentityHash(workDir)
+	endpoint := dwsevent.IPCEndpoint(workDir, "open", dwsevent.SourceKindAppStream, identityHash)
+	runDone := make(chan error, 1)
+	go func() {
+		runDone <- Run(context.Background(), Config{
+			WorkDir:      workDir,
+			IPCEndpoint:  endpoint,
+			ClientID:     "ding_ipc_stop_test",
+			Edition:      "open",
+			SourceKind:   dwsevent.SourceKindAppStream,
+			IdentityHash: identityHash,
+			Source:       &fakeSource{},
+		})
+	}()
+
+	var conn net.Conn
+	deadline := time.Now().Add(3 * time.Second)
+	for time.Now().Before(deadline) {
+		var err error
+		conn, err = transport.Dial(endpoint)
+		if err == nil {
+			break
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	if conn == nil {
+		t.Fatal("bus IPC endpoint did not become ready")
+	}
+	w := transport.NewWriter(conn)
+	r := transport.NewReader(conn)
+	if err := w.WriteJSON(transport.Hello{
+		Type:        transport.FrameTypeHello,
+		ConsumerPID: os.Getpid(),
+		Role:        transport.HelloRoleStop,
+	}); err != nil {
+		t.Fatalf("write stop hello: %v", err)
+	}
+	var bye transport.Bye
+	if err := r.ReadJSON(&bye); err != nil {
+		t.Fatalf("read stop response: %v", err)
+	}
+	_ = conn.Close()
+	if bye.Type != transport.FrameTypeBye || bye.Reason != "stop_request" {
+		t.Fatalf("stop response = %#v", bye)
+	}
+
+	select {
+	case err := <-runDone:
+		if err != nil {
+			t.Fatalf("Run returned after IPC stop: %v", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("Run did not return after IPC stop")
+	}
+	if pid := ReadHolderPID(filepath.Join(workDir, LockFileName)); pid != 0 {
+		t.Fatalf("bus lock retained pid %d after IPC stop", pid)
+	}
+}
+
 func TestDaemon_ConsumerReceivesEvents(t *testing.T) {
 	skipOnWindows(t, "uses Unix socket dial")
 	workDir := shortTempDir(t)

internal/event/busctl/discover.go

@@ -112,8 +112,9 @@ func Discover(cfg DiscoverConfig) (net.Conn, error) {
 		return nil, fmt.Errorf("busctl: mkdir workdir: %w", err)
 	}
 	_, spawnErr := cfg.Spawn(SpawnConfig{
-		ClientID:  cfg.ClientID,
-		ExtraArgs: cfg.SpawnExtraArgs,
+		ClientID:    cfg.ClientID,
+		IPCEndpoint: cfg.IPCEndpoint,
+		ExtraArgs:   cfg.SpawnExtraArgs,
 	})
 	if spawnErr != nil && !errors.Is(spawnErr, ErrSpawnFailed) {
 		// Hard error (couldn't even exec the child). Stop here — no bus

internal/event/busctl/discover_test.go

@@ -122,7 +122,10 @@ func TestDiscover_NoBus_SpawnSucceeds(t *testing.T) {
 			closer()
 		}
 	})
-	fakeSpawn := func(SpawnConfig) (int, error) {
+	fakeSpawn := func(cfg SpawnConfig) (int, error) {
+		if cfg.IPCEndpoint != sock {
+			t.Fatalf("spawn IPC endpoint = %q, want %q", cfg.IPCEndpoint, sock)
+		}
 		closer = startStubBus(t, sock)
 		return 12345, nil
 	}

internal/event/busctl/spawn.go

@@ -43,18 +43,14 @@ var (
 	spawnPipe       = os.Pipe
 )
 
-// ErrSpawnFailed is returned when the child reports startup failure via
-// the ready pipe ('E' byte). The child's exit error / log file holds the
-// actual cause; this sentinel just lets the caller distinguish "ready
-// pipe said no" from "ready pipe timed out / closed early".
-var ErrSpawnFailed = errors.New("busctl: bus child reported startup failure on ready pipe")
+// ErrSpawnFailed is returned when the child reports startup failure through
+// the Unix ready pipe or exits before binding the Windows named pipe.
+var ErrSpawnFailed = errors.New("busctl: bus child reported startup failure")
 
 // ErrSpawnTimeout is returned when ReadyTimeout elapses without any signal.
 var ErrSpawnTimeout = errors.New("busctl: bus child did not signal readiness within deadline")
 
-// SpawnConfig describes one spawn attempt. ClientID is the only field
-// inspected by the child; the rest govern process attributes the parent
-// applies before exec.
+// SpawnConfig describes one spawn attempt.
 type SpawnConfig struct {
 	// ExecPath is the dws binary to exec. Default os.Executable().
 	ExecPath string
@@ -62,18 +58,23 @@ type SpawnConfig struct {
 	// ClientID is passed as `--client-id` to `dws event _bus`. Required.
 	ClientID string
 
+	// IPCEndpoint is the bus endpoint the child binds. Windows uses the
+	// existing named pipe as its readiness handshake because os/exec does not
+	// support ExtraFiles there. Unix keeps the inherited ready-pipe protocol.
+	IPCEndpoint string
+
 	// ExtraArgs are appended after `--client-id`. Empty for normal use; tests
 	// pass `--extra-flag-for-test` etc.
 	ExtraArgs []string
 
-	// Env to pass to the child. Defaults to os.Environ(). The ReadyFDEnv
-	// entry is appended automatically.
+	// Env to pass to the child. Defaults to os.Environ(). Unix appends the
+	// ReadyFDEnv entry automatically; Windows removes any inherited copy.
 	Env []string
 }
 
-// Spawn forks a detached `dws event _bus --client-id <id>` child process and
-// waits for it to signal readiness via the ready pipe. Returns the child's
-// PID on success — the caller can then dial the bus IPC endpoint.
+// spawnWithReadyPipe is the Unix implementation behind Spawn. It forks a
+// detached `dws event _bus --client-id <id>` child and waits for the inherited
+// ready pipe. Windows has a named-pipe implementation in spawn_windows.go.
 //
 // stdio detach (plan invariant #7):
 //   - cmd.Stdout / cmd.Stderr set to nil so the child's own writes don't
@@ -89,7 +90,7 @@ type SpawnConfig struct {
 // Parent (this function):
 //   - Holds the read end open until either 1 byte is read or ReadyTimeout
 //   - Returns ErrSpawnFailed for 'E', ErrSpawnTimeout otherwise
-func Spawn(cfg SpawnConfig) (pid int, err error) {
+func spawnWithReadyPipe(cfg SpawnConfig) (pid int, err error) {
 	if cfg.ClientID == "" {
 		return 0, errors.New("busctl: SpawnConfig.ClientID is required")
 	}

internal/event/busctl/spawn_test.go

@@ -22,6 +22,8 @@ import (
 	"strings"
 	"testing"
 	"time"
+
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
 )
 
 // Test child mode: when DWS_BUSCTL_TEST_CHILD is set, this test binary
@@ -35,7 +37,8 @@ import (
 // env-marker pattern is what Go's own os/exec tests use and stays
 // confined to this file.
 const (
-	childEnvMarker = "DWS_BUSCTL_TEST_CHILD"
+	childEnvMarker   = "DWS_BUSCTL_TEST_CHILD"
+	childEndpointEnv = "DWS_BUSCTL_TEST_ENDPOINT"
 	// values:
 	//   "ready"        — write 'R' then sleep 30s (parent should see ready)
 	//   "fail"         — write 'E' then exit (parent should see ErrSpawnFailed)
@@ -70,6 +73,26 @@ func TestMain(m *testing.M) {
 		writeReady('R')
 		time.Sleep(30 * time.Second)
 		os.Exit(0)
+	case "windows-ready":
+		listener, err := transport.Listen(os.Getenv(childEndpointEnv))
+		if err != nil {
+			os.Exit(3)
+		}
+		defer listener.Close()
+		conn, err := listener.Accept()
+		if err != nil {
+			os.Exit(4)
+		}
+		_ = conn.Close()
+		time.Sleep(30 * time.Second)
+		os.Exit(0)
+	case "windows-fail":
+		os.Exit(5)
+	case "windows-exit":
+		os.Exit(0)
+	case "windows-stall":
+		time.Sleep(30 * time.Second)
+		os.Exit(0)
 	}
 	os.Exit(m.Run())
 }

internal/event/busctl/spawn_unix.go

@@ -20,6 +20,13 @@ import (
 	"syscall"
 )
 
+// Spawn starts the detached bus and waits for its inherited ready pipe.
+// ExtraFiles is intentionally confined to Unix: Go does not support it on
+// Windows.
+func Spawn(cfg SpawnConfig) (pid int, err error) {
+	return spawnWithReadyPipe(cfg)
+}
+
 // applyDetach configures the child to live past parent death and not share
 // the parent's controlling terminal. Setsid puts the child in a new
 // session, so SIGHUP on the parent's controlling tty (e.g. SSH disconnect)

internal/event/busctl/spawn_windows.go

@@ -16,14 +16,101 @@
 package busctl
 
 import (
+	"fmt"
+	"os"
 	"os/exec"
+	"strings"
 	"syscall"
+	"time"
+
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
 )
 
 // CREATE_NEW_PROCESS_GROUP (0x00000200) prevents the child from receiving
 // the parent's Ctrl+C signal, similar in spirit to Setsid on Unix.
 const createNewProcessGroup = 0x00000200
 
+var (
+	spawnWindowsDial         = transport.Dial
+	spawnWindowsPollInterval = 25 * time.Millisecond
+)
+
+// Spawn starts a detached Windows bus without cmd.ExtraFiles (unsupported by
+// Go on Windows). Readiness is confirmed by dialing the bus's existing named
+// pipe. The child is reaped in the background after readiness, while an early
+// process exit is surfaced as ErrSpawnFailed.
+func Spawn(cfg SpawnConfig) (pid int, err error) {
+	if strings.TrimSpace(cfg.ClientID) == "" {
+		return 0, fmt.Errorf("busctl: SpawnConfig.ClientID is required")
+	}
+	if strings.TrimSpace(cfg.IPCEndpoint) == "" {
+		return 0, fmt.Errorf("busctl: SpawnConfig.IPCEndpoint is required on Windows")
+	}
+	if cfg.ExecPath == "" {
+		execPath, err := spawnExecutable()
+		if err != nil {
+			return 0, fmt.Errorf("busctl: locate executable: %w", err)
+		}
+		cfg.ExecPath = execPath
+	}
+	if cfg.Env == nil {
+		cfg.Env = os.Environ()
+	}
+
+	args := append([]string{"event", "_bus", "--client-id", cfg.ClientID}, cfg.ExtraArgs...)
+	cmd := exec.Command(cfg.ExecPath, args...)
+	cmd.Env = withoutReadyFDEnv(cfg.Env)
+	cmd.Stdin = nil
+	cmd.Stdout = nil
+	cmd.Stderr = nil
+	applyDetach(cmd)
+	if err := cmd.Start(); err != nil {
+		return 0, fmt.Errorf("busctl: start %s: %w", cfg.ExecPath, err)
+	}
+	pid = cmd.Process.Pid
+
+	waitDone := make(chan error, 1)
+	go func() { waitDone <- cmd.Wait() }()
+	timer := time.NewTimer(ReadyTimeout)
+	defer timer.Stop()
+	ticker := time.NewTicker(spawnWindowsPollInterval)
+	defer ticker.Stop()
+
+	for {
+		if conn, dialErr := spawnWindowsDial(cfg.IPCEndpoint); dialErr == nil {
+			_ = conn.Close()
+			return pid, nil
+		}
+		select {
+		case waitErr := <-waitDone:
+			if waitErr == nil {
+				return pid, fmt.Errorf("%w: child exited before binding named pipe", ErrSpawnFailed)
+			}
+			return pid, fmt.Errorf("%w: %v", ErrSpawnFailed, waitErr)
+		case <-timer.C:
+			_ = cmd.Process.Kill()
+			select {
+			case <-waitDone:
+			case <-time.After(time.Second):
+			}
+			return pid, ErrSpawnTimeout
+		case <-ticker.C:
+		}
+	}
+}
+
+func withoutReadyFDEnv(env []string) []string {
+	prefix := strings.ToUpper(ReadyFDEnv) + "="
+	out := make([]string, 0, len(env))
+	for _, entry := range env {
+		if strings.HasPrefix(strings.ToUpper(entry), prefix) {
+			continue
+		}
+		out = append(out, entry)
+	}
+	return out
+}
+
 func applyDetach(cmd *exec.Cmd) {
 	cmd.SysProcAttr = &syscall.SysProcAttr{
 		CreationFlags: createNewProcessGroup,

internal/event/busctl/spawn_windows_test.go

@@ -0,0 +1,108 @@
+// 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 busctl
+
+import (
+	"errors"
+	"fmt"
+	"os"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/testseam"
+)
+
+func TestCrossPlatformCoverageWindowsSpawnUsesNamedPipeReadiness(t *testing.T) {
+	endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-test-%d-%d`, os.Getpid(), time.Now().UnixNano())
+	pid, err := spawnWithMarker(t, "windows-ready", func(cfg *SpawnConfig) {
+		cfg.IPCEndpoint = endpoint
+		cfg.Env = append(cfg.Env, childEndpointEnv+"="+endpoint)
+	})
+	if err != nil {
+		t.Fatalf("Spawn() = %v", err)
+	}
+	if pid <= 0 {
+		t.Fatalf("Spawn() pid = %d", pid)
+	}
+	proc, err := os.FindProcess(pid)
+	if err != nil {
+		t.Fatal(err)
+	}
+	_ = proc.Kill()
+}
+
+func TestCrossPlatformCoverageWindowsSpawnReportsEarlyExit(t *testing.T) {
+	endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-fail-%d-%d`, os.Getpid(), time.Now().UnixNano())
+	_, err := spawnWithMarker(t, "windows-fail", func(cfg *SpawnConfig) {
+		cfg.IPCEndpoint = endpoint
+	})
+	if !errors.Is(err, ErrSpawnFailed) {
+		t.Fatalf("Spawn() error = %v, want ErrSpawnFailed", err)
+	}
+}
+
+func TestCrossPlatformCoverageWindowsSpawnReportsCleanEarlyExit(t *testing.T) {
+	endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-exit-%d-%d`, os.Getpid(), time.Now().UnixNano())
+	_, err := spawnWithMarker(t, "windows-exit", func(cfg *SpawnConfig) {
+		cfg.IPCEndpoint = endpoint
+	})
+	if !errors.Is(err, ErrSpawnFailed) {
+		t.Fatalf("Spawn() error = %v, want ErrSpawnFailed", err)
+	}
+}
+
+func TestCrossPlatformCoverageWindowsSpawnValidationAndStartErrors(t *testing.T) {
+	endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-errors-%d-%d`, os.Getpid(), time.Now().UnixNano())
+	if _, err := Spawn(SpawnConfig{ClientID: "client"}); err == nil {
+		t.Fatal("missing endpoint unexpectedly succeeded")
+	}
+	testseam.Swap(t, &spawnExecutable, func() (string, error) {
+		return "", errors.New("executable unavailable")
+	})
+	if _, err := Spawn(SpawnConfig{ClientID: "client", IPCEndpoint: endpoint}); err == nil {
+		t.Fatal("executable lookup unexpectedly succeeded")
+	}
+	if _, err := Spawn(SpawnConfig{
+		ClientID:    "client",
+		IPCEndpoint: endpoint,
+		ExecPath:    `C:\\definitely-missing-dws.exe`,
+	}); err == nil {
+		t.Fatal("missing executable unexpectedly started")
+	}
+}
+
+func TestCrossPlatformCoverageWindowsSpawnReadyTimeoutTerminatesChild(t *testing.T) {
+	endpoint := fmt.Sprintf(`\\.\pipe\dws-event-spawn-stall-%d-%d`, os.Getpid(), time.Now().UnixNano())
+	testseam.Swap(t, &ReadyTimeout, 100*time.Millisecond)
+	_, err := spawnWithMarker(t, "windows-stall", func(cfg *SpawnConfig) {
+		cfg.IPCEndpoint = endpoint
+	})
+	if !errors.Is(err, ErrSpawnTimeout) {
+		t.Fatalf("Spawn() error = %v, want ErrSpawnTimeout", err)
+	}
+}
+
+func TestCrossPlatformCoverageWindowsSpawnRemovesInheritedReadyFD(t *testing.T) {
+	env := withoutReadyFDEnv([]string{
+		"PATH=C:\\Windows",
+		ReadyFDEnv + "=3",
+		strings.ToLower(ReadyFDEnv) + "=4",
+	})
+	if len(env) != 1 || !strings.HasPrefix(env[0], "PATH=") {
+		t.Fatalf("withoutReadyFDEnv() = %#v", env)
+	}
+}

internal/event/busctl/stop.go

@@ -17,15 +17,17 @@ import (
 	"errors"
 	"fmt"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/bus"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/process"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
 )
 
-// DefaultStopTimeout is the wall-clock budget Stop waits for the bus to
-// exit after the signal is sent. 5s covers the bus's own graceful tear-down
-// (broadcast Bye → consumer goroutines drain → cleanup) with margin.
+// DefaultStopTimeout is the wall-clock budget Stop waits for each graceful or
+// fallback termination phase. 5s covers the bus's own tear-down (broadcast
+// Bye → consumer goroutines drain → cleanup) with margin.
 const DefaultStopTimeout = 5 * time.Second
 
 // ErrNotRunning indicates bus.lock either does not exist or its recorded
@@ -38,15 +40,19 @@ var (
 	stopAlive         = process.Alive
 	stopFindProcess   = os.FindProcess
 	stopSignalProcess = func(proc *os.Process, signal os.Signal) error { return proc.Signal(signal) }
+	stopRequest       = requestBusStop
 )
 
 // StopConfig identifies the target bus and tunes timing.
 type StopConfig struct {
 	// WorkDir holds bus.lock; Stop reads the PID from there.
 	WorkDir string
-	// Timeout is the total wall-clock budget for graceful exit. After this,
-	// Stop returns an error; it does NOT escalate to SIGKILL — leave that
-	// to the operator.
+	// IPCEndpoint enables the cross-platform graceful stop RPC. Callers that
+	// know the bus identity should always provide it. Empty preserves the
+	// legacy signal-only fallback for compatibility and focused tests.
+	IPCEndpoint string
+	// Timeout is the wall-clock budget for graceful exit and, if required,
+	// the subsequent platform termination fallback.
 	Timeout time.Duration
 }
 
@@ -54,13 +60,12 @@ type StopConfig struct {
 // for the process to actually die. Returns ErrNotRunning if no bus is
 // running for that work dir.
 //
-// Implementation note: on Unix we send SIGTERM. The bus daemon's Run loop
-// watches its parent ctx for cancellation; the cobra `event _bus` command
-// wires signal.NotifyContext so SIGTERM triggers ctx.Done() → graceful
-// shutdown path. On Windows we use os.Process.Signal(os.Interrupt) which
-// the Go runtime maps to TerminateProcess for processes outside our
-// console group; for v1 that's acceptable (Windows graceful shutdown is
-// future work — plan §16 v2).
+// Stop first asks the bus to shut down through its owner-only IPC endpoint.
+// This is the normal path on every platform and lets the daemon cancel its
+// cloud source, notify consumers, and release its lock. If the endpoint is
+// unavailable (for example an older bus) or graceful shutdown times out, the
+// platform signal is used as a compatibility fallback: SIGTERM on Unix and
+// TerminateProcess via os.Kill on Windows.
 func Stop(cfg StopConfig) error {
 	if cfg.WorkDir == "" {
 		return errors.New("busctl: StopConfig.WorkDir is required")
@@ -81,6 +86,12 @@ func Stop(cfg StopConfig) error {
 	if err != nil {
 		return fmt.Errorf("busctl: find process %d: %w", pid, err)
 	}
+	if strings.TrimSpace(cfg.IPCEndpoint) != "" {
+		if err := stopRequest(cfg.IPCEndpoint); err == nil && waitForBusExit(pid, cfg.Timeout) {
+			return nil
+		}
+	}
+
 	if err := stopSignalProcess(proc, stopSignal()); err != nil {
 		// On many Unix platforms Signal returns "process already finished"
 		// when the bus has just exited on its own — treat that as success.
@@ -90,13 +101,45 @@ func Stop(cfg StopConfig) error {
 		return fmt.Errorf("busctl: signal bus pid=%d: %w", pid, err)
 	}
 
-	// Poll for actual exit.
-	deadline := time.Now().Add(cfg.Timeout)
-	for time.Now().Before(deadline) {
-		if !stopAlive(pid) {
-			return nil
-		}
-		time.Sleep(50 * time.Millisecond)
+	if waitForBusExit(pid, cfg.Timeout) {
+		return nil
 	}
 	return fmt.Errorf("busctl: bus pid=%d did not exit within %s", pid, cfg.Timeout)
 }
+
+func requestBusStop(endpoint string) error {
+	conn, err := transport.Dial(endpoint)
+	if err != nil {
+		return fmt.Errorf("busctl: dial bus for stop: %w", err)
+	}
+	defer conn.Close()
+	_ = conn.SetDeadline(time.Now().Add(DefaultStatusRPCTimeout))
+	w := transport.NewWriter(conn)
+	r := transport.NewReader(conn)
+	if err := w.WriteJSON(transport.Hello{
+		Type:        transport.FrameTypeHello,
+		ConsumerPID: os.Getpid(),
+		Role:        transport.HelloRoleStop,
+	}); err != nil {
+		return fmt.Errorf("busctl: write stop hello: %w", err)
+	}
+	var bye transport.Bye
+	if err := r.ReadJSON(&bye); err != nil {
+		return fmt.Errorf("busctl: read stop response: %w", err)
+	}
+	if bye.Type != transport.FrameTypeBye || bye.Reason != "stop_request" {
+		return fmt.Errorf("busctl: unexpected stop response type=%q reason=%q", bye.Type, bye.Reason)
+	}
+	return nil
+}
+
+func waitForBusExit(pid int, timeout time.Duration) bool {
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		if !stopAlive(pid) {
+			return true
+		}
+		time.Sleep(50 * time.Millisecond)
+	}
+	return !stopAlive(pid)
+}

internal/event/busctl/stop_test.go

@@ -23,7 +23,10 @@ import (
 	"testing"
 	"time"
 
+	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/bus"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/testseam"
 )
 
 func TestStop_NotRunningWhenLockMissing(t *testing.T) {
@@ -46,6 +49,87 @@ func TestStop_NotRunningWhenPIDDead(t *testing.T) {
 	}
 }
 
+func TestCrossPlatformCoverageStopPrefersGracefulIPC(t *testing.T) {
+	const pid = 4242
+	stopped := false
+	signals := 0
+	testseam.Swap(t, &stopReadHolderPID, func(string) int { return pid })
+	testseam.Swap(t, &stopAlive, func(int) bool { return !stopped })
+	proc, err := os.FindProcess(os.Getpid())
+	if err != nil {
+		t.Fatal(err)
+	}
+	testseam.Swap(t, &stopFindProcess, func(int) (*os.Process, error) { return proc, nil })
+	testseam.Swap(t, &stopRequest, func(endpoint string) error {
+		if endpoint != "test-endpoint" {
+			t.Fatalf("stop endpoint = %q", endpoint)
+		}
+		stopped = true
+		return nil
+	})
+	testseam.Swap(t, &stopSignalProcess, func(*os.Process, os.Signal) error {
+		signals++
+		return nil
+	})
+
+	if err := Stop(StopConfig{
+		WorkDir:     "test-workdir",
+		IPCEndpoint: "test-endpoint",
+		Timeout:     time.Second,
+	}); err != nil {
+		t.Fatalf("Stop() = %v", err)
+	}
+	if signals != 0 {
+		t.Fatalf("graceful IPC stop used %d process signals", signals)
+	}
+}
+
+func TestCrossPlatformCoverageRequestBusStopProtocol(t *testing.T) {
+	dir := shortTempDir(t)
+	endpoint := dwsevent.IPCEndpoint(
+		dir,
+		"open",
+		dwsevent.SourceKindAppStream,
+		dwsevent.IdentityHash(dir),
+	)
+	listener, err := transport.Listen(endpoint)
+	if err != nil {
+		t.Fatalf("Listen() = %v", err)
+	}
+	defer listener.Close()
+	serverDone := make(chan error, 1)
+	go func() {
+		conn, err := listener.Accept()
+		if err != nil {
+			serverDone <- err
+			return
+		}
+		defer conn.Close()
+		r := transport.NewReader(conn)
+		w := transport.NewWriter(conn)
+		var hello transport.Hello
+		if err := r.ReadJSON(&hello); err != nil {
+			serverDone <- err
+			return
+		}
+		if hello.Type != transport.FrameTypeHello || hello.Role != transport.HelloRoleStop {
+			serverDone <- errors.New("unexpected stop hello")
+			return
+		}
+		serverDone <- w.WriteJSON(transport.Bye{
+			Type:   transport.FrameTypeBye,
+			Reason: "stop_request",
+		})
+	}()
+
+	if err := requestBusStop(endpoint); err != nil {
+		t.Fatalf("requestBusStop() = %v", err)
+	}
+	if err := <-serverDone; err != nil {
+		t.Fatalf("stop protocol server = %v", err)
+	}
+}
+
 func TestStop_SignalsLiveProcess(t *testing.T) {
 	skipOnWindows(t)
 	dir := shortTempDir(t)

internal/event/busctl/stop_windows.go

@@ -17,8 +17,7 @@ package busctl
 
 import "os"
 
-// stopSignal returns the graceful-shutdown signal for Windows. The Go
-// runtime maps os.Interrupt to TerminateProcess for non-console-group
-// processes — not truly graceful, but acceptable for v1 (Windows graceful
-// shutdown via Ctrl+Break is in the v2 backlog, plan §16).
-func stopSignal() os.Signal { return os.Interrupt }
+// stopSignal is only the fallback after the graceful IPC stop path fails or
+// times out. Go maps os.Kill to TerminateProcess on Windows; os.Interrupt is
+// unsupported and returns syscall.EWINDOWS.
+func stopSignal() os.Signal { return os.Kill }

internal/event/busctl/stop_windows_test.go

@@ -0,0 +1,27 @@
+// 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 busctl
+
+import (
+	"os"
+	"testing"
+)
+
+func TestCrossPlatformCoverageWindowsStopFallbackUsesKill(t *testing.T) {
+	if stopSignal() != os.Kill {
+		t.Fatalf("stopSignal() = %v, want os.Kill", stopSignal())
+	}
+}