// 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 app

import (
	"context"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log/slog"
	"os"
	"os/signal"
	"path/filepath"
	"sort"
	"strings"
	"syscall"
	"text/tabwriter"
	"time"

	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/helpers"

	authpkg "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/auth"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/cli"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/corecmd/contract"
	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/busctl"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/consume"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/registry"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/source"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/edition"
	"github.com/spf13/cobra"
)

var (
	eventRunPersonalConsume    = runPersonalEventConsume
	eventRunPersonalList       = runPersonalEventList
	eventRunPersonalStatus     = runPersonalEventStatus
	eventRunPersonalStop       = runPersonalEventStop
	eventNormalizeAs           = normalizeEventAs
	eventResolveCredentials    = resolveEventCredentials
	eventConsumeRun            = consume.Run
	eventRunForeground         = runForegroundBus
	eventNewEventSource        = newEventSource
	eventNewDingtalkSource     = source.New
	eventResolveAccessToken    = ResolveAuxiliaryAccessToken
	eventForceRefreshRejected  = forceRefreshRejectedAccessToken
	eventBusRun                = bus.Run
	eventReadyFDFromEnv        = busctl.ReadyFDFromEnv
	eventResolvePersonal       = resolvePersonalEventIdentity
	eventNewPersonalSource     = newPersonalStreamSource
	eventMkdirAll              = os.MkdirAll
	eventOpenFile              = os.OpenFile
	eventEnumerateBuses        = busctl.EnumerateBuses
	eventFindBus               = busctl.FindBusByClientID
	eventQueryEntry            = busctl.QueryEntry
	eventStopBus               = busctl.Stop
	eventResolveAppCredentials = authpkg.ResolveAppCredentialsStrict
)

// newEventCommand returns the `event` parent command and all its subcommands.
// Wired into root.go's utilityCommands list.
func newEventCommand(globalFlags ...*GlobalFlags) *cobra.Command {
	// Product-level Agent routing Decl (migrated from selection/event.json
	// products.event). Catalog assembly stamps provenance contract_final.
	contract.RegisterProductDecl(contract.ProductDecl{
		ID: "event",
		Selection: contract.ProductSelectionDecl{
			AgentSummary: "实时监听当前用户相关的个人 IM 与 OA 审批事件，并管理订阅生命周期",
			UseWhen: []string{
				"需要实时监听未来发生的个人消息、消息动作、群生命周期或 OA 审批任务/实例事件，或管理个人事件订阅生命周期",
			},
			AvoidWhen: []string{
				"查历史聊天或主动发消息用 chat；查询或处理审批实例/任务用 oa；配置开放平台应用事件回调用 dev app event",
			},
		},
	})
	cmd := &cobra.Command{
		Use:               "event",
		Short:             "事件订阅 (DingTalk Stream 长连接)",
		Long:              "通过 DingTalk Stream 长连接订阅事件并以 NDJSON 输出到 stdout。详见 `dws event consume --help`。",
		Args:              cobra.NoArgs,
		TraverseChildren:  true,
		DisableAutoGenTag: true,
		RunE:              func(c *cobra.Command, _ []string) error { return c.Help() },
	}
	cmd.AddCommand(
		newEventListenIMCommand(globalFlags...),
		newEventConsumeCommand(globalFlags...),
		newEventListCommand(),
		newEventSchemaCommand(),
		newEventStatusCommandWithFlags(globalFlags...),
		newEventStopCommandWithFlags(globalFlags...),
		newEventBusCommand(),
	)
	return cmd
}

// ─────────────────────────────────────────────────────────────────────
//  event consume
// ─────────────────────────────────────────────────────────────────────

