Merge pull request #950 from DingTalk-Real-AI/codex/changelog-v1.0.58-beta.3

docs: seal v1.0.58-beta.3 changelog

CHANGELOG.md

@@ -6,6 +6,8 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
 
 ## [Unreleased]
 
+## [1.0.58-beta.3] - 2026-08-11
+
 ### Added
 
 - **Aitable workflow execution and history** — adds `dws aitable workflow run` for confirmed asynchronous execution of scheduled or record-triggered workflows, plus `dws aitable workflow history` for status-, time-, and page-filtered execution records. The commands map directly to `aitable/run_workflow` and `aitable/get_flow_record_list`, validate trigger-specific arguments locally, and document the `executionId` / `instanceId` correlation.
@@ -13,6 +15,9 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
   `--at-open-dingtalk-ids` and `--at-all` for group cards, passing mention
   targets to the initial card-creation request and prepending its returned
   `atTag` to the automatic streaming update.
+- **Personal OA approval events** — personal event consumers now support task
+  creation, completion, redirection, instance start, termination, and
+  completion events, with typed output and matching usage documentation.
 
 ### Fixed
 
@@ -22,6 +27,18 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
   a successful local write. The result includes the saved path and byte size;
   document exports additionally report the node, requested format, job/task
   ID, and final status.
+- **IM search and card-write safety** — conversation-scoped search now fails
+  closed when the target cannot be verified, and streaming-card updates require
+  business evidence rather than a transport-only success response.
+- **Document shortcut reliability** — document write, readback verification,
+  pagination, template/version discovery, export, media, and local-file
+  workflows now preserve compatibility while rejecting ambiguous write results.
+- **Event runtime-token handoff** — personal `event consume`, `status`,
+  `stop`, and `+listen-im` honor the root `--token` without falling back to a
+  stale OAuth profile. Detached buses negotiate an owner-only, memory-only IPC
+  credential channel; tokens are never placed in child argv, environment,
+  profiles, logs, or run-state files.
+
 ### Changed
 
 - **Minutes `permission apply --policy` type** — `--policy` is now declared as
@@ -31,6 +48,15 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
 - **Minutes skill references** — document `permission apply` in both Minutes
   skill references: list it in the command trees, describe its policy values and
   how it differs from `permission add`, and add its intent routing.
+- **Chat paging guidance** — typed chat message commands now document
+  `--page-all`, aggregate result shapes, and cursor behavior in CLI Help and
+  Agent selection examples.
+- **Calendar skill parity** — mono and multi Calendar references are aligned to
+  prevent documentation drift without changing CLI behavior.
+- **Release engineering** — CI now shards helper-package changes through the
+  full race suite, widens a flaky stdio idempotency test budget, governs exact
+  reviewed CLI/Schema type migrations, and lets authorized maintainers trigger
+  internal MCP evaluation with a reviewed `/eval` PR comment.
 
 ## [1.0.58-beta.2] - 2026-08-10
 
