feat(event): add personal OA approval events

Use the Event-specific pre-release control and stream ticket endpoints by default.

internal/app/event_command.go

@@ -343,7 +343,7 @@ SIGTERM、关 stdin，或先用 dws event stop <subscribe_id> --dry-run 预览
 	f.StringVar(&personalOpts.GroupID, "group", "",
 		"group 规则：openConversationId")
 	f.StringVar(&personalOpts.ControlBaseURL, "personal-event-base-url", "",
-		"个人事件控制面 base URL；默认由 MCP base 派生 /dws")
+		"个人事件控制面 base URL；当前默认预发 https://pre-mcp.dingtalk.com/dws")
 	f.BoolVar(&personalOpts.DebugRawEvents, "debug-raw-events", false,
 		"个人事件联调：绕过本地 event type/subscribe_id 过滤，输出当前 personal stream bus 收到的所有事件")
 	f.StringVar(&streamOpts.Mode, "stream-ticket-mode", strings.TrimSpace(os.Getenv("DWS_STREAM_TICKET_MODE")),
@@ -351,7 +351,7 @@ SIGTERM、关 stdin，或先用 dws event stop <subscribe_id> --dry-run 预览
 	f.StringVar(&streamOpts.SourceID, "stream-source-id", strings.TrimSpace(os.Getenv("DWS_STREAM_SOURCE_ID")),
 		"个人 Stream sourceId；开源版默认 open，可由 edition 覆盖")
 	f.StringVar(&streamOpts.TicketURL, "stream-ticket-url", strings.TrimSpace(os.Getenv("DWS_STREAM_TICKET_URL")),
-		"个人 Stream 取票 URL；默认由 MCP base 派生 /stream/connections/ticket")
+		"个人 Stream 取票 URL；当前默认预发 https://pre-mcp.dingtalk.com/stream/connections/ticket")
 	hideEventInternalFlags(cmd, "as")
 	cli.AnnotateRuntimePositionals(cmd, cli.RuntimeSchemaPositional{
 		Name:        "event_key",
@@ -792,7 +792,7 @@ func newEventStatusCommand() *cobra.Command {
 	cmd.Flags().StringVar(&personalOpts.EventKey, "event", "", "个人事件 event_key 过滤")
 	cmd.Flags().StringVar(&personalOpts.Status, "status", "active", "个人订阅状态过滤: active|paused|error|deleted|all")
 	cmd.Flags().StringVar(&personalOpts.SubscribeID, "subscribe-id", "", "个人订阅 ID 过滤")
-	cmd.Flags().StringVar(&personalOpts.ControlBaseURL, "personal-event-base-url", "", "个人事件控制面 base URL；默认由 MCP base 派生 /dws")
+	cmd.Flags().StringVar(&personalOpts.ControlBaseURL, "personal-event-base-url", "", "个人事件控制面 base URL；当前默认预发 https://pre-mcp.dingtalk.com/dws")
 	cmd.Flags().StringVar(&personalOpts.StreamSourceID, "stream-source-id", strings.TrimSpace(os.Getenv("DWS_STREAM_SOURCE_ID")),
 		"个人事件 sourceId；开源版默认 open，可由 edition 覆盖")
 	hideEventInternalFlags(cmd, "as", "all", "all-editions", "client-id", "fail-on-orphan")
@@ -1090,7 +1090,7 @@ func newEventStopCommand() *cobra.Command {
 		},
 	}
 	cmd.Flags().StringVar(&asIdentity, "as", "user", "事件身份: user")
-	cmd.Flags().StringVar(&opts.ControlBaseURL, "personal-event-base-url", "", "个人事件控制面 base URL；默认由 MCP base 派生 /dws")
+	cmd.Flags().StringVar(&opts.ControlBaseURL, "personal-event-base-url", "", "个人事件控制面 base URL；当前默认预发 https://pre-mcp.dingtalk.com/dws")
 	cmd.Flags().StringVar(&opts.StreamSourceID, "stream-source-id", strings.TrimSpace(os.Getenv("DWS_STREAM_SOURCE_ID")),
 		"个人事件 sourceId；开源版默认 open，可由 edition 覆盖")
 	cmd.Flags().BoolVar(&opts.All, "all", false, "取消当前身份下本地记录的所有个人订阅")

internal/app/event_personal_command.go

@@ -38,11 +38,12 @@ import (
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/source"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/transport"
-	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/config"
 	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/pkg/edition"
 	"github.com/spf13/cobra"
 )
 
+const defaultPersonalEventMCPBaseURL = "https://pre-mcp.dingtalk.com"
+
 type commonConsumeOptions struct {
 	EventTypes []string
 	Filter     string
@@ -229,6 +230,9 @@ func runPersonalEventConsumeSingle(c *cobra.Command, opts personalConsumeOptions
 	if err := ensurePublicPersonalEvent(opts.EventKey); err != nil {
 		return err
 	}
+	if err := validatePersonalOAOptions(opts.EventKey, opts); err != nil {
+		return fmt.Errorf("event consume --as user: %w", err)
+	}
 	rawFormat := ""
 	if f := c.Flags().Lookup("format"); f != nil && f.Changed {
 		rawFormat = opts.Common.FormatRaw
@@ -567,6 +571,9 @@ func preparePersonalMultiOptions(opts personalConsumeOptions) ([]personalConsume
 		if !def.Public {
 			return nil, personal.PublicAvailabilityError(eventKey)
 		}
+		if err := validatePersonalOAOptions(eventKey, opts); err != nil {
+			return nil, err
+		}
 		switch def.RuleType {
 		case "singleChat", "sender":
 			hasUserScope = true
@@ -693,6 +700,9 @@ func applyPersonalConsumeFilters(cfg *consume.Config, opts personalConsumeOption
 }
 
 func validatePersonalSubscriptionOptions(opts personalConsumeOptions) error {
+	if err := validatePersonalOAOptions(opts.EventKey, opts); err != nil {
+		return err
+	}
 	if _, _, err := personal.BuildRuleParam(opts.EventKey, personal.RuleOptions{
 		RuleType:       opts.Rule,
 		UserID:         opts.UserID,
@@ -705,12 +715,49 @@ func validatePersonalSubscriptionOptions(opts personalConsumeOptions) error {
 	return err
 }
 
+func validatePersonalOAOptions(eventKey string, opts personalConsumeOptions) error {
+	changed := personalOAOptionNames(opts)
+	if len(changed) == 0 {
+		return nil
+	}
+	def, ok := personalLookupDefinition(strings.TrimSpace(eventKey))
+	if !ok || def.Category != "oa" {
+		return nil
+	}
+	return fmt.Errorf("%s not supported for OA event %s", strings.Join(changed, ", "), eventKey)
+}
+
+func personalOAOptionNames(opts personalConsumeOptions) []string {
+	var changed []string
+	for _, item := range []struct {
+		name  string
+		value string
+	}{
+		{name: "--user", value: opts.UserID},
+		{name: "--open-dingtalk-id", value: opts.OpenDingTalkID},
+		{name: "--group", value: opts.GroupID},
+		{name: "--query", value: opts.QueryCSV},
+		{name: "--filter-json", value: opts.FilterJSON},
+	} {
+		if strings.TrimSpace(item.value) != "" {
+			changed = append(changed, item.name)
+		}
+	}
+	return changed
+}
+
 func ensurePersonalSubscription(ctx context.Context, client *personal.Client, identity personal.Identity, opts personalConsumeOptions) (*personal.Subscription, string, string, error) {
 	if strings.TrimSpace(opts.SubscribeID) != "" {
 		sub, err := personalGetSubscription(client, ctx, opts.SubscribeID)
 		if err != nil {
 			return nil, "", "", err
 		}
+		lookupEventKey := strings.TrimSpace(sub.EventKey)
+		if lookupEventKey != "" {
+			if err := validatePersonalOAOptions(lookupEventKey, opts); err != nil {
+				return nil, "", "", err
+			}
+		}
 		eventKey := firstNonEmptyPersonalString(opts.EventKey, sub.EventKey)
 		if eventKey == "" {
 			return nil, "", "", fmt.Errorf("event_key is required when --subscribe-id lookup returns no event_key")
@@ -718,6 +765,9 @@ func ensurePersonalSubscription(ctx context.Context, client *personal.Client, id
 		if err := ensurePublicPersonalEvent(eventKey); err != nil {
 			return nil, "", "", err
 		}
+		if err := validatePersonalOAOptions(eventKey, opts); err != nil {
+			return nil, "", "", err
+		}
 		ruleType := firstNonEmptyPersonalString(sub.RuleType, opts.Rule)
 		if ruleType == "" {
 			if def, ok := personal.Lookup(eventKey); ok {
@@ -733,6 +783,9 @@ func ensurePersonalSubscription(ctx context.Context, client *personal.Client, id
 	if err := ensurePublicPersonalEvent(opts.EventKey); err != nil {
 		return nil, "", "", err
 	}
+	if err := validatePersonalOAOptions(opts.EventKey, opts); err != nil {
+		return nil, "", "", err
+	}
 	ruleType, ruleParam, err := personal.BuildRuleParam(opts.EventKey, personal.RuleOptions{
 		RuleType:       opts.Rule,
 		UserID:         opts.UserID,
@@ -1300,7 +1353,7 @@ func personalEventMCPBaseURL(configDir string) string {
 	if v := configuredMCPBaseURL(configDir); v != "" {
 		return strings.TrimRight(v, "/")
 	}
-	return strings.TrimRight(config.DefaultMCPBaseURL, "/")
+	return defaultPersonalEventMCPBaseURL
 }
 
 func configuredMCPBaseURL(configDir string) string {

internal/app/event_personal_identity_test.go

@@ -133,18 +133,18 @@ func TestResolvePersonalEventIdentityDefaultsSourceIDToOpen(t *testing.T) {
 	}
 }
 
-func TestPersonalEventDefaultsUseProductionWithoutMCPConfig(t *testing.T) {
+func TestPersonalEventDefaultsUsePreReleaseWithoutMCPConfig(t *testing.T) {
 	dir := t.TempDir()
 	t.Setenv("DWS_CONFIG_DIR", dir)
 	prev := edition.Get()
 	edition.Override(&edition.Hooks{})
 	t.Cleanup(func() { edition.Override(prev) })
 
-	if got := personalEventControlBaseURL("", dir); got != "https://mcp.dingtalk.com/dws" {
-		t.Fatalf("personalEventControlBaseURL() = %q, want production control URL", got)
+	if got := personalEventControlBaseURL("", dir); got != "https://pre-mcp.dingtalk.com/dws" {
+		t.Fatalf("personalEventControlBaseURL() = %q, want pre-release control URL", got)
 	}
-	if got := personalEventStreamTicketURL("", dir); got != "https://mcp.dingtalk.com/stream/connections/ticket" {
-		t.Fatalf("personalEventStreamTicketURL() = %q, want production ticket URL", got)
+	if got := personalEventStreamTicketURL("", dir); got != "https://pre-mcp.dingtalk.com/stream/connections/ticket" {
+		t.Fatalf("personalEventStreamTicketURL() = %q, want pre-release ticket URL", got)
 	}
 	if got := personalEventStreamSourceID(""); got != "open" {
 		t.Fatalf("personalEventStreamSourceID() = %q, want open", got)

internal/app/event_personal_oa_test.go

@@ -0,0 +1,484 @@
+// Copyright 2026 Alibaba Group
+// Licensed under the Apache License, Version 2.0 (the "License");
+
+package app
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"io"
+	"reflect"
+	"strings"
+	"testing"
+
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/consume"
+	"github.com/DingTalk-Real-AI/dingtalk-workspace-cli/internal/event/personal"
+	"github.com/spf13/cobra"
+)
+
+func TestPersonalOAEventListAndSchemaCommands(t *testing.T) {
+	list := newEventListCommand()
+	list.SilenceUsage = true
+	list.SilenceErrors = true
+	var listOut bytes.Buffer
+	list.SetOut(&listOut)
+	list.SetArgs([]string{"--category", "oa"})
+	if err := list.Execute(); err != nil {
+		t.Fatalf("event list --category oa error = %v", err)
+	}
+	for _, eventKey := range []string{
+		personal.EventOAApprovalTaskCreated,
+		personal.EventOAApprovalInstanceFinished,
+	} {
+		if !strings.Contains(listOut.String(), eventKey) {
+			t.Fatalf("OA event list missing %s:\n%s", eventKey, listOut.String())
+		}
+
+		schema := newEventSchemaCommand()
+		schema.SilenceUsage = true
+		schema.SilenceErrors = true
+		var schemaOut bytes.Buffer
+		schema.SetOut(&schemaOut)
+		schema.SetArgs([]string{eventKey, "--flatten"})
+		if err := schema.Execute(); err != nil {
+			t.Fatalf("event schema %s --flatten error = %v", eventKey, err)
+		}
+		var doc map[string]any
+		if err := json.Unmarshal(schemaOut.Bytes(), &doc); err != nil {
+			t.Fatalf("decode schema for %s: %v\n%s", eventKey, err, schemaOut.String())
+		}
+		if doc["event_key"] != eventKey || doc["rule_type"] != "all" || doc["jq_root_path"] != "." {
+			t.Fatalf("schema document for %s = %#v", eventKey, doc)
+		}
+		schemaBody, ok := doc["schema"].(map[string]any)
+		if !ok {
+			t.Fatalf("schema body for %s = %#v", eventKey, doc["schema"])
+		}
+		properties, ok := schemaBody["properties"].(map[string]any)
+		if !ok {
+			t.Fatalf("schema properties for %s = %#v", eventKey, schemaBody["properties"])
+		}
+		payload, ok := properties["payload"].(map[string]any)
+		if !ok || payload["additionalProperties"] != true {
+			t.Fatalf("schema payload for %s = %#v, want open object", eventKey, properties["payload"])
+		}
+	}
+	if strings.Contains(listOut.String(), personal.EventMention) {
+		t.Fatalf("OA category list leaked IM event:\n%s", listOut.String())
+	}
+}
+
+func TestPersonalOAEventConsumeDryRunAndValidation(t *testing.T) {
+	oldIdentity := personalResolveEventIdentity
+	t.Cleanup(func() { personalResolveEventIdentity = oldIdentity })
+	personalResolveEventIdentity = func(context.Context, string, string) (personal.Identity, error) {
+		return personal.Identity{
+			AccessToken:  "token",
+			LocalSubject: "subject",
+			ClientID:     "client",
+			SourceID:     "open",
+		}, nil
+	}
+
+	oaEvents := []string{
+		personal.EventOAApprovalTaskCreated,
+		personal.EventOAApprovalInstanceFinished,
+	}
+	for _, eventKey := range oaEvents {
+		t.Run(eventKey+"/dry-run", func(t *testing.T) {
+			cmd := newEventConsumeCommand()
+			cmd.SilenceUsage = true
+			cmd.SilenceErrors = true
+			var stderr bytes.Buffer
+			cmd.SetOut(io.Discard)
+			cmd.SetErr(&stderr)
+			cmd.SetArgs([]string{eventKey, "--dry-run"})
+			if err := cmd.Execute(); err != nil {
+				t.Fatalf("OA dry-run error = %v", err)
+			}
+			if !strings.Contains(stderr.String(), "event_types      : "+eventKey) {
+				t.Fatalf("OA dry-run does not select %s:\n%s", eventKey, stderr.String())
+			}
+		})
+
+		for _, args := range [][]string{
+			{"--user", "user-1"},
+			{"--open-dingtalk-id", "open-user-1"},
+			{"--group", "cid-1"},
+			{"--query", "urgent"},
+			{"--filter-json", `{"field":"content","op":"eq","value":"urgent"}`},
+		} {
+			name := strings.TrimPrefix(args[0], "--")
+			t.Run(eventKey+"/reject-"+name, func(t *testing.T) {
+				cmd := newEventConsumeCommand()
+				cmd.SilenceUsage = true
+				cmd.SilenceErrors = true
+				cmd.SetOut(io.Discard)
+				cmd.SetErr(io.Discard)
+				cmd.SetArgs(append([]string{eventKey}, append(args, "--dry-run")...))
+				err := cmd.Execute()
+				if err == nil || !strings.Contains(err.Error(), "not supported") {
+					t.Fatalf("OA consume %s error = %v, want unsupported option", args[0], err)
+				}
+			})
+		}
+	}
+
+	t.Run("multi-dry-run", func(t *testing.T) {
+		cmd := newEventConsumeCommand()
+		cmd.SilenceUsage = true
+		cmd.SilenceErrors = true
+		var stderr bytes.Buffer
+		cmd.SetOut(io.Discard)
+		cmd.SetErr(&stderr)
+		cmd.SetArgs([]string{oaEvents[0], oaEvents[1], "--dry-run"})
+		if err := cmd.Execute(); err != nil {
+			t.Fatalf("multi OA dry-run error = %v", err)
+		}
+		for _, eventKey := range oaEvents {
+			want := "event_key=" + eventKey + " rule_type=all rule_param={}"
+			if !strings.Contains(stderr.String(), want) {
+				t.Fatalf("multi OA dry-run missing %q:\n%s", want, stderr.String())
+			}
+		}
+	})
+
+	reuseOverrides := [][]string{
+		{"--user", "user-1"},
+		{"--open-dingtalk-id", "open-user-1"},
+		{"--group", "cid-1"},
+		{"--query", "urgent"},
+		{"--filter-json", `{"field":"content","op":"eq","value":"urgent"}`},
+	}
+	for _, explicitEventKey := range []bool{true, false} {
+		mode := "implicit-event-key"
+		if explicitEventKey {
+			mode = "explicit-event-key"
+		}
+		for _, override := range reuseOverrides {
+			flag := override[0]
+			t.Run("reuse-dry-run/"+mode+"/"+strings.TrimPrefix(flag, "--"), func(t *testing.T) {
+				args := make([]string, 0, 6)
+				if explicitEventKey {
+					args = append(args, personal.EventOAApprovalTaskCreated)
+				}
+				args = append(args, "--subscribe-id", "oa-sub-task", flag, override[1], "--dry-run")
+				cmd := newEventConsumeCommand()
+				cmd.SilenceUsage = true
+				cmd.SilenceErrors = true
+				cmd.SetOut(io.Discard)
+				cmd.SetErr(io.Discard)
+				cmd.SetArgs(args)
+				err := cmd.Execute()
+				if explicitEventKey {
+					if err == nil || !strings.Contains(err.Error(), flag+" not supported for OA event") {
+						t.Fatalf("explicit reused OA dry-run %s error = %v", flag, err)
+					}
+					return
+				}
+				if err != nil {
+					t.Fatalf("implicit reused IM-compatible subscription dry-run %s error = %v", flag, err)
+				}
+			})
+		}
+	}
+
+	for _, override := range reuseOverrides {
+		flag, value := override[0], override[1]
+		t.Run("multi-reject-"+strings.TrimPrefix(flag, "--"), func(t *testing.T) {
+			cmd := newEventConsumeCommand()
+			cmd.SilenceUsage = true
+			cmd.SilenceErrors = true
+			cmd.SetOut(io.Discard)
+			cmd.SetErr(io.Discard)
+			cmd.SetArgs([]string{oaEvents[0], oaEvents[1], flag, value, "--dry-run"})
+			err := cmd.Execute()
+			if err == nil || !strings.Contains(err.Error(), "not supported for OA event") {
+				t.Fatalf("multi OA consume %s error = %v", flag, err)
+			}
+		})
+	}
+
+	for _, test := range []struct {
+		name string
+		args []string
+	}{
+		{
+			name: "message query remains supported",
+			args: []string{personal.EventMention, "--query", "urgent", "--dry-run"},
+		},
+		{
+			name: "single group lifecycle filter remains supported",
+			args: []string{
+				personal.EventGroupUpdated,
+				"--group", "cid-1",
+				"--filter-json", `{"field":"future","op":"eq","value":"value"}`,
+				"--dry-run",
+			},
+		},
+	} {
+		t.Run(test.name, func(t *testing.T) {
+			cmd := newEventConsumeCommand()
+			cmd.SilenceUsage = true
+			cmd.SilenceErrors = true
+			cmd.SetOut(io.Discard)
+			cmd.SetErr(io.Discard)
+			cmd.SetArgs(test.args)
+			if err := cmd.Execute(); err != nil {
+				t.Fatalf("existing IM consume behavior changed: %v", err)
+			}
+		})
+	}
+}
+
+func TestPersonalOAMultiConsumeCreatesIndependentAllSubscriptionsOnSharedBus(t *testing.T) {
+	restoreMany := installPersonalManySeams(t)
+	defer restoreMany()
+	oldCreate := personalCreateSubscription
+	defer func() { personalCreateSubscription = oldCreate }()
+	t.Setenv("DWS_CONFIG_DIR", t.TempDir())
+
+	identity := personal.Identity{
+		AccessToken:  "token",
+		LocalSubject: "subject",
+		ClientID:     "client",
+		SourceID:     "open",
+	}
+	personalResolveEventIdentity = func(context.Context, string, string) (personal.Identity, error) {
+		return identity, nil
+	}
+	var requests []personal.CreateSubscriptionRequest
+	personalCreateSubscription = func(_ *personal.Client, _ context.Context, req personal.CreateSubscriptionRequest) (*personal.Subscription, error) {
+		requests = append(requests, req)
+		return &personal.Subscription{SubscribeID: "sub-" + req.EventKey}, nil
+	}
+	personalEnsureSubscription = ensurePersonalSubscription
+	var states []personal.RunState
+	personalUpsertRunState = func(_ string, state personal.RunState) error {
+		states = append(states, state)
+		return nil
+	}
+	personalDeleteSubscription = func(*personal.Client, context.Context, string) error { return nil }
+	personalRemoveRunStates = func(string, []string) error { return nil }
+	personalValidateConsumeConfig = func(consume.Config) error { return nil }
+	runManyCalls := 0
+	var gotSpecs []consume.ConsumerSpec
+	personalConsumeRunMany = func(_ context.Context, _ consume.Config, specs []consume.ConsumerSpec) error {
+		runManyCalls++
+		gotSpecs = append([]consume.ConsumerSpec(nil), specs...)
+		return nil
+	}
+
+	eventKeys := []string{
+		personal.EventOAApprovalTaskCreated,
+		personal.EventOAApprovalInstanceFinished,
+	}
+	if err := runPersonalEventConsume(newPersonalCoverageCommand(), personalConsumeOptions{
+		EventKeys: eventKeys,
+		Flatten:   true,
+	}); err != nil {
+		t.Fatalf("multi OA consume error = %v", err)
+	}
+	if runManyCalls != 1 {
+		t.Fatalf("RunMany calls = %d, want one shared-bus consume call", runManyCalls)
+	}
+	if len(requests) != len(eventKeys) || len(states) != len(eventKeys) || len(gotSpecs) != len(eventKeys) {
+		t.Fatalf("requests=%d states=%d specs=%d, want %d each", len(requests), len(states), len(gotSpecs), len(eventKeys))
+	}
+	for i, eventKey := range eventKeys {
+		req := requests[i]
+		if req.EventKey != eventKey || req.RuleType != "all" || req.RuleParam == nil || len(req.RuleParam) != 0 || req.Filter != nil {
+			t.Fatalf("subscription request[%d] = %#v, want %s all/{}", i, req, eventKey)
+		}
+		if states[i].EventKey != eventKey || states[i].RuleType != "all" {
+			t.Fatalf("run state[%d] = %#v", i, states[i])
+		}
+		wantSpec := consume.ConsumerSpec{
+			EventKey:         eventKey,
+			EventTypes:       []string{eventKey},
+			SubscribeID:      "sub-" + eventKey,
+			ReadySubscribeID: "sub-" + eventKey,
+		}
+		if !reflect.DeepEqual(gotSpecs[i], wantSpec) {
+			t.Fatalf("consumer spec[%d] = %#v, want %#v", i, gotSpecs[i], wantSpec)
+		}
+	}
+}
+
+func TestPersonalOAReusedSubscriptionRejectsDefinitionOverridesAtRuntime(t *testing.T) {
+	oldGet := personalGetSubscription
+	t.Cleanup(func() { personalGetSubscription = oldGet })
+	getCalls := 0
+	personalGetSubscription = func(*personal.Client, context.Context, string) (*personal.Subscription, error) {
+		getCalls++
+		return &personal.Subscription{
+			SubscribeID: "oa-sub-task",
+			EventKey:    personal.EventOAApprovalTaskCreated,
+			RuleType:    "all",
+		}, nil
+	}
+
+	tests := []struct {
+		name string
+		set  func(*personalConsumeOptions)
+	}{
+		{name: "user", set: func(opts *personalConsumeOptions) { opts.UserID = "user-1" }},
+		{name: "open-dingtalk-id", set: func(opts *personalConsumeOptions) { opts.OpenDingTalkID = "open-user-1" }},
+		{name: "group", set: func(opts *personalConsumeOptions) { opts.GroupID = "cid-1" }},
+		{name: "query", set: func(opts *personalConsumeOptions) { opts.QueryCSV = "urgent" }},
+		{name: "filter-json", set: func(opts *personalConsumeOptions) { opts.FilterJSON = `{"field":"content","op":"eq","value":"urgent"}` }},
+	}
+	for _, explicitEventKey := range []bool{true, false} {
+		mode := "implicit-event-key"
+		if explicitEventKey {
+			mode = "explicit-event-key"
+		}
+		for _, test := range tests {
+			t.Run(mode+"/"+test.name, func(t *testing.T) {
+				opts := personalConsumeOptions{SubscribeID: "oa-sub-task"}
+				if explicitEventKey {
+					opts.EventKey = personal.EventOAApprovalTaskCreated
+				}
+				test.set(&opts)
+				before := getCalls
+				_, _, _, err := ensurePersonalSubscription(
+					context.Background(),
+					nil,
+					personal.Identity{},
+					opts,
+				)
+				if err == nil || !strings.Contains(err.Error(), "--"+test.name+" not supported for OA event") {
+					t.Fatalf("reused OA subscription %s error = %v", test.name, err)
+				}
+				if getCalls != before+1 {
+					t.Fatalf("subscription lookup calls = %d, want %d", getCalls, before+1)
+				}
+			})
+		}
+	}
+}
+
+func TestPersonalOAImplicitReuseRuntimeLooksUpEventBeforeValidation(t *testing.T) {
+	oldIdentity := personalResolveEventIdentity
+	oldGet := personalGetSubscription
+	t.Cleanup(func() {
+		personalResolveEventIdentity = oldIdentity
+		personalGetSubscription = oldGet
+	})
+	personalResolveEventIdentity = func(context.Context, string, string) (personal.Identity, error) {
+		return personal.Identity{
+			AccessToken:  "token",
+			LocalSubject: "subject",
+			ClientID:     "client",
+			SourceID:     "open",
+		}, nil
+	}
+	getCalls := 0
+	personalGetSubscription = func(*personal.Client, context.Context, string) (*personal.Subscription, error) {
+		getCalls++
+		return &personal.Subscription{
+			SubscribeID: "oa-sub-task",
+			EventKey:    personal.EventOAApprovalTaskCreated,
+			RuleType:    "all",
+		}, nil
+	}
+
+	cmd := newEventConsumeCommand()
+	cmd.SilenceUsage = true
+	cmd.SilenceErrors = true
+	cmd.SetOut(io.Discard)
+	cmd.SetErr(io.Discard)
+	cmd.SetArgs([]string{"--subscribe-id", "oa-sub-task", "--group", "cid-1"})
+	err := cmd.Execute()
+	if err == nil || !strings.Contains(err.Error(), "--group not supported for OA event "+personal.EventOAApprovalTaskCreated) {
+		t.Fatalf("implicit reused OA runtime error = %v", err)
+	}
+	if getCalls != 1 {
+		t.Fatalf("subscription lookup calls = %d, want 1", getCalls)
+	}
+}
+
+func TestPersonalIMReusedSubscriptionWithExistingOverridesRemainsSupported(t *testing.T) {
+	oldGet := personalGetSubscription
+	t.Cleanup(func() { personalGetSubscription = oldGet })
+	personalGetSubscription = func(*personal.Client, context.Context, string) (*personal.Subscription, error) {
+		return &personal.Subscription{
+			SubscribeID: "im-sub",
+			EventKey:    personal.EventSingleChat,
+			RuleType:    "singleChat",
+		}, nil
+	}
+
+	sub, eventKey, ruleType, err := ensurePersonalSubscription(
+		context.Background(),
+		nil,
+		personal.Identity{},
+		personalConsumeOptions{
+			SubscribeID: "im-sub",
+			EventKey:    personal.EventSingleChat,
+			UserID:      "user-1",
+			QueryCSV:    "urgent",
+			FilterJSON:  `{"field":"content","op":"eq","value":"urgent"}`,
+		},
+	)
+	if err != nil {
+		t.Fatalf("reused IM subscription error = %v", err)
+	}
+	if sub.SubscribeID != "im-sub" || eventKey != personal.EventSingleChat || ruleType != "singleChat" {
+		t.Fatalf("reused IM subscription = %#v, event=%q rule=%q", sub, eventKey, ruleType)
+	}
+}
+
+func TestPersonalOAStatusAndStopCommandWiring(t *testing.T) {
+	oldStatus := eventRunPersonalStatus
+	oldStop := eventRunPersonalStop
+	t.Cleanup(func() {
+		eventRunPersonalStatus = oldStatus
+		eventRunPersonalStop = oldStop
+	})
+
+	var statusOpts personalStatusOptions
+	eventRunPersonalStatus = func(_ *cobra.Command, opts personalStatusOptions) error {
+		statusOpts = opts
+		return nil
+	}
+	status := newEventStatusCommand()
+	status.SilenceUsage = true
+	status.SilenceErrors = true
+	status.SetOut(io.Discard)
+	status.SetErr(io.Discard)
+	status.SetArgs([]string{
+		"--event", personal.EventOAApprovalTaskCreated,
+		"--subscribe-id", "oa-sub-task",
+		"--status", "all",
+	})
+	if err := status.Execute(); err != nil {
+		t.Fatalf("OA event status error = %v", err)
+	}
+	if statusOpts.EventKey != personal.EventOAApprovalTaskCreated ||
+		statusOpts.SubscribeID != "oa-sub-task" ||
+		statusOpts.Status != "all" {
+		t.Fatalf("OA status options = %#v", statusOpts)
+	}
+
+	var stopOpts personalStopOptions
+	eventRunPersonalStop = func(_ *cobra.Command, opts personalStopOptions) error {
+		stopOpts = opts
+		return nil
+	}
+	root := &cobra.Command{Use: "dws", SilenceUsage: true, SilenceErrors: true}
+	root.SetOut(io.Discard)
+	root.SetErr(io.Discard)
+	root.PersistentFlags().Bool("yes", false, "")
+	event := &cobra.Command{Use: "event"}
+	event.AddCommand(newEventStopCommand())
+	root.AddCommand(event)
+	root.SetArgs([]string{"event", "stop", "oa-sub-task", "--yes"})
+	if err := root.Execute(); err != nil {
+		t.Fatalf("OA event stop error = %v", err)
+	}
+	if stopOpts.SubscribeID != "oa-sub-task" || stopOpts.All {
+		t.Fatalf("OA stop options = %#v", stopOpts)
+	}
+}