func newEventConsumeCommand(globalFlags ...*GlobalFlags) *cobra.Command {
	var (
		eventTypes   []string
		filter       string
		compact      bool
		formatRaw    string
		outputDir    string
		routesRaw    []string
		maxEvents    int
		duration     time.Duration
		quiet        bool
		force        bool
		dryRun       bool
		foreground   bool
		asIdentity   string
		flatten      bool
		personalOpts personalConsumeOptions
		streamOpts   eventStreamTicketOptions
	)

	cmd := &cobra.Command{
		Use:   "consume [event_key...]",
		Short: "订阅事件流并输出到 stdout",
		Long: `订阅 DingTalk 个人事件并将每条事件以 NDJSON 输出到 stdout。

输出格式（事件流默认 ndjson；显式 -f json/pretty/raw 可覆盖；-f table/csv 对
事件流无意义会 fallback 到 ndjson）：
  ndjson  (默认)  一行一对象，适合 jq / 管道处理
  json            每事件多行美化 JSON（必须配 --max-events 或 --duration）
  pretty          同 json，未来加颜色
  raw             仅 SDK 原始 payload，无外层封装
  compact         单行紧凑 JSON；不传 --flatten 时沿用原 compact processor

数据结构：
  ndjson/json/pretty 默认保持 transport envelope（type/event_type/data/headers）
  --flatten          结构化格式输出稳定的顶层业务字段，适合 Agent / 脚本直接消费

默认使用当前 OAuth 登录态自动创建/复用个人订阅并建立个人长连接；非默认组织加
--profile。连上后 stderr 打就绪行 [event] ready，等它出现再读 stdout；停机用
SIGTERM、关 stdin，或先用 dws event stop <subscribe_id> --dry-run 预览、确认后加
--yes，绝不要 kill -9。

可提供多个 event_key。多事件会各自创建订阅和本地 consumer，但共享同一 bus、
远程连接、输出和 duration/max-events。用户类事件必须共享一个 --user 或
--open-dingtalk-id，群类事件必须共享一个 --group；用户类与群类不能混用。
全部 consumer 就绪后 stderr 输出 [event] ready event_count=<n> bus_pid=<pid>。
--event-types/--filter 只影响本地 bus → consume 这一段投递；普通个人事件消费
通常不需要设置。`,
		Args:              cobra.ArbitraryArgs,
		DisableAutoGenTag: true,
		RunE: func(c *cobra.Command, args []string) error {
			as, err := eventNormalizeAs(asIdentity)
			if err != nil {
				return err
			}
			if as == "user" {
				personalOpts.ExplicitToken = eventExplicitToken(globalFlags)
				personalOpts.ClientIDOverride = eventExplicitClientID(globalFlags)
				personalOpts.EventKeys = dedupePersonalEventKeys(args)
				personalOpts.EventKey = firstArg(personalOpts.EventKeys)
				personalOpts.Flatten = flatten
				if len(personalOpts.EventKeys) > 1 {
					if err := rejectPersonalMultiEventFlags(c,
						"subscribe-id", "rule", "event-types", "filter",
						"foreground", "force", "debug-raw-events",
					); err != nil {
						return fmt.Errorf("event consume: %w", personalSubscriptionValidationError(err))
					}
				}
				personalOpts.Common = commonConsumeOptions{
					EventTypes: eventTypes,
					Filter:     filter,
					Compact:    compact,
					FormatRaw:  formatRaw,
					OutputDir:  outputDir,
					RoutesRaw:  routesRaw,
					MaxEvents:  maxEvents,
					Duration:   duration,
					Quiet:      quiet,
					Force:      force,
					DryRun:     dryRun,
					Foreground: foreground,
				}
				personalOpts.StreamTicketMode = streamOpts.Mode
				personalOpts.StreamTicketURL = streamOpts.TicketURL
				personalOpts.StreamSourceID = streamOpts.SourceID
				return eventRunPersonalConsume(c, personalOpts)
			}
			if personalOpts.DebugRawEvents {
				return fmt.Errorf("event consume: --debug-raw-events is only supported with --as user")
			}
			if err := rejectChangedFlags(c, "user",
				"flatten",
				"subscribe-id",
				"rule",
				"name",
				"filter-json",
				"query",
				"ttl",
				"ephemeral",
				"user",
				"open-dingtalk-id",
				"group",
				"personal-event-base-url",
			); err != nil {
				return fmt.Errorf("event consume: %w", err)
			}
			if len(args) > 0 {
				return fmt.Errorf("event consume: event_key is only supported with --as user")
			}
			ctx, cancel := signal.NotifyContext(c.Context(), os.Interrupt, syscall.SIGTERM)
			defer cancel()

			// Step 1: resolve credentials.
			// Portal ticket normal mode uses portal-managed app credentials, so
			// local ClientSecret is intentionally not required there.
			configDir := defaultConfigDir()
			clientID, clientSecret, err := eventResolveCredentials(configDir, streamOpts)
			if err != nil {
				return fmt.Errorf("event consume: %w", err)
			}
			if !streamOpts.usesPortalNormalMode() && authpkg.EnvHalfSet() {
				fmt.Fprintln(c.ErrOrStderr(),
					"WARN: only one of DWS_CLIENT_ID/DWS_CLIENT_SECRET is set; env fallback disabled, using keychain/app config")
			}

			// Step 2: derive bus working directory + IPC endpoint.
			editionName := editionNameOrDefault()
			clientIDHash := dwsevent.ClientIDHash(clientID)
			workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindAppStream, clientIDHash)
			ipcEndpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindAppStream, clientIDHash)

			// Step 3: parse routes.
			routes, err := consume.ParseRoutes(routesRaw)
			if err != nil {
				return fmt.Errorf("event consume: %w", err)
			}

			// Step 4: resolve format. Event command's default is ndjson;
			// we check Changed on the inherited global -f/--format flag so
			// the user can still override (-f json etc.). table/csv fall
			// back to ndjson with a stderr WARN. See plan §3.1 输出约束.
			rawFormat := ""
			if f := c.Flags().Lookup("format"); f != nil && f.Changed {
				rawFormat = formatRaw
			}
			normalised, fellback := consume.NormalizeFormat(rawFormat)
			if fellback && !quiet {
				fmt.Fprintf(c.ErrOrStderr(),
					"WARN: --format %q has no meaning for event stream; using ndjson\n", rawFormat)
			}

			cfg := consume.Config{
				WorkDir:        workDir,
				IPCEndpoint:    ipcEndpoint,
				ClientID:       clientID,
				EventTypes:     eventTypesWithDefault(eventTypes),
				Filter:         filter,
				Compact:        compact,
				MaxEvents:      maxEvents,
				Duration:       duration,
				Format:         normalised,
				OutputDir:      outputDir,
				Routes:         routes,
				Stdout:         c.OutOrStdout(),
				Stderr:         c.ErrOrStderr(),
				Quiet:          quiet,
				Foreground:     foreground,
				Force:          force,
				DryRun:         dryRun,
				SpawnExtraArgs: streamOpts.spawnArgs(),
			}
			// Arm the stdin-EOF shutdown watcher only for a pipe-style,
			// unbounded run (see shouldWatchStdinEOF).
			applyEventConsumeStdin(&cfg, maxEvents, duration, c.InOrStdin())

			// Step 5: validation (flag-only rules).
			if err := consume.ValidateConfig(cfg); err != nil {
				return err
			}
			// --output-dir/--route vs global hidden -o/--output.
			if o := c.Flags().Lookup("output"); o != nil && o.Changed {
				if err := consume.ValidateNoOutputConflict(cfg, o.Value.String()); err != nil {
					return err
				}
			}

			// Step 6: foreground mode runs the bus in-process. Otherwise
			// consume.Run discovers / forks the bus and dials it.
			if foreground {
				return eventRunForeground(ctx, cfg, configDir, clientSecret, streamOpts)
			}
			return eventConsumeRun(ctx, cfg)
		},
	}

	f := cmd.Flags()
	f.StringVar(&asIdentity, "as", "user", "事件身份: user")
	f.StringSliceVar(&eventTypes, "event-types", nil,
		"逗号分隔事件类型；省略时按 event_key 过滤")
	f.StringVar(&filter, "filter", "",
		"客户端正则过滤事件类型 (下推到 bus 减少 IPC 投递量)")
	f.BoolVar(&compact, "compact", false,
		"提示 bus 客户端期望 compact 渲染（语义透传，bus 仍按原 payload 投递）")
	f.StringVarP(&formatRaw, "format", "f", "ndjson",
		"输出格式 (ndjson/json/pretty/raw/compact)；事件流默认 ndjson")
	f.BoolVar(&flatten, "flatten", false,
		"将个人事件 transport envelope 投影为稳定的顶层业务字段")
	f.StringVar(&outputDir, "output-dir", "",
		"每事件写一个文件到该目录 ({type}_{id}_{ts}.json)；与 stdout 互斥")
	f.StringArrayVar(&routesRaw, "route", nil,
		"按 regex 路由事件到目录：'<regex>=dir:<path>'，可重复；未命中走 stdout/--output-dir")
	f.IntVar(&maxEvents, "max-events", 0,
		"收到 N 条后退出 (0 = 不限)")
	f.DurationVar(&duration, "duration", 0,
		"运行时长上限 (Go duration 如 30s/5m)；事件流专用，不复用全局 --timeout")
	f.BoolVar(&quiet, "quiet", false,
		"抑制 stderr 状态信息")
	f.BoolVar(&force, "force", false,
		"仅 --foreground 模式生效：跳过单实例锁 (慎用：会让云事件被随机切分)")
	f.BoolVar(&dryRun, "dry-run", false,
		"仅打印解析后的配置；不创建订阅、不连接 bus；复用 --subscribe-id 时会只读查询控制面")
	f.BoolVar(&foreground, "foreground", false,
		"当前进程直接跑 bus 服务、不 fork、不打印事件（给 systemd/k8s 托管用）；读事件不要用它")
	f.StringVar(&personalOpts.SubscribeID, "subscribe-id", "",
		"个人事件订阅 ID；传入后复用已有订阅")
	f.StringVar(&personalOpts.Rule, "rule", "",
		"个人事件规则类型；默认根据 event_key 推断")
	f.StringVar(&personalOpts.Name, "name", "",
		"个人事件订阅名称")
	f.StringVar(&personalOpts.FilterJSON, "filter-json", "",
		"个人事件 Filter DSL JSON")
	f.StringVar(&personalOpts.QueryCSV, "query", "",
		"按消息文本关键词过滤，逗号分隔")
	f.DurationVar(&personalOpts.TTL, "ttl", 0,
		"个人订阅 TTL (Go duration，如 24h；0 表示不过期)")
	f.BoolVar(&personalOpts.Ephemeral, "ephemeral", false,
		"强制退出时取消个人订阅。默认已按归属清理：本次新建的订阅退出即取消，"+
			"用 --subscribe-id 复用的订阅保留。优雅停可用 SIGTERM、关闭 stdin，"+
			"或从外部先用 dws event stop <subscribe_id> --dry-run 预览、确认后加 --yes（会一并退订）；"+
			"请勿 kill -9（会跳过退订、泄漏服务端订阅）")
	f.StringVar(&personalOpts.UserID, "user", "",
		"单聊对端或指定发送人的 userId（与 --open-dingtalk-id 二选一）")
	f.StringVar(&personalOpts.OpenDingTalkID, "open-dingtalk-id", "",
		"单聊对端或指定发送人的 openDingtalkId（与 --user 二选一）")
	f.StringVar(&personalOpts.GroupID, "group", "",
		"group 规则：openConversationId")
	f.StringVar(&personalOpts.ControlBaseURL, "personal-event-base-url", "",
		"个人事件控制面 base URL；默认由 MCP base 派生 /dws")
	f.BoolVar(&personalOpts.DebugRawEvents, "debug-raw-events", false,
		"个人事件联调：绕过本地 event type/subscribe_id 过滤，输出当前 personal stream bus 收到的所有事件")
	f.StringVar(&streamOpts.Mode, "stream-ticket-mode", strings.TrimSpace(os.Getenv("DWS_STREAM_TICKET_MODE")),
		"个人 Stream 建联模式；默认 normal")
	f.StringVar(&streamOpts.SourceID, "stream-source-id", strings.TrimSpace(os.Getenv("DWS_STREAM_SOURCE_ID")),
		"个人 Stream sourceId；开源版默认 open，可由 edition 覆盖")
	f.StringVar(&streamOpts.TicketURL, "stream-ticket-url", strings.TrimSpace(os.Getenv("DWS_STREAM_TICKET_URL")),
		"个人 Stream 取票 URL；默认由 MCP base 派生 /stream/connections/ticket")
	hideEventInternalFlags(cmd, "as")
	cli.AnnotateRuntimePositionals(cmd, contract.RuntimeSchemaPositional{
		Name:        "event_key",
		Type:        "string",
		Description: "要消费的一个或多个个人事件码；多个事件必须共享同一目标和过滤上下文",
		Required:    false,
		Variadic:    true,
		Index:       0,
	})
	helpers.DeclareLeafMetadata(cmd, helpers.LeafSpec{
		Safety: contract.SafetySpec{
			Effect: "write", Risk: "medium",
			Confirmation: "not_required", Idempotency: "non_idempotent",
		},
		Contract: helpers.LeafContract{
			Identity: contract.ToolIdentitySpec{
				ProductID:      "event",
				Name:           "consume",
				CanonicalPath:  "event.consume",
				CLIPath:        "event consume",
				PrimaryCLIPath: "event consume",
			},
			Description: "订阅并持续消费一个或多个兼容的个人事件；Agent 使用 --flatten 输出顶层业务 NDJSON",
			Interface: &contract.InterfaceSpec{
				Mode:         "composite",
				Availability: "available",
				Reason:       "Reviewed composite workflow: the command creates or reuses a remote personal-event subscription and coordinates the local event bus and Stream consumer; no single pinned RPC represents the workflow.",
			},
			Selection: contract.SelectionSpec{
				AgentSummary: "消费 OA、群生命周期或需要底层控制的个人事件流；Agent 通常使用 --flatten 输出 NDJSON",
				UseWhen: []string{
					"需要监听七个公开 OA 审批任务/实例 EventKey 中的一个或多个事件",
					"需要监听指定群的标题变更、成员进退群或群解散事件",
					"用户显式给出原始 EventKey、Filter DSL、subscribe_id，要求原始 transport envelope，或需要普通 IM facade 不提供的高级多事件控制",
				},
				AvoidWhen: []string{
					"普通 @我、指定发送人/群、全部单聊/群聊及 message/reaction/read/recall 监听优先使用 event +listen-im",
					"只查历史聊天记录时用 chat 查询命令",
					"查询、同意、拒绝、转交、撤销或发起审批时用 oa；配置应用事件回调时用 dev app event",
					"只看事件目录/字段时用 event list / event schema",
				},
				Examples: []string{
					"dws event consume user_oa_approval_task_created user_oa_approval_instance_finished --flatten --duration 10m --format ndjson",
					"dws event consume user_im_group_member_added --group cid-example --flatten --max-events 1 --format ndjson",
				},
			},
		},
	})
	return cmd
}

