feat(credential): resolve the active app from the invocation context

Each lark-cli invocation can now carry its own app identity, so callers
bound to different apps no longer compete over a single stored default
profile.

cmd/bootstrap.go

@@ -6,14 +6,21 @@ package cmd
 import (
 	"errors"
 	"io"
+	"os"
+	"strings"
 
 	"github.com/larksuite/cli/internal/cmdutil"
+	"github.com/larksuite/cli/internal/envvars"
 	"github.com/spf13/pflag"
 )
 
 // BootstrapInvocationContext extracts global invocation options before
 // the real command tree is built, so provider-backed config resolution sees
 // the correct profile from the start.
+//
+// Profile resolution: --profile flag > CliRuntimeAppID env > "" (defers to
+// MultiAppConfig.CurrentApp). The env value flows through FindApp which
+// matches by Name first, then by AppId — so callers can pass either form.
 func BootstrapInvocationContext(args []string) (cmdutil.InvocationContext, error) {
 	var globals GlobalOptions
 
@@ -26,5 +33,9 @@ func BootstrapInvocationContext(args []string) (cmdutil.InvocationContext, error
 	if err := fs.Parse(args); err != nil && !errors.Is(err, pflag.ErrHelp) {
 		return cmdutil.InvocationContext{}, err
 	}
-	return cmdutil.InvocationContext{Profile: globals.Profile}, nil
+	profile := globals.Profile
+	if profile == "" {
+		profile = strings.TrimSpace(os.Getenv(envvars.CliRuntimeAppID))
+	}
+	return cmdutil.InvocationContext{Profile: profile}, nil
 }

cmd/bootstrap_test.go

@@ -3,7 +3,11 @@
 
 package cmd
 
-import "testing"
+import (
+	"testing"
+
+	"github.com/larksuite/cli/internal/envvars"
+)
 
 func TestBootstrapInvocationContext_ProfileFlag(t *testing.T) {
 	inv, err := BootstrapInvocationContext([]string{"--profile", "target", "auth", "status"})
@@ -70,3 +74,25 @@ func TestBootstrapInvocationContext_HelpWithProfile(t *testing.T) {
 		t.Fatalf("profile = %q, want %q", inv.Profile, "target")
 	}
 }
+
+func TestBootstrapInvocationContext_RuntimeAppIDEnv(t *testing.T) {
+	t.Setenv(envvars.CliRuntimeAppID, " env-app ")
+	inv, err := BootstrapInvocationContext([]string{"auth", "status"})
+	if err != nil {
+		t.Fatalf("BootstrapInvocationContext() error = %v", err)
+	}
+	if inv.Profile != "env-app" {
+		t.Fatalf("profile = %q, want %q (env value, trimmed)", inv.Profile, "env-app")
+	}
+}
+
+func TestBootstrapInvocationContext_ProfileFlagBeatsRuntimeAppIDEnv(t *testing.T) {
+	t.Setenv(envvars.CliRuntimeAppID, "env-app")
+	inv, err := BootstrapInvocationContext([]string{"--profile", "flag-app", "auth", "status"})
+	if err != nil {
+		t.Fatalf("BootstrapInvocationContext() error = %v", err)
+	}
+	if inv.Profile != "flag-app" {
+		t.Fatalf("profile = %q, want %q (--profile wins over env)", inv.Profile, "flag-app")
+	}
+}

cmd/config/bind.go

@@ -7,6 +7,7 @@ import (
 	"encoding/json"
 	"fmt"
 	"os"
+	"sort"
 	"strings"
 
 	"github.com/charmbracelet/huh"
@@ -55,6 +56,11 @@ type BindOptions struct {
 	// configBindRun; downstream branches read this instead of rechecking
 	// IOStreams.IsTerminal. Do not set from outside — it is overwritten.
 	IsTUI bool
+
+	// All binds every account exposed by the agent source as a multi-app
+	// config. Hidden flag, currently openclaw source only, mutually
+	// exclusive with --app-id.
+	All bool
 }
 
 // NewCmdConfigBind creates the config bind subcommand.
@@ -107,6 +113,8 @@ Interactive terminal use: run with no flags to enter the TUI form.`,
 	cmd.Flags().StringVar(&opts.Identity, "identity", "", "identity preset (bot-only|user-default); defaults to bot-only in flag mode (safer: no impersonation)")
 	cmd.Flags().BoolVar(&opts.Force, "force", false, "confirm a risky transition (currently: bot-only → user-default identity change in flag mode)")
 	cmd.Flags().StringVar(&opts.Lang, "lang", "", "language preference (e.g. zh or zh_cn)")
+	cmd.Flags().BoolVar(&opts.All, "all", false, "bind every account from the agent source as a multi-app config (currently openclaw source only)")
+	_ = cmd.Flags().MarkHidden("all")
 	cmdutil.SetRisk(cmd, "write")
 
 	return cmd
@@ -124,6 +132,10 @@ func configBindRun(opts *BindOptions) error {
 	// opts.IsTUI instead of re-checking IOStreams.IsTerminal.
 	opts.IsTUI = opts.Source == "" && opts.Factory.IOStreams.IsTerminal
 
+	if opts.All {
+		return configBindAllRun(opts)
+	}
+
 	source, err := finalizeSource(opts)
 	if err != nil {
 		return err
@@ -632,6 +644,15 @@ func validateBindFlags(opts *BindOptions) error {
 			return errs.NewValidationError(errs.SubtypeInvalidArgument, "invalid --identity %q; valid values: bot-only, user-default", opts.Identity).WithParam("--identity")
 		}
 	}
+	if opts.All {
+		if opts.AppID != "" {
+			return errs.NewValidationError(errs.SubtypeInvalidArgument, "--all is mutually exclusive with --app-id").WithParam("--all")
+		}
+		src := strings.TrimSpace(strings.ToLower(opts.Source))
+		if src != "" && src != "openclaw" {
+			return errs.NewValidationError(errs.SubtypeInvalidArgument, "--all only supports --source openclaw").WithParam("--all")
+		}
+	}
 	lang, err := cmdutil.ParseLangFlag(opts.Lang)
 	if err != nil {
 		return err
@@ -677,3 +698,173 @@ func tuiSelectIdentity(opts *BindOptions) (string, error) {
 	}
 	return value, nil
 }
+
+// configBindAllRun binds every account exposed by the agent source as a
+// multi-app config. Flag-mode only; bot-only identity by default;
+// silently overwrites prior bindings.
+func configBindAllRun(opts *BindOptions) error {
+	// --all is flag-mode only: resolve the source through the same flag/env
+	// reconciliation as a single bind (including the explicit-vs-detected
+	// mismatch guard), but never prompt.
+	opts.IsTUI = false
+	source, err := finalizeSource(opts)
+	if err != nil {
+		return err
+	}
+	if source != "openclaw" {
+		return errs.NewValidationError(errs.SubtypeInvalidArgument,
+			"--all only supports --source openclaw").WithParam("--all")
+	}
+	core.SetCurrentWorkspace(core.Workspace(source))
+	targetConfigPath := core.GetConfigPath()
+	previousConfigBytes, _ := vfs.ReadFile(targetConfigPath)
+
+	binder, err := newBinder(source, opts)
+	if err != nil {
+		return err
+	}
+	candidates, err := binder.ListCandidates()
+	if err != nil {
+		return err
+	}
+	if len(candidates) == 0 {
+		return errs.NewConfigError(errs.SubtypeNotConfigured,
+			"no Feishu accounts found in openclaw.json").
+			WithHint("configure channels.feishu.accounts first")
+	}
+
+	if opts.Identity == "" {
+		opts.Identity = "bot-only"
+	}
+	if err := warnIdentityEscalation(opts, previousConfigBytes); err != nil {
+		return err
+	}
+	noticeUserDefaultRisk(opts)
+	prior := priorLang(previousConfigBytes)
+
+	oc, ok := binder.(*openclawBinder)
+	if !ok {
+		return errs.NewInternalError(errs.SubtypeSDKError, "internal: --all requires openclawBinder, got %T", binder)
+	}
+	apps, err := oc.BuildAll()
+	if err != nil {
+		return err
+	}
+	for i := range apps {
+		applyPreferences(&apps[i], opts, prior)
+	}
+	sortAppsDefaultFirst(apps)
+	apps = dedupAndOrderApps(apps)
+
+	return commitMultiAppBinding(opts, apps, previousConfigBytes, source, targetConfigPath)
+}
+
+// sortAppsDefaultFirst orders the bound apps deterministically: the "default"
+// alias first, then named accounts alphabetically. Account enumeration
+// upstream follows Go map order, and CurrentApp is taken from the first
+// entry — without a stable order the selected default app would vary
+// between runs of the same bind.
+func sortAppsDefaultFirst(apps []core.AppConfig) {
+	sort.SliceStable(apps, func(i, j int) bool {
+		di, dj := strings.EqualFold(apps[i].Name, "default"), strings.EqualFold(apps[j].Name, "default")
+		if di != dj {
+			return di
+		}
+		return apps[i].Name < apps[j].Name
+	})
+}
+
+// dedupAndOrderApps collapses entries that share an AppId. Per OpenClaw
+// semantics, "default" is an implicit fallback alias that inherits the
+// top-level appId; when an explicit named account exposes the same appId,
+// "default" is redundant and discarded. Insertion order is preserved.
+func dedupAndOrderApps(apps []core.AppConfig) []core.AppConfig {
+	byID := make(map[string]int, len(apps))
+	out := make([]core.AppConfig, 0, len(apps))
+	for _, a := range apps {
+		if idx, seen := byID[a.AppId]; seen {
+			if strings.EqualFold(out[idx].Name, "default") && !strings.EqualFold(a.Name, "default") {
+				out[idx] = a
+			}
+			continue
+		}
+		byID[a.AppId] = len(out)
+		out = append(out, a)
+	}
+	return out
+}
+
+// commitMultiAppBinding writes a MultiAppConfig containing every bound
+// account. CurrentApp defaults to the first entry so existing single-app
+// callers keep their behaviour; callers select another profile with
+// --profile or LARKSUITE_CLI_RUNTIME_APP_ID.
+func commitMultiAppBinding(opts *BindOptions, apps []core.AppConfig, previousConfigBytes []byte, source, configPath string) error {
+	multi := &core.MultiAppConfig{
+		CurrentApp: apps[0].ProfileName(),
+		Apps:       apps,
+	}
+
+	if err := vfs.MkdirAll(core.GetConfigDir(), 0700); err != nil {
+		return errs.NewInternalError(errs.SubtypeFileIO, "failed to create workspace directory: %v", err).WithCause(err)
+	}
+	data, err := json.MarshalIndent(multi, "", "  ")
+	if err != nil {
+		return errs.NewInternalError(errs.SubtypeStorage, "failed to marshal config: %v", err).WithCause(err)
+	}
+	if err := validate.AtomicWrite(configPath, append(data, '\n'), 0600); err != nil {
+		return errs.NewInternalError(errs.SubtypeStorage, "failed to write config %s: %v", configPath, err).WithCause(err)
+	}
+
+	replaced := previousConfigBytes != nil
+	if replaced {
+		cleanupKeychainFromDataMulti(opts.Factory.Keychain, previousConfigBytes, apps)
+	}
+
+	appIDs := make([]string, len(apps))
+	for i := range apps {
+		appIDs[i] = apps[i].AppId
+	}
+
+	uiMsg := getBindMsg(opts.UILang)
+	display := sourceDisplayName(source)
+	fmt.Fprintln(opts.Factory.IOStreams.ErrOut,
+		fmt.Sprintf(uiMsg.BindSuccessHeader, display)+"\n"+
+			fmt.Sprintf("bound %d Feishu apps: %s", len(appIDs), strings.Join(appIDs, ", ")))
+
+	envelope := map[string]interface{}{
+		"ok":          true,
+		"workspace":   source,
+		"app_ids":     appIDs,
+		"config_path": configPath,
+		"replaced":    replaced,
+		"identity":    opts.Identity,
+		"all":         true,
+	}
+	resultJSON, _ := json.Marshal(envelope)
+	fmt.Fprintln(opts.Factory.IOStreams.Out, string(resultJSON))
+	return nil
+}
+
+// cleanupKeychainFromDataMulti is cleanupKeychainFromData with a keep set,
+// used when re-binding multiple apps in one pass.
+func cleanupKeychainFromDataMulti(kc keychain.KeychainAccess, data []byte, keep []core.AppConfig) {
+	var multi core.MultiAppConfig
+	if err := json.Unmarshal(data, &multi); err != nil {
+		return
+	}
+	keepIDs := make(map[string]struct{}, len(keep))
+	for i := range keep {
+		ref := keep[i].AppSecret.Ref
+		if ref != nil && ref.Source == "keychain" {
+			keepIDs[ref.ID] = struct{}{}
+		}
+	}
+	for _, app := range multi.Apps {
+		if app.AppSecret.Ref != nil && app.AppSecret.Ref.Source == "keychain" {
+			if _, ok := keepIDs[app.AppSecret.Ref.ID]; ok {
+				continue
+			}
+		}
+		core.RemoveSecretStore(app.AppSecret, kc)
+	}
+}

cmd/config/bind_test.go

@@ -1907,3 +1907,270 @@ func TestConfigBindRun_LangExplicit_PrintsConfirmation(t *testing.T) {
 		t.Errorf("stderr = %q, want it to contain confirmation %q", got, want)
 	}
 }
+
+// ── --all flag tests (hidden multi-account bind) ──
+
+func TestConfigBindCmd_AllFlagParsing(t *testing.T) {
+	f, _, _, _ := cmdutil.TestFactory(t, nil)
+
+	var gotOpts *BindOptions
+	cmd := NewCmdConfigBind(f, func(opts *BindOptions) error {
+		gotOpts = opts
+		return nil
+	})
+	cmd.SetArgs([]string{"--source", "openclaw", "--all"})
+	if err := cmd.Execute(); err != nil {
+		t.Fatalf("unexpected error: %v", err)
+	}
+	if !gotOpts.All {
+		t.Error("expected All=true")
+	}
+
+	// --all is hidden but still parseable.
+	flag := cmd.Flags().Lookup("all")
+	if flag == nil {
+		t.Fatal("--all flag must be registered")
+	}
+	if !flag.Hidden {
+		t.Error("--all must be hidden")
+	}
+}
+
+// requireAllValidationParam asserts err is a typed validation error of
+// subtype invalid_argument carrying the named param, locking the contract
+// beyond the prose message.
+func requireAllValidationParam(t *testing.T, err error, wantParam string) {
+	t.Helper()
+	var ve *errs.ValidationError
+	if !errors.As(err, &ve) {
+		t.Fatalf("expected *errs.ValidationError, got %T (%v)", err, err)
+	}
+	if ve.Category != errs.CategoryValidation || ve.Subtype != errs.SubtypeInvalidArgument {
+		t.Fatalf("expected validation/invalid_argument, got %s/%s", ve.Category, ve.Subtype)
+	}
+	if ve.Param != wantParam {
+		t.Fatalf("param = %q, want %q", ve.Param, wantParam)
+	}
+}
+
+func TestConfigBindRun_AllRejectsAppID(t *testing.T) {
+	f, _, _, _ := cmdutil.TestFactory(t, nil)
+	err := configBindRun(&BindOptions{
+		Factory: f,
+		Source:  "openclaw",
+		AppID:   "cli_x",
+		All:     true,
+	})
+	if err == nil {
+		t.Fatal("expected validation error")
+	}
+	if !strings.Contains(err.Error(), "--all is mutually exclusive with --app-id") {
+		t.Errorf("unexpected error: %v", err)
+	}
+	requireAllValidationParam(t, err, "--all")
+}
+
+func TestDedupAndOrderApps_NamedWinsOverDefault(t *testing.T) {
+	// OpenClaw semantics: "default" is an implicit fallback alias; when a
+	// named account exposes the same appId, default is dropped.
+	apps := []core.AppConfig{
+		{Name: "acct-a", AppId: "cli_main"},
+		{Name: "acct-b", AppId: "cli_product"},
+		{Name: "default", AppId: "cli_main"},
+	}
+	got := dedupAndOrderApps(apps)
+	if len(got) != 2 {
+		t.Fatalf("len=%d, want 2", len(got))
+	}
+	if got[0].Name != "acct-a" || got[0].AppId != "cli_main" {
+		t.Errorf("apps[0]=%+v, want acct-a cli_main", got[0])
+	}
+	if got[1].Name != "acct-b" {
+		t.Errorf("apps[1]=%+v, want acct-b", got[1])
+	}
+}
+
+func TestDedupAndOrderApps_DefaultFirstThenNamedOverwrites(t *testing.T) {
+	// default arrives before its named alias — named still wins.
+	apps := []core.AppConfig{
+		{Name: "default", AppId: "cli_main"},
+		{Name: "acct-a", AppId: "cli_main"},
+	}
+	got := dedupAndOrderApps(apps)
+	if len(got) != 1 {
+		t.Fatalf("len=%d, want 1", len(got))
+	}
+	if got[0].Name != "acct-a" {
+		t.Errorf("apps[0]=%+v, want acct-a", got[0])
+	}
+}
+
+func TestDedupAndOrderApps_NoDuplicatesPreservesOrder(t *testing.T) {
+	apps := []core.AppConfig{
+		{Name: "acct-a", AppId: "cli_main"},
+		{Name: "acct-b", AppId: "cli_product"},
+		{Name: "acct-c", AppId: "cli_extra"},
+	}
+	got := dedupAndOrderApps(apps)
+	if len(got) != 3 {
+		t.Fatalf("len=%d, want 3", len(got))
+	}
+	want := []string{"acct-a", "acct-b", "acct-c"}
+	for i := range got {
+		if got[i].Name != want[i] {
+			t.Errorf("apps[%d]=%q, want %q", i, got[i].Name, want[i])
+		}
+	}
+}
+
+func TestDedupAndOrderApps_DefaultAlreadyFirst(t *testing.T) {
+	apps := []core.AppConfig{
+		{Name: "default", AppId: "cli_main"},
+		{Name: "acct-b", AppId: "cli_product"},
+	}
+	got := dedupAndOrderApps(apps)
+	if len(got) != 2 || got[0].Name != "default" {
+		t.Fatalf("got %+v, want default first", got)
+	}
+}
+
+func TestConfigBindRun_AllRejectsNonOpenClawSource(t *testing.T) {
+	f, _, _, _ := cmdutil.TestFactory(t, nil)
+	err := configBindRun(&BindOptions{
+		Factory: f,
+		Source:  "hermes",
+		All:     true,
+	})
+	if err == nil {
+		t.Fatal("expected validation error")
+	}
+	if !strings.Contains(err.Error(), "--all only supports --source openclaw") {
+		t.Errorf("unexpected error: %v", err)
+	}
+	requireAllValidationParam(t, err, "--all")
+}
+
+func TestConfigBindRun_AllRejectsSourceEnvMismatch(t *testing.T) {
+	saveWorkspace(t)
+	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
+	clearAgentEnv(t)
+	t.Setenv("HERMES_HOME", t.TempDir()) // env says hermes, flag says openclaw
+
+	f, _, _, _ := cmdutil.TestFactory(t, nil)
+	err := configBindRun(&BindOptions{Factory: f, Source: "openclaw", All: true})
+	if err == nil {
+		t.Fatal("expected source/env mismatch error")
+	}
+	if !strings.Contains(err.Error(), "does not match detected Agent environment") {
+		t.Errorf("unexpected error: %v", err)
+	}
+	requireAllValidationParam(t, err, "--source")
+}
+
+func TestConfigBindRun_AllBindsEveryAccount(t *testing.T) {
+	saveWorkspace(t)
+	configDir := t.TempDir()
+	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", configDir)
+	clearAgentEnv(t)
+
+	openclawHome := t.TempDir()
+	t.Setenv("OPENCLAW_HOME", openclawHome)
+	openclawDir := filepath.Join(openclawHome, ".openclaw")
+	if err := os.MkdirAll(openclawDir, 0700); err != nil {
+		t.Fatalf("mkdir: %v", err)
+	}
+	// Top-level credentials surface as a "default" candidate aliasing the
+	// "work" account; dedup must collapse them and keep the named entry.
+	cfg := `{"channels":{"feishu":{"appId":"cli_main","appSecret":"sec_main","domain":"feishu","accounts":{"work":{"appId":"cli_main","appSecret":"sec_main"},"personal":{"appId":"cli_personal","appSecret":"sec_personal"}}}}}`
+	if err := os.WriteFile(filepath.Join(openclawDir, "openclaw.json"), []byte(cfg), 0600); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	f, stdout, _, _ := cmdutil.TestFactory(t, nil)
+	err := configBindRun(&BindOptions{Factory: f, All: true})
+	if err != nil {
+		t.Fatalf("expected success, got %v", err)
+	}
+
+	configPath := filepath.Join(configDir, "openclaw", "config.json")
+	assertEnvelope(t, stdout.Bytes(), map[string]any{
+		"ok":          true,
+		"workspace":   "openclaw",
+		"app_ids":     []any{"cli_main", "cli_personal"},
+		"config_path": configPath,
+		"replaced":    false,
+		"identity":    "bot-only",
+		"all":         true,
+	})
+
+	data, err := os.ReadFile(configPath)
+	if err != nil {
+		t.Fatalf("read written config: %v", err)
+	}
+	var multi core.MultiAppConfig
+	if err := json.Unmarshal(data, &multi); err != nil {
+		t.Fatalf("written config is not a MultiAppConfig: %v", err)
+	}
+	if len(multi.Apps) != 2 {
+		t.Fatalf("apps = %d, want 2", len(multi.Apps))
+	}
+	// The account aliased by top-level "default" stays first and becomes
+	// the initial CurrentApp.
+	if multi.CurrentApp != "work" {
+		t.Errorf("currentApp = %q, want %q", multi.CurrentApp, "work")
+	}
+	if multi.Apps[0].Name != "work" || multi.Apps[1].Name != "personal" {
+		t.Errorf("app order = [%q, %q], want [work, personal]", multi.Apps[0].Name, multi.Apps[1].Name)
+	}
+}
+
+func TestConfigBindRun_AllEscalationRequiresForce(t *testing.T) {
+	saveWorkspace(t)
+	configDir := t.TempDir()
+	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", configDir)
+	clearAgentEnv(t)
+
+	openclawHome := t.TempDir()
+	t.Setenv("OPENCLAW_HOME", openclawHome)
+	openclawDir := filepath.Join(openclawHome, ".openclaw")
+	if err := os.MkdirAll(openclawDir, 0700); err != nil {
+		t.Fatalf("mkdir: %v", err)
+	}
+	cfg := `{"channels":{"feishu":{"appId":"cli_main","appSecret":"sec_main","domain":"feishu"}}}`
+	if err := os.WriteFile(filepath.Join(openclawDir, "openclaw.json"), []byte(cfg), 0600); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	// Pre-existing binding locked to bot-only.
+	ocDir := filepath.Join(configDir, "openclaw")
+	if err := os.MkdirAll(ocDir, 0700); err != nil {
+		t.Fatalf("mkdir: %v", err)
+	}
+	locked := `{"apps":[{"appId":"cli_old","strictMode":"bot"}]}`
+	if err := os.WriteFile(filepath.Join(ocDir, "config.json"), []byte(locked), 0600); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	f, _, _, _ := cmdutil.TestFactory(t, nil)
+	err := configBindRun(&BindOptions{Factory: f, All: true, Identity: "user-default"})
+	if err == nil {
+		t.Fatal("expected confirmation-required error without --force")
+	}
+	var ce *errs.ConfirmationRequiredError
+	if !errors.As(err, &ce) {
+		t.Fatalf("expected *errs.ConfirmationRequiredError, got %T (%v)", err, err)
+	}
+
+	// Same escalation with --force proceeds.
+	f2, stdout, _, _ := cmdutil.TestFactory(t, nil)
+	if err := configBindRun(&BindOptions{Factory: f2, All: true, Identity: "user-default", Force: true}); err != nil {
+		t.Fatalf("expected --force to allow escalation, got %v", err)
+	}
+	envelope := map[string]any{}
+	if err := json.Unmarshal(stdout.Bytes(), &envelope); err != nil {
+		t.Fatalf("invalid JSON output: %v", err)
+	}
+	if envelope["replaced"] != true || envelope["identity"] != "user-default" {
+		t.Errorf("envelope = %v, want replaced=true identity=user-default", envelope)
+	}
+}

cmd/config/binder.go

@@ -172,40 +172,54 @@ func (b *openclawBinder) Build(appID string) (*core.AppConfig, error) {
 	if b.cfg == nil {
 		return nil, errs.NewInternalError(errs.SubtypeSDKError, "internal: Build called before ListCandidates")
 	}
-
-	var selected *binding.CandidateApp
 	for i := range b.rawApps {
 		if b.rawApps[i].AppID == appID {
-			selected = &b.rawApps[i]
-			break
+			return b.buildFromCandidate(&b.rawApps[i])
 		}
 	}
-	if selected == nil {
-		return nil, errs.NewInternalError(errs.SubtypeSDKError, "internal: appID %q not in candidates", appID)
-	}
+	return nil, errs.NewInternalError(errs.SubtypeSDKError, "internal: appID %q not in candidates", appID)
+}
 
-	if selected.AppSecret.IsZero() {
-		return nil, errs.NewConfigError(errs.SubtypeInvalidClient, "appSecret is empty for app %s in %s", selected.AppID, b.path).
+// BuildAll builds every candidate without appID lookup so duplicate-appID
+// entries (e.g. OpenClaw's "default" alias of a named account) keep their own
+// Label. --all callers must use this; Build(appID) collapses duplicates.
+func (b *openclawBinder) BuildAll() ([]core.AppConfig, error) {
+	if b.cfg == nil {
+		return nil, errs.NewInternalError(errs.SubtypeSDKError, "internal: BuildAll called before ListCandidates")
+	}
+	out := make([]core.AppConfig, 0, len(b.rawApps))
+	for i := range b.rawApps {
+		cfg, err := b.buildFromCandidate(&b.rawApps[i])
+		if err != nil {
+			return nil, err
+		}
+		out = append(out, *cfg)
+	}
+	return out, nil
+}
+
+func (b *openclawBinder) buildFromCandidate(c *binding.CandidateApp) (*core.AppConfig, error) {
+	if c.AppSecret.IsZero() {
+		return nil, errs.NewConfigError(errs.SubtypeInvalidClient, "appSecret is empty for app %s in %s", c.AppID, b.path).
 			WithHint("configure channels.feishu.appSecret in openclaw.json")
 	}
-	secret, err := binding.ResolveSecretInput(selected.AppSecret, b.cfg.Secrets, os.Getenv)
+	secret, err := binding.ResolveSecretInput(c.AppSecret, b.cfg.Secrets, os.Getenv)
 	if err != nil {
-		return nil, errs.NewConfigError(errs.SubtypeInvalidClient, "failed to resolve appSecret for %s: %v", selected.AppID, err).
+		return nil, errs.NewConfigError(errs.SubtypeInvalidClient, "failed to resolve appSecret for %s: %v", c.AppID, err).
 			WithHint("check appSecret configuration in %s", b.path).
 			WithCause(err)
 	}
-
-	stored, err := core.ForStorage(selected.AppID, core.PlainSecret(secret), b.opts.Factory.Keychain)
+	stored, err := core.ForStorage(c.AppID, core.PlainSecret(secret), b.opts.Factory.Keychain)
 	if err != nil {
 		return nil, errs.NewInternalError(errs.SubtypeStorage, "keychain unavailable: %v", err).
 			WithHint("use file: reference in config to bypass keychain").
 			WithCause(err)
 	}
-
 	return &core.AppConfig{
-		AppId:     selected.AppID,
+		Name:      c.Label,
+		AppId:     c.AppID,
 		AppSecret: stored,
-		Brand:     core.LarkBrand(normalizeBrand(selected.Brand)),
+		Brand:     core.LarkBrand(normalizeBrand(c.Brand)),
 	}, nil
 }
 

cmd/event/format_helpers_test.go

@@ -143,6 +143,79 @@ func TestWriteStatusText_CoversAllStates(t *testing.T) {
 	}
 }
 
+func TestWriteStatusText_ShowsSubColumn(t *testing.T) {
+	var buf bytes.Buffer
+	writeStatusText(&buf, []appStatus{
+		{
+			AppID:     "cli_RUNNINGXXXXXXXXX",
+			State:     stateRunning,
+			PID:       1234,
+			UptimeSec: 60,
+			Active:    2,
+			Consumers: []protocol.ConsumerInfo{
+				{PID: 1001, EventKey: "mail.x", SubscriptionID: "mail.x:alice", Received: 5, Dropped: 0},
+				{PID: 1002, EventKey: "mail.x", SubscriptionID: "mail.x:bob", Received: 3, Dropped: 0},
+			},
+		},
+	})
+	out := buf.String()
+	if !strings.Contains(out, "SUB") {
+		t.Errorf("missing SUB column header: %s", out)
+	}
+	if !strings.Contains(out, "alice") {
+		t.Errorf("missing alice suffix in SUB column: %s", out)
+	}
+	if !strings.Contains(out, "bob") {
+		t.Errorf("missing bob suffix in SUB column: %s", out)
+	}
+}
+
+func TestWriteStatusText_LegacySubscriptionID_RendersDash(t *testing.T) {
+	var buf bytes.Buffer
+	writeStatusText(&buf, []appStatus{
+		{
+			AppID:     "cli_RUNNINGXXXXXXXXX",
+			State:     stateRunning,
+			PID:       1234,
+			UptimeSec: 60,
+			Active:    1,
+			Consumers: []protocol.ConsumerInfo{
+				{PID: 1001, EventKey: "im.x", SubscriptionID: "", Received: 5},
+			},
+		},
+	})
+	out := buf.String()
+	if !strings.Contains(out, "SUB") {
+		t.Errorf("missing SUB header: %s", out)
+	}
+	if !strings.Contains(out, "-") {
+		t.Errorf("missing dash placeholder for empty SubscriptionID: %s", out)
+	}
+}
+
+func TestWriteStatusText_EventKeyEqualSubscriptionID_RendersDash(t *testing.T) {
+	var buf bytes.Buffer
+	writeStatusText(&buf, []appStatus{
+		{
+			AppID:     "cli_RUNNINGXXXXXXXXX",
+			State:     stateRunning,
+			PID:       1234,
+			UptimeSec: 60,
+			Active:    1,
+			Consumers: []protocol.ConsumerInfo{
+				{PID: 1001, EventKey: "im.x", SubscriptionID: "im.x", Received: 5},
+			},
+		},
+	})
+	out := buf.String()
+	if !strings.Contains(out, "SUB") {
+		t.Errorf("missing SUB header: %s", out)
+	}
+	if !strings.Contains(out, "-") {
+		t.Errorf("missing dash placeholder when SubscriptionID==EventKey: %s", out)
+	}
+}
+
 func TestWriteStatusJSON_OrphanHint(t *testing.T) {
 	var buf bytes.Buffer
 	if err := writeStatusJSON(&buf, []appStatus{

cmd/event/schema.go

@@ -134,12 +134,16 @@ func runSchema(f *cmdutil.Factory, key string, asJSON bool) error {
 	if len(def.Params) > 0 {
 		fmt.Fprintf(out, "\nParameters:\n")
 		w := tabwriter.NewWriter(out, 0, 4, 2, ' ', 0)
-		fmt.Fprintf(w, "  NAME\tTYPE\tREQUIRED\tDEFAULT\tDESCRIPTION\n")
+		fmt.Fprintf(w, "  NAME\tTYPE\tREQUIRED\tSUB-KEY\tDEFAULT\tDESCRIPTION\n")
 		for _, p := range def.Params {
 			required := "no"
 			if p.Required {
 				required = "yes"
 			}
+			subKey := "no"
+			if p.SubscriptionKey {
+				subKey = "yes"
+			}
 			defaultVal := p.Default
 			if defaultVal == "" {
 				defaultVal = "-"
@@ -148,7 +152,7 @@ func runSchema(f *cmdutil.Factory, key string, asJSON bool) error {
 			if desc == "" {
 				desc = "-"
 			}
-			fmt.Fprintf(w, "  %s\t%s\t%s\t%s\t%s\n", p.Name, p.Type, required, defaultVal, desc)
+			fmt.Fprintf(w, "  %s\t%s\t%s\t%s\t%s\t%s\n", p.Name, p.Type, required, subKey, defaultVal, desc)
 		}
 		w.Flush()
 

cmd/event/schema_test.go

@@ -96,6 +96,79 @@ func TestRunSchema_JSONOutput(t *testing.T) {
 	}
 }
 
+func TestSchema_RendersSubscriptionKeyMarker(t *testing.T) {
+	const syntheticKey = "test.evt_sub"
+	t.Cleanup(func() { eventlib.UnregisterKeyForTest(syntheticKey) })
+
+	eventlib.RegisterKey(eventlib.KeyDefinition{
+		Key:       syntheticKey,
+		EventType: syntheticKey,
+		Params: []eventlib.ParamDef{
+			{Name: "mailbox", SubscriptionKey: true, Description: "subscription id source"},
+			{Name: "folders", Description: "filter only"},
+		},
+		Schema: eventlib.SchemaDef{Native: &eventlib.SchemaSpec{Type: reflect.TypeOf(struct{ X string }{})}},
+	})
+
+	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
+	if err := runSchema(f, syntheticKey, false); err != nil {
+		t.Fatalf("runSchema: %v", err)
+	}
+
+	out := stdout.String()
+	if !strings.Contains(out, "SUB-KEY") {
+		t.Errorf("missing SUB-KEY column header in:\n%s", out)
+	}
+
+	// Find the mailbox row and verify "yes" is present
+	var mailboxRow string
+	for _, ln := range strings.Split(out, "\n") {
+		if strings.Contains(ln, "mailbox") && !strings.Contains(ln, "NAME") {
+			mailboxRow = ln
+			break
+		}
+	}
+	if !strings.Contains(mailboxRow, "yes") {
+		t.Errorf("mailbox row missing yes SUB-KEY marker: %q", mailboxRow)
+	}
+
+	// Find the folders row and verify "no" is present
+	var foldersRow string
+	for _, ln := range strings.Split(out, "\n") {
+		if strings.Contains(ln, "folders") && !strings.Contains(ln, "NAME") {
+			foldersRow = ln
+			break
+		}
+	}
+	if !strings.Contains(foldersRow, "no") {
+		t.Errorf("folders row missing no SUB-KEY marker: %q", foldersRow)
+	}
+}
+
+func TestSchema_JSON_IncludesSubscriptionKey(t *testing.T) {
+	const syntheticKey = "test.evt_json"
+	t.Cleanup(func() { eventlib.UnregisterKeyForTest(syntheticKey) })
+
+	eventlib.RegisterKey(eventlib.KeyDefinition{
+		Key:       syntheticKey,
+		EventType: syntheticKey,
+		Params:    []eventlib.ParamDef{{Name: "mailbox", SubscriptionKey: true}},
+		Schema:    eventlib.SchemaDef{Native: &eventlib.SchemaSpec{Type: reflect.TypeOf(struct{ X string }{})}},
+	})
+
+	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
+	if err := runSchema(f, syntheticKey, true); err != nil {
+		t.Fatalf("runSchema json: %v", err)
+	}
+
+	if !strings.Contains(stdout.String(), `"subscription_key"`) {
+		t.Errorf("JSON output missing subscription_key field: %s", stdout.String())
+	}
+	if !strings.Contains(stdout.String(), `true`) {
+		t.Errorf("JSON output missing subscription_key: true value: %s", stdout.String())
+	}
+}
+
 func TestResolveSchemaJSON_CustomWithOverlay(t *testing.T) {
 	const syntheticKey = "t.custom.overlay"
 	t.Cleanup(func() { eventlib.UnregisterKeyForTest(syntheticKey) })

cmd/event/status.go

@@ -7,6 +7,7 @@ import (
 	"fmt"
 	"io"
 	"sort"
+	"strings"
 	"sync"
 	"time"
 
@@ -242,12 +243,17 @@ func writeStatusText(out io.Writer, statuses []appStatus) {
 				s.PID, (time.Duration(s.UptimeSec) * time.Second).String())
 			fmt.Fprintf(out, "  Active consumers: %d\n", s.Active)
 			if len(s.Consumers) > 0 {
-				headers := []string{"CONSUMER", "EVENT KEY", "RECEIVED", "DROPPED"}
+				headers := []string{"CONSUMER", "EVENT KEY", "SUB", "RECEIVED", "DROPPED"}
 				rows := make([][]string, 0, len(s.Consumers))
 				for _, c := range s.Consumers {
+					subDisplay := "-"
+					if c.SubscriptionID != "" && c.SubscriptionID != c.EventKey {
+						subDisplay = strings.TrimPrefix(c.SubscriptionID, c.EventKey+":")
+					}
 					rows = append(rows, []string{
 						fmt.Sprintf("pid=%d", c.PID),
 						c.EventKey,
+						subDisplay,
 						fmt.Sprintf("%d", c.Received),
 						fmt.Sprintf("%d", c.Dropped),
 					})

events/minutes/preconsume.go

@@ -13,8 +13,8 @@ import (
 
 const cleanupTimeout = 5 * time.Second
 
-func subscriptionPreConsume(eventType, subscribePath, unsubscribePath string) func(context.Context, event.APIClient, map[string]string) (func(), error) {
-	return func(ctx context.Context, rt event.APIClient, _ map[string]string) (func(), error) {
+func subscriptionPreConsume(eventType, subscribePath, unsubscribePath string) func(context.Context, event.APIClient, map[string]string) (func() error, error) {
+	return func(ctx context.Context, rt event.APIClient, _ map[string]string) (func() error, error) {
 		if rt == nil {
 			return nil, errs.NewInternalError(errs.SubtypeUnknown,
 				"runtime API client is required for pre-consume subscription")
@@ -25,10 +25,13 @@ func subscriptionPreConsume(eventType, subscribePath, unsubscribePath string) fu
 			return nil, err
 		}
 
-		return func() {
+		return func() error {
 			cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
 			defer cancel()
-			_, _ = rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body)
+			if _, err := rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body); err != nil {
+				return err
+			}
+			return nil
 		}, nil
 	}
 }

events/vc/preconsume.go

@@ -13,8 +13,8 @@ import (
 
 const cleanupTimeout = 5 * time.Second
 
-func subscriptionPreConsume(eventType, subscribePath, unsubscribePath string) func(context.Context, event.APIClient, map[string]string) (func(), error) {
-	return func(ctx context.Context, rt event.APIClient, _ map[string]string) (func(), error) {
+func subscriptionPreConsume(eventType, subscribePath, unsubscribePath string) func(context.Context, event.APIClient, map[string]string) (func() error, error) {
+	return func(ctx context.Context, rt event.APIClient, _ map[string]string) (func() error, error) {
 		if rt == nil {
 			return nil, errs.NewInternalError(errs.SubtypeUnknown,
 				"runtime API client is required for pre-consume subscription")
@@ -25,10 +25,13 @@ func subscriptionPreConsume(eventType, subscribePath, unsubscribePath string) fu
 			return nil, err
 		}
 
-		return func() {
+		return func() error {
 			cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
 			defer cancel()
-			_, _ = rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body)
+			if _, err := rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body); err != nil {
+				return err
+			}
+			return nil
 		}, nil
 	}
 }

events/whiteboard/preconsume.go

@@ -22,8 +22,8 @@ const cleanupTimeout = 5 * time.Second
 //
 // board.whiteboard.updated_v1 is subscribed per-whiteboard (by whiteboard_id),
 // so the path contains a :whiteboard_id placeholder that must be supplied via params.
-func whiteboardSubscriptionPreConsume(eventType string) func(context.Context, event.APIClient, map[string]string) (func(), error) {
-	return func(ctx context.Context, rt event.APIClient, params map[string]string) (func(), error) {
+func whiteboardSubscriptionPreConsume(eventType string) func(context.Context, event.APIClient, map[string]string) (func() error, error) {
+	return func(ctx context.Context, rt event.APIClient, params map[string]string) (func() error, error) {
 		if rt == nil {
 			return nil, errs.NewInternalError(errs.SubtypeUnknown,
 				"runtime API client is required for pre-consume subscription")
@@ -44,10 +44,13 @@ func whiteboardSubscriptionPreConsume(eventType string) func(context.Context, ev
 			return nil, err
 		}
 
-		return func() {
+		return func() error {
 			cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
 			defer cancel()
-			_, _ = rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body)
+			if _, err := rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body); err != nil {
+				return err
+			}
+			return nil
 		}, nil
 	}
 }

internal/credential/default_provider.go

@@ -68,6 +68,8 @@ func (p *DefaultAccountProvider) ResolveAccount(ctx context.Context) (*Account,
 		return nil, core.NotConfiguredError()
 	}
 
+	// p.profile already incorporates --profile / CliRuntimeAppID precedence
+	// (resolved in cmd/bootstrap.go before Factory construction).
 	cfg, err := core.ResolveConfigFromMulti(multi, p.keychain(), p.profile)
 	if err != nil {
 		return nil, err

internal/envvars/envvars.go

@@ -4,8 +4,11 @@
 package envvars
 
 const (
-	CliAppID             = "LARKSUITE_CLI_APP_ID"
-	CliAppSecret         = "LARKSUITE_CLI_APP_SECRET"
+	CliAppID     = "LARKSUITE_CLI_APP_ID"
+	CliAppSecret = "LARKSUITE_CLI_APP_SECRET"
+	// CliRuntimeAppID pins a profile by appId when --profile is unset.
+	// Selector only; credentials still come from config.json / keychain.
+	CliRuntimeAppID      = "LARKSUITE_CLI_RUNTIME_APP_ID"
 	CliBrand             = "LARKSUITE_CLI_BRAND"
 	CliUserAccessToken   = "LARKSUITE_CLI_USER_ACCESS_TOKEN"
 	CliTenantAccessToken = "LARKSUITE_CLI_TENANT_ACCESS_TOKEN"

internal/event/bus/bus.go

@@ -262,19 +262,23 @@ func (b *Bus) handleConn(conn net.Conn) {
 
 // handleHello registers a consume connection with the hub; reader carries bytes already pulled off conn.
 func (b *Bus) handleHello(conn net.Conn, reader *bufio.Reader, hello *protocol.Hello) {
-	bc := NewConn(conn, reader, hello.EventKey, hello.EventTypes, hello.PID)
+	subID := hello.SubscriptionID
+	if subID == "" {
+		subID = hello.EventKey
+	}
+	bc := NewConn(conn, reader, hello.EventKey, hello.EventTypes, hello.PID, subID)
 	bc.SetLogger(b.logger)
 
 	// Register + isFirst under one lock; blocks on any in-progress cleanup lock for the same EventKey.
 	firstForKey := b.hub.RegisterAndIsFirst(bc)
 
-	bc.SetCheckLastForKey(func(eventKey string) bool {
-		return b.hub.AcquireCleanupLock(eventKey)
+	bc.SetCheckLastForKey(func(scope string) bool {
+		return b.hub.AcquireCleanupLock(scope)
 	})
 	bc.SetOnClose(func(c *Conn) {
 		b.hub.UnregisterAndIsLast(c)
 		// Release is idempotent and must fire on every disconnect path so waiters don't block forever.
-		b.hub.ReleaseCleanupLock(c.EventKey())
+		b.hub.ReleaseCleanupLock(c.SubscriptionID())
 		b.mu.Lock()
 		delete(b.conns, c)
 		remaining := len(b.conns)

internal/event/bus/bus_shutdown_test.go

@@ -33,7 +33,7 @@ func TestRunShutdownWithMultipleConns(t *testing.T) {
 		server, client := net.Pipe()
 		pipes = append(pipes, server, client)
 
-		bc := NewConn(server, nil, "im.msg", []string{"im.message.receive_v1"}, 1000+i)
+		bc := NewConn(server, nil, "im.msg", []string{"im.message.receive_v1"}, 1000+i, "")
 		bc.SetLogger(logger)
 		hub.RegisterAndIsFirst(bc)
 

internal/event/bus/conn.go

@@ -29,9 +29,10 @@ type Conn struct {
 	writeMu         sync.Mutex // serialises all net.Conn writes (Encode+SetWriteDeadline is a 2-call sequence)
 	eventKey        string
 	eventTypes      []string
+	subID           string
 	pid             int
 	onClose         func(*Conn)
-	checkLastForKey func(eventKey string) bool
+	checkLastForKey func(scope string) bool
 	logger          *log.Logger
 	closed          chan struct{}
 	closeOnce       sync.Once
@@ -41,7 +42,7 @@ type Conn struct {
 }
 
 // NewConn creates a Conn; pass a reader with pre-buffered bytes (handoff from Bus.handleConn) or nil for a fresh one.
-func NewConn(conn net.Conn, reader *bufio.Reader, eventKey string, eventTypes []string, pid int) *Conn {
+func NewConn(conn net.Conn, reader *bufio.Reader, eventKey string, eventTypes []string, pid int, subID string) *Conn {
 	if reader == nil {
 		reader = bufio.NewReader(conn)
 	}
@@ -52,10 +53,20 @@ func NewConn(conn net.Conn, reader *bufio.Reader, eventKey string, eventTypes []
 		eventKey:   eventKey,
 		eventTypes: eventTypes,
 		pid:        pid,
+		subID:      subID,
 		closed:     make(chan struct{}),
 	}
 }
 
+// SubscriptionID returns the subscription identity. Falls back to EventKey
+// when the stored subID is empty (legacy clients / no-SubscriptionKey EventKeys).
+func (c *Conn) SubscriptionID() string {
+	if c.subID == "" {
+		return c.eventKey
+	}
+	return c.subID
+}
+
 func (c *Conn) SetOnClose(fn func(*Conn)) { c.onClose = fn }
 
 // SetCheckLastForKey: returning true means "you are the last subscriber, run cleanup".
@@ -132,13 +143,19 @@ func (c *Conn) ReaderLoop() {
 }
 
 func (c *Conn) handleControlMessage(msg interface{}) {
-	switch m := msg.(type) {
+	switch msg.(type) {
 	case *protocol.Bye:
 		c.shutdown()
 	case *protocol.PreShutdownCheck:
+		// Use the connection's own authoritative subscription identity rather
+		// than recomputing from the incoming message: a stale or mismatched
+		// PreShutdownCheck must not ask about the wrong scope (which would
+		// suppress or mistrigger per-subscription cleanup). Conn.SubscriptionID()
+		// already falls back to EventKey when its stored subID is empty.
+		scope := c.SubscriptionID()
 		lastForKey := true
 		if c.checkLastForKey != nil {
-			lastForKey = c.checkLastForKey(m.EventKey)
+			lastForKey = c.checkLastForKey(scope)
 		}
 		ack := protocol.NewPreShutdownAck(lastForKey)
 		if err := c.writeFrame(ack); err != nil && c.logger != nil {

internal/event/bus/conn_test.go

@@ -21,7 +21,7 @@ func TestConn_SenderWritesEvents(t *testing.T) {
 	defer server.Close()
 	defer client.Close()
 
-	bc := NewConn(server, nil, "im.msg", []string{"im.message.receive_v1"}, 12345)
+	bc := NewConn(server, nil, "im.msg", []string{"im.message.receive_v1"}, 12345, "")
 	go bc.SenderLoop()
 
 	bc.SendCh() <- &protocol.Event{
@@ -62,7 +62,7 @@ func TestConn_ConcurrentWritesSerialised(t *testing.T) {
 	defer client.Close()
 
 	det := &serializingDetector{Conn: server}
-	bc := NewConn(det, nil, "im.msg", []string{"im.msg"}, 12345)
+	bc := NewConn(det, nil, "im.msg", []string{"im.msg"}, 12345, "")
 
 	go func() { _, _ = io.Copy(io.Discard, client) }()
 
@@ -106,7 +106,7 @@ func TestConn_TrySend_NonEvicting(t *testing.T) {
 	server, client := net.Pipe()
 	defer server.Close()
 	defer client.Close()
-	bc := NewConn(server, nil, "im.msg", []string{"im.msg"}, 12345)
+	bc := NewConn(server, nil, "im.msg", []string{"im.msg"}, 12345, "")
 
 	for i := 0; i < sendChCap; i++ {
 		if !bc.TrySend(i) {
@@ -126,7 +126,7 @@ func TestConn_ReaderDetectsEOF(t *testing.T) {
 	server, client := net.Pipe()
 	defer server.Close()
 
-	bc := NewConn(server, nil, "im.msg", []string{"im.msg"}, 12345)
+	bc := NewConn(server, nil, "im.msg", []string{"im.msg"}, 12345, "")
 
 	done := make(chan struct{})
 	go func() {
@@ -142,3 +142,23 @@ func TestConn_ReaderDetectsEOF(t *testing.T) {
 		t.Fatal("ReaderLoop did not exit on EOF")
 	}
 }
+
+func TestConn_SubscriptionID(t *testing.T) {
+	c1, c2 := net.Pipe()
+	defer c1.Close()
+	defer c2.Close()
+	conn := NewConn(c1, nil, "mail.x", []string{"mail.x"}, 999, "mail.x:abc")
+	if got := conn.SubscriptionID(); got != "mail.x:abc" {
+		t.Errorf("SubscriptionID() = %q, want %q", got, "mail.x:abc")
+	}
+}
+
+func TestConn_SubscriptionID_EmptyFallsBackToEventKey(t *testing.T) {
+	c1, c2 := net.Pipe()
+	defer c1.Close()
+	defer c2.Close()
+	conn := NewConn(c1, nil, "mail.x", []string{"mail.x"}, 999, "")
+	if got := conn.SubscriptionID(); got != "mail.x" {
+		t.Errorf("SubscriptionID() with empty input = %q, want fallback %q", got, "mail.x")
+	}
+}

internal/event/bus/handle_hello_test.go

@@ -63,3 +63,134 @@ func TestHandleHello_HelloAckWriteFailureUnregisters(t *testing.T) {
 		t.Errorf("b.conns after failed HelloAck = %d entries, want 0", remaining)
 	}
 }
+
+// TestHandleHello_LegacyClient_FallsBackToEventKey: a Hello with empty
+// subscription_id registers under EventKey (today's behavior preserved).
+func TestHandleHello_LegacyClient_FallsBackToEventKey(t *testing.T) {
+	logger := log.New(io.Discard, "", 0)
+	hub := NewHub()
+	b := &Bus{
+		hub:        hub,
+		logger:     logger,
+		conns:      make(map[*Conn]struct{}),
+		idleTimer:  time.NewTimer(30 * time.Second),
+		shutdownCh: make(chan struct{}, 1),
+	}
+
+	server, client := net.Pipe()
+	defer server.Close()
+	defer client.Close()
+
+	// Legacy client: no subscription_id field (empty string).
+	hello := &protocol.Hello{
+		PID:            9999,
+		EventKey:       "im.message",
+		EventTypes:     []string{"im.message.receive_v1"},
+		SubscriptionID: "", // legacy: empty, should fallback to EventKey
+	}
+
+	br := bufio.NewReader(server)
+
+	done := make(chan struct{})
+	go func() {
+		b.handleHello(server, br, hello)
+		close(done)
+	}()
+
+	// Read the HelloAck from client side to let handleHello complete.
+	clientReader := bufio.NewReader(client)
+	ackLine, err := clientReader.ReadString('\n')
+	if err != nil {
+		t.Fatalf("failed to read HelloAck: %v", err)
+	}
+
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("handleHello did not return within 3s")
+	}
+
+	// Assertions: registered under EventKey (not a qualified subscription ID).
+	if got := hub.ConnCount(); got != 1 {
+		t.Errorf("hub.ConnCount = %d, want 1", got)
+	}
+	if got := hub.EventKeyCount("im.message"); got != 1 {
+		t.Errorf("hub.EventKeyCount(im.message) = %d, want 1", got)
+	}
+	if got := hub.SubCount("im.message"); got != 1 {
+		t.Errorf("hub.SubCount(im.message) = %d, want 1 (legacy fallback to EventKey)", got)
+	}
+	if got := hub.SubCount("im.message:something"); got != 0 {
+		t.Errorf("hub.SubCount(im.message:something) = %d, want 0 (should not exist)", got)
+	}
+
+	if ackLine == "" {
+		t.Fatal("HelloAck was empty")
+	}
+}
+
+// TestHandleHello_ModernClient_UsesSubscriptionID: a Hello with
+// non-empty subscription_id registers under that ID, not EventKey.
+func TestHandleHello_ModernClient_UsesSubscriptionID(t *testing.T) {
+	logger := log.New(io.Discard, "", 0)
+	hub := NewHub()
+	b := &Bus{
+		hub:        hub,
+		logger:     logger,
+		conns:      make(map[*Conn]struct{}),
+		idleTimer:  time.NewTimer(30 * time.Second),
+		shutdownCh: make(chan struct{}, 1),
+	}
+
+	server, client := net.Pipe()
+	defer server.Close()
+	defer client.Close()
+
+	// Modern client: subscription_id explicitly set.
+	subscriptionID := "mail.message:alice@example.com"
+	hello := &protocol.Hello{
+		PID:            8888,
+		EventKey:       "mail.message",
+		EventTypes:     []string{"mail.message.receive_v1"},
+		SubscriptionID: subscriptionID, // modern: per-resource subscription
+	}
+
+	br := bufio.NewReader(server)
+
+	done := make(chan struct{})
+	go func() {
+		b.handleHello(server, br, hello)
+		close(done)
+	}()
+
+	// Read the HelloAck from client side to let handleHello complete.
+	clientReader := bufio.NewReader(client)
+	ackLine, err := clientReader.ReadString('\n')
+	if err != nil {
+		t.Fatalf("failed to read HelloAck: %v", err)
+	}
+
+	select {
+	case <-done:
+	case <-time.After(3 * time.Second):
+		t.Fatal("handleHello did not return within 3s")
+	}
+
+	// Assertions: registered under the subscription_id, not bare EventKey.
+	if got := hub.ConnCount(); got != 1 {
+		t.Errorf("hub.ConnCount = %d, want 1", got)
+	}
+	if got := hub.EventKeyCount("mail.message"); got != 1 {
+		t.Errorf("hub.EventKeyCount(mail.message) = %d, want 1", got)
+	}
+	if got := hub.SubCount(subscriptionID); got != 1 {
+		t.Errorf("hub.SubCount(%q) = %d, want 1 (modern: uses SubscriptionID)", subscriptionID, got)
+	}
+	if got := hub.SubCount("mail.message"); got != 0 {
+		t.Errorf("hub.SubCount(mail.message) = %d, want 0 (modern: NOT registered under bare EventKey)", got)
+	}
+
+	if ackLine == "" {
+		t.Fatal("HelloAck was empty")
+	}
+}

internal/event/bus/hub.go

@@ -16,6 +16,9 @@ import (
 // Subscriber is the interface a connection must satisfy for Hub registration.
 type Subscriber interface {
 	EventKey() string
+	// SubscriptionID identifies the per-resource subscription for dedup purposes.
+	// When no resource qualifier is needed it equals EventKey.
+	SubscriptionID() string
 	EventTypes() []string
 	SendCh() chan interface{}
 	PID() int
@@ -34,8 +37,11 @@ type Subscriber interface {
 type Hub struct {
 	mu          sync.RWMutex
 	subscribers map[Subscriber]struct{}
-	keyCounts   map[string]int
-	// cleanupInProgress[key] holds a channel closed on release; presence means a cleanup lock is held.
+	// subCounts is keyed by SubscriptionID (not EventKey) so that different
+	// per-resource subscriptions sharing the same EventKey are deduped independently.
+	subCounts map[string]int
+	// cleanupInProgress[subscriptionID] holds a channel closed on release;
+	// presence means a cleanup lock is held for that subscription.
 	cleanupInProgress map[string]chan struct{}
 	logger            atomic.Pointer[log.Logger]
 }
@@ -43,7 +49,7 @@ type Hub struct {
 func NewHub() *Hub {
 	return &Hub{
 		subscribers:       make(map[Subscriber]struct{}),
-		keyCounts:         make(map[string]int),
+		subCounts:         make(map[string]int),
 		cleanupInProgress: make(map[string]chan struct{}),
 	}
 }
@@ -51,7 +57,7 @@ func NewHub() *Hub {
 // SetLogger attaches a logger (nil tolerated).
 func (h *Hub) SetLogger(l *log.Logger) { h.logger.Store(l) }
 
-// UnregisterAndIsLast removes s and reports whether it was last for its EventKey; stale unregisters are no-ops.
+// UnregisterAndIsLast removes s and reports whether it was last for its SubscriptionID; stale unregisters are no-ops.
 func (h *Hub) UnregisterAndIsLast(s Subscriber) bool {
 	h.mu.Lock()
 	defer h.mu.Unlock()
@@ -59,34 +65,35 @@ func (h *Hub) UnregisterAndIsLast(s Subscriber) bool {
 		return false
 	}
 	delete(h.subscribers, s)
-	h.keyCounts[s.EventKey()]--
-	isLast := h.keyCounts[s.EventKey()] == 0
+	sid := s.SubscriptionID()
+	h.subCounts[sid]--
+	isLast := h.subCounts[sid] == 0
 	if isLast {
-		delete(h.keyCounts, s.EventKey())
+		delete(h.subCounts, sid)
 	}
 	return isLast
 }
 
-// AcquireCleanupLock reserves cleanup rights iff exactly one subscriber exists for eventKey and no lock is held.
+// AcquireCleanupLock reserves cleanup rights iff exactly one subscriber exists for subscriptionID and no lock is held.
 // Count==0 is rejected (would block future Register calls). On true return, caller MUST Release.
-func (h *Hub) AcquireCleanupLock(eventKey string) bool {
+func (h *Hub) AcquireCleanupLock(subscriptionID string) bool {
 	h.mu.Lock()
 	defer h.mu.Unlock()
-	if h.keyCounts[eventKey] != 1 {
+	if h.subCounts[subscriptionID] != 1 {
 		return false
 	}
-	if _, alreadyLocked := h.cleanupInProgress[eventKey]; alreadyLocked {
+	if _, alreadyLocked := h.cleanupInProgress[subscriptionID]; alreadyLocked {
 		return false
 	}
-	h.cleanupInProgress[eventKey] = make(chan struct{})
+	h.cleanupInProgress[subscriptionID] = make(chan struct{})
 	return true
 }
 
 // ReleaseCleanupLock is idempotent; OnClose calls unconditionally.
-func (h *Hub) ReleaseCleanupLock(eventKey string) {
+func (h *Hub) ReleaseCleanupLock(subscriptionID string) {
 	h.mu.Lock()
-	ch := h.cleanupInProgress[eventKey]
-	delete(h.cleanupInProgress, eventKey)
+	ch := h.cleanupInProgress[subscriptionID]
+	delete(h.cleanupInProgress, subscriptionID)
 	h.mu.Unlock()
 	if ch != nil {
 		close(ch)
@@ -94,23 +101,24 @@ func (h *Hub) ReleaseCleanupLock(eventKey string) {
 }
 
 // RegisterAndIsFirst adds s to the hub and reports whether it's the first
-// subscriber for its EventKey. If a cleanup is in progress for
-// s.EventKey() (another conn holds the cleanup lock), this waits until
+// subscriber for its SubscriptionID. If a cleanup is in progress for
+// s.SubscriptionID() (another conn holds the cleanup lock), this waits until
 // cleanup releases before registering — closing the PreShutdownCheck ×
 // Hello TOCTOU race. The wait releases h.mu before blocking on the
-// channel, so concurrent operations on other keys aren't stalled.
+// channel, so concurrent operations on other subscriptions aren't stalled.
 func (h *Hub) RegisterAndIsFirst(s Subscriber) bool {
+	sid := s.SubscriptionID()
 	for {
 		h.mu.Lock()
-		ch, locked := h.cleanupInProgress[s.EventKey()]
+		ch, locked := h.cleanupInProgress[sid]
 		if locked {
 			h.mu.Unlock()
 			<-ch // wait for release, then re-check (defensive against races)
 			continue
 		}
-		isFirst := h.keyCounts[s.EventKey()] == 0
+		isFirst := h.subCounts[sid] == 0
 		h.subscribers[s] = struct{}{}
-		h.keyCounts[s.EventKey()]++
+		h.subCounts[sid]++
 		h.mu.Unlock()
 		return isFirst
 	}
@@ -176,11 +184,25 @@ func (h *Hub) ConnCount() int {
 	return len(h.subscribers)
 }
 
-// EventKeyCount returns the number of subscribers registered for eventKey.
+// EventKeyCount returns total subscribers for the given EventKey, aggregating
+// across all SubscriptionIDs. For per-subscription counts use SubCount.
 func (h *Hub) EventKeyCount(eventKey string) int {
 	h.mu.RLock()
 	defer h.mu.RUnlock()
-	return h.keyCounts[eventKey]
+	count := 0
+	for s := range h.subscribers {
+		if s.EventKey() == eventKey {
+			count++
+		}
+	}
+	return count
+}
+
+// SubCount returns the count of subscribers for the given SubscriptionID.
+func (h *Hub) SubCount(subscriptionID string) int {
+	h.mu.RLock()
+	defer h.mu.RUnlock()
+	return h.subCounts[subscriptionID]
 }
 
 // BroadcastSourceStatus fans out a source-level status change to every
@@ -205,10 +227,11 @@ func (h *Hub) Consumers() []protocol.ConsumerInfo {
 	result := make([]protocol.ConsumerInfo, 0, len(h.subscribers))
 	for s := range h.subscribers {
 		result = append(result, protocol.ConsumerInfo{
-			PID:      s.PID(),
-			EventKey: s.EventKey(),
-			Received: s.Received(),
-			Dropped:  s.DroppedCount(),
+			PID:            s.PID(),
+			EventKey:       s.EventKey(),
+			SubscriptionID: s.SubscriptionID(),
+			Received:       s.Received(),
+			Dropped:        s.DroppedCount(),
 		})
 	}
 	return result

internal/event/bus/hub_observability_test.go

@@ -17,7 +17,7 @@ func TestHubDroppedCountIncrements(t *testing.T) {
 	server, client := testNetPipe(t)
 	defer server.Close()
 	defer client.Close()
-	c := NewConn(server, nil, "k", []string{"t"}, 1)
+	c := NewConn(server, nil, "k", []string{"t"}, 1, "")
 	c.sendCh = make(chan interface{}, 1)
 	h.RegisterAndIsFirst(c)
 
@@ -35,7 +35,7 @@ func TestPublishAssignsIncrementalSeq(t *testing.T) {
 	server, client := testNetPipe(t)
 	defer server.Close()
 	defer client.Close()
-	c := NewConn(server, nil, "k", []string{"t"}, 1)
+	c := NewConn(server, nil, "k", []string{"t"}, 1, "")
 	c.sendCh = make(chan interface{}, 10)
 	h.RegisterAndIsFirst(c)
 
@@ -60,7 +60,7 @@ func TestPublishPopulatesEventIDAndSourceTime(t *testing.T) {
 	server, client := testNetPipe(t)
 	defer server.Close()
 	defer client.Close()
-	c := NewConn(server, nil, "k", []string{"t"}, 1)
+	c := NewConn(server, nil, "k", []string{"t"}, 1, "")
 	c.sendCh = make(chan interface{}, 1)
 	h.RegisterAndIsFirst(c)
 
@@ -87,7 +87,7 @@ func TestPublishSourceTimeTakesPrecedence(t *testing.T) {
 	server, client := testNetPipe(t)
 	defer server.Close()
 	defer client.Close()
-	c := NewConn(server, nil, "k", []string{"t"}, 1)
+	c := NewConn(server, nil, "k", []string{"t"}, 1, "")
 	c.sendCh = make(chan interface{}, 1)
 	h.RegisterAndIsFirst(c)
 
@@ -111,7 +111,7 @@ func TestPublishSourceTimeFallback(t *testing.T) {
 	server, client := testNetPipe(t)
 	defer server.Close()
 	defer client.Close()
-	c := NewConn(server, nil, "k", []string{"t"}, 1)
+	c := NewConn(server, nil, "k", []string{"t"}, 1, "")
 	c.sendCh = make(chan interface{}, 1)
 	h.RegisterAndIsFirst(c)
 

internal/event/bus/hub_publish_race_test.go

@@ -111,6 +111,7 @@ type alwaysFailSubscriber struct {
 }
 
 func (s *alwaysFailSubscriber) EventKey() string         { return s.eventKey }
+func (s *alwaysFailSubscriber) SubscriptionID() string   { return s.eventKey }
 func (s *alwaysFailSubscriber) EventTypes() []string     { return s.eventTypes }
 func (s *alwaysFailSubscriber) SendCh() chan interface{} { return s.sendCh }
 func (s *alwaysFailSubscriber) PID() int                 { return 0 }
@@ -153,6 +154,7 @@ func newRaceSubscriber(key string, types []string, capacity int) *raceSubscriber
 }
 
 func (s *raceSubscriber) EventKey() string         { return s.eventKey }
+func (s *raceSubscriber) SubscriptionID() string   { return s.eventKey }
 func (s *raceSubscriber) EventTypes() []string     { return s.eventTypes }
 func (s *raceSubscriber) SendCh() chan interface{} { return s.sendCh }
 func (s *raceSubscriber) PID() int                 { return s.pid }

internal/event/bus/hub_test.go

@@ -5,6 +5,7 @@ package bus
 
 import (
 	"encoding/json"
+	"net"
 	"sync"
 	"sync/atomic"
 	"testing"
@@ -235,7 +236,10 @@ func newTestConn(eventKey string, eventTypes []string) *testConn {
 	}
 }
 
-func (c *testConn) EventKey() string         { return c.eventKey }
+func (c *testConn) EventKey() string { return c.eventKey }
+
+// SubscriptionID falls back to EventKey for test mocks that don't set a separate subscription ID.
+func (c *testConn) SubscriptionID() string   { return c.eventKey }
 func (c *testConn) EventTypes() []string     { return c.eventTypes }
 func (c *testConn) SendCh() chan interface{} { return c.sendCh }
 func (c *testConn) PID() int                 { return c.pid }
@@ -275,3 +279,79 @@ func (c *testConn) TrySend(msg interface{}) bool {
 		return false
 	}
 }
+
+func TestHub_SubscriptionID_Isolation(t *testing.T) {
+	h := NewHub()
+	c1, _ := net.Pipe()
+	c2, _ := net.Pipe()
+	defer c1.Close()
+	defer c2.Close()
+	s1 := NewConn(c1, nil, "mail.x", []string{"mail.x"}, 1, "mail.x:alice")
+	s2 := NewConn(c2, nil, "mail.x", []string{"mail.x"}, 2, "mail.x:bob")
+
+	if !h.RegisterAndIsFirst(s1) {
+		t.Error("s1 should be first for its subscription")
+	}
+	if !h.RegisterAndIsFirst(s2) {
+		t.Error("s2 should ALSO be first (different SubscriptionID)")
+	}
+	if !h.UnregisterAndIsLast(s1) {
+		t.Error("s1 should be last for mail.x:alice")
+	}
+	if !h.UnregisterAndIsLast(s2) {
+		t.Error("s2 should be last for mail.x:bob")
+	}
+}
+
+func TestHub_SameSubscriptionID_NotFirst(t *testing.T) {
+	h := NewHub()
+	c1, _ := net.Pipe()
+	c2, _ := net.Pipe()
+	defer c1.Close()
+	defer c2.Close()
+	s1 := NewConn(c1, nil, "mail.x", []string{"mail.x"}, 1, "mail.x:alice")
+	s2 := NewConn(c2, nil, "mail.x", []string{"mail.x"}, 2, "mail.x:alice")
+
+	if !h.RegisterAndIsFirst(s1) {
+		t.Error("s1 first")
+	}
+	if h.RegisterAndIsFirst(s2) {
+		t.Error("s2 same SubscriptionID should NOT be first")
+	}
+}
+
+func TestHub_EventKeyCount_AggregatesAcrossSubscriptions(t *testing.T) {
+	h := NewHub()
+	c1, _ := net.Pipe()
+	c2, _ := net.Pipe()
+	defer c1.Close()
+	defer c2.Close()
+	s1 := NewConn(c1, nil, "mail.x", []string{"mail.x"}, 1, "mail.x:alice")
+	s2 := NewConn(c2, nil, "mail.x", []string{"mail.x"}, 2, "mail.x:bob")
+	h.RegisterAndIsFirst(s1)
+	h.RegisterAndIsFirst(s2)
+	if got := h.EventKeyCount("mail.x"); got != 2 {
+		t.Errorf("EventKeyCount(mail.x) = %d, want 2 (aggregated across subscriptions)", got)
+	}
+	if got := h.SubCount("mail.x:alice"); got != 1 {
+		t.Errorf("SubCount(mail.x:alice) = %d, want 1", got)
+	}
+	if got := h.SubCount("mail.x:bob"); got != 1 {
+		t.Errorf("SubCount(mail.x:bob) = %d, want 1", got)
+	}
+}
+
+func TestHub_Consumers_PopulatesSubscriptionID(t *testing.T) {
+	h := NewHub()
+	c1, _ := net.Pipe()
+	defer c1.Close()
+	s1 := NewConn(c1, nil, "mail.x", []string{"mail.x"}, 1, "mail.x:alice")
+	h.RegisterAndIsFirst(s1)
+	consumers := h.Consumers()
+	if len(consumers) != 1 {
+		t.Fatalf("got %d consumers, want 1", len(consumers))
+	}
+	if consumers[0].SubscriptionID != "mail.x:alice" {
+		t.Errorf("Consumers()[0].SubscriptionID = %q, want %q", consumers[0].SubscriptionID, "mail.x:alice")
+	}
+}

internal/event/consume/consume.go

@@ -61,6 +61,22 @@ func Run(ctx context.Context, tr transport.IPC, appID, profileName, domain strin
 		}
 	}
 
+	// Normalize params (resolve aliases like "me" -> real email) before fingerprint
+	// compute, PreConsume, Match, Process. Must happen BEFORE doHello so the
+	// SubscriptionID we send to bus reflects canonical values.
+	if keyDef.NormalizeParams != nil {
+		if err := keyDef.NormalizeParams(ctx, opts.Runtime, opts.Params); err != nil {
+			if _, ok := errs.ProblemOf(err); ok {
+				return err
+			}
+			return errs.NewInternalError(errs.SubtypeUnknown,
+				"normalize params for %s: %s", opts.EventKey, err).WithCause(err)
+		}
+	}
+
+	// Compute subscription identity from normalized params + SubscriptionKey flags.
+	subscriptionID := ComputeSubscriptionID(keyDef, opts.Params)
+
 	if opts.Timeout > 0 {
 		var cancel context.CancelFunc
 		ctx, cancel = context.WithTimeout(ctx, opts.Timeout)
@@ -81,13 +97,13 @@ func Run(ctx context.Context, tr transport.IPC, appID, profileName, domain strin
 	}
 	defer conn.Close()
 
-	ack, br, err := doHello(conn, opts.EventKey, []string{keyDef.EventType})
+	ack, br, err := doHello(conn, opts.EventKey, []string{keyDef.EventType}, subscriptionID)
 	if err != nil {
 		return errs.NewInternalError(errs.SubtypeUnknown,
 			"event bus handshake failed: %s", err).WithCause(err)
 	}
 
-	var cleanup func()
+	var cleanup func() error
 	if ack.FirstForKey && keyDef.PreConsume != nil {
 		if !opts.Quiet {
 			fmt.Fprintf(errOut, "[event] running pre-consume setup...\n")
@@ -113,14 +129,22 @@ func Run(ctx context.Context, tr transport.IPC, appID, profileName, domain strin
 		if cleanup != nil {
 			switch {
 			case r != nil:
-				fmt.Fprintf(errOut, "WARN: panic recovered; running cleanup unconditionally (may affect other consumers of %s)\n", opts.EventKey)
-				cleanup()
+				fmt.Fprintf(errOut,
+					"WARN: panic recovered; running cleanup unconditionally (may affect other consumers of %s)\n",
+					opts.EventKey)
+				if cleanupErr := cleanup(); cleanupErr != nil {
+					fmt.Fprintf(errOut,
+						"WARN: cleanup also failed during panic recovery: %v\n", cleanupErr)
+				}
 			case lastForKey:
 				if !opts.Quiet {
 					fmt.Fprintf(errOut, "[event] running cleanup...\n")
 				}
-				cleanup()
-				if !opts.Quiet {
+				if cleanupErr := cleanup(); cleanupErr != nil {
+					fmt.Fprintf(errOut,
+						"WARN: cleanup failed: %v (server-side subscribe is idempotent — residual record will be overwritten on next subscribe)\n",
+						cleanupErr)
+				} else if !opts.Quiet {
 					fmt.Fprintf(errOut, "[event] cleanup done.\n")
 				}
 			}
@@ -144,7 +168,7 @@ func Run(ctx context.Context, tr transport.IPC, appID, profileName, domain strin
 
 	writeReadyMarker(errOut, opts)
 
-	return consumeLoop(ctx, conn, br, keyDef, opts, &lastForKey, &emitted)
+	return consumeLoop(ctx, conn, br, keyDef, opts, subscriptionID, &lastForKey, &emitted)
 }
 
 func truncateDuration(d time.Duration) time.Duration {

internal/event/consume/consume_test.go

@@ -0,0 +1,101 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package consume
+
+import (
+	"bufio"
+	"bytes"
+	"context"
+	"encoding/json"
+	"errors"
+	"net"
+	"strings"
+	"testing"
+
+	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/protocol"
+	"github.com/larksuite/cli/internal/event/transport"
+)
+
+// fakeRT is a minimal event.APIClient mock.
+type fakeRT struct {
+	err error
+}
+
+func (f *fakeRT) CallAPI(_ context.Context, _, _ string, _ interface{}) (json.RawMessage, error) {
+	return nil, f.err
+}
+
+func TestNormalizeParams_ErrorIsWrappedWithEventKey(t *testing.T) {
+	// Drives the real Run() path: NormalizeParams fails before EnsureBus, so no
+	// bus is contacted, yet the production error-wrapping is exercised — if Run()
+	// ever stops wrapping, this test fails.
+	const key = "test.evt_normalize_fail"
+	event.RegisterKey(event.KeyDefinition{
+		Key:       key,
+		EventType: key,
+		Schema:    event.SchemaDef{Custom: &event.SchemaSpec{Raw: json.RawMessage(`{"type":"object"}`)}},
+		NormalizeParams: func(_ context.Context, _ event.APIClient, _ map[string]string) error {
+			return errors.New("simulated normalize failure")
+		},
+	})
+	defer event.UnregisterKeyForTest(key)
+
+	err := Run(context.Background(), transport.New(), "app", "", "", Options{
+		EventKey: key,
+		Runtime:  &fakeRT{},
+		Quiet:    true,
+	})
+	if err == nil {
+		t.Fatal("expected Run to fail when NormalizeParams errors")
+	}
+	if !strings.Contains(err.Error(), "normalize params for "+key+":") {
+		t.Errorf("error not wrapped with EventKey prefix: %v", err)
+	}
+	if !strings.Contains(err.Error(), "simulated normalize failure") {
+		t.Errorf("underlying error not propagated: %v", err)
+	}
+}
+
+func TestDoHello_PassesSubscriptionIDToWire(t *testing.T) {
+	a, b := net.Pipe()
+	defer a.Close()
+	defer b.Close()
+
+	// Server-side: read Hello, decode, assert SubscriptionID, send ack
+	done := make(chan string, 1)
+	go func() {
+		br := bufio.NewReader(b)
+		line, err := protocol.ReadFrame(br)
+		if err != nil {
+			done <- "READ_ERR:" + err.Error()
+			return
+		}
+		msg, err := protocol.Decode(bytes.TrimRight(line, "\n"))
+		if err != nil {
+			done <- "DECODE_ERR:" + err.Error()
+			return
+		}
+		if hello, ok := msg.(*protocol.Hello); ok {
+			done <- hello.SubscriptionID
+			// send ack so client can return
+			ack := protocol.NewHelloAck("v1", true)
+			_ = protocol.EncodeWithDeadline(b, ack, protocol.WriteTimeout)
+		} else {
+			done <- "WRONG_TYPE"
+		}
+	}()
+
+	ack, _, err := doHello(a, "mail.x", []string{"mail.x"}, "mail.x:alice")
+	if err != nil {
+		t.Fatalf("doHello error: %v", err)
+	}
+	if ack == nil {
+		t.Fatal("got nil ack")
+	}
+	got := <-done
+	if got != "mail.x:alice" {
+		t.Errorf("Hello.SubscriptionID on wire = %q, want %q", got, "mail.x:alice")
+	}
+}

internal/event/consume/fingerprint.go

@@ -0,0 +1,41 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package consume
+
+import (
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"sort"
+
+	"github.com/larksuite/cli/internal/event"
+)
+
+// ComputeSubscriptionID returns a stable identifier scoped to (EventKey, values
+// of the ParamDefs marked SubscriptionKey); the framework uses it to dedup
+// PreConsume/cleanup gates and key Hub counts per-subscription. No SubscriptionKey
+// params -> returns def.Key verbatim (legacy one-dimensional behavior).
+//
+// Stability contract: same EventKey + same normalized param values -> same ID
+// across CLI versions; changing the encoding requires a wire-format bump.
+func ComputeSubscriptionID(def *event.KeyDefinition, params map[string]string) string {
+	type kv struct {
+		Name  string `json:"name"`
+		Value string `json:"value"`
+	}
+	var subParams []kv
+	for _, p := range def.Params {
+		if !p.SubscriptionKey {
+			continue
+		}
+		subParams = append(subParams, kv{Name: p.Name, Value: params[p.Name]})
+	}
+	if len(subParams) == 0 {
+		return def.Key
+	}
+	sort.Slice(subParams, func(i, j int) bool { return subParams[i].Name < subParams[j].Name })
+	raw, _ := json.Marshal(subParams) // err impossible: kv has no unmarshalable fields
+	sum := sha256.Sum256(raw)
+	return def.Key + ":" + base64.RawURLEncoding.EncodeToString(sum[:12])
+}

internal/event/consume/fingerprint_test.go

@@ -0,0 +1,126 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package consume
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/larksuite/cli/internal/event"
+)
+
+func TestComputeSubscriptionID(t *testing.T) {
+	makeDef := func(subKeyNames ...string) *event.KeyDefinition {
+		def := &event.KeyDefinition{Key: "test.evt"}
+		marked := make(map[string]bool, len(subKeyNames))
+		for _, n := range subKeyNames {
+			marked[n] = true
+		}
+		for _, n := range []string{"alpha", "beta", "gamma"} {
+			def.Params = append(def.Params, event.ParamDef{Name: n, SubscriptionKey: marked[n]})
+		}
+		return def
+	}
+
+	t.Run("no SubscriptionKey params returns EventKey verbatim", func(t *testing.T) {
+		def := makeDef()
+		got := ComputeSubscriptionID(def, map[string]string{"alpha": "x", "beta": "y"})
+		if got != "test.evt" {
+			t.Errorf("got %q, want %q", got, "test.evt")
+		}
+	})
+
+	t.Run("single SubscriptionKey param: non-sub params do not leak into ID", func(t *testing.T) {
+		def := makeDef("alpha")
+		id1 := ComputeSubscriptionID(def, map[string]string{"alpha": "value1", "beta": "ignored"})
+		id2 := ComputeSubscriptionID(def, map[string]string{"alpha": "value1", "beta": "different"})
+		if id1 != id2 {
+			t.Errorf("non-SubscriptionKey param change leaked into ID: %q vs %q", id1, id2)
+		}
+	})
+
+	t.Run("different SubscriptionKey value produces different ID", func(t *testing.T) {
+		def := makeDef("alpha")
+		id1 := ComputeSubscriptionID(def, map[string]string{"alpha": "v1"})
+		id2 := ComputeSubscriptionID(def, map[string]string{"alpha": "v2"})
+		if id1 == id2 {
+			t.Errorf("different values produced same ID: %q", id1)
+		}
+	})
+}
+
+func TestComputeSubscriptionID_Stability(t *testing.T) {
+	// Param order in the ParamDef list must not affect the result (sorted by name internally).
+	def1 := &event.KeyDefinition{
+		Key: "test.evt",
+		Params: []event.ParamDef{
+			{Name: "b", SubscriptionKey: true},
+			{Name: "a", SubscriptionKey: true},
+		},
+	}
+	def2 := &event.KeyDefinition{
+		Key: "test.evt",
+		Params: []event.ParamDef{
+			{Name: "a", SubscriptionKey: true},
+			{Name: "b", SubscriptionKey: true},
+		},
+	}
+	id1 := ComputeSubscriptionID(def1, map[string]string{"a": "1", "b": "2"})
+	id2 := ComputeSubscriptionID(def2, map[string]string{"a": "1", "b": "2"})
+	if id1 != id2 {
+		t.Errorf("order-sensitive: id1=%q id2=%q", id1, id2)
+	}
+}
+
+func TestComputeSubscriptionID_Format(t *testing.T) {
+	def := &event.KeyDefinition{
+		Key:    "mail.user_mailbox.event.message_received_v1",
+		Params: []event.ParamDef{{Name: "mailbox", SubscriptionKey: true}},
+	}
+	id := ComputeSubscriptionID(def, map[string]string{"mailbox": "liuxinyang@example.com"})
+	prefix := "mail.user_mailbox.event.message_received_v1:"
+	if !strings.HasPrefix(id, prefix) {
+		t.Fatalf("missing prefix: %q", id)
+	}
+	suffix := strings.TrimPrefix(id, prefix)
+	if len(suffix) != 16 {
+		t.Errorf("fingerprint length = %d, want 16", len(suffix))
+	}
+	for _, c := range suffix {
+		isValid := (c >= 'A' && c <= 'Z') || (c >= 'a' && c <= 'z') || (c >= '0' && c <= '9') || c == '-' || c == '_'
+		if !isValid {
+			t.Errorf("non-base64URL char in fingerprint: %q", suffix)
+			break
+		}
+	}
+}
+
+func TestComputeSubscriptionID_UnicodeAndSpecialChars(t *testing.T) {
+	def := &event.KeyDefinition{
+		Key:    "test.evt",
+		Params: []event.ParamDef{{Name: "value", SubscriptionKey: true}},
+	}
+	for _, val := range []string{"中文", "emoji🚀", "with spaces", "with:colons", "with\"quotes"} {
+		id := ComputeSubscriptionID(def, map[string]string{"value": val})
+		if !strings.HasPrefix(id, "test.evt:") || len(id) != len("test.evt:")+16 {
+			t.Errorf("ID malformed for value=%q: %q (len=%d)", val, id, len(id))
+		}
+	}
+}
+
+func TestComputeSubscriptionID_EmptyValue(t *testing.T) {
+	def := &event.KeyDefinition{
+		Key:    "test.evt",
+		Params: []event.ParamDef{{Name: "x", SubscriptionKey: true}},
+	}
+	id1 := ComputeSubscriptionID(def, map[string]string{"x": ""})
+	id2 := ComputeSubscriptionID(def, map[string]string{}) // missing entirely
+	if id1 != id2 {
+		t.Errorf("empty value should be indistinguishable from missing: %q vs %q", id1, id2)
+	}
+	id3 := ComputeSubscriptionID(def, map[string]string{"x": "nonempty"})
+	if id1 == id3 {
+		t.Errorf("empty and nonempty produced same ID: %q", id1)
+	}
+}

internal/event/consume/handshake.go

@@ -18,8 +18,8 @@ const helloAckTimeout = 5 * time.Second // symmetric with bus-side hello read de
 
 // doHello returns a bufio.Reader holding any bytes already pulled off conn so events
 // buffered with the ack in one TCP segment aren't dropped.
-func doHello(conn net.Conn, eventKey string, eventTypes []string) (*protocol.HelloAck, *bufio.Reader, error) {
-	hello := protocol.NewHello(os.Getpid(), eventKey, eventTypes, "v1")
+func doHello(conn net.Conn, eventKey string, eventTypes []string, subscriptionID string) (*protocol.HelloAck, *bufio.Reader, error) {
+	hello := protocol.NewHello(os.Getpid(), eventKey, eventTypes, "v1", subscriptionID)
 	if err := protocol.EncodeWithDeadline(conn, hello, protocol.WriteTimeout); err != nil {
 		return nil, nil, err
 	}

internal/event/consume/handshake_test.go

@@ -27,7 +27,7 @@ func TestDoHello_ReadDeadline(t *testing.T) {
 	start := time.Now()
 	done := make(chan error, 1)
 	go func() {
-		_, _, err := doHello(client, "im.msg", []string{"im.msg"})
+		_, _, err := doHello(client, "im.msg", []string{"im.msg"}, "")
 		done <- err
 	}()
 

internal/event/consume/loop.go

@@ -22,7 +22,7 @@ import (
 )
 
 // consumeLoop reads events and dispatches to workers; cancels on terminal sink errors.
-func consumeLoop(ctx context.Context, conn net.Conn, br *bufio.Reader, keyDef *event.KeyDefinition, opts Options, lastForKey *bool, emitted *atomic.Int64) error {
+func consumeLoop(ctx context.Context, conn net.Conn, br *bufio.Reader, keyDef *event.KeyDefinition, opts Options, subscriptionID string, lastForKey *bool, emitted *atomic.Int64) error {
 	ctx, cancel := context.WithCancel(ctx)
 	defer cancel()
 
@@ -185,7 +185,7 @@ func consumeLoop(ctx context.Context, conn net.Conn, br *bufio.Reader, keyDef *e
 		close(stopReader)
 		<-readerDone
 		conn.SetReadDeadline(time.Time{})
-		*lastForKey = checkLastForKey(conn, opts.EventKey)
+		*lastForKey = checkLastForKey(conn, opts.EventKey, subscriptionID)
 		conn.Close()
 	case <-allDone:
 		// bus-side close; can't query, assume last
@@ -199,13 +199,19 @@ func consumeLoop(ctx context.Context, conn net.Conn, br *bufio.Reader, keyDef *e
 
 // processAndOutput returns (wrote, err); err non-nil only for sink.Write failures.
 func processAndOutput(ctx context.Context, keyDef *event.KeyDefinition, evt *protocol.Event, opts Options, sink Sink, jqCode *gojq.Code) (bool, error) {
+	raw := &event.RawEvent{
+		EventType: evt.EventType,
+		Payload:   evt.Payload,
+	}
+
+	// Synchronous Match filter runs before any work (Process / sink write).
+	if keyDef.Match != nil && !keyDef.Match(raw, opts.Params) {
+		return false, nil
+	}
+
 	var result json.RawMessage
 
 	if keyDef.Process != nil {
-		raw := &event.RawEvent{
-			EventType: evt.EventType,
-			Payload:   evt.Payload,
-		}
 		var err error
 		result, err = keyDef.Process(ctx, opts.Runtime, raw, opts.Params)
 		if err != nil {

internal/event/consume/loop_test.go

@@ -89,7 +89,7 @@ func TestConsumeLoop_DeliversEventsAndExitsOnMaxEvents(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	err := consumeLoop(ctx, client, bufio.NewReader(client), echoKeyDef("test.key"), opts, &lastForKey, &emitted)
+	err := consumeLoop(ctx, client, bufio.NewReader(client), echoKeyDef("test.key"), opts, "", &lastForKey, &emitted)
 	if err != nil {
 		t.Fatalf("consumeLoop: %v", err)
 	}
@@ -132,7 +132,7 @@ func TestConsumeLoop_SeqGapEmitsWarning(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	if err := consumeLoop(ctx, client, bufio.NewReader(client), echoKeyDef("test.key"), opts, &lastForKey, &emitted); err != nil {
+	if err := consumeLoop(ctx, client, bufio.NewReader(client), echoKeyDef("test.key"), opts, "", &lastForKey, &emitted); err != nil {
 		t.Fatalf("consumeLoop: %v", err)
 	}
 	if got := emitted.Load(); got != 2 {
@@ -169,7 +169,7 @@ func TestConsumeLoop_JQFilterAppliedPerEvent(t *testing.T) {
 	ctx, cancel := context.WithTimeout(context.Background(), 5*time.Second)
 	defer cancel()
 
-	if err := consumeLoop(ctx, client, bufio.NewReader(client), echoKeyDef("test.key"), opts, &lastForKey, &emitted); err != nil {
+	if err := consumeLoop(ctx, client, bufio.NewReader(client), echoKeyDef("test.key"), opts, "", &lastForKey, &emitted); err != nil {
 		t.Fatalf("consumeLoop: %v", err)
 	}
 	if got := emitted.Load(); got != 1 {
@@ -196,12 +196,96 @@ func TestConsumeLoop_CompileJQFailsEarly(t *testing.T) {
 
 	var lastForKey bool
 	var emitted atomic.Int64
-	err := consumeLoop(context.Background(), client, bufio.NewReader(client), echoKeyDef("test.key"), opts, &lastForKey, &emitted)
+	err := consumeLoop(context.Background(), client, bufio.NewReader(client), echoKeyDef("test.key"), opts, "", &lastForKey, &emitted)
 	if err == nil {
 		t.Fatal("consumeLoop should fail immediately on bad jq expression")
 	}
 }
 
+// captureSink is a minimal Sink for unit-testing processAndOutput directly.
+type captureSink struct {
+	written []json.RawMessage
+}
+
+func (s *captureSink) Write(data json.RawMessage) error {
+	s.written = append(s.written, data)
+	return nil
+}
+
+func TestProcessAndOutput_Match_DropsEvent(t *testing.T) {
+	calledProcess := false
+	keyDef := &event.KeyDefinition{
+		Key: "test.evt",
+		Match: func(raw *event.RawEvent, params map[string]string) bool {
+			return false
+		},
+		Process: func(ctx context.Context, rt event.APIClient, raw *event.RawEvent, params map[string]string) (json.RawMessage, error) {
+			calledProcess = true
+			return json.RawMessage(`{}`), nil
+		},
+	}
+	sink := &captureSink{}
+	wrote, err := processAndOutput(context.Background(), keyDef,
+		&protocol.Event{Type: protocol.MsgTypeEvent, EventType: "test.evt", Payload: json.RawMessage(`{"x":1}`)},
+		Options{}, sink, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if wrote {
+		t.Error("Match returned false but event was written")
+	}
+	if calledProcess {
+		t.Error("Process was called even though Match returned false")
+	}
+	if len(sink.written) != 0 {
+		t.Errorf("sink received %d events, want 0", len(sink.written))
+	}
+}
+
+func TestProcessAndOutput_Match_NilAcceptsAll(t *testing.T) {
+	keyDef := &event.KeyDefinition{Key: "test.evt"} // no Match, no Process
+	sink := &captureSink{}
+	wrote, err := processAndOutput(context.Background(), keyDef,
+		&protocol.Event{Type: protocol.MsgTypeEvent, EventType: "test.evt", Payload: json.RawMessage(`{"x":1}`)},
+		Options{}, sink, nil)
+	if err != nil || !wrote {
+		t.Errorf("expected wrote=true err=nil; got wrote=%v err=%v", wrote, err)
+	}
+	if len(sink.written) != 1 {
+		t.Errorf("sink received %d events, want 1", len(sink.written))
+	}
+}
+
+func TestProcessAndOutput_Match_RunsBeforeProcess(t *testing.T) {
+	// Record the actual call sequence — a bare call-count check would still
+	// pass if Process ran before Match.
+	var order []string
+	keyDef := &event.KeyDefinition{
+		Key: "test.evt",
+		Match: func(raw *event.RawEvent, params map[string]string) bool {
+			order = append(order, "match")
+			return true
+		},
+		Process: func(ctx context.Context, rt event.APIClient, raw *event.RawEvent, params map[string]string) (json.RawMessage, error) {
+			order = append(order, "process")
+			return raw.Payload, nil
+		},
+	}
+	sink := &captureSink{}
+	wrote, err := processAndOutput(context.Background(), keyDef,
+		&protocol.Event{Type: protocol.MsgTypeEvent, EventType: "test.evt", Payload: json.RawMessage(`{}`)},
+		Options{}, sink, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if !wrote {
+		t.Error("expected wrote=true")
+	}
+	if len(order) != 2 || order[0] != "match" || order[1] != "process" {
+		t.Errorf("call order = %v, want [match process]", order)
+	}
+}
+
 func TestIsTerminalSinkError(t *testing.T) {
 	for _, tc := range []struct {
 		name string

internal/event/consume/shutdown.go

@@ -16,8 +16,8 @@ const preShutdownAckTimeout = 2 * time.Second
 
 // checkLastForKey atomically reserves a cleanup lock; on any error defaults to true
 // (cleanup-on-error is safer than leaking server state). Discards non-ack frames in flight.
-func checkLastForKey(conn net.Conn, eventKey string) bool {
-	msg := protocol.NewPreShutdownCheck(eventKey)
+func checkLastForKey(conn net.Conn, eventKey string, subscriptionID string) bool {
+	msg := protocol.NewPreShutdownCheck(eventKey, subscriptionID)
 	if err := protocol.EncodeWithDeadline(conn, msg, protocol.WriteTimeout); err != nil {
 		return true
 	}

internal/event/consume/shutdown_test.go

@@ -4,6 +4,8 @@
 package consume
 
 import (
+	"bufio"
+	"bytes"
 	"encoding/json"
 	"io"
 	"net"
@@ -38,7 +40,7 @@ func TestCheckLastForKey_IgnoresNonAckFrames(t *testing.T) {
 		}
 	}()
 
-	got := checkLastForKey(client, "im.msg")
+	got := checkLastForKey(client, "im.msg", "")
 	if got != false {
 		t.Errorf("checkLastForKey = %v, want false", got)
 	}
@@ -62,7 +64,7 @@ func TestCheckLastForKey_ReturnsAckValue(t *testing.T) {
 		_ = protocol.Encode(server, ack)
 	}()
 
-	got := checkLastForKey(client, "im.msg")
+	got := checkLastForKey(client, "im.msg", "")
 	if got != true {
 		t.Errorf("checkLastForKey = %v, want true", got)
 	}
@@ -83,7 +85,7 @@ func TestCheckLastForKey_DefaultsToTrueOnTimeout(t *testing.T) {
 	}()
 
 	start := time.Now()
-	got := checkLastForKey(client, "im.msg")
+	got := checkLastForKey(client, "im.msg", "")
 	elapsed := time.Since(start)
 
 	if got != true {
@@ -93,3 +95,39 @@ func TestCheckLastForKey_DefaultsToTrueOnTimeout(t *testing.T) {
 		t.Errorf("elapsed = %v, expected ~%v (timeout-bounded)", elapsed, preShutdownAckTimeout)
 	}
 }
+
+func TestCheckLastForKey_SendsSubscriptionID(t *testing.T) {
+	a, b := net.Pipe()
+	defer a.Close()
+	defer b.Close()
+
+	done := make(chan string, 1)
+	go func() {
+		br := bufio.NewReader(b)
+		line, err := protocol.ReadFrame(br)
+		if err != nil {
+			done <- "READ_ERR"
+			return
+		}
+		msg, err := protocol.Decode(bytes.TrimRight(line, "\n"))
+		if err != nil {
+			done <- "DECODE_ERR"
+			return
+		}
+		check, ok := msg.(*protocol.PreShutdownCheck)
+		if !ok {
+			done <- "WRONG_TYPE"
+			return
+		}
+		done <- check.SubscriptionID
+		// Reply with ack so client returns
+		ack := protocol.NewPreShutdownAck(true)
+		_ = protocol.EncodeWithDeadline(b, ack, protocol.WriteTimeout)
+	}()
+
+	_ = checkLastForKey(a, "mail.x", "mail.x:alice")
+	got := <-done
+	if got != "mail.x:alice" {
+		t.Errorf("PreShutdownCheck.SubscriptionID on wire = %q, want %q", got, "mail.x:alice")
+	}
+}

internal/event/protocol/codec_test.go

@@ -77,3 +77,88 @@ func TestDecodeUnknownType(t *testing.T) {
 		t.Error("expected error for unknown type")
 	}
 }
+
+func TestEncodeDecodeHello_WithSubscriptionID(t *testing.T) {
+	msg := &Hello{
+		Type:           MsgTypeHello,
+		PID:            12345,
+		EventKey:       "mail.user_mailbox.event.message_received_v1",
+		EventTypes:     []string{"mail.user_mailbox.event.message_received_v1"},
+		Version:        "v1",
+		SubscriptionID: "mail.user_mailbox.event.message_received_v1:a7Bx9Kp2Lm3Qv4Rs",
+	}
+	buf := &bytes.Buffer{}
+	if err := Encode(buf, msg); err != nil {
+		t.Fatal(err)
+	}
+	line := buf.Bytes()
+	if !bytes.Contains(line, []byte(`"subscription_id":"mail.user_mailbox.event.message_received_v1:a7Bx9Kp2Lm3Qv4Rs"`)) {
+		t.Errorf("subscription_id not serialized: %s", string(line))
+	}
+	decoded, err := Decode(bytes.TrimRight(line, "\n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	hello, ok := decoded.(*Hello)
+	if !ok {
+		t.Fatalf("expected *Hello, got %T", decoded)
+	}
+	if hello.SubscriptionID != msg.SubscriptionID {
+		t.Errorf("roundtrip subscription_id: got %q want %q", hello.SubscriptionID, msg.SubscriptionID)
+	}
+}
+
+func TestEncodeDecodeHello_EmptySubscriptionIDOmitted(t *testing.T) {
+	msg := &Hello{
+		Type:       MsgTypeHello,
+		PID:        1,
+		EventKey:   "k",
+		EventTypes: []string{"k"},
+		Version:    "v1",
+	}
+	buf := &bytes.Buffer{}
+	if err := Encode(buf, msg); err != nil {
+		t.Fatal(err)
+	}
+	if bytes.Contains(buf.Bytes(), []byte("subscription_id")) {
+		t.Errorf("empty subscription_id should be omitted: %s", buf.String())
+	}
+	decoded, _ := Decode(bytes.TrimRight(buf.Bytes(), "\n"))
+	hello := decoded.(*Hello)
+	if hello.SubscriptionID != "" {
+		t.Errorf("got %q, want empty", hello.SubscriptionID)
+	}
+}
+
+func TestEncodeDecodePreShutdownCheck_WithSubscriptionID(t *testing.T) {
+	msg := &PreShutdownCheck{
+		Type:           MsgTypePreShutdownCheck,
+		EventKey:       "mail.x",
+		SubscriptionID: "mail.x:abc",
+	}
+	buf := &bytes.Buffer{}
+	if err := Encode(buf, msg); err != nil {
+		t.Fatal(err)
+	}
+	decoded, err := Decode(bytes.TrimRight(buf.Bytes(), "\n"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := decoded.(*PreShutdownCheck)
+	if got.SubscriptionID != msg.SubscriptionID {
+		t.Errorf("roundtrip: got %q want %q", got.SubscriptionID, msg.SubscriptionID)
+	}
+}
+
+func TestStatusResponse_ConsumerInfo_SubscriptionID(t *testing.T) {
+	msg := NewStatusResponse(7, 120, 1, []ConsumerInfo{
+		{PID: 99, EventKey: "mail.x", SubscriptionID: "mail.x:abc", Received: 5, Dropped: 0},
+	})
+	buf := &bytes.Buffer{}
+	if err := Encode(buf, msg); err != nil {
+		t.Fatal(err)
+	}
+	if !bytes.Contains(buf.Bytes(), []byte(`"subscription_id":"mail.x:abc"`)) {
+		t.Errorf("ConsumerInfo.SubscriptionID missing from JSON: %s", buf.String())
+	}
+}

internal/event/protocol/messages.go

@@ -34,11 +34,12 @@ type SourceStatus struct {
 }
 
 type Hello struct {
-	Type       string   `json:"type"`
-	PID        int      `json:"pid"`
-	EventKey   string   `json:"event_key"`
-	EventTypes []string `json:"event_types"`
-	Version    string   `json:"version"`
+	Type           string   `json:"type"`
+	PID            int      `json:"pid"`
+	EventKey       string   `json:"event_key"`
+	EventTypes     []string `json:"event_types"`
+	Version        string   `json:"version"`
+	SubscriptionID string   `json:"subscription_id,omitempty"` // empty = fallback to EventKey on bus side
 }
 
 type HelloAck struct {
@@ -61,10 +62,11 @@ type Bye struct {
 	Type string `json:"type"`
 }
 
-// PreShutdownCheck atomically reserves the cleanup lock for EventKey.
+// PreShutdownCheck atomically reserves the cleanup lock for (EventKey, SubscriptionID).
 type PreShutdownCheck struct {
-	Type     string `json:"type"`
-	EventKey string `json:"event_key"`
+	Type           string `json:"type"`
+	EventKey       string `json:"event_key"`
+	SubscriptionID string `json:"subscription_id,omitempty"` // empty = fallback to EventKey
 }
 
 type PreShutdownAck struct {
@@ -77,10 +79,11 @@ type StatusQuery struct {
 }
 
 type ConsumerInfo struct {
-	PID      int    `json:"pid"`
-	EventKey string `json:"event_key"`
-	Received int64  `json:"received"`
-	Dropped  int64  `json:"dropped"`
+	PID            int    `json:"pid"`
+	EventKey       string `json:"event_key"`
+	SubscriptionID string `json:"subscription_id,omitempty"`
+	Received       int64  `json:"received"`
+	Dropped        int64  `json:"dropped"`
 }
 
 type StatusResponse struct {
@@ -95,13 +98,14 @@ type Shutdown struct {
 	Type string `json:"type"`
 }
 
-func NewHello(pid int, eventKey string, eventTypes []string, version string) *Hello {
+func NewHello(pid int, eventKey string, eventTypes []string, version string, subscriptionID string) *Hello {
 	return &Hello{
-		Type:       MsgTypeHello,
-		PID:        pid,
-		EventKey:   eventKey,
-		EventTypes: eventTypes,
-		Version:    version,
+		Type:           MsgTypeHello,
+		PID:            pid,
+		EventKey:       eventKey,
+		EventTypes:     eventTypes,
+		Version:        version,
+		SubscriptionID: subscriptionID,
 	}
 }
 
@@ -124,8 +128,8 @@ func NewEvent(eventType, eventID, sourceTime string, seq uint64, payload json.Ra
 	}
 }
 
-func NewPreShutdownCheck(eventKey string) *PreShutdownCheck {
-	return &PreShutdownCheck{Type: MsgTypePreShutdownCheck, EventKey: eventKey}
+func NewPreShutdownCheck(eventKey, subscriptionID string) *PreShutdownCheck {
+	return &PreShutdownCheck{Type: MsgTypePreShutdownCheck, EventKey: eventKey, SubscriptionID: subscriptionID}
 }
 
 func NewPreShutdownAck(lastForKey bool) *PreShutdownAck {

internal/event/protocol/messages_test.go

@@ -17,7 +17,7 @@ import (
 
 // Every NewXxx helper must set the Type discriminator (Decode rejects messages without it).
 func TestConstructors_PinTypeField(t *testing.T) {
-	if got := NewHello(1, "k", []string{"t"}, "v1"); got.Type != MsgTypeHello {
+	if got := NewHello(1, "k", []string{"t"}, "v1", ""); got.Type != MsgTypeHello {
 		t.Errorf("NewHello.Type = %q, want %q", got.Type, MsgTypeHello)
 	}
 	if got := NewHelloAck("v1", true); got.Type != MsgTypeHelloAck || !got.FirstForKey {
@@ -26,7 +26,7 @@ func TestConstructors_PinTypeField(t *testing.T) {
 	if got := NewEvent("im.msg", "e1", "", 7, json.RawMessage(`{}`)); got.Type != MsgTypeEvent || got.Seq != 7 {
 		t.Errorf("NewEvent mismatch: %+v", got)
 	}
-	if got := NewPreShutdownCheck("k"); got.Type != MsgTypePreShutdownCheck || got.EventKey != "k" {
+	if got := NewPreShutdownCheck("k", ""); got.Type != MsgTypePreShutdownCheck || got.EventKey != "k" {
 		t.Errorf("NewPreShutdownCheck mismatch: %+v", got)
 	}
 	if got := NewPreShutdownAck(true); got.Type != MsgTypePreShutdownAck || !got.LastForKey {
@@ -63,7 +63,7 @@ func TestEncode_DecodeRoundtripAllTypes(t *testing.T) {
 		}
 	}
 	roundtrip(t, NewHelloAck("v1", true), &HelloAck{})
-	roundtrip(t, NewPreShutdownCheck("im.msg"), &PreShutdownCheck{})
+	roundtrip(t, NewPreShutdownCheck("im.msg", ""), &PreShutdownCheck{})
 	roundtrip(t, NewPreShutdownAck(false), &PreShutdownAck{})
 	roundtrip(t, NewStatusQuery(), &StatusQuery{})
 	roundtrip(t, NewStatusResponse(7, 120, 1, []ConsumerInfo{{PID: 99, EventKey: "k"}}), &StatusResponse{})

internal/event/types.go

@@ -55,6 +55,23 @@ type ParamDef struct {
 	Default     string       `json:"default,omitempty"`
 	Description string       `json:"description"`
 	Values      []ParamValue `json:"values,omitempty"`
+
+	// SubscriptionKey marks this param as part of the subscription identity.
+	// Two consumers of the same EventKey but different values for any
+	// SubscriptionKey-marked param are treated as DISTINCT subscriptions:
+	// PreConsume runs once per (EventKey, SubscriptionID), cleanup runs once per
+	// (EventKey, SubscriptionID).
+	//
+	// CONTRACT: only mark a param SubscriptionKey if the EventKey's server-side
+	// subscribe/unsubscribe API is itself scoped to that resource. Lark keys the
+	// subscription record by (app, user, event_type) and overwrites it rather
+	// than reference-counting, so for a non-per-resource API the cleanup of one
+	// resource's last consumer unsubscribes the shared record and silently cuts
+	// off every other resource sharing that event_type.
+	//
+	// Default false = the param is a filter / formatting / metadata param
+	// and does not affect subscription identity.
+	SubscriptionKey bool `json:"subscription_key,omitempty"`
 }
 
 type ProcessFunc = func(ctx context.Context, rt APIClient, raw *RawEvent, params map[string]string) (json.RawMessage, error)
@@ -83,10 +100,44 @@ type KeyDefinition struct {
 
 	Schema SchemaDef `json:"schema"`
 
+	// NormalizeParams canonicalizes param values BEFORE fingerprint compute,
+	// PreConsume, Match, and Process. Mutates the params map in place.
+	// May call OAPI; runs once per consumer at startup.
+	//
+	// Use cases: resolve aliases ("me" -> real email, a name -> an ID),
+	// trim whitespace. On error, consume fails (no retry); caller gets the
+	// wrapped error.
+	//
+	// Default nil = no normalization, params pass through unchanged.
+	NormalizeParams func(ctx context.Context, rt APIClient, params map[string]string) error `json:"-"`
+
 	// Process required when Schema.Custom is Processed output; must be nil when Native is used.
+	//
+	// Convention: returning (nil, nil) signals "drop this event" — the
+	// consumer loop will skip writing it to sink and not advance the
+	// emitted counter. Useful for async filtering (e.g. fetch metadata,
+	// drop if folder doesn't match). For sync filters that don't need
+	// OAPI, use Match instead.
 	Process func(ctx context.Context, rt APIClient, raw *RawEvent, params map[string]string) (json.RawMessage, error) `json:"-"`
 
-	PreConsume func(ctx context.Context, rt APIClient, params map[string]string) (cleanup func(), err error) `json:"-"`
+	// Match is a synchronous payload filter run on every received event
+	// BEFORE Process. Return false to drop the event without further work.
+	//
+	// Signature deliberately omits ctx/rt to physically enforce "no OAPI
+	// calls in Match". For filters that need a metadata fetch first, use
+	// Process and return nil to drop.
+	//
+	// Default nil = accept all events.
+	Match func(raw *RawEvent, params map[string]string) bool `json:"-"`
+
+	// PreConsume runs once per (EventKey, SubscriptionID) when this consumer
+	// is first for that scope. Returns a cleanup function that the framework
+	// invokes when this consumer is the last for its scope.
+	//
+	// The cleanup's error return is honored: on nil the framework prints
+	// "[event] cleanup done."; on non-nil it prints a WARN with an
+	// idempotency note.
+	PreConsume func(ctx context.Context, rt APIClient, params map[string]string) (cleanup func() error, err error) `json:"-"`
 
 	Scopes []string `json:"scopes,omitempty"`
 

shortcuts/apps/walk_html_publish_candidates.go

@@ -56,6 +56,15 @@ func walkHTMLPublishCandidates(fio fileio.FileIO, rootPath string) ([]htmlPublis
 		if walkErr != nil {
 			return walkErr
 		}
+		// Skip a stray git repo: a directory named .git skips the whole subtree,
+		// and a .git file (the gitdir pointer used by submodules/worktrees) is
+		// skipped too.
+		if d.Name() == ".git" {
+			if d.IsDir() {
+				return filepath.SkipDir
+			}
+			return nil
+		}
 		if d.IsDir() {
 			return nil
 		}

shortcuts/apps/walk_html_publish_candidates_test.go

@@ -113,6 +113,102 @@ func TestIsUnsafeRelPath(t *testing.T) {
 	}
 }
 
+func TestWalkHTMLPublishCandidates_ExcludesGitDir(t *testing.T) {
+	dir := t.TempDir()
+	files := map[string]string{
+		"index.html":               "<html></html>",
+		".git/config":              "[core]\n",
+		".git/HEAD":                "ref: refs/heads/main\n",
+		".git/objects/ab/cdef123":  "binary",
+		".github/workflows/ci.yml": "on: push\n",
+		".gitignore":               "node_modules\n",
+	}
+	for rel, content := range files {
+		full := filepath.Join(dir, rel)
+		if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
+			t.Fatalf("mkdir: %v", err)
+		}
+		if err := os.WriteFile(full, []byte(content), 0o644); err != nil {
+			t.Fatalf("write: %v", err)
+		}
+	}
+
+	got, err := walkHTMLPublishCandidates(newTestFIO(), dir)
+	if err != nil {
+		t.Fatalf("err=%v", err)
+	}
+	rels := make(map[string]bool)
+	for _, c := range got {
+		rels[c.RelPath] = true
+	}
+	// .git 子树整体排除
+	for _, banned := range []string{".git/config", ".git/HEAD", ".git/objects/ab/cdef123"} {
+		if rels[banned] {
+			t.Errorf("%q should be excluded from candidates", banned)
+		}
+	}
+	// 相邻名不受影响
+	for _, kept := range []string{"index.html", ".github/workflows/ci.yml", ".gitignore"} {
+		if !rels[kept] {
+			t.Errorf("%q should be kept in candidates, got %+v", kept, got)
+		}
+	}
+}
+
+func TestWalkHTMLPublishCandidates_ExcludesNestedGitDir(t *testing.T) {
+	dir := t.TempDir()
+	files := map[string]string{
+		"index.html":      "<html></html>",
+		"sub/.git/config": "[core]\n",
+		"sub/page.html":   "<html></html>",
+	}
+	for rel, content := range files {
+		full := filepath.Join(dir, rel)
+		if err := os.MkdirAll(filepath.Dir(full), 0o755); err != nil {
+			t.Fatalf("mkdir: %v", err)
+		}
+		if err := os.WriteFile(full, []byte(content), 0o644); err != nil {
+			t.Fatalf("write: %v", err)
+		}
+	}
+
+	got, err := walkHTMLPublishCandidates(newTestFIO(), dir)
+	if err != nil {
+		t.Fatalf("err=%v", err)
+	}
+	rels := make(map[string]bool)
+	for _, c := range got {
+		rels[c.RelPath] = true
+	}
+	if rels["sub/.git/config"] {
+		t.Errorf("nested sub/.git/config should be excluded, got %+v", got)
+	}
+	if !rels["sub/page.html"] || !rels["index.html"] {
+		t.Errorf("non-git files should be kept, got %+v", got)
+	}
+}
+
+func TestWalkHTMLPublishCandidates_ExcludesGitFile(t *testing.T) {
+	// submodule / worktree 场景：.git 是个 gitdir 指针文件，不是目录。
+	dir := t.TempDir()
+	if err := os.WriteFile(filepath.Join(dir, "index.html"), []byte("<html></html>"), 0o644); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+	if err := os.WriteFile(filepath.Join(dir, ".git"), []byte("gitdir: /elsewhere\n"), 0o644); err != nil {
+		t.Fatalf("write: %v", err)
+	}
+
+	got, err := walkHTMLPublishCandidates(newTestFIO(), dir)
+	if err != nil {
+		t.Fatalf("err=%v", err)
+	}
+	for _, c := range got {
+		if c.RelPath == ".git" {
+			t.Errorf(".git pointer file should be excluded, got %+v", got)
+		}
+	}
+}
+
 func TestWalkHTMLPublishCandidates_SymlinkSkipped(t *testing.T) {
 	// Walker 只接受 regular file —— symlink 跳过（避免 loop + out-of-root 引用，
 	// 且 fio.Open 对 symlink 行为不一致）。real.html 仍然被收，link.html 不在结果里。

shortcuts/im/builders_test.go

@@ -317,6 +317,17 @@ func TestShortcutValidateBranches(t *testing.T) {
 		}
 	})
 
+	t.Run("ImChatSearch invalid chat-modes value", func(t *testing.T) {
+		runtime := newTestRuntimeContext(t, map[string]string{
+			"query":      "ok",
+			"chat-modes": "group,bogus",
+		}, nil)
+		err := ImChatSearch.Validate(context.Background(), runtime)
+		if err == nil || !strings.Contains(err.Error(), "invalid --chat-modes value") {
+			t.Fatalf("ImChatSearch.Validate() error = %v", err)
+		}
+	})
+
 	t.Run("ImChatUpdate requires fields", func(t *testing.T) {
 		runtime := newTestRuntimeContext(t, map[string]string{
 			"chat-id": "oc_123",
@@ -693,6 +704,39 @@ func TestShortcutDryRunShapes(t *testing.T) {
 		}
 	})
 
+	t.Run("ImChatSearch dry run maps chat-modes to wire values", func(t *testing.T) {
+		runtime := newTestRuntimeContext(t, map[string]string{
+			"query":      "team-alpha",
+			"chat-modes": "group,topic",
+		}, nil)
+		got := mustMarshalDryRun(t, ImChatSearch.DryRun(context.Background(), runtime))
+		if !strings.Contains(got, `"chat_modes":["default","thread"]`) {
+			t.Fatalf("ImChatSearch.DryRun() chat_modes mapping = %s", got)
+		}
+	})
+
+	t.Run("ImChatSearch dry run maps single chat-mode topic", func(t *testing.T) {
+		runtime := newTestRuntimeContext(t, map[string]string{
+			"query":      "team-alpha",
+			"chat-modes": "topic",
+		}, nil)
+		got := mustMarshalDryRun(t, ImChatSearch.DryRun(context.Background(), runtime))
+		if !strings.Contains(got, `"chat_modes":["thread"]`) {
+			t.Fatalf("ImChatSearch.DryRun() chat_modes mapping = %s", got)
+		}
+	})
+
+	t.Run("ImChatSearch dry run dedupes chat-modes", func(t *testing.T) {
+		runtime := newTestRuntimeContext(t, map[string]string{
+			"query":      "team-alpha",
+			"chat-modes": "group, group",
+		}, nil)
+		got := mustMarshalDryRun(t, ImChatSearch.DryRun(context.Background(), runtime))
+		if !strings.Contains(got, `"chat_modes":["default"]`) {
+			t.Fatalf("ImChatSearch.DryRun() chat_modes dedupe = %s", got)
+		}
+	})
+
 	t.Run("ImMessagesSearch dry run uses messages search endpoint", func(t *testing.T) {
 		runtime := newMessagesSearchTestRuntimeContext(t, map[string]string{
 			"query":      "incident",

shortcuts/im/im_chat_search.go

@@ -31,6 +31,7 @@ var ImChatSearch = common.Shortcut{
 	Flags: []common.Flag{
 		{Name: "query", Desc: "search keyword (max 64 chars)"},
 		{Name: "search-types", Desc: "chat types, comma-separated (private, external, public_joined, public_not_joined)"},
+		{Name: "chat-modes", Desc: "filter by chat mode, comma-separated (group, topic)"},
 		{Name: "member-ids", Desc: "filter by member open_ids, comma-separated"},
 		{Name: "is-manager", Type: "bool", Desc: "only show chats you created or manage"},
 		{Name: "disable-search-by-user", Type: "bool", Desc: "disable search-by-member-name (default: search by member name first, then group name)"},
@@ -72,6 +73,13 @@ var ImChatSearch = common.Shortcut{
 				}
 			}
 		}
+		if cm := runtime.Str("chat-modes"); cm != "" {
+			for _, mode := range common.SplitCSV(cm) {
+				if mode != "group" && mode != "topic" {
+					return errs.NewValidationError(errs.SubtypeInvalidArgument, "invalid --chat-modes value %q: expected one of group, topic", mode).WithParam("--chat-modes")
+				}
+			}
+		}
 		if mi := runtime.Str("member-ids"); mi != "" {
 			ids := common.SplitCSV(mi)
 			if len(ids) > 50 {
@@ -217,6 +225,24 @@ func buildSearchChatBody(runtime *common.RuntimeContext) map[string]interface{}
 	if st := runtime.Str("search-types"); st != "" {
 		filter["search_types"] = common.SplitCSV(st)
 	}
+	// chat_modes is a server-side filter. The CLI exposes group/topic; the wire
+	// expects default/thread. Map and dedupe (the API caps the list at 2, and
+	// there are only 2 distinct modes) while preserving the user's order.
+	if cm := runtime.Str("chat-modes"); cm != "" {
+		seen := map[string]bool{}
+		var modes []string
+		for _, mode := range common.SplitCSV(cm) {
+			wire := map[string]string{"group": "default", "topic": "thread"}[mode]
+			if wire == "" || seen[wire] {
+				continue
+			}
+			seen[wire] = true
+			modes = append(modes, wire)
+		}
+		if len(modes) > 0 {
+			filter["chat_modes"] = modes
+		}
+	}
 	if mi := runtime.Str("member-ids"); mi != "" {
 		filter["member_ids"] = common.SplitCSV(mi)
 	}

skills/lark-apps/references/lark-apps-html-publish.md

@@ -38,7 +38,7 @@ lark-cli apps +html-publish --app-id app_xxx --path ./index.html --dry-run
 - 用户只说“用 HTML 写个 PPT/页面给我看看”时，先生成本地文件或目录，返回路径并问是否发布到妙搭分享；不要默认创建应用或部署。
 - 用户明确说“部署出去/发链接/可分享”时，才创建 `html` 应用并用 `+html-publish`。
 - 用户要发布但没有 app_id 时，先 `+create --app-type html` 创建应用；应用名可从页面/站点主题生成，不要让用户手动提供 app_id。
-- 若产物首页不是 `index.html`，发布前改名或复制为 `index.html`；目录发布时只传干净产物目录，例如 `./dist`，不要把 `.git`、`node_modules`、源码缓存一起带上。
+- 若产物首页不是 `index.html`，发布前改名或复制为 `index.html`；目录发布时只传干净产物目录，例如 `./dist`。`.git` 目录会被自动排除，不会进入压缩包；`node_modules`、源码缓存等仍建议手动精简以控制包体。
 - 重新部署同一个 HTML 应用时复用原 `app_id`，只重新执行 `+html-publish --app-id <id> --path <dir-or-index.html>`。
 
 ## 安全规则

skills/lark-doc/references/lark-doc-xml.md

@@ -78,6 +78,12 @@ p, h1-h9, ul, ol, li, table, thead, tbody, tr, th, td, blockquote, pre, code, hr
    ```
 
 
+## 用户名写入规则
+
+- 当从 IM 消息、日历、审批、任务等来源获取到用户的 `open_id` 时，写入文档**必须**使用 `<cite type="user" user-id="open_id">` 标签，而非纯文本名字。这样文档中会渲染为可点击的 @人。
+- 典型场景：IM 消息的 `sender`、`mentions`、reactions 的 `operator`、卡片消息中引用的用户、系统消息中的用户名、合并转发中的用户名。
+- 当只有纯文本名字而没有 `open_id` 时（如系统消息、合并转发内容），先通过 `lark-cli contact +search-user --query "名字" --as user` 反查 `open_id`，再写入 cite 标签。
+
 ## 表格扩展
 标准 HTML table 结构不变，扩展点：
 - `<colgroup>` / `<col>` 定义列宽，紧跟 `<table>` 之后：`<col span="2" width="100"/>`

skills/lark-doc/references/style/lark-doc-style.md

@@ -7,7 +7,7 @@
 1. **结构优于文字**：能用结构化 block 表达的信息，不用纯文本段落
 2. **Front-load 结论**：文档以 `<callout>` 开头概括核心结论；每章节首段点明要旨
 3. **视觉节奏**：连续纯文本不超过 3 段；不同主题章节间用 `<hr/>` 分隔
-4. **最少惊讶**：同类信息使用同类元素，全篇风格统一
+4. **风格一致**：同类信息使用同类元素，全篇风格统一
 5. **重要信息画板化**：核心流程、架构、对比、风险、路线图、指标趋势等重要信息优先使用画板表达
 
 ## 二、元素选择指南
@@ -24,7 +24,7 @@
 | 代码片段                                       | `<pre lang="x" caption="说明">`         |
 | 引用 / 公式                                    | `<blockquote>` / `<latex>`            |
 | 操作入口 / 跳转链接                                | `<button>` / `<a type="url-preview">` |
-| 流程图 / 时间线 / 示意图 / 自定义图形 / 架构图 / 数据图 /思维导图等 | 画板图表                                  |
+| 流程图 / 时间线 / 示意图 / 自定义图形 / 架构图 / 数据图 / 思维导图等 | 画板图表                                  |
 
 
 ### 画板意图识别
@@ -69,7 +69,7 @@
 ## 四、排版规范
 
 - 标题层级 ≤ 4 层，段落单段 ≤ 5 行，列表嵌套 ≤ 2 层，Grid ≤ 3 列
-- 文档开头用 `<callout>` front-load 结论；
+- 文档开头用 `<callout>` front-load 结论。
 
 ## 五、丰富度自检
 

skills/lark-im/SKILL.md

@@ -134,6 +134,21 @@ lark-cli im <resource> <method> [flags] # 调用 API
   - `delete` — 将用户或机器人移出群聊。Identity: supports `user` and `bot`; only group owner, admin, or creator bot can remove others; max 50 users or 5 bots per request.
   - `get` — 获取群成员列表。Identity: supports `user` and `bot`; the caller must be in the target chat and must belong to the same tenant for internal chats.
 
+### chat.user_setting
+
+  - `batch_query` — 批量查询当前用户在群内的个人偏好设置 (e.g. `is_muted` mutes normal messages, `is_mute_at_all` mutes @all messages); up to 10 chats per request. Identity: `user` only (`user_access_token`); the caller must be in each target chat.
+  - `batch_update` — 批量更新当前用户在群内的个人偏好设置 (e.g. `is_muted` mutes normal messages, `is_mute_at_all` mutes @all messages); up to 10 chats per request. Identity: `user` only (`user_access_token`); the caller must be in each target chat.
+
+### chat.managers
+
+  - `add_managers` — 指定群管理员。Identity: supports `user` and `bot`; only the group owner can add managers; max 10 managers per chat (20 for super-large chats), and at most 5 bots per request.
+  - `delete_managers` — 删除群管理员。Identity: supports `user` and `bot`; only the group owner can remove managers; max 50 users or 5 bots per request.
+
+### chat.moderation
+
+  - `get` — 获取群成员发言权限。Identity: supports `user` and `bot`; the caller must be in the target chat and belong to the same tenant.
+  - `update` — 更新群发言权限。Identity: supports `user` and `bot`; only the group owner (or creator bot with `im:chat:operate_as_owner`) can update; the caller must be in the chat.
+
 ### messages
 
   - `delete` — 撤回消息。Identity: supports `user` and `bot`; for `bot` calls, the bot must be in the chat to revoke group messages; to revoke another user's group message, the bot must be the owner, an admin, or the creator; for user P2P recalls, the target user must be within the bot's availability.
@@ -186,6 +201,12 @@ lark-cli im <resource> <method> [flags] # 调用 API
 | `chat.members.create` | `im:chat.members:write_only` |
 | `chat.members.delete` | `im:chat.members:write_only` |
 | `chat.members.get` | `im:chat.members:read` |
+| `chat.user_setting.batch_query` | `im:chat.user_setting:read` |
+| `chat.user_setting.batch_update` | `im:chat.user_setting:write` |
+| `chat.managers.add_managers` | `im:chat.managers:write_only` |
+| `chat.managers.delete_managers` | `im:chat.managers:write_only` |
+| `chat.moderation.get` | `im:chat.moderation:read` |
+| `chat.moderation.update` | `im:chat:moderation:write_only` |
 | `messages.delete` | `im:message:recall` |
 | `messages.forward` | `im:message` |
 | `messages.merge_forward` | `im:message` |

skills/lark-im/references/lark-im-chat-search.md

@@ -15,6 +15,9 @@ lark-cli im +chat-search --query "project"
 # Restrict by search types
 lark-cli im +chat-search --query "project" --search-types "private,public_joined"
 
+# Filter by chat mode (group = regular group, topic = topic/thread group)
+lark-cli im +chat-search --query "project" --chat-modes "topic"
+
 # Filter by member open_ids (with keyword)
 lark-cli im +chat-search --query "project" --member-ids "ou_xxx,ou_yyy"
 
@@ -43,6 +46,7 @@ lark-cli im +chat-search --query "project" --dry-run
 |------|------|------|------|
 | `--query <keyword>` | No (at least one of `--query` / `--member-ids` required) | Max 64 characters | Search keyword. Supports matching localized chat names, member names, multilingual search, pinyin, and prefix fuzzy search. If the query contains `-`, it is automatically wrapped in quotes |
 | `--search-types <types>` | No | Comma-separated: `private`, `external`, `public_joined`, `public_not_joined` | Restrict the visible chat types returned by search |
+| `--chat-modes <modes>` | No | Comma-separated: `group`, `topic` | Filter by chat mode (server-side): `group` = regular group, `topic` = topic/thread group |
 | `--member-ids <ids>` | No (at least one of `--query` / `--member-ids` required) | Up to 50, format `ou_xxx` | Filter by member open_ids; can be used alone or combined with `--query` |
 | `--is-manager` | No | - | Only show chats you created or manage |
 | `--disable-search-by-user` | No | - | Disable member-name-based matching and search by group name only |

tests/cli_e2e/apps/apps_html_publish_dryrun_test.go

@@ -87,9 +87,11 @@ func TestAppsHTMLPublishDryRun(t *testing.T) {
 		assert.Equal(t, "page.html", gjson.Get(result.Stdout, "files.0").String())
 	})
 
-	t.Run("HiddenFilesIncluded", func(t *testing.T) {
-		// Walker MUST NOT silently filter .git / .DS_Store — that's an explicit
-		// design decision so users pass clean ./dist trees, not source repos.
+	t.Run("HiddenFilesIncludedExceptGit", func(t *testing.T) {
+		// The walker filters the .git directory (and a .git gitdir pointer file)
+		// so a stray repo under --path doesn't ship its history / remote URL to a
+		// public share URL. Generic hidden files like .DS_Store are NOT filtered —
+		// only .git is — so users still see everything else they pointed --path at.
 		dir := t.TempDir()
 		require.NoError(t, os.MkdirAll(filepath.Join(dir, "dist"), 0o755))
 		require.NoError(t, os.WriteFile(filepath.Join(dir, "dist", "index.html"), []byte("<html/>"), 0o644))
@@ -112,8 +114,17 @@ func TestAppsHTMLPublishDryRun(t *testing.T) {
 		})
 		require.NoError(t, err)
 		result.AssertExitCode(t, 0)
-		assert.Equal(t, int64(3), gjson.Get(result.Stdout, "file_count").Int(),
-			"walker must include hidden files; got: %s", result.Stdout)
+		// index.html + .DS_Store kept; .git/HEAD filtered out → 2 files.
+		assert.Equal(t, int64(2), gjson.Get(result.Stdout, "file_count").Int(),
+			"walker must keep non-.git hidden files but drop .git; got: %s", result.Stdout)
+		names := gjson.Get(result.Stdout, "files").Array()
+		var got []string
+		for _, n := range names {
+			got = append(got, n.String())
+		}
+		assert.Contains(t, got, "index.html")
+		assert.Contains(t, got, ".DS_Store")
+		assert.NotContains(t, got, ".git/HEAD", "walker must exclude .git contents")
 	})
 
 	t.Run("EmptyDir_ManifestEmpty", func(t *testing.T) {