@@ -60,6 +86,10 @@ The format is inspired by [Keep a Changelog](https://keepachangelog.com/) and th
 - **CLI 接口兼容门禁支持 reviewed flag 类型豁免**（无用户可见变更）— `authoritative-interface-integrity` 与 `check-command-compatibility.sh` 此前一律拒绝历史命令的 flag 类型变更，即使新类型只是把同一套校验从 RunE 前移到解析期，也没有任何评审通道。现在两道门禁各带一张精确豁免表：命令路径 + flag 名 + 旧类型 → 新类型四元组全等才命中、方向敏感（`string`→`int` 与 `int`→`string` 是两个不同的键，只有被评审的方向可用），且仅当该 flag 的其他契约（shorthand / required / hidden / no-opt / scope）纹丝不动时才放行，因此豁免夹带不了别的破坏。首条也是目前唯一一条登记的是 `dws minutes permission apply --policy` 的 `string` → `int`（配合 #912）：旧实现在 RunE 里做 `strconv.ParseInt(v, 10, 64)` 再校验 `[2,4]`，新实现由 pflag 以 `strconv.ParseInt(s, 0, 64)` 解析后仍校验 `[2,4]`，**历史上能成功的调用集是新调用集的子集**（base 0 额外接受 `0x3` 这类写法，只放宽不收紧），非法值依然失败、只是报错文案与时机前移；flag 默认值由 `""` 变 `"0"` 是类型的必然结果，两道门禁都不比较默认值，且该 flag 必须显式给出、默认值不可达。两张表必须逐字一致并有守卫测试锚定漂移——重复是被迫的而非选择：`check-authoritative-interface-baselines.sh` 会把整个 `scripts/policy/interface-baseline` 目录复制进检出历史版本的 worktree 再编译，那份拷贝不能 import 本分支新增的包。
 - **Schema 兼容门禁支持 reviewed 参数类型豁免**（无用户可见变更）— 接上一条。`schema-compatibility` 是同一个 `Interface Integrity` job 里排在两道 CLI 接口门禁之后的第三道检查，此前也一律拒绝已发布参数的 `type` 变更。由于前两道先失败、`set -e` 让它从未在 CI 上暴露，上一条豁免只解决了三分之二。现在 `checkParameterCompatibility` 也带一张精确豁免表：`<product>/<tool id>` + 参数名 + 旧类型 + 新类型四元组全等才命中、方向敏感，且仅当该参数**除 `type` 外的全部已发布字段逐字段相等**时才放行。这里刻意用相等性比较而非「没有产生其他兼容性错误」：放宽 `required` / `cli_required`、清空 `required_when`、扩宽 `enum`、清空 `interface_type`、经 reviewed mapping exclusion 清空 `property`——这些变化单独看都是兼容的、根本不产生错误，若以错误列表代替相等性检查，它们就能搭着一次已评审的类型迁移一起蒙混过关。结构体整体比较还意味着将来给 `parameterSchema` 新增字段时会自动纳入守卫，而不是悄悄放宽每一条既有条目。唯一条目是 `minutes/minutes.apply_minutes_permission` 的 `policy` 由 `"string"` 迁移到 `"integer"`（配合 #912）：该 `type` 由 Cobra flag 类型投影而来（provenance `cobra_flag_type`），描述的是 CLI 如何接受取值；消费方据此拼装的是命令行，而 `--policy 4` 在两种声明下是同一个 argv，加引号的 `--policy "4"` 到 pflag 仍是 4，RunE 也仍校验 `[2,4]`——而且该参数映射的 property `policyId` 一直以数字上报，新声明比旧声明更贴近真实请求。表里的类型值必须是 `schemaType` 实际产出的带引号形态（`"string"` 而非裸 `string`），守卫测试用 `schemaType` 复算并校验类型名属于 JSON Schema 的封闭取值集合——`reviewedInterfaceRefRedirect` 曾因键的书写形态错误两次静默失效，这里不重犯。
 
+### Fixed
+
+- **Event runtime-token handoff** — personal `event consume`, `status`, `stop`, and `+listen-im` now honor the existing root `--token` instead of falling back to a stale local OAuth profile. Detached personal-event buses negotiate the credential only after an additive capability handshake, receive and rotate it through owner-only local IPC, and keep it in memory; the token is never forwarded through child argv, environment variables, profiles, logs, or run-state files. Existing OAuth and multi-profile behavior is unchanged when `--token` is absent. A new client refuses to send a runtime token to an older bus and leaves its existing consumers and subscriptions untouched; the recovery message asks users to inspect `event status --as user`, preview `event stop --as user --all --dry-run`, and explicitly confirm `event stop --as user --all --yes` before retrying.
+
 ## [1.0.58-beta.1] - 2026-08-07
 
 ### Added

docs/event-subprocess-contract.md

@@ -4,7 +4,7 @@ Defines the stable `dws event consume` subprocess contract so an
 orchestrator can determine when the consumer is ready, stop it cleanly,
 and machine-read why it exited.
 
-Scope of this branch: the five **contract** items below. Reconnect
+Scope of this branch: the six **contract** items below. Reconnect
 resilience (keeping the stream alive across a transient upstream drop) is
 tracked separately and intentionally out of scope here.
 
@@ -159,6 +159,46 @@ marker; reconnecting an established Stream remains a separate mechanism.
   `terminal_hold`, and identity-scoped cleanup; skill/docs tests pin the
   operational recovery instructions.
 
+### 6. Host runtime-token handoff
+
+When the root command carries an explicit host-supplied `--token`, personal
+event control requests and the foreground Stream use that token with higher
+priority than local OAuth. A detached bus receives it only through the
+owner-only local IPC transport:
+
+1. The child starts in runtime-token mode with non-sensitive identity and
+   ticket metadata only; neither its argv nor environment contains the token.
+2. The consumer sends `Hello` with `credential_mode=runtime_token`.
+3. The bus advertises the additive `runtime_token_v1` capability and its
+   in-memory credential generation in `HelloAck`.
+4. Only after that capability is confirmed does the consumer send a bounded
+   `credential_update` frame. The bus applies it with generation CAS, replies
+   with `credential_update_ack`, and registers the consumer only on success.
+
+The bus blocks ticket acquisition until the first runtime credential arrives.
+A later invocation may rotate Token A to Token B on a compatible existing bus;
+the current WebSocket remains connected and the next ticket request or natural
+reconnect uses B. If a 401 rejects the current runtime token, only an already
+installed newer generation is retried; the runtime path never refreshes or
+falls back to a local OAuth profile and never suggests `dws auth login`.
+
+Clients do not send a token to a bus that lacks the capability, do not stop
+other consumers automatically, and fail before printing the ready marker. With
+no explicit `--token`, the original OAuth, refresh, profile, and old-client to
+new-bus protocol behavior remains unchanged.
+
+**Verification**
+- T6a: a stale local Token A and root Token B produce control and ticket
+  requests authenticated only with B.
+- T6b: compatible bus reuse supports A-to-B rotation and generation conflicts;
+  401 retries only an already-installed newer runtime token.
+- T6c: an old bus receives no credential and remains running; the new consumer
+  exits before its ready marker.
+- T6d: a canary credential is absent from child argv/environment, dry-run,
+  stdout/stderr, `bus.meta`, `bus.log`, run state, and returned errors.
+- T6e: no-token OAuth, refresh, multi-profile, marker/cache, and bus-reuse tests
+  continue to pass.
+
 ## Out of scope (next branch)
 
 **Reconnect resilience** — today `personal source` retries only

internal/app/event_command.go

@@ -38,6 +38,7 @@ import (
 	"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"
@@ -74,7 +75,7 @@ var (
 
 // newEventCommand returns the `event` parent command and all its subcommands.
 // Wired into root.go's utilityCommands list.
-func newEventCommand() *cobra.Command {
+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{
@@ -99,12 +100,12 @@ func newEventCommand() *cobra.Command {
 		RunE:              func(c *cobra.Command, _ []string) error { return c.Help() },
 	}
 	cmd.AddCommand(
-		newEventListenIMCommand(),
-		newEventConsumeCommand(),
+		newEventListenIMCommand(globalFlags...),
+		newEventConsumeCommand(globalFlags...),
 		newEventListCommand(),
 		newEventSchemaCommand(),
-		newEventStatusCommand(),
-		newEventStopCommand(),
+		newEventStatusCommandWithFlags(globalFlags...),
+		newEventStopCommandWithFlags(globalFlags...),
 		newEventBusCommand(),
 	)
 	return cmd
@@ -114,7 +115,7 @@ func newEventCommand() *cobra.Command {
 //  event consume
 // ─────────────────────────────────────────────────────────────────────
 
-func newEventConsumeCommand() *cobra.Command {
+func newEventConsumeCommand(globalFlags ...*GlobalFlags) *cobra.Command {
 	var (
 		eventTypes   []string
 		filter       string
@@ -170,6 +171,8 @@ SIGTERM、关 stdin，或先用 dws event stop <subscribe_id> --dry-run 预览
 				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
@@ -563,6 +566,8 @@ func newEventBusCommand() *cobra.Command {
 		clientIDOverride string
 		idleTimeout      time.Duration
 		sourceKindRaw    string
+		runtimeTokenMode bool
+		identityHashFlag string
 		streamOpts       eventStreamTicketOptions
 	)
 	cmd := &cobra.Command{
@@ -599,23 +604,45 @@ func newEventBusCommand() *cobra.Command {
 				sourceKind = dwsevent.SourceKindAppStream
 			}
 			if sourceKind == dwsevent.SourceKindPersonalStream {
-				identity, err := eventResolvePersonal(ctx, configDir, streamOpts.SourceID)
-				if err != nil {
-					return failEarly(fmt.Errorf("event _bus: %w", err))
+				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())
 				}
-				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)
@@ -628,17 +655,18 @@ func newEventBusCommand() *cobra.Command {
 					}
 				}
 				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(),
+					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)
@@ -698,12 +726,18 @@ func newEventBusCommand() *cobra.Command {
 		"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
 }
 
@@ -822,6 +856,10 @@ func newEventListCommand() *cobra.Command {
 // ─────────────────────────────────────────────────────────────────────
 
 func newEventStatusCommand() *cobra.Command {
+	return newEventStatusCommandWithFlags()
+}
+
+func newEventStatusCommandWithFlags(globalFlags ...*GlobalFlags) *cobra.Command {
 	var (
 		all          bool
 		allEditions  bool
@@ -847,6 +885,8 @@ func newEventStatusCommand() *cobra.Command {
 					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 {
@@ -1138,6 +1178,10 @@ func renderStatusBlock(w io.Writer, qs busctl.EntryStatus) {
 }
 
 func newEventStopCommand() *cobra.Command {
+	return newEventStopCommandWithFlags()
+}
+
+func newEventStopCommandWithFlags(globalFlags ...*GlobalFlags) *cobra.Command {
 	var asIdentity string
 	var opts personalStopOptions
 	cmd := &cobra.Command{
@@ -1158,6 +1202,8 @@ func newEventStopCommand() *cobra.Command {
 			}
 			if as == "user" {
 				opts.SubscribeID = firstArg(args)
+				opts.ExplicitToken = eventExplicitToken(globalFlags)
+				opts.ClientIDOverride = eventExplicitClientID(globalFlags)
 				if eventStopDryRun(c) {
 					return writeEventStopDryRun(c, as, opts)
 				}
@@ -1261,6 +1307,20 @@ func eventStopDryRun(cmd *cobra.Command) bool {
 	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,

internal/app/event_listen_im.go

@@ -65,7 +65,7 @@ func (eventTargetReader) CallMCPData(product, tool string, params map[string]any
 
 var eventListenIMReader = func() targetresolver.Reader { return eventTargetReader{} }
 
-func newEventListenIMCommand() *cobra.Command {
+func newEventListenIMCommand(globalFlags ...*GlobalFlags) *cobra.Command {
 	var opts listenIMOptions
 	cmd := &cobra.Command{
 		Use:   "+listen-im",
@@ -91,6 +91,8 @@ func newEventListenIMCommand() *cobra.Command {
 				StreamTicketMode: opts.StreamTicketMode,
 				StreamTicketURL:  opts.StreamTicketURL,
 				StreamSourceID:   opts.StreamSourceID,
+				ExplicitToken:    eventExplicitToken(globalFlags),
+				ClientIDOverride: eventExplicitClientID(globalFlags),
 				Common: commonConsumeOptions{
 					FormatRaw: "ndjson",
 					MaxEvents: opts.MaxEvents,

internal/app/event_personal_attempts.go

@@ -28,6 +28,7 @@ import (
 
 	apperrors "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/errors"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
 )
 
@@ -184,6 +185,25 @@ func (r *personalSubscriptionAttemptReservation) completeSuccess() error {
 	return nil
 }
 
+// releaseRuntimeTokenFailure releases the in-flight claim without recording a
+// cross-invocation hold. A host may supply a fresh token on the very next
+// command, which must be allowed to retry immediately.
+func (r *personalSubscriptionAttemptReservation) releaseRuntimeTokenFailure() error {
+	if r == nil {
+		return runtimecred.ErrRuntimeTokenRejected
+	}
+	if r.store == nil || r.claim == nil {
+		return personalSubscriptionGuardError(errors.Join(
+			runtimecred.ErrRuntimeTokenRejected,
+			errors.New("personal event: subscription attempt reservation is incomplete"),
+		))
+	}
+	if err := r.store.Release(r.claim); err != nil {
+		return personalSubscriptionGuardError(errors.Join(runtimecred.ErrRuntimeTokenRejected, err))
+	}
+	return runtimecred.ErrRuntimeTokenRejected
+}
+
 func (r *personalSubscriptionAttemptReservation) completeFailure(
 	ctx context.Context,
 	failedIndex int,

internal/app/event_personal_command.go

@@ -14,6 +14,7 @@
 package app
 
 import (
+	"bytes"
 	"context"
 	"crypto/sha256"
 	"encoding/hex"
@@ -40,6 +41,7 @@ import (
 	"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/runtimecred"
 	"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"
@@ -82,6 +84,8 @@ type personalConsumeOptions struct {
 	StreamTicketMode string
 	StreamTicketURL  string
 	StreamSourceID   string
+	ExplicitToken    string
+	ClientIDOverride string
 }
 
 type personalListOptions struct {
@@ -92,19 +96,23 @@ type personalListOptions struct {
 }
 
 type personalStatusOptions struct {
-	EventKey       string
-	Status         string
-	SubscribeID    string
-	Format         string
-	ControlBaseURL string
-	StreamSourceID string
+	EventKey         string
+	Status           string
+	SubscribeID      string
+	Format           string
+	ControlBaseURL   string
+	StreamSourceID   string
+	ExplicitToken    string
+	ClientIDOverride string
 }
 
 type personalStopOptions struct {
-	SubscribeID    string
-	All            bool
-	ControlBaseURL string
-	StreamSourceID string
+	SubscribeID      string
+	All              bool
+	ControlBaseURL   string
+	StreamSourceID   string
+	ExplicitToken    string
+	ClientIDOverride string
 }
 
 type personalStreamSourceOptions struct {
@@ -113,6 +121,8 @@ type personalStreamSourceOptions struct {
 	TicketMode       string
 	TicketURL        string
 	ClientIDOverride string
+	CredentialBroker *runtimecred.Broker
+	RuntimeTokenMode bool
 }
 
 var (
@@ -142,10 +152,19 @@ var (
 	personalResolveAuxiliaryAccessToken = ResolveAuxiliaryAccessToken
 	personalForceRefreshRejectedToken   = forceRefreshRejectedAccessToken
 	personalLoadTokenData               = authpkg.LoadTokenData
+	personalLoadProfiles                = authpkg.LoadProfiles
 	personalClientID                    = authpkg.ClientID
+	personalRuntimeEventClientID        = runtimePersonalEventClientID
 	personalResolveAppCredentialsStrict = authpkg.ResolveAppCredentialsStrict
 )
 
+func runtimePersonalEventClientID() string {
+	if clientID := strings.TrimSpace(edition.Get().AuthClientID); clientID != "" {
+		return clientID
+	}
+	return strings.TrimSpace(os.Getenv("DWS_CLIENT_ID"))
+}
+
 func newEventSchemaCommand() *cobra.Command {
 	var asIdentity string
 	var formatRaw string
@@ -283,7 +302,7 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
 	projector := personalEventProjector(opts.DebugRawEvents, opts.Flatten)
 
 	configDir := defaultConfigDir()
-	identity, err := personalResolveEventIdentity(ctx, configDir, opts.StreamSourceID)
+	identity, err := resolvePersonalEventIdentityForToken(ctx, configDir, opts.StreamSourceID, opts.ExplicitToken, opts.ClientIDOverride)
 	if err != nil {
 		return fmt.Errorf("event consume --as user: %w", err)
 	}
@@ -291,13 +310,24 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
 	editionName := editionNameOrDefault()
 	workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
 	ipcEndpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
-	spawnProfileSelector := personalBusProfileSelector(configDir, identity)
+	spawnProfileSelector := ""
+	if strings.TrimSpace(opts.ExplicitToken) == "" {
+		spawnProfileSelector = personalBusProfileSelector(configDir, identity)
+	}
+	spawnArgs := personalBusSpawnArgsForToken(
+		identity,
+		identityHash,
+		opts.StreamTicketMode,
+		opts.StreamTicketURL,
+		spawnProfileSelector,
+		opts.ExplicitToken,
+	)
 
 	routes, err := consume.ParseRoutes(opts.Common.RoutesRaw)
 	if err != nil {
 		return fmt.Errorf("event consume --as user: %w", personalSubscriptionValidationError(err))
 	}
-	client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity)
+	client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity, opts.ExplicitToken)
 	if opts.Common.DryRun {
 		if strings.TrimSpace(opts.SubscribeID) == "" {
 			if err := validatePersonalSubscriptionOptions(opts); err != nil {
@@ -314,7 +344,7 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
 			WorkDir:        workDir,
 			IPCEndpoint:    ipcEndpoint,
 			ClientID:       identity.ClientID,
-			SpawnExtraArgs: personalBusSpawnArgs(identity, opts.StreamTicketMode, personalEventStreamTicketURL(opts.StreamTicketURL, configDir), spawnProfileSelector),
+			SpawnExtraArgs: personalBusSpawnArgsForToken(identity, identityHash, opts.StreamTicketMode, personalEventStreamTicketURL(opts.StreamTicketURL, configDir), spawnProfileSelector, opts.ExplicitToken),
 			Compact:        opts.Common.Compact,
 			MaxEvents:      opts.Common.MaxEvents,
 			Duration:       opts.Common.Duration,
@@ -341,7 +371,8 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
 		WorkDir:        workDir,
 		IPCEndpoint:    ipcEndpoint,
 		ClientID:       identity.ClientID,
-		SpawnExtraArgs: personalBusSpawnArgs(identity, opts.StreamTicketMode, opts.StreamTicketURL, spawnProfileSelector),
+		SpawnExtraArgs: spawnArgs,
+		RuntimeToken:   strings.TrimSpace(opts.ExplicitToken),
 		Compact:        opts.Common.Compact,
 		MaxEvents:      opts.Common.MaxEvents,
 		Duration:       opts.Common.Duration,
@@ -370,13 +401,25 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
 		}
 	}
 
-	var foregroundSource *source.PersonalSource
+	var (
+		foregroundSource *source.PersonalSource
+		foregroundBroker *runtimecred.Broker
+	)
 	if opts.Common.Foreground {
+		explicitToken := strings.TrimSpace(opts.ExplicitToken)
+		foregroundBroker = newPersonalCredentialBroker(configDir, explicitToken != "", false)
+		if explicitToken != "" {
+			if _, err := foregroundBroker.Update(0, explicitToken); err != nil {
+				return personalSubscriptionValidationError(err)
+			}
+		}
 		foregroundSource, err = personalNewStreamSource(ctx, personalStreamSourceOptions{
-			ConfigDir:  configDir,
-			Identity:   identity,
-			TicketMode: opts.StreamTicketMode,
-			TicketURL:  opts.StreamTicketURL,
+			ConfigDir:        configDir,
+			Identity:         identity,
+			TicketMode:       opts.StreamTicketMode,
+			TicketURL:        opts.StreamTicketURL,
+			CredentialBroker: foregroundBroker,
+			RuntimeTokenMode: explicitToken != "",
 		})
 		if err != nil {
 			return personalSubscriptionValidationError(err)
@@ -398,6 +441,10 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
 	}
 	sub, eventKey, ruleType, err := personalEnsureSubscription(ctx, client, identity, opts)
 	if err != nil {
+		if strings.TrimSpace(opts.ExplicitToken) != "" && personalRuntimeTokenControlRejection(err) {
+			err = attempt.releaseRuntimeTokenFailure()
+			return fmt.Errorf("event consume --as user: %w", err)
+		}
 		err = attempt.completeFailure(ctx, 0, 0, err, nil)
 		return fmt.Errorf("event consume --as user: %w", err)
 	}
@@ -476,14 +523,15 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
 	}
 	if opts.Common.Foreground {
 		busCfg := bus.Config{
-			WorkDir:      workDir,
-			IPCEndpoint:  ipcEndpoint,
-			ClientID:     identity.ClientID,
-			Edition:      editionName,
-			SourceKind:   dwsevent.SourceKindPersonalStream,
-			IdentityHash: identityHash,
-			SourceID:     identity.SourceID,
-			Source:       foregroundSource,
+			WorkDir:          workDir,
+			IPCEndpoint:      ipcEndpoint,
+			ClientID:         identity.ClientID,
+			Edition:          editionName,
+			SourceKind:       dwsevent.SourceKindPersonalStream,
+			IdentityHash:     identityHash,
+			SourceID:         identity.SourceID,
+			Source:           foregroundSource,
+			CredentialBroker: foregroundBroker,
 		}
 		bus.ApplyEnvTuning(&busCfg)
 		return personalBusRun(ctx, busCfg)
@@ -517,7 +565,7 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
 
 	ctx := c.Context()
 	configDir := defaultConfigDir()
-	identity, err := personalResolveEventIdentity(ctx, configDir, opts.StreamSourceID)
+	identity, err := resolvePersonalEventIdentityForToken(ctx, configDir, opts.StreamSourceID, opts.ExplicitToken, opts.ClientIDOverride)
 	if err != nil {
 		return fmt.Errorf("event consume --as user: %w", err)
 	}
@@ -525,7 +573,10 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
 	editionName := editionNameOrDefault()
 	workDir := eventWorkDir(configDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
 	ipcEndpoint := defaultIPCEndpoint(workDir, editionName, dwsevent.SourceKindPersonalStream, identityHash)
-	spawnProfileSelector := personalBusProfileSelector(configDir, identity)
+	spawnProfileSelector := ""
+	if strings.TrimSpace(opts.ExplicitToken) == "" {
+		spawnProfileSelector = personalBusProfileSelector(configDir, identity)
+	}
 	routes, err := consume.ParseRoutes(opts.Common.RoutesRaw)
 	if err != nil {
 		return fmt.Errorf("event consume --as user: %w", personalSubscriptionValidationError(err))
@@ -534,7 +585,7 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
 		WorkDir:        workDir,
 		IPCEndpoint:    ipcEndpoint,
 		ClientID:       identity.ClientID,
-		SpawnExtraArgs: personalBusSpawnArgs(identity, opts.StreamTicketMode, personalEventStreamTicketURL(opts.StreamTicketURL, configDir), spawnProfileSelector),
+		SpawnExtraArgs: personalBusSpawnArgsForToken(identity, identityHash, opts.StreamTicketMode, personalEventStreamTicketURL(opts.StreamTicketURL, configDir), spawnProfileSelector, opts.ExplicitToken),
 		Compact:        opts.Common.Compact,
 		MaxEvents:      opts.Common.MaxEvents,
 		Duration:       opts.Common.Duration,
@@ -560,8 +611,9 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
 		printPersonalMultiDryRun(c.ErrOrStderr(), baseCfg, plans)
 		return nil
 	}
+	baseCfg.RuntimeToken = strings.TrimSpace(opts.ExplicitToken)
 
-	client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity)
+	client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity, opts.ExplicitToken)
 	attempt, err := reservePersonalSubscriptionAttempts(
 		workDir,
 		client,
@@ -598,6 +650,10 @@ func runPersonalEventConsumeMany(c *cobra.Command, opts personalConsumeOptions)
 		if personalSubscriptionCanceled(ctx, cause) {
 			cleanupCtx = ctx
 		}
+		if strings.TrimSpace(opts.ExplicitToken) != "" && personalRuntimeTokenControlRejection(cause) {
+			cleanup(cleanupCtx)
+			return attempt.releaseRuntimeTokenFailure()
+		}
 		completed := attempt.completeFailure(ctx, failedIndex, succeededCount, cause, override)
 		// Persist the hold (or release a canceled claim) before any potentially
 		// slow remote rollback. Otherwise the attempt lease can expire while
@@ -985,7 +1041,7 @@ func runPersonalEventStatus(c *cobra.Command, opts personalStatusOptions) error
 		return err
 	}
 	configDir := defaultConfigDir()
-	identity, err := personalResolveEventIdentity(ctx, configDir, opts.StreamSourceID)
+	identity, err := resolvePersonalEventIdentityForToken(ctx, configDir, opts.StreamSourceID, opts.ExplicitToken, opts.ClientIDOverride)
 	if err != nil {
 		return fmt.Errorf("event status --as user: %w", err)
 	}
@@ -1017,7 +1073,7 @@ func runPersonalEventStatus(c *cobra.Command, opts personalStatusOptions) error
 	if status == "" || status == "all" {
 		status = ""
 	}
-	subs, err := personalListSubscriptions(newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity), ctx, personal.ListOptions{
+	subs, err := personalListSubscriptions(newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity, opts.ExplicitToken), ctx, personal.ListOptions{
 		Status:      status,
 		EventKey:    opts.EventKey,
 		SubscribeID: opts.SubscribeID,
@@ -1038,6 +1094,15 @@ func runPersonalEventStatus(c *cobra.Command, opts personalStatusOptions) error
 	return nil
 }
 
+func personalRuntimeTokenControlRejection(err error) bool {
+	var apiErr *personal.APIError
+	if !errors.As(err, &apiErr) || apiErr == nil {
+		return false
+	}
+	return apiErr.HTTPStatus == http.StatusUnauthorized ||
+		strings.EqualFold(strings.TrimSpace(apiErr.Code), "RUNTIME_TOKEN_REJECTED")
+}
+
 func ensurePublicPersonalEvent(eventKey string) error {
 	eventKey = strings.TrimSpace(eventKey)
 	if eventKey == "" {
@@ -1120,7 +1185,7 @@ func runPersonalEventStop(c *cobra.Command, opts personalStopOptions) error {
 	}
 
 	configDir := defaultConfigDir()
-	identity, err := personalResolveEventIdentity(ctx, configDir, opts.StreamSourceID)
+	identity, err := resolvePersonalEventIdentityForToken(ctx, configDir, opts.StreamSourceID, opts.ExplicitToken, opts.ClientIDOverride)
 	if err != nil {
 		return fmt.Errorf("event stop --as user: %w", err)
 	}
@@ -1132,7 +1197,7 @@ func runPersonalEventStop(c *cobra.Command, opts personalStopOptions) error {
 	if err != nil {
 		return fmt.Errorf("event stop --as user: %w", err)
 	}
-	client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity)
+	client := newPersonalEventControlClient(configDir, personalEventControlBaseURL(opts.ControlBaseURL, configDir), identity, opts.ExplicitToken)
 	for _, id := range subscribeIDs {
 		if err := personalDeleteSubscription(client, ctx, id); err != nil {
 			return fmt.Errorf("event stop --as user: cancel subscription %s: %w", id, err)
@@ -1248,6 +1313,138 @@ func printPersonalStopResult(w io.Writer, subscribeIDs []string, single bool, bu
 	fmt.Fprintf(w, "cancelled %d personal subscription(s); %s\n", len(subscribeIDs), busState)
 }
 
+func resolvePersonalEventIdentityForToken(ctx context.Context, configDir, sourceIDOverride, explicitToken string, clientIDOverrides ...string) (personal.Identity, error) {
+	explicitToken = strings.TrimSpace(explicitToken)
+	if explicitToken == "" {
+		return personalResolveEventIdentity(ctx, configDir, sourceIDOverride)
+	}
+	clientIDOverride := ""
+	if len(clientIDOverrides) > 0 {
+		clientIDOverride = strings.TrimSpace(clientIDOverrides[0])
+	}
+	return resolvePersonalEventIdentityWithToken(ctx, configDir, sourceIDOverride, explicitToken, clientIDOverride)
+}
+
+// resolvePersonalEventIdentityWithToken resolves only non-sensitive identity
+// metadata around a caller-supplied bearer token. It intentionally does not
+// call LoadTokenData or any refresh-capable token resolver: an explicit root
+// --token must never be replaced with, persisted into, or used to refresh a
+// local OAuth profile.
+func resolvePersonalEventIdentityWithToken(ctx context.Context, configDir, sourceIDOverride, explicitToken string, clientIDOverrides ...string) (personal.Identity, error) {
+	explicitToken = strings.TrimSpace(explicitToken)
+	if explicitToken == "" {
+		return resolvePersonalEventIdentity(ctx, configDir, sourceIDOverride)
+	}
+	if strings.Contains(strings.TrimSpace(authpkg.RuntimeProfile()), ",") {
+		return personal.Identity{}, fmt.Errorf("personal events require exactly one --profile")
+	}
+
+	corpID := resolveRuntimeDefault(ctx, "$corpId")
+	userID := resolveRuntimeDefault(ctx, "$currentUserId")
+	clientID := ""
+	if len(clientIDOverrides) > 0 {
+		clientID = strings.TrimSpace(clientIDOverrides[0])
+	}
+	if clientID == "" {
+		// An edition hook or explicit environment value is runtime identity,
+		// not persisted app state. Resolve it before profiles.json so a complete
+		// host context never depends on local OAuth metadata health.
+		clientID = strings.TrimSpace(personalRuntimeEventClientID())
+	}
+	explicitProfile := strings.TrimSpace(authpkg.RuntimeProfile()) != ""
+	if explicitProfile || corpID == "" || userID == "" || clientID == "" {
+		profile, err := personalEventProfileMetadata(configDir)
+		if err != nil {
+			// A user-selected --profile remains a strict contract. Without an
+			// explicit selector, profiles.json is optional metadata for a
+			// host-managed bearer: malformed or stale persisted state must not
+			// override complete runtime defaults or prevent the later global
+			// client-id fallback.
+			if explicitProfile {
+				return personal.Identity{}, fmt.Errorf("load OAuth identity metadata: %w", err)
+			}
+			profile = nil
+		}
+		if profile != nil {
+			if corpID == "" {
+				corpID = strings.TrimSpace(profile.CorpID)
+			}
+			if userID == "" {
+				userID = strings.TrimSpace(profile.UserID)
+			}
+			if clientID == "" {
+				clientID = strings.TrimSpace(profile.ClientID)
+			}
+		}
+	}
+	if clientID == "" {
+		// Persisted/global app credentials are only a fallback after the
+		// selected profile, so an old app config cannot override profile.ClientID.
+		clientID = strings.TrimSpace(personalClientID())
+	}
+	if clientID == "" {
+		if id, _, _, _, resolveErr := personalResolveAppCredentialsStrict(configDir); resolveErr == nil {
+			clientID = strings.TrimSpace(id)
+		}
+	}
+	if clientID == "" {
+		return personal.Identity{}, fmt.Errorf("cannot resolve OAuth client_id for personal events")
+	}
+
+	sourceID := strings.TrimSpace(sourceIDOverride)
+	if sourceID == "" {
+		sourceID = personalEventStreamSourceID("")
+	}
+	localSubject := ""
+	if corpID == "" || userID == "" {
+		localSubject = personalTokenSubject("access", explicitToken)
+	}
+	return personal.Identity{
+		LocalSubject: localSubject,
+		CorpID:       corpID,
+		UserID:       userID,
+		ClientID:     clientID,
+		SourceID:     sourceID,
+	}, nil
+}
+
+func personalEventProfileMetadata(configDir string) (*authpkg.Profile, error) {
+	cfg, err := personalLoadProfiles(configDir)
+	if err != nil {
+		return nil, err
+	}
+	selector := strings.TrimSpace(authpkg.RuntimeProfile())
+	explicitSelector := selector != ""
+	if strings.Contains(selector, ",") {
+		return nil, fmt.Errorf("personal events require exactly one --profile")
+	}
+	if cfg == nil || len(cfg.Profiles) == 0 {
+		if explicitSelector {
+			return nil, fmt.Errorf("profile %q not found", selector)
+		}
+		return nil, nil
+	}
+	if selector == "" {
+		selector = strings.TrimSpace(cfg.CurrentProfile)
+	}
+	if selector == "" {
+		return nil, nil
+	}
+	profile, err := selectPersonalEventProfileMetadata(cfg, selector, make(map[string]struct{}))
+	if err != nil && !explicitSelector {
+		// A stale persisted CurrentProfile must not make a host-provided bearer
+		// unusable. Runtime defaults and the one-way local subject are sufficient
+		// to isolate the event bus without consulting local OAuth credentials.
+		return nil, nil
+	}
+	return profile, err
+}
+
+func selectPersonalEventProfileMetadata(cfg *authpkg.ProfilesConfig, selector string, visited map[string]struct{}) (*authpkg.Profile, error) {
+	_ = visited // retained for the focused compatibility seam used by app tests.
+	return authpkg.ResolveProfileMetadata(cfg, strings.TrimSpace(selector))
+}
+
 func resolvePersonalEventIdentity(ctx context.Context, configDir string, sourceIDOverride string) (personal.Identity, error) {
 	accessToken, err := personalResolveAuxiliaryAccessToken(ctx, configDir, "")
 	if err != nil {
@@ -1302,7 +1499,11 @@ func resolvePersonalEventIdentity(ctx context.Context, configDir string, sourceI
 	}, nil
 }
 
-func newPersonalEventControlClient(configDir, baseURL string, identity personal.Identity) *personal.Client {
+func newPersonalEventControlClient(configDir, baseURL string, identity personal.Identity, explicitTokens ...string) *personal.Client {
+	explicitToken := ""
+	if len(explicitTokens) > 0 {
+		explicitToken = strings.TrimSpace(explicitTokens[0])
+	}
 	identity.AccessToken = ""
 	client := personal.NewClient(baseURL, identity)
 	version := strings.TrimSpace(RawVersion())
@@ -1311,12 +1512,146 @@ func newPersonalEventControlClient(configDir, baseURL string, identity personal.
 	}
 	client.ClientVersion = version
 	client.UserAgent = "dws-cli/" + version
-	client.AccessTokenProvider = func(ctx context.Context) (string, error) {
-		return personalResolveAuxiliaryAccessToken(ctx, configDir, "")
+	if explicitToken != "" {
+		client.AccessTokenProvider = func(context.Context) (string, error) { return explicitToken, nil }
+		client.HTTPClient.Transport = runtimeTokenControlTransport{base: http.DefaultTransport, token: explicitToken}
+		client.HTTPClient.CheckRedirect = runtimeTokenRedirectPolicy
+	} else {
+		client.AccessTokenProvider = func(ctx context.Context) (string, error) {
+			return personalResolveAuxiliaryAccessToken(ctx, configDir, "")
+		}
 	}
 	return client
 }
 
+// runtimeTokenRedirectPolicy prevents Go's redirect machinery from copying
+// DWS's custom x-user-access-token header to another authority. Returning
+// ErrUseLastResponse keeps the 3xx response available to the caller without a
+// url.Error that could echo an attacker-controlled Location value.
+func runtimeTokenRedirectPolicy(req *http.Request, via []*http.Request) error {
+	if len(via) == 0 || req == nil || req.URL == nil || via[0] == nil || via[0].URL == nil {
+		return http.ErrUseLastResponse
+	}
+	origin := via[0].URL
+	if !strings.EqualFold(strings.TrimSpace(req.URL.Host), strings.TrimSpace(origin.Host)) {
+		return http.ErrUseLastResponse
+	}
+	if strings.EqualFold(origin.Scheme, "https") && !strings.EqualFold(req.URL.Scheme, "https") {
+		return http.ErrUseLastResponse
+	}
+	return nil
+}
+
+const runtimeTokenControlErrorBody = `{"code":"RUNTIME_TOKEN_REJECTED","message":"event runtime token was rejected; retry with a fresh host credential"}`
+
+// runtimeTokenControlTransport scrubs an explicit bearer from every response
+// body and diagnostic header before the control client decodes or logs it. A
+// 401 is replaced with a fixed rejection envelope so untrusted response text
+// can never escape through stderr or debug logs.
+type runtimeTokenControlTransport struct {
+	base  http.RoundTripper
+	token string
+}
+
+func (t runtimeTokenControlTransport) RoundTrip(req *http.Request) (*http.Response, error) {
+	base := t.base
+	if base == nil {
+		base = http.DefaultTransport
+	}
+	resp, err := base.RoundTrip(req)
+	if err != nil {
+		if token := strings.TrimSpace(t.token); token != "" && strings.Contains(err.Error(), token) {
+			return nil, errors.New("personal event: runtime-token control request failed")
+		}
+		return nil, err
+	}
+	if resp == nil {
+		return resp, err
+	}
+	token := strings.TrimSpace(t.token)
+	for key, values := range resp.Header {
+		for i := range values {
+			if token != "" {
+				values[i] = strings.ReplaceAll(values[i], token, "<redacted-runtime-token>")
+			}
+		}
+		resp.Header[key] = values
+	}
+	var responseBody []byte
+	if resp.Body != nil {
+		responseBody, err = io.ReadAll(io.LimitReader(resp.Body, config.MaxResponseBodySize))
+		_ = resp.Body.Close()
+		if err != nil {
+			return nil, errors.New("personal event: read runtime-token control response")
+		}
+	}
+	if resp.StatusCode == http.StatusUnauthorized {
+		responseBody = []byte(runtimeTokenControlErrorBody)
+	} else if token != "" {
+		responseBody = redactRuntimeTokenResponseBody(responseBody, token)
+	}
+	resp.Body = io.NopCloser(bytes.NewReader(responseBody))
+	resp.ContentLength = int64(len(responseBody))
+	if resp.Header == nil {
+		resp.Header = make(http.Header)
+	}
+	resp.Header.Set("Content-Type", "application/json")
+	resp.Header.Set("Content-Length", fmt.Sprintf("%d", len(responseBody)))
+	return resp, nil
+}
+
+func redactRuntimeTokenResponseBody(data []byte, token string) []byte {
+	token = strings.TrimSpace(token)
+	if len(data) == 0 || token == "" {
+		return data
+	}
+	decoder := json.NewDecoder(bytes.NewReader(data))
+	decoder.UseNumber()
+	var decoded any
+	if err := decoder.Decode(&decoded); err == nil {
+		var trailing any
+		if trailingErr := decoder.Decode(&trailing); errors.Is(trailingErr, io.EOF) {
+			if redacted, changed := redactRuntimeTokenJSONValue(decoded, token); changed {
+				if encoded, marshalErr := json.Marshal(redacted); marshalErr == nil {
+					return encoded
+				}
+			}
+		}
+	}
+	return bytes.ReplaceAll(data, []byte(token), []byte("<redacted-runtime-token>"))
+}
+
+func redactRuntimeTokenJSONValue(value any, token string) (any, bool) {
+	switch typed := value.(type) {
+	case string:
+		redacted := strings.ReplaceAll(typed, token, "<redacted-runtime-token>")
+		return redacted, redacted != typed
+	case []any:
+		changed := false
+		for i := range typed {
+			var itemChanged bool
+			typed[i], itemChanged = redactRuntimeTokenJSONValue(typed[i], token)
+			changed = changed || itemChanged
+		}
+		return typed, changed
+	case map[string]any:
+		changed := false
+		redactedMap := make(map[string]any, len(typed))
+		for key, item := range typed {
+			redactedKey := strings.ReplaceAll(key, token, "<redacted-runtime-token>")
+			redacted, itemChanged := redactRuntimeTokenJSONValue(item, token)
+			redactedMap[redactedKey] = redacted
+			changed = changed || itemChanged || redactedKey != key
+		}
+		if !changed {
+			return typed, false
+		}
+		return redactedMap, true
+	default:
+		return value, false
+	}
+}
+
 func personalTokenSubject(kind, token string) string {
 	token = strings.TrimSpace(token)
 	if token == "" {
@@ -1326,6 +1661,15 @@ func personalTokenSubject(kind, token string) string {
 	return strings.TrimSpace(kind) + ":" + hex.EncodeToString(sum[:])
 }
 
+func validPersonalIdentityHash(value string) bool {
+	value = strings.TrimSpace(value)
+	if len(value) != 16 {
+		return false
+	}
+	_, err := hex.DecodeString(value)
+	return err == nil
+}
+
 func resolveRuntimeDefault(ctx context.Context, key string) string {
 	if fnMap := edition.Get().RuntimeDefaults; fnMap != nil {
 		if fn := fnMap()[key]; fn != nil {
@@ -1363,20 +1707,42 @@ func newPersonalStreamSource(ctx context.Context, opts personalStreamSourceOptio
 		}
 		clientSecret = secret
 	}
+	credentialBroker := opts.CredentialBroker
+	if credentialBroker == nil {
+		credentialBroker = newPersonalCredentialBroker(opts.ConfigDir, false, false)
+	}
+	httpClient := &http.Client{Timeout: 30 * time.Second}
+	if opts.RuntimeTokenMode {
+		httpClient.CheckRedirect = runtimeTokenRedirectPolicy
+	}
 	_ = ctx
 	return source.NewPersonal(source.PersonalConfig{
 		AccessTokenProvider: func(ctx context.Context) (string, error) {
-			return personalResolveAuxiliaryAccessToken(ctx, opts.ConfigDir, "")
+			return credentialBroker.Resolve(ctx)
 		},
 		ForceRefreshToken: func(ctx context.Context, rejectedToken string) (string, error) {
-			return personalForceRefreshRejectedToken(ctx, opts.ConfigDir, rejectedToken)
+			return credentialBroker.RefreshRejected(ctx, rejectedToken)
+		},
+		ClassifyRetryReject: credentialBroker.ClassifyRejectedAfterRetry,
+		ClientID:            clientID,
+		ClientSecret:        clientSecret,
+		SourceID:            opts.Identity.SourceID,
+		TicketURL:           ticketURL,
+		TicketMode:          mode,
+		HTTPClient:          httpClient,
+	})
+}
+
+func newPersonalCredentialBroker(configDir string, requireSeed, requireActivation bool) *runtimecred.Broker {
+	return runtimecred.New(runtimecred.Config{
+		RequireSeed:       requireSeed,
+		RequireActivation: requireActivation,
+		LocalResolve: func(ctx context.Context) (string, error) {
+			return personalResolveAuxiliaryAccessToken(ctx, configDir, "")
+		},
+		LocalRefresh: func(ctx context.Context, rejectedToken string) (string, error) {
+			return personalForceRefreshRejectedToken(ctx, configDir, rejectedToken)
 		},
-		ClientID:     clientID,
-		ClientSecret: clientSecret,
-		SourceID:     opts.Identity.SourceID,
-		TicketURL:    ticketURL,
-		TicketMode:   mode,
-		HTTPClient:   &http.Client{Timeout: 30 * time.Second},
 	})
 }
 
@@ -1441,6 +1807,25 @@ func personalBusSpawnArgs(identity personal.Identity, ticketMode, ticketURL stri
 	return args
 }
 
+func personalBusSpawnArgsForToken(identity personal.Identity, identityHash, ticketMode, ticketURL, profileSelector, explicitToken string) []string {
+	if strings.TrimSpace(explicitToken) == "" {
+		return personalBusSpawnArgs(identity, ticketMode, ticketURL, profileSelector)
+	}
+	args := []string{
+		"--source-kind", string(dwsevent.SourceKindPersonalStream),
+		"--runtime-token-mode",
+		"--identity-hash", strings.TrimSpace(identityHash),
+		"--stream-source-id", strings.TrimSpace(identity.SourceID),
+	}
+	if strings.TrimSpace(ticketMode) != "" {
+		args = append(args, "--stream-ticket-mode", strings.TrimSpace(ticketMode))
+	}
+	if strings.TrimSpace(ticketURL) != "" {
+		args = append(args, "--stream-ticket-url", strings.TrimSpace(ticketURL))
+	}
+	return args
+}
+
 func personalEventTypes(eventKey string, explicit []string) []string {
 	if len(explicit) > 0 {
 		return explicit

internal/app/event_runtime_token_detached_e2e_test.go

@@ -0,0 +1,396 @@
+// 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.
+
+package app
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io/fs"
+	"log/slog"
+	"os"
+	"os/signal"
+	"path/filepath"
+	"runtime"
+	"strings"
+	"syscall"
+	"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/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/runtimecred"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
+)
+
+const (
+	runtimeTokenDetachedChildEnv     = "DWS_EVENT_RUNTIME_TOKEN_E2E_CHILD"
+	runtimeTokenDetachedWorkDirEnv   = "DWS_EVENT_RUNTIME_TOKEN_E2E_WORKDIR"
+	runtimeTokenDetachedEndpointEnv  = "DWS_EVENT_RUNTIME_TOKEN_E2E_ENDPOINT"
+	runtimeTokenDetachedEvidenceEnv  = "DWS_EVENT_RUNTIME_TOKEN_E2E_EVIDENCE"
+	runtimeTokenDetachedCanaryA      = "dws-runtime-e2e-A-9f34c8d10b7e"
+	runtimeTokenDetachedCanaryB      = "dws-runtime-e2e-B-2ad761e5c490"
+	runtimeTokenDetachedClientID     = "runtime-e2e-client"
+	runtimeTokenDetachedIdentityHash = "90abcdef12345678"
+	runtimeTokenDetachedSourceID     = "runtime-e2e-source"
+)
+
+// runRuntimeTokenDetachedE2EChild is called at the very start of TestMain.
+// busctl.Spawn executes this test binary with production-style `event _bus`
+// arguments; the env marker lets the child run a real bus daemon before the Go
+// test runner attempts to parse those CLI arguments.
+func runRuntimeTokenDetachedE2EChild() (int, bool) {
+	if os.Getenv(runtimeTokenDetachedChildEnv) != "1" {
+		return 0, false
+	}
+	workDir := strings.TrimSpace(os.Getenv(runtimeTokenDetachedWorkDirEnv))
+	endpoint := strings.TrimSpace(os.Getenv(runtimeTokenDetachedEndpointEnv))
+	evidence := strings.TrimSpace(os.Getenv(runtimeTokenDetachedEvidenceEnv))
+	if workDir == "" || endpoint == "" || evidence == "" {
+		return 91, true
+	}
+
+	argvClean := !runtimeTokenDetachedContainsCanary(strings.Join(os.Args, "\x00"))
+	envClean := !runtimeTokenDetachedContainsCanary(strings.Join(os.Environ(), "\x00"))
+	if err := appendRuntimeTokenDetachedEvidence(evidence,
+		fmt.Sprintf("child_start argv_clean=%t env_clean=%t", argvClean, envClean)); err != nil {
+		return 92, true
+	}
+	if !argvClean || !envClean {
+		return 93, true
+	}
+
+	logFile, err := os.OpenFile(filepath.Join(workDir, "bus.log"), os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
+	if err != nil {
+		return 94, true
+	}
+	defer logFile.Close()
+
+	broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
+	ctx, cancel := signal.NotifyContext(context.Background(), os.Interrupt, syscall.SIGTERM)
+	defer cancel()
+	err = bus.Run(ctx, bus.Config{
+		WorkDir:          workDir,
+		IPCEndpoint:      endpoint,
+		ClientID:         runtimeTokenDetachedClientID,
+		SourceKind:       dwsevent.SourceKindPersonalStream,
+		IdentityHash:     runtimeTokenDetachedIdentityHash,
+		SourceID:         runtimeTokenDetachedSourceID,
+		Edition:          "open",
+		SDKVersion:       "runtime-e2e",
+		Source:           &runtimeTokenDetachedSource{broker: broker, evidence: evidence},
+		CredentialBroker: broker,
+		ReadyPipe:        busctl.ReadyFDFromEnv(),
+		Logger:           slog.New(slog.NewTextHandler(logFile, nil)),
+	})
+	if err != nil && !errors.Is(err, context.Canceled) {
+		_ = appendRuntimeTokenDetachedEvidence(evidence, "bus_exit clean=false")
+		return 95, true
+	}
+	_ = appendRuntimeTokenDetachedEvidence(evidence, "bus_exit clean=true")
+	return 0, true
+}
+
+type runtimeTokenDetachedSource struct {
+	broker   *runtimecred.Broker
+	evidence string
+}
+
+// Start models the credential-sensitive part of a reconnecting Stream source
+// without network access. It resolves A for the first connection, waits until a
+// second consumer rotates the broker to B, then exercises the exact 401 path:
+// RefreshRejected(A) must return B and must not fall back to local OAuth.
+func (s *runtimeTokenDetachedSource) Start(ctx context.Context, _ dwsevent.EmitFn) error {
+	first, err := s.broker.Resolve(ctx)
+	if err != nil {
+		return errors.New("runtime e2e: initial credential unavailable")
+	}
+	if first != runtimeTokenDetachedCanaryA || s.broker.Generation() != 1 {
+		return errors.New("runtime e2e: initial credential mismatch")
+	}
+	if err := appendRuntimeTokenDetachedEvidence(s.evidence, "resolved_a=true generation=1"); err != nil {
+		return errors.New("runtime e2e: record initial connection")
+	}
+
+	ticker := time.NewTicker(5 * time.Millisecond)
+	defer ticker.Stop()
+	for s.broker.Generation() < 2 {
+		select {
+		case <-ctx.Done():
+			return ctx.Err()
+		case <-ticker.C:
+		}
+	}
+	rotated, err := s.broker.RefreshRejected(ctx, first)
+	if err != nil || rotated != runtimeTokenDetachedCanaryB {
+		return errors.New("runtime e2e: rotated credential unavailable")
+	}
+	if err := appendRuntimeTokenDetachedEvidence(s.evidence, "rejected_a=true resolved_b=true reconnect=true generation=2"); err != nil {
+		return errors.New("runtime e2e: record reconnect")
+	}
+	<-ctx.Done()
+	return ctx.Err()
+}
+
+func appendRuntimeTokenDetachedEvidence(path, line string) error {
+	f, err := os.OpenFile(path, os.O_CREATE|os.O_APPEND|os.O_WRONLY, 0o600)
+	if err != nil {
+		return err
+	}
+	defer f.Close()
+	_, err = fmt.Fprintln(f, line)
+	return err
+}
+
+func runtimeTokenDetachedContainsCanary(value string) bool {
+	return strings.Contains(value, runtimeTokenDetachedCanaryA) ||
+		strings.Contains(value, runtimeTokenDetachedCanaryB)
+}
+
+func TestCrossPlatformCoverageUnixDetachedRuntimeTokenLifecycleAndCanaryLeakScan(t *testing.T) {
+	if runtime.GOOS == "windows" {
+		t.Skip("real detached-process lifecycle is Unix-only; Windows named-pipe code is cross-compiled separately")
+	}
+
+	root, err := os.MkdirTemp("/tmp", "dws-runtime-token-e2e-")
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = os.RemoveAll(root) })
+	workDir := filepath.Join(root, "events", "open", string(dwsevent.SourceKindPersonalStream), runtimeTokenDetachedIdentityHash)
+	if err := os.MkdirAll(workDir, 0o700); err != nil {
+		t.Fatal(err)
+	}
+	endpoint := dwsevent.IPCEndpoint(workDir, "open", dwsevent.SourceKindPersonalStream, runtimeTokenDetachedIdentityHash)
+	evidencePath := filepath.Join(workDir, "runtime-e2e.evidence")
+
+	identity := personal.Identity{
+		ClientID: runtimeTokenDetachedClientID,
+		SourceID: runtimeTokenDetachedSourceID,
+		CorpID:   "runtime-e2e-corp",
+		UserID:   "runtime-e2e-user",
+	}
+	spawnArgs := personalBusSpawnArgsForToken(identity, runtimeTokenDetachedIdentityHash, "", "", "corp:user", runtimeTokenDetachedCanaryA)
+	assertRuntimeTokenDetachedClean(t, "spawn argv", []byte(strings.Join(spawnArgs, "\x00")))
+
+	childEnv := append([]string{}, os.Environ()...)
+	childEnv = append(childEnv,
+		runtimeTokenDetachedChildEnv+"=1",
+		runtimeTokenDetachedWorkDirEnv+"="+workDir,
+		runtimeTokenDetachedEndpointEnv+"="+endpoint,
+		runtimeTokenDetachedEvidenceEnv+"="+evidencePath,
+	)
+	assertRuntimeTokenDetachedClean(t, "spawn environment", []byte(strings.Join(childEnv, "\x00")))
+
+	pid, err := busctl.Spawn(busctl.SpawnConfig{
+		ExecPath:  os.Args[0],
+		ClientID:  runtimeTokenDetachedClientID,
+		ExtraArgs: spawnArgs,
+		Env:       childEnv,
+	})
+	if err != nil {
+		failRuntimeTokenDetachedError(t, "spawn detached runtime bus", err)
+	}
+	stopped := false
+	t.Cleanup(func() {
+		if !stopped {
+			_ = busctl.Stop(busctl.StopConfig{WorkDir: workDir, Timeout: 2 * time.Second})
+			if proc, findErr := os.FindProcess(pid); findErr == nil {
+				_ = proc.Kill()
+			}
+		}
+	})
+
+	waitRuntimeTokenDetachedFile(t, evidencePath, "child_start argv_clean=true env_clean=true", 3*time.Second)
+
+	var stdoutA, stderrA bytes.Buffer
+	err = consume.Run(context.Background(), runtimeTokenDetachedConsumeConfig(
+		workDir, endpoint, "sub-runtime-a", runtimeTokenDetachedCanaryA, 500*time.Millisecond, &stdoutA, &stderrA,
+	))
+	if err != nil {
+		failRuntimeTokenDetachedError(t, "consume token A", err)
+	}
+	waitRuntimeTokenDetachedFile(t, evidencePath, "resolved_a=true generation=1", 3*time.Second)
+
+	if err := personal.UpsertRunState(workDir, personal.RunState{
+		SubscribeID:  "sub-runtime-b",
+		EventKey:     personal.EventMention,
+		ClientID:     runtimeTokenDetachedClientID,
+		SourceID:     runtimeTokenDetachedSourceID,
+		IdentityHash: runtimeTokenDetachedIdentityHash,
+	}); err != nil {
+		failRuntimeTokenDetachedError(t, "persist non-sensitive run state", err)
+	}
+
+	var stdoutB, stderrB bytes.Buffer
+	consumeDone := make(chan error, 1)
+	go func() {
+		consumeDone <- consume.Run(context.Background(), runtimeTokenDetachedConsumeConfig(
+			workDir, endpoint, "sub-runtime-b", runtimeTokenDetachedCanaryB, 5*time.Second, &stdoutB, &stderrB,
+		))
+	}()
+
+	status := waitRuntimeTokenDetachedStatus(t, endpoint, "sub-runtime-b", 3*time.Second)
+	if status.Bus.PID != pid || status.Bus.IdentityHash != runtimeTokenDetachedIdentityHash {
+		t.Fatalf("status bus identity = %#v, want pid=%d identity=%s", status.Bus, pid, runtimeTokenDetachedIdentityHash)
+	}
+	waitRuntimeTokenDetachedFile(t, evidencePath, "rejected_a=true resolved_b=true reconnect=true generation=2", 3*time.Second)
+
+	stopResp, err := busctl.StopConsumers(endpoint, []string{"sub-runtime-b"})
+	if err != nil {
+		failRuntimeTokenDetachedError(t, "targeted consumer stop", err)
+	}
+	if len(stopResp.Stopped) != 1 || stopResp.Stopped[0] != "sub-runtime-b" {
+		t.Fatalf("targeted stop response = %#v", stopResp)
+	}
+	select {
+	case err := <-consumeDone:
+		if err != nil {
+			failRuntimeTokenDetachedError(t, "consume token B after targeted stop", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("token B consumer did not exit after targeted stop")
+	}
+	status = waitRuntimeTokenDetachedStatus(t, endpoint, "", 3*time.Second)
+	if len(status.Consumers) != 0 {
+		t.Fatalf("status consumers after stop = %#v", status.Consumers)
+	}
+
+	if err := busctl.Stop(busctl.StopConfig{WorkDir: workDir, Timeout: 4 * time.Second}); err != nil {
+		failRuntimeTokenDetachedError(t, "stop detached bus", err)
+	}
+	stopped = true
+	waitRuntimeTokenDetachedFile(t, evidencePath, "bus_exit clean=true", 3*time.Second)
+
+	statusJSON, err := json.Marshal(status)
+	if err != nil {
+		t.Fatal(err)
+	}
+	stopJSON, err := json.Marshal(stopResp)
+	if err != nil {
+		t.Fatal(err)
+	}
+	for name, artifact := range map[string][]byte{
+		"consume A stdout": stdoutA.Bytes(),
+		"consume A stderr": stderrA.Bytes(),
+		"consume B stdout": stdoutB.Bytes(),
+		"consume B stderr": stderrB.Bytes(),
+		"status response":  statusJSON,
+		"stop response":    stopJSON,
+	} {
+		assertRuntimeTokenDetachedClean(t, name, artifact)
+	}
+	assertRuntimeTokenDetachedTreeClean(t, root)
+
+	for _, required := range []string{
+		filepath.Join(workDir, bus.MetaFileName),
+		filepath.Join(workDir, "bus.log"),
+		filepath.Join(workDir, personal.StateFileName),
+		evidencePath,
+	} {
+		if info, statErr := os.Stat(required); statErr != nil || !info.Mode().IsRegular() {
+			t.Fatalf("expected runtime artifact %s: info=%v err=%v", required, info, statErr)
+		}
+	}
+}
+
+func runtimeTokenDetachedConsumeConfig(workDir, endpoint, subscribeID, token string, duration time.Duration, stdout, stderr *bytes.Buffer) consume.Config {
+	return consume.Config{
+		WorkDir:          workDir,
+		IPCEndpoint:      endpoint,
+		ClientID:         runtimeTokenDetachedClientID,
+		RuntimeToken:     token,
+		EventTypes:       []string{personal.EventMention},
+		EventKey:         personal.EventMention,
+		SubscribeID:      subscribeID,
+		ReadySubscribeID: subscribeID,
+		Duration:         duration,
+		Format:           consume.FormatNDJSON,
+		Stdout:           stdout,
+		Stderr:           stderr,
+	}
+}
+
+func waitRuntimeTokenDetachedFile(t *testing.T, path, want string, timeout time.Duration) string {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	for time.Now().Before(deadline) {
+		data, err := os.ReadFile(path)
+		if err == nil && strings.Contains(string(data), want) {
+			return string(data)
+		}
+		time.Sleep(10 * time.Millisecond)
+	}
+	data, err := os.ReadFile(path)
+	if runtimeTokenDetachedContainsCanary(string(data)) {
+		t.Fatalf("runtime credential leaked into child evidence while waiting for %q", want)
+	}
+	t.Fatalf("evidence %s missing %q: data=%q err=%v", path, want, data, err)
+	return ""
+}
+
+func waitRuntimeTokenDetachedStatus(t *testing.T, endpoint, subscribeID string, timeout time.Duration) *transport.StatusResp {
+	t.Helper()
+	deadline := time.Now().Add(timeout)
+	var lastErr error
+	for time.Now().Before(deadline) {
+		status, err := busctl.QueryStatus(endpoint)
+		if err == nil {
+			if subscribeID == "" && len(status.Consumers) == 0 {
+				return status
+			}
+			for _, consumer := range status.Consumers {
+				if consumer.SubscribeID == subscribeID {
+					return status
+				}
+			}
+		}
+		lastErr = err
+		time.Sleep(10 * time.Millisecond)
+	}
+	t.Fatalf("status never reached subscribe_id=%q: %v", subscribeID, lastErr)
+	return nil
+}
+
+func assertRuntimeTokenDetachedTreeClean(t *testing.T, root string) {
+	t.Helper()
+	err := filepath.WalkDir(root, func(path string, entry fs.DirEntry, walkErr error) error {
+		if walkErr != nil {
+			return walkErr
+		}
+		if entry.IsDir() || !entry.Type().IsRegular() {
+			return nil
+		}
+		data, err := os.ReadFile(path)
+		if err != nil {
+			return err
+		}
+		assertRuntimeTokenDetachedClean(t, path, data)
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("scan runtime artifacts: %v", err)
+	}
+}
+
+func assertRuntimeTokenDetachedClean(t *testing.T, name string, artifact []byte) {
+	t.Helper()
+	if runtimeTokenDetachedContainsCanary(string(artifact)) {
+		t.Fatalf("runtime credential leaked into %s", name)
+	}
+}
+
+func failRuntimeTokenDetachedError(t *testing.T, step string, err error) {
+	t.Helper()
+	if err != nil && runtimeTokenDetachedContainsCanary(err.Error()) {
+		t.Fatalf("%s failed and exposed a runtime credential", step)
+	}
+	t.Fatalf("%s: %v", step, err)
+}

internal/app/event_runtime_token_platform_gap_test.go

@@ -0,0 +1,335 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+
+package app
+
+import (
+	"context"
+	"errors"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+	"time"
+
+	authpkg "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/auth"
+	"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/runtimecred"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/edition"
+)
+
+func TestCrossPlatformCoverageRuntimeTokenBusRejectsIncompleteIdentity(t *testing.T) {
+	t.Setenv("DWS_CONFIG_DIR", t.TempDir())
+	for _, tc := range []struct {
+		name string
+		args []string
+		want string
+	}{
+		{
+			name: "invalid identity hash",
+			args: []string{"--source-kind", "personal_stream", "--runtime-token-mode", "--identity-hash", "not-a-hash", "--client-id", "client"},
+			want: "16-character hexadecimal identity hash",
+		},
+		{
+			name: "missing client id",
+			args: []string{"--source-kind", "personal_stream", "--runtime-token-mode", "--identity-hash", "0123456789abcdef"},
+			want: "--client-id is required",
+		},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			cmd := newEventBusCommand()
+			cmd.SetOut(io.Discard)
+			cmd.SetErr(io.Discard)
+			cmd.SetArgs(tc.args)
+			err := cmd.Execute()
+			if err == nil || !strings.Contains(err.Error(), tc.want) {
+				t.Fatalf("Execute() error = %v, want %q", err, tc.want)
+			}
+		})
+	}
+}
+
+type eventRuntimeTokenReleaseErrorStore struct {
+	err error
+}
+
+func (*eventRuntimeTokenReleaseErrorStore) Claim([]personal.AttemptSpec, time.Duration) (*personal.AttemptClaim, error) {
+	return nil, nil
+}
+
+func (*eventRuntimeTokenReleaseErrorStore) CompleteSuccess(*personal.AttemptClaim) error {
+	return nil
+}
+
+func (*eventRuntimeTokenReleaseErrorStore) CompleteFailure(*personal.AttemptClaim, []string, personal.AttemptFailure) (personal.AttemptHold, error) {
+	return personal.AttemptHold{}, nil
+}
+
+func (s *eventRuntimeTokenReleaseErrorStore) Release(*personal.AttemptClaim) error {
+	return s.err
+}
+
+func TestCrossPlatformCoverageRuntimeTokenAttemptReleaseGuardEdges(t *testing.T) {
+	var nilReservation *personalSubscriptionAttemptReservation
+	if err := nilReservation.releaseRuntimeTokenFailure(); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
+		t.Fatalf("nil reservation error = %v", err)
+	}
+
+	incomplete := &personalSubscriptionAttemptReservation{}
+	if err := incomplete.releaseRuntimeTokenFailure(); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) ||
+		!strings.Contains(err.Error(), "reservation is incomplete") {
+		t.Fatalf("incomplete reservation error = %v", err)
+	}
+
+	wantErr := errors.New("release failed")
+	reservation := &personalSubscriptionAttemptReservation{
+		store: &eventRuntimeTokenReleaseErrorStore{err: wantErr},
+		claim: &personal.AttemptClaim{AttemptID: "attempt"},
+	}
+	if err := reservation.releaseRuntimeTokenFailure(); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) ||
+		!errors.Is(err, wantErr) {
+		t.Fatalf("release failure error = %v", err)
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeTokenConsumeRejectionAndOversizeEdges(t *testing.T) {
+	t.Setenv("DWS_CONFIG_DIR", t.TempDir())
+	oldEdition := edition.Get()
+	oldProfile := authpkg.RuntimeProfile()
+	oldLoadProfiles := personalLoadProfiles
+	oldValidate := personalValidateConsumeConfig
+	oldConflict := personalValidateNoOutputConflict
+	oldAttemptStore := personalNewSubscriptionAttemptStore
+	oldEnsure := personalEnsureSubscription
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		authpkg.SetRuntimeProfile(oldProfile)
+		personalLoadProfiles = oldLoadProfiles
+		personalValidateConsumeConfig = oldValidate
+		personalValidateNoOutputConflict = oldConflict
+		personalNewSubscriptionAttemptStore = oldAttemptStore
+		personalEnsureSubscription = oldEnsure
+	})
+	edition.Override(&edition.Hooks{})
+	authpkg.SetRuntimeProfile("")
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) { return nil, nil }
+	personalValidateConsumeConfig = func(consume.Config) error { return nil }
+	personalValidateNoOutputConflict = func(consume.Config, string) error { return nil }
+
+	oversized := strings.Repeat("x", runtimecred.DefaultMaxTokenBytes+1)
+	err := runPersonalEventConsumeSingle(newPersonalCoverageCommand(), personalConsumeOptions{
+		EventKey:         personal.EventMention,
+		ExplicitToken:    oversized,
+		ClientIDOverride: "runtime-client",
+		Common:           commonConsumeOptions{Foreground: true},
+	})
+	if !errors.Is(err, runtimecred.ErrTokenTooLarge) {
+		t.Fatalf("oversized foreground token error = %v", err)
+	}
+
+	rejection := &personal.APIError{
+		Code:       "RUNTIME_TOKEN_REJECTED",
+		HTTPStatus: http.StatusUnauthorized,
+	}
+	if personalRuntimeTokenControlRejection(errors.New("ordinary failure")) {
+		t.Fatal("ordinary error classified as runtime-token rejection")
+	}
+
+	singleStore := &personalRecordingAttemptStore{}
+	personalNewSubscriptionAttemptStore = func(string) personalSubscriptionAttemptStore { return singleStore }
+	personalEnsureSubscription = func(context.Context, *personal.Client, personal.Identity, personalConsumeOptions) (*personal.Subscription, string, string, error) {
+		return nil, "", "", rejection
+	}
+	err = runPersonalEventConsumeSingle(newPersonalCoverageCommand(), personalConsumeOptions{
+		EventKey:         personal.EventMention,
+		ExplicitToken:    "runtime-token-single",
+		ClientIDOverride: "runtime-client",
+		ControlBaseURL:   "https://control.example.test",
+	})
+	if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) || singleStore.releaseCalls != 1 || singleStore.failureCalls != 0 {
+		t.Fatalf("single rejection = %v, release=%d failure=%d", err, singleStore.releaseCalls, singleStore.failureCalls)
+	}
+
+	manyStore := &personalRecordingAttemptStore{}
+	personalNewSubscriptionAttemptStore = func(string) personalSubscriptionAttemptStore { return manyStore }
+	err = runPersonalEventConsumeMany(newPersonalCoverageCommand(), personalConsumeOptions{
+		EventKeys:        []string{personal.EventMention, personal.EventAllSingleChat},
+		ExplicitToken:    "runtime-token-many",
+		ClientIDOverride: "runtime-client",
+		ControlBaseURL:   "https://control.example.test",
+	})
+	if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) || manyStore.releaseCalls != 1 || manyStore.failureCalls != 0 {
+		t.Fatalf("multi rejection = %v, release=%d failure=%d", err, manyStore.releaseCalls, manyStore.failureCalls)
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeTokenIdentityFallbackEdges(t *testing.T) {
+	configDir := t.TempDir()
+	oldEdition := edition.Get()
+	oldProfile := authpkg.RuntimeProfile()
+	oldResolveIdentity := personalResolveEventIdentity
+	oldResolveAuxiliary := personalResolveAuxiliaryAccessToken
+	oldLoadTokenData := personalLoadTokenData
+	oldLoadProfiles := personalLoadProfiles
+	oldRuntimeClientID := personalRuntimeEventClientID
+	oldClientID := personalClientID
+	oldResolveCredentials := personalResolveAppCredentialsStrict
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		authpkg.SetRuntimeProfile(oldProfile)
+		personalResolveEventIdentity = oldResolveIdentity
+		personalResolveAuxiliaryAccessToken = oldResolveAuxiliary
+		personalLoadTokenData = oldLoadTokenData
+		personalLoadProfiles = oldLoadProfiles
+		personalRuntimeEventClientID = oldRuntimeClientID
+		personalClientID = oldClientID
+		personalResolveAppCredentialsStrict = oldResolveCredentials
+	})
+
+	legacy := personal.Identity{ClientID: "legacy-client", SourceID: "legacy-source"}
+	personalResolveEventIdentity = func(context.Context, string, string) (personal.Identity, error) { return legacy, nil }
+	identity, err := resolvePersonalEventIdentityForToken(context.Background(), configDir, "", "  ")
+	if err != nil || identity.ClientID != legacy.ClientID {
+		t.Fatalf("wrapper empty-token fallback = %#v, %v", identity, err)
+	}
+	personalResolveAuxiliaryAccessToken = func(context.Context, string, string) (string, error) {
+		return "legacy-access", nil
+	}
+	personalLoadTokenData = func(string) (*authpkg.TokenData, error) {
+		return &authpkg.TokenData{
+			CorpID: "legacy-corp", UserID: "legacy-user", ClientID: "direct-client",
+		}, nil
+	}
+	identity, err = resolvePersonalEventIdentityWithToken(context.Background(), configDir, "", "  ")
+	if err != nil || identity.ClientID != "direct-client" {
+		t.Fatalf("direct empty-token fallback = %#v, %v", identity, err)
+	}
+
+	edition.Override(&edition.Hooks{})
+	personalRuntimeEventClientID = func() string { return "" }
+	personalClientID = func() string { return "" }
+	wantMetadataErr := errors.New("profiles unreadable")
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) { return nil, wantMetadataErr }
+	authpkg.SetRuntimeProfile("corp:user")
+	if _, err := resolvePersonalEventIdentityWithToken(context.Background(), configDir, "", "token", "runtime-client"); !errors.Is(err, wantMetadataErr) {
+		t.Fatalf("explicit profile metadata error = %v", err)
+	}
+
+	authpkg.SetRuntimeProfile("")
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) { return nil, nil }
+	personalResolveAppCredentialsStrict = func(string) (string, string, authpkg.CredentialSource, authpkg.CredentialSource, error) {
+		return "app-client", "", "", "", nil
+	}
+	identity, err = resolvePersonalEventIdentityWithToken(context.Background(), configDir, "", "runtime-token")
+	if err != nil || identity.ClientID != "app-client" || !strings.HasPrefix(identity.LocalSubject, "access:") {
+		t.Fatalf("app-credential fallback identity = %#v, %v", identity, err)
+	}
+
+	personalResolveAppCredentialsStrict = func(string) (string, string, authpkg.CredentialSource, authpkg.CredentialSource, error) {
+		return "", "", "", "", errors.New("missing app credentials")
+	}
+	if _, err := resolvePersonalEventIdentityWithToken(context.Background(), configDir, "", "runtime-token"); err == nil || !strings.Contains(err.Error(), "cannot resolve OAuth client_id") {
+		t.Fatalf("missing client ID error = %v", err)
+	}
+
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		return &authpkg.ProfilesConfig{
+			CurrentProfile: "stale",
+			Profiles:       []authpkg.Profile{{Name: "other", CorpID: "corp", UserID: "user"}},
+		}, nil
+	}
+	profile, err := personalEventProfileMetadata(configDir)
+	if err != nil || profile != nil {
+		t.Fatalf("stale implicit current profile = %#v, %v", profile, err)
+	}
+
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		return &authpkg.ProfilesConfig{Profiles: []authpkg.Profile{{Name: "other"}}}, nil
+	}
+	profile, err = personalEventProfileMetadata(configDir)
+	if err != nil || profile != nil {
+		t.Fatalf("empty implicit selector = %#v, %v", profile, err)
+	}
+
+	authpkg.SetRuntimeProfile("corp-a:user-a,corp-b:user-b")
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) { return nil, nil }
+	if _, err := personalEventProfileMetadata(configDir); err == nil || !strings.Contains(err.Error(), "exactly one --profile") {
+		t.Fatalf("multi-profile metadata error = %v", err)
+	}
+}
+
+type eventRuntimeTokenReadErrorBody struct{}
+
+func (eventRuntimeTokenReadErrorBody) Read([]byte) (int, error) {
+	return 0, errors.New("body read failed")
+}
+
+func (eventRuntimeTokenReadErrorBody) Close() error { return nil }
+
+func TestCrossPlatformCoverageRuntimeTokenControlTransportErrorEdges(t *testing.T) {
+	const token = "runtime-control-edge-canary"
+
+	unsupported, err := http.NewRequest(http.MethodGet, "unsupported://control.example.test/path", nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := (runtimeTokenControlTransport{}).RoundTrip(unsupported); err == nil {
+		t.Fatal("nil base unexpectedly accepted an unsupported protocol")
+	}
+
+	wantTransportErr := errors.New("ordinary transport failure")
+	ordinary := runtimeTokenControlTransport{base: eventRuntimeRoundTripFunc(func(*http.Request) (*http.Response, error) {
+		return nil, wantTransportErr
+	})}
+	if _, err := ordinary.RoundTrip(unsupported); !errors.Is(err, wantTransportErr) {
+		t.Fatalf("ordinary transport error = %v", err)
+	}
+
+	leaking := runtimeTokenControlTransport{
+		token: token,
+		base: eventRuntimeRoundTripFunc(func(*http.Request) (*http.Response, error) {
+			return nil, errors.New("reflected " + token)
+		}),
+	}
+	if _, err := leaking.RoundTrip(unsupported); err == nil || strings.Contains(err.Error(), token) ||
+		err.Error() != "personal event: runtime-token control request failed" {
+		t.Fatalf("redacted transport error = %v", err)
+	}
+
+	nilResponse := runtimeTokenControlTransport{base: eventRuntimeRoundTripFunc(func(*http.Request) (*http.Response, error) {
+		return nil, nil
+	})}
+	if resp, err := nilResponse.RoundTrip(unsupported); resp != nil || err != nil {
+		t.Fatalf("nil response = %#v, %v", resp, err)
+	}
+
+	readFailure := runtimeTokenControlTransport{base: eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
+		return &http.Response{
+			StatusCode: http.StatusInternalServerError,
+			Header:     make(http.Header),
+			Body:       eventRuntimeTokenReadErrorBody{},
+			Request:    req,
+		}, nil
+	})}
+	if _, err := readFailure.RoundTrip(unsupported); err == nil || !strings.Contains(err.Error(), "read runtime-token control response") {
+		t.Fatalf("body read error = %v", err)
+	}
+
+	nilHeader := runtimeTokenControlTransport{base: eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
+		return &http.Response{StatusCode: http.StatusOK, Request: req}, nil
+	})}
+	resp, err := nilHeader.RoundTrip(unsupported)
+	if err != nil || resp == nil || resp.Header == nil || resp.Header.Get("Content-Type") != "application/json" {
+		t.Fatalf("nil-header response = %#v, %v", resp, err)
+	}
+
+	if got := redactRuntimeTokenResponseBody(nil, token); len(got) != 0 {
+		t.Fatalf("empty response redaction = %q", got)
+	}
+	value, changed := redactRuntimeTokenJSONValue([]any{"plain", "prefix-" + token}, token)
+	items, ok := value.([]any)
+	if !ok || !changed || len(items) != 2 || strings.Contains(items[1].(string), token) {
+		t.Fatalf("array redaction = %#v changed=%t", value, changed)
+	}
+}

internal/app/event_runtime_token_wiring_test.go

@@ -0,0 +1,917 @@
+// 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.
+
+package app
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"io"
+	"log/slog"
+	"net/http"
+	"net/http/httptest"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	authpkg "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/auth"
+	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/consume"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/source"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/edition"
+	"github.com/spf13/cobra"
+)
+
+func TestCrossPlatformCoverageEventCommandsWireTrimmedRootRuntimeToken(t *testing.T) {
+	oldConsume := eventRunPersonalConsume
+	oldStatus := eventRunPersonalStatus
+	oldStop := eventRunPersonalStop
+	t.Cleanup(func() {
+		eventRunPersonalConsume = oldConsume
+		eventRunPersonalStatus = oldStatus
+		eventRunPersonalStop = oldStop
+	})
+
+	flags := &GlobalFlags{Token: "  runtime-canary  ", ClientID: " root-client "}
+	assertIdentity := func(token, clientID string) {
+		t.Helper()
+		if token != "runtime-canary" || clientID != "root-client" {
+			t.Fatalf("runtime identity = token %q client %q", token, clientID)
+		}
+	}
+
+	eventRunPersonalConsume = func(_ *cobra.Command, opts personalConsumeOptions) error {
+		assertIdentity(opts.ExplicitToken, opts.ClientIDOverride)
+		return nil
+	}
+	consumeCmd := newEventConsumeCommand(flags)
+	if err := consumeCmd.RunE(consumeCmd, []string{personal.EventMention}); err != nil {
+		t.Fatalf("consume RunE() error = %v", err)
+	}
+
+	eventRunPersonalStatus = func(_ *cobra.Command, opts personalStatusOptions) error {
+		assertIdentity(opts.ExplicitToken, opts.ClientIDOverride)
+		return nil
+	}
+	statusCmd := newEventStatusCommandWithFlags(flags)
+	if err := statusCmd.RunE(statusCmd, nil); err != nil {
+		t.Fatalf("status RunE() error = %v", err)
+	}
+
+	eventRunPersonalStop = func(_ *cobra.Command, opts personalStopOptions) error {
+		assertIdentity(opts.ExplicitToken, opts.ClientIDOverride)
+		return nil
+	}
+	stopCmd := newEventStopCommandWithFlags(flags)
+	stopRoot := &cobra.Command{Use: "dws"}
+	stopRoot.PersistentFlags().Bool("yes", true, "")
+	stopRoot.AddCommand(stopCmd)
+	if err := stopCmd.RunE(stopCmd, []string{"sub-runtime"}); err != nil {
+		t.Fatalf("stop RunE() error = %v", err)
+	}
+
+	listenCmd := newEventListenIMCommand(flags)
+	if err := listenCmd.RunE(listenCmd, nil); err != nil {
+		t.Fatalf("listen-im RunE() error = %v", err)
+	}
+}
+
+func TestCrossPlatformCoverageEventConsumeParsesRootRuntimeTokenBeforeAndAfterSubcommand(t *testing.T) {
+	oldConsume := eventRunPersonalConsume
+	t.Cleanup(func() { eventRunPersonalConsume = oldConsume })
+	for _, tc := range []struct {
+		name string
+		args []string
+	}{
+		{name: "before", args: []string{"--token", "runtime-before", "event", "consume", personal.EventMention}},
+		{name: "after", args: []string{"event", "consume", personal.EventMention, "--token", "runtime-after"}},
+	} {
+		t.Run(tc.name, func(t *testing.T) {
+			flags := &GlobalFlags{}
+			root := &cobra.Command{Use: "dws", SilenceErrors: true, SilenceUsage: true}
+			bindPersistentFlags(root, flags)
+			root.AddCommand(newEventCommand(flags))
+			var got string
+			eventRunPersonalConsume = func(_ *cobra.Command, opts personalConsumeOptions) error {
+				got = opts.ExplicitToken
+				return nil
+			}
+			root.SetArgs(tc.args)
+			if err := root.Execute(); err != nil {
+				t.Fatalf("Execute() error = %v", err)
+			}
+			want := "runtime-" + tc.name
+			if got != want {
+				t.Fatalf("ExplicitToken = %q, want %q", got, want)
+			}
+		})
+	}
+}
+
+func TestCrossPlatformCoverageResolvePersonalEventIdentityWithTokenUsesMetadataOnly(t *testing.T) {
+	oldEdition := edition.Get()
+	oldLoadProfiles := personalLoadProfiles
+	oldLoadToken := personalLoadTokenData
+	oldAux := personalResolveAuxiliaryAccessToken
+	oldClientID := personalClientID
+	oldCredentials := personalResolveAppCredentialsStrict
+	previousProfile := authpkg.RuntimeProfile()
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		personalLoadProfiles = oldLoadProfiles
+		personalLoadTokenData = oldLoadToken
+		personalResolveAuxiliaryAccessToken = oldAux
+		personalClientID = oldClientID
+		personalResolveAppCredentialsStrict = oldCredentials
+		authpkg.SetRuntimeProfile(previousProfile)
+	})
+
+	personalLoadTokenData = func(string) (*authpkg.TokenData, error) {
+		t.Fatal("explicit token identity read sensitive TokenData")
+		return nil, nil
+	}
+	personalResolveAuxiliaryAccessToken = func(context.Context, string, string) (string, error) {
+		t.Fatal("explicit token identity resolved local OAuth")
+		return "", nil
+	}
+	personalClientID = func() string {
+		t.Fatal("explicit root client ID was not preferred")
+		return ""
+	}
+	personalResolveAppCredentialsStrict = func(string) (string, string, authpkg.CredentialSource, authpkg.CredentialSource, error) {
+		t.Fatal("explicit root client ID unexpectedly fell back to app credentials")
+		return "", "", "", "", nil
+	}
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		return &authpkg.ProfilesConfig{
+			Version: 2,
+			Profiles: []authpkg.Profile{{
+				Name:     "Runtime profile",
+				CorpID:   "profile-corp",
+				CorpName: "Runtime Org",
+				UserID:   "profile-user",
+				UserName: "Runtime User",
+				ClientID: "profile-client",
+			}},
+		}, nil
+	}
+	authpkg.SetRuntimeProfile("Runtime Org:Runtime User")
+	edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
+		return map[string]edition.RuntimeDefaultFn{
+			"$corpId":        func(context.Context) (string, bool) { return "runtime-corp", true },
+			"$currentUserId": func(context.Context) (string, bool) { return "", false },
+		}
+	}})
+
+	identity, err := resolvePersonalEventIdentityWithToken(
+		context.Background(), "unused", "runtime-source", " runtime-canary ", "root-client",
+	)
+	if err != nil {
+		t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
+	}
+	if identity.CorpID != "runtime-corp" || identity.UserID != "profile-user" || identity.ClientID != "root-client" {
+		t.Fatalf("identity metadata = %#v", identity)
+	}
+	if identity.AccessToken != "" {
+		t.Fatalf("identity retained raw runtime token: %q", identity.AccessToken)
+	}
+	if identity.LocalSubject != "" {
+		t.Fatalf("complete identity LocalSubject = %q, want empty", identity.LocalSubject)
+	}
+}
+
+func TestCrossPlatformCoverageResolvePersonalEventIdentityWithCompleteRuntimeMetadataSkipsProfiles(t *testing.T) {
+	oldEdition := edition.Get()
+	oldLoadProfiles := personalLoadProfiles
+	oldRuntimeClientID := personalRuntimeEventClientID
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		personalLoadProfiles = oldLoadProfiles
+		personalRuntimeEventClientID = oldRuntimeClientID
+	})
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		t.Fatal("complete host metadata unexpectedly read profiles.json")
+		return nil, nil
+	}
+	edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
+		return map[string]edition.RuntimeDefaultFn{
+			"$corpId":        func(context.Context) (string, bool) { return "runtime-corp", true },
+			"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
+		}
+	}})
+	personalRuntimeEventClientID = func() string { return "edition-client" }
+	identity, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary", "root-client")
+	if err != nil {
+		t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
+	}
+	if identity.CorpID != "runtime-corp" || identity.UserID != "runtime-user" || identity.ClientID != "root-client" {
+		t.Fatalf("identity = %#v", identity)
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeEventClientIDPrefersEditionBeforeEnvironment(t *testing.T) {
+	oldEdition := edition.Get()
+	t.Cleanup(func() { edition.Override(oldEdition) })
+	t.Setenv("DWS_CLIENT_ID", "environment-client")
+
+	edition.Override(&edition.Hooks{AuthClientID: "edition-client"})
+	if got := runtimePersonalEventClientID(); got != "edition-client" {
+		t.Fatalf("runtime client ID = %q, want edition hook", got)
+	}
+	edition.Override(&edition.Hooks{})
+	if got := runtimePersonalEventClientID(); got != "environment-client" {
+		t.Fatalf("runtime client ID = %q, want environment fallback", got)
+	}
+}
+
+func TestCrossPlatformCoverageCompleteRuntimeIdentityUsesEditionClientBeforeProfiles(t *testing.T) {
+	oldEdition := edition.Get()
+	oldLoadProfiles := personalLoadProfiles
+	oldRuntimeClientID := personalRuntimeEventClientID
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		personalLoadProfiles = oldLoadProfiles
+		personalRuntimeEventClientID = oldRuntimeClientID
+	})
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		t.Fatal("complete host metadata unexpectedly read profiles.json")
+		return nil, errors.New("unreachable")
+	}
+	personalRuntimeEventClientID = func() string { return "edition-client" }
+	edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
+		return map[string]edition.RuntimeDefaultFn{
+			"$corpId":        func(context.Context) (string, bool) { return "runtime-corp", true },
+			"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
+		}
+	}})
+	identity, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary")
+	if err != nil {
+		t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
+	}
+	if identity.CorpID != "runtime-corp" || identity.UserID != "runtime-user" || identity.ClientID != "edition-client" {
+		t.Fatalf("identity = %#v", identity)
+	}
+}
+
+func TestCrossPlatformCoverageSelectedProfileClientPrecedesPersistedGlobalClient(t *testing.T) {
+	oldEdition := edition.Get()
+	oldLoadProfiles := personalLoadProfiles
+	oldRuntimeClientID := personalRuntimeEventClientID
+	oldClientID := personalClientID
+	previousProfile := authpkg.RuntimeProfile()
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		personalLoadProfiles = oldLoadProfiles
+		personalRuntimeEventClientID = oldRuntimeClientID
+		personalClientID = oldClientID
+		authpkg.SetRuntimeProfile(previousProfile)
+	})
+	edition.Override(&edition.Hooks{})
+	personalRuntimeEventClientID = func() string { return "" }
+	personalClientID = func() string { return "stale-global-client" }
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		return &authpkg.ProfilesConfig{
+			Version:        3,
+			CurrentProfile: "corp:user",
+			Profiles: []authpkg.Profile{{
+				Name: "Selected", CorpID: "corp", UserID: "user", ClientID: "profile-client",
+			}},
+		}, nil
+	}
+	authpkg.SetRuntimeProfile("corp:user")
+	identity, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary")
+	if err != nil {
+		t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
+	}
+	if identity.ClientID != "profile-client" {
+		t.Fatalf("ClientID = %q, want selected profile client", identity.ClientID)
+	}
+}
+
+func TestCrossPlatformCoverageMalformedPersistedProfilesDoNotBlockRuntimeDefaultsAndGlobalClient(t *testing.T) {
+	oldEdition := edition.Get()
+	oldLoadProfiles := personalLoadProfiles
+	oldRuntimeClientID := personalRuntimeEventClientID
+	oldClientID := personalClientID
+	previousProfile := authpkg.RuntimeProfile()
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		personalLoadProfiles = oldLoadProfiles
+		personalRuntimeEventClientID = oldRuntimeClientID
+		personalClientID = oldClientID
+		authpkg.SetRuntimeProfile(previousProfile)
+	})
+	authpkg.SetRuntimeProfile("")
+	personalRuntimeEventClientID = func() string { return "" }
+	personalClientID = func() string { return "global-client" }
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		return nil, errors.New("malformed persisted profiles")
+	}
+	edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
+		return map[string]edition.RuntimeDefaultFn{
+			"$corpId":        func(context.Context) (string, bool) { return "runtime-corp", true },
+			"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
+		}
+	}})
+
+	identity, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary")
+	if err != nil {
+		t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
+	}
+	if identity.CorpID != "runtime-corp" || identity.UserID != "runtime-user" || identity.ClientID != "global-client" {
+		t.Fatalf("identity = %#v", identity)
+	}
+}
+
+func TestCrossPlatformCoverageResolvePersonalEventIdentityWithTokenRejectsMultipleProfilesBeforeMetadata(t *testing.T) {
+	oldEdition := edition.Get()
+	oldLoadProfiles := personalLoadProfiles
+	previousProfile := authpkg.RuntimeProfile()
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		personalLoadProfiles = oldLoadProfiles
+		authpkg.SetRuntimeProfile(previousProfile)
+	})
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		t.Fatal("multiple runtime profiles unexpectedly reached metadata loading")
+		return nil, nil
+	}
+	authpkg.SetRuntimeProfile("corp-a:user-a,corp-b:user-b")
+	edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
+		return map[string]edition.RuntimeDefaultFn{
+			"$corpId":        func(context.Context) (string, bool) { return "runtime-corp", true },
+			"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
+		}
+	}})
+	_, err := resolvePersonalEventIdentityWithToken(context.Background(), "unused", "source", "canary", "root-client")
+	if err == nil || !strings.Contains(err.Error(), "exactly one --profile") {
+		t.Fatalf("multiple-profile error = %v", err)
+	}
+}
+
+func TestCrossPlatformCoverageExplicitProfileRequiresMetadataRegistry(t *testing.T) {
+	oldLoadProfiles := personalLoadProfiles
+	defer func() { personalLoadProfiles = oldLoadProfiles }()
+	oldProfile := authpkg.RuntimeProfile()
+	authpkg.SetRuntimeProfile("missing-profile")
+	defer authpkg.SetRuntimeProfile(oldProfile)
+
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		return &authpkg.ProfilesConfig{}, nil
+	}
+	_, err := personalEventProfileMetadata(t.TempDir())
+	if err == nil || !strings.Contains(err.Error(), `profile "missing-profile" not found`) {
+		t.Fatalf("personalEventProfileMetadata() error = %v", err)
+	}
+}
+
+func TestCrossPlatformCoverageCompleteRuntimeIdentityStillValidatesExplicitProfile(t *testing.T) {
+	oldEdition := edition.Get()
+	oldLoadProfiles := personalLoadProfiles
+	oldRuntimeClientID := personalRuntimeEventClientID
+	oldProfile := authpkg.RuntimeProfile()
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		personalLoadProfiles = oldLoadProfiles
+		personalRuntimeEventClientID = oldRuntimeClientID
+		authpkg.SetRuntimeProfile(oldProfile)
+	})
+
+	authpkg.SetRuntimeProfile("missing-profile")
+	personalLoadProfiles = func(string) (*authpkg.ProfilesConfig, error) {
+		return &authpkg.ProfilesConfig{}, nil
+	}
+	personalRuntimeEventClientID = func() string { return "runtime-client" }
+	edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
+		return map[string]edition.RuntimeDefaultFn{
+			"$corpId":        func(context.Context) (string, bool) { return "runtime-corp", true },
+			"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
+		}
+	}})
+
+	_, err := resolvePersonalEventIdentityWithToken(context.Background(), t.TempDir(), "source", "canary")
+	if err == nil || !strings.Contains(err.Error(), `profile "missing-profile" not found`) {
+		t.Fatalf("resolvePersonalEventIdentityWithToken() error = %v", err)
+	}
+}
+
+func TestCrossPlatformCoveragePersonalProfileMetadataOrganizationCurrentBeatsUnresolved(t *testing.T) {
+	cfg := &authpkg.ProfilesConfig{
+		Version: 3,
+		Profiles: []authpkg.Profile{
+			{Name: "Historical", CorpID: "corp-1"},
+			{Name: "Exact", CorpID: "corp-1", UserID: "user-1"},
+		},
+		OrgCurrentProfiles: map[string]string{"corp-1": "corp-1:user-1"},
+	}
+	profile, err := selectPersonalEventProfileMetadata(cfg, "corp-1", make(map[string]struct{}))
+	if err != nil {
+		t.Fatalf("selectPersonalEventProfileMetadata() error = %v", err)
+	}
+	if profile == nil || profile.UserID != "user-1" {
+		t.Fatalf("selected profile = %#v, want organization current account", profile)
+	}
+}
+
+func TestCrossPlatformCoverageExplicitTokenControlClientRedactsReflected401(t *testing.T) {
+	const token = "runtime-control-canary"
+	oldLogger := slog.Default()
+	var logs bytes.Buffer
+	slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug})))
+	t.Cleanup(func() { slog.SetDefault(oldLogger) })
+
+	client := newPersonalEventControlClient("unused", "https://control.invalid", personal.Identity{
+		ClientID: "client", SourceID: "source",
+	}, token)
+	wrapped, ok := client.HTTPClient.Transport.(runtimeTokenControlTransport)
+	if !ok {
+		t.Fatalf("control transport = %T, want runtimeTokenControlTransport", client.HTTPClient.Transport)
+	}
+	var authorization string
+	wrapped.base = eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
+		authorization = req.Header.Get("Authorization")
+		body := `{"code":"UNAUTHORIZED","message":"rejected ` + token + `"}`
+		header := make(http.Header)
+		header.Set("X-Request-Id", "request-"+token)
+		header.Set("X-Trace-Id", "trace-"+token)
+		return &http.Response{
+			StatusCode: http.StatusUnauthorized,
+			Header:     header,
+			Body:       io.NopCloser(strings.NewReader(body)),
+			Request:    req,
+		}, nil
+	})
+	client.HTTPClient.Transport = wrapped
+
+	_, err := client.ListSubscriptions(context.Background(), personal.ListOptions{})
+	if err == nil {
+		t.Fatal("ListSubscriptions() unexpectedly succeeded")
+	}
+	if authorization != "Bearer "+token {
+		t.Fatalf("Authorization = %q", authorization)
+	}
+	if strings.Contains(err.Error(), token) || strings.Contains(logs.String(), token) {
+		t.Fatalf("runtime token leaked: error=%q logs=%q", err, logs.String())
+	}
+	if !strings.Contains(err.Error(), "RUNTIME_TOKEN_REJECTED") {
+		t.Fatalf("error = %q, want fixed runtime token rejection", err)
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeTokenRedirectGuardDoesNotForwardCustomHeader(t *testing.T) {
+	const token = "runtime-redirect-canary"
+	var controlTargetHits, ticketTargetHits atomic.Int32
+	controlTarget := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		controlTargetHits.Add(1)
+		if r.Header.Get("x-user-access-token") == token {
+			t.Error("control redirect forwarded runtime token")
+		}
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer controlTarget.Close()
+	controlOrigin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("x-user-access-token") != token {
+			t.Error("control origin did not receive runtime token")
+		}
+		http.Redirect(w, r, controlTarget.URL, http.StatusFound)
+	}))
+	defer controlOrigin.Close()
+
+	client := newPersonalEventControlClient("unused", controlOrigin.URL, personal.Identity{
+		ClientID: "client", SourceID: "source",
+	}, token)
+	_, controlErr := client.ListSubscriptions(context.Background(), personal.ListOptions{})
+	if controlErr == nil {
+		t.Fatal("cross-host control redirect unexpectedly succeeded")
+	}
+	if controlTargetHits.Load() != 0 || strings.Contains(controlErr.Error(), token) {
+		t.Fatalf("control redirect hits=%d error=%q", controlTargetHits.Load(), controlErr)
+	}
+
+	ticketTarget := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		ticketTargetHits.Add(1)
+		if r.Header.Get("x-user-access-token") == token {
+			t.Error("ticket redirect forwarded runtime token")
+		}
+		w.WriteHeader(http.StatusNoContent)
+	}))
+	defer ticketTarget.Close()
+	ticketOrigin := httptest.NewServer(http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
+		if r.Header.Get("x-user-access-token") != token {
+			t.Error("ticket origin did not receive runtime token")
+		}
+		http.Redirect(w, r, ticketTarget.URL, http.StatusFound)
+	}))
+	defer ticketOrigin.Close()
+
+	broker := runtimecred.New(runtimecred.Config{RequireSeed: true})
+	if _, err := broker.Update(0, token); err != nil {
+		t.Fatal(err)
+	}
+	src, err := newPersonalStreamSource(context.Background(), personalStreamSourceOptions{
+		ConfigDir:        "unused",
+		Identity:         personal.Identity{ClientID: "client", SourceID: "source"},
+		TicketURL:        ticketOrigin.URL,
+		CredentialBroker: broker,
+		RuntimeTokenMode: true,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	err = src.Start(context.Background(), func(*dwsevent.RawEvent) {})
+	if err == nil {
+		t.Fatal("cross-host ticket redirect unexpectedly succeeded")
+	}
+	if ticketTargetHits.Load() != 0 || strings.Contains(err.Error(), token) {
+		t.Fatalf("ticket redirect hits=%d error=%q", ticketTargetHits.Load(), err)
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeTokenRedirectPolicyBranches(t *testing.T) {
+	origin, _ := http.NewRequest(http.MethodGet, "https://control.example/start", nil)
+	sameHost, _ := http.NewRequest(http.MethodGet, "https://control.example/next", nil)
+	if err := runtimeTokenRedirectPolicy(sameHost, []*http.Request{origin}); err != nil {
+		t.Fatalf("same-host HTTPS redirect rejected: %v", err)
+	}
+	for name, request := range map[string]*http.Request{
+		"cross-host": func() *http.Request {
+			r, _ := http.NewRequest(http.MethodGet, "https://other.example/next", nil)
+			return r
+		}(),
+		"downgrade": func() *http.Request {
+			r, _ := http.NewRequest(http.MethodGet, "http://control.example/next", nil)
+			return r
+		}(),
+	} {
+		if err := runtimeTokenRedirectPolicy(request, []*http.Request{origin}); !errors.Is(err, http.ErrUseLastResponse) {
+			t.Fatalf("%s redirect policy error = %v", name, err)
+		}
+	}
+	if err := runtimeTokenRedirectPolicy(nil, nil); !errors.Is(err, http.ErrUseLastResponse) {
+		t.Fatalf("empty redirect chain error = %v", err)
+	}
+}
+
+func TestCrossPlatformCoverageExplicitTokenControlClientRedactsEveryErrorEnvelope(t *testing.T) {
+	const token = "runtime-control-all-status-canary"
+	tests := []struct {
+		name   string
+		status int
+		body   string
+	}{
+		{name: "bad-request", status: http.StatusBadRequest, body: `{"code":"BAD_REQUEST","message":"` + token + `"}`},
+		{name: "server-error", status: http.StatusInternalServerError, body: `{"code":"INTERNAL","message":"` + token + `"}`},
+		{name: "success-false", status: http.StatusOK, body: `{"success":false,"errorCode":"DENIED","errorMsg":"` + token + `"}`},
+	}
+	for _, tc := range tests {
+		t.Run(tc.name, func(t *testing.T) {
+			oldLogger := slog.Default()
+			var logs bytes.Buffer
+			slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug})))
+			t.Cleanup(func() { slog.SetDefault(oldLogger) })
+
+			client := newPersonalEventControlClient("unused", "https://control.invalid", personal.Identity{
+				ClientID: "client", SourceID: "source",
+			}, token)
+			wrapped := client.HTTPClient.Transport.(runtimeTokenControlTransport)
+			wrapped.base = eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
+				header := make(http.Header)
+				header.Set("X-Request-Id", "request-"+token)
+				header.Set("X-Trace-Id", "trace-"+token)
+				return &http.Response{
+					StatusCode: tc.status,
+					Header:     header,
+					Body:       io.NopCloser(strings.NewReader(tc.body)),
+					Request:    req,
+				}, nil
+			})
+			client.HTTPClient.Transport = wrapped
+
+			_, err := client.ListSubscriptions(context.Background(), personal.ListOptions{})
+			if err == nil {
+				t.Fatal("ListSubscriptions() unexpectedly succeeded")
+			}
+			if strings.Contains(err.Error(), token) || strings.Contains(logs.String(), token) {
+				t.Fatalf("runtime token leaked: error=%q logs=%q", err, logs.String())
+			}
+		})
+	}
+}
+
+func TestCrossPlatformCoverageExplicitTokenControlTransportPreservesSuccessfulResponse(t *testing.T) {
+	client := newPersonalEventControlClient("unused", "https://control.invalid", personal.Identity{
+		ClientID: "client", SourceID: "source",
+	}, "runtime-success-canary")
+	wrapped := client.HTTPClient.Transport.(runtimeTokenControlTransport)
+	wrapped.base = eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
+		return &http.Response{
+			StatusCode: http.StatusOK,
+			Header:     make(http.Header),
+			Body:       io.NopCloser(strings.NewReader(`{"success":true,"result":{"items":[],"total":0}}`)),
+			Request:    req,
+		}, nil
+	})
+	client.HTTPClient.Transport = wrapped
+	if _, err := client.ListSubscriptions(context.Background(), personal.ListOptions{}); err != nil {
+		t.Fatalf("ListSubscriptions() successful response error = %v", err)
+	}
+}
+
+func TestCrossPlatformCoverageExplicitTokenControlClientRedactsJSONEscapedToken(t *testing.T) {
+	const token = "runtime<escaped>&canary"
+	body, err := json.Marshal(map[string]any{"code": "BAD_REQUEST", "message": "rejected " + token})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if bytes.Contains(body, []byte(token)) {
+		t.Fatalf("fixture was not JSON-escaped: %s", body)
+	}
+	oldLogger := slog.Default()
+	var logs bytes.Buffer
+	slog.SetDefault(slog.New(slog.NewTextHandler(&logs, &slog.HandlerOptions{Level: slog.LevelDebug})))
+	t.Cleanup(func() { slog.SetDefault(oldLogger) })
+
+	client := newPersonalEventControlClient("unused", "https://control.invalid", personal.Identity{
+		ClientID: "client", SourceID: "source",
+	}, token)
+	wrapped := client.HTTPClient.Transport.(runtimeTokenControlTransport)
+	wrapped.base = eventRuntimeRoundTripFunc(func(req *http.Request) (*http.Response, error) {
+		return &http.Response{
+			StatusCode: http.StatusBadRequest,
+			Header:     make(http.Header),
+			Body:       io.NopCloser(bytes.NewReader(body)),
+			Request:    req,
+		}, nil
+	})
+	client.HTTPClient.Transport = wrapped
+	_, err = client.ListSubscriptions(context.Background(), personal.ListOptions{})
+	if err == nil {
+		t.Fatal("ListSubscriptions() unexpectedly succeeded")
+	}
+	if strings.Contains(err.Error(), token) || strings.Contains(logs.String(), token) {
+		t.Fatalf("escaped runtime token leaked: error=%q logs=%q", err, logs.String())
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeTokenBusModeSkipsLocalOAuthIdentity(t *testing.T) {
+	t.Setenv("DWS_CONFIG_DIR", t.TempDir())
+	oldResolve := eventResolvePersonal
+	oldSource := eventNewPersonalSource
+	oldRun := eventBusRun
+	t.Cleanup(func() {
+		eventResolvePersonal = oldResolve
+		eventNewPersonalSource = oldSource
+		eventBusRun = oldRun
+	})
+
+	resolvedLocal := false
+	eventResolvePersonal = func(context.Context, string, string) (personal.Identity, error) {
+		resolvedLocal = true
+		return personal.Identity{}, nil
+	}
+	var sourceOpts personalStreamSourceOptions
+	eventNewPersonalSource = func(_ context.Context, opts personalStreamSourceOptions) (*source.PersonalSource, error) {
+		sourceOpts = opts
+		return nil, nil
+	}
+	var busCfg bus.Config
+	eventBusRun = func(_ context.Context, cfg bus.Config) error {
+		busCfg = cfg
+		return nil
+	}
+
+	cmd := newEventBusCommand()
+	cmd.SetArgs([]string{
+		"--source-kind", "personal_stream",
+		"--runtime-token-mode",
+		"--identity-hash", "0123456789abcdef",
+		"--client-id", "runtime-client",
+		"--stream-source-id", "runtime-source",
+		"--idle-timeout", "0",
+	})
+	if err := cmd.Execute(); err != nil {
+		t.Fatalf("event _bus runtime mode error = %v", err)
+	}
+	if resolvedLocal {
+		t.Fatal("runtime token bus resolved local OAuth identity")
+	}
+	if sourceOpts.CredentialBroker == nil || busCfg.CredentialBroker != sourceOpts.CredentialBroker {
+		t.Fatal("personal source and bus did not share one credential broker")
+	}
+	if busCfg.IdentityHash != "0123456789abcdef" || busCfg.ClientID != "runtime-client" || busCfg.SourceID != "runtime-source" {
+		t.Fatalf("bus identity = %#v", busCfg)
+	}
+	generation, err := sourceOpts.CredentialBroker.Update(0, "detached-activation-canary")
+	if err != nil {
+		t.Fatalf("seed detached broker: %v", err)
+	}
+	waitCtx, cancel := context.WithTimeout(context.Background(), 20*time.Millisecond)
+	defer cancel()
+	if _, err := sourceOpts.CredentialBroker.Resolve(waitCtx); !errors.Is(err, context.DeadlineExceeded) {
+		t.Fatalf("detached broker resolved before consumer activation: %v", err)
+	}
+	if _, err := sourceOpts.CredentialBroker.Activate(generation); err != nil {
+		t.Fatalf("activate detached broker: %v", err)
+	}
+	if resolved, err := sourceOpts.CredentialBroker.Resolve(context.Background()); err != nil || resolved == "" {
+		t.Fatalf("detached broker did not resolve after activation: %v", err)
+	}
+}
+
+func TestCrossPlatformCoverageForegroundRuntimeBrokerDoesNotRequireActivation(t *testing.T) {
+	broker := newPersonalCredentialBroker(t.TempDir(), true, false)
+	if _, err := broker.Update(0, "foreground-activation-canary"); err != nil {
+		t.Fatalf("seed foreground broker: %v", err)
+	}
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second)
+	defer cancel()
+	if resolved, err := broker.Resolve(ctx); err != nil || resolved == "" {
+		t.Fatalf("foreground broker unexpectedly waited for activation: %v", err)
+	}
+}
+
+func TestCrossPlatformCoveragePersonalRuntimeBusSpawnArgsContainNoSecretOrProfile(t *testing.T) {
+	const token = "runtime-spawn-canary"
+	args := personalBusSpawnArgsForToken(personal.Identity{
+		ClientID: "client", SourceID: "source", CorpID: "corp", UserID: "user",
+	}, "identity-hash", "normal", "https://ticket.invalid", "corp:user", token)
+	joined := strings.Join(args, " ")
+	for _, forbidden := range []string{token, "--profile", "corp:user"} {
+		if strings.Contains(joined, forbidden) {
+			t.Fatalf("spawn args leaked %q: %q", forbidden, joined)
+		}
+	}
+	for _, required := range []string{"--runtime-token-mode", "--identity-hash", "identity-hash", "--stream-source-id", "source"} {
+		if !strings.Contains(joined, required) {
+			t.Fatalf("spawn args %q missing %q", joined, required)
+		}
+	}
+}
+
+func TestCrossPlatformCoverageUnsupportedOldBusDoesNotDeleteReusedSubscription(t *testing.T) {
+	t.Setenv("DWS_CONFIG_DIR", t.TempDir())
+	oldEdition := edition.Get()
+	oldEnsure := personalEnsureSubscription
+	oldUpsert := personalUpsertRunState
+	oldDelete := personalDeleteSubscription
+	oldRemove := personalRemoveRunStates
+	oldConsume := personalConsumeRun
+	oldValidate := personalValidateConsumeConfig
+	oldConflict := personalValidateNoOutputConflict
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		personalEnsureSubscription = oldEnsure
+		personalUpsertRunState = oldUpsert
+		personalDeleteSubscription = oldDelete
+		personalRemoveRunStates = oldRemove
+		personalConsumeRun = oldConsume
+		personalValidateConsumeConfig = oldValidate
+		personalValidateNoOutputConflict = oldConflict
+	})
+	edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
+		return map[string]edition.RuntimeDefaultFn{
+			"$corpId":        func(context.Context) (string, bool) { return "runtime-corp", true },
+			"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
+		}
+	}})
+	personalEnsureSubscription = func(context.Context, *personal.Client, personal.Identity, personalConsumeOptions) (*personal.Subscription, string, string, error) {
+		return &personal.Subscription{SubscribeID: "sub-existing"}, personal.EventMention, "at", nil
+	}
+	personalUpsertRunState = func(string, personal.RunState) error { return nil }
+	deleteCalls := 0
+	personalDeleteSubscription = func(*personal.Client, context.Context, string) error {
+		deleteCalls++
+		return nil
+	}
+	var removed []string
+	personalRemoveRunStates = func(_ string, ids []string) error {
+		removed = append(removed, ids...)
+		return nil
+	}
+	personalValidateConsumeConfig = func(consume.Config) error { return nil }
+	personalValidateNoOutputConflict = func(consume.Config, string) error { return nil }
+	personalConsumeRun = func(_ context.Context, cfg consume.Config) error {
+		if strings.TrimSpace(cfg.RuntimeToken) == "" {
+			t.Fatal("runtime token was not wired to consume")
+		}
+		return &consume.RuntimeTokenUnsupportedError{BusPID: 72}
+	}
+
+	err := runPersonalEventConsumeSingle(newPersonalCoverageCommand(), personalConsumeOptions{
+		SubscribeID:      "sub-existing",
+		ExplicitToken:    "old-bus-cleanup-canary",
+		ClientIDOverride: "runtime-client",
+	})
+	if !errors.Is(err, consume.ErrRuntimeTokenUnsupported) {
+		t.Fatalf("consume error = %v", err)
+	}
+	if deleteCalls != 0 {
+		t.Fatalf("reused remote subscription was deleted %d time(s)", deleteCalls)
+	}
+	if len(removed) != 0 {
+		t.Fatalf("reused local run-state was removed: %#v", removed)
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeTokenReusedDryRunUsesExplicitControlCredential(t *testing.T) {
+	const token = "runtime-dry-run-control-canary"
+	t.Setenv("DWS_CONFIG_DIR", t.TempDir())
+	oldEdition := edition.Get()
+	oldEnsure := personalEnsureSubscription
+	oldUpsert := personalUpsertRunState
+	oldConsume := personalConsumeRun
+	oldBusRun := personalBusRun
+	t.Cleanup(func() {
+		edition.Override(oldEdition)
+		personalEnsureSubscription = oldEnsure
+		personalUpsertRunState = oldUpsert
+		personalConsumeRun = oldConsume
+		personalBusRun = oldBusRun
+	})
+	edition.Override(&edition.Hooks{RuntimeDefaults: func() map[string]edition.RuntimeDefaultFn {
+		return map[string]edition.RuntimeDefaultFn{
+			"$corpId":        func(context.Context) (string, bool) { return "runtime-corp", true },
+			"$currentUserId": func(context.Context) (string, bool) { return "runtime-user", true },
+		}
+	}})
+
+	personalEnsureSubscription = func(ctx context.Context, client *personal.Client, _ personal.Identity, _ personalConsumeOptions) (*personal.Subscription, string, string, error) {
+		if _, ok := client.HTTPClient.Transport.(runtimeTokenControlTransport); !ok {
+			t.Fatalf("control transport = %T, want runtimeTokenControlTransport", client.HTTPClient.Transport)
+		}
+		got, err := client.AccessTokenProvider(ctx)
+		if err != nil || got != token {
+			t.Fatalf("control token = %q, %v", got, err)
+		}
+		return &personal.Subscription{SubscribeID: "sub-existing"}, personal.EventMention, "at", nil
+	}
+	personalUpsertRunState = func(string, personal.RunState) error {
+		t.Fatal("dry-run unexpectedly persisted run state")
+		return nil
+	}
+	consumeCalls := 0
+	personalConsumeRun = func(_ context.Context, cfg consume.Config) error {
+		consumeCalls++
+		if !cfg.DryRun {
+			t.Fatal("consume config is not dry-run")
+		}
+		if strings.Contains(strings.Join(cfg.SpawnExtraArgs, " "), token) {
+			t.Fatal("dry-run spawn args leaked runtime token")
+		}
+		return nil
+	}
+	personalBusRun = func(context.Context, bus.Config) error {
+		t.Fatal("dry-run unexpectedly started a bus")
+		return nil
+	}
+
+	err := runPersonalEventConsumeSingle(newPersonalCoverageCommand(), personalConsumeOptions{
+		SubscribeID:      "sub-existing",
+		ExplicitToken:    token,
+		ClientIDOverride: "runtime-client",
+		Common:           commonConsumeOptions{DryRun: true},
+	})
+	if err != nil {
+		t.Fatalf("dry-run consume error = %v", err)
+	}
+	if consumeCalls != 1 {
+		t.Fatalf("dry-run consume calls = %d, want 1", consumeCalls)
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeTokenControlRejectionReleasesSubscriptionClaim(t *testing.T) {
+	store := &personalRecordingAttemptStore{}
+	reservation := &personalSubscriptionAttemptReservation{
+		store: store,
+		claim: &personal.AttemptClaim{AttemptID: "runtime-token-attempt"},
+		items: []personalSubscriptionAttemptItem{{eventKey: personal.EventMention, fingerprint: strings.Repeat("a", 64)}},
+	}
+	cause := &personal.APIError{
+		Code:       "RUNTIME_TOKEN_REJECTED",
+		Message:    "event runtime token was rejected; retry with a fresh host credential",
+		HTTPStatus: http.StatusUnauthorized,
+	}
+	if !personalRuntimeTokenControlRejection(cause) {
+		t.Fatal("runtime token control rejection was not classified")
+	}
+	err := reservation.releaseRuntimeTokenFailure()
+	if err == nil || !strings.Contains(err.Error(), "runtime token was rejected") {
+		t.Fatalf("releaseRuntimeTokenFailure() error = %v", err)
+	}
+	if store.releaseCalls != 1 || store.failureCalls != 0 {
+		t.Fatalf("attempt store release=%d failure=%d, want release only", store.releaseCalls, store.failureCalls)
+	}
+}
+
+type eventRuntimeRoundTripFunc func(*http.Request) (*http.Response, error)
+
+func (f eventRuntimeRoundTripFunc) RoundTrip(req *http.Request) (*http.Response, error) {
+	return f(req)
+}

internal/app/root.go

@@ -472,7 +472,7 @@ func newRootCommandWithEngine(rootCtx context.Context, engine *pipeline.Engine,
 		newConfigCommand(),
 		newDoctorCommand(),
 		newRecoveryCommand(),
-		newEventCommand(),
+		newEventCommand(flags),
 		newAuditCommand(),
 		newCompletionCommand(root),
 		newUpgradeCommand(),

internal/app/testmain_test.go

@@ -44,6 +44,10 @@ import (
 // test binary never launches a page on the developer's machine; tests that
 // need to assert the URL can still replace openBrowserFunc locally.
 func TestMain(m *testing.M) {
+	if code, ok := runRuntimeTokenDetachedE2EChild(); ok {
+		os.Exit(code)
+	}
+
 	tmpDir, err := os.MkdirTemp("", "dws-app-test-keychain-")
 	if err != nil {
 		panic("create test keychain tempdir: " + err.Error())

internal/auth/profile_metadata_test.go

@@ -0,0 +1,40 @@
+// 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.
+
+package auth
+
+import "testing"
+
+func TestCrossPlatformCoverageResolveProfileMetadataUsesSelectorGrammarAndReturnsCopy(t *testing.T) {
+	cfg := &ProfilesConfig{
+		Version: 3,
+		Profiles: []Profile{
+			{Name: "Historical", CorpID: "corp-1", CorpName: "Example Org"},
+			{Name: "Exact", CorpID: "corp-1", CorpName: "Example Org", UserID: "user-1", UserName: "Example User", ClientID: "client-1"},
+		},
+		OrgCurrentProfiles: map[string]string{"corp-1": "corp-1:user-1"},
+	}
+
+	for _, selector := range []string{"corp-1", "Example Org", "corp-1:Example User", "Example Org:Example User"} {
+		profile, err := ResolveProfileMetadata(cfg, selector)
+		if err != nil {
+			t.Fatalf("ResolveProfileMetadata(%q) error = %v", selector, err)
+		}
+		if profile == nil || profile.UserID != "user-1" || profile.ClientID != "client-1" {
+			t.Fatalf("ResolveProfileMetadata(%q) = %#v", selector, profile)
+		}
+	}
+
+	profile, err := ResolveProfileMetadata(cfg, "corp-1:user-1")
+	if err != nil {
+		t.Fatalf("ResolveProfileMetadata(exact) error = %v", err)
+	}
+	profile.Name = "mutated copy"
+	if cfg.Profiles[1].Name != "Exact" {
+		t.Fatalf("ResolveProfileMetadata returned registry-owned pointer")
+	}
+	if _, err := ResolveProfileMetadata(cfg, "missing"); err == nil {
+		t.Fatal("ResolveProfileMetadata(missing) unexpectedly succeeded")
+	}
+}

internal/auth/profiles.go

@@ -738,6 +738,20 @@ func ResolveProfileWithScope(configDir, selector string) (*Profile, bool, error)
 	return result, exact, err
 }
 
+// ResolveProfileMetadata applies the public profile-selector grammar to an
+// already-loaded, non-sensitive profiles registry. It performs no migration,
+// keychain access, token loading, or persistence, making it suitable for
+// callers that carry an externally managed bearer credential and need only
+// corp/user/client identity metadata.
+func ResolveProfileMetadata(cfg *ProfilesConfig, selector string) (*Profile, error) {
+	profile, _, err := resolveProfileSelection("", cfg, selector)
+	if err != nil || profile == nil {
+		return nil, err
+	}
+	copy := *profile
+	return &copy, nil
+}
+
 func resolveProfileWithScopeLocked(configDir, selector string) (*Profile, bool, error) {
 	if err := profilesEnsureMigration(configDir); err != nil {
 		return nil, false, err

internal/event/bus/coverage_edges_test.go

@@ -17,6 +17,7 @@ import (
 
 	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
 	eventlock "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/lock"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
 )
 
@@ -324,6 +325,21 @@ func TestCrossPlatformCoverageRunStartupAndSourceEdges(t *testing.T) {
 	if err := Run(context.Background(), base); !errors.Is(err, errBusInjected) {
 		t.Fatalf("source error = %v", err)
 	}
+
+	runtimeWorkDir := shortTempDir(t)
+	base.WorkDir = runtimeWorkDir
+	base.IPCEndpoint = dwsevent.IPCEndpoint(
+		runtimeWorkDir,
+		"open",
+		dwsevent.SourceKindPersonalStream,
+		dwsevent.IdentityHash(runtimeWorkDir),
+	)
+	base.Source = edgeSource{start: func(context.Context, dwsevent.EmitFn) error {
+		return runtimecred.ErrRuntimeTokenRejected
+	}}
+	if err := Run(context.Background(), base); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
+		t.Fatalf("runtime source error = %v", err)
+	}
 }
 
 type scriptedListener struct {

internal/event/bus/cross_platform_coverage_event_core_agent_test.go

@@ -0,0 +1,275 @@
+package bus
+
+import (
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"net"
+	"runtime"
+	"testing"
+	"time"
+
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
+)
+
+func eventCoreDaemon(broker *runtimecred.Broker) *daemon {
+	return &daemon{
+		cfg:      Config{CredentialBroker: broker, IdleTimeout: time.Second},
+		log:      slog.New(slog.NewTextHandler(io.Discard, nil)),
+		hub:      NewHub(4),
+		started:  time.Now(),
+		idleStop: make(chan struct{}),
+	}
+}
+
+func eventCoreConnection(d *daemon, wrap func(net.Conn) net.Conn) (net.Conn, *transport.Writer, *transport.Reader, <-chan struct{}) {
+	server, client := net.Pipe()
+	if wrap != nil {
+		server = wrap(server)
+	}
+	done := make(chan struct{})
+	go func() {
+		d.handleConnection(context.Background(), server)
+		close(done)
+	}()
+	return client, transport.NewWriter(client), transport.NewReader(client), done
+}
+
+func eventCoreWaitDone(t *testing.T, done <-chan struct{}) {
+	t.Helper()
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("connection handler did not stop")
+	}
+}
+
+type eventCoreWriteHookConn struct {
+	net.Conn
+	writes int
+	hook   func(int)
+}
+
+func (c *eventCoreWriteHookConn) Write(p []byte) (int, error) {
+	n, err := c.Conn.Write(p)
+	c.writes++
+	if err == nil && c.hook != nil {
+		c.hook(c.writes)
+	}
+	return n, err
+}
+
+func TestCrossPlatformCoverageEventCoreDaemonHandshakeEdges(t *testing.T) {
+	t.Run("incompatible ack write failure", func(t *testing.T) {
+		d := eventCoreDaemon(nil)
+		client, w, _, done := eventCoreConnection(d, nil)
+		if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: "unsupported"}); err != nil {
+			t.Fatal(err)
+		}
+		_ = client.Close()
+		eventCoreWaitDone(t, done)
+	})
+
+	t.Run("runtime ack write failure", func(t *testing.T) {
+		d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
+		client, w, _, done := eventCoreConnection(d, nil)
+		if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
+			t.Fatal(err)
+		}
+		_ = client.Close()
+		eventCoreWaitDone(t, done)
+	})
+
+	t.Run("terminal runtime hello", func(t *testing.T) {
+		d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
+		d.setTerminalReason(transport.ByeReasonRuntimeTokenRejected)
+		client, w, r, done := eventCoreConnection(d, nil)
+		defer client.Close()
+		if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
+			t.Fatal(err)
+		}
+		var ack transport.HelloAck
+		if err := r.ReadJSON(&ack); err != nil || ack.TerminalReason != transport.ByeReasonRuntimeTokenRejected {
+			t.Fatalf("terminal ack = %#v, %v", ack, err)
+		}
+		eventCoreWaitDone(t, done)
+	})
+
+	t.Run("malformed credential update", func(t *testing.T) {
+		d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
+		client, w, r, done := eventCoreConnection(d, nil)
+		defer client.Close()
+		if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
+			t.Fatal(err)
+		}
+		var ack transport.HelloAck
+		if err := r.ReadJSON(&ack); err != nil {
+			t.Fatal(err)
+		}
+		if _, err := client.Write([]byte("{\n")); err != nil {
+			t.Fatal(err)
+		}
+		eventCoreWaitDone(t, done)
+	})
+
+	t.Run("unexpected credential update", func(t *testing.T) {
+		d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
+		client, w, r, done := eventCoreConnection(d, nil)
+		defer client.Close()
+		if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
+			t.Fatal(err)
+		}
+		var ack transport.HelloAck
+		if err := r.ReadJSON(&ack); err != nil {
+			t.Fatal(err)
+		}
+		if err := w.WriteJSON(transport.Heartbeat{Type: transport.FrameTypeHeartbeat}); err != nil {
+			t.Fatal(err)
+		}
+		var updateAck transport.CredentialUpdateAck
+		if err := r.ReadJSON(&updateAck); err != nil || updateAck.ErrorCode != transport.CredentialErrorInvalid {
+			t.Fatalf("unexpected-frame ack = %#v, %v", updateAck, err)
+		}
+		eventCoreWaitDone(t, done)
+	})
+
+	t.Run("credential ack write failure", func(t *testing.T) {
+		d := eventCoreDaemon(runtimecred.New(runtimecred.Config{}))
+		client, w, r, done := eventCoreConnection(d, nil)
+		if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
+			t.Fatal(err)
+		}
+		var ack transport.HelloAck
+		if err := r.ReadJSON(&ack); err != nil {
+			t.Fatal(err)
+		}
+		if err := w.WriteJSON(transport.CredentialUpdate{
+			Type: transport.FrameTypeCredentialUpdate, ExpectedGeneration: ack.CredentialGeneration, Token: "token",
+		}); err != nil {
+			t.Fatal(err)
+		}
+		_ = client.Close()
+		eventCoreWaitDone(t, done)
+	})
+
+	t.Run("activation conflict", func(t *testing.T) {
+		broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
+		d := eventCoreDaemon(broker)
+		client, w, r, done := eventCoreConnection(d, func(conn net.Conn) net.Conn {
+			return &eventCoreWriteHookConn{Conn: conn, hook: func(write int) {
+				if write == 2 {
+					_, _ = broker.Update(1, "newer-token")
+				}
+			}}
+		})
+		defer client.Close()
+		if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, CredentialMode: transport.CredentialModeRuntimeToken}); err != nil {
+			t.Fatal(err)
+		}
+		var ack transport.HelloAck
+		if err := r.ReadJSON(&ack); err != nil {
+			t.Fatal(err)
+		}
+		if err := w.WriteJSON(transport.CredentialUpdate{Type: transport.FrameTypeCredentialUpdate, Token: "first-token"}); err != nil {
+			t.Fatal(err)
+		}
+		var updateAck transport.CredentialUpdateAck
+		if err := r.ReadJSON(&updateAck); err != nil || !updateAck.Accepted {
+			t.Fatalf("credential ack = %#v, %v", updateAck, err)
+		}
+		var bye transport.Bye
+		if err := r.ReadJSON(&bye); err != nil || bye.Reason != "runtime_credential_activation_failed" {
+			t.Fatalf("activation failure bye = %#v, %v", bye, err)
+		}
+		eventCoreWaitDone(t, done)
+	})
+}
+
+func TestCrossPlatformCoverageEventCoreDaemonWriterStopEdges(t *testing.T) {
+	originalProcs := runtime.GOMAXPROCS(1)
+	t.Cleanup(func() { runtime.GOMAXPROCS(originalProcs) })
+
+	run := func(t *testing.T, queueEvent bool) {
+		t.Helper()
+		d := eventCoreDaemon(nil)
+		client, w, r, done := eventCoreConnection(d, nil)
+		defer client.Close()
+		if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, SubscribeID: "writer-stop"}); err != nil {
+			t.Fatal(err)
+		}
+		var ack transport.HelloAck
+		if err := r.ReadJSON(&ack); err != nil {
+			t.Fatal(err)
+		}
+		deadline := time.Now().Add(time.Second)
+		for d.hub.Len() != 1 && time.Now().Before(deadline) {
+			time.Sleep(time.Millisecond)
+		}
+		time.Sleep(5 * time.Millisecond)
+		d.hub.mu.RLock()
+		var consumer *Consumer
+		for _, candidate := range d.hub.consumers {
+			consumer = candidate
+		}
+		d.hub.mu.RUnlock()
+		if consumer == nil {
+			t.Fatal("consumer not registered")
+		}
+		if queueEvent {
+			consumer.SendCh <- transport.Heartbeat{Type: transport.FrameTypeHeartbeat}
+		}
+		consumer.StopCh <- "writer-stop"
+		var bye transport.Bye
+		if err := r.ReadJSON(&bye); err != nil || bye.Reason != "writer-stop" {
+			t.Fatalf("writer stop bye = %#v, %v", bye, err)
+		}
+		eventCoreWaitDone(t, done)
+	}
+
+	t.Run("recheck after event", func(t *testing.T) { run(t, true) })
+	t.Run("blocked stop select", func(t *testing.T) { run(t, false) })
+}
+
+func TestCrossPlatformCoverageEventCoreDaemonHelpersAndStopAll(t *testing.T) {
+	var nilDaemon *daemon
+	nilDaemon.setTerminalReason("ignored")
+	if nilDaemon.getTerminalReason() != "" {
+		t.Fatal("nil daemon returned terminal reason")
+	}
+	d := eventCoreDaemon(nil)
+	d.setTerminalReason("ignored")
+	if d.getTerminalReason() != "" {
+		t.Fatal("invalid terminal reason was stored")
+	}
+	d.setTerminalReason(transport.ByeReasonRuntimeTokenRejected)
+	if d.getTerminalReason() != transport.ByeReasonRuntimeTokenRejected {
+		t.Fatal("terminal reason was not stored")
+	}
+
+	if code, _ := classifyCredentialUpdateError(runtimecred.ErrEmptyToken); code != transport.CredentialErrorInvalid {
+		t.Fatalf("empty-token classification = %q", code)
+	}
+	if code, message := classifyCredentialUpdateError(errors.New("internal detail")); code != transport.CredentialErrorInternal || message != "runtime credential update failed" {
+		t.Fatalf("internal classification = %q, %q", code, message)
+	}
+
+	hub := NewHub(1)
+	consumer, err := hub.Register(transport.Hello{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if stopped := hub.StopAll("   "); stopped != 1 {
+		t.Fatalf("StopAll = %d", stopped)
+	}
+	select {
+	case reason := <-consumer.StopCh:
+		if reason != "shutdown" {
+			t.Fatalf("default stop reason = %q", reason)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("default stop reason not delivered")
+	}
+	hub.Unregister(consumer.ID)
+}

internal/event/bus/daemon.go

@@ -30,6 +30,7 @@ import (
 
 	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/dedup"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
 )
@@ -78,6 +79,11 @@ type Config struct {
 	// Source is the cloud adapter. Required.
 	Source SourceAdapter
 
+	// CredentialBroker enables additive runtime-token handoff over the
+	// owner-only local IPC transport. Nil preserves the original protocol and
+	// does not advertise runtime-token support.
+	CredentialBroker *runtimecred.Broker
+
 	// IdleTimeout: bus self-exits after this long with zero consumers.
 	// Zero disables (bus runs until SIGTERM).
 	IdleTimeout time.Duration
@@ -106,11 +112,13 @@ type Config struct {
 }
 
 var (
-	daemonMkdirAll        = os.MkdirAll
-	daemonAcquire         = Acquire
-	daemonWriteMeta       = WriteMeta
-	daemonListen          = transport.Listen
-	daemonShutdownTimeout = 2 * time.Second
+	daemonMkdirAll                   = os.MkdirAll
+	daemonAcquire                    = Acquire
+	daemonWriteMeta                  = WriteMeta
+	daemonListen                     = transport.Listen
+	daemonShutdownTimeout            = 2 * time.Second
+	daemonByeDrainTimeout            = 100 * time.Millisecond
+	daemonCredentialHandshakeTimeout = 10 * time.Second
 )
 
 // Run starts the bus daemon. Lifecycle (plan §4 invariant #6):
@@ -237,12 +245,22 @@ func Run(ctx context.Context, cfg Config) error {
 
 	// 6. Wait for shutdown trigger.
 	var exitErr error
+	shutdownReason := "shutdown"
 	select {
 	case <-ctx.Done():
 		log.Info("bus: shutdown requested by ctx", "reason", ctx.Err())
 	case err := <-srcErr:
 		log.Error("bus: source exited", "err", err)
 		exitErr = err
+		shutdownReason = sourceShutdownReason(err)
+		if shutdownReason == transport.ByeReasonRuntimeTokenRejected {
+			// A runtime token can fail immediately after Broker.Update. Serialize
+			// terminal publication with that handshake so the initiating consumer
+			// is registered (or receives a terminal HelloAck) before shutdown.
+			d.credentialHandoffMu.Lock()
+			d.setTerminalReason(shutdownReason)
+			d.credentialHandoffMu.Unlock()
+		}
 	case <-d.idleStop:
 		log.Info("bus: idle timeout reached, shutting down")
 	}
@@ -252,7 +270,7 @@ func Run(ctx context.Context, cfg Config) error {
 	// consumers. The accept-loop barrier is required before WaitGroup.Wait:
 	// sync.WaitGroup forbids a positive Add racing with Wait.
 	cancelRun()
-	d.shutdown(acceptDone)
+	d.shutdown(acceptDone, shutdownReason)
 	<-idleDone
 	<-dropWarnDone
 
@@ -274,6 +292,10 @@ type daemon struct {
 	shutdownMu   sync.Mutex
 	shuttingDown atomic.Bool
 	idleStop     chan struct{}
+
+	credentialHandoffMu sync.Mutex
+	terminalMu          sync.RWMutex
+	terminalReason      string
 }
 
 // closeOnceConn makes every connection close path idempotent. A live consumer
@@ -330,7 +352,8 @@ func (d *daemon) acceptLoop(ctx context.Context) {
 }
 
 // handleConnection processes one IPC connection's full lifecycle: read
-// Hello → register with Hub → spawn writer goroutine → read until EOF/Bye.
+// Hello → optional runtime credential negotiation → register with Hub → spawn
+// writer goroutine → read until EOF/Bye.
 // Always Unregisters and Closes on exit (plan invariant #5).
 func (d *daemon) handleConnection(ctx context.Context, conn net.Conn) {
 	conn = ensureCloseOnce(conn)
@@ -372,29 +395,141 @@ func (d *daemon) handleConnection(ctx context.Context, conn net.Conn) {
 		return
 	}
 
-	// Regular consumer registration
+	// HelloAck — credentials_source fields are filled in by the daemon
+	// runner (which knows from the strict resolver) and exposed via the
+	// adapter for forward-compat. v1 leaves them empty here; daemon.Run
+	// passes them through future config if the caller wishes.
+	ack := d.helloAck()
+	handoffLocked := false
+	runtimeGeneration := uint64(0)
+	defer func() {
+		if handoffLocked {
+			d.credentialHandoffMu.Unlock()
+		}
+	}()
+
+	// Runtime credentials use a two-phase additive handshake. The first ack
+	// proves capability before the client sends any secret. Only a successful
+	// CAS and credential ack permit Hub registration.
+	if hello.CredentialMode != "" {
+		if hello.CredentialMode != transport.CredentialModeRuntimeToken || d.cfg.CredentialBroker == nil {
+			ack.Capabilities = nil
+			ack.CredentialGeneration = 0
+			if err := w.WriteJSON(ack); err != nil {
+				d.log.Warn("bus: incompatible helloack write failed", "err", err)
+			}
+			return
+		}
+		d.credentialHandoffMu.Lock()
+		handoffLocked = true
+		// Terminal state may have been published while this Hello waited for a
+		// concurrent credential handoff. Rebuild the ack while holding the gate.
+		ack = d.helloAck()
+		if err := w.WriteJSON(ack); err != nil {
+			d.log.Warn("bus: runtime helloack write failed", "err", err)
+			return
+		}
+		if ack.TerminalReason == transport.ByeReasonRuntimeTokenRejected {
+			return
+		}
+
+		var update transport.CredentialUpdate
+		_ = conn.SetReadDeadline(time.Now().Add(daemonCredentialHandshakeTimeout))
+		if err := r.ReadJSON(&update); err != nil {
+			// Do not include the decoder error: malformed JSON may contain
+			// fragments of the credential.
+			d.log.Warn("bus: malformed runtime credential update")
+			return
+		}
+		_ = conn.SetReadDeadline(time.Time{})
+		if update.Type != transport.FrameTypeCredentialUpdate {
+			_ = w.WriteJSON(transport.CredentialUpdateAck{
+				Type:                 transport.FrameTypeCredentialUpdateAck,
+				Accepted:             false,
+				CredentialGeneration: d.cfg.CredentialBroker.Generation(),
+				ErrorCode:            transport.CredentialErrorInvalid,
+				Error:                "unexpected credential update frame",
+			})
+			return
+		}
+		// Validate registration before applying the credential or sending an
+		// accepted ack. Hub.Register performs the same deterministic compile
+		// before mutating the Hub; this preflight keeps invalid filters from
+		// producing a ready marker after credential negotiation.
+		if _, err := compileMatcher(hello.EventTypes, hello.Filter, hello.SubscribeID); err != nil {
+			update.Token = ""
+			_ = w.WriteJSON(transport.CredentialUpdateAck{
+				Type:                 transport.FrameTypeCredentialUpdateAck,
+				Accepted:             false,
+				CredentialGeneration: d.cfg.CredentialBroker.Generation(),
+				ErrorCode:            transport.CredentialErrorRegistration,
+				Error:                "consumer registration validation failed",
+			})
+			d.log.Warn("bus: runtime consumer registration validation failed")
+			return
+		}
+
+		generation, updateErr := d.cfg.CredentialBroker.Update(update.ExpectedGeneration, update.Token)
+		runtimeGeneration = generation
+		update.Token = ""
+		credentialAck := transport.CredentialUpdateAck{
+			Type:                 transport.FrameTypeCredentialUpdateAck,
+			Accepted:             updateErr == nil,
+			CredentialGeneration: generation,
+		}
+		if updateErr != nil {
+			credentialAck.ErrorCode, credentialAck.Error = classifyCredentialUpdateError(updateErr)
+		}
+		if err := w.WriteJSON(credentialAck); err != nil {
+			d.log.Warn("bus: credential update ack write failed", "err", err)
+			return
+		}
+		if updateErr != nil {
+			d.log.Warn("bus: runtime credential update rejected", "error_code", credentialAck.ErrorCode)
+			return
+		}
+	}
+
+	// Regular consumer registration. Local clients retain the original
+	// register-before-HelloAck ordering; runtime clients were already acked by
+	// the additive handshake above.
 	c, err := d.hub.Register(hello)
 	if err != nil {
 		d.log.Warn("bus: register failed", "err", err, "pid", hello.ConsumerPID)
 		_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "register_failed: " + err.Error()})
 		return
 	}
-
-	// HelloAck — credentials_source fields are filled in by the daemon
-	// runner (which knows from the strict resolver) and exposed via the
-	// adapter for forward-compat. v1 leaves them empty here; daemon.Run
-	// passes them through future config if the caller wishes.
-	idleSecs := int(d.cfg.IdleTimeout / time.Second)
-	if err := w.WriteJSON(transport.HelloAck{
-		Type:            transport.FrameTypeHelloAck,
-		BusPID:          os.Getpid(),
-		SourceState:     "connected", // best-effort; full state machine pushed via SourceState frames
-		StateSource:     "inferred",
-		IdleTimeoutSecs: idleSecs,
-	}); err != nil {
-		d.log.Warn("bus: helloack write failed", "err", err)
+	if handoffLocked {
+		// The runtime broker deliberately keeps the seed pending until the
+		// initiating consumer is registered. This prevents ticket acquisition
+		// (and an immediate 401) from racing ahead of the only connection that
+		// can observe the typed terminal reason.
+		if _, activateErr := d.cfg.CredentialBroker.Activate(runtimeGeneration); activateErr != nil {
+			d.log.Error("bus: runtime credential activation failed")
+			_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "runtime_credential_activation_failed"})
+			d.hub.Unregister(c.ID)
+			return
+		}
+	}
+	// A local/legacy consumer can arrive after terminal publication but after
+	// StopAll took its snapshot. Refuse it synchronously so it cannot observe a
+	// clean EOF for a runtime-token rejection.
+	if terminalReason := d.getTerminalReason(); terminalReason != "" {
+		_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: terminalReason})
+		d.hub.Unregister(c.ID)
 		return
 	}
+	if handoffLocked {
+		d.credentialHandoffMu.Unlock()
+		handoffLocked = false
+	}
+	if hello.CredentialMode == "" {
+		if err := w.WriteJSON(ack); err != nil {
+			d.log.Warn("bus: helloack write failed", "err", err)
+			d.hub.Unregister(c.ID)
+			return
+		}
+	}
 
 	// Writer goroutine pulls from SendCh and writes to the wire.
 	writerDone := make(chan struct{})
@@ -416,6 +551,16 @@ func (d *daemon) handleConnection(ctx context.Context, conn net.Conn) {
 				if !ok {
 					return
 				}
+				// A stop may have arrived while both channels were ready and the
+				// scheduler selected the buffered event. Re-check before starting a
+				// potentially blocking event write so terminal reasons stay prompt.
+				select {
+				case reason := <-c.StopCh:
+					_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: reason})
+					_ = conn.Close()
+					return
+				default:
+				}
 				if err := w.WriteJSON(frame); err != nil {
 					return
 				}
@@ -454,6 +599,51 @@ func (d *daemon) handleConnection(ctx context.Context, conn net.Conn) {
 	_ = ctx // for future use (writer ctx-cancel propagation)
 }
 
+func (d *daemon) helloAck() transport.HelloAck {
+	ack := transport.HelloAck{
+		Type:            transport.FrameTypeHelloAck,
+		BusPID:          os.Getpid(),
+		SourceState:     "connected", // best-effort; full state machine pushed via SourceState frames
+		StateSource:     "inferred",
+		IdleTimeoutSecs: int(d.cfg.IdleTimeout / time.Second),
+	}
+	if d.cfg.CredentialBroker != nil {
+		ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
+		ack.CredentialGeneration = d.cfg.CredentialBroker.Generation()
+	}
+	ack.TerminalReason = d.getTerminalReason()
+	return ack
+}
+
+func (d *daemon) setTerminalReason(reason string) {
+	if d == nil || reason != transport.ByeReasonRuntimeTokenRejected {
+		return
+	}
+	d.terminalMu.Lock()
+	d.terminalReason = reason
+	d.terminalMu.Unlock()
+}
+
+func (d *daemon) getTerminalReason() string {
+	if d == nil {
+		return ""
+	}
+	d.terminalMu.RLock()
+	defer d.terminalMu.RUnlock()
+	return d.terminalReason
+}
+
+func classifyCredentialUpdateError(err error) (string, string) {
+	var conflict *runtimecred.GenerationConflictError
+	if errors.As(err, &conflict) {
+		return transport.CredentialErrorGenerationConflict, conflict.Error()
+	}
+	if errors.Is(err, runtimecred.ErrEmptyToken) || errors.Is(err, runtimecred.ErrTokenTooLarge) {
+		return transport.CredentialErrorInvalid, err.Error()
+	}
+	return transport.CredentialErrorInternal, "runtime credential update failed"
+}
+
 func (d *daemon) handleConsumerStopRPC(w *transport.Writer, r *transport.Reader) {
 	var req transport.ConsumerStopReq
 	if err := r.ReadJSON(&req); err != nil {
@@ -577,38 +767,64 @@ func (d *daemon) triggerShutdown(reason string) {
 //  4. wait for acceptLoop to return so no future consumerWG.Add can occur
 //  5. close all accepted connections and wait for handlers to drain
 //  6. lock + meta cleanup via Run's defers
-func (d *daemon) shutdown(acceptDone <-chan struct{}) {
+func (d *daemon) shutdown(acceptDone <-chan struct{}, reasons ...string) {
 	d.shutdownMu.Lock()
 	defer d.shutdownMu.Unlock()
 	if !d.shuttingDown.CompareAndSwap(false, true) {
 		return
 	}
-	d.hub.Broadcast(transport.Bye{Type: transport.FrameTypeBye, Reason: "shutdown"})
+	reason := normalizedShutdownReason(reasons...)
+	if reason == transport.ByeReasonRuntimeTokenRejected {
+		d.hub.StopAll(reason)
+	} else {
+		d.hub.Broadcast(transport.Bye{Type: transport.FrameTypeBye, Reason: reason})
+	}
 	_ = d.listener.Close()
 	<-acceptDone
-	// Force-close all open IPC connections so any reader goroutine blocked
-	// on Read() returns with a network error and exits cleanly. Without
-	// this the consumerWG never drains and Run hangs forever.
-	d.conns.Range(func(k, _ any) bool {
-		if c, ok := k.(net.Conn); ok {
-			_ = c.Close()
-		}
-		return true
-	})
-	// Give consumers a brief moment to drain final frames before we tear
-	// down their channels.
+	// Let local consumers drain the final Bye before force-closing their
+	// connections. This short grace period is what makes typed shutdown
+	// reasons (notably runtime_token_rejected) observable instead of racing
+	// with EOF. Consumers close their side immediately after reading Bye.
 	doneCh := make(chan struct{})
 	go func() {
 		d.consumerWG.Wait()
 		close(doneCh)
 	}()
 	select {
+	case <-doneCh:
+		return
+	case <-time.After(daemonByeDrainTimeout):
+	}
+
+	// A wedged/old consumer may not close after Bye. Force-close remaining
+	// connections so the daemon still has a bounded shutdown.
+	d.conns.Range(func(k, _ any) bool {
+		if c, ok := k.(net.Conn); ok {
+			_ = c.Close()
+		}
+		return true
+	})
+	select {
 	case <-doneCh:
 	case <-time.After(daemonShutdownTimeout):
 		d.log.Warn("bus: shutdown: consumer goroutines did not drain within 2s")
 	}
 }
 
+func sourceShutdownReason(err error) string {
+	if errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
+		return transport.ByeReasonRuntimeTokenRejected
+	}
+	return "shutdown"
+}
+
+func normalizedShutdownReason(reasons ...string) string {
+	if len(reasons) > 0 && reasons[0] == transport.ByeReasonRuntimeTokenRejected {
+		return transport.ByeReasonRuntimeTokenRejected
+	}
+	return "shutdown"
+}
+
 // signalReady writes a single 'R' byte to the ready pipe (if provided) and
 // closes it. The parent process (busctl/spawn) reads one byte and proceeds.
 func signalReady(p *os.File) {

internal/event/bus/hub.go

@@ -246,6 +246,31 @@ func (h *Hub) StopConsumers(subscribeIDs []string, reason string) []string {
 	return out
 }
 
+// StopAll requests a high-priority graceful close for every live consumer.
+// Unlike Broadcast(Bye), this uses the writer's priority StopCh and therefore
+// cannot sit behind a full event buffer during a terminal source failure.
+func (h *Hub) StopAll(reason string) int {
+	reason = strings.TrimSpace(reason)
+	if reason == "" {
+		reason = "shutdown"
+	}
+	h.mu.RLock()
+	consumers := make([]*Consumer, 0, len(h.consumers))
+	for _, consumer := range h.consumers {
+		consumers = append(consumers, consumer)
+	}
+	h.mu.RUnlock()
+	stopped := 0
+	for _, consumer := range consumers {
+		select {
+		case consumer.StopCh <- reason:
+			stopped++
+		default:
+		}
+	}
+	return stopped
+}
+
 // Unregister removes a consumer by ID and closes its sendCh. Idempotent —
 // calling twice or on an unknown ID is a no-op. closeSend shares the same
 // per-consumer lock as Deliver/Broadcast, so a stale Hub snapshot cannot send

internal/event/bus/hub_test.go

@@ -346,6 +346,27 @@ func TestHub_StopConsumersCoalescesQueuedStop(t *testing.T) {
 	}
 }
 
+func TestCrossPlatformCoverageHubStopAllBypassesFullEventBuffer(t *testing.T) {
+	hub := NewHub(1)
+	consumer, err := hub.Register(transport.Hello{Type: transport.FrameTypeHello})
+	if err != nil {
+		t.Fatal(err)
+	}
+	consumer.SendCh <- transport.Event{Type: transport.FrameTypeEvent}
+
+	if stopped := hub.StopAll(transport.ByeReasonRuntimeTokenRejected); stopped != 1 {
+		t.Fatalf("StopAll() = %d, want 1", stopped)
+	}
+	select {
+	case reason := <-consumer.StopCh:
+		if reason != transport.ByeReasonRuntimeTokenRejected {
+			t.Fatalf("StopCh reason = %q", reason)
+		}
+	default:
+		t.Fatal("terminal stop was blocked behind the full event buffer")
+	}
+}
+
 func TestHub_ConcurrentStopConsumersRegisterUnregister(t *testing.T) {
 	h := NewHub(4)
 	const workers = 32

internal/event/bus/runtime_credential_test.go

@@ -0,0 +1,407 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+
+package bus
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"log/slog"
+	"net"
+	"path/filepath"
+	"strings"
+	"testing"
+	"time"
+
+	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
+)
+
+type runtimeCredentialRejectSource struct {
+	broker *runtimecred.Broker
+}
+
+func (s *runtimeCredentialRejectSource) Start(ctx context.Context, _ dwsevent.EmitFn) error {
+	if _, err := s.broker.Resolve(ctx); err != nil {
+		return err
+	}
+	return runtimecred.ErrRuntimeTokenRejected
+}
+
+func runtimeCredentialDaemon(broker *runtimecred.Broker, logOutput io.Writer) *daemon {
+	if logOutput == nil {
+		logOutput = io.Discard
+	}
+	return &daemon{
+		cfg:      Config{CredentialBroker: broker},
+		log:      slog.New(slog.NewTextHandler(logOutput, nil)),
+		hub:      NewHub(2),
+		started:  time.Now(),
+		idleStop: make(chan struct{}),
+	}
+}
+
+func runRuntimeCredentialConnection(t *testing.T, d *daemon) (net.Conn, *transport.Writer, *transport.Reader, <-chan struct{}) {
+	t.Helper()
+	server, client := net.Pipe()
+	done := make(chan struct{})
+	go func() {
+		d.handleConnection(context.Background(), server)
+		close(done)
+	}()
+	return client, transport.NewWriter(client), transport.NewReader(client), done
+}
+
+func TestCrossPlatformCoverageDaemonRuntimeCredentialHandshakeBeforeRegister(t *testing.T) {
+	broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
+	d := runtimeCredentialDaemon(broker, nil)
+	client, w, r, done := runRuntimeCredentialConnection(t, d)
+	defer client.Close()
+
+	if err := w.WriteJSON(transport.Hello{
+		Type:           transport.FrameTypeHello,
+		ConsumerPID:    42,
+		CredentialMode: transport.CredentialModeRuntimeToken,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var helloAck transport.HelloAck
+	if err := r.ReadJSON(&helloAck); err != nil {
+		t.Fatal(err)
+	}
+	if !hasTransportCapability(helloAck.Capabilities, transport.CapabilityRuntimeTokenV1) || helloAck.CredentialGeneration != 0 {
+		t.Fatalf("hello ack = %#v", helloAck)
+	}
+	if d.hub.Len() != 0 {
+		t.Fatalf("consumer registered before credential update: %d", d.hub.Len())
+	}
+
+	const canary = "ipc-canary-runtime-token"
+	if err := w.WriteJSON(transport.CredentialUpdate{
+		Type:               transport.FrameTypeCredentialUpdate,
+		ExpectedGeneration: helloAck.CredentialGeneration,
+		Token:              canary,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var updateAck transport.CredentialUpdateAck
+	if err := r.ReadJSON(&updateAck); err != nil {
+		t.Fatal(err)
+	}
+	if !updateAck.Accepted || updateAck.CredentialGeneration != 1 {
+		t.Fatalf("credential ack = %#v", updateAck)
+	}
+	deadline := time.Now().Add(time.Second)
+	for d.hub.Len() != 1 && time.Now().Before(deadline) {
+		time.Sleep(time.Millisecond)
+	}
+	if d.hub.Len() != 1 {
+		t.Fatal("consumer was not registered after credential ack")
+	}
+	if token, err := broker.Resolve(context.Background()); err != nil || token != canary {
+		t.Fatalf("broker did not resolve installed runtime token: %v", err)
+	}
+
+	if err := w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "done"}); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("connection did not close")
+	}
+}
+
+func TestCrossPlatformCoverageDaemonCompatibleBusRotatesRuntimeCredentialAcrossConnections(t *testing.T) {
+	broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
+	d := runtimeCredentialDaemon(broker, nil)
+
+	handshake := func(token string, wantHelloGeneration, wantAckGeneration uint64) {
+		t.Helper()
+		client, w, r, done := runRuntimeCredentialConnection(t, d)
+		if err := w.WriteJSON(transport.Hello{
+			Type:           transport.FrameTypeHello,
+			CredentialMode: transport.CredentialModeRuntimeToken,
+		}); err != nil {
+			t.Fatal(err)
+		}
+		var helloAck transport.HelloAck
+		if err := r.ReadJSON(&helloAck); err != nil {
+			t.Fatal(err)
+		}
+		if helloAck.CredentialGeneration != wantHelloGeneration {
+			t.Fatalf("hello generation = %d, want %d", helloAck.CredentialGeneration, wantHelloGeneration)
+		}
+		if err := w.WriteJSON(transport.CredentialUpdate{
+			Type:               transport.FrameTypeCredentialUpdate,
+			ExpectedGeneration: helloAck.CredentialGeneration,
+			Token:              token,
+		}); err != nil {
+			t.Fatal(err)
+		}
+		var updateAck transport.CredentialUpdateAck
+		if err := r.ReadJSON(&updateAck); err != nil {
+			t.Fatal(err)
+		}
+		if !updateAck.Accepted || updateAck.CredentialGeneration != wantAckGeneration {
+			t.Fatalf("credential ack accepted=%v generation=%d, want true/%d", updateAck.Accepted, updateAck.CredentialGeneration, wantAckGeneration)
+		}
+		if err := w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "done"}); err != nil {
+			t.Fatal(err)
+		}
+		_ = client.Close()
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Fatal("connection did not close")
+		}
+	}
+
+	handshake("runtime-token-a", 0, 1)
+	if token, err := broker.Resolve(context.Background()); err != nil || token != "runtime-token-a" {
+		t.Fatalf("broker did not retain first token: %v", err)
+	}
+	handshake("runtime-token-b", 1, 2)
+	if token, err := broker.Resolve(context.Background()); err != nil || token != "runtime-token-b" {
+		t.Fatalf("broker did not rotate to second token: %v", err)
+	}
+}
+
+func TestCrossPlatformCoverageDaemonRuntimeCredentialMissingCapabilityDoesNotRegister(t *testing.T) {
+	d := runtimeCredentialDaemon(nil, nil)
+	client, w, r, done := runRuntimeCredentialConnection(t, d)
+	defer client.Close()
+	if err := w.WriteJSON(transport.Hello{
+		Type:           transport.FrameTypeHello,
+		CredentialMode: transport.CredentialModeRuntimeToken,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var ack transport.HelloAck
+	if err := r.ReadJSON(&ack); err != nil {
+		t.Fatal(err)
+	}
+	if hasTransportCapability(ack.Capabilities, transport.CapabilityRuntimeTokenV1) {
+		t.Fatalf("unsupported daemon advertised capability: %#v", ack)
+	}
+	if d.hub.Len() != 0 {
+		t.Fatalf("unsupported daemon registered consumer: %d", d.hub.Len())
+	}
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("unsupported connection did not close")
+	}
+}
+
+func TestCrossPlatformCoverageDaemonTerminalStateRejectsLateLegacyConsumer(t *testing.T) {
+	d := runtimeCredentialDaemon(nil, nil)
+	d.setTerminalReason(transport.ByeReasonRuntimeTokenRejected)
+	client, w, r, done := runRuntimeCredentialConnection(t, d)
+	defer client.Close()
+
+	if err := w.WriteJSON(transport.Hello{Type: transport.FrameTypeHello, ConsumerPID: 7}); err != nil {
+		t.Fatal(err)
+	}
+	var bye transport.Bye
+	if err := r.ReadJSON(&bye); err != nil {
+		t.Fatal(err)
+	}
+	if bye.Type != transport.FrameTypeBye || bye.Reason != transport.ByeReasonRuntimeTokenRejected {
+		t.Fatalf("late consumer frame = %#v", bye)
+	}
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("late consumer connection did not close")
+	}
+	if d.hub.Len() != 0 {
+		t.Fatalf("late terminal consumer remained registered: %d", d.hub.Len())
+	}
+}
+
+func TestCrossPlatformCoverageDaemonRuntimeCredentialConflictDoesNotLeakOrRegister(t *testing.T) {
+	broker := runtimecred.New(runtimecred.Config{})
+	if _, err := broker.Update(0, "installed-secret"); err != nil {
+		t.Fatal(err)
+	}
+	var logs bytes.Buffer
+	d := runtimeCredentialDaemon(broker, &logs)
+	client, w, r, done := runRuntimeCredentialConnection(t, d)
+	defer client.Close()
+	if err := w.WriteJSON(transport.Hello{
+		Type:           transport.FrameTypeHello,
+		CredentialMode: transport.CredentialModeRuntimeToken,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var helloAck transport.HelloAck
+	if err := r.ReadJSON(&helloAck); err != nil {
+		t.Fatal(err)
+	}
+	const canary = "rejected-canary-secret"
+	if err := w.WriteJSON(transport.CredentialUpdate{
+		Type:               transport.FrameTypeCredentialUpdate,
+		ExpectedGeneration: 0,
+		Token:              canary,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var updateAck transport.CredentialUpdateAck
+	if err := r.ReadJSON(&updateAck); err != nil {
+		t.Fatal(err)
+	}
+	if updateAck.Accepted || updateAck.ErrorCode != transport.CredentialErrorGenerationConflict || updateAck.CredentialGeneration != 1 {
+		t.Fatalf("conflict ack = %#v", updateAck)
+	}
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("rejected connection did not close")
+	}
+	if strings.Contains(updateAck.Error, canary) || strings.Contains(logs.String(), canary) {
+		t.Fatal("credential appeared in acknowledgement or logs")
+	}
+	if d.hub.Len() != 0 {
+		t.Fatalf("rejected consumer registered: %d", d.hub.Len())
+	}
+}
+
+func TestCrossPlatformCoverageDaemonRuntimeCredentialInvalidFilterRejectedBeforeUpdate(t *testing.T) {
+	broker := runtimecred.New(runtimecred.Config{})
+	d := runtimeCredentialDaemon(broker, nil)
+	client, w, r, done := runRuntimeCredentialConnection(t, d)
+	defer client.Close()
+	if err := w.WriteJSON(transport.Hello{
+		Type:           transport.FrameTypeHello,
+		CredentialMode: transport.CredentialModeRuntimeToken,
+		Filter:         "[",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var helloAck transport.HelloAck
+	if err := r.ReadJSON(&helloAck); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.WriteJSON(transport.CredentialUpdate{
+		Type:               transport.FrameTypeCredentialUpdate,
+		ExpectedGeneration: helloAck.CredentialGeneration,
+		Token:              "filter-canary-secret",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var updateAck transport.CredentialUpdateAck
+	if err := r.ReadJSON(&updateAck); err != nil {
+		t.Fatal(err)
+	}
+	if updateAck.Accepted || updateAck.ErrorCode != transport.CredentialErrorRegistration {
+		t.Fatalf("invalid filter ack = %#v", updateAck)
+	}
+	if broker.Generation() != 0 || d.hub.Len() != 0 {
+		t.Fatalf("invalid filter mutated state: generation=%d consumers=%d", broker.Generation(), d.hub.Len())
+	}
+	select {
+	case <-done:
+	case <-time.After(time.Second):
+		t.Fatal("invalid filter connection did not close")
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeCredentialShutdownReason(t *testing.T) {
+	if got := sourceShutdownReason(runtimecred.ErrRuntimeTokenRejected); got != transport.ByeReasonRuntimeTokenRejected {
+		t.Fatalf("runtime source shutdown reason = %q", got)
+	}
+	if got := sourceShutdownReason(errors.New("local source failed")); got != "shutdown" {
+		t.Fatalf("local source shutdown reason = %q", got)
+	}
+	if got := normalizedShutdownReason(transport.ByeReasonRuntimeTokenRejected); got != transport.ByeReasonRuntimeTokenRejected {
+		t.Fatalf("normalized runtime shutdown reason = %q", got)
+	}
+	for _, reasons := range [][]string{nil, {"peer-controlled"}} {
+		if got := normalizedShutdownReason(reasons...); got != "shutdown" {
+			t.Fatalf("normalized untrusted shutdown reason = %q", got)
+		}
+	}
+}
+
+func TestCrossPlatformCoverageDaemonRuntimeCredentialSourceRejectionBroadcastsTypedBye(t *testing.T) {
+	skipOnWindows(t, "uses Unix socket dial")
+	workDir := shortTempDir(t)
+	sockPath := filepath.Join(workDir, "bus.sock")
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	broker := runtimecred.New(runtimecred.Config{RequireSeed: true, RequireActivation: true})
+	runDone := make(chan error, 1)
+	go func() {
+		runDone <- Run(ctx, Config{
+			WorkDir:          workDir,
+			IPCEndpoint:      sockPath,
+			ClientID:         "runtime-client",
+			Edition:          "open",
+			SourceKind:       dwsevent.SourceKindPersonalStream,
+			IdentityHash:     "0123456789abcdef",
+			SourceID:         "open",
+			Source:           &runtimeCredentialRejectSource{broker: broker},
+			CredentialBroker: broker,
+		})
+	}()
+	waitForFile(t, sockPath, 2*time.Second)
+
+	conn, err := transport.Dial(sockPath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer conn.Close()
+	r, w := transport.NewReader(conn), transport.NewWriter(conn)
+	if err := w.WriteJSON(transport.Hello{
+		Type:           transport.FrameTypeHello,
+		CredentialMode: transport.CredentialModeRuntimeToken,
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var helloAck transport.HelloAck
+	if err := r.ReadJSON(&helloAck); err != nil {
+		t.Fatal(err)
+	}
+	if err := w.WriteJSON(transport.CredentialUpdate{
+		Type:               transport.FrameTypeCredentialUpdate,
+		ExpectedGeneration: helloAck.CredentialGeneration,
+		Token:              "runtime-rejected-canary",
+	}); err != nil {
+		t.Fatal(err)
+	}
+	var updateAck transport.CredentialUpdateAck
+	if err := r.ReadJSON(&updateAck); err != nil || !updateAck.Accepted {
+		t.Fatalf("credential update ack = %#v, %v", updateAck, err)
+	}
+
+	var bye transport.Bye
+	if err := r.ReadJSON(&bye); err != nil {
+		t.Fatalf("read typed shutdown: %v", err)
+	}
+	if bye.Type != transport.FrameTypeBye || bye.Reason != transport.ByeReasonRuntimeTokenRejected {
+		t.Fatalf("shutdown frame = %#v", bye)
+	}
+	_ = conn.Close()
+	select {
+	case err := <-runDone:
+		if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
+			t.Fatalf("Run() error = %v", err)
+		}
+	case <-time.After(3 * time.Second):
+		t.Fatal("runtime credential bus did not stop")
+	}
+}
+
+func hasTransportCapability(capabilities []string, want string) bool {
+	for _, capability := range capabilities {
+		if capability == want {
+			return true
+		}
+	}
+	return false
+}

internal/event/consume/cross_platform_coverage_event_core_agent_test.go

@@ -0,0 +1,101 @@
+package consume
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"strings"
+	"testing"
+
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
+)
+
+func eventCoreCredentialReader(t *testing.T, frame any) *transport.Reader {
+	t.Helper()
+	var buffer bytes.Buffer
+	if err := transport.NewWriter(&buffer).WriteJSON(frame); err != nil {
+		t.Fatal(err)
+	}
+	return transport.NewReader(&buffer)
+}
+
+func TestCrossPlatformCoverageEventCoreRuntimeNegotiationEdges(t *testing.T) {
+	capable := transport.HelloAck{Capabilities: []string{transport.CapabilityRuntimeTokenV1}}
+	if err := negotiateRuntimeToken(
+		transport.NewWriter(io.Discard),
+		transport.NewReader(strings.NewReader("")),
+		transport.HelloAck{TerminalReason: transport.ByeReasonRuntimeTokenRejected},
+		"token",
+	); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
+		t.Fatalf("terminal hello error = %v", err)
+	}
+
+	wantWriteErr := errors.New("write failed")
+	if err := negotiateRuntimeToken(
+		transport.NewWriter(errorWriter{err: wantWriteErr}),
+		transport.NewReader(strings.NewReader("")),
+		capable,
+		"token",
+	); !errors.Is(err, wantWriteErr) {
+		t.Fatalf("credential write error = %v", err)
+	}
+
+	if err := negotiateRuntimeToken(
+		transport.NewWriter(io.Discard),
+		transport.NewReader(strings.NewReader("")),
+		capable,
+		"token",
+	); !errors.Is(err, io.EOF) {
+		t.Fatalf("credential ack read error = %v", err)
+	}
+
+	if err := negotiateRuntimeToken(
+		transport.NewWriter(io.Discard),
+		eventCoreCredentialReader(t, transport.Heartbeat{Type: transport.FrameTypeHeartbeat}),
+		capable,
+		"token",
+	); err == nil || !strings.Contains(err.Error(), "unexpected runtime credential response") {
+		t.Fatalf("unexpected credential frame error = %v", err)
+	}
+
+	if err := negotiateRuntimeToken(
+		transport.NewWriter(io.Discard),
+		eventCoreCredentialReader(t, transport.CredentialUpdateAck{
+			Type:      transport.FrameTypeCredentialUpdateAck,
+			Accepted:  false,
+			ErrorCode: transport.CredentialErrorRuntimeRejected,
+		}),
+		capable,
+		"token",
+	); !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
+		t.Fatalf("runtime rejected ack error = %v", err)
+	}
+
+	for _, code := range []string{
+		transport.CredentialErrorGenerationConflict,
+		transport.CredentialErrorInvalid,
+		transport.CredentialErrorRegistration,
+		transport.CredentialErrorRuntimeRejected,
+		transport.CredentialErrorInternal,
+	} {
+		if got := safeCredentialErrorCode(code); got != code {
+			t.Fatalf("safeCredentialErrorCode(%q) = %q", code, got)
+		}
+	}
+	if got := safeCredentialErrorCode("peer-controlled"); got != transport.CredentialErrorInternal {
+		t.Fatalf("unknown credential error code = %q", got)
+	}
+}
+
+func TestCrossPlatformCoverageEventCoreRunManyHandshakeFailure(t *testing.T) {
+	bus := newManyFakeBus(901, nil)
+	installManyDiscover(t, bus)
+	cfg := manyTestConfig(io.Discard, io.Discard)
+	cfg.RuntimeToken = "runtime-token"
+	err := RunMany(context.Background(), cfg, manyTestSpecs())
+	if !errors.Is(err, ErrRuntimeTokenUnsupported) || !strings.Contains(err.Error(), "runtime credential handshake") {
+		t.Fatalf("RunMany handshake error = %v", err)
+	}
+}

internal/event/consume/run.go

@@ -21,9 +21,11 @@ import (
 	"io"
 	"net"
 	"os"
+	"strings"
 	"time"
 
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
 )
 
@@ -45,6 +47,11 @@ type Config struct {
 	// personal_stream.
 	SpawnExtraArgs []string
 
+	// RuntimeToken is a host-supplied credential handed to a compatible bus
+	// only after capability negotiation over owner-only local IPC. It is never
+	// included in dry-run output, child argv, environment, or persisted state.
+	RuntimeToken string `json:"-" yaml:"-"`
+
 	// EventTypes / Filter / Compact are forwarded to the bus via Hello
 	// for server-side pushdown filtering.
 	EventTypes  []string
@@ -132,6 +139,36 @@ type Config struct {
 
 var discoverBus = busctl.Discover
 
+var (
+	ErrRuntimeTokenUnsupported = errors.New("consume: event bus does not support secure runtime-token handoff")
+	ErrRuntimeTokenUpdate      = errors.New("consume: event bus rejected runtime-token update")
+)
+
+// RuntimeTokenUnsupportedError is returned before the token is sent when the
+// connected bus lacks the runtime_token_v1 capability.
+type RuntimeTokenUnsupportedError struct {
+	BusPID int
+}
+
+func (e *RuntimeTokenUnsupportedError) Error() string {
+	return fmt.Sprintf("consume: running event bus (pid %d) does not support secure runtime-token handoff; let existing consumers exit, inspect with `dws event status --as user --format json`, preview cleanup with `dws event stop --as user --all --dry-run`, then confirm with `dws event stop --as user --all --yes` and retry", e.BusPID)
+}
+
+func (e *RuntimeTokenUnsupportedError) Unwrap() error { return ErrRuntimeTokenUnsupported }
+
+// RuntimeTokenUpdateError reports a rejected credential CAS without carrying
+// either the credential or peer-provided free-form error text.
+type RuntimeTokenUpdateError struct {
+	Code       string
+	Generation uint64
+}
+
+func (e *RuntimeTokenUpdateError) Error() string {
+	return fmt.Sprintf("consume: event bus rejected runtime-token update (code=%s, generation=%d)", e.Code, e.Generation)
+}
+
+func (e *RuntimeTokenUpdateError) Unwrap() error { return ErrRuntimeTokenUpdate }
+
 // Run dials the bus (forking one if necessary), sends Hello, and writes
 // each received Event frame as one NDJSON line to stdout. Blocks until
 // ctx is cancelled, MaxEvents is reached, the bus sends Bye, or the
@@ -144,6 +181,7 @@ func Run(ctx context.Context, cfg Config) error {
 	if cfg.WorkDir == "" || cfg.IPCEndpoint == "" || cfg.ClientID == "" {
 		return errors.New("consume: WorkDir, IPCEndpoint, and ClientID are required")
 	}
+	cfg.RuntimeToken = strings.TrimSpace(cfg.RuntimeToken)
 	if cfg.Stdout == nil {
 		cfg.Stdout = os.Stdout
 	}
@@ -225,6 +263,9 @@ func Run(ctx context.Context, cfg Config) error {
 		SubscribeID: cfg.SubscribeID,
 		Compact:     cfg.Compact,
 	}
+	if cfg.RuntimeToken != "" {
+		hello.CredentialMode = transport.CredentialModeRuntimeToken
+	}
 	if err := w.WriteJSON(hello); err != nil {
 		return fmt.Errorf("consume: write hello: %w", err)
 	}
@@ -236,6 +277,9 @@ func Run(ctx context.Context, cfg Config) error {
 	if ack.Type != transport.FrameTypeHelloAck {
 		return fmt.Errorf("consume: unexpected first frame type %q", ack.Type)
 	}
+	if err := negotiateRuntimeToken(w, r, ack, cfg.RuntimeToken); err != nil {
+		return err
+	}
 	if !cfg.Quiet {
 		// Contract: a fixed ready line on stderr BEFORE any stdout event.
 		// Parents block on stderr until this appears, then read stdout.
@@ -331,6 +375,9 @@ func Run(ctx context.Context, cfg Config) error {
 		case transport.FrameTypeBye:
 			var bye transport.Bye
 			_ = json.Unmarshal(raw, &bye)
+			if bye.Reason == transport.ByeReasonRuntimeTokenRejected {
+				return fmt.Errorf("consume: %w", runtimecred.ErrRuntimeTokenRejected)
+			}
 			if !cfg.Quiet {
 				fmt.Fprintf(cfg.Stderr, "[event] bus closing: %s\n", bye.Reason)
 			}
@@ -350,6 +397,62 @@ func Run(ctx context.Context, cfg Config) error {
 	}
 }
 
+func negotiateRuntimeToken(w *transport.Writer, r *transport.Reader, ack transport.HelloAck, token string) error {
+	if token == "" {
+		return nil
+	}
+	if ack.TerminalReason == transport.ByeReasonRuntimeTokenRejected {
+		return fmt.Errorf("consume: %w", runtimecred.ErrRuntimeTokenRejected)
+	}
+	if !hasCapability(ack.Capabilities, transport.CapabilityRuntimeTokenV1) {
+		return &RuntimeTokenUnsupportedError{BusPID: ack.BusPID}
+	}
+	if err := w.WriteJSON(transport.CredentialUpdate{
+		Type:               transport.FrameTypeCredentialUpdate,
+		ExpectedGeneration: ack.CredentialGeneration,
+		Token:              token,
+	}); err != nil {
+		return fmt.Errorf("consume: write runtime credential update: %w", err)
+	}
+	var updateAck transport.CredentialUpdateAck
+	if err := r.ReadJSON(&updateAck); err != nil {
+		return fmt.Errorf("consume: read runtime credential update ack: %w", err)
+	}
+	if updateAck.Type != transport.FrameTypeCredentialUpdateAck {
+		return errors.New("consume: unexpected runtime credential response frame")
+	}
+	if !updateAck.Accepted {
+		code := safeCredentialErrorCode(updateAck.ErrorCode)
+		if code == transport.CredentialErrorRuntimeRejected {
+			return fmt.Errorf("consume: %w", runtimecred.ErrRuntimeTokenRejected)
+		}
+		return &RuntimeTokenUpdateError{Code: code, Generation: updateAck.CredentialGeneration}
+	}
+	return nil
+}
+
+func safeCredentialErrorCode(code string) string {
+	switch code {
+	case transport.CredentialErrorGenerationConflict,
+		transport.CredentialErrorInvalid,
+		transport.CredentialErrorRegistration,
+		transport.CredentialErrorRuntimeRejected,
+		transport.CredentialErrorInternal:
+		return code
+	default:
+		return transport.CredentialErrorInternal
+	}
+}
+
+func hasCapability(capabilities []string, want string) bool {
+	for _, capability := range capabilities {
+		if capability == want {
+			return true
+		}
+	}
+	return false
+}
+
 // closeOnContext spawns a goroutine that closes conn when ctx is done.
 // This unblocks any pending Read on conn so the main loop can return.
 func closeOnContext(ctx context.Context, conn net.Conn) {

internal/event/consume/run_many.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
 )
 
@@ -53,6 +54,7 @@ func RunMany(ctx context.Context, cfg Config, specs []ConsumerSpec) error {
 	if cfg.WorkDir == "" || cfg.IPCEndpoint == "" || cfg.ClientID == "" {
 		return errors.New("consume: WorkDir, IPCEndpoint, and ClientID are required")
 	}
+	cfg.RuntimeToken = strings.TrimSpace(cfg.RuntimeToken)
 	if len(specs) < 2 {
 		return errors.New("consume: RunMany requires at least two consumers")
 	}
@@ -127,14 +129,18 @@ func RunMany(ctx context.Context, cfg Config, specs []ConsumerSpec) error {
 		}
 		sessions = append(sessions, session)
 		closeOnContext(ctx, session.conn)
-		if err := session.w.WriteJSON(transport.Hello{
+		hello := transport.Hello{
 			Type:        transport.FrameTypeHello,
 			ConsumerPID: os.Getpid(),
 			EventTypes:  spec.EventTypes,
 			Filter:      spec.Filter,
 			SubscribeID: spec.SubscribeID,
 			Compact:     cfg.Compact,
-		}); err != nil {
+		}
+		if cfg.RuntimeToken != "" {
+			hello.CredentialMode = transport.CredentialModeRuntimeToken
+		}
+		if err := session.w.WriteJSON(hello); err != nil {
 			return fmt.Errorf("consume: write hello for %s: %w", spec.EventKey, err)
 		}
 		if err := session.r.ReadJSON(&session.ack); err != nil {
@@ -143,9 +149,16 @@ func RunMany(ctx context.Context, cfg Config, specs []ConsumerSpec) error {
 		if session.ack.Type != transport.FrameTypeHelloAck {
 			return fmt.Errorf("consume: unexpected first frame type %q for %s", session.ack.Type, spec.EventKey)
 		}
+		// Verify that every connection reached the same bus before handing a
+		// runtime credential to it. Discovery is expected to converge on one
+		// daemon, but a stale endpoint/race must not propagate the host token to
+		// an unrelated process merely so we can report the PID mismatch later.
 		if len(sessions) > 1 && session.ack.BusPID != sessions[0].ack.BusPID {
 			return fmt.Errorf("consume: consumers connected to different bus processes (%d and %d)", sessions[0].ack.BusPID, session.ack.BusPID)
 		}
+		if err := negotiateRuntimeToken(session.w, session.r, session.ack, cfg.RuntimeToken); err != nil {
+			return fmt.Errorf("consume: runtime credential handshake for %s: %w", spec.EventKey, err)
+		}
 	}
 
 	if !cfg.Quiet {
@@ -225,6 +238,9 @@ func RunMany(ctx context.Context, cfg Config, specs []ConsumerSpec) error {
 			case transport.FrameTypeBye:
 				var bye transport.Bye
 				_ = json.Unmarshal(frame.raw, &bye)
+				if bye.Reason == transport.ByeReasonRuntimeTokenRejected {
+					return fmt.Errorf("consume: %w", runtimecred.ErrRuntimeTokenRejected)
+				}
 				if bye.Reason == transport.ByeReasonSubscriptionStopped {
 					delete(active, frame.index)
 					_ = sessions[frame.index].conn.Close()

internal/event/consume/run_many_test.go

@@ -16,6 +16,7 @@ import (
 	"time"
 
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
 )
 
@@ -37,23 +38,25 @@ func (b *synchronizedBuffer) String() string {
 }
 
 type manyFakeBus struct {
-	client  net.Conn
-	server  net.Conn
-	hello   chan transport.Hello
-	acked   chan struct{}
-	ackGate <-chan struct{}
-	ack     transport.HelloAck
-	send    chan any
+	client           net.Conn
+	server           net.Conn
+	hello            chan transport.Hello
+	credentialUpdate chan transport.CredentialUpdate
+	acked            chan struct{}
+	ackGate          <-chan struct{}
+	ack              transport.HelloAck
+	send             chan any
 }
 
 func newManyFakeBus(busPID int, ackGate <-chan struct{}) *manyFakeBus {
 	client, server := net.Pipe()
 	f := &manyFakeBus{
-		client:  client,
-		server:  server,
-		hello:   make(chan transport.Hello, 1),
-		acked:   make(chan struct{}),
-		ackGate: ackGate,
+		client:           client,
+		server:           server,
+		hello:            make(chan transport.Hello, 1),
+		credentialUpdate: make(chan transport.CredentialUpdate, 1),
+		acked:            make(chan struct{}),
+		ackGate:          ackGate,
 		ack: transport.HelloAck{
 			Type:            transport.FrameTypeHelloAck,
 			BusPID:          busPID,
@@ -85,6 +88,23 @@ func (f *manyFakeBus) serve() {
 	if err := w.WriteJSON(f.ack); err != nil {
 		return
 	}
+	if hello.CredentialMode == transport.CredentialModeRuntimeToken {
+		if !hasCapability(f.ack.Capabilities, transport.CapabilityRuntimeTokenV1) {
+			return
+		}
+		var update transport.CredentialUpdate
+		if err := r.ReadJSON(&update); err != nil {
+			return
+		}
+		f.credentialUpdate <- update
+		if err := w.WriteJSON(transport.CredentialUpdateAck{
+			Type:                 transport.FrameTypeCredentialUpdateAck,
+			Accepted:             true,
+			CredentialGeneration: f.ack.CredentialGeneration + 1,
+		}); err != nil {
+			return
+		}
+	}
 	close(f.acked)
 	go func() {
 		for {
@@ -221,6 +241,91 @@ func TestRunManyWaitsForAllConsumersAndStopsOneAtATime(t *testing.T) {
 	}
 }
 
+func TestCrossPlatformCoverageRunManyRuntimeTokenUsesEachConnectionGeneration(t *testing.T) {
+	const canary = "many-runtime-canary"
+	busA := newManyFakeBus(111, nil)
+	busB := newManyFakeBus(111, nil)
+	busA.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
+	busB.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
+	busA.ack.CredentialGeneration = 2
+	busB.ack.CredentialGeneration = 8
+	installManyDiscover(t, busA, busB)
+
+	cfg := manyTestConfig(io.Discard, io.Discard)
+	cfg.RuntimeToken = canary
+	done := make(chan error, 1)
+	go func() { done <- RunMany(context.Background(), cfg, manyTestSpecs()) }()
+
+	updateA := <-busA.credentialUpdate
+	updateB := <-busB.credentialUpdate
+	if updateA.Token != canary || updateA.ExpectedGeneration != 2 {
+		t.Fatal("first connection used the wrong credential or generation")
+	}
+	if updateB.Token != canary || updateB.ExpectedGeneration != 8 {
+		t.Fatal("second connection used the wrong credential or generation")
+	}
+	busA.send <- transport.Bye{Type: transport.FrameTypeBye, Reason: "shutdown"}
+	select {
+	case err := <-done:
+		if err != nil {
+			t.Fatal(err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("RunMany did not stop")
+	}
+}
+
+func TestCrossPlatformCoverageRunManyRuntimeTokenRejectsDifferentBusBeforeSecondCredential(t *testing.T) {
+	const canary = "many-mismatched-bus-canary"
+	busA := newManyFakeBus(111, nil)
+	busB := newManyFakeBus(222, nil)
+	busA.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
+	busB.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
+	installManyDiscover(t, busA, busB)
+
+	cfg := manyTestConfig(io.Discard, io.Discard)
+	cfg.RuntimeToken = canary
+	err := RunMany(context.Background(), cfg, manyTestSpecs())
+	if err == nil || !strings.Contains(err.Error(), "different bus processes") {
+		t.Fatalf("RunMany() error = %v", err)
+	}
+	if update := <-busA.credentialUpdate; update.Token != canary {
+		t.Fatal("first bus did not receive the negotiated credential")
+	}
+	select {
+	case update := <-busB.credentialUpdate:
+		t.Fatalf("mismatched second bus received credential: generation=%d", update.ExpectedGeneration)
+	default:
+	}
+}
+
+func TestCrossPlatformCoverageRunManyRuntimeTokenRejectedByeReturnsTypedError(t *testing.T) {
+	busA := newManyFakeBus(333, nil)
+	busB := newManyFakeBus(333, nil)
+	busA.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
+	busB.ack.Capabilities = []string{transport.CapabilityRuntimeTokenV1}
+	installManyDiscover(t, busA, busB)
+
+	cfg := manyTestConfig(io.Discard, io.Discard)
+	cfg.RuntimeToken = "many-runtime-rejected-canary"
+	done := make(chan error, 1)
+	go func() { done <- RunMany(context.Background(), cfg, manyTestSpecs()) }()
+	<-busA.credentialUpdate
+	<-busB.credentialUpdate
+	busA.send <- transport.Bye{
+		Type:   transport.FrameTypeBye,
+		Reason: transport.ByeReasonRuntimeTokenRejected,
+	}
+	select {
+	case err := <-done:
+		if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
+			t.Fatalf("RunMany() error = %v", err)
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("RunMany did not return runtime credential rejection")
+	}
+}
+
 func TestRunManyMaxEventsIsSharedAcrossConsumers(t *testing.T) {
 	busA := newManyFakeBus(202, nil)
 	busB := newManyFakeBus(202, nil)

internal/event/consume/runtime_credential_test.go

@@ -0,0 +1,242 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+
+package consume
+
+import (
+	"bytes"
+	"context"
+	"errors"
+	"io"
+	"net"
+	"strings"
+	"testing"
+	"time"
+
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/busctl"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
+)
+
+func installRuntimeCredentialDiscover(t *testing.T, serve func(net.Conn)) {
+	t.Helper()
+	oldDiscover := discoverBus
+	done := make(chan struct{})
+	discoverBus = func(busctl.DiscoverConfig) (net.Conn, error) {
+		client, server := net.Pipe()
+		go func() {
+			defer close(done)
+			defer server.Close()
+			serve(server)
+		}()
+		return client, nil
+	}
+	t.Cleanup(func() {
+		discoverBus = oldDiscover
+		select {
+		case <-done:
+		case <-time.After(time.Second):
+			t.Error("fake runtime credential bus did not stop")
+		}
+	})
+}
+
+func TestCrossPlatformCoverageRunRuntimeTokenNegotiatesBeforeReady(t *testing.T) {
+	const canary = "consume-runtime-canary"
+	helloSeen := make(chan transport.Hello, 1)
+	updateSeen := make(chan transport.CredentialUpdate, 1)
+	installRuntimeCredentialDiscover(t, func(conn net.Conn) {
+		r, w := transport.NewReader(conn), transport.NewWriter(conn)
+		var hello transport.Hello
+		if err := r.ReadJSON(&hello); err != nil {
+			return
+		}
+		helloSeen <- hello
+		_ = w.WriteJSON(transport.HelloAck{
+			Type:                 transport.FrameTypeHelloAck,
+			BusPID:               71,
+			Capabilities:         []string{transport.CapabilityRuntimeTokenV1},
+			CredentialGeneration: 3,
+		})
+		var update transport.CredentialUpdate
+		if err := r.ReadJSON(&update); err != nil {
+			return
+		}
+		updateSeen <- update
+		_ = w.WriteJSON(transport.CredentialUpdateAck{
+			Type:                 transport.FrameTypeCredentialUpdateAck,
+			Accepted:             true,
+			CredentialGeneration: 4,
+		})
+		_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "done"})
+	})
+
+	var stderr bytes.Buffer
+	cfg := validRunConfig()
+	cfg.RuntimeToken = "  " + canary + "  "
+	cfg.Stderr = &stderr
+	if err := Run(context.Background(), cfg); err != nil {
+		t.Fatal(err)
+	}
+	hello := <-helloSeen
+	if hello.CredentialMode != transport.CredentialModeRuntimeToken {
+		t.Fatalf("credential mode = %q", hello.CredentialMode)
+	}
+	update := <-updateSeen
+	if update.ExpectedGeneration != 3 || update.Token != canary {
+		t.Fatal("credential update used the wrong token or generation")
+	}
+	if !strings.Contains(stderr.String(), "[event] ready bus_pid=71") {
+		t.Fatalf("ready marker missing: %s", stderr.String())
+	}
+}
+
+func TestCrossPlatformCoverageRunRuntimeTokenMissingCapabilityFailsBeforeSendingSecretOrReady(t *testing.T) {
+	const canary = "unsupported-canary-secret"
+	peerBytes := make(chan string, 1)
+	installRuntimeCredentialDiscover(t, func(conn net.Conn) {
+		r, w := transport.NewReader(conn), transport.NewWriter(conn)
+		var hello transport.Hello
+		if err := r.ReadJSON(&hello); err != nil {
+			return
+		}
+		_ = w.WriteJSON(transport.HelloAck{Type: transport.FrameTypeHelloAck, BusPID: 72})
+		raw, _ := io.ReadAll(conn)
+		peerBytes <- string(raw)
+	})
+
+	var stderr bytes.Buffer
+	cfg := validRunConfig()
+	cfg.RuntimeToken = canary
+	cfg.Stderr = &stderr
+	err := Run(context.Background(), cfg)
+	if !errors.Is(err, ErrRuntimeTokenUnsupported) {
+		t.Fatalf("Run error = %v", err)
+	}
+	if strings.Contains(err.Error(), canary) {
+		t.Fatal("unsupported-bus error contained runtime token")
+	}
+	for _, recoveryStep := range []string{
+		"dws event status --as user --format json",
+		"dws event stop --as user --all --dry-run",
+		"dws event stop --as user --all --yes",
+	} {
+		if !strings.Contains(err.Error(), recoveryStep) {
+			t.Fatalf("unsupported-bus error missing recovery step %q: %v", recoveryStep, err)
+		}
+	}
+	if strings.Contains(stderr.String(), "[event] ready") {
+		t.Fatal("ready marker was written before capability negotiation succeeded")
+	}
+	if got := <-peerBytes; strings.Contains(got, canary) || got != "" {
+		t.Fatal("client sent data after unsupported capability acknowledgement")
+	}
+}
+
+func TestCrossPlatformCoverageRunRuntimeTokenRejectedDoesNotSurfacePeerText(t *testing.T) {
+	const canary = "rejected-canary-secret"
+	installRuntimeCredentialDiscover(t, func(conn net.Conn) {
+		r, w := transport.NewReader(conn), transport.NewWriter(conn)
+		var hello transport.Hello
+		if err := r.ReadJSON(&hello); err != nil {
+			return
+		}
+		_ = w.WriteJSON(transport.HelloAck{
+			Type:         transport.FrameTypeHelloAck,
+			Capabilities: []string{transport.CapabilityRuntimeTokenV1},
+		})
+		var update transport.CredentialUpdate
+		if err := r.ReadJSON(&update); err != nil {
+			return
+		}
+		_ = w.WriteJSON(transport.CredentialUpdateAck{
+			Type:      transport.FrameTypeCredentialUpdateAck,
+			Accepted:  false,
+			ErrorCode: "malicious-code-" + update.Token,
+			Error:     "malicious echo " + update.Token,
+		})
+	})
+
+	cfg := validRunConfig()
+	cfg.RuntimeToken = canary
+	err := Run(context.Background(), cfg)
+	if !errors.Is(err, ErrRuntimeTokenUpdate) {
+		t.Fatalf("Run error = %v", err)
+	}
+	if strings.Contains(err.Error(), canary) {
+		t.Fatal("credential update error contained peer-provided token text")
+	}
+}
+
+func TestCrossPlatformCoverageRuntimeTokenNeverAppearsInDryRun(t *testing.T) {
+	const canary = "dry-run-canary-secret"
+	var output bytes.Buffer
+	cfg := validRunConfig()
+	cfg.RuntimeToken = canary
+	PrintDryRun(&output, cfg)
+	if strings.Contains(output.String(), canary) || strings.Contains(output.String(), "RuntimeToken") {
+		t.Fatal("dry-run output contained runtime-token data")
+	}
+}
+
+func TestCrossPlatformCoverageRunWhitespaceRuntimeTokenUsesLegacyHandshake(t *testing.T) {
+	helloSeen := make(chan transport.Hello, 1)
+	installRuntimeCredentialDiscover(t, func(conn net.Conn) {
+		r, w := transport.NewReader(conn), transport.NewWriter(conn)
+		var hello transport.Hello
+		if err := r.ReadJSON(&hello); err != nil {
+			return
+		}
+		helloSeen <- hello
+		_ = w.WriteJSON(transport.HelloAck{Type: transport.FrameTypeHelloAck})
+		_ = w.WriteJSON(transport.Bye{Type: transport.FrameTypeBye, Reason: "done"})
+	})
+	cfg := validRunConfig()
+	cfg.RuntimeToken = "   "
+	if err := Run(context.Background(), cfg); err != nil {
+		t.Fatal(err)
+	}
+	if hello := <-helloSeen; hello.CredentialMode != "" {
+		t.Fatalf("whitespace token enabled mode %q", hello.CredentialMode)
+	}
+}
+
+func TestCrossPlatformCoverageRunRuntimeTokenRejectedByeReturnsTypedError(t *testing.T) {
+	installRuntimeCredentialDiscover(t, func(conn net.Conn) {
+		r, w := transport.NewReader(conn), transport.NewWriter(conn)
+		var hello transport.Hello
+		if err := r.ReadJSON(&hello); err != nil {
+			return
+		}
+		_ = w.WriteJSON(transport.HelloAck{
+			Type:         transport.FrameTypeHelloAck,
+			BusPID:       73,
+			Capabilities: []string{transport.CapabilityRuntimeTokenV1},
+		})
+		var update transport.CredentialUpdate
+		if err := r.ReadJSON(&update); err != nil {
+			return
+		}
+		_ = w.WriteJSON(transport.CredentialUpdateAck{
+			Type:                 transport.FrameTypeCredentialUpdateAck,
+			Accepted:             true,
+			CredentialGeneration: 1,
+		})
+		_ = w.WriteJSON(transport.Bye{
+			Type:   transport.FrameTypeBye,
+			Reason: transport.ByeReasonRuntimeTokenRejected,
+		})
+	})
+
+	var stderr bytes.Buffer
+	cfg := validRunConfig()
+	cfg.RuntimeToken = "runtime-rejected-canary"
+	cfg.Stderr = &stderr
+	err := Run(context.Background(), cfg)
+	if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
+		t.Fatalf("Run() error = %v", err)
+	}
+	if strings.Contains(stderr.String(), "reason: bus_shutdown") {
+		t.Fatalf("runtime rejection was reported as a successful exit: %s", stderr.String())
+	}
+}

internal/event/consume/validate.go

@@ -113,8 +113,9 @@ func IsValidationError(err error) bool {
 // human-readable block. Called by Run when cfg.DryRun is true. Format
 // avoids JSON so users can `dws event consume --dry-run | head` cleanly.
 //
-// Secret-bearing fields are never present in Config (credentials never
-// reach this layer), so no redaction is required here.
+// RuntimeToken is the only secret-bearing Config field and is deliberately
+// not read or rendered here. Keep this function allowlist-based: never switch
+// it to generic struct serialization.
 func PrintDryRun(w io.Writer, cfg Config) {
 	if w == nil {
 		return

internal/event/runtimecred/broker.go

@@ -0,0 +1,300 @@
+// 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 runtimecred provides an in-memory credential broker for event bus
+// processes. Runtime credentials are never persisted by this package.
+package runtimecred
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"strings"
+	"sync"
+)
+
+// DefaultMaxTokenBytes bounds credentials accepted over the local IPC
+// channel. Access tokens are normally only a few KiB; the larger limit leaves
+// ample room for future token formats while avoiding accidental large secret
+// allocations.
+const DefaultMaxTokenBytes = 64 << 10
+
+var (
+	ErrEmptyToken              = errors.New("runtime credential: token is empty")
+	ErrTokenTooLarge           = errors.New("runtime credential: token exceeds size limit")
+	ErrCredentialUnavailable   = errors.New("runtime credential: no credential resolver is available")
+	ErrLocalRefreshUnavailable = errors.New("runtime credential: no local refresh callback is available")
+	ErrRuntimeTokenRejected    = &RuntimeTokenRejectedError{}
+)
+
+// RuntimeTokenRejectedError means the currently installed runtime token was
+// rejected and no newer runtime token is available. It deliberately carries
+// no token or server response data so it is safe to surface to users and logs.
+type RuntimeTokenRejectedError struct{}
+
+func (*RuntimeTokenRejectedError) Error() string {
+	return "event runtime token was rejected; retry with a fresh host credential"
+}
+
+func (*RuntimeTokenRejectedError) Is(target error) bool {
+	_, ok := target.(*RuntimeTokenRejectedError)
+	return ok
+}
+
+// GenerationConflictError reports a failed compare-and-swap update.
+type GenerationConflictError struct {
+	Expected uint64
+	Actual   uint64
+}
+
+func (e *GenerationConflictError) Error() string {
+	return fmt.Sprintf("runtime credential: generation conflict (expected %d, actual %d)", e.Expected, e.Actual)
+}
+
+// ResolveFunc resolves the existing local OAuth credential when no runtime
+// credential has been installed.
+type ResolveFunc func(context.Context) (string, error)
+
+// RefreshFunc refreshes a rejected local OAuth credential. It is never called
+// after a runtime credential has been installed.
+type RefreshFunc func(context.Context, string) (string, error)
+
+type Config struct {
+	LocalResolve ResolveFunc
+	LocalRefresh RefreshFunc
+	RequireSeed  bool
+	// RequireActivation keeps the first installed runtime credential pending
+	// until Activate is called. Detached buses use it to register the consumer
+	// before ticket acquisition can emit or fail.
+	RequireActivation bool
+	MaxTokenBytes     int
+}
+
+// Broker holds at most one runtime credential. All state, including the
+// credential generation, is process-local and concurrency-safe.
+type Broker struct {
+	localResolve      ResolveFunc
+	localRefresh      RefreshFunc
+	requireSeed       bool
+	requireActivation bool
+	maxBytes          int
+
+	mu         sync.Mutex
+	token      string
+	generation uint64
+	active     bool
+	changed    chan struct{}
+}
+
+func New(cfg Config) *Broker {
+	maxBytes := cfg.MaxTokenBytes
+	if maxBytes <= 0 {
+		maxBytes = DefaultMaxTokenBytes
+	}
+	return &Broker{
+		localResolve:      cfg.LocalResolve,
+		localRefresh:      cfg.LocalRefresh,
+		requireSeed:       cfg.RequireSeed,
+		requireActivation: cfg.RequireActivation,
+		maxBytes:          maxBytes,
+		active:            !cfg.RequireActivation,
+		changed:           make(chan struct{}),
+	}
+}
+
+// Generation returns the current runtime credential generation. Generation 0
+// means that no runtime credential has been installed yet.
+func (b *Broker) Generation() uint64 {
+	if b == nil {
+		return 0
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.generation
+}
+
+// Update atomically installs token when expectedGeneration matches the
+// current generation. Reinstalling the same token is idempotent, including
+// when another concurrent writer already installed it.
+func (b *Broker) Update(expectedGeneration uint64, token string) (uint64, error) {
+	if b == nil {
+		return 0, ErrCredentialUnavailable
+	}
+	normalized, err := b.validate(token)
+	if err != nil {
+		return b.Generation(), err
+	}
+
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if b.token == normalized {
+		return b.generation, nil
+	}
+	if expectedGeneration != b.generation {
+		return b.generation, &GenerationConflictError{Expected: expectedGeneration, Actual: b.generation}
+	}
+	b.token = normalized
+	b.generation++
+	if b.active {
+		b.signalChangedLocked()
+	}
+	return b.generation, nil
+}
+
+// Activate publishes a pending first runtime credential after the bus has
+// registered the initiating consumer. It is an idempotent generation-checked
+// no-op for brokers that do not require activation.
+func (b *Broker) Activate(expectedGeneration uint64) (uint64, error) {
+	if b == nil {
+		return 0, ErrCredentialUnavailable
+	}
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	if expectedGeneration != b.generation {
+		return b.generation, &GenerationConflictError{Expected: expectedGeneration, Actual: b.generation}
+	}
+	if b.token == "" {
+		return b.generation, ErrCredentialUnavailable
+	}
+	if b.active {
+		return b.generation, nil
+	}
+	b.active = true
+	b.signalChangedLocked()
+	return b.generation, nil
+}
+
+// Resolve returns the runtime credential when installed. In RequireSeed mode
+// it waits until Update installs one; otherwise it preserves the existing
+// local resolver behavior until a runtime credential arrives.
+func (b *Broker) Resolve(ctx context.Context) (string, error) {
+	if b == nil {
+		return "", ErrCredentialUnavailable
+	}
+	for {
+		token, wait, requireSeed, active := b.snapshot()
+		if token != "" && active {
+			return token, nil
+		}
+		if requireSeed || (token != "" && !active) {
+			select {
+			case <-ctx.Done():
+				return "", ctx.Err()
+			case <-wait:
+				continue
+			}
+		}
+		if b.localResolve == nil {
+			return "", ErrCredentialUnavailable
+		}
+		local, err := b.localResolve(ctx)
+		if err != nil {
+			return "", err
+		}
+		if runtime, _, _, runtimeActive := b.snapshot(); runtime != "" && runtimeActive {
+			return runtime, nil
+		}
+		return validateResolved(local)
+	}
+}
+
+// RefreshRejected returns a newer runtime token if one was installed after
+// rejectedToken was used. If the installed runtime token itself was rejected,
+// it returns RuntimeTokenRejectedError and never invokes local OAuth refresh.
+func (b *Broker) RefreshRejected(ctx context.Context, rejectedToken string) (string, error) {
+	if b == nil {
+		return "", ErrCredentialUnavailable
+	}
+	for {
+		token, wait, requireSeed, active := b.snapshot()
+		if token != "" && active {
+			if token != rejectedToken {
+				return token, nil
+			}
+			return "", &RuntimeTokenRejectedError{}
+		}
+		if requireSeed || (token != "" && !active) {
+			select {
+			case <-ctx.Done():
+				return "", ctx.Err()
+			case <-wait:
+				continue
+			}
+		}
+		if b.localRefresh == nil {
+			return "", ErrLocalRefreshUnavailable
+		}
+		refreshed, err := b.localRefresh(ctx, rejectedToken)
+		if err != nil {
+			return "", err
+		}
+		if runtime, _, _, runtimeActive := b.snapshot(); runtime != "" && runtimeActive {
+			if runtime != rejectedToken {
+				return runtime, nil
+			}
+			return "", &RuntimeTokenRejectedError{}
+		}
+		return validateResolved(refreshed)
+	}
+}
+
+// ClassifyRejectedAfterRetry is called when a token returned by
+// RefreshRejected was itself rejected. When a still newer runtime credential
+// is already installed, superseded is true so the source may reconnect and
+// resolve that generation without a second in-attempt retry. When the rejected
+// token is still current, the fixed typed rejection is returned. With no
+// runtime credential installed it preserves local OAuth behavior by returning
+// (false, nil).
+func (b *Broker) ClassifyRejectedAfterRetry(rejectedToken string) (superseded bool, err error) {
+	if b == nil {
+		return false, ErrCredentialUnavailable
+	}
+	token, _, _, active := b.snapshot()
+	if token == "" || !active {
+		return false, nil
+	}
+	if token != strings.TrimSpace(rejectedToken) {
+		return true, nil
+	}
+	return false, &RuntimeTokenRejectedError{}
+}
+
+func (b *Broker) snapshot() (string, <-chan struct{}, bool, bool) {
+	b.mu.Lock()
+	defer b.mu.Unlock()
+	return b.token, b.changed, b.requireSeed, b.active
+}
+
+func (b *Broker) signalChangedLocked() {
+	close(b.changed)
+	b.changed = make(chan struct{})
+}
+
+func (b *Broker) validate(token string) (string, error) {
+	normalized := strings.TrimSpace(token)
+	if normalized == "" {
+		return "", ErrEmptyToken
+	}
+	if len(normalized) > b.maxBytes {
+		return "", ErrTokenTooLarge
+	}
+	return normalized, nil
+}
+
+func validateResolved(token string) (string, error) {
+	token = strings.TrimSpace(token)
+	if token == "" {
+		return "", ErrCredentialUnavailable
+	}
+	return token, nil
+}

internal/event/runtimecred/broker_test.go

@@ -0,0 +1,203 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+
+package runtimecred
+
+import (
+	"context"
+	"errors"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+)
+
+func TestCrossPlatformCoverageBrokerLocalFallbackAndRuntimePrecedence(t *testing.T) {
+	var localCalls atomic.Int32
+	b := New(Config{LocalResolve: func(context.Context) (string, error) {
+		localCalls.Add(1)
+		return "local", nil
+	}})
+	if got, err := b.Resolve(context.Background()); err != nil || got != "local" {
+		t.Fatalf("local Resolve = %q, %v", got, err)
+	}
+	gen, err := b.Update(0, " runtime ")
+	if err != nil || gen != 1 {
+		t.Fatalf("Update = %d, %v", gen, err)
+	}
+	if got, err := b.Resolve(context.Background()); err != nil || got != "runtime" {
+		t.Fatalf("runtime Resolve = %q, %v", got, err)
+	}
+	if localCalls.Load() != 1 {
+		t.Fatalf("local resolver calls = %d", localCalls.Load())
+	}
+}
+
+func TestCrossPlatformCoverageBrokerRequireSeedWaitsAndCancels(t *testing.T) {
+	b := New(Config{RequireSeed: true})
+	ctx, cancel := context.WithCancel(context.Background())
+	done := make(chan error, 1)
+	go func() { _, err := b.Resolve(ctx); done <- err }()
+	cancel()
+	if err := <-done; !errors.Is(err, context.Canceled) {
+		t.Fatalf("Resolve error = %v", err)
+	}
+
+	resolved := make(chan string, 1)
+	go func() {
+		token, _ := b.Resolve(context.Background())
+		resolved <- token
+	}()
+	select {
+	case <-resolved:
+		t.Fatal("Resolve returned before seed")
+	case <-time.After(20 * time.Millisecond):
+	}
+	if _, err := b.Update(0, "seed"); err != nil {
+		t.Fatal(err)
+	}
+	if got := <-resolved; got != "seed" {
+		t.Fatalf("Resolve = %q", got)
+	}
+}
+
+func TestCrossPlatformCoverageBrokerUpdateCASIdempotenceAndConcurrentConflict(t *testing.T) {
+	b := New(Config{})
+	if gen, err := b.Update(0, "same"); err != nil || gen != 1 {
+		t.Fatalf("first update = %d, %v", gen, err)
+	}
+	if gen, err := b.Update(0, "same"); err != nil || gen != 1 {
+		t.Fatalf("idempotent stale update = %d, %v", gen, err)
+	}
+	if _, err := b.Update(0, "different"); err == nil {
+		t.Fatal("expected generation conflict")
+	} else {
+		var conflict *GenerationConflictError
+		if !errors.As(err, &conflict) || conflict.Actual != 1 {
+			t.Fatalf("conflict = %#v, %v", conflict, err)
+		}
+	}
+
+	b = New(Config{})
+	start := make(chan struct{})
+	var successes atomic.Int32
+	var conflicts atomic.Int32
+	var wg sync.WaitGroup
+	for _, token := range []string{"a", "b"} {
+		wg.Add(1)
+		go func(token string) {
+			defer wg.Done()
+			<-start
+			_, err := b.Update(0, token)
+			if err == nil {
+				successes.Add(1)
+				return
+			}
+			var conflict *GenerationConflictError
+			if errors.As(err, &conflict) {
+				conflicts.Add(1)
+			}
+		}(token)
+	}
+	close(start)
+	wg.Wait()
+	if successes.Load() != 1 || conflicts.Load() != 1 {
+		t.Fatalf("successes=%d conflicts=%d", successes.Load(), conflicts.Load())
+	}
+}
+
+func TestCrossPlatformCoverageBrokerRefreshRejectedRuntimeNeverFallsBack(t *testing.T) {
+	var refreshCalls atomic.Int32
+	b := New(Config{LocalRefresh: func(context.Context, string) (string, error) {
+		refreshCalls.Add(1)
+		return "local-new", nil
+	}})
+	if _, err := b.Update(0, "runtime-a"); err != nil {
+		t.Fatal(err)
+	}
+	if _, err := b.RefreshRejected(context.Background(), "runtime-a"); !errors.Is(err, ErrRuntimeTokenRejected) {
+		t.Fatalf("same token refresh error = %v", err)
+	}
+	if refreshCalls.Load() != 0 {
+		t.Fatalf("local refresh called %d times", refreshCalls.Load())
+	}
+	if _, err := b.Update(1, "runtime-b"); err != nil {
+		t.Fatal(err)
+	}
+	if got, err := b.RefreshRejected(context.Background(), "runtime-a"); err != nil || got != "runtime-b" {
+		t.Fatalf("rotated refresh = %q, %v", got, err)
+	}
+	if superseded, err := b.ClassifyRejectedAfterRetry("runtime-b"); superseded || !errors.Is(err, ErrRuntimeTokenRejected) {
+		t.Fatalf("current retry rejection = superseded %v, error %v", superseded, err)
+	}
+	if _, err := b.Update(2, "runtime-c"); err != nil {
+		t.Fatal(err)
+	}
+	if superseded, err := b.ClassifyRejectedAfterRetry("runtime-b"); !superseded || err != nil {
+		t.Fatalf("newer generation classification = superseded %v, error %v", superseded, err)
+	}
+	localOnly := New(Config{})
+	if superseded, err := localOnly.ClassifyRejectedAfterRetry("local"); superseded || err != nil {
+		t.Fatalf("local-only classification = superseded %v, error %v", superseded, err)
+	}
+}
+
+func TestCrossPlatformCoverageBrokerDefersSeedUntilActivation(t *testing.T) {
+	b := New(Config{RequireSeed: true, RequireActivation: true})
+	ctx, cancel := context.WithCancel(context.Background())
+	defer cancel()
+
+	resolved := make(chan string, 1)
+	errs := make(chan error, 1)
+	go func() {
+		token, err := b.Resolve(ctx)
+		if err != nil {
+			errs <- err
+			return
+		}
+		resolved <- token
+	}()
+
+	generation, err := b.Update(0, "runtime-a")
+	if err != nil || generation != 1 {
+		t.Fatalf("Update() = generation %d, err %v", generation, err)
+	}
+	select {
+	case token := <-resolved:
+		t.Fatalf("Resolve() returned before activation: %q", token)
+	case err := <-errs:
+		t.Fatalf("Resolve() failed before activation: %v", err)
+	case <-time.After(20 * time.Millisecond):
+	}
+
+	if _, err := b.Activate(0); err == nil {
+		t.Fatal("Activate() with stale generation unexpectedly succeeded")
+	}
+	if activeGeneration, err := b.Activate(generation); err != nil || activeGeneration != generation {
+		t.Fatalf("Activate() = generation %d, err %v", activeGeneration, err)
+	}
+	select {
+	case token := <-resolved:
+		if token != "runtime-a" {
+			t.Fatalf("Resolve() token = %q", token)
+		}
+	case err := <-errs:
+		t.Fatalf("Resolve() failed after activation: %v", err)
+	case <-time.After(time.Second):
+		t.Fatal("Resolve() remained blocked after activation")
+	}
+}
+
+func TestCrossPlatformCoverageBrokerRejectsInvalidTokensWithoutEcho(t *testing.T) {
+	b := New(Config{MaxTokenBytes: 4})
+	for _, token := range []string{"   ", "secret-token"} {
+		_, err := b.Update(0, token)
+		if err == nil {
+			t.Fatalf("Update(%q) succeeded", token)
+		}
+		if strings.Contains(err.Error(), token) {
+			t.Fatal("validation error contained rejected token")
+		}
+	}
+}

internal/event/runtimecred/cross_platform_coverage_event_core_agent_test.go

@@ -0,0 +1,150 @@
+package runtimecred
+
+import (
+	"context"
+	"errors"
+	"testing"
+	"time"
+)
+
+func TestCrossPlatformCoverageEventCoreBrokerEdges(t *testing.T) {
+	rejected := &RuntimeTokenRejectedError{}
+	if rejected.Error() == "" || !errors.Is(rejected, ErrRuntimeTokenRejected) || rejected.Is(errors.New("other")) {
+		t.Fatal("runtime rejection error contract failed")
+	}
+	conflict := &GenerationConflictError{Expected: 2, Actual: 3}
+	if conflict.Error() == "" {
+		t.Fatal("generation conflict error is empty")
+	}
+
+	var nilBroker *Broker
+	if nilBroker.Generation() != 0 {
+		t.Fatal("nil broker generation is non-zero")
+	}
+	if _, err := nilBroker.Update(0, "token"); !errors.Is(err, ErrCredentialUnavailable) {
+		t.Fatalf("nil Update error = %v", err)
+	}
+	if _, err := nilBroker.Activate(0); !errors.Is(err, ErrCredentialUnavailable) {
+		t.Fatalf("nil Activate error = %v", err)
+	}
+	if _, err := nilBroker.Resolve(context.Background()); !errors.Is(err, ErrCredentialUnavailable) {
+		t.Fatalf("nil Resolve error = %v", err)
+	}
+	if _, err := nilBroker.RefreshRejected(context.Background(), "token"); !errors.Is(err, ErrCredentialUnavailable) {
+		t.Fatalf("nil RefreshRejected error = %v", err)
+	}
+	if superseded, err := nilBroker.ClassifyRejectedAfterRetry("token"); superseded || !errors.Is(err, ErrCredentialUnavailable) {
+		t.Fatalf("nil classification = %v, %v", superseded, err)
+	}
+
+	b := New(Config{})
+	if _, err := b.Activate(0); !errors.Is(err, ErrCredentialUnavailable) {
+		t.Fatalf("empty Activate error = %v", err)
+	}
+	if _, err := b.Update(0, "token"); err != nil {
+		t.Fatal(err)
+	}
+	if generation, err := b.Activate(1); err != nil || generation != 1 {
+		t.Fatalf("active Activate = %d, %v", generation, err)
+	}
+
+	if _, err := New(Config{}).Resolve(context.Background()); !errors.Is(err, ErrCredentialUnavailable) {
+		t.Fatalf("missing local resolver error = %v", err)
+	}
+	wantResolveErr := errors.New("resolve failed")
+	if _, err := New(Config{LocalResolve: func(context.Context) (string, error) {
+		return "", wantResolveErr
+	}}).Resolve(context.Background()); !errors.Is(err, wantResolveErr) {
+		t.Fatalf("local resolve error = %v", err)
+	}
+	if _, err := New(Config{LocalResolve: func(context.Context) (string, error) {
+		return "   ", nil
+	}}).Resolve(context.Background()); !errors.Is(err, ErrCredentialUnavailable) {
+		t.Fatalf("empty resolved credential error = %v", err)
+	}
+	var racingResolve *Broker
+	racingResolve = New(Config{LocalResolve: func(context.Context) (string, error) {
+		_, err := racingResolve.Update(0, "runtime-wins")
+		return "local", err
+	}})
+	if token, err := racingResolve.Resolve(context.Background()); err != nil || token != "runtime-wins" {
+		t.Fatalf("runtime precedence after resolve = %q, %v", token, err)
+	}
+
+	seed := New(Config{RequireSeed: true})
+	canceled, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, err := seed.Resolve(canceled); !errors.Is(err, context.Canceled) {
+		t.Fatalf("seed cancellation = %v", err)
+	}
+
+	if _, err := New(Config{}).RefreshRejected(context.Background(), "local"); !errors.Is(err, ErrLocalRefreshUnavailable) {
+		t.Fatalf("missing local refresh error = %v", err)
+	}
+	wantRefreshErr := errors.New("refresh failed")
+	if _, err := New(Config{LocalRefresh: func(context.Context, string) (string, error) {
+		return "", wantRefreshErr
+	}}).RefreshRejected(context.Background(), "local"); !errors.Is(err, wantRefreshErr) {
+		t.Fatalf("local refresh error = %v", err)
+	}
+	if _, err := New(Config{LocalRefresh: func(context.Context, string) (string, error) {
+		return " ", nil
+	}}).RefreshRejected(context.Background(), "local"); !errors.Is(err, ErrCredentialUnavailable) {
+		t.Fatalf("empty refreshed credential error = %v", err)
+	}
+
+	var racingRefresh *Broker
+	racingRefresh = New(Config{LocalRefresh: func(context.Context, string) (string, error) {
+		_, err := racingRefresh.Update(0, "runtime-new")
+		return "local-new", err
+	}})
+	if token, err := racingRefresh.RefreshRejected(context.Background(), "runtime-old"); err != nil || token != "runtime-new" {
+		t.Fatalf("runtime precedence after refresh = %q, %v", token, err)
+	}
+	var sameRefresh *Broker
+	sameRefresh = New(Config{LocalRefresh: func(context.Context, string) (string, error) {
+		_, err := sameRefresh.Update(0, "same")
+		return "local-new", err
+	}})
+	if _, err := sameRefresh.RefreshRejected(context.Background(), "same"); !errors.Is(err, ErrRuntimeTokenRejected) {
+		t.Fatalf("same runtime after refresh error = %v", err)
+	}
+
+	pending := New(Config{RequireSeed: true, RequireActivation: true})
+	generation, err := pending.Update(0, "pending")
+	if err != nil {
+		t.Fatal(err)
+	}
+	waitCtx, waitCancel := context.WithCancel(context.Background())
+	waitCancel()
+	if _, err := pending.RefreshRejected(waitCtx, "pending"); !errors.Is(err, context.Canceled) {
+		t.Fatalf("pending refresh cancellation = %v", err)
+	}
+	refreshDone := make(chan error, 1)
+	go func() {
+		_, err := pending.RefreshRejected(context.Background(), "pending")
+		refreshDone <- err
+	}()
+	// Let RefreshRejected take the pending credential's changed-channel path
+	// before activation publishes it.
+	time.Sleep(10 * time.Millisecond)
+	if _, err := pending.Activate(generation); err != nil {
+		t.Fatal(err)
+	}
+	select {
+	case err := <-refreshDone:
+		if !errors.Is(err, ErrRuntimeTokenRejected) {
+			t.Fatalf("activated pending refresh error = %v", err)
+		}
+	case <-time.After(time.Second):
+		t.Fatal("pending refresh did not resume")
+	}
+
+	inactive := New(Config{RequireActivation: true})
+	if _, err := inactive.Update(0, "inactive"); err != nil {
+		t.Fatal(err)
+	}
+	if superseded, err := inactive.ClassifyRejectedAfterRetry("other"); superseded || err != nil {
+		t.Fatalf("inactive classification = %v, %v", superseded, err)
+	}
+}

internal/event/source/cross_platform_coverage_event_core_agent_test.go

@@ -0,0 +1,40 @@
+package source
+
+import (
+	"context"
+	"io"
+	"net/http"
+	"strings"
+	"testing"
+)
+
+func TestCrossPlatformCoverageEventCorePersonalSupersededRetry(t *testing.T) {
+	source, err := NewPersonal(PersonalConfig{
+		AccessToken: "stale",
+		ForceRefreshToken: func(context.Context, string) (string, error) {
+			return "rotated", nil
+		},
+		ClassifyRetryReject: func(token string) (bool, error) {
+			if token != "rotated" {
+				t.Fatalf("classified token = %q", token)
+			}
+			return true, nil
+		},
+		ClientID:  "client",
+		SourceID:  "source",
+		TicketURL: "https://ticket.test",
+		HTTPClient: &http.Client{Transport: roundTripFunc(func(*http.Request) (*http.Response, error) {
+			return &http.Response{
+				StatusCode: http.StatusUnauthorized,
+				Body:       io.NopCloser(strings.NewReader("rejected")),
+				Header:     make(http.Header),
+			}, nil
+		})},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if _, err := source.fetchTicket(context.Background()); !isRetryablePersonalError(err) {
+		t.Fatalf("superseded retry error = %v", err)
+	}
+}

internal/event/source/personal.go

@@ -45,6 +45,7 @@ type PersonalConfig struct {
 	AccessToken         string
 	AccessTokenProvider AccessTokenProvider
 	ForceRefreshToken   ForceRefreshTokenFn
+	ClassifyRetryReject RetryRejectClassifier
 	ClientID            string
 	ClientSecret        string
 	SourceID            string
@@ -66,6 +67,12 @@ type AccessTokenProvider func(context.Context) (string, error)
 // when nil a 401 stays fatal, matching the previous behavior.
 type ForceRefreshTokenFn func(ctx context.Context, rejectedToken string) (string, error)
 
+// RetryRejectClassifier classifies a 401 from the one refreshed-token retry.
+// superseded means a newer credential is already available and the outer
+// reconnect loop should start a fresh attempt; err is a terminal typed
+// rejection. A nil callback preserves the historical local OAuth behavior.
+type RetryRejectClassifier func(rejectedToken string) (superseded bool, err error)
+
 type PersonalSource struct {
 	cfg     PersonalConfig
 	machine *Machine
@@ -223,7 +230,19 @@ func (s *PersonalSource) fetchTicket(ctx context.Context) (*ticketResponse, erro
 			return nil, refreshErr
 		}
 		// Retry once with the freshly rotated token; a second 401 stays fatal.
-		ticket, _, err = s.fetchTicketAttempt(ctx, refreshed)
+		var retryStatus int
+		var retryErr error
+		ticket, retryStatus, retryErr = s.fetchTicketAttempt(ctx, refreshed)
+		if retryStatus == http.StatusUnauthorized && s.cfg.ClassifyRetryReject != nil {
+			superseded, classifyErr := s.cfg.ClassifyRetryReject(refreshed)
+			if classifyErr != nil {
+				return nil, classifyErr
+			}
+			if superseded {
+				return nil, retryPersonal(retryErr)
+			}
+		}
+		err = retryErr
 	}
 	return ticket, err
 }

internal/event/source/portal_ticket_refresh_test.go

@@ -14,6 +14,7 @@ import (
 	"time"
 
 	dwsevent "github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/runtimecred"
 	"github.com/gorilla/websocket"
 	"github.com/open-dingtalk/dingtalk-stream-sdk-go/payload"
 )
@@ -316,6 +317,46 @@ func TestCrossPlatformCoveragePersonalFetchTicket401RefreshFailureStaysFatal(t *
 	}
 }
 
+func TestCrossPlatformCoveragePersonalRuntimeTokenAtoBSecond401IsTyped(t *testing.T) {
+	broker := runtimecred.New(runtimecred.Config{})
+	if _, err := broker.Update(0, "runtime-a"); err != nil {
+		t.Fatal(err)
+	}
+	var attemptTokens []string
+	src, err := NewPersonal(PersonalConfig{
+		AccessTokenProvider: broker.Resolve,
+		ForceRefreshToken:   broker.RefreshRejected,
+		ClassifyRetryReject: broker.ClassifyRejectedAfterRetry,
+		ClientID:            "client",
+		SourceID:            "source",
+		TicketURL:           "https://ticket.test",
+		HTTPClient: &http.Client{Transport: roundTripFunc(func(req *http.Request) (*http.Response, error) {
+			token := req.Header.Get("x-user-access-token")
+			attemptTokens = append(attemptTokens, token)
+			if token == "runtime-a" {
+				if _, err := broker.Update(1, "runtime-b"); err != nil {
+					t.Fatalf("rotate runtime credential: %v", err)
+				}
+			}
+			return &http.Response{
+				StatusCode: http.StatusUnauthorized,
+				Body:       io.NopCloser(strings.NewReader("must-not-surface")),
+				Header:     make(http.Header),
+			}, nil
+		})},
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = src.fetchTicket(context.Background())
+	if !errors.Is(err, runtimecred.ErrRuntimeTokenRejected) {
+		t.Fatalf("fetchTicket() error = %v", err)
+	}
+	if len(attemptTokens) != 2 || attemptTokens[0] != "runtime-a" || attemptTokens[1] != "runtime-b" {
+		t.Fatalf("attempt tokens = %v", attemptTokens)
+	}
+}
+
 // brokenBody simulates a response body that fails mid-read, e.g. the server
 // closing the connection before the error payload is fully written.
 type brokenBody struct{}

internal/event/transport/protocol.go

@@ -26,16 +26,28 @@ import (
 type FrameType string
 
 const (
-	FrameTypeHello            FrameType = "hello"              // consume → bus
-	FrameTypeHelloAck         FrameType = "hello_ack"          // bus → consume
-	FrameTypeEvent            FrameType = "event"              // bus → consume
-	FrameTypeHeartbeat        FrameType = "heartbeat"          // bidirectional
-	FrameTypeSourceState      FrameType = "source_state"       // bus → consume
-	FrameTypeBye              FrameType = "bye"                // bidirectional
-	FrameTypeStatusReq        FrameType = "status_req"         // consume/ad-hoc → bus
-	FrameTypeStatusResp       FrameType = "status_resp"        // bus → consume/ad-hoc
-	FrameTypeConsumerStopReq  FrameType = "consumer_stop_req"  // ad-hoc → bus
-	FrameTypeConsumerStopResp FrameType = "consumer_stop_resp" // bus → ad-hoc
+	FrameTypeHello               FrameType = "hello"                 // consume → bus
+	FrameTypeHelloAck            FrameType = "hello_ack"             // bus → consume
+	FrameTypeEvent               FrameType = "event"                 // bus → consume
+	FrameTypeHeartbeat           FrameType = "heartbeat"             // bidirectional
+	FrameTypeSourceState         FrameType = "source_state"          // bus → consume
+	FrameTypeBye                 FrameType = "bye"                   // bidirectional
+	FrameTypeStatusReq           FrameType = "status_req"            // consume/ad-hoc → bus
+	FrameTypeStatusResp          FrameType = "status_resp"           // bus → consume/ad-hoc
+	FrameTypeConsumerStopReq     FrameType = "consumer_stop_req"     // ad-hoc → bus
+	FrameTypeConsumerStopResp    FrameType = "consumer_stop_resp"    // bus → ad-hoc
+	FrameTypeCredentialUpdate    FrameType = "credential_update"     // consume → bus
+	FrameTypeCredentialUpdateAck FrameType = "credential_update_ack" // bus → consume
+)
+
+// CredentialMode declares that a consumer needs an additive credential
+// negotiation before it can register for events. The zero value preserves the
+// original protocol.
+type CredentialMode string
+
+const (
+	CredentialModeRuntimeToken CredentialMode = "runtime_token"
+	CapabilityRuntimeTokenV1                  = "runtime_token_v1"
 )
 
 // Hello is the first frame a consumer sends after dialing the bus. The bus
@@ -51,7 +63,8 @@ type Hello struct {
 	// Role distinguishes a real consumer (registered for events) from an
 	// ad-hoc tooling connection (status/list/stop). Ad-hoc connections do
 	// NOT register with the Hub.
-	Role HelloRole `json:"role,omitempty"`
+	Role           HelloRole      `json:"role,omitempty"`
+	CredentialMode CredentialMode `json:"credential_mode,omitempty"`
 }
 
 // HelloRole tags the purpose of a Hello connection.
@@ -76,6 +89,38 @@ type HelloAck struct {
 	ClientIDSource     string    `json:"client_id_source"`            // auth.CredentialSource string
 	ClientSecretSource string    `json:"client_secret_source"`        // auth.CredentialSource string
 	IdleTimeoutSecs    int       `json:"idle_timeout_secs,omitempty"` // bus's IdleTimeout for diagnostics
+	Capabilities       []string  `json:"capabilities,omitempty"`
+	// CredentialGeneration is process-local and contains no secret data.
+	CredentialGeneration uint64 `json:"credential_generation"`
+	// TerminalReason is a fixed, non-sensitive bus terminal state. A runtime
+	// client checks it before sending credential material.
+	TerminalReason string `json:"terminal_reason,omitempty"`
+}
+
+// CredentialUpdate installs a host-supplied runtime token into a compatible
+// bus after the bus has advertised CapabilityRuntimeTokenV1. Token is carried
+// only over the owner-only local IPC connection and must never be logged.
+type CredentialUpdate struct {
+	Type               FrameType `json:"type"`
+	ExpectedGeneration uint64    `json:"expected_generation"`
+	Token              string    `json:"token"`
+}
+
+const (
+	CredentialErrorGenerationConflict = "generation_conflict"
+	CredentialErrorInvalid            = "invalid_credential"
+	CredentialErrorRegistration       = "registration_failed"
+	CredentialErrorRuntimeRejected    = "runtime_token_rejected"
+	CredentialErrorInternal           = "internal_error"
+)
+
+// CredentialUpdateAck reports the CAS result without echoing any credential.
+type CredentialUpdateAck struct {
+	Type                 FrameType `json:"type"`
+	Accepted             bool      `json:"accepted"`
+	CredentialGeneration uint64    `json:"credential_generation"`
+	ErrorCode            string    `json:"error_code,omitempty"`
+	Error                string    `json:"error,omitempty"`
 }
 
 // Event wraps one delivered RawEvent for the wire. We keep the payload as
@@ -126,7 +171,10 @@ type Bye struct {
 	Reason string    `json:"reason"`
 }
 
-const ByeReasonSubscriptionStopped = "subscription_stopped"
+const (
+	ByeReasonSubscriptionStopped  = "subscription_stopped"
+	ByeReasonRuntimeTokenRejected = "runtime_token_rejected"
+)
 
 // ConsumerStopReq asks the bus to close consumers whose exact personal
 // subscription IDs match. It is an additive local IPC control operation;

internal/event/transport/protocol_test.go

@@ -23,21 +23,23 @@ import (
 // to guard against accidental wire-format changes (the JSON field tags
 // double as part of the cross-version protocol contract).
 
-func TestFrameType_StableWireValues(t *testing.T) {
+func TestCrossPlatformCoverageFrameTypeStableWireValues(t *testing.T) {
 	// If any of these strings change we've made a protocol-breaking
 	// change. The test value list is duplicated here on purpose so a
 	// reviewer renaming a constant is forced to also update the test.
 	wants := map[FrameType]string{
-		FrameTypeHello:            "hello",
-		FrameTypeHelloAck:         "hello_ack",
-		FrameTypeEvent:            "event",
-		FrameTypeHeartbeat:        "heartbeat",
-		FrameTypeSourceState:      "source_state",
-		FrameTypeBye:              "bye",
-		FrameTypeStatusReq:        "status_req",
-		FrameTypeStatusResp:       "status_resp",
-		FrameTypeConsumerStopReq:  "consumer_stop_req",
-		FrameTypeConsumerStopResp: "consumer_stop_resp",
+		FrameTypeHello:               "hello",
+		FrameTypeHelloAck:            "hello_ack",
+		FrameTypeEvent:               "event",
+		FrameTypeHeartbeat:           "heartbeat",
+		FrameTypeSourceState:         "source_state",
+		FrameTypeBye:                 "bye",
+		FrameTypeStatusReq:           "status_req",
+		FrameTypeStatusResp:          "status_resp",
+		FrameTypeConsumerStopReq:     "consumer_stop_req",
+		FrameTypeConsumerStopResp:    "consumer_stop_resp",
+		FrameTypeCredentialUpdate:    "credential_update",
+		FrameTypeCredentialUpdateAck: "credential_update_ack",
 	}
 	for ft, want := range wants {
 		if string(ft) != want {
@@ -104,54 +106,91 @@ func roundTrip(t *testing.T, in any, dst any) {
 	}
 }
 
-func TestHello_Roundtrip(t *testing.T) {
+func TestCrossPlatformCoverageHelloRoundtrip(t *testing.T) {
 	in := Hello{
-		Type:        FrameTypeHello,
-		ConsumerPID: 42,
-		EventTypes:  []string{"im.*", "approval.task"},
-		Filter:      `^im\.`,
-		Compact:     true,
-		Role:        HelloRoleStatus,
+		Type:           FrameTypeHello,
+		ConsumerPID:    42,
+		EventTypes:     []string{"im.*", "approval.task"},
+		Filter:         `^im\.`,
+		Compact:        true,
+		Role:           HelloRoleStatus,
+		CredentialMode: CredentialModeRuntimeToken,
 	}
 	var out Hello
 	roundTrip(t, in, &out)
 	if out.Type != in.Type || out.ConsumerPID != in.ConsumerPID || out.Filter != in.Filter ||
-		out.Compact != in.Compact || out.Role != in.Role || len(out.EventTypes) != len(in.EventTypes) {
+		out.Compact != in.Compact || out.Role != in.Role || out.CredentialMode != in.CredentialMode || len(out.EventTypes) != len(in.EventTypes) {
 		t.Fatalf("roundtrip mismatch: %+v != %+v", out, in)
 	}
 }
 
-func TestHello_OmitemptyForDefaults(t *testing.T) {
+func TestCrossPlatformCoverageHelloOmitemptyForDefaults(t *testing.T) {
 	// Default values should NOT appear in the wire form so old/new readers
 	// stay tolerant of each other (each new field comes in with its
 	// zero value by default).
 	in := Hello{Type: FrameTypeHello, ConsumerPID: 1}
 	b, _ := json.Marshal(in)
 	s := string(b)
-	for _, k := range []string{`"event_types"`, `"filter"`, `"compact"`, `"role"`} {
+	for _, k := range []string{`"event_types"`, `"filter"`, `"compact"`, `"role"`, `"credential_mode"`} {
 		if strings.Contains(s, k) {
 			t.Errorf("zero-value field %s leaked into wire form: %s", k, s)
 		}
 	}
 }
 
-func TestHelloAck_Roundtrip(t *testing.T) {
+func TestCrossPlatformCoverageHelloAckRoundtrip(t *testing.T) {
 	in := HelloAck{
-		Type:               FrameTypeHelloAck,
-		BusPID:             12345,
-		SourceState:        "connected",
-		StateSource:        "inferred",
-		ClientIDSource:     "env",
-		ClientSecretSource: "env",
-		IdleTimeoutSecs:    300,
+		Type:                 FrameTypeHelloAck,
+		BusPID:               12345,
+		SourceState:          "connected",
+		StateSource:          "inferred",
+		ClientIDSource:       "env",
+		ClientSecretSource:   "env",
+		IdleTimeoutSecs:      300,
+		Capabilities:         []string{CapabilityRuntimeTokenV1},
+		CredentialGeneration: 7,
 	}
 	var out HelloAck
 	roundTrip(t, in, &out)
-	if out != in {
+	if out.Type != in.Type || out.BusPID != in.BusPID || out.SourceState != in.SourceState ||
+		out.StateSource != in.StateSource || out.ClientIDSource != in.ClientIDSource ||
+		out.ClientSecretSource != in.ClientSecretSource || out.IdleTimeoutSecs != in.IdleTimeoutSecs ||
+		out.CredentialGeneration != in.CredentialGeneration || strings.Join(out.Capabilities, ",") != CapabilityRuntimeTokenV1 {
 		t.Fatalf("HelloAck roundtrip: %+v != %+v", out, in)
 	}
 }
 
+func TestCrossPlatformCoverageCredentialFramesRoundtripWithoutTokenInAck(t *testing.T) {
+	update := CredentialUpdate{
+		Type:               FrameTypeCredentialUpdate,
+		ExpectedGeneration: 4,
+		Token:              "canary-secret",
+	}
+	var decodedUpdate CredentialUpdate
+	roundTrip(t, update, &decodedUpdate)
+	if decodedUpdate != update {
+		t.Fatal("CredentialUpdate roundtrip mismatch")
+	}
+
+	ack := CredentialUpdateAck{
+		Type:                 FrameTypeCredentialUpdateAck,
+		Accepted:             true,
+		CredentialGeneration: 5,
+	}
+	var decodedAck CredentialUpdateAck
+	roundTrip(t, ack, &decodedAck)
+	if decodedAck != ack {
+		t.Fatalf("CredentialUpdateAck roundtrip = %#v", decodedAck)
+	}
+	b, err := json.Marshal(ack)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(string(b), update.Token) {
+		t.Fatal("credential acknowledgement contained token")
+	}
+}
+
 func TestEvent_Roundtrip(t *testing.T) {
 	in := Event{
 		Type:              FrameTypeEvent,
@@ -245,7 +284,7 @@ func TestStatusResp_Roundtrip(t *testing.T) {
 // PeekType is used by the daemon to dispatch incoming frames before
 // fully decoding into the typed struct. Behaviour at boundaries matters.
 
-func TestPeekType_EachFrameVariant(t *testing.T) {
+func TestCrossPlatformCoveragePeekTypeEachFrameVariant(t *testing.T) {
 	cases := []struct {
 		v   any
 		typ FrameType
@@ -258,6 +297,8 @@ func TestPeekType_EachFrameVariant(t *testing.T) {
 		{Bye{Type: FrameTypeBye, Reason: "x"}, FrameTypeBye},
 		{StatusReq{Type: FrameTypeStatusReq}, FrameTypeStatusReq},
 		{StatusResp{Type: FrameTypeStatusResp}, FrameTypeStatusResp},
+		{CredentialUpdate{Type: FrameTypeCredentialUpdate}, FrameTypeCredentialUpdate},
+		{CredentialUpdateAck{Type: FrameTypeCredentialUpdateAck}, FrameTypeCredentialUpdateAck},
 	}
 	for _, c := range cases {
 		b, _ := json.Marshal(c.v)

skills/mono/references/products/calendar.md

@@ -5,7 +5,8 @@
 - **二级子命令（必选其一）**：`event`（日程）、`attendee`（参会人）、`room`（会议室）、`busy`（闲忙）、`attachment`（日程附件）、`book`（我能看哪些日历本）、`acl`（我的日历共享给了谁）。`dws calendar` 后**必须**紧跟上述之一；**禁止**只执行 `dws calendar`（无子命令）。
 - **个人日程 / 给自己留时间块 / 专注时段**：统一走 **`dws calendar event create`**。当前**没有**单独的 `personal schedule create` / `calendar create` 命令。
 - **查日程列表**：`dws calendar event list --start "<ISO-8601>" --end "<ISO-8601>" --format json`，或优先使用脚本 `python scripts/calendar_today_agenda.py [today|tomorrow|week]`（见文末「自动化脚本」）。
-- **查用户日历本列表**：`dws calendar book list`（返回字段为 `calendarId`，主日历 `calendarId == "primary"` 等；注意无 `id` 字段，按 `.id` 提取会取空）。**重要**可以查询他人共享给自己的日历本，根据日历本id可以进一步查询对方的日程信息。
+- **查循环日程实例**：`dws calendar event instances --id <EVENT_ID> --start "<ISO-8601>" --end "<ISO-8601>" --format json`，用于按时间范围展开重复日程的每一个实例。**注意：此接口只能查询重复性日程；普通非循环日程将查不到任何实例信息。**
+- **查用户日历本列表**：`dws calendar book list`（返回主日历 `id == "primary"` 等）。**重要**可以查询他人共享给自己的日历本，根据日历本id可以进一步查询对方的日程信息。
 - **CLI 不存在**独立的 `dws calendar list`；若误跑无子命令的 `dws calendar`，会打印整段 Usage，**切勿**将该段 help 当作工具结果再次塞进对话（会急剧增加 token 与首字延迟）。
 - **必须**遵循指令说明进行调用。**绝对禁止**使用虚构指令，使用虚构参数。
 
@@ -13,7 +14,6 @@
 1. **禁止**执行 `dws calendar` 且不带二级子命令（会刷出大量帮助文本）。合法二级子命令：`event` / `attendee` / `room` / `busy` / `attachment` / `book` / `acl`。
 2. **禁止**使用不存在的子命令试探（如臆造 `dws calendar list`）；需要日程列表时一律使用 **`dws calendar event list`**（带 `--start` / `--end`，见下文「查询日程列表」示例）；需要日历本列表时使用 **`dws calendar book list`**。
 3. **禁止**将完整 `--help`/Usage 输出作为「观察」重复提交给模型；若误触，应直接改用本节黄金路径中的合法命令并重试。
-4. **禁止**继续使用已弃用的 `--max-results` 参数 —— 它们仍可被解析但**会被丢弃**，不再透传到 MCP，模型生成命令时也不要再带。
 5. **禁止**为已有日程重新创建日程来预订会议室。若日程已存在（同一会话中刚创建、或用户明确指向某日程），必须使用 `room add --event <已有EVENT_ID> --rooms <ROOM_ID>` 追加会议室，**绝不能**再调一次 `event create --rooms`（会创建重复日程）。
 6. **禁止**用 `--location` 替代会议室预订。`--location` 是纯文本地点备注字段，填入会议室名称**不会**完成任何预订或占用。预订会议室必须通过 `room add --rooms <roomId>` 或 `event create --rooms <roomId>`，roomId 来自 `room search` 返回；`--location` 与 `--rooms` 是两个独立字段，用途完全不同。
 7. **禁止**只传 `--recurrence-*` 部分 flag **并不是彼此独立的参数**：只传其中一项（比如只改 `--recurrence-count`、只设 `--recurrence-type`）会让服务端收到不完整的 recurrence 结构，CLI 现已前置校验并直接拒绝这类调用。**修改已有周期日程的任何一个循环字段时，都必须重新提供完整的 pattern+range 字段集合**——必要时先 `event get` 读取现有 `recurrence`，再在命令中整体重传。
@@ -26,7 +26,7 @@
 日程实例（event instance）：日程的具体时间实例，可以通过event list指令查询时间段内的所有实例。1个普通日程和对应1个Instance，而1个重复性日程(SeriesMaster)对应N个Instance（同属一个日程序列）。
   - 同一个日程序列具有相同的iCalUid，并且重复性日程，其eventId和iCalUid的值相同。因此可以通过重复性日程实例的iCalUid得到重复性日程(SeriesMaster)的eventId
 重复规则（recurrence rule）：定义重复性日程的重复规则。
-参会人（attendee）：日程的参与者。常用通讯录工具查询userId，dws contact user search --keyword "姓名"。
+参会人（attendee）：日程的参与者。按姓名统一使用 `dws aisearch person --keyword "姓名" --dimension name --format json` 查询 userId。
 响应状态（response）：参会人对日程的回应，包括：未响应、接受、待定、拒绝。
 忙闲时间（busy）：查询用户在指定时间段的忙闲状态，查询会议室在指定时间段的预定状态，用于会议时间协调。
 会议室（room）：room是 会议室 ，room可视为日程的资源类参会人，需要加入日程完成预订。注意和location区分，location只是地点，和room不同。
@@ -42,6 +42,8 @@
 dws calendar event [create|update|get|delete|respond] [flags]
 # 按时间范围批量查询
 dws calendar event list [flags]
+# 查询循环日程的实例列表（按时间范围展开重复日程）
+dws calendar event instances [flags]
 # 对于非明确时间或一段时间范围的约会场景，可基于所有参会人的忙闲状态，推荐多个可用的时间块方案
 dws calendar event suggest [flags]
 ```
@@ -64,7 +66,6 @@ dws calendar room add [flags]
 dws calendar room delete [flags]
 ```
 > room是会议室，用于线下开会场景。
-> **组织限制**：部分企业使用自建会议室系统，未接入钉钉会议室能力。此时 `room search` / `room list-groups` 会返回业务错误 `400056`（"所选组织不支持预定钉钉会议室"）。这是组织级配置限制，不是命令用法问题；应直接告知用户该组织需在钉钉客户端手动预订会议室，不要重试或换参。
 
 ### busy 相关三级子命令
 ```
@@ -109,17 +110,21 @@ Usage:
   dws calendar event list [flags]
 Example:
   dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T18:00:00+08:00"
+  dws calendar event list --start "2026-03-10T00:00:00+08:00" --end "2026-03-31T23:59:59+08:00" --limit 50
   dws calendar event list --calendar-id primary
+  dws calendar event list --cursor "<nextCursor从上一次查询结果获取>"
 Flags:
       --calendar-id string   日历 ID (默认 primary 主日历，仅在查询其他日历本时填写；通过 `book list` 获取)
+      --cursor string        分页游标 (从上一次返回的 nextCursor 获取，首次查询无需传入)
       --end string           结束时间 ISO-8601 (例如 2026-03-10T18:00:00+08:00)
+      --limit int            每页返回条数 (默认 100，最大 100)
       --start string         开始时间 ISO-8601 (例如 2026-03-10T14:00:00+08:00)
 ```
 
-> `--max-results` 已弃用：MCP 不再支持，CLI 仍接受但参数会被丢弃，结果固定最多返回 100 条。
-
 **默认行为**：不传 `--start` / `--end` 时，默认返回今天的日程（00:00:00 ~ 23:59:59）。
 **权限**：查询共享日历下的日程时，至少要有reader权限。
+**分页**：单次最多返回 `--limit` 指定的条数（默认/最大 100）；当结果超过 limit 时，返回体包含 `nextCursor` 字段。首次查询无需传 `--cursor`，仅在翻页时将上一次返回的 `nextCursor` 作为 `--cursor` 传入。
+
 
 ### 获取日程详情
 ```
@@ -133,6 +138,28 @@ Flags:
       --calendar-id string   日历 ID (默认 primary 主日历)
 ```
 
+### 查询循环日程实例
+```
+Usage:
+  dws calendar event instances [flags]
+Example:
+  dws calendar event instances --id <EVENT_ID>
+  dws calendar event instances --id <EVENT_ID> --start "2026-03-10T00:00:00+08:00" --end "2026-03-31T23:59:59+08:00"
+  dws calendar event instances --id <EVENT_ID> --limit 50
+  dws calendar event instances --id <EVENT_ID> --cursor "<nextCursor>"
+Flags:
+      --id string            日程 ID (必填，重复性日程 SeriesMaster 的 eventId)
+      --calendar-id string   日历 ID (默认 primary 主日历，仅在查询其他日历本时填写；通过 `book list` 获取)
+      --start string         开始时间 ISO-8601 (例如 2026-03-10T00:00:00+08:00，不传则默认今天 00:00:00)
+      --end string           结束时间 ISO-8601 (例如 2026-03-31T23:59:59+08:00，不传则默认今天 23:59:59)
+      --limit int            每页返回条数 (默认 100，最大 100)
+      --cursor string        分页游标 (从上一次返回的 nextCursor 获取，首次查询无需传入)
+```
+
+> **说明**：用于按时间范围展开重复日程（SeriesMaster）的每一个实例。**此接口只能查询重复性日程；若传入普通非循环日程，将查不到任何实例信息。**`--id` 必须是重复性日程的 eventId，可通过 `event list` 获取。
+> **默认行为**：不传 `--start` / `--end` 时，默认返回今天的实例（00:00:00 ~ 23:59:59）。
+> **分页**：单次最多返回 `--limit` 指定的条数（默认/最大 100）；当结果超过 limit 时，返回体包含 `nextCursor` 字段。
+
 ### 创建日程
 ```
 Usage:
@@ -149,10 +176,17 @@ Example:
   dws calendar event create --title "每日站会" \
     --start "2026-03-10T09:00:00+08:00" --end "2026-03-10T09:30:00+08:00" \
     --recurrence-type daily --recurrence-interval 1 --recurrence-range-type numbered --recurrence-count 10
+  dws calendar event create --title "团队周会" \
+    --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" \
+    --calendar-id <SHARED_CALENDAR_ID>     # 在指定日历本下创建日程
+  dws calendar event create --title "重要会议" \
+    --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" \
+    --remind-minutes 5,10     # 开始前5分钟和10分钟各提醒一次
 Flags:
       --title string                    日程标题 (必填，最大2048字符)
       --start string                    开始时间 ISO-8601 (必填，例如 2026-03-10T14:00:00+08:00)
       --end string                      结束时间 ISO-8601 (必填，例如 2026-03-10T15:00:00+08:00)
+      --calendar-id string              日历 ID (可选，默认 primary 主日历；仅在共享/订阅日历本下创建时填写，通过 `book list` 获取)
       --timezone string                 时区 IANA 格式 (例如 Asia/Shanghai，默认 Asia/Shanghai)
       --desc string                     日程描述 (最大5000字符)
       --attendees string                参会人 userId 列表，逗号分隔 (最多500人) 日程组织人自动放入参会人列表，无需传入userId
@@ -172,6 +206,7 @@ Flags:
       --rich-text-desc string           html格式的富文本类型日程描述，用于复杂内容的展示
       --location string                 地点信息（纯文本备注，如‘3号楼A区’；**不等于**预订会议室）
       --free-busy string                此日程的忙碌状态，默认值为busy。busy - 在忙闲视图中，此日程时间段为忙碌; free - 此日程不占用忙闲
+      --remind-minutes string           日程开始前提醒，逗号分隔分钟数 (可选，例如 5,10,15 表示开始前5/10/15分钟提醒；不传则默认15分钟提醒)
 ```
 
 > **说明**：个人日程也走 `event create`。如果只是给自己安排时间，不传 `--attendees` / `--open-dingtalk-ids` 即可。
@@ -185,8 +220,10 @@ Example:
   dws calendar event update --id <EVENT_ID> --desc "新描述" --timezone Asia/Tokyo
   dws calendar event update --id <EVENT_ID> --recurrence-type daily --recurrence-interval 1 \
     --recurrence-range-type numbered --recurrence-count 5
+  dws calendar event update --id <EVENT_ID> --calendar-id <SHARED_CALENDAR_ID> --title "新标题"   # 修改其他日历本下的日程
 Flags:
       --id string                       日程 ID (必填)
+      --calendar-id string              日历 ID (可选，默认 primary 主日历；指定其他日历本时填写，可通过 `book list` 获取)
       --title string                    新标题
       --start string                    新开始时间 ISO-8601
       --end string                      新结束时间 ISO-8601
@@ -218,8 +255,10 @@ Usage:
   dws calendar event delete [flags]
 Example:
   dws calendar event delete --id <EVENT_ID> --yes
+  dws calendar event delete --id <EVENT_ID> --calendar-id <SHARED_CALENDAR_ID> --yes   # 删除其他日历本下的日程
 Flags:
-      --id string   日程 ID (必填)
+      --id string            日程 ID (必填)
+      --calendar-id string   日历 ID (可选，默认 primary 主日历；指定其他日历本时填写，可通过 `book list` 获取)
 ```
 
 ### 查看参会人
@@ -228,8 +267,10 @@ Usage:
   dws calendar attendee list [flags]
 Example:
   dws calendar attendee list --event <EVENT_ID>
+  dws calendar attendee list --event <EVENT_ID> --calendar-id <SHARED_CALENDAR_ID>   # 查看其他日历本下日程的参会人
 Flags:
-      --event string   日程 ID (必填)
+      --event string         日程 ID (必填)
+      --calendar-id string   日历 ID (可选，默认 primary 主日历；指定其他日历本时填写，可通过 `book list` 获取, 注意：订阅日历下的日程无参会人，因此不可查看)
 ```
 
 ### 添加参会人
@@ -239,10 +280,12 @@ Usage:
 Example:
   dws calendar attendee add --event <EVENT_ID> --attendees <USER_ID_1>,<USER_ID_2>
   dws calendar attendee add --event <EVENT_ID> --attendees <USER_ID> --optional
+  dws calendar attendee add --event <EVENT_ID> --attendees <USER_ID> --calendar-id <SHARED_CALENDAR_ID>   # 给其他日历本下的日程添加参会人
 Flags:
-      --event string       日程 ID (必填)
-      --attendees string   参会人 userId 列表，逗号分隔 (必填，最多500人)
-      --optional           参会人可选 (默认必选参会人)
+      --event string         日程 ID (必填)
+      --attendees string     参会人 userId 列表，逗号分隔 (必填，最多500人)
+      --optional             参会人可选 (默认必选参会人)
+      --calendar-id string   日历 ID (可选，默认 primary 主日历；指定其他日历本时填写，可通过 `book list` 获取，注意：订阅日历下的日程无参会人，因此不可添加)
 ```
 
 ### 移除参会人
@@ -254,36 +297,54 @@ Usage:
   dws calendar attendee delete [flags]
 Example:
   dws calendar attendee delete --event <EVENT_ID> --attendees <USER_ID> --yes
+  dws calendar attendee delete --event <EVENT_ID> --attendees <USER_ID> --calendar-id <SHARED_CALENDAR_ID> --yes   # 移除其他日历本下日程的参会人
 Flags:
-      --event string       日程 ID (必填)
-      --attendees string   参会人 userId 列表，逗号分隔 (必填)
+      --event string         日程 ID (必填)
+      --attendees string     参会人 userId 列表，逗号分隔 (必填)
+      --calendar-id string   日历 ID (可选，默认 primary 主日历；指定其他日历本时填写，可通过 `book list` 获取，注意：订阅日历下的日程无参会人，因此不可移除)
 ```
 
 ### 搜索会议室
-> 此指令可用于搜索当前用户可用的会议室。**注意**，大部分会议室**仅在工作时间**可用，如果检索时间不在工作时间可能查不到任何结果。
+> 此指令支持两种模式：**按名称搜索**（不传 --start/--end）和**按时间段搜索可用会议室**（传 --start/--end 或不传任何参数）。
 > 此指令搜索到的会议室结果中，有两个值需要注意：
- 		- customApprovalProcess: true - 表示该会议室设置了自定义审批流程，只能通过客户端完成预订。
- 		- supportRecurring: true - 表示该会议室支持循环预定；false - 表示不支持循环预定，直接加入到循环日程会失败。
+> - customApprovalProcess: true - 表示该会议室设置了自定义审批流程，只能通过客户端完成预订。
+> - supportRecurring: true - 表示该会议室支持循环预定；false - 表示不支持循环预定，直接加入到循环日程会失败。
+
+**模式路由规则**：
+- **按名称搜索**：仅传 `--room-name`，不传 `--start`/`--end` → 返回所有匹配名称的会议室，**不检查可用性**。适用于「找到某个会议室」的场景。
+- **按时间段搜索可用会议室**：传 `--start`/`--end`，或不传任何参数 → 返回指定时间段内**可用**的会议室。不传时间时默认当前时间起 1 小时。
 
 ```
 Usage:
   dws calendar room search [flags]
 Example:
+  # 按名称搜索（不检查可用性，返回所有匹配的会议室）
+  dws calendar room search --room-name 永澄亭   # 注意：用户即使说「永澄亭会议室」，也应仅传「永澄亭」
+
+  # 按时间段搜索可用会议室
   dws calendar room search --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00"
   dws calendar room search --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" --group-id <GROUP_ID>
-  dws calendar room search --room-name 永澄亭   # 注意：用户即使说「永澄亭会议室」，也应仅传「永澄亭」
   dws calendar room search   # 不传 --start/--end 时默认当前时间起 1 小时
+
+  # 名称 + 时间段：搜索指定名称的可用会议室
+  dws calendar room search --room-name 永澄亭 --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00"
+
+  # 分页（仅按时间段搜索时有效）
+  dws calendar room search --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" --limit 20 --page 0
 Flags:
       --start string        开始时间 ISO-8601 (可选，不传则默认当前时间+1分钟缓冲）
       --end string          结束时间 ISO-8601 (可选，不传则默认当前时间+1 小时)
-      --group-id string     会议室分组ID（可选，留空查根目录；超100条时需按分组查询）
-      --room-name string    按会议室名称过滤（可选，服务端模糊匹配；传入前必须由调用方精简，只保留核心专名）
+      --group-id string     会议室分组ID（可选，留空查根目录；超100条时需按分组查询；仅按时间段搜索时有效）
+      --room-name string    会议室名称（按名称搜索时必填；按时间段搜索时可选，用于过滤）
+      --limit string        页大小 (可选，不填默认 100，超过 100 按 100 处理；仅按时间段搜索时有效)
+      --page string         分页起始位置 (可选，不填默认 0；仅按时间段搜索时有效)
 ```
 
 > **时间约束（API 限制）**：`start` 必须是未来的时间（服务端校验：start can not less current time）。
 > - 若传入的 `--start` 早于当前时间，CLI 会自动修正为 `now + 1min`，调用方无需额外处理。
 > - 若传入的 `--end` 早于当前时间，CLI 直接报错——无法检索已过去的时间段。
 > - **最佳实践**：调用方在组装时间参数时应确保 start/end 都是未来时间；若不确定，可省略 `--start`/`--end` 让 CLI 使用默认值（当前时间起 1 小时）。
+> - **注意**：时间约束仅在「按时间段搜索」模式下生效。按名称搜索（仅传 `--room-name`）不受时间约束。
 
 **名称过滤使用规范**：`--room-name` 适用于用户说「预定永澄亭」「约西湖厅」这类按名找会议室的场景。
 - **服务端是模糊匹配，但匹配词越精简命中率越高**，关键疗法：**调用方必须在调用 CLI 前自行精简名称，CLI 不会再做任何删减**。
@@ -293,13 +354,16 @@ Flags:
   - 用户：「西湖厅有空吗」 → `--room-name 西湖厅`（本身就是专名，不删即可）
   - 用户：「预定贡嘎山大会议室」 → `--room-name 贡嘎山`
 
-**优先路径**：当用户给了会议室中文名时，**优先用 `room search --room-name <核心专名> --start <开始时间> --end <结束时间>`**代替「先走 `list-groups`再遍历」的旧流程。`--room-name` 可与 `--group-id` 同时使用，表示「在指定分组内按名称过滤」。若返回空列表，再降级使用 `list-groups` 定位分组再查。
+**优先路径**：
+- 当用户仅给出会议室名称、未指定时间段时，用 `room search --room-name <核心专名>` 快速找到会议室（不检查可用性）。
+- 当用户给出会议室名称且指定了时间段时，用 `room search --room-name <核心专名> --start <开始时间> --end <结束时间>` 查询该名称的可用会议室。
+- 若返回空列表，再降级使用 `list-groups` 定位分组再查。`--room-name` 可与 `--group-id` 同时使用（仅按时间段搜索时），表示「在指定分组内按名称过滤」。
 
 **`roomId` 与用户说的话不是一回事**：用户说的「C6-4-06-N 贡嘎山」等是**展示名/编号文案**，**绝不能**直接填进 `room add --rooms`。`--rooms` 只接受上一步 `room search`（或同类接口）返回 JSON 里的 **`rooms[].roomId`**。形态上多为**小写十六进制串**（长度以接口为准，例如 `e6b7b65b8b30fb707afcf6c3b699f028003e6834fdd7fee7`）。含**中文、空格、连字符拼接的楼层编号**、或凭空调 UUID/纯数字「试格式」——一律视为非法，必须先搜房再取返回字段。
 
 > 如果知道roomId，想查该会议室的预订记录，直接用dws calendar busy search 指令
 
---- 
+---
 
 ### 预定会议室
 ```
@@ -307,9 +371,11 @@ Usage:
   dws calendar room add [flags]
 Example:
   dws calendar room add --event <EVENT_ID> --rooms <ROOM_ID>
+  dws calendar room add --event <EVENT_ID> --rooms <ROOM_ID> --calendar-id <SHARED_CALENDAR_ID>   # 给其他日历本下的日程预定会议室
 Flags:
-      --event string   日程 ID (必填)
-      --rooms string   会议室 ID 列表 (必填)
+      --event string         日程 ID (必填)
+      --rooms string         会议室 ID 列表 (必填)
+      --calendar-id string   日历 ID (可选，默认 primary 主日历；指定其他日历本时填写，可通过 `book list` 获取，注意：订阅日历下的日程不可添加会议室)
 ```
 > room是会议室，用于线下开会场景。将room加入到日程完成预订
 > 重复性日程，预订会议室时，必须设置 循环结束时间（recurrence-end-date），noEnd 或者 指定循环次数 都无法完成预定。
@@ -324,9 +390,11 @@ Usage:
   dws calendar room delete [flags]
 Example:
   dws calendar room delete --event <EVENT_ID> --rooms <ROOM_ID> --yes
+  dws calendar room delete --event <EVENT_ID> --rooms <ROOM_ID> --calendar-id <SHARED_CALENDAR_ID> --yes   # 移除其他日历本下日程的会议室
 Flags:
-      --event string   日程 ID (必填)
-      --rooms string   会议室 ID 列表 (必填)
+      --event string         日程 ID (必填)
+      --rooms string         会议室 ID 列表 (必填)
+      --calendar-id string   日历 ID (可选，默认 primary 主日历；指定其他日历本时填写，可通过 `book list` 获取。注意：订阅日历下的日程不可添加会议室)
 ```
 
 ### 会议室分组列表
@@ -347,22 +415,24 @@ Usage:
   dws calendar attachment add [flags]
 Example:
   dws calendar attachment add --event <EVENT_ID> --files <FILE_ID>:report.pdf,<FILE_ID2>:slides.pptx
+  dws calendar attachment add --event <EVENT_ID> --files <FILE_ID>:report.pdf --calendar-id <SHARED_CALENDAR_ID>   # 给其他日历本下的日程添加附件
 Flags:
-      --event string   日程 ID (必填)
-      --files string   附件列表，格式 <fileId>:<name>，多项逗号分隔 (必填)
+      --event string         日程 ID (必填)
+      --files string         附件列表，格式 <fileId>:<name>，多项逗号分隔 (必填)
+      --calendar-id string   日历 ID (可选，默认 primary 主日历；指定其他日历本时填写，可通过 `book list` 获取，注意：订阅日历下的日程不可添加附件)
 ```
 
 > 上传文件得到 `fileId` 需配合钉盘相关流程；本命令只负责把已上传的文件挂载到日程上。
 
-### 查询用户日历列表
+### 查询我能看的所有日历本
 ```
 Usage:
   dws calendar book list [flags]
 Example:
   dws calendar book list
 ```
-> 通过此接口可查询当前用户的日历列表，包含 主日历本、他人共享的日历、订阅的公共/团队日历。查的是"我能看哪些日历本"；注意区分：`acl list` 是查"我的日历共享**给了谁**"，方向相反。
-> 共享日历本中有来自 xxx 的，且权限大于reader，那么通过 `event list --calendar-id <xxx的日历本id> `可查到xxx完整的日程安排
+> 查询"我能看哪些日历本"，包含：我的主日历、他人共享**给我**的日历、订阅的公共/团队日历。注意区分：`acl list` 是查"我的日历共享**给了谁**"，方向相反。
+> 共享日历本中有来自 xxx 的，且权限大于reader，那么通过 `event list --calendar-id <xxx的日历本id> `可查到xxx完整的日程安排。
 > 主日历 `id` 固定为 `primary`，绝大多数日程操作都默认走主日历，只有当用户明确要求查/写其他日历本时才需要带 `--calendar-id`。
 
 ### 查询指定日历本
@@ -495,9 +565,11 @@ Example:
   dws calendar event respond --id <EVENT_ID> --status accepted
   dws calendar event respond --id <EVENT_ID> --status declined
   dws calendar event respond --id <EVENT_ID> --status tentative
+  dws calendar event respond --id <EVENT_ID> --status accepted --calendar-id <SHARED_CALENDAR_ID>   # 响应其他日历本下的日程
 Flags:
-      --id string       日程 ID (必填)
-      --status string   响应状态: needsAction(未操作)|accepted(接受)|declined(拒绝)|tentative(暂定) (必填)
+      --id string            日程 ID (必填)
+      --status string        响应状态: needsAction(未操作)|accepted(接受)|declined(拒绝)|tentative(暂定) (必填)
+      --calendar-id string   日历 ID (可选，默认 primary 主日历；指定其他日历本时填写，可通过 `book list` 获取。注意：订阅日历下的日程无参会人，因此不可响应)
 ```
 
 > **说明**：作为日程参会人，设置自己的响应状态（接受、拒绝、暂定）。`--status` 可选值：`needsAction`（未操作，默认值）、`accepted`（接受）、`declined`（拒绝）、`tentative`（暂定）。
@@ -507,11 +579,12 @@ Flags:
 用户说"日程/会议/约会/日历":
 - 查看 → `event list`
 - 详情 → `event get`
-- 创建/约/给自己留时间块/个人日程 → `event create`（带参会人时加 `--attendees`，循环日程加 `--recurrence-*`）
+- 创建/约/给自己留时间块/个人日程 → `event create`（带参会人时加 `--attendees`，循环日程加 `--recurrence-*`，自定义提醒加 `--remind-minutes`）
 - 修改/改时间/改描述 → `event update`（支持修改标题、时间、描述、时区、循环规则）
 - 取消/删除 → `event delete`
 - 推荐时间/什么时候有空/协调时间 → `event suggest`
 - 接受/拒绝/暂定日程 → `event respond`
+- 查询循环日程/重复日程的每次实例/展开循环日程 → `event instances`
 
 用户说"参会人/与会者":
 - 查看 → `attendee list`
@@ -519,13 +592,14 @@ Flags:
 - 移除 → `attendee delete --attendees <USER_ID>`
 
 用户说"会议室/订会议室":
-- 哪个空闲 → `room search`
-- 按名找会议室（如「永澄亭」「永澄亭会议室」「约西湖厅」）→ 先在模型层精简名称（剔除「会议室」等通用后缀），再用 `room search --room-name <核心专名>`
+- 哪个空闲 → `room search`（默认查当前时间起 1 小时内可用会议室）
+- 按名找会议室（如「永澄亭」「永澄亭会议室」「约西湖厅」，未提时间段）→ 先在模型层精简名称（剔除「会议室」等通用后缀），再用 `room search --room-name <核心专名>`（按名称搜索，不检查可用性）
+- 按名找可用会议室（如「永澄亭下午 2 点有空吗」）→ `room search --room-name <核心专名> --start <开始时间> --end <结束时间>`（按名称+时间段搜索可用会议室）
 - 预订
   - 给已有日程订会议室 → `room add --event <已有EVENT_ID> --rooms <ROOM_ID>`
   - 创建新日程并订会议室 → `event create --rooms`（仅当日程尚不存在时）
 - 取消预定 → `room delete`
-- 分组 → `room list-groups`，取 groupId 后 `room search --group-id`（可再叠加 `--room-name` 在分组内过滤）
+- 分组 → `room list-groups`，取 groupId 后 `room search --group-id`（需配合 `--start`/`--end` 按时间段搜索；可再叠加 `--room-name` 在分组内过滤）
 
 用户说"有空吗/忙不忙/闲忙":
 - 查询用户闲忙 → `busy search --users <USER_ID>`
@@ -580,11 +654,11 @@ dws calendar event create --title "Q1 复盘会" \
 **方式二：先创建日程，再单独添加参会人 / 会议室**
 
 ```bash
-# Step 1: 创建日程 — 提取 eventId
+# Step 1: 创建日程 — 从 result.id 提取日程 ID
 dws calendar event create --title "Q1 复盘会" \
   --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15:00:00+08:00" --format json
 
-# Step 2: 添加参会人（必须用 Step 1 返回的 eventId）
+# Step 2: 添加参会人（必须用 Step 1 返回的 result.id）
 dws calendar attendee add --event <EVENT_ID> --attendees userId1,userId2 --format json
 
 # Step 3: 搜索空闲会议室
@@ -612,13 +686,14 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
 
 | 操作 | 从返回中提取 | 用于 |
 |------|-------------|------|
-| `event create` | `eventId` | attendee/room/attachment 操作的 --event |
-| `event list` | `events[].eventId` | event get/update/delete/respond 的 --id |
+| `event create` | `result.id` | attendee/room/attachment 操作的 --event |
+| `event list` | `result.events[].id`, `nextCursor` | event get/update/delete/respond 的 --id；下一页 --cursor |
 | `event suggest` | 推荐的时间段 | event create 的 --start/--end |
 | `event respond` | 响应结果 | — |
+| `event instances` | `result.events[].id`, `nextCursor` | event get/update/delete/respond 的 --id；下一页 --cursor |
 | `room search` | `rooms[].roomId` | room add 的 --rooms 或 event create 的 --rooms |
 | `room list-groups` | `groups[].groupId` | room search 的 --group-id |
-| `book list` | `calendarId`（如 `primary`） | event list/get 的 --calendar-id, book get/update 的 --id |
+| `book list` | `id`（如 `primary`） | event list/get 的 --calendar-id, book get/update 的 --id |
 | `book get` | 日历详细信息 | — |
 | `book search` | 匹配的日历列表 | book get/update 的 --id |
 | `acl list` | `aclId` | acl delete 的 --acl-id |
@@ -638,20 +713,23 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
 - `room list-groups` 支持 `--limit` / `--page` 分页（schema 类型为字符串）
 - **`event create --rooms` / `room add --rooms` 的唯一合法来源**：最近一次（同一会话、同一时段窗口）`room search` 返回体中的 `roomId`；禁止把用户自然语言会议室名当 `roomId` 传入（否则会 `roomId invalid` 等错误）
 - **搜房无结果**：在符合早停/用户限定范围内，`room search`（含按分组逐组查）全部返回空或无空闲 → 应**直接向用户报错/说明失败**并结束订房；**禁止**假设 roomId、禁止无合法 `roomId` 时调用 `room add` / `event create --rooms` 试探、禁止用 `event get` 等绕路推断 roomId
-- **评测 / 自动化断言**：凡涉及 `room add` / `event create --rooms` 的流程，`--rooms` 只能填上游 `room search`（或等价接口）返回 JSON 中的 **`rooms[].roomId`**；不得以会议室展示名、楼层文案或用户口语当作 `roomId`
+- **自动化校验**：凡涉及 `room add` / `event create --rooms` 的流程，`--rooms` 只能填上游 `room search`（或等价接口）返回 JSON 中的 **`rooms[].roomId`**；不得以会议室展示名、楼层文案或用户口语当作 `roomId`
 - **附件**：`attachment add` 仅负责挂载，**不上传**文件；fileId 必须先通过钉盘流程取得；`--files` 多附件用 `<fileId>:<name>` 元素逗号分隔
-- **日历本**：`book list` 返回的 `calendarId` 字段才是合法日历 id（无 `id` 字段）；如无明确说明，`event list` / `event get` 都不要带 `--calendar-id`，让接口默认走 primary 主日历
-- **已弃用入参**：`--max-results`（event list）在新 MCP schema 中被移除；CLI 仍接受但**不会**透传到 MCP；模型生成命令时**禁止**继续使用
+- **日历本**：`book list` 返回的 `id` 才是合法 `calendarId`；如无明确说明，`event list` / `event get` 都不要带 `--calendar-id`，让接口默认走 primary 主日历
+- **分页查询**：`event list` / `event instances` 均支持 `--limit`（控制每页条数，默认/最大 100）和 `--cursor`（翻页游标）；**首次查询无需传 `--cursor`**，仅当返回体中包含 `nextCursor` 时，将其作为 `--cursor` 传入可获取下一页
+- **循环日程实例**：`event instances` 用于按时间范围展开重复日程（SeriesMaster）的每一个实例；**普通非循环日程调用该命令将查不到任何实例信息**
+- **日程提醒**：`event create` 支持 `--remind-minutes` 设置开始前提醒，逗号分隔多个分钟数（如 `--remind-minutes 5,10,15`），不传则默认15分钟提醒
+- **会议室分页**：`room search` 支持 `--limit`（每页条数，默认100，最大100）和 `--page`（分页起始位置，默认0），与 `room list-groups` 分页风格一致
 
 ## 自动化脚本
 
 | 脚本 | 场景 | 用法 |
 |------|------|------|
-| [calendar_today_agenda.py](../../scripts/calendar_today_agenda.py) | 查看今天/明天/本周日程安排 | `python calendar_today_agenda.py today` |
-| [calendar_schedule_meeting.py](../../scripts/calendar_schedule_meeting.py) | 一键创建日程+添加参会人+预定会议室；搜房失败时输出明确原因并返回非零退出码 | `python calendar_schedule_meeting.py --title "复盘会" --start "2026-03-15T14:00" --end "2026-03-15T15:00" --users userId1 --book-room` |
-| [calendar_free_slot_finder.py](../../scripts/calendar_free_slot_finder.py) | 查询多人共同空闲时段 | `python calendar_free_slot_finder.py --users userId1,userId2 --date 2026-03-15` |
+| [calendar_today_agenda.py](../scripts/calendar_today_agenda.py) | 查看今天/明天/本周日程安排 | `python calendar_today_agenda.py today` |
+| [calendar_schedule_meeting.py](../scripts/calendar_schedule_meeting.py) | 一键创建日程+添加参会人+预定会议室；搜房失败时输出明确原因并返回非零退出码 | `python calendar_schedule_meeting.py --title "复盘会" --start "2026-03-15T14:00" --end "2026-03-15T15:00" --users userId1 --book-room` |
+| [calendar_free_slot_finder.py](../scripts/calendar_free_slot_finder.py) | 查询多人共同空闲时段 | `python calendar_free_slot_finder.py --users userId1,userId2 --date 2026-03-15` |
 
 ## 相关产品
 
-- conference 视频会议 — 当前 CLI 不支持发起/预约/邀请入会/会中控制；请在钉钉客户端操作
-- [contact](./contact.md) — 搜索同事 userId，用于 attendee add --attendees
+- conference（视频会议预约） — 仅视频会议预约（返回入会链接），不含参会人/会议室管理
+- [contact](../../dingtalk-contact/references/contact.md) — 搜索同事 userId，用于 attendee add --attendees

skills/multi/dingtalk-calendar/references/calendar.md

@@ -5,6 +5,7 @@
 - **二级子命令（必选其一）**：`event`（日程）、`attendee`（参会人）、`room`（会议室）、`busy`（闲忙）、`attachment`（日程附件）、`book`（我能看哪些日历本）、`acl`（我的日历共享给了谁）。`dws calendar` 后**必须**紧跟上述之一；**禁止**只执行 `dws calendar`（无子命令）。
 - **个人日程 / 给自己留时间块 / 专注时段**：统一走 **`dws calendar event create`**。当前**没有**单独的 `personal schedule create` / `calendar create` 命令。
 - **查日程列表**：`dws calendar event list --start "<ISO-8601>" --end "<ISO-8601>" --format json`，或优先使用脚本 `python scripts/calendar_today_agenda.py [today|tomorrow|week]`（见文末「自动化脚本」）。
+- **查循环日程实例**：`dws calendar event instances --id <EVENT_ID> --start "<ISO-8601>" --end "<ISO-8601>" --format json`，用于按时间范围展开重复日程的每一个实例。**注意：此接口只能查询重复性日程；普通非循环日程将查不到任何实例信息。**
 - **查用户日历本列表**：`dws calendar book list`（返回主日历 `id == "primary"` 等）。**重要**可以查询他人共享给自己的日历本，根据日历本id可以进一步查询对方的日程信息。
 - **CLI 不存在**独立的 `dws calendar list`；若误跑无子命令的 `dws calendar`，会打印整段 Usage，**切勿**将该段 help 当作工具结果再次塞进对话（会急剧增加 token 与首字延迟）。
 - **必须**遵循指令说明进行调用。**绝对禁止**使用虚构指令，使用虚构参数。
@@ -41,6 +42,8 @@
 dws calendar event [create|update|get|delete|respond] [flags]
 # 按时间范围批量查询
 dws calendar event list [flags]
+# 查询循环日程的实例列表（按时间范围展开重复日程）
+dws calendar event instances [flags]
 # 对于非明确时间或一段时间范围的约会场景，可基于所有参会人的忙闲状态，推荐多个可用的时间块方案
 dws calendar event suggest [flags]
 ```
@@ -135,6 +138,28 @@ Flags:
       --calendar-id string   日历 ID (默认 primary 主日历)
 ```
 
+### 查询循环日程实例
+```
+Usage:
+  dws calendar event instances [flags]
+Example:
+  dws calendar event instances --id <EVENT_ID>
+  dws calendar event instances --id <EVENT_ID> --start "2026-03-10T00:00:00+08:00" --end "2026-03-31T23:59:59+08:00"
+  dws calendar event instances --id <EVENT_ID> --limit 50
+  dws calendar event instances --id <EVENT_ID> --cursor "<nextCursor>"
+Flags:
+      --id string            日程 ID (必填，重复性日程 SeriesMaster 的 eventId)
+      --calendar-id string   日历 ID (默认 primary 主日历，仅在查询其他日历本时填写；通过 `book list` 获取)
+      --start string         开始时间 ISO-8601 (例如 2026-03-10T00:00:00+08:00，不传则默认今天 00:00:00)
+      --end string           结束时间 ISO-8601 (例如 2026-03-31T23:59:59+08:00，不传则默认今天 23:59:59)
+      --limit int            每页返回条数 (默认 100，最大 100)
+      --cursor string        分页游标 (从上一次返回的 nextCursor 获取，首次查询无需传入)
+```
+
+> **说明**：用于按时间范围展开重复日程（SeriesMaster）的每一个实例。**此接口只能查询重复性日程；若传入普通非循环日程，将查不到任何实例信息。**`--id` 必须是重复性日程的 eventId，可通过 `event list` 获取。
+> **默认行为**：不传 `--start` / `--end` 时，默认返回今天的实例（00:00:00 ~ 23:59:59）。
+> **分页**：单次最多返回 `--limit` 指定的条数（默认/最大 100）；当结果超过 limit 时，返回体包含 `nextCursor` 字段。
+
 ### 创建日程
 ```
 Usage:
@@ -559,6 +584,7 @@ Flags:
 - 取消/删除 → `event delete`
 - 推荐时间/什么时候有空/协调时间 → `event suggest`
 - 接受/拒绝/暂定日程 → `event respond`
+- 查询循环日程/重复日程的每次实例/展开循环日程 → `event instances`
 
 用户说"参会人/与会者":
 - 查看 → `attendee list`
@@ -664,6 +690,7 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
 | `event list` | `result.events[].id`, `nextCursor` | event get/update/delete/respond 的 --id；下一页 --cursor |
 | `event suggest` | 推荐的时间段 | event create 的 --start/--end |
 | `event respond` | 响应结果 | — |
+| `event instances` | `result.events[].id`, `nextCursor` | event get/update/delete/respond 的 --id；下一页 --cursor |
 | `room search` | `rooms[].roomId` | room add 的 --rooms 或 event create 的 --rooms |
 | `room list-groups` | `groups[].groupId` | room search 的 --group-id |
 | `book list` | `id`（如 `primary`） | event list/get 的 --calendar-id, book get/update 的 --id |
@@ -689,7 +716,8 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
 - **自动化校验**：凡涉及 `room add` / `event create --rooms` 的流程，`--rooms` 只能填上游 `room search`（或等价接口）返回 JSON 中的 **`rooms[].roomId`**；不得以会议室展示名、楼层文案或用户口语当作 `roomId`
 - **附件**：`attachment add` 仅负责挂载，**不上传**文件；fileId 必须先通过钉盘流程取得；`--files` 多附件用 `<fileId>:<name>` 元素逗号分隔
 - **日历本**：`book list` 返回的 `id` 才是合法 `calendarId`；如无明确说明，`event list` / `event get` 都不要带 `--calendar-id`，让接口默认走 primary 主日历
-- **分页查询**：`event list` 支持 `--limit`（控制每页条数，默认/最大 100）和 `--cursor`（翻页游标）；**首次查询无需传 `--cursor`**，仅当返回体中包含 `nextCursor` 时，将其作为 `--cursor` 传入可获取下一页
+- **分页查询**：`event list` / `event instances` 均支持 `--limit`（控制每页条数，默认/最大 100）和 `--cursor`（翻页游标）；**首次查询无需传 `--cursor`**，仅当返回体中包含 `nextCursor` 时，将其作为 `--cursor` 传入可获取下一页
+- **循环日程实例**：`event instances` 用于按时间范围展开重复日程（SeriesMaster）的每一个实例；**普通非循环日程调用该命令将查不到任何实例信息**
 - **日程提醒**：`event create` 支持 `--remind-minutes` 设置开始前提醒，逗号分隔多个分钟数（如 `--remind-minutes 5,10,15`），不传则默认15分钟提醒
 - **会议室分页**：`room search` 支持 `--limit`（每页条数，默认100，最大100）和 `--page`（分页起始位置，默认0），与 `room list-groups` 分页风格一致
 
@@ -703,5 +731,5 @@ dws calendar event list --start "2026-03-10T14:00:00+08:00" --end "2026-03-10T15
 
 ## 相关产品
 
-- 视频会议（conference）— 当前 CLI 不支持发起/预约/邀请入会/会中控制；请在钉钉客户端操作
+- conference（视频会议预约） — 仅视频会议预约（返回入会链接），不含参会人/会议室管理
 - [contact](../../dingtalk-contact/references/contact.md) — 搜索同事 userId，用于 attendee add --attendees