// runForegroundBus implements --foreground: instead of dialing an existing
// bus or forking a new one, the current process IS the bus. We construct a
// source.DingtalkSource with the same credentials, then bus.Run in this
// goroutine with a ready pipe wired to stderr-only.
//
// Implementation note: a true --foreground run does NOT spawn a second
// process to consume — the consumer's NDJSON output would then have no
// stdout. For v1 we keep it simple: --foreground runs bus only; the user
// can run `dws event consume` from another shell to consume the events.
// v2 may add a "foreground + in-process consumer" combined mode.
func runForegroundBus(ctx context.Context, cfg consume.Config, configDir, clientSecret string, streamOpts eventStreamTicketOptions) error {
	src, err := eventNewEventSource(ctx, configDir, cfg.ClientID, clientSecret, streamOpts)
	if err != nil {
		return err
	}
	busCfg := bus.Config{
		WorkDir:      cfg.WorkDir,
		IPCEndpoint:  cfg.IPCEndpoint,
		ClientID:     cfg.ClientID,
		Edition:      editionNameOrDefault(),
		SourceKind:   dwsevent.SourceKindAppStream,
		IdentityHash: dwsevent.ClientIDHash(cfg.ClientID),
		Source:       src,
		Logger:       slog.Default(),
	}
	bus.ApplyEnvTuning(&busCfg)
	return eventBusRun(ctx, busCfg)
}

type eventStreamTicketOptions struct {
	Mode      string
	SourceID  string
	TicketURL string
}

func (o eventStreamTicketOptions) enabled() bool {
	return strings.TrimSpace(o.Mode) != ""
}

func (o eventStreamTicketOptions) normalizedMode() string {
	return strings.ToLower(strings.TrimSpace(o.Mode))
}

func (o eventStreamTicketOptions) usesPortalNormalMode() bool {
	return o.enabled() && o.normalizedMode() == source.PortalTicketModeNormal
}

func (o eventStreamTicketOptions) spawnArgs() []string {
	if !o.enabled() {
		return nil
	}
	args := []string{"--stream-ticket-mode", strings.TrimSpace(o.Mode)}
	if sourceID := strings.TrimSpace(o.SourceID); sourceID != "" {
		args = append(args, "--stream-source-id", sourceID)
	}
	if ticketURL := strings.TrimSpace(o.TicketURL); ticketURL != "" {
		args = append(args, "--stream-ticket-url", ticketURL)
	}
	return args
}

func resolveEventCredentials(configDir string, streamOpts eventStreamTicketOptions) (clientID, clientSecret string, err error) {
	if !streamOpts.usesPortalNormalMode() {
		clientID, clientSecret, _, _, err = authpkg.ResolveAppCredentialsStrict(configDir)
		return clientID, clientSecret, err
	}
	return eventStreamBusID(streamOpts), "", nil
}

func eventStreamBusID(streamOpts eventStreamTicketOptions) string {
	sourceID := eventStreamSourceID(streamOpts.SourceID)
	return "portal-ticket-normal:" + sourceID
}

func newEventSource(_ context.Context, configDir, clientID, clientSecret string, streamOpts eventStreamTicketOptions) (*source.DingtalkSource, error) {
	if !streamOpts.enabled() {
		return eventNewDingtalkSource(source.Config{
			ClientID:     clientID,
			ClientSecret: clientSecret,
		})
	}

	portalClientID := clientID
	portalClientSecret := clientSecret
	if streamOpts.usesPortalNormalMode() {
		portalClientID = ""
		portalClientSecret = ""
	}

	return eventNewDingtalkSource(source.Config{
		ClientID:     portalClientID,
		ClientSecret: portalClientSecret,
		PortalTicket: &source.PortalTicketConfig{
			TicketURL: eventStreamTicketURL(streamOpts.TicketURL),
			AccessTokenProvider: func(ctx context.Context) (string, error) {
				return eventResolveAccessToken(ctx, configDir, "")
			},
			ForceRefreshToken: func(ctx context.Context, rejectedToken string) (string, error) {
				return eventForceRefreshRejected(ctx, configDir, rejectedToken)
			},
			SourceID:     eventStreamSourceID(streamOpts.SourceID),
			Mode:         streamOpts.Mode,
			ClientID:     portalClientID,
			ClientSecret: portalClientSecret,
			UserAgent:    "dws-event-consume",
		},
	})
}

func eventStreamTicketURL(raw string) string {
	if v := strings.TrimSpace(raw); v != "" {
		return v
	}
	return strings.TrimRight(config.GetMCPBaseURL(), "/") + "/stream/connections/ticket"
}

func defaultEventStreamSourceID() string {
	if v := strings.TrimSpace(os.Getenv("DWS_STREAM_SOURCE_ID")); v != "" {
		return v
	}
	base := strings.ToLower(config.GetMCPBaseURL())
	if strings.Contains(base, "pre-mcp") {
		return "pre_open_source"
	}
	return "open"
}

func eventStreamSourceID(raw string) string {
	if v := strings.TrimSpace(raw); v != "" {
		return v
	}
	return defaultEventStreamSourceID()
}

// ─────────────────────────────────────────────────────────────────────
//  event _bus  (hidden — auto-forked by consume)
// ─────────────────────────────────────────────────────────────────────

func newEventBusCommand() *cobra.Command {
	var (
		clientIDOverride string
		idleTimeout      time.Duration
		sourceKindRaw    string
		runtimeTokenMode bool
		identityHashFlag string
		streamOpts       eventStreamTicketOptions
	)
	cmd := &cobra.Command{
		Use:               "_bus",
		Short:             "Internal event bus daemon (do not call directly)",
		Long:              "Hidden subcommand auto-spawned by `dws event consume`. Do not invoke directly.",
		Hidden:            true,
		Args:              cobra.NoArgs,
		DisableAutoGenTag: true,
		RunE: func(c *cobra.Command, _ []string) error {
			ctx, cancel := signal.NotifyContext(c.Context(), os.Interrupt, syscall.SIGTERM)
			defer cancel()

			// Acquire ReadyPipe early so pre-bus.Run failures can signal
			// 'E' to the parent process instead of silently dying.
			readyPipe := eventReadyFDFromEnv()
			failEarly := func(err error) error {
				if readyPipe != nil {
					// 'E' signals failure; the trailing text lets the parent
					// (busctl.waitReady) surface the real startup error to the
					// user instead of an opaque "startup failure on ready pipe".
					_, _ = readyPipe.Write([]byte{'E'})
					if err != nil {
						_, _ = io.WriteString(readyPipe, err.Error())
					}
					_ = readyPipe.Close()
				}
				return err
			}

			configDir := defaultConfigDir()
			sourceKind := dwsevent.SourceKind(strings.TrimSpace(sourceKindRaw))
			if sourceKind == "" {
				sourceKind = dwsevent.SourceKindAppStream
			}
			if sourceKind == dwsevent.SourceKindPersonalStream {
				var (
					identity     personal.Identity
					identityHash string
				)
				if runtimeTokenMode {
					identityHash = strings.TrimSpace(identityHashFlag)
					if !validPersonalIdentityHash(identityHash) {
						return failEarly(errors.New("event _bus: --identity-hash must be a 16-character hexadecimal identity hash in runtime token mode"))
					}
					if strings.TrimSpace(clientIDOverride) == "" {
						return failEarly(errors.New("event _bus: --client-id is required in runtime token mode"))
					}
					identity = personal.Identity{
						ClientID: strings.TrimSpace(clientIDOverride),
						SourceID: personalEventStreamSourceID(streamOpts.SourceID),
					}
				} else {
					var err error
					identity, err = eventResolvePersonal(ctx, configDir, streamOpts.SourceID)
					if err != nil {
						return failEarly(fmt.Errorf("event _bus: %w", err))
					}
					if clientIDOverride != "" {
						identity.ClientID = clientIDOverride
					}
					identityHash = dwsevent.IdentityHash(identity.Key())
				}
				editionName := editionNameOrDefault()
				workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
				endpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
				credentialBroker := newPersonalCredentialBroker(configDir, runtimeTokenMode, runtimeTokenMode)
				src, err := eventNewPersonalSource(ctx, personalStreamSourceOptions{
					ConfigDir:        configDir,
					Identity:         identity,
					TicketMode:       streamOpts.Mode,
					TicketURL:        streamOpts.TicketURL,
					ClientIDOverride: clientIDOverride,
					CredentialBroker: credentialBroker,
					RuntimeTokenMode: runtimeTokenMode,
				})
				if err != nil {
					return failEarly(err)
				}
				if err := eventMkdirAll(workDir, config.DirPerm); err == nil {
					if lf, ferr := eventOpenFile(filepath.Join(workDir, "bus.log"),
						os.O_CREATE|os.O_WRONLY|os.O_APPEND, config.FilePerm); ferr == nil {
						defer lf.Close()
						slog.SetDefault(slog.New(slog.NewTextHandler(lf, &slog.HandlerOptions{Level: slog.LevelInfo})))
					}
				}
				busCfg := bus.Config{
					WorkDir:          workDir,
					IPCEndpoint:      endpoint,
					ClientID:         identity.ClientID,
					Edition:          editionName,
					SourceKind:       dwsevent.SourceKindPersonalStream,
					IdentityHash:     identityHash,
					SourceID:         identity.SourceID,
					Source:           src,
					IdleTimeout:      idleTimeout,
					ReadyPipe:        readyPipe,
					Logger:           slog.Default(),
					CredentialBroker: credentialBroker,
				}
				bus.ApplyEnvTuning(&busCfg)
				return eventBusRun(ctx, busCfg)
			}

			resolvedID, secret, err := eventResolveCredentials(configDir, streamOpts)
			if err != nil {
				return failEarly(fmt.Errorf("event _bus: %w", err))
			}
			clientID := resolvedID
			if clientIDOverride != "" {
				clientID = clientIDOverride
			}
			editionName := editionNameOrDefault()
			clientIDHash := dwsevent.ClientIDHash(clientID)
			workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindAppStream, clientIDHash)
			endpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindAppStream, clientIDHash)

			src, err := eventNewEventSource(ctx, configDir, clientID, secret, streamOpts)
			if err != nil {
				return failEarly(err)
			}

			// busLogger writes to bus.log inside WorkDir so the daemon's
			// own log lines never pollute stdout/stderr (which busctl/Spawn
			// detached). Best-effort: if mkdir / open fails we fall back
			// to slog.Default (stderr) so we at least see startup errors.
			if err := eventMkdirAll(workDir, config.DirPerm); err == nil {
				if lf, ferr := eventOpenFile(filepath.Join(workDir, "bus.log"),
					os.O_CREATE|os.O_WRONLY|os.O_APPEND, config.FilePerm); ferr == nil {
					defer lf.Close()
					slog.SetDefault(slog.New(slog.NewTextHandler(lf, &slog.HandlerOptions{Level: slog.LevelInfo})))
				}
			}

			busCfg := bus.Config{
				WorkDir:      workDir,
				IPCEndpoint:  endpoint,
				ClientID:     clientID,
				Edition:      editionName,
				SourceKind:   dwsevent.SourceKindAppStream,
				IdentityHash: clientIDHash,
				Source:       src,
				IdleTimeout:  idleTimeout,
				ReadyPipe:    readyPipe,
				Logger:       slog.Default(),
			}
			// env-var tuning (only fills in fields left at zero; explicit
			// flags above keep precedence).
			bus.ApplyEnvTuning(&busCfg)
			return eventBusRun(ctx, busCfg)
		},
	}
	cmd.Flags().StringVar(&clientIDOverride, "client-id", "",
		"override clientID resolved from app config / env (used by busctl/Spawn)")
	cmd.Flags().DurationVar(&idleTimeout, "idle-timeout", 5*time.Minute,
		"exit after this long with zero consumers (0 = disabled)")
	cmd.Flags().StringVar(&sourceKindRaw, "source-kind", string(dwsevent.SourceKindAppStream),
		"event source kind: app_stream|personal_stream")
	cmd.Flags().BoolVar(&runtimeTokenMode, "runtime-token-mode", false,
		"use an owner-injected in-memory runtime credential")
	cmd.Flags().StringVar(&identityHashFlag, "identity-hash", "",
		"pre-resolved non-sensitive personal identity hash")
	cmd.Flags().StringVar(&streamOpts.Mode, "stream-ticket-mode", strings.TrimSpace(os.Getenv("DWS_STREAM_TICKET_MODE")),
		"用户 Stream 建联模式：空=SDK app credential；normal/custom=portal 取票")
	cmd.Flags().StringVar(&streamOpts.SourceID, "stream-source-id", strings.TrimSpace(os.Getenv("DWS_STREAM_SOURCE_ID")),
		"用户 Stream sourceId；personal_stream 开源版默认 open")
	cmd.Flags().StringVar(&streamOpts.TicketURL, "stream-ticket-url", strings.TrimSpace(os.Getenv("DWS_STREAM_TICKET_URL")),
		"用户 Stream 取票 URL；personal_stream 默认由 MCP base URL 派生")
	_ = cmd.Flags().MarkHidden("runtime-token-mode")
	_ = cmd.Flags().MarkHidden("identity-hash")
	return cmd
}

// ─────────────────────────────────────────────────────────────────────
//  event list
// ─────────────────────────────────────────────────────────────────────

// listEntry is one row in `dws event list` output. Combines on-disk bus
// metadata with the live status RPC results so consumers from multiple
// buses can be rendered as a single flat table.
type listEntry struct {
	ClientID     string                     `json:"client_id"`
	ClientIDHash string                     `json:"client_id_hash"`
	SourceKind   dwsevent.SourceKind        `json:"source_kind,omitempty"`
	SourceID     string                     `json:"source_id,omitempty"`
	Edition      string                     `json:"edition"`
	BusPID       int                        `json:"bus_pid"`
	BusState     busctl.BusEntryState       `json:"bus_state"`
	WorkDir      string                     `json:"workdir"`
	Consumers    []transport.StatusConsumer `json:"consumers,omitempty"`
	// PerType is included in JSON only — text mode renders it in `status`.
	PerType map[string]transport.Counters `json:"per_event_type,omitempty"`
}

func newEventListCommand() *cobra.Command {
	var (
		all            bool
		allEditions    bool
		formatRaw      string
		clientIDOver   string
		asIdentity     string
		category       string
		enabledOnly    bool
		includePending bool
	)
	cmd := &cobra.Command{
		Use:               "list",
		Short:             "列出个人事件目录",
		Long:              "列出当前支持的个人事件目录。",
		Args:              cobra.NoArgs,
		DisableAutoGenTag: true,
		RunE: func(c *cobra.Command, _ []string) error {
			as, err := eventNormalizeAs(asIdentity)
			if err != nil {
				return err
			}
			if as == "user" {
				if err := rejectPersonalEventUnsupportedFlags(c, "all", "all-editions", "client-id"); err != nil {
					return fmt.Errorf("event list: %w", err)
				}
				return eventRunPersonalList(c, personalListOptions{
					Category:       category,
					EnabledOnly:    enabledOnly,
					IncludePending: includePending,
					Format:         formatRaw,
				})
			}
			if err := rejectChangedFlags(c, "user", "category", "enabled-only", "include-pending"); err != nil {
				return fmt.Errorf("event list: %w", err)
			}
			entries, err := collectEntries(c, clientIDOver, all, allEditions)
			if err != nil {
				return err
			}
			rendered := make([]listEntry, 0, len(entries))
			for _, qs := range entries {
				rendered = append(rendered, buildListEntry(qs))
			}
			return renderList(c.OutOrStdout(), rendered, formatRaw)
		},
	}
	cmd.Flags().BoolVar(&all, "all", false, "列出当前 edition 下所有 ClientID 的消费者")
	cmd.Flags().BoolVar(&allEditions, "all-editions", false, "跨 edition 列出（罕用，调试用）")
	cmd.Flags().StringVar(&clientIDOver, "client-id", "", "指定具体 ClientID（覆盖凭证解析）")
	cmd.Flags().StringVarP(&formatRaw, "format", "f", "table", "输出格式: table|json")
	cmd.Flags().StringVar(&asIdentity, "as", "user", "事件身份: user")
	cmd.Flags().StringVar(&category, "category", "", "个人事件目录分类")
	cmd.Flags().BoolVar(&enabledOnly, "enabled-only", false, "个人事件目录只显示 enabled")
	cmd.Flags().BoolVar(&includePending, "include-pending", false, "个人事件目录包含 pending 项")
	hideEventInternalFlags(cmd, "as", "all", "all-editions", "client-id")
	helpers.DeclareLeafMetadata(cmd, helpers.LeafSpec{
		Safety: contract.SafetySpec{
			Effect: "read", Risk: "low",
			Confirmation: "not_required", Idempotency: "idempotent",
		},
		Contract: helpers.LeafContract{
			Identity: contract.ToolIdentitySpec{
				ProductID:      "event",
				Name:           "list",
				CanonicalPath:  "event.list",
				CLIPath:        "event list",
				PrimaryCLIPath: "event list",
			},
			Description: "列出支持的个人事件目录与状态说明",
			Interface: &contract.InterfaceSpec{
				Mode:         "local",
				Availability: "available",
				Reason:       "命令读取 CLI 内置的个人事件目录，不绑定 pinned MCP RPC",
			},
			Selection: contract.SelectionSpec{
				AgentSummary: "列出支持的个人事件目录与状态说明",
				UseWhen:      []string{"尚不知道可用 event_key，需要先盘点个人事件目录"},
				AvoidWhen: []string{
					"已知 event_key 要看 payload 字段时用 event schema",
					"要开始监听时用 event consume",
				},
				Examples: []string{"dws event list --format json"},
			},
		},
	})
	return cmd
}

// ─────────────────────────────────────────────────────────────────────
//  event status
// ─────────────────────────────────────────────────────────────────────

func newEventStatusCommand() *cobra.Command {
	return newEventStatusCommandWithFlags()
}

func newEventStatusCommandWithFlags(globalFlags ...*GlobalFlags) *cobra.Command {
	var (
		all          bool
		allEditions  bool
		formatRaw    string
		clientIDOver string
		failOnOrphan bool
		asIdentity   string
		personalOpts personalStatusOptions
	)
	cmd := &cobra.Command{
		Use:               "status",
		Short:             "显示个人事件订阅和本地消费状态",
		Long:              "显示当前用户个人事件订阅、personal bus 状态和本地 consume 进程。",
		Args:              cobra.NoArgs,
		DisableAutoGenTag: true,
		RunE: func(c *cobra.Command, _ []string) error {
			as, err := eventNormalizeAs(asIdentity)
			if err != nil {
				return err
			}
			if as == "user" {
				if err := rejectPersonalEventUnsupportedFlags(c, "all", "all-editions", "client-id", "fail-on-orphan"); err != nil {
					return fmt.Errorf("event status: %w", err)
				}
				personalOpts.Format = formatRaw
				personalOpts.ExplicitToken = eventExplicitToken(globalFlags)
				personalOpts.ClientIDOverride = eventExplicitClientID(globalFlags)
				return eventRunPersonalStatus(c, personalOpts)
			}
			if err := rejectChangedFlags(c, "user", "event", "status", "subscribe-id", "personal-event-base-url", "stream-source-id"); err != nil {
				return fmt.Errorf("event status: %w", err)
			}
			entries, err := collectEntries(c, clientIDOver, all, allEditions)
			if err != nil {
				return err
			}
			if err := renderStatus(c.OutOrStdout(), entries, formatRaw); err != nil {
				return err
			}
			if failOnOrphan {
				for _, qs := range entries {
					if qs.Entry.State == busctl.BusStateOrphan {
						return &consume.ValidationError{Msg: "orphan bus detected (--fail-on-orphan set)"}
					}
				}
			}
			return nil
		},
	}
	cmd.Flags().BoolVar(&all, "all", false, "当前 edition 下所有 ClientID")
	cmd.Flags().BoolVar(&allEditions, "all-editions", false, "跨 edition")
	cmd.Flags().StringVar(&clientIDOver, "client-id", "", "指定具体 ClientID")
	cmd.Flags().StringVarP(&formatRaw, "format", "f", "text", "输出格式: text|json")
	cmd.Flags().BoolVar(&failOnOrphan, "fail-on-orphan", false, "检测到 orphan 时退出码 2")
	cmd.Flags().StringVar(&asIdentity, "as", "user", "事件身份: user")
	cmd.Flags().StringVar(&personalOpts.EventKey, "event", "", "个人事件 event_key 过滤")
	cmd.Flags().StringVar(&personalOpts.Status, "status", "active", "个人订阅状态过滤: active|paused|error|deleted|all")
	cmd.Flags().StringVar(&personalOpts.SubscribeID, "subscribe-id", "", "个人订阅 ID 过滤")
	cmd.Flags().StringVar(&personalOpts.ControlBaseURL, "personal-event-base-url", "", "个人事件控制面 base URL；默认由 MCP base 派生 /dws")
	cmd.Flags().StringVar(&personalOpts.StreamSourceID, "stream-source-id", strings.TrimSpace(os.Getenv("DWS_STREAM_SOURCE_ID")),
		"个人事件 sourceId；开源版默认 open，可由 edition 覆盖")
	hideEventInternalFlags(cmd, "as", "all", "all-editions", "client-id", "fail-on-orphan")
	helpers.DeclareLeafMetadata(cmd, helpers.LeafSpec{
		Safety: contract.SafetySpec{
			Effect: "read", Risk: "low",
			Confirmation: "not_required", Idempotency: "idempotent",
		},
		Contract: helpers.LeafContract{
			Identity: contract.ToolIdentitySpec{
				ProductID:      "event",
				Name:           "status",
				CanonicalPath:  "event.status",
				CLIPath:        "event status",
				PrimaryCLIPath: "event status",
			},
			Description: "查看个人事件订阅、本地 bus 与消费进程状态",
			Interface: &contract.InterfaceSpec{
				Mode:         "composite",
				Availability: "available",
				Reason:       "Reviewed composite workflow: the command reads the remote personal-event subscription control plane and combines it with local bus and consumer state; no single pinned RPC represents the result.",
			},
			Selection: contract.SelectionSpec{
				AgentSummary: "查看个人事件订阅、本地 bus 与消费进程状态",
				UseWhen:      []string{"需要确认订阅是否活跃、bus/consume 是否仍在运行"},
				AvoidWhen:    []string{"取消订阅用 event stop；列事件目录用 event list"},
				Examples:     []string{"dws event status --format json"},
			},
		},
	})
	return cmd
}

// ─────────────────────────────────────────────────────────────────────
//  list/status 共享 helpers
// ─────────────────────────────────────────────────────────────────────

// collectEntries resolves which BusEntries the command should operate on
// and queries each one's live status. Returns at most one entry when
// neither --all nor --all-editions is set.
func collectEntries(c *cobra.Command, clientIDOver string, all, allEditions bool) ([]busctl.EntryStatus, error) {
	configDir := defaultConfigDir()
	editionName := editionNameOrDefault()

	// --all-editions trumps --all (scan whole tree)
	if allEditions {
		entries, err := eventEnumerateBuses(configDir, "")
		if err != nil {
			return nil, err
		}
		return queryAll(entries), nil
	}
	if all {
		entries, err := eventEnumerateBuses(configDir, editionName)
		if err != nil {
			return nil, err
		}
		return queryAll(entries), nil
	}

	// Single ClientID path. If --client-id passed use it directly,
	// otherwise resolve via strict resolver.
	clientID := clientIDOver
	if clientID == "" {
		resolved, _, _, _, err := eventResolveAppCredentials(configDir)
		if err != nil {
			return nil, fmt.Errorf("event status: resolve credentials: %w (or pass --client-id)", err)
		}
		clientID = resolved
	}
	hash := dwsevent.ClientIDHash(clientID)
	entry := eventFindBus(configDir, editionName, hash)
	if entry == nil {
		// No directory at all — render an empty "not running" so the user
		// sees a useful answer instead of an error.
		return []busctl.EntryStatus{
			{Entry: busctl.BusEntry{
				WorkDir:      eventWorkDir(configDir, editionName, dwsevent.SourceKindAppStream, hash),
				Edition:      editionName,
				SourceKind:   dwsevent.SourceKindAppStream,
				ClientIDHash: hash,
				IdentityHash: hash,
				State:        busctl.BusStateNotRunning,
				Meta: &bus.Meta{
					ClientID:     clientID,
					Edition:      editionName,
					SourceKind:   dwsevent.SourceKindAppStream,
					IdentityHash: hash,
				},
			}},
		}, nil
	}
	if entry.Meta == nil {
		entry.Meta = &bus.Meta{ClientID: clientID, Edition: editionName}
	}
	return []busctl.EntryStatus{eventQueryEntry(*entry)}, nil
}

func queryAll(entries []busctl.BusEntry) []busctl.EntryStatus {
	out := make([]busctl.EntryStatus, 0, len(entries))
	for _, e := range entries {
		out = append(out, eventQueryEntry(e))
	}
	return out
}

func buildListEntry(qs busctl.EntryStatus) listEntry {
	le := listEntry{
		ClientIDHash: qs.Entry.ClientIDHash,
		SourceKind:   qs.Entry.SourceKind,
		Edition:      qs.Entry.Edition,
		BusPID:       qs.Entry.HolderPID,
		BusState:     qs.Entry.State,
		WorkDir:      qs.Entry.WorkDir,
	}
	if qs.Entry.Meta != nil {
		le.ClientID = qs.Entry.Meta.ClientID
		le.SourceID = qs.Entry.Meta.SourceID
	}
	if qs.Live != nil {
		le.Consumers = qs.Live.Consumers
		le.PerType = qs.Live.PerEventTypeCounters
	}
	return le
}

// renderList prints either a table or a JSON array.
func renderList(w io.Writer, entries []listEntry, format string) error {
	if format == "json" {
		enc := json.NewEncoder(w)
		enc.SetIndent("", "  ")
		return enc.Encode(entries)
	}
	// Table: one line per consumer, prefixed by client_id. Buses with no
	// consumers still get one row (so users see they exist).
	tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
	fmt.Fprintln(tw, "SOURCE\tCLIENT_ID\tBUS\tCONSUMER PID\tEVENT KEYS\tSUBSCRIBE_ID\tRECEIVED\tDROPPED")
	for _, le := range entries {
		busDisplay := fmt.Sprintf("%s pid=%d", le.BusState, le.BusPID)
		if le.BusState == busctl.BusStateNotRunning {
			busDisplay = string(le.BusState)
		}
		clientLabel := le.ClientID
		if clientLabel == "" {
			clientLabel = "(unknown — hash=" + le.ClientIDHash + ")"
		}
		if len(le.Consumers) == 0 {
			fmt.Fprintf(tw, "%s\t%s\t%s\t-\t-\t-\t-\t-\n", sourceKindLabel(le.SourceKind), clientLabel, busDisplay)
			continue
		}
		for _, cs := range le.Consumers {
			keys := strings.Join(cs.EventTypes, ",")
			if keys == "" {
				keys = "(catch-all)"
			}
			subID := cs.SubscribeID
			if subID == "" {
				subID = "-"
			}
			fmt.Fprintf(tw, "%s\t%s\t%s\t%d\t%s\t%s\t%d\t%d\n",
				sourceKindLabel(le.SourceKind), clientLabel, busDisplay, cs.PID, keys, subID, cs.Received, cs.Dropped)
		}
	}
	return tw.Flush()
}

// renderStatus prints a multi-line block per bus. JSON mode dumps the raw
// EntryStatus slice.
func renderStatus(w io.Writer, entries []busctl.EntryStatus, format string) error {
	if format == "json" {
		enc := json.NewEncoder(w)
		enc.SetIndent("", "  ")
		return enc.Encode(entries)
	}
	for i, qs := range entries {
		if i > 0 {
			fmt.Fprintln(w)
		}
		renderStatusBlock(w, qs)
	}
	return nil
}

func renderStatusBlock(w io.Writer, qs busctl.EntryStatus) {
	clientLabel := qs.Entry.ClientIDHash
	if qs.Entry.Meta != nil && qs.Entry.Meta.ClientID != "" {
		clientLabel = qs.Entry.Meta.ClientID
	}
	fmt.Fprintf(w, "ClientID: %s\n", clientLabel)
	fmt.Fprintf(w, "  Edition  : %s\n", qs.Entry.Edition)
	fmt.Fprintf(w, "  Source   : %s\n", sourceKindLabel(qs.Entry.SourceKind))
	if qs.Entry.Meta != nil && qs.Entry.Meta.SourceID != "" {
		fmt.Fprintf(w, "  SourceID : %s\n", qs.Entry.Meta.SourceID)
	}
	fmt.Fprintf(w, "  Workdir  : %s\n", qs.Entry.WorkDir)

	switch qs.Entry.State {
	case busctl.BusStateNotRunning:
		fmt.Fprintln(w, "  Bus      : not_running")
		return
	case busctl.BusStateOrphan:
		fmt.Fprintf(w, "  Bus      : orphan  (last_pid=%d not alive)\n", qs.Entry.HolderPID)
		if qs.Entry.Meta != nil && !qs.Entry.Meta.StartedAt.IsZero() {
			fmt.Fprintf(w, "  Started  : %s (from bus.meta)\n", qs.Entry.Meta.StartedAt.Format(time.RFC3339))
		}
		fmt.Fprintln(w, "  Action   : run `dws event consume` to force-restart, or rm -rf the workdir")
		return
	}

	// Running: include live RPC results when present.
	uptime := ""
	if qs.Live != nil {
		uptime = fmt.Sprintf("uptime=%ds", qs.Live.Bus.UptimeSecs)
	} else if qs.Entry.Meta != nil && !qs.Entry.Meta.StartedAt.IsZero() {
		uptime = fmt.Sprintf("uptime=%s (from bus.meta)", time.Since(qs.Entry.Meta.StartedAt).Round(time.Second))
	}
	fmt.Fprintf(w, "  Bus      : running  pid=%d  %s\n", qs.Entry.HolderPID, uptime)

	if qs.Live == nil {
		fmt.Fprintln(w, "  (status RPC failed — bus may be shutting down)")
		return
	}
	live := qs.Live
	fmt.Fprintf(w, "  Source   : state=%s  state_source=%s  reconnects=%d\n",
		live.SourceState.State, live.SourceState.Source, live.SourceState.ReconnectCount)
	fmt.Fprintf(w, "  Consumers: %d active\n", len(live.Consumers))

	if len(live.PerEventTypeCounters) > 0 {
		fmt.Fprintln(w, "  Per-event-type counters (since bus start):")
		keys := make([]string, 0, len(live.PerEventTypeCounters))
		for k := range live.PerEventTypeCounters {
			keys = append(keys, k)
		}
		sort.Strings(keys)
		for _, k := range keys {
			c := live.PerEventTypeCounters[k]
			fmt.Fprintf(w, "    %s  received=%d  dropped=%d\n", k, c.Received, c.Dropped)
		}
	}
	if len(live.Consumers) > 0 {
		fmt.Fprintln(w, "  Consumers:")
		tw := tabwriter.NewWriter(w, 0, 0, 2, ' ', 0)
		fmt.Fprintln(tw, "    PID\tEVENT KEYS\tSUBSCRIBE_ID\tRECEIVED\tDROPPED")
		for _, cs := range live.Consumers {
			keys := strings.Join(cs.EventTypes, ",")
			if keys == "" {
				keys = "(catch-all)"
			}
			subID := cs.SubscribeID
			if subID == "" {
				subID = "-"
			}
			fmt.Fprintf(tw, "    %d\t%s\t%s\t%d\t%d\n", cs.PID, keys, subID, cs.Received, cs.Dropped)
		}
		_ = tw.Flush()
	}
}

func newEventStopCommand() *cobra.Command {
	return newEventStopCommandWithFlags()
}

func newEventStopCommandWithFlags(globalFlags ...*GlobalFlags) *cobra.Command {
	var asIdentity string
	var opts personalStopOptions
	cmd := &cobra.Command{
		Use:   "stop [subscribe_id]",
		Short: "取消个人事件订阅并停止本地消费",
		Long: `取消个人事件订阅并停止本地消费，清理对应本地消费状态，并尝试停止对应前台 consume。

传 subscribe_id 时取消该个人订阅；传 --all 时取消当前身份下本地记录的全部个人订阅。
取消后如果没有剩余本地个人订阅，会停止 personal bus。`,
		Args:              cobra.MaximumNArgs(1),
		DisableAutoGenTag: true,
		RunE: func(c *cobra.Command, args []string) error {
			// Local validation runs in DeclareLeafMetadata Validate (PreRunE)
			// before ConfirmSafety. Confirmation is the framework wrap.
			as, err := eventNormalizeAs(asIdentity)
			if err != nil {
				return err
			}
			if as == "user" {
				opts.SubscribeID = firstArg(args)
				opts.ExplicitToken = eventExplicitToken(globalFlags)
				opts.ClientIDOverride = eventExplicitClientID(globalFlags)
				if eventStopDryRun(c) {
					return writeEventStopDryRun(c, as, opts)
				}
				return eventRunPersonalStop(c, opts)
			}
			if eventStopDryRun(c) {
				return writeEventStopDryRun(c, as, opts)
			}
			configDir := defaultConfigDir()
			clientID, _, _, _, err := eventResolveAppCredentials(configDir)
			if err != nil {
				return fmt.Errorf("event stop: %w", err)
			}
			editionName := editionNameOrDefault()
			clientIDHash := dwsevent.ClientIDHash(clientID)
			workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindAppStream, clientIDHash)
			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
				}
				return err
			}
			fmt.Fprintln(c.OutOrStdout(), "bus stopped")
			return nil
		},
	}
	cmd.Flags().StringVar(&asIdentity, "as", "user", "事件身份: user")
	cmd.Flags().StringVar(&opts.ControlBaseURL, "personal-event-base-url", "", "个人事件控制面 base URL；默认由 MCP base 派生 /dws")
	cmd.Flags().StringVar(&opts.StreamSourceID, "stream-source-id", strings.TrimSpace(os.Getenv("DWS_STREAM_SOURCE_ID")),
		"个人事件 sourceId；开源版默认 open，可由 edition 覆盖")
	cmd.Flags().BoolVar(&opts.All, "all", false, "取消当前身份下本地记录的所有个人订阅")
	hideEventInternalFlags(cmd, "as")
	cli.AnnotateRuntimePositionals(cmd, contract.RuntimeSchemaPositional{
		Name:        "subscribe_id",
		Type:        "string",
		Description: "要取消的个人事件订阅 ID；与 --all 二选一",
		Required:    false,
		Index:       0,
	})
	helpers.DeclareLeafMetadata(cmd, helpers.LeafSpec{
		Safety: contract.SafetySpec{
			Effect: "destructive", Risk: "high",
			Confirmation: "user_required", Idempotency: "unknown",
		},
		Validate: func(c *cobra.Command, args []string) error {
			as, err := eventNormalizeAs(asIdentity)
			if err != nil {
				return err
			}
			if as == "user" {
				subscribeID := firstArg(args)
				hasSubscribeID := strings.TrimSpace(subscribeID) != ""
				if hasSubscribeID && opts.All {
					return fmt.Errorf("event stop --as user: subscribe_id and --all are mutually exclusive")
				}
				if !hasSubscribeID && !opts.All {
					return fmt.Errorf("event stop --as user: subscribe_id is required unless --all is set")
				}
				return nil
			}
			if err := rejectChangedFlags(c, "user", "all", "personal-event-base-url", "stream-source-id"); err != nil {
				return fmt.Errorf("event stop: %w", err)
			}
			if len(args) > 0 {
				return fmt.Errorf("event stop: subscribe_id is only supported with --as user")
			}
			return nil
		},
		Contract: helpers.LeafContract{
			Identity: contract.ToolIdentitySpec{
				ProductID:      "event",
				Name:           "stop",
				CanonicalPath:  "event.stop",
				CLIPath:        "event stop",
				PrimaryCLIPath: "event stop",
			},
			Description: "取消个人事件订阅并停止对应本地消费",
			DryRun:      &contract.DryRunSpec{PreviewKind: "request", RemoteReads: false},
			Interface: &contract.InterfaceSpec{
				Mode:         "composite",
				Availability: "available",
				Reason:       "Reviewed composite workflow: the command deletes remote personal-event subscriptions, interrupts local consumers, updates local state, and may stop the local bus; no single pinned RPC represents the workflow.",
			},
			Selection: contract.SelectionSpec{
				AgentSummary: "取消个人事件订阅并停止对应本地消费",
				UseWhen:      []string{"用户明确要求取消已知 subscribe_id（或清理全部）并停止消费"},
				AvoidWhen: []string{
					"只需检查状态时用 event status",
					"目标订阅不明确或用户未确认时不要停止",
				},
				Examples: []string{"dws event stop SUBSCRIBE_ID --dry-run"},
			},
		},
	})
	return cmd
}

func eventStopDryRun(cmd *cobra.Command) bool {
	value, _ := cmd.Flags().GetBool("dry-run")
	return value
}

func eventExplicitToken(globalFlags []*GlobalFlags) string {
	if len(globalFlags) == 0 || globalFlags[0] == nil {
		return ""
	}
	return strings.TrimSpace(globalFlags[0].Token)
}

func eventExplicitClientID(globalFlags []*GlobalFlags) string {
	if len(globalFlags) == 0 || globalFlags[0] == nil {
		return ""
	}
	return strings.TrimSpace(globalFlags[0].ClientID)
}

func writeEventStopDryRun(cmd *cobra.Command, identity string, opts personalStopOptions) error {
	payload := map[string]any{
		"dry_run":  true,
		"action":   "event.stop",
		"identity": strings.TrimSpace(identity),
		"all":      opts.All,
	}
	if subscribeID := strings.TrimSpace(opts.SubscribeID); subscribeID != "" {
		payload["subscribe_id"] = subscribeID
	}
	encoder := json.NewEncoder(cmd.OutOrStdout())
	encoder.SetIndent("", "  ")
	return encoder.Encode(payload)
}

// ─────────────────────────────────────────────────────────────────────
//  helpers
// ─────────────────────────────────────────────────────────────────────

// editionNameOrDefault returns edition.Get().Name with "open" fallback.
// Centralised so every event subcommand agrees on the path prefix.
func editionNameOrDefault() string {
	name := edition.Get().Name
	if name == "" {
		return "open"
	}
	return name
}

// defaultIPCEndpoint returns the canonical Unix socket path / Windows pipe
// name for the bus. Thin wrapper over dwsevent.IPCEndpoint, which owns the
// platform shape and the too-long-socket-path fallback.
func defaultIPCEndpoint(workDir, editionName string, sourceKind dwsevent.SourceKind, identityHash string) string {
	return dwsevent.IPCEndpoint(workDir, editionName, sourceKind, identityHash)
}

func eventWorkDir(configDir, editionName string, sourceKind dwsevent.SourceKind, identityHash string) string {
	if sourceKind == "" {
		sourceKind = dwsevent.SourceKindAppStream
	}
	return filepath.Join(configDir, "events", editionName, string(sourceKind), identityHash)
}

func sourceKindLabel(kind dwsevent.SourceKind) string {
	if kind == "" {
		return string(dwsevent.SourceKindAppStream)
	}
	return string(kind)
}

func normalizeEventAs(v string) (string, error) {
	v = strings.ToLower(strings.TrimSpace(v))
	switch v {
	case "", "user":
		return "user", nil
	case "app", "bot":
		// App stream implementation is intentionally retained for future use,
		// while the public event command currently exposes only user events.
		return "", fmt.Errorf("app event is not publicly available yet")
	default:
		return "", fmt.Errorf("--as only supports user")
	}
}

func hideEventInternalFlags(cmd *cobra.Command, names ...string) {
	for _, name := range names {
		_ = cmd.Flags().MarkHidden(name)
	}
}

func rejectPersonalEventUnsupportedFlags(c *cobra.Command, names ...string) error {
	changed := make([]string, 0, len(names))
	for _, name := range names {
		if f := c.Flags().Lookup(name); f != nil && f.Changed {
			changed = append(changed, "--"+name)
		}
	}
	if len(changed) == 0 {
		return nil
	}
	return fmt.Errorf("%s are not supported for personal events", strings.Join(changed, ", "))
}

func rejectPersonalMultiEventFlags(c *cobra.Command, names ...string) error {
	changed := make([]string, 0, len(names))
	for _, name := range names {
		if f := c.Flags().Lookup(name); f != nil && f.Changed {
			changed = append(changed, "--"+name)
		}
	}
	if len(changed) == 0 {
		return nil
	}
	return fmt.Errorf("%s are not supported when consuming multiple events", strings.Join(changed, ", "))
}

func rejectChangedFlags(c *cobra.Command, supportedAs string, names ...string) error {
	changed := make([]string, 0, len(names))
	for _, name := range names {
		if f := c.Flags().Lookup(name); f != nil && f.Changed {
			changed = append(changed, "--"+name)
		}
	}
	if len(changed) == 0 {
		return nil
	}
	return fmt.Errorf("%s are only supported with --as %s", strings.Join(changed, ", "), supportedAs)
}

func firstArg(args []string) string {
	if len(args) == 0 {
		return ""
	}
	return args[0]
}

// shouldWatchStdinEOF gates the stdin-EOF shutdown watcher (AI-subprocess
// contract). It arms only for a parent-controlled, pipe-style stdin on an
// unbounded run:
//   - bounded runs (--max-events / --duration) already have their own
//     lifecycle, so stdin is irrelevant;
//   - char devices (an interactive TTY, or /dev/null) are excluded, so a
//     terminal Ctrl-D and the common `< /dev/null` launch do NOT trigger a
//     surprise shutdown. Only a pipe / regular file — an stdin a parent
//     holds and can close to stop us — arms the watcher.
func shouldWatchStdinEOF(maxEvents int, duration time.Duration) bool {
	if maxEvents > 0 || duration > 0 {
		return false
	}
	fi, err := os.Stdin.Stat()
	if err != nil {
		return false
	}
	return fi.Mode()&os.ModeCharDevice == 0
}

func applyEventConsumeStdin(cfg *consume.Config, maxEvents int, duration time.Duration, stdin io.Reader) {
	if cfg != nil && shouldWatchStdinEOF(maxEvents, duration) {
		cfg.Stdin = stdin
	}
}

// eventTypesWithDefault picks the catch-all list from registry when the
// user did not pass --event-types.
func eventTypesWithDefault(types []string) []string {
	if len(types) > 0 {
		return types
	}
	return registry.CatchAllEventTypes()
}