feat(event): list valid domains in the --domain flag help

Frozen baseline outputs now validate against their declared schemas;
process-error diagnostics truncate before payload fragments can reach
stderr; the render contract gate recurses into nested named types; the
snapshot deep-copies field-override enums; hosts are gated against
importing the platform SDK; a credential-vocabulary guard keeps future
sensitive parameters inside the redaction net.

cmd/build.go

@@ -20,7 +20,6 @@ import (
 	"github.com/larksuite/cli/cmd/skill"
 	cmdupdate "github.com/larksuite/cli/cmd/update"
 	"github.com/larksuite/cli/cmd/whoami"
-	_ "github.com/larksuite/cli/events"
 	"github.com/larksuite/cli/internal/apicatalog"
 	"github.com/larksuite/cli/internal/build"
 	"github.com/larksuite/cli/internal/cmdpolicy"

cmd/event/bus.go

@@ -16,12 +16,14 @@ import (
 	"github.com/larksuite/cli/internal/cmdutil"
 	"github.com/larksuite/cli/internal/core"
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/adapter/lark/websocket"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/transport"
 	"github.com/larksuite/cli/internal/event/bus"
-	"github.com/larksuite/cli/internal/event/transport"
+	"github.com/larksuite/cli/internal/event/catalog"
 )
 
 // NewCmdBus creates the hidden `event _bus` daemon subcommand, forked by the consume client; fork argv lives in consume/startup.go.
-func NewCmdBus(f *cmdutil.Factory) *cobra.Command {
+func NewCmdBus(f *cmdutil.Factory, snap *catalog.Snapshot) *cobra.Command {
 	var domain string
 
 	cmd := &cobra.Command{
@@ -44,7 +46,13 @@ func NewCmdBus(f *cmdutil.Factory) *cobra.Command {
 			}
 
 			tr := transport.New()
-			b := bus.NewBus(cfg.AppID, cfg.AppSecret, domain, tr, logger)
+			ingress := &websocket.FeishuSource{
+				AppID:     cfg.AppID,
+				AppSecret: cfg.AppSecret,
+				Domain:    domain,
+				Logger:    logger,
+			}
+			b := bus.NewBus(cfg.AppID, cfg.AppSecret, domain, tr, logger, snap, ingress)
 
 			ctx, cancel := context.WithCancel(cmd.Context())
 			defer cancel()

cmd/event/bus_test.go

@@ -27,7 +27,7 @@ func TestBusCommandLoggerSetupFailureIsTypedFileIO(t *testing.T) {
 	f, _, _, _ := cmdutil.TestFactory(t, &core.CliConfig{
 		AppID: "cli_bus_test", AppSecret: "secret", Brand: core.BrandFeishu,
 	})
-	cmd := NewCmdBus(f)
+	cmd := NewCmdBus(f, compileCatalog())
 	cmd.SetArgs([]string{})
 
 	err := cmd.Execute()

cmd/event/consume.go

@@ -16,6 +16,7 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"github.com/larksuite/cli/cmd/event/render"
 	"github.com/larksuite/cli/errs"
 	"github.com/larksuite/cli/internal/appmeta"
 	"github.com/larksuite/cli/internal/auth"
@@ -23,8 +24,10 @@ import (
 	"github.com/larksuite/cli/internal/core"
 	"github.com/larksuite/cli/internal/credential"
 	eventlib "github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/transport"
+	appconsume "github.com/larksuite/cli/internal/event/application/consume"
+	"github.com/larksuite/cli/internal/event/catalog"
 	"github.com/larksuite/cli/internal/event/consume"
-	"github.com/larksuite/cli/internal/event/transport"
 	"github.com/larksuite/cli/internal/output"
 	"github.com/larksuite/cli/internal/validate"
 )
@@ -37,9 +40,10 @@ type consumeCmdOpts struct {
 
 	maxEvents int
 	timeout   time.Duration
+	dryRun    bool
 }
 
-func NewCmdConsume(f *cmdutil.Factory) *cobra.Command {
+func NewCmdConsume(f *cmdutil.Factory, snap *catalog.Snapshot) *cobra.Command {
 	var o consumeCmdOpts
 
 	cmd := &cobra.Command{
@@ -57,7 +61,7 @@ Use 'event list' to see all available EventKeys.
 Use 'event schema <EventKey>' for parameter details.`,
 		Args: cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			return runConsume(cmd, f, args[0], o)
+			return runConsume(cmd, f, snap, args[0], o)
 		},
 	}
 
@@ -66,6 +70,7 @@ Use 'event schema <EventKey>' for parameter details.`,
 	cmd.Flags().BoolVar(&o.quiet, "quiet", false, "Suppress informational messages on stderr")
 	cmd.Flags().StringVar(&o.outputDir, "output-dir", "", "Write each event as a file in this directory (relative paths only; absolute paths and ~ are rejected to prevent path traversal)")
 	cmd.Flags().IntVar(&o.maxEvents, "max-events", 0, "Exit after N successful emits (0 = unlimited). Multi-worker EventKeys may emit up to workers-1 past N before all workers stop. Bounded runs ignore stdin EOF.")
+	cmd.Flags().BoolVar(&o.dryRun, "dry-run", false, "Decide and preview the consume (identity, preconditions, side effects) without performing any of them, then exit")
 	cmd.Flags().DurationVar(&o.timeout, "timeout", 0, "Exit after DURATION (e.g. 30s, 2m). 0 = no timeout. Timeout is a normal exit (code 0; stderr 'reason: timeout'). Bounded runs ignore stdin EOF.")
 	cmd.Flags().String("as", "auto", "identity type: user | bot | auto (must match EventKey's declared AuthTypes)")
 	_ = cmd.RegisterFlagCompletionFunc("as", func(cmd *cobra.Command, args []string, toComplete string) ([]string, cobra.ShellCompDirective) {
@@ -76,7 +81,7 @@ Use 'event schema <EventKey>' for parameter details.`,
 	return cmd
 }
 
-func runConsume(cmd *cobra.Command, f *cmdutil.Factory, eventKey string, o consumeCmdOpts) error {
+func runConsume(cmd *cobra.Command, f *cmdutil.Factory, snap *catalog.Snapshot, eventKey string, o consumeCmdOpts) error {
 	// Pipe-close (e.g. `... | head -n 1`) must reach the EPIPE error path in the loop, not SIGPIPE-kill.
 	ignoreBrokenPipe()
 
@@ -90,10 +95,11 @@ func runConsume(cmd *cobra.Command, f *cmdutil.Factory, eventKey string, o consu
 		return err
 	}
 
-	keyDef, ok := eventlib.Lookup(eventKey)
+	entry, ok := snap.Resolve(eventKey)
 	if !ok {
-		return unknownEventKeyErr(eventKey)
+		return unknownEventKeyErr(snap, eventKey)
 	}
+	keyDef := entry.Definition()
 
 	identity, err := resolveIdentity(cmd, f, keyDef)
 	if err != nil {
@@ -120,9 +126,16 @@ func runConsume(cmd *cobra.Command, f *cmdutil.Factory, eventKey string, o consu
 
 	domain := core.ResolveEndpoints(cfg.Brand).Open
 
-	// Surface auth errors before forking the bus daemon.
+	// Surface auth errors before forking the bus daemon. A dry run instead
+	// reports the unusable credential as a blocked precondition: the caller
+	// asked what would happen, and "a real run would refuse to authenticate"
+	// is a legitimate part of that answer.
+	var tokenErr error
 	if _, err := resolveTenantToken(cmd.Context(), f, cfg.AppID); err != nil {
-		return err
+		if !o.dryRun {
+			return err
+		}
+		tokenErr = err
 	}
 
 	apiClient, err := f.NewAPIClient()
@@ -169,11 +182,31 @@ func runConsume(cmd *cobra.Command, f *cmdutil.Factory, eventKey string, o consu
 		appVer:              appVer,
 		subscribedCallbacks: subscribedCallbacks,
 	}
-	if err := preflightEventTypes(pf); err != nil {
+
+	svc := &appconsume.Service{
+		Strategies: consumeStrategies,
+		Identity:   identityResolverFunc(func(context.Context, *catalog.Entry) (string, error) { return string(identity), nil }),
+		Preflight: preflightReaderFunc(func(ctx context.Context, _ *catalog.Entry, _ string) ([]appconsume.Precondition, error) {
+			return readPreconditions(ctx, pf, appVerErr, tokenErr), nil
+		}),
+	}
+	req := appconsume.Request{
+		EventKey:  eventKey,
+		Params:    paramMap,
+		JQExpr:    o.jqExpr,
+		OutputDir: outputDir,
+		DryRun:    o.dryRun,
+		MaxEvents: o.maxEvents,
+		Timeout:   o.timeout,
+		IsTTY:     f.IOStreams.IsTerminal,
+	}
+	decision, err := svc.Decide(cmd.Context(), entry, req, appconsume.ExecutionContext{API: runtime})
+	if err != nil {
 		return err
 	}
-	if err := preflightScopes(cmd.Context(), pf); err != nil {
-		return err
+
+	if o.dryRun {
+		return render.WriteDecisionJSON(f.IOStreams.Out, f.IOStreams.ErrOut, string(identity), decision.View())
 	}
 
 	ctx, cancel := context.WithCancel(cmd.Context())
@@ -204,23 +237,26 @@ func runConsume(cmd *cobra.Command, f *cmdutil.Factory, eventKey string, o consu
 		watchStdinEOF(os.Stdin, cancel, errOut)
 	}
 
-	if err := consume.Run(ctx, transport.New(), cfg.AppID, cfg.ProfileName, domain, consume.Options{
-		EventKey:        eventKey,
-		Params:          paramMap,
-		JQExpr:          o.jqExpr,
-		Quiet:           o.quiet,
-		OutputDir:       outputDir,
-		Runtime:         runtime,
-		Out:             f.IOStreams.Out,
-		ErrOut:          errOut,
-		RemoteAPIClient: botRuntime,
-		MaxEvents:       o.maxEvents,
-		Timeout:         o.timeout,
-		IsTTY:           f.IOStreams.IsTerminal,
-	}); err != nil {
-		return err
-	}
-	return nil
+	runner := streamRunnerFunc(func(ctx context.Context, prepare appconsume.PrepareFunc) error {
+		return consume.Run(ctx, transport.New(), cfg.AppID, cfg.ProfileName, domain, consume.Options{
+			EventKey:         eventKey,
+			Def:              keyDef,
+			Params:           decision.NormalizedParams(),
+			ParamsNormalized: true,
+			JQExpr:           o.jqExpr,
+			Quiet:            o.quiet,
+			OutputDir:        outputDir,
+			Runtime:          runtime,
+			Out:              f.IOStreams.Out,
+			ErrOut:           errOut,
+			RemoteAPIClient:  botRuntime,
+			MaxEvents:        o.maxEvents,
+			Timeout:          o.timeout,
+			IsTTY:            f.IOStreams.IsTerminal,
+			Prepare:          prepare,
+		})
+	})
+	return svc.Execute(ctx, entry, decision, runner, appconsume.ExecutionContext{API: runtime})
 }
 
 // resolveIdentity resolves the session identity and enforces keyDef.AuthTypes as a whitelist.
@@ -248,10 +284,14 @@ type preflightCtx struct {
 	subscribedCallbacks []string
 }
 
-// preflightScopes compares required scopes against session-available scopes (user: UAT stored; bot: appVer.TenantScopes).
-func preflightScopes(ctx context.Context, pf *preflightCtx) error {
+// preflightScopes compares required scopes against session-available scopes
+// (user: UAT stored; bot: appVer.TenantScopes). checked reports whether a
+// comparison actually happened: "the ledger was unavailable" and "the check
+// passed" are different answers, and only the caller can decide how loudly to
+// say the first one.
+func preflightScopes(ctx context.Context, pf *preflightCtx) (checked bool, err error) {
 	if len(pf.keyDef.Scopes) == 0 || pf.identity == "" {
-		return nil
+		return true, nil
 	}
 	if ctx == nil {
 		ctx = context.Background()
@@ -261,24 +301,24 @@ func preflightScopes(ctx context.Context, pf *preflightCtx) error {
 	switch {
 	case pf.identity.IsBot():
 		if pf.appVer == nil {
-			return nil
+			return false, nil
 		}
 		storedScopes = strings.Join(pf.appVer.TenantScopes, " ")
 	case pf.identity == core.AsUser:
 		result, err := pf.factory.Credential.ResolveToken(ctx, credential.NewTokenSpec(pf.identity, pf.appID))
 		if err != nil || result == nil || result.Scopes == "" {
-			return nil //nolint:nilerr // best-effort: bus handshake will surface real auth error
+			return false, nil //nolint:nilerr // best-effort: the bus handshake surfaces the real auth error
 		}
 		storedScopes = result.Scopes
 	default:
-		return nil
+		return false, nil
 	}
 
 	missing := auth.MissingScopes(storedScopes, pf.keyDef.Scopes)
 	if len(missing) == 0 {
-		return nil
+		return true, nil
 	}
-	return errs.NewPermissionError(errs.SubtypeMissingScope,
+	return true, errs.NewPermissionError(errs.SubtypeMissingScope,
 		"missing required scopes for EventKey %s (as %s): %s",
 		pf.eventKey, pf.identity, strings.Join(missing, ", ")).
 		WithIdentity(string(pf.identity)).

cmd/event/consume_dryrun_test.go

@@ -0,0 +1,75 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/larksuite/cli/internal/cmdutil"
+	"github.com/larksuite/cli/internal/core"
+)
+
+// A dry run in a degraded environment (the test factory has no reachable
+// platform, so every weak read-only check comes back unanswered) still exits
+// zero with a structured decision that honestly says "unknown" — and performs
+// none of its declared write effects.
+func TestDryRun_DegradedEnvironmentStaysHonestAndSideEffectFree(t *testing.T) {
+	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "cli_test"})
+	snap := compileCatalog()
+
+	tmp := t.TempDir()
+	prevWD, _ := os.Getwd()
+	if err := os.Chdir(tmp); err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = os.Chdir(prevWD) })
+
+	cmd := NewCmdConsume(f, snap)
+	cmd.SetArgs([]string{"im.message.receive_v1", "--as", "bot", "--dry-run", "--output-dir", "events-out"})
+	cmd.SilenceUsage = true
+	cmd.SilenceErrors = true
+	if err := cmd.Execute(); err != nil {
+		t.Fatalf("dry-run must not fail on unusable credentials, got: %v", err)
+	}
+
+	var envelope struct {
+		OK     bool `json:"ok"`
+		DryRun bool `json:"dry_run"`
+		Data   struct {
+			Decision struct {
+				Status        string `json:"status"`
+				Preconditions []struct {
+					Name   string `json:"name"`
+					Status string `json:"status"`
+				} `json:"preconditions"`
+				WouldWrite []string `json:"would_write"`
+			} `json:"decision"`
+		} `json:"data"`
+	}
+	if err := json.Unmarshal(stdout.Bytes(), &envelope); err != nil {
+		t.Fatalf("stdout is not a decision envelope: %v\n%s", err, stdout.String())
+	}
+	if !envelope.OK || !envelope.DryRun {
+		t.Errorf("want ok=true dry_run=true, got: %s", stdout.String())
+	}
+	if envelope.Data.Decision.Status != "unknown" {
+		t.Errorf("unanswerable weak checks must render unknown, not fake readiness; got status %q", envelope.Data.Decision.Status)
+	}
+	names := map[string]string{}
+	for _, p := range envelope.Data.Decision.Preconditions {
+		names[p.Name] = p.Status
+	}
+	if names["credentials_available"] == "" || names["console_event_published"] == "" || names["scopes_granted"] == "" {
+		t.Errorf("preconditions must name every check, got: %v", names)
+	}
+
+	// The declared write side effects must stay declarations: the requested
+	// output dir must not exist after a dry run.
+	if _, err := os.Stat(filepath.Join(tmp, "events-out")); !os.IsNotExist(err) {
+		t.Error("dry-run created the output directory; the preview performed a side effect")
+	}
+}

cmd/event/event.go

@@ -18,12 +18,13 @@ func NewCmdEvents(f *cmdutil.Factory) *cobra.Command {
 		SilenceUsage: true,
 	}
 
-	cmd.AddCommand(NewCmdConsume(f))
-	cmd.AddCommand(NewCmdList(f))
-	cmd.AddCommand(NewCmdSchema(f))
+	snap := compileCatalog()
+	cmd.AddCommand(NewCmdConsume(f, snap))
+	cmd.AddCommand(NewCmdList(f, snap))
+	cmd.AddCommand(NewCmdSchema(f, snap))
 	cmd.AddCommand(NewCmdStatus(f))
 	cmd.AddCommand(NewCmdStop(f))
-	cmd.AddCommand(NewCmdBus(f))
+	cmd.AddCommand(NewCmdBus(f, snap))
 
 	return cmd
 }

cmd/event/format_helpers_test.go

@@ -12,7 +12,7 @@ import (
 
 	"github.com/larksuite/cli/internal/cmdutil"
 	"github.com/larksuite/cli/internal/core"
-	"github.com/larksuite/cli/internal/event/protocol"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/protocol"
 	"github.com/larksuite/cli/internal/output"
 )
 
@@ -288,9 +288,10 @@ func errorAs(err error, target interface{}) bool {
 
 func TestNewCmdFactories_WireFlags(t *testing.T) {
 	f, _, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "cli_XXXXXXXXXXXXXXXX"})
+	snap := compileCatalog()
 
 	t.Run("consume", func(t *testing.T) {
-		cmd := NewCmdConsume(f)
+		cmd := NewCmdConsume(f, snap)
 		for _, flag := range []string{"param", "jq", "quiet", "output-dir", "max-events", "timeout", "as"} {
 			if cmd.Flags().Lookup(flag) == nil {
 				t.Errorf("consume missing --%s flag", flag)
@@ -320,14 +321,22 @@ func TestNewCmdFactories_WireFlags(t *testing.T) {
 	})
 
 	t.Run("list", func(t *testing.T) {
-		cmd := NewCmdList(f)
+		cmd := NewCmdList(f, snap)
 		if cmd.Flags().Lookup("json") == nil {
 			t.Error("list missing --json flag")
 		}
+		domainFlag := cmd.Flags().Lookup("domain")
+		if domainFlag == nil {
+			t.Fatal("list missing --domain flag")
+		}
+		wantUsage := "Only list EventKeys of this domain. Valid domains: " + strings.Join(snap.Domains(), ", ")
+		if domainFlag.Usage != wantUsage {
+			t.Errorf("--domain usage = %q, want %q", domainFlag.Usage, wantUsage)
+		}
 	})
 
 	t.Run("bus", func(t *testing.T) {
-		cmd := NewCmdBus(f)
+		cmd := NewCmdBus(f, snap)
 		if !cmd.Hidden {
 			t.Error("bus should be hidden (internal daemon entrypoint)")
 		}

cmd/event/golden_test.go

@@ -0,0 +1,81 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"flag"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/larksuite/cli/internal/cmdutil"
+	"github.com/larksuite/cli/internal/core"
+)
+
+var updateGolden = flag.Bool("update", false, "rewrite golden files instead of comparing")
+
+// goldenSchemaKeys picks one key per rendering path so every branch of the
+// list/schema output stays pinned: a processed key with a flat custom schema,
+// a native key with field overrides, a callback key with a single consumer,
+// and a key with a required parameter plus a pre-consume hook.
+var goldenSchemaKeys = map[string]string{
+	"schema_im_message_receive":  "im.message.receive_v1",
+	"schema_im_chat_updated":     "im.chat.updated_v1",
+	"schema_card_action_trigger": "card.action.trigger",
+	"schema_board_whiteboard":    "board.whiteboard.updated_v1",
+}
+
+// The golden files pin stdout byte-for-byte. The output is deterministic:
+// the snapshot keeps keys sorted, encoding/json sorts object keys, and nothing
+// on the rendering path reads the clock or randomness. Regenerate with:
+//
+//	go test ./cmd/event/ -run TestGolden -update
+func TestGolden_ListOutput(t *testing.T) {
+	snap := compileCatalog()
+	for name, asJSON := range map[string]bool{"list_text": false, "list_json": true} {
+		t.Run(name, func(t *testing.T) {
+			f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
+			if err := runList(f, snap, "", asJSON); err != nil {
+				t.Fatalf("runList: %v", err)
+			}
+			assertGolden(t, name, stdout.String())
+		})
+	}
+}
+
+func TestGolden_SchemaOutput(t *testing.T) {
+	snap := compileCatalog()
+	for name, key := range goldenSchemaKeys {
+		for suffix, asJSON := range map[string]bool{"_text": false, "_json": true} {
+			t.Run(name+suffix, func(t *testing.T) {
+				f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
+				if err := runSchema(f, snap, key, asJSON); err != nil {
+					t.Fatalf("runSchema(%s): %v", key, err)
+				}
+				assertGolden(t, name+suffix, stdout.String())
+			})
+		}
+	}
+}
+
+func assertGolden(t *testing.T, name, got string) {
+	t.Helper()
+	path := filepath.Join("testdata", "golden", name+".golden")
+	if *updateGolden {
+		if err := os.MkdirAll(filepath.Dir(path), 0o755); err != nil {
+			t.Fatal(err)
+		}
+		if err := os.WriteFile(path, []byte(got), 0o644); err != nil {
+			t.Fatal(err)
+		}
+		return
+	}
+	want, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatalf("missing golden %s (regenerate with -update): %v", name, err)
+	}
+	if string(want) != got {
+		t.Errorf("output drifted from golden %s\n--- want\n%s\n--- got\n%s", name, want, got)
+	}
+}

cmd/event/list.go

@@ -10,31 +10,44 @@ import (
 
 	"github.com/spf13/cobra"
 
+	"github.com/larksuite/cli/errs"
 	"github.com/larksuite/cli/internal/cmdutil"
 	eventlib "github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
 	"github.com/larksuite/cli/internal/output"
 )
 
-func NewCmdList(f *cmdutil.Factory) *cobra.Command {
+func NewCmdList(f *cmdutil.Factory, snap *catalog.Snapshot) *cobra.Command {
 	var asJSON bool
+	var domain string
 	cmd := &cobra.Command{
 		Use:   "list",
 		Short: "List all available EventKeys",
-		Long:  "Show all registered EventKeys grouped by domain (first segment of the key). Use --json for machine-readable output.",
+		Long:  "Show all registered EventKeys grouped by domain (first segment of the key). Use --domain to keep one domain only, --json for machine-readable output.",
 		RunE: func(cmd *cobra.Command, args []string) error {
-			return runList(f, asJSON)
+			return runList(f, snap, domain, asJSON)
 		},
 	}
 	cmd.Flags().BoolVar(&asJSON, "json", false, "Emit the full EventKey list as JSON (for AI / scripts)")
+	cmd.Flags().StringVar(&domain, "domain", "", fmt.Sprintf(
+		"Only list EventKeys of this domain. Valid domains: %s",
+		strings.Join(snap.Domains(), ", "),
+	))
 	cmdutil.SetRisk(cmd, "read")
 	return cmd
 }
 
-func runList(f *cmdutil.Factory, asJSON bool) error {
-	all := eventlib.ListAll()
-
+func runList(f *cmdutil.Factory, snap *catalog.Snapshot, domain string, asJSON bool) error {
+	entries, err := entriesForDomain(snap, domain)
+	if err != nil {
+		return err
+	}
 	if asJSON {
-		return writeListJSON(f, all)
+		return writeListJSON(f, entries)
+	}
+	all := make([]*eventlib.KeyDefinition, 0, len(entries))
+	for _, entry := range entries {
+		all = append(all, entry.Definition())
 	}
 
 	if len(all) == 0 {
@@ -104,18 +117,43 @@ func runList(f *cmdutil.Factory, asJSON bool) error {
 	return nil
 }
 
-func writeListJSON(f *cmdutil.Factory, all []*eventlib.KeyDefinition) error {
-	type row struct {
-		*eventlib.KeyDefinition
-		ResolvedSchema json.RawMessage `json:"resolved_output_schema,omitempty"`
+// listRow is the JSON shape of one `event list --json` row. It is a named
+// type (not a function-local literal) so the render contract test can walk
+// its fields and reject accidental additions to the public output.
+type listRow struct {
+	*eventlib.KeyDefinition
+	ResolvedSchema json.RawMessage `json:"resolved_output_schema,omitempty"`
+}
+
+// entriesForDomain filters at the snapshot query layer: without a domain the
+// full catalog comes back untouched; with one, rows are only removed, never
+// reshaped. An unknown domain is rejected with the valid set spelled out.
+func entriesForDomain(snap *catalog.Snapshot, domain string) ([]*catalog.Entry, error) {
+	if domain == "" {
+		return snap.Entries(), nil
 	}
-	rows := make([]row, len(all))
-	for i, def := range all {
-		resolved, _, err := resolveSchemaJSON(def)
-		if err != nil {
-			return err
+	var filtered []*catalog.Entry
+	for _, entry := range snap.Entries() {
+		if entry.Descriptor().Domain == domain {
+			filtered = append(filtered, entry)
+		}
+	}
+	if len(filtered) == 0 {
+		return nil, errs.NewValidationError(errs.SubtypeInvalidArgument,
+			"unknown domain: %s", domain).
+			WithParam("--domain").
+			WithHint("valid domains: %s", strings.Join(snap.Domains(), ", "))
+	}
+	return filtered, nil
+}
+
+func writeListJSON(f *cmdutil.Factory, entries []*catalog.Entry) error {
+	rows := make([]listRow, len(entries))
+	for i, entry := range entries {
+		rows[i] = listRow{
+			KeyDefinition:  entry.Definition(),
+			ResolvedSchema: entry.Output().SchemaJSON,
 		}
-		rows[i] = row{KeyDefinition: def, ResolvedSchema: resolved}
 	}
 	output.PrintJson(f.IOStreams.Out, rows)
 	return nil

cmd/event/list_domain_test.go

@@ -0,0 +1,96 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"encoding/json"
+	"strings"
+	"testing"
+
+	"github.com/larksuite/cli/errs"
+	"github.com/larksuite/cli/internal/cmdutil"
+	"github.com/larksuite/cli/internal/core"
+)
+
+// Filtering only removes rows: the vc selection must be exactly the catalog's
+// vc keys, and every remaining row keeps the unfiltered field set.
+func TestListDomain_FilterKeepsExactlyTheRequestedDomain(t *testing.T) {
+	snap := compileCatalog()
+
+	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
+	if err := runList(f, snap, "vc", true); err != nil {
+		t.Fatal(err)
+	}
+	var rows []map[string]json.RawMessage
+	if err := json.Unmarshal(stdout.Bytes(), &rows); err != nil {
+		t.Fatal(err)
+	}
+
+	want := map[string]bool{}
+	for _, key := range snap.Keys() {
+		if strings.HasPrefix(key, "vc.") {
+			want[key] = true
+		}
+	}
+	if len(want) == 0 {
+		t.Fatal("the catalog has no vc keys; the filter test proves nothing")
+	}
+	got := map[string]bool{}
+	for _, row := range rows {
+		var key string
+		_ = json.Unmarshal(row["key"], &key)
+		got[key] = true
+		for _, field := range []string{"event_type", "schema", "resolved_output_schema"} {
+			if _, ok := row[field]; !ok {
+				t.Errorf("%s: filtering must not reshape rows; %q is missing", key, field)
+			}
+		}
+	}
+	if len(got) != len(want) {
+		t.Fatalf("filtered rows = %v, want the exact vc set %v", got, want)
+	}
+	for key := range want {
+		if !got[key] {
+			t.Errorf("vc key missing from the filtered list: %s", key)
+		}
+	}
+}
+
+func TestListDomain_TextFilter(t *testing.T) {
+	snap := compileCatalog()
+	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
+	if err := runList(f, snap, "im", false); err != nil {
+		t.Fatal(err)
+	}
+	out := stdout.String()
+	if !strings.Contains(out, "im.message.receive_v1") {
+		t.Error("im keys must be listed")
+	}
+	for _, foreign := range []string{"vc.", "minutes.", "board.", "approval."} {
+		if strings.Contains(out, foreign) {
+			t.Errorf("foreign domain %q leaked into the filtered text output", foreign)
+		}
+	}
+}
+
+func TestListDomain_UnknownDomainIsRejectedWithTheValidSet(t *testing.T) {
+	snap := compileCatalog()
+	f, _, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
+	err := runList(f, snap, "definitely-bogus", true)
+	if err == nil {
+		t.Fatal("an unknown domain must be rejected")
+	}
+	problem, ok := errs.ProblemOf(err)
+	if !ok || problem.Subtype != errs.SubtypeInvalidArgument {
+		t.Fatalf("want invalid_argument, got %v", err)
+	}
+	if !strings.Contains(err.Error(), "unknown domain: definitely-bogus") {
+		t.Errorf("error must name the rejected value, got %v", err)
+	}
+	for _, domain := range []string{"application", "approval", "board", "card", "im", "minutes", "task", "vc"} {
+		if !strings.Contains(problem.Hint, domain) {
+			t.Errorf("hint must list valid domain %q, got %q", domain, problem.Hint)
+		}
+	}
+}

cmd/event/list_test.go

@@ -10,20 +10,18 @@ import (
 
 	"github.com/larksuite/cli/internal/cmdutil"
 	"github.com/larksuite/cli/internal/core"
-	eventlib "github.com/larksuite/cli/internal/event"
-
-	_ "github.com/larksuite/cli/events"
 )
 
 func TestEventLookup_VCMeetingLifecycleKeys(t *testing.T) {
+	snap := compileCatalog()
 	for _, key := range []string{
 		"approval.instance.status_changed_v4",
 		"approval.task.status_changed_v4",
 		"vc.meeting.participant_meeting_started_v1",
 		"vc.meeting.participant_meeting_joined_v1",
 	} {
-		if _, ok := eventlib.Lookup(key); !ok {
-			t.Fatalf("event.Lookup(%q) should succeed", key)
+		if _, ok := snap.Resolve(key); !ok {
+			t.Fatalf("snap.Resolve(%q) should succeed", key)
 		}
 	}
 }
@@ -31,7 +29,7 @@ func TestEventLookup_VCMeetingLifecycleKeys(t *testing.T) {
 func TestRunList_TextOutput(t *testing.T) {
 	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-	if err := runList(f, false); err != nil {
+	if err := runList(f, compileCatalog(), "", false); err != nil {
 		t.Fatalf("runList: %v", err)
 	}
 
@@ -55,7 +53,7 @@ func TestRunList_TextOutput(t *testing.T) {
 func TestRunList_JSONOutput(t *testing.T) {
 	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-	if err := runList(f, true); err != nil {
+	if err := runList(f, compileCatalog(), "", true); err != nil {
 		t.Fatalf("runList json: %v", err)
 	}
 

cmd/event/preconditions_test.go

@@ -0,0 +1,65 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"context"
+	"errors"
+	"testing"
+
+	"github.com/larksuite/cli/internal/core"
+	eventlib "github.com/larksuite/cli/internal/event"
+	appconsume "github.com/larksuite/cli/internal/event/application/consume"
+)
+
+func preconditionByName(list []appconsume.Precondition, name string) *appconsume.Precondition {
+	for i := range list {
+		if list[i].Name == name {
+			return &list[i]
+		}
+	}
+	return nil
+}
+
+// An unusable credential blocks the decision and carries the exact error a
+// real run would have returned, so both paths refuse for the same reason.
+func TestReadPreconditions_TokenErrorBlocksWithTheSameError(t *testing.T) {
+	tokenErr := errors.New("no tenant token available")
+	pf := &preflightCtx{
+		appID:    "cli_test",
+		identity: core.AsBot,
+		keyDef:   &eventlib.KeyDefinition{Key: "demo.thing.updated_v1"},
+	}
+	got := readPreconditions(context.Background(), pf, nil, tokenErr)
+
+	cred := preconditionByName(got, "credentials_available")
+	if cred == nil {
+		t.Fatal("credentials_available precondition missing")
+	}
+	if cred.Status != appconsume.PreconditionBlocked || !errors.Is(cred.BlockErr, tokenErr) {
+		t.Errorf("token failure must block with the original error, got %+v", cred)
+	}
+}
+
+// A scope ledger nobody could read is reported as unknown — never as ok.
+func TestReadPreconditions_UnreadableScopesAreUnknown(t *testing.T) {
+	pf := &preflightCtx{
+		appID:    "cli_test",
+		identity: core.AsBot,
+		keyDef: &eventlib.KeyDefinition{
+			Key:    "demo.thing.updated_v1",
+			Scopes: []string{"demo:read"},
+		},
+		appVer: nil, // no published version: the bot scope ledger is unreadable
+	}
+	got := readPreconditions(context.Background(), pf, nil, nil)
+
+	scopes := preconditionByName(got, "scopes_granted")
+	if scopes == nil {
+		t.Fatal("scopes_granted precondition missing")
+	}
+	if scopes.Status != appconsume.PreconditionUnknown {
+		t.Errorf("an unreadable ledger must report unknown, got %q", scopes.Status)
+	}
+}

cmd/event/preflight_test.go

@@ -108,7 +108,7 @@ func TestPreflightScopes_Bot_NoAppVer_SkipsCheck(t *testing.T) {
 		Key:    "im.message.text",
 		Scopes: []string{"im:message", "im:message.group_at_msg"},
 	}
-	err := preflightScopes(nil, newPreflightCtx("cli_x", "feishu", core.AsBot, def, nil))
+	_, err := preflightScopes(nil, newPreflightCtx("cli_x", "feishu", core.AsBot, def, nil))
 	if err != nil {
 		t.Fatalf("bot + nil appVer should skip, got: %v", err)
 	}
@@ -124,7 +124,7 @@ func TestPreflightScopes_Bot_AllGranted_Passes(t *testing.T) {
 		"im:message.group_at_msg",
 		"contact:user:readonly",
 	}}
-	err := preflightScopes(nil, newPreflightCtx("cli_x", "feishu", core.AsBot, def, appVer))
+	_, err := preflightScopes(nil, newPreflightCtx("cli_x", "feishu", core.AsBot, def, appVer))
 	if err != nil {
 		t.Fatalf("all scopes granted, unexpected error: %v", err)
 	}
@@ -136,7 +136,7 @@ func TestPreflightScopes_Bot_MissingBlocks(t *testing.T) {
 		Scopes: []string{"im:message", "im:message.group_at_msg"},
 	}
 	appVer := &appmeta.AppVersion{TenantScopes: []string{"im:message"}}
-	err := preflightScopes(nil, newPreflightCtx("cli_x", "feishu", core.AsBot, def, appVer))
+	_, err := preflightScopes(nil, newPreflightCtx("cli_x", "feishu", core.AsBot, def, appVer))
 	if err == nil {
 		t.Fatal("expected error for missing scope")
 	}
@@ -169,7 +169,7 @@ func TestPreflightScopes_Bot_MissingBlocks(t *testing.T) {
 
 func TestPreflightScopes_NoRequiredScopes_SkipsCheck(t *testing.T) {
 	def := &eventlib.KeyDefinition{Key: "x"}
-	if err := preflightScopes(nil, newPreflightCtx("cli_x", "feishu", core.AsBot, def, nil)); err != nil {
+	if _, err := preflightScopes(nil, newPreflightCtx("cli_x", "feishu", core.AsBot, def, nil)); err != nil {
 		t.Fatalf("no required scopes means nothing to verify, got: %v", err)
 	}
 }

cmd/event/render/decision.go

@@ -0,0 +1,96 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+// Package render turns consume decisions into user-facing output. It is the
+// only place a decision becomes JSON; the application layer never formats
+// anything itself.
+package render
+
+import (
+	"io"
+	"regexp"
+
+	appconsume "github.com/larksuite/cli/internal/event/application/consume"
+	"github.com/larksuite/cli/internal/output"
+)
+
+// sensitiveParamName matches parameter names whose values must never be
+// echoed back in a rendered decision. Names are matched, not values: a
+// credential-bearing parameter is identifiable by its declaration, and
+// guessing at value shapes would miss more than it catches.
+var sensitiveParamName = regexp.MustCompile(`(?i)(token|secret|password|credential|cookie)`)
+
+func redactParams(params map[string]string) map[string]string {
+	out := make(map[string]string, len(params))
+	for name, value := range params {
+		if sensitiveParamName.MatchString(name) {
+			out[name] = "[redacted]"
+			continue
+		}
+		out[name] = value
+	}
+	return out
+}
+
+// decisionPayload is the JSON shape under data.decision — snake_case, stable,
+// documented in the event skill. Field additions must be additive.
+type decisionPayload struct {
+	EventKey      string             `json:"event_key"`
+	Domain        string             `json:"domain"`
+	Identity      string             `json:"identity"`
+	Status        string             `json:"status"`
+	Params        map[string]string  `json:"params"`
+	Scope         string             `json:"scope"`
+	Preconditions []preconditionView `json:"preconditions"`
+	Preparation   *preparationView   `json:"preparation,omitempty"`
+	WouldRead     []string           `json:"would_read"`
+	WouldWrite    []string           `json:"would_write"`
+}
+
+type preconditionView struct {
+	Name   string `json:"name"`
+	Status string `json:"status"`
+	Detail string `json:"detail,omitempty"`
+}
+
+type preparationView struct {
+	Strategy  string `json:"strategy"`
+	Condition string `json:"condition"`
+	Action    string `json:"action"`
+}
+
+// WriteDecisionJSON emits the decision inside the standard success envelope
+// with the envelope's own top-level dry_run marker set.
+func WriteDecisionJSON(out, errOut io.Writer, identity string, v appconsume.DecisionView) error {
+	return output.WriteSuccessEnvelope(map[string]any{
+		"decision": toPayload(v),
+	}, output.SuccessEnvelopeOptions{
+		CommandPath: "event consume",
+		Identity:    identity,
+		DryRun:      true,
+		Out:         out,
+		ErrOut:      errOut,
+	})
+}
+
+func toPayload(v appconsume.DecisionView) decisionPayload {
+	p := decisionPayload{
+		EventKey:   v.EventKey,
+		Domain:     v.Domain,
+		Identity:   v.Identity,
+		Status:     v.Status,
+		Params:     redactParams(v.Params),
+		Scope:      v.Scope,
+		WouldRead:  v.WouldRead,
+		WouldWrite: v.WouldWrite,
+	}
+	p.Preconditions = make([]preconditionView, 0, len(v.Preconditions))
+	for _, pc := range v.Preconditions {
+		p.Preconditions = append(p.Preconditions, preconditionView(pc))
+	}
+	if v.Preparation != nil {
+		pv := preparationView(*v.Preparation)
+		p.Preparation = &pv
+	}
+	return p
+}

cmd/event/render/decision_test.go

@@ -0,0 +1,114 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package render
+
+import (
+	"bytes"
+	"encoding/json"
+	"strings"
+	"testing"
+
+	appconsume "github.com/larksuite/cli/internal/event/application/consume"
+)
+
+func sampleView() appconsume.DecisionView {
+	return appconsume.DecisionView{
+		EventKey: "vc.note.generated_v1",
+		Domain:   "vc",
+		Identity: "user",
+		Status:   "ready",
+		Params:   map[string]string{"whiteboard_id": "wb-1", "access_token": "sk-SENSITIVE-VALUE"},
+		Scope:    "vc.note.generated_v1",
+		Preconditions: []appconsume.PreconditionView{
+			{Name: "console_event_published", Status: "ok"},
+			{Name: "scopes_granted", Status: "ok"},
+		},
+		Preparation: &appconsume.PreparationView{
+			Strategy: "legacy_preconsume", Condition: "first_consumer_for_scope", Action: "register_event_delivery",
+		},
+		WouldRead:  []string{"local_bus_probe", "app_metadata_preflight"},
+		WouldWrite: []string{"start_or_reuse_local_bus", "register_consumer", "run_preparation_when_first", "open_event_stream"},
+	}
+}
+
+// The JSON contract: dry_run is the envelope's own top-level marker (never a
+// data field), and the decision sits under data.decision with its documented
+// members.
+func TestWriteDecisionJSON_EnvelopeContract(t *testing.T) {
+	var out, errOut bytes.Buffer
+	if err := WriteDecisionJSON(&out, &errOut, "user", sampleView()); err != nil {
+		t.Fatal(err)
+	}
+
+	var envelope map[string]json.RawMessage
+	if err := json.Unmarshal(out.Bytes(), &envelope); err != nil {
+		t.Fatalf("stdout is not a JSON envelope: %v\n%s", err, out.String())
+	}
+	if string(envelope["ok"]) != "true" || string(envelope["dry_run"]) != "true" {
+		t.Errorf("envelope must carry top-level ok=true and dry_run=true, got %s", out.String())
+	}
+	if _, misplaced := envelope["decision"]; misplaced {
+		t.Error("decision must live under data, not at the envelope top level")
+	}
+
+	var data struct {
+		Decision struct {
+			EventKey    string            `json:"event_key"`
+			Domain      string            `json:"domain"`
+			Identity    string            `json:"identity"`
+			Status      string            `json:"status"`
+			Params      map[string]string `json:"params"`
+			Scope       string            `json:"scope"`
+			Preparation *struct {
+				Strategy  string `json:"strategy"`
+				Condition string `json:"condition"`
+				Action    string `json:"action"`
+			} `json:"preparation"`
+			WouldRead  []string `json:"would_read"`
+			WouldWrite []string `json:"would_write"`
+			DryRun     *bool    `json:"dry_run"`
+		} `json:"decision"`
+	}
+	if err := json.Unmarshal(envelope["data"], &data); err != nil {
+		t.Fatalf("data.decision does not match the documented shape: %v", err)
+	}
+	d := data.Decision
+	if d.EventKey != "vc.note.generated_v1" || d.Domain != "vc" || d.Identity != "user" || d.Status != "ready" {
+		t.Errorf("identity facts drifted: %+v", d)
+	}
+	if d.Preparation == nil || d.Preparation.Condition != "first_consumer_for_scope" {
+		t.Errorf("conditional preparation must be stated: %+v", d.Preparation)
+	}
+	if len(d.WouldRead) == 0 || len(d.WouldWrite) == 0 {
+		t.Error("would_read / would_write must be present")
+	}
+	if d.DryRun != nil {
+		t.Error("dry_run inside data.decision would duplicate the envelope marker")
+	}
+}
+
+// Sensitive parameter values never reach the rendered output. The control
+// assertion first proves the sentinel would be visible if leaked.
+func TestWriteDecision_RedactsSensitiveParams(t *testing.T) {
+	const sentinel = "sk-SENSITIVE-VALUE"
+	view := sampleView()
+	if !strings.Contains(view.Params["access_token"], sentinel) {
+		t.Fatal("control failed: the sentinel is not in the input, the test cannot prove redaction")
+	}
+
+	var jsonOut, jsonErr bytes.Buffer
+	if err := WriteDecisionJSON(&jsonOut, &jsonErr, "user", view); err != nil {
+		t.Fatal(err)
+	}
+	if strings.Contains(jsonOut.String(), sentinel) {
+		t.Errorf("JSON output leaks a sensitive param value: %s", jsonOut.String())
+	}
+	compact := strings.ReplaceAll(strings.ReplaceAll(jsonOut.String(), "\n", ""), " ", "")
+	if !strings.Contains(compact, `"access_token":"[redacted]"`) {
+		t.Errorf("sensitive param must render as redacted, got: %s", jsonOut.String())
+	}
+	if !strings.Contains(compact, `"whiteboard_id":"wb-1"`) {
+		t.Errorf("non-sensitive params must render verbatim, got: %s", jsonOut.String())
+	}
+}

cmd/event/render/redaction_guard_test.go

@@ -0,0 +1,121 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package render
+
+import (
+	"strings"
+	"testing"
+
+	"github.com/larksuite/cli/events"
+	"github.com/larksuite/cli/internal/event/catalog"
+)
+
+// This file is a guard, not a contract: it does not pin what the redaction
+// regex matches, it hunts for declared parameter names that smell like
+// credentials yet would render verbatim. The detector wordlist is therefore
+// deliberately wider than the production sensitiveParamName pattern — a hit
+// here means either the parameter should be renamed or the production
+// pattern must grow, decided by a human, never by loosening this list.
+
+// credentialWords are matched against whole '_'/'-'/'.'-separated segments of
+// a parameter name, so chat_key or tokenizer_mode cannot trip them. The bare
+// word "key" is intentionally absent (identifier names like whiteboard_id or
+// a hypothetical chat_key are not credentials); the api/key pairing is what
+// carries credential semantics and is detected as a pair below.
+var credentialWords = map[string]bool{
+	"token":       true,
+	"secret":      true,
+	"password":    true,
+	"credential":  true,
+	"credentials": true,
+	"cookie":      true,
+	"auth":        true,
+	"signature":   true,
+	"bearer":      true,
+	"apikey":      true,
+}
+
+// smellsLikeCredential reports whether a parameter name carries credential
+// semantics per the guard wordlist: any single segment in credentialWords,
+// or the adjacent segment pair api+key.
+func smellsLikeCredential(name string) bool {
+	segments := strings.FieldsFunc(strings.ToLower(name), func(r rune) bool {
+		return r == '_' || r == '-' || r == '.'
+	})
+	for i, seg := range segments {
+		if credentialWords[seg] {
+			return true
+		}
+		if seg == "api" && i+1 < len(segments) && segments[i+1] == "key" {
+			return true
+		}
+	}
+	return false
+}
+
+// unredactedCredentialParams returns the names that smell like credentials
+// but are NOT matched by the production redaction pattern — every such name
+// would render its value verbatim in a dry-run decision.
+func unredactedCredentialParams(names []string) []string {
+	var findings []string
+	for _, name := range names {
+		if smellsLikeCredential(name) && !sensitiveParamName.MatchString(name) {
+			findings = append(findings, name)
+		}
+	}
+	return findings
+}
+
+// The detector itself must bite before the live scan means anything: known
+// credential-shaped names that the production pattern misses must be caught,
+// and ordinary identifier names must pass.
+func TestRedactionGuardDetector_SelfCheck(t *testing.T) {
+	// Credential-shaped and covered by the production pattern: no finding.
+	for _, name := range []string{"access_token", "client_secret", "user_password", "session_cookie", "sso_credential"} {
+		if got := unredactedCredentialParams([]string{name}); len(got) != 0 {
+			t.Errorf("%q is redacted by the production pattern, the guard must not flag it, got %v", name, got)
+		}
+	}
+	// Credential-shaped but NOT covered by the production pattern today: the
+	// guard must flag these, otherwise it can never catch a real gap.
+	for _, name := range []string{"api_key", "auth_code", "request_signature", "bearer_value"} {
+		if got := unredactedCredentialParams([]string{name}); len(got) != 1 {
+			t.Errorf("%q smells like a credential and is not redacted; the guard must flag it, got %v", name, got)
+		}
+	}
+	// Ordinary identifiers, including the wide-false-positive shapes the
+	// wordlist is segment-matched to avoid: no finding.
+	for _, name := range []string{"whiteboard_id", "chat_key", "tokenizer_mode", "author", "meeting_no"} {
+		if got := unredactedCredentialParams([]string{name}); len(got) != 0 {
+			t.Errorf("%q is an ordinary identifier, the guard must not flag it, got %v", name, got)
+		}
+	}
+}
+
+// Every declared parameter of every compiled EventKey either carries no
+// credential semantics or is caught by the production redaction pattern.
+func TestRedactionGuard_CatalogParamsHaveNoUnredactedCredentials(t *testing.T) {
+	snap, err := catalog.Compile(events.All(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("compile catalog: %v", err)
+	}
+
+	var names []string
+	for _, entry := range snap.Entries() {
+		desc := entry.Descriptor()
+		for _, p := range desc.Params {
+			names = append(names, desc.Key+": "+p.Name)
+			if findings := unredactedCredentialParams([]string{p.Name}); len(findings) != 0 {
+				t.Errorf("EventKey %s declares param %q which smells like a credential but is not matched by the redaction pattern; rename the param or extend sensitiveParamName deliberately", desc.Key, p.Name)
+			}
+		}
+	}
+	// A scan that visited no parameters proves nothing.
+	if len(names) == 0 {
+		t.Fatal("the compiled catalog declares no parameters at all; the guard scanned nothing")
+	}
+}

cmd/event/render_contract_test.go

@@ -0,0 +1,163 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"reflect"
+	"strings"
+	"testing"
+)
+
+// renderedDeclarationFields lists every JSON field the list/schema commands
+// are allowed to expose, each with the reason it belongs to the public
+// contract. Golden files pin today's bytes; this gate protects tomorrow: a
+// field added to the rendered structs (promoted through the embedded
+// definition or nested anywhere under it) must either appear here
+// deliberately or be tagged `json:"-"`. The set is flat — an entry admits its
+// rendered name at any nesting level, which is the same latitude
+// encoding/json gives a name.
+var renderedDeclarationFields = map[string]string{
+	"key":                     "stable identifier agents subscribe by",
+	"domain":                  "declared domain override; empty for every shipped key (filtering reads the derived descriptor value), so legacy output is byte-identical",
+	"display_name":            "human-readable name for pickers",
+	"description":             "what the event means (KeyDefinition) / what the parameter does (ParamDef)",
+	"event_type":              "upstream event type behind this key",
+	"subscription_type":       "which console ledger the precheck reads",
+	"params":                  "declared consume parameters",
+	"schema":                  "declared schema source (native/custom markers)",
+	"scopes":                  "OAuth scopes required to consume",
+	"auth_types":              "identities the key accepts",
+	"required_console_events": "console switches that must be enabled",
+	"buffer_size":             "delivery buffer size after normalization",
+	"workers":                 "worker count after normalization",
+	"single_consumer":         "whether a second consumer is rejected",
+	"resolved_output_schema":  "fully resolved JSON schema of stdout events",
+	"jq_root_path":            "schema command only: jq root for consuming stdout",
+
+	// Nested under params (ParamDef): everything an agent needs to pass the
+	// parameter correctly.
+	"name":             "parameter name as passed via --param",
+	"type":             "parameter value type (string/enum/multi/bool/int)",
+	"required":         "whether the parameter must be provided",
+	"default":          "value applied when the parameter is omitted",
+	"values":           "allowed values for enum/multi parameters",
+	"subscription_key": "whether the parameter is part of the subscription identity",
+
+	// Nested under params.values (ParamValue).
+	"value": "one allowed parameter value",
+	"desc":  "what choosing this value means",
+
+	// Nested under schema (SchemaDef / SchemaSpec): declaration markers only;
+	// the resolved schema is the sibling resolved_output_schema.
+	"native":          "marker for keys delivering the raw V2 envelope",
+	"custom":          "marker for keys delivering processed output",
+	"field_overrides": "per-field annotations overriding the reflected schema",
+	"raw":             "raw declared schema bytes; empty for reflected types",
+
+	// Nested under schema.field_overrides (schemas.FieldMeta). The type has
+	// no json tags, so encoding/json renders the Go field names — pinned
+	// as-is because retagging them would change the public bytes.
+	"Description": "override for the field's schema description",
+	"Enum":        "override for the field's allowed values",
+	"Kind":        "override rendered as the field's schema format",
+}
+
+// TestRenderContract_NoRuntimeFieldLeaksIntoJSON walks both rendered shapes,
+// following embedded struct promotion and recursing into every named type
+// reachable through the rendered fields, and fails on any exported member
+// that is neither allowlisted nor explicitly excluded from JSON.
+func TestRenderContract_NoRuntimeFieldLeaksIntoJSON(t *testing.T) {
+	emitted := map[string]bool{}
+	for _, typ := range []reflect.Type{
+		reflect.TypeFor[listRow](),
+		reflect.TypeFor[schemaPayload](),
+	} {
+		walkRenderedFields(t, typ, emitted, map[reflect.Type]bool{})
+	}
+
+	if len(emitted) == 0 {
+		t.Fatal("no rendered fields were visited; the gate scanned nothing")
+	}
+	// The embedded definition is where leaks would hide: prove promotion was
+	// actually followed by requiring fields that only exist on it. The nested
+	// sentinels prove each recursion path is really taken: subscription_key
+	// (slice-of-struct: ParamDef), desc (slice inside a nested struct:
+	// ParamValue), raw (pointer-to-struct: SchemaSpec), Enum (map value:
+	// FieldMeta, rendered under its Go name because the type is untagged).
+	for _, sentinel := range []string{
+		"key", "event_type", "resolved_output_schema",
+		"subscription_key", "desc", "raw", "Enum",
+	} {
+		if !emitted[sentinel] {
+			t.Fatalf("field %q was not visited; the walker no longer reaches every rendered shape", sentinel)
+		}
+	}
+	for name := range renderedDeclarationFields {
+		if !emitted[name] {
+			t.Errorf("allowlist entry %q is stale: no rendered struct emits it", name)
+		}
+	}
+}
+
+// walkRenderedFields records every JSON field name typ can render: embedded
+// structs promote into the parent object, and any struct reachable through a
+// field's type — behind pointers, slice/array elements, or map values — is
+// walked in turn, so a field added to a nested type like ParamDef cannot
+// escape the gate. visited breaks cycles; a type already recorded in this
+// walk contributes nothing new.
+func walkRenderedFields(t *testing.T, typ reflect.Type, emitted map[string]bool, visited map[reflect.Type]bool) {
+	t.Helper()
+	typ = nestedStructType(typ)
+	if typ == nil || visited[typ] {
+		return
+	}
+	visited[typ] = true
+	for i := 0; i < typ.NumField(); i++ {
+		field := typ.Field(i)
+		if !field.IsExported() {
+			continue
+		}
+		tag := field.Tag.Get("json")
+		if tag == "-" {
+			continue
+		}
+		if field.Anonymous && tag == "" {
+			if ft := nestedStructType(field.Type); ft != nil {
+				// Embedded struct without a tag: fields promote into the
+				// parent JSON object.
+				walkRenderedFields(t, ft, emitted, visited)
+				continue
+			}
+		}
+		name, _, _ := strings.Cut(tag, ",")
+		if name == "" {
+			// encoding/json renders an untagged exported field under its Go
+			// name (schemas.FieldMeta does this today); the rendered name is
+			// what the contract governs, so it is what must be declared.
+			name = field.Name
+		}
+		if _, ok := renderedDeclarationFields[name]; !ok {
+			t.Errorf("%s.%s renders JSON field %q that is not in the declared output contract; add it deliberately or exclude it with json:\"-\"", typ.Name(), field.Name, name)
+		}
+		emitted[name] = true
+		walkRenderedFields(t, field.Type, emitted, visited)
+	}
+}
+
+// nestedStructType unwraps pointers, slice/array elements, and map values
+// until it reaches the struct that would render as a JSON object; nil means
+// the type renders as a leaf (scalar, string, raw bytes) and holds no fields
+// to govern.
+func nestedStructType(typ reflect.Type) reflect.Type {
+	for {
+		switch typ.Kind() {
+		case reflect.Pointer, reflect.Slice, reflect.Array, reflect.Map:
+			typ = typ.Elem()
+		case reflect.Struct:
+			return typ
+		default:
+			return nil
+		}
+	}
+}

cmd/event/schema.go

@@ -11,75 +11,13 @@ import (
 
 	"github.com/spf13/cobra"
 
-	"github.com/larksuite/cli/errs"
 	"github.com/larksuite/cli/internal/cmdutil"
 	eventlib "github.com/larksuite/cli/internal/event"
-	"github.com/larksuite/cli/internal/event/schemas"
+	"github.com/larksuite/cli/internal/event/catalog"
 	"github.com/larksuite/cli/internal/output"
 )
 
-// resolveSchemaJSON returns the final JSON Schema for an EventKey (reflected base, V2-wrapped for Native, overlay applied); orphans lists unresolved FieldOverrides pointers.
-func resolveSchemaJSON(def *eventlib.KeyDefinition) (json.RawMessage, []string, error) {
-	spec, isNative := pickSpec(def.Schema)
-	if spec == nil {
-		return nil, nil, nil
-	}
-
-	base, err := renderSpec(spec)
-	if err != nil {
-		return nil, nil, err
-	}
-	if base == nil {
-		return nil, nil, nil
-	}
-
-	if isNative {
-		base = schemas.WrapV2Envelope(base)
-	}
-
-	if len(def.Schema.FieldOverrides) > 0 {
-		var parsed map[string]interface{}
-		if err := json.Unmarshal(base, &parsed); err != nil {
-			return nil, nil, errs.NewInternalError(errs.SubtypeUnknown,
-				"parse base schema for field overrides: %s", err).WithCause(err)
-		}
-		orphans := schemas.ApplyFieldOverrides(parsed, def.Schema.FieldOverrides)
-		out, err := json.Marshal(parsed)
-		if err != nil {
-			return nil, nil, errs.NewInternalError(errs.SubtypeUnknown,
-				"serialize schema with field overrides: %s", err).WithCause(err)
-		}
-		return out, orphans, nil
-	}
-
-	return base, nil, nil
-}
-
-// pickSpec returns the non-nil spec and whether it is Native (requires V2 envelope wrap).
-func pickSpec(s eventlib.SchemaDef) (*eventlib.SchemaSpec, bool) {
-	if s.Native != nil {
-		return s.Native, true
-	}
-	if s.Custom != nil {
-		return s.Custom, false
-	}
-	return nil, false
-}
-
-// renderSpec produces a JSON Schema from Type (reflected) or Raw (copied).
-func renderSpec(s *eventlib.SchemaSpec) (json.RawMessage, error) {
-	if s.Type != nil {
-		return schemas.FromType(s.Type), nil
-	}
-	if len(s.Raw) > 0 {
-		buf := make(json.RawMessage, len(s.Raw))
-		copy(buf, s.Raw)
-		return buf, nil
-	}
-	return nil, errs.NewInternalError(errs.SubtypeUnknown, "schemaSpec has neither Type nor Raw")
-}
-
-func NewCmdSchema(f *cmdutil.Factory) *cobra.Command {
+func NewCmdSchema(f *cmdutil.Factory, snap *catalog.Snapshot) *cobra.Command {
 	var asJSON bool
 	cmd := &cobra.Command{
 		Use:   "schema <EventKey>",
@@ -87,7 +25,7 @@ func NewCmdSchema(f *cmdutil.Factory) *cobra.Command {
 		Long:  "Display detailed information about an EventKey including type, events, parameters, and response schema. Use --json for machine-readable output.",
 		Args:  cobra.ExactArgs(1),
 		RunE: func(cmd *cobra.Command, args []string) error {
-			return runSchema(f, args[0], asJSON)
+			return runSchema(f, snap, args[0], asJSON)
 		},
 	}
 	cmd.Flags().BoolVar(&asJSON, "json", false, "Emit the EventKey definition + resolved schema as JSON (for AI / scripts)")
@@ -95,14 +33,15 @@ func NewCmdSchema(f *cmdutil.Factory) *cobra.Command {
 	return cmd
 }
 
-func runSchema(f *cmdutil.Factory, key string, asJSON bool) error {
-	def, ok := eventlib.Lookup(key)
+func runSchema(f *cmdutil.Factory, snap *catalog.Snapshot, key string, asJSON bool) error {
+	entry, ok := snap.Resolve(key)
 	if !ok {
-		return unknownEventKeyErr(key)
+		return unknownEventKeyErr(snap, key)
 	}
+	def := entry.Definition()
 
 	if asJSON {
-		return writeSchemaJSON(f, def)
+		return writeSchemaJSON(f, entry)
 	}
 
 	out := f.IOStreams.Out
@@ -170,10 +109,7 @@ func runSchema(f *cmdutil.Factory, key string, asJSON bool) error {
 		}
 	}
 
-	resolved, _, err := resolveSchemaJSON(def)
-	if err != nil {
-		return err
-	}
+	resolved := entry.Output().SchemaJSON
 	if resolved != nil {
 		fmt.Fprintf(out, "\nOutput Schema:\n")
 		printIndentedJSON(out, resolved)
@@ -202,30 +138,22 @@ func printIndentedJSON(out io.Writer, raw json.RawMessage) {
 	fmt.Fprintf(out, "  %s\n", string(formatted))
 }
 
+// schemaPayload is the JSON shape of `event schema --json`. It is a named
+// type (not a function-local literal) so the render contract test can walk
+// its fields and reject accidental additions to the public output.
+type schemaPayload struct {
+	*eventlib.KeyDefinition
+	ResolvedSchema json.RawMessage `json:"resolved_output_schema,omitempty"`
+	JQRootPath     string          `json:"jq_root_path,omitempty"`
+}
+
 // writeSchemaJSON emits the EventKey definition plus resolved schema; jq_root_path tells callers whether fields live at `.` or `.event`.
-func writeSchemaJSON(f *cmdutil.Factory, def *eventlib.KeyDefinition) error {
-	type payload struct {
-		*eventlib.KeyDefinition
-		ResolvedSchema json.RawMessage `json:"resolved_output_schema,omitempty"`
-		JQRootPath     string          `json:"jq_root_path,omitempty"`
-	}
-	resolved, _, err := resolveSchemaJSON(def)
-	if err != nil {
-		return err
-	}
-	var jqRootPath string
-	if resolved != nil {
-		// Native → V2 envelope ⇒ `.event.xxx`; Custom → flat ⇒ `.`.
-		_, isNative := pickSpec(def.Schema)
-		jqRootPath = "."
-		if isNative {
-			jqRootPath = ".event"
-		}
-	}
-	output.PrintJson(f.IOStreams.Out, payload{
-		KeyDefinition:  def,
-		ResolvedSchema: resolved,
-		JQRootPath:     jqRootPath,
+func writeSchemaJSON(f *cmdutil.Factory, entry *catalog.Entry) error {
+	contract := entry.Output()
+	output.PrintJson(f.IOStreams.Out, schemaPayload{
+		KeyDefinition:  entry.Definition(),
+		ResolvedSchema: contract.SchemaJSON,
+		JQRootPath:     contract.JQRootPath,
 	})
 	return nil
 }

cmd/event/schema_test.go

@@ -10,15 +10,27 @@ import (
 	"strings"
 	"testing"
 
-	"github.com/larksuite/cli/errs"
 	"github.com/larksuite/cli/internal/cmdutil"
 	"github.com/larksuite/cli/internal/core"
 	eventlib "github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
 	"github.com/larksuite/cli/internal/event/schemas"
-
-	_ "github.com/larksuite/cli/events"
 )
 
+// compileTestSnapshot compiles synthetic declarations into a snapshot using
+// the same strategy set the production wiring provides.
+func compileTestSnapshot(t *testing.T, defs ...eventlib.KeyDefinition) *catalog.Snapshot {
+	t.Helper()
+	snap, err := catalog.Compile(defs, catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("compile test catalog: %v", err)
+	}
+	return snap
+}
+
 type approvalSchemaJSONPayload struct {
 	JQRootPath           string                           `json:"jq_root_path"`
 	AuthTypes            []string                         `json:"auth_types"`
@@ -45,7 +57,7 @@ type approvalSchemaJSONProperty struct {
 func TestRunSchema_ProcessedKey_Text(t *testing.T) {
 	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-	if err := runSchema(f, "im.message.receive_v1", false); err != nil {
+	if err := runSchema(f, compileCatalog(), "im.message.receive_v1", false); err != nil {
 		t.Fatalf("runSchema: %v", err)
 	}
 
@@ -65,7 +77,7 @@ func TestRunSchema_ProcessedKey_Text(t *testing.T) {
 func TestRunSchema_NativeKey_WrapsEnvelope(t *testing.T) {
 	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-	if err := runSchema(f, "im.message.message_read_v1", false); err != nil {
+	if err := runSchema(f, compileCatalog(), "im.message.message_read_v1", false); err != nil {
 		t.Fatalf("runSchema: %v", err)
 	}
 
@@ -85,7 +97,7 @@ func TestRunSchema_NativeKey_WrapsEnvelope(t *testing.T) {
 func TestRunSchema_UnknownKey_SuggestsAlternatives(t *testing.T) {
 	f, _, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-	err := runSchema(f, "im.message.recieve_v1", false)
+	err := runSchema(f, compileCatalog(), "im.message.recieve_v1", false)
 	if err == nil {
 		t.Fatal("expected error for unknown key")
 	}
@@ -101,7 +113,7 @@ func TestRunSchema_UnknownKey_SuggestsAlternatives(t *testing.T) {
 func TestRunSchema_JSONOutput(t *testing.T) {
 	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-	if err := runSchema(f, "im.message.receive_v1", true); err != nil {
+	if err := runSchema(f, compileCatalog(), "im.message.receive_v1", true); err != nil {
 		t.Fatalf("runSchema json: %v", err)
 	}
 
@@ -122,7 +134,7 @@ func TestRunSchema_JSONOutput(t *testing.T) {
 func TestRunSchema_ReceiveMessageAgentFieldsJSON(t *testing.T) {
 	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-	if err := runSchema(f, "im.message.receive_v1", true); err != nil {
+	if err := runSchema(f, compileCatalog(), "im.message.receive_v1", true); err != nil {
 		t.Fatalf("runSchema json: %v", err)
 	}
 
@@ -156,7 +168,7 @@ func TestRunSchema_ReceiveMessageAgentFieldsJSON(t *testing.T) {
 func TestRunSchema_TaskUpdateUserAccessJSON(t *testing.T) {
 	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-	if err := runSchema(f, "task.task.update_user_access_v2", true); err != nil {
+	if err := runSchema(f, compileCatalog(), "task.task.update_user_access_v2", true); err != nil {
 		t.Fatalf("runSchema json: %v", err)
 	}
 
@@ -195,7 +207,7 @@ func TestRunSchema_ApprovalStatusChangedJSON(t *testing.T) {
 			t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
 			f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-			if err := runSchema(f, tc.key, true); err != nil {
+			if err := runSchema(f, compileCatalog(), tc.key, true); err != nil {
 				t.Fatalf("runSchema json: %v", err)
 			}
 
@@ -243,7 +255,7 @@ func TestRunSchema_JSONOutput_VCMeetingLifecycleKeys(t *testing.T) {
 		t.Run(key, func(t *testing.T) {
 			f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
 
-			if err := runSchema(f, key, true); err != nil {
+			if err := runSchema(f, compileCatalog(), key, true); err != nil {
 				t.Fatalf("runSchema json: %v", err)
 			}
 
@@ -276,9 +288,8 @@ func TestRunSchema_JSONOutput_VCMeetingLifecycleKeys(t *testing.T) {
 
 func TestSchema_RendersSubscriptionKeyMarker(t *testing.T) {
 	const syntheticKey = "test.evt_sub"
-	t.Cleanup(func() { eventlib.UnregisterKeyForTest(syntheticKey) })
 
-	eventlib.RegisterKey(eventlib.KeyDefinition{
+	snap := compileTestSnapshot(t, eventlib.KeyDefinition{
 		Key:       syntheticKey,
 		EventType: syntheticKey,
 		Params: []eventlib.ParamDef{
@@ -289,7 +300,7 @@ func TestSchema_RendersSubscriptionKeyMarker(t *testing.T) {
 	})
 
 	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
-	if err := runSchema(f, syntheticKey, false); err != nil {
+	if err := runSchema(f, snap, syntheticKey, false); err != nil {
 		t.Fatalf("runSchema: %v", err)
 	}
 
@@ -325,9 +336,8 @@ func TestSchema_RendersSubscriptionKeyMarker(t *testing.T) {
 
 func TestSchema_JSON_IncludesSubscriptionKey(t *testing.T) {
 	const syntheticKey = "test.evt_json"
-	t.Cleanup(func() { eventlib.UnregisterKeyForTest(syntheticKey) })
 
-	eventlib.RegisterKey(eventlib.KeyDefinition{
+	snap := compileTestSnapshot(t, eventlib.KeyDefinition{
 		Key:       syntheticKey,
 		EventType: syntheticKey,
 		Params:    []eventlib.ParamDef{{Name: "mailbox", SubscriptionKey: true}},
@@ -335,7 +345,7 @@ func TestSchema_JSON_IncludesSubscriptionKey(t *testing.T) {
 	})
 
 	f, stdout, _, _ := cmdutil.TestFactory(t, &core.CliConfig{AppID: "test"})
-	if err := runSchema(f, syntheticKey, true); err != nil {
+	if err := runSchema(f, snap, syntheticKey, true); err != nil {
 		t.Fatalf("runSchema json: %v", err)
 	}
 
@@ -349,12 +359,13 @@ func TestSchema_JSON_IncludesSubscriptionKey(t *testing.T) {
 
 func TestResolveSchemaJSON_CustomWithOverlay(t *testing.T) {
 	const syntheticKey = "t.custom.overlay"
-	t.Cleanup(func() { eventlib.UnregisterKeyForTest(syntheticKey) })
 
 	type out struct {
 		SenderID string `json:"sender_id"`
 	}
-	eventlib.RegisterKey(eventlib.KeyDefinition{
+	// A compile that succeeds proves the overlay left no orphan pointers; the
+	// entry's output contract carries the resolved schema.
+	snap := compileTestSnapshot(t, eventlib.KeyDefinition{
 		Key:       syntheticKey,
 		EventType: syntheticKey,
 		Schema: eventlib.SchemaDef{
@@ -367,13 +378,12 @@ func TestResolveSchemaJSON_CustomWithOverlay(t *testing.T) {
 			return nil, nil
 		},
 	})
-	def, _ := eventlib.Lookup(syntheticKey)
-	resolved, orphans, err := resolveSchemaJSON(def)
-	if err != nil || len(orphans) != 0 {
-		t.Fatalf("resolve: err=%v orphans=%v", err, orphans)
+	entry, ok := snap.Resolve(syntheticKey)
+	if !ok {
+		t.Fatalf("snap.Resolve(%q) should succeed", syntheticKey)
 	}
 	var parsed map[string]interface{}
-	if err := json.Unmarshal(resolved, &parsed); err != nil {
+	if err := json.Unmarshal(entry.Output().SchemaJSON, &parsed); err != nil {
 		t.Fatal(err)
 	}
 	got := parsed["properties"].(map[string]interface{})["sender_id"].(map[string]interface{})["format"]
@@ -382,37 +392,35 @@ func TestResolveSchemaJSON_CustomWithOverlay(t *testing.T) {
 	}
 }
 
-func TestRenderSpec_EmptySpecIsTypedInternalError(t *testing.T) {
-	_, err := renderSpec(&eventlib.SchemaSpec{})
+func TestCompile_EmptySpecIsRejected(t *testing.T) {
+	_, err := catalog.Compile([]eventlib.KeyDefinition{{
+		Key:       "synthetic.empty.spec",
+		EventType: "synthetic.empty.spec",
+		Schema:    eventlib.SchemaDef{Native: &eventlib.SchemaSpec{}},
+	}}, catalog.StrategyRefs{catalog.StrategyNone})
 	if err == nil {
 		t.Fatal("expected error for spec with neither Type nor Raw")
 	}
-	p, ok := errs.ProblemOf(err)
-	if !ok {
-		t.Fatalf("expected typed errs error, got %T: %v", err, err)
-	}
-	if p.Category != errs.CategoryInternal {
-		t.Errorf("category = %s, want %s", p.Category, errs.CategoryInternal)
+	if !strings.Contains(err.Error(), "exactly one of Type or Raw") {
+		t.Errorf("error should reject the empty spec, got: %v", err)
 	}
 }
 
-func TestResolveSchemaJSON_InvalidBaseWithOverridesIsTypedInternalError(t *testing.T) {
-	def := &eventlib.KeyDefinition{
-		Key: "synthetic.invalid.base",
+func TestCompile_InvalidBaseWithOverridesIsRejected(t *testing.T) {
+	_, err := catalog.Compile([]eventlib.KeyDefinition{{
+		Key:       "synthetic.invalid.base",
+		EventType: "synthetic.invalid.base",
 		Schema: eventlib.SchemaDef{
 			Custom:         &eventlib.SchemaSpec{Raw: json.RawMessage("{not json")},
 			FieldOverrides: map[string]schemas.FieldMeta{"x": {}},
 		},
-	}
-	_, _, err := resolveSchemaJSON(def)
+	}}, catalog.StrategyRefs{catalog.StrategyNone})
 	if err == nil {
 		t.Fatal("expected error for unparsable base schema")
 	}
-	p, ok := errs.ProblemOf(err)
-	if !ok {
-		t.Fatalf("expected typed errs error, got %T: %v", err, err)
-	}
-	if p.Category != errs.CategoryInternal {
-		t.Errorf("category = %s, want %s", p.Category, errs.CategoryInternal)
+	// Garbage raw bytes are rejected by the spec check itself, before the
+	// overlay machinery would even try to parse them.
+	if !strings.Contains(err.Error(), "is not a JSON object") {
+		t.Errorf("error should reject the unparsable base schema, got: %v", err)
 	}
 }

cmd/event/service_adapters.go

@@ -0,0 +1,85 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"context"
+
+	eventlib "github.com/larksuite/cli/internal/event"
+	appconsume "github.com/larksuite/cli/internal/event/application/consume"
+	"github.com/larksuite/cli/internal/event/catalog"
+)
+
+// consumeStrategies is the executable strategy set for this binary. The same
+// registry is handed to catalog compilation, so a reference the compiler
+// accepted is guaranteed to resolve here.
+var consumeStrategies = appconsume.DefaultRegistry()
+
+type identityResolverFunc func(ctx context.Context, entry *catalog.Entry) (string, error)
+
+func (f identityResolverFunc) Resolve(ctx context.Context, entry *catalog.Entry) (string, error) {
+	return f(ctx, entry)
+}
+
+type preflightReaderFunc func(ctx context.Context, entry *catalog.Entry, identity string) ([]appconsume.Precondition, error)
+
+func (f preflightReaderFunc) Read(ctx context.Context, entry *catalog.Entry, identity string) ([]appconsume.Precondition, error) {
+	return f(ctx, entry, identity)
+}
+
+type streamRunnerFunc func(ctx context.Context, prepare appconsume.PrepareFunc) error
+
+func (f streamRunnerFunc) Run(ctx context.Context, prepare appconsume.PrepareFunc) error {
+	return f(ctx, prepare)
+}
+
+// readPreconditions classifies the existing read-only preflight checks into
+// named preconditions. Weak dependencies that could not answer stay visible
+// as "unknown" instead of silently passing; a failed check carries the exact
+// error a real run returns, so refusal is identical on both paths.
+func readPreconditions(ctx context.Context, pf *preflightCtx, appVerErr, tokenErr error) []appconsume.Precondition {
+	credentials := appconsume.Precondition{Name: "credentials_available", Status: appconsume.PreconditionOK}
+	if tokenErr != nil {
+		credentials.Status = appconsume.PreconditionBlocked
+		credentials.Detail = tokenErr.Error()
+		credentials.BlockErr = tokenErr
+	}
+
+	console := appconsume.Precondition{Name: "console_event_published", Status: appconsume.PreconditionOK}
+	switch {
+	case len(pf.keyDef.RequiredConsoleEvents) == 0:
+		// nothing to verify
+	case pf.keyDef.SubscriptionType == eventlib.SubTypeCallback && pf.subscribedCallbacks == nil,
+		pf.keyDef.SubscriptionType != eventlib.SubTypeCallback && pf.appVer == nil:
+		console.Status = appconsume.PreconditionUnknown
+		if appVerErr != nil {
+			console.Detail = describeAppMetaErr(appVerErr)
+		} else {
+			console.Detail = "console ledger unavailable"
+		}
+	default:
+		if err := preflightEventTypes(pf); err != nil {
+			console.Status = appconsume.PreconditionBlocked
+			console.Detail = err.Error()
+			console.BlockErr = err
+		}
+	}
+
+	scopes := appconsume.Precondition{Name: "scopes_granted", Status: appconsume.PreconditionOK}
+	checked, err := preflightScopes(ctx, pf)
+	switch {
+	case err != nil:
+		scopes.Status = appconsume.PreconditionBlocked
+		scopes.Detail = err.Error()
+		scopes.BlockErr = err
+	case !checked:
+		// The scope ledger could not be read (no published version for bots,
+		// no resolvable token for users). Saying "ok" here would dress up
+		// "nobody looked" as "it was verified".
+		scopes.Status = appconsume.PreconditionUnknown
+		scopes.Detail = "granted scopes could not be read for this identity"
+	}
+
+	return []appconsume.Precondition{credentials, console, scopes}
+}

cmd/event/status.go

@@ -14,10 +14,10 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/larksuite/cli/internal/cmdutil"
-	"github.com/larksuite/cli/internal/event/busctl"
-	"github.com/larksuite/cli/internal/event/busdiscover"
-	"github.com/larksuite/cli/internal/event/protocol"
-	"github.com/larksuite/cli/internal/event/transport"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/busctl"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/busdiscover"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/protocol"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/transport"
 	"github.com/larksuite/cli/internal/output"
 )
 

cmd/event/status_orphan_test.go

@@ -11,8 +11,8 @@ import (
 	"testing"
 	"time"
 
-	"github.com/larksuite/cli/internal/event/busdiscover"
-	"github.com/larksuite/cli/internal/event/protocol"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/busdiscover"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/protocol"
 )
 
 type fakeScanner struct {

cmd/event/stop.go

@@ -13,9 +13,9 @@ import (
 	"github.com/spf13/cobra"
 
 	"github.com/larksuite/cli/internal/cmdutil"
-	"github.com/larksuite/cli/internal/event/busctl"
-	"github.com/larksuite/cli/internal/event/busdiscover"
-	"github.com/larksuite/cli/internal/event/transport"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/busctl"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/busdiscover"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/transport"
 	"github.com/larksuite/cli/internal/output"
 )
 

cmd/event/stop_discover_test.go

@@ -9,7 +9,7 @@ import (
 	"sort"
 	"testing"
 
-	"github.com/larksuite/cli/internal/event/busdiscover"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/busdiscover"
 )
 
 func TestDiscoverAppIDs_OnlyLiveLockHolders(t *testing.T) {

cmd/event/stop_integration_test.go

@@ -13,7 +13,7 @@ import (
 	"testing"
 	"time"
 
-	"github.com/larksuite/cli/internal/event/protocol"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/protocol"
 )
 
 type mockTransport struct {

cmd/event/suggestions.go

@@ -9,14 +9,14 @@ import (
 	"strings"
 
 	"github.com/larksuite/cli/errs"
-	eventlib "github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
 	"github.com/larksuite/cli/internal/suggest"
 )
 
 const maxSuggestions = 3
 
 // suggestEventKeys returns up to maxSuggestions keys resembling input (substring match beats edit distance).
-func suggestEventKeys(input string) []string {
+func suggestEventKeys(snap *catalog.Snapshot, input string) []string {
 	type match struct {
 		key  string
 		dist int
@@ -24,13 +24,13 @@ func suggestEventKeys(input string) []string {
 	var hits []match
 	threshold := max(2, len(input)/5)
 
-	for _, def := range eventlib.ListAll() {
-		if strings.Contains(def.Key, input) {
-			hits = append(hits, match{def.Key, 0})
+	for _, key := range snap.Keys() {
+		if strings.Contains(key, input) {
+			hits = append(hits, match{key, 0})
 			continue
 		}
-		if d := suggest.Levenshtein(input, def.Key); d <= threshold {
-			hits = append(hits, match{def.Key, d})
+		if d := suggest.Levenshtein(input, key); d <= threshold {
+			hits = append(hits, match{key, d})
 		}
 	}
 	sort.Slice(hits, func(i, j int) bool { return hits[i].dist < hits[j].dist })
@@ -59,9 +59,9 @@ func formatSuggestions(keys []string) string {
 }
 
 // unknownEventKeyErr builds the shared "unknown EventKey" error with a suggestion tail when available.
-func unknownEventKeyErr(key string) error {
+func unknownEventKeyErr(snap *catalog.Snapshot, key string) error {
 	msg := fmt.Sprintf("unknown EventKey: %s", key)
-	if guesses := suggestEventKeys(key); len(guesses) > 0 {
+	if guesses := suggestEventKeys(snap, key); len(guesses) > 0 {
 		msg += " — did you mean " + formatSuggestions(guesses) + "?"
 	}
 	return errs.NewValidationError(errs.SubtypeInvalidArgument, "%s", msg).

cmd/event/suggestions_test.go

@@ -6,11 +6,10 @@ package event
 import (
 	"strings"
 	"testing"
-
-	_ "github.com/larksuite/cli/events"
 )
 
 func TestSuggestEventKeys(t *testing.T) {
+	snap := compileCatalog()
 	cases := []struct {
 		name              string
 		input             string
@@ -41,7 +40,7 @@ func TestSuggestEventKeys(t *testing.T) {
 	}
 	for _, tc := range cases {
 		t.Run(tc.name, func(t *testing.T) {
-			got := suggestEventKeys(tc.input)
+			got := suggestEventKeys(snap, tc.input)
 			if tc.wantEmpty {
 				if len(got) != 0 {
 					t.Errorf("expected empty slice, got %v", got)
@@ -98,7 +97,7 @@ func TestFormatSuggestions(t *testing.T) {
 }
 
 func TestUnknownEventKeyErr_IncludesSuggestion(t *testing.T) {
-	err := unknownEventKeyErr("im.message.recieve_v1")
+	err := unknownEventKeyErr(compileCatalog(), "im.message.recieve_v1")
 	if err == nil {
 		t.Fatal("expected error")
 	}
@@ -115,7 +114,7 @@ func TestUnknownEventKeyErr_IncludesSuggestion(t *testing.T) {
 }
 
 func TestUnknownEventKeyErr_NoSuggestion(t *testing.T) {
-	err := unknownEventKeyErr("xyzzy_no_such_event_key_at_all")
+	err := unknownEventKeyErr(compileCatalog(), "xyzzy_no_such_event_key_at_all")
 	if err == nil {
 		t.Fatal("expected error")
 	}

cmd/event/testdata/golden/list_text.golden

@@ -0,0 +1,42 @@
+KEY                                             AUTH      PARAMS  DESCRIPTION
+
+── application ──
+application.bot.menu_v6                         bot       0       Triggered when a user clicks a custom bot menu item whose action is configured as a push event.
+
+── approval ──
+approval.instance.status_changed_v4             user      1       Triggered after an approval instance status becomes visible to the requester or approval participants
+approval.task.status_changed_v4                 user      1       Triggered after an approval task status becomes visible to the requester or task approver
+
+── board ──
+board.whiteboard.updated_v1                     user|bot  1       Pushed when the whiteboard content is updated.
+
+── card ──
+card.action.trigger                             bot       0       Triggered when a user interacts with an interactive card (button click, form submit, dropdown select, etc.). Output includes: token (valid 30 min, max 2 updates), action details (tag, value, name, form_value), and card_content (original card in userDSL text format, auto-fetched at consume time). To update the card: parse card_content to understand the current state, construct the new card JSON, then call `lark-cli api POST /open-apis/interactive/v1/card/update` with the token (see lark-im-card-action-reply.md).
+
+── im ──
+im.chat.disbanded_v1                            bot       0       Triggered after a chat is disbanded
+im.chat.member.bot.added_v1                     bot       0       Triggered when the bot is added to a chat
+im.chat.member.bot.deleted_v1                   bot       0       Triggered after the bot is removed from a chat
+im.chat.member.user.added_v1                    bot       0       Triggered when a new user joins a chat (including topic chats)
+im.chat.member.user.deleted_v1                  bot       0       Triggered when a user leaves or is removed from a chat
+im.chat.member.user.withdrawn_v1                bot       0       Triggered after a pending user invite is withdrawn
+im.chat.updated_v1                              bot       0       Triggered after chat settings (owner, avatar, name, permissions, etc.) are updated
+im.message.message_read_v1                      bot       0       Triggered after a user reads a P2P message sent by the bot
+im.message.reaction.created_v1                  bot       0       Triggered when a reaction is added to a message
+im.message.reaction.deleted_v1                  bot       0       Triggered when a reaction is removed from a message
+im.message.receive_v1                           bot       0       Receive IM messages
+
+── minutes ──
+minutes.minute.generated_v1                     user      0       Triggered when a minute has been generated
+
+── task ──
+task.task.update_user_access_v2                 user|bot  0       Triggered when tasks visible to the current user or app are created, deleted, or updated
+
+── vc ──
+vc.meeting.participant_meeting_ended_v1         user      0       Triggered when a meeting the current user participates in has ended
+vc.meeting.participant_meeting_joined_v1        user      0       Triggered when the current user joins a meeting
+vc.meeting.participant_meeting_started_v1       user      0       Triggered when a meeting the current user participates in has started
+vc.note.generated_v1                            user      0       Triggered when a note has been generated
+vc.recording.recording_ended_v1                 user      0       Triggered when a recording_bean recording ends and uploads successfully; only generated when connected to Feishu software.
+vc.recording.recording_started_v1               user      0       Triggered when a recording_bean recording starts; only generated when connected to Feishu software.
+vc.recording.recording_transcript_generated_v1  user      0       Triggered when recording_bean transcript items are generated; only generated when connected to Feishu software.

cmd/event/testdata/golden/schema_board_whiteboard_json.golden

@@ -0,0 +1,127 @@
+{
+  "key": "board.whiteboard.updated_v1",
+  "display_name": "Whiteboard updated",
+  "description": "Pushed when the whiteboard content is updated.",
+  "event_type": "board.whiteboard.updated_v1",
+  "subscription_type": "event",
+  "params": [
+    {
+      "name": "whiteboard_id",
+      "type": "string",
+      "required": true,
+      "description": "Whiteboard id to subscribe; subscription is per-whiteboard.",
+      "subscription_key": true
+    }
+  ],
+  "schema": {
+    "native": {},
+    "field_overrides": {
+      "/event/operator_ids/*/open_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "open_id"
+      },
+      "/event/operator_ids/*/union_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "union_id"
+      },
+      "/event/operator_ids/*/user_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "user_id"
+      },
+      "/event/whiteboard_id": {
+        "Description": "whiteboard id to subscribe",
+        "Enum": null,
+        "Kind": "whiteboard_id"
+      }
+    }
+  },
+  "scopes": [
+    "board:whiteboard:node:read"
+  ],
+  "auth_types": [
+    "user",
+    "bot"
+  ],
+  "required_console_events": [
+    "board.whiteboard.updated_v1"
+  ],
+  "buffer_size": 100,
+  "workers": 1,
+  "resolved_output_schema": {
+    "description": "飞书事件",
+    "properties": {
+      "event": {
+        "properties": {
+          "operator_ids": {
+            "items": {
+              "properties": {
+                "open_id": {
+                  "format": "open_id",
+                  "type": "string"
+                },
+                "union_id": {
+                  "format": "union_id",
+                  "type": "string"
+                },
+                "user_id": {
+                  "format": "user_id",
+                  "type": "string"
+                }
+              },
+              "type": "object"
+            },
+            "type": "array"
+          },
+          "whiteboard_id": {
+            "description": "whiteboard id to subscribe",
+            "format": "whiteboard_id",
+            "type": "string"
+          }
+        },
+        "type": "object"
+      },
+      "header": {
+        "description": "事件头，所有事件结构一致",
+        "properties": {
+          "app_id": {
+            "description": "接收事件的应用 ID",
+            "type": "string"
+          },
+          "create_time": {
+            "description": "事件创建时间，毫秒时间戳字符串",
+            "type": "string"
+          },
+          "event_id": {
+            "description": "事件唯一 ID",
+            "type": "string"
+          },
+          "event_type": {
+            "description": "事件类型，用于路由",
+            "type": "string"
+          },
+          "tenant_key": {
+            "description": "租户唯一标识",
+            "type": "string"
+          },
+          "token": {
+            "description": "回调校验 token",
+            "type": "string"
+          }
+        },
+        "type": "object"
+      },
+      "schema": {
+        "description": "飞书事件协议版本",
+        "enum": [
+          "2.0"
+        ],
+        "type": "string"
+      }
+    },
+    "type": "object"
+  },
+  "jq_root_path": ".event"
+}

cmd/event/testdata/golden/schema_board_whiteboard_text.golden

@@ -0,0 +1,89 @@
+Key:         board.whiteboard.updated_v1
+Description: Pushed when the whiteboard content is updated.
+Event:       board.whiteboard.updated_v1
+Pre-consume: yes
+
+Required Scopes:
+  - board:whiteboard:node:read
+
+Required Console Events (must be enabled in developer console):
+  - board.whiteboard.updated_v1
+
+Parameters:
+  NAME           TYPE    REQUIRED  SUB-KEY  DEFAULT  DESCRIPTION
+  whiteboard_id  string  yes       yes      -        Whiteboard id to subscribe; subscription is per-whiteboard.
+
+Output Schema:
+  {
+    "description": "飞书事件",
+    "properties": {
+      "event": {
+        "properties": {
+          "operator_ids": {
+            "items": {
+              "properties": {
+                "open_id": {
+                  "format": "open_id",
+                  "type": "string"
+                },
+                "union_id": {
+                  "format": "union_id",
+                  "type": "string"
+                },
+                "user_id": {
+                  "format": "user_id",
+                  "type": "string"
+                }
+              },
+              "type": "object"
+            },
+            "type": "array"
+          },
+          "whiteboard_id": {
+            "description": "whiteboard id to subscribe",
+            "format": "whiteboard_id",
+            "type": "string"
+          }
+        },
+        "type": "object"
+      },
+      "header": {
+        "description": "事件头，所有事件结构一致",
+        "properties": {
+          "app_id": {
+            "description": "接收事件的应用 ID",
+            "type": "string"
+          },
+          "create_time": {
+            "description": "事件创建时间，毫秒时间戳字符串",
+            "type": "string"
+          },
+          "event_id": {
+            "description": "事件唯一 ID",
+            "type": "string"
+          },
+          "event_type": {
+            "description": "事件类型，用于路由",
+            "type": "string"
+          },
+          "tenant_key": {
+            "description": "租户唯一标识",
+            "type": "string"
+          },
+          "token": {
+            "description": "回调校验 token",
+            "type": "string"
+          }
+        },
+        "type": "object"
+      },
+      "schema": {
+        "description": "飞书事件协议版本",
+        "enum": [
+          "2.0"
+        ],
+        "type": "string"
+      }
+    },
+    "type": "object"
+  }

cmd/event/testdata/golden/schema_card_action_trigger_json.golden

@@ -0,0 +1,104 @@
+{
+  "key": "card.action.trigger",
+  "display_name": "Card action",
+  "description": "Triggered when a user interacts with an interactive card (button click, form submit, dropdown select, etc.). Output includes: token (valid 30 min, max 2 updates), action details (tag, value, name, form_value), and card_content (original card in userDSL text format, auto-fetched at consume time). To update the card: parse card_content to understand the current state, construct the new card JSON, then call `lark-cli api POST /open-apis/interactive/v1/card/update` with the token (see lark-im-card-action-reply.md).",
+  "event_type": "card.action.trigger",
+  "subscription_type": "callback",
+  "schema": {
+    "custom": {}
+  },
+  "scopes": [
+    "im:message:readonly"
+  ],
+  "auth_types": [
+    "bot"
+  ],
+  "required_console_events": [
+    "card.action.trigger"
+  ],
+  "buffer_size": 100,
+  "workers": 1,
+  "single_consumer": true,
+  "resolved_output_schema": {
+    "type": "object",
+    "properties": {
+      "action_name": {
+        "type": "string",
+        "description": "Element name attribute"
+      },
+      "action_tag": {
+        "type": "string",
+        "description": "Triggered element type: button/select_static/input/checker/etc"
+      },
+      "action_value": {
+        "type": "string",
+        "description": "Developer-defined action value as JSON string"
+      },
+      "card_content": {
+        "type": "string",
+        "description": "Original card JSON content (body.content) auto-fetched via message get API at consume time using message_id; empty if message_id absent or fetch fails"
+      },
+      "chat_id": {
+        "type": "string",
+        "description": "Chat ID",
+        "format": "chat_id"
+      },
+      "checked": {
+        "type": "boolean",
+        "description": "Checkbox state (for checkbox elements)"
+      },
+      "event_id": {
+        "type": "string",
+        "description": "Globally unique event ID"
+      },
+      "form_value": {
+        "type": "string",
+        "description": "Form submission values as JSON string (only on form submit)"
+      },
+      "host": {
+        "type": "string",
+        "description": "Host type: im_message / im_top_notice"
+      },
+      "input_value": {
+        "type": "string",
+        "description": "Input field value (only for input elements)"
+      },
+      "message_id": {
+        "type": "string",
+        "description": "Message ID of the card",
+        "format": "message_id"
+      },
+      "operator_id": {
+        "type": "string",
+        "description": "Operator open_id",
+        "format": "open_id"
+      },
+      "option": {
+        "type": "string",
+        "description": "Selected option value (for single-select dropdown)"
+      },
+      "options": {
+        "type": "string",
+        "description": "Selected options, comma-separated (for multi-select)"
+      },
+      "timestamp": {
+        "type": "string",
+        "description": "Event delivery time (ms timestamp string)",
+        "format": "timestamp_ms"
+      },
+      "timezone": {
+        "type": "string",
+        "description": "User timezone for date/time picker interactions"
+      },
+      "token": {
+        "type": "string",
+        "description": "Token for delay card update (valid 30 min, max 2 updates)"
+      },
+      "type": {
+        "type": "string",
+        "description": "Event type; always card.action.trigger"
+      }
+    }
+  },
+  "jq_root_path": "."
+}

cmd/event/testdata/golden/schema_card_action_trigger_text.golden

@@ -0,0 +1,92 @@
+Key:         card.action.trigger
+Description: Triggered when a user interacts with an interactive card (button click, form submit, dropdown select, etc.). Output includes: token (valid 30 min, max 2 updates), action details (tag, value, name, form_value), and card_content (original card in userDSL text format, auto-fetched at consume time). To update the card: parse card_content to understand the current state, construct the new card JSON, then call `lark-cli api POST /open-apis/interactive/v1/card/update` with the token (see lark-im-card-action-reply.md).
+Event:       card.action.trigger
+
+Required Scopes:
+  - im:message:readonly
+
+Required Console Events (must be enabled in developer console):
+  - card.action.trigger
+
+Output Schema:
+  {
+    "type": "object",
+    "properties": {
+      "action_name": {
+        "type": "string",
+        "description": "Element name attribute"
+      },
+      "action_tag": {
+        "type": "string",
+        "description": "Triggered element type: button/select_static/input/checker/etc"
+      },
+      "action_value": {
+        "type": "string",
+        "description": "Developer-defined action value as JSON string"
+      },
+      "card_content": {
+        "type": "string",
+        "description": "Original card JSON content (body.content) auto-fetched via message get API at consume time using message_id; empty if message_id absent or fetch fails"
+      },
+      "chat_id": {
+        "type": "string",
+        "description": "Chat ID",
+        "format": "chat_id"
+      },
+      "checked": {
+        "type": "boolean",
+        "description": "Checkbox state (for checkbox elements)"
+      },
+      "event_id": {
+        "type": "string",
+        "description": "Globally unique event ID"
+      },
+      "form_value": {
+        "type": "string",
+        "description": "Form submission values as JSON string (only on form submit)"
+      },
+      "host": {
+        "type": "string",
+        "description": "Host type: im_message / im_top_notice"
+      },
+      "input_value": {
+        "type": "string",
+        "description": "Input field value (only for input elements)"
+      },
+      "message_id": {
+        "type": "string",
+        "description": "Message ID of the card",
+        "format": "message_id"
+      },
+      "operator_id": {
+        "type": "string",
+        "description": "Operator open_id",
+        "format": "open_id"
+      },
+      "option": {
+        "type": "string",
+        "description": "Selected option value (for single-select dropdown)"
+      },
+      "options": {
+        "type": "string",
+        "description": "Selected options, comma-separated (for multi-select)"
+      },
+      "timestamp": {
+        "type": "string",
+        "description": "Event delivery time (ms timestamp string)",
+        "format": "timestamp_ms"
+      },
+      "timezone": {
+        "type": "string",
+        "description": "User timezone for date/time picker interactions"
+      },
+      "token": {
+        "type": "string",
+        "description": "Token for delay card update (valid 30 min, max 2 updates)"
+      },
+      "type": {
+        "type": "string",
+        "description": "Event type; always card.action.trigger"
+      }
+    }
+  }

cmd/event/testdata/golden/schema_im_chat_updated_json.golden

@@ -0,0 +1,430 @@
+{
+  "key": "im.chat.updated_v1",
+  "display_name": "Chat updated",
+  "description": "Triggered after chat settings (owner, avatar, name, permissions, etc.) are updated",
+  "event_type": "im.chat.updated_v1",
+  "subscription_type": "event",
+  "schema": {
+    "native": {},
+    "field_overrides": {
+      "/event/after_change/owner_id/open_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "open_id"
+      },
+      "/event/after_change/owner_id/union_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "union_id"
+      },
+      "/event/after_change/owner_id/user_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "user_id"
+      },
+      "/event/before_change/owner_id/open_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "open_id"
+      },
+      "/event/before_change/owner_id/union_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "union_id"
+      },
+      "/event/before_change/owner_id/user_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "user_id"
+      },
+      "/event/chat_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "chat_id"
+      },
+      "/event/moderator_list/added_member_list/*/user_id/open_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "open_id"
+      },
+      "/event/moderator_list/added_member_list/*/user_id/union_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "union_id"
+      },
+      "/event/moderator_list/added_member_list/*/user_id/user_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "user_id"
+      },
+      "/event/moderator_list/removed_member_list/*/user_id/open_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "open_id"
+      },
+      "/event/moderator_list/removed_member_list/*/user_id/union_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "union_id"
+      },
+      "/event/moderator_list/removed_member_list/*/user_id/user_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "user_id"
+      },
+      "/event/operator_id/open_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "open_id"
+      },
+      "/event/operator_id/union_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "union_id"
+      },
+      "/event/operator_id/user_id": {
+        "Description": "",
+        "Enum": null,
+        "Kind": "user_id"
+      }
+    }
+  },
+  "scopes": [
+    "im:chat:read"
+  ],
+  "auth_types": [
+    "bot"
+  ],
+  "required_console_events": [
+    "im.chat.updated_v1"
+  ],
+  "buffer_size": 100,
+  "workers": 1,
+  "resolved_output_schema": {
+    "description": "飞书事件",
+    "properties": {
+      "event": {
+        "properties": {
+          "after_change": {
+            "properties": {
+              "add_member_permission": {
+                "type": "string"
+              },
+              "at_all_permission": {
+                "type": "string"
+              },
+              "avatar": {
+                "type": "string"
+              },
+              "description": {
+                "type": "string"
+              },
+              "edit_permission": {
+                "type": "string"
+              },
+              "group_message_type": {
+                "type": "string"
+              },
+              "i18n_names": {
+                "properties": {
+                  "en_us": {
+                    "type": "string"
+                  },
+                  "ja_jp": {
+                    "type": "string"
+                  },
+                  "zh_cn": {
+                    "type": "string"
+                  }
+                },
+                "type": "object"
+              },
+              "join_message_visibility": {
+                "type": "string"
+              },
+              "labels": {
+                "items": {
+                  "type": "string"
+                },
+                "type": "array"
+              },
+              "leave_message_visibility": {
+                "type": "string"
+              },
+              "membership_approval": {
+                "type": "string"
+              },
+              "moderation_permission": {
+                "type": "string"
+              },
+              "name": {
+                "type": "string"
+              },
+              "owner_id": {
+                "properties": {
+                  "open_id": {
+                    "format": "open_id",
+                    "type": "string"
+                  },
+                  "union_id": {
+                    "format": "union_id",
+                    "type": "string"
+                  },
+                  "user_id": {
+                    "format": "user_id",
+                    "type": "string"
+                  }
+                },
+                "type": "object"
+              },
+              "restricted_mode_setting": {
+                "properties": {
+                  "download_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "message_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "screenshot_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "status": {
+                    "type": "boolean"
+                  }
+                },
+                "type": "object"
+              },
+              "share_card_permission": {
+                "type": "string"
+              }
+            },
+            "type": "object"
+          },
+          "before_change": {
+            "properties": {
+              "add_member_permission": {
+                "type": "string"
+              },
+              "at_all_permission": {
+                "type": "string"
+              },
+              "avatar": {
+                "type": "string"
+              },
+              "description": {
+                "type": "string"
+              },
+              "edit_permission": {
+                "type": "string"
+              },
+              "group_message_type": {
+                "type": "string"
+              },
+              "i18n_names": {
+                "properties": {
+                  "en_us": {
+                    "type": "string"
+                  },
+                  "ja_jp": {
+                    "type": "string"
+                  },
+                  "zh_cn": {
+                    "type": "string"
+                  }
+                },
+                "type": "object"
+              },
+              "join_message_visibility": {
+                "type": "string"
+              },
+              "labels": {
+                "items": {
+                  "type": "string"
+                },
+                "type": "array"
+              },
+              "leave_message_visibility": {
+                "type": "string"
+              },
+              "membership_approval": {
+                "type": "string"
+              },
+              "moderation_permission": {
+                "type": "string"
+              },
+              "name": {
+                "type": "string"
+              },
+              "owner_id": {
+                "properties": {
+                  "open_id": {
+                    "format": "open_id",
+                    "type": "string"
+                  },
+                  "union_id": {
+                    "format": "union_id",
+                    "type": "string"
+                  },
+                  "user_id": {
+                    "format": "user_id",
+                    "type": "string"
+                  }
+                },
+                "type": "object"
+              },
+              "restricted_mode_setting": {
+                "properties": {
+                  "download_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "message_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "screenshot_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "status": {
+                    "type": "boolean"
+                  }
+                },
+                "type": "object"
+              },
+              "share_card_permission": {
+                "type": "string"
+              }
+            },
+            "type": "object"
+          },
+          "chat_id": {
+            "format": "chat_id",
+            "type": "string"
+          },
+          "external": {
+            "type": "boolean"
+          },
+          "moderator_list": {
+            "properties": {
+              "added_member_list": {
+                "items": {
+                  "properties": {
+                    "tenant_key": {
+                      "type": "string"
+                    },
+                    "user_id": {
+                      "properties": {
+                        "open_id": {
+                          "format": "open_id",
+                          "type": "string"
+                        },
+                        "union_id": {
+                          "format": "union_id",
+                          "type": "string"
+                        },
+                        "user_id": {
+                          "format": "user_id",
+                          "type": "string"
+                        }
+                      },
+                      "type": "object"
+                    }
+                  },
+                  "type": "object"
+                },
+                "type": "array"
+              },
+              "removed_member_list": {
+                "items": {
+                  "properties": {
+                    "tenant_key": {
+                      "type": "string"
+                    },
+                    "user_id": {
+                      "properties": {
+                        "open_id": {
+                          "format": "open_id",
+                          "type": "string"
+                        },
+                        "union_id": {
+                          "format": "union_id",
+                          "type": "string"
+                        },
+                        "user_id": {
+                          "format": "user_id",
+                          "type": "string"
+                        }
+                      },
+                      "type": "object"
+                    }
+                  },
+                  "type": "object"
+                },
+                "type": "array"
+              }
+            },
+            "type": "object"
+          },
+          "operator_id": {
+            "properties": {
+              "open_id": {
+                "format": "open_id",
+                "type": "string"
+              },
+              "union_id": {
+                "format": "union_id",
+                "type": "string"
+              },
+              "user_id": {
+                "format": "user_id",
+                "type": "string"
+              }
+            },
+            "type": "object"
+          },
+          "operator_tenant_key": {
+            "type": "string"
+          }
+        },
+        "type": "object"
+      },
+      "header": {
+        "description": "事件头，所有事件结构一致",
+        "properties": {
+          "app_id": {
+            "description": "接收事件的应用 ID",
+            "type": "string"
+          },
+          "create_time": {
+            "description": "事件创建时间，毫秒时间戳字符串",
+            "type": "string"
+          },
+          "event_id": {
+            "description": "事件唯一 ID",
+            "type": "string"
+          },
+          "event_type": {
+            "description": "事件类型，用于路由",
+            "type": "string"
+          },
+          "tenant_key": {
+            "description": "租户唯一标识",
+            "type": "string"
+          },
+          "token": {
+            "description": "回调校验 token",
+            "type": "string"
+          }
+        },
+        "type": "object"
+      },
+      "schema": {
+        "description": "飞书事件协议版本",
+        "enum": [
+          "2.0"
+        ],
+        "type": "string"
+      }
+    },
+    "type": "object"
+  },
+  "jq_root_path": ".event"
+}

cmd/event/testdata/golden/schema_im_chat_updated_text.golden

@@ -0,0 +1,337 @@
+Key:         im.chat.updated_v1
+Description: Triggered after chat settings (owner, avatar, name, permissions, etc.) are updated
+Event:       im.chat.updated_v1
+
+Required Scopes:
+  - im:chat:read
+
+Required Console Events (must be enabled in developer console):
+  - im.chat.updated_v1
+
+Output Schema:
+  {
+    "description": "飞书事件",
+    "properties": {
+      "event": {
+        "properties": {
+          "after_change": {
+            "properties": {
+              "add_member_permission": {
+                "type": "string"
+              },
+              "at_all_permission": {
+                "type": "string"
+              },
+              "avatar": {
+                "type": "string"
+              },
+              "description": {
+                "type": "string"
+              },
+              "edit_permission": {
+                "type": "string"
+              },
+              "group_message_type": {
+                "type": "string"
+              },
+              "i18n_names": {
+                "properties": {
+                  "en_us": {
+                    "type": "string"
+                  },
+                  "ja_jp": {
+                    "type": "string"
+                  },
+                  "zh_cn": {
+                    "type": "string"
+                  }
+                },
+                "type": "object"
+              },
+              "join_message_visibility": {
+                "type": "string"
+              },
+              "labels": {
+                "items": {
+                  "type": "string"
+                },
+                "type": "array"
+              },
+              "leave_message_visibility": {
+                "type": "string"
+              },
+              "membership_approval": {
+                "type": "string"
+              },
+              "moderation_permission": {
+                "type": "string"
+              },
+              "name": {
+                "type": "string"
+              },
+              "owner_id": {
+                "properties": {
+                  "open_id": {
+                    "format": "open_id",
+                    "type": "string"
+                  },
+                  "union_id": {
+                    "format": "union_id",
+                    "type": "string"
+                  },
+                  "user_id": {
+                    "format": "user_id",
+                    "type": "string"
+                  }
+                },
+                "type": "object"
+              },
+              "restricted_mode_setting": {
+                "properties": {
+                  "download_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "message_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "screenshot_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "status": {
+                    "type": "boolean"
+                  }
+                },
+                "type": "object"
+              },
+              "share_card_permission": {
+                "type": "string"
+              }
+            },
+            "type": "object"
+          },
+          "before_change": {
+            "properties": {
+              "add_member_permission": {
+                "type": "string"
+              },
+              "at_all_permission": {
+                "type": "string"
+              },
+              "avatar": {
+                "type": "string"
+              },
+              "description": {
+                "type": "string"
+              },
+              "edit_permission": {
+                "type": "string"
+              },
+              "group_message_type": {
+                "type": "string"
+              },
+              "i18n_names": {
+                "properties": {
+                  "en_us": {
+                    "type": "string"
+                  },
+                  "ja_jp": {
+                    "type": "string"
+                  },
+                  "zh_cn": {
+                    "type": "string"
+                  }
+                },
+                "type": "object"
+              },
+              "join_message_visibility": {
+                "type": "string"
+              },
+              "labels": {
+                "items": {
+                  "type": "string"
+                },
+                "type": "array"
+              },
+              "leave_message_visibility": {
+                "type": "string"
+              },
+              "membership_approval": {
+                "type": "string"
+              },
+              "moderation_permission": {
+                "type": "string"
+              },
+              "name": {
+                "type": "string"
+              },
+              "owner_id": {
+                "properties": {
+                  "open_id": {
+                    "format": "open_id",
+                    "type": "string"
+                  },
+                  "union_id": {
+                    "format": "union_id",
+                    "type": "string"
+                  },
+                  "user_id": {
+                    "format": "user_id",
+                    "type": "string"
+                  }
+                },
+                "type": "object"
+              },
+              "restricted_mode_setting": {
+                "properties": {
+                  "download_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "message_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "screenshot_has_permission_setting": {
+                    "type": "string"
+                  },
+                  "status": {
+                    "type": "boolean"
+                  }
+                },
+                "type": "object"
+              },
+              "share_card_permission": {
+                "type": "string"
+              }
+            },
+            "type": "object"
+          },
+          "chat_id": {
+            "format": "chat_id",
+            "type": "string"
+          },
+          "external": {
+            "type": "boolean"
+          },
+          "moderator_list": {
+            "properties": {
+              "added_member_list": {
+                "items": {
+                  "properties": {
+                    "tenant_key": {
+                      "type": "string"
+                    },
+                    "user_id": {
+                      "properties": {
+                        "open_id": {
+                          "format": "open_id",
+                          "type": "string"
+                        },
+                        "union_id": {
+                          "format": "union_id",
+                          "type": "string"
+                        },
+                        "user_id": {
+                          "format": "user_id",
+                          "type": "string"
+                        }
+                      },
+                      "type": "object"
+                    }
+                  },
+                  "type": "object"
+                },
+                "type": "array"
+              },
+              "removed_member_list": {
+                "items": {
+                  "properties": {
+                    "tenant_key": {
+                      "type": "string"
+                    },
+                    "user_id": {
+                      "properties": {
+                        "open_id": {
+                          "format": "open_id",
+                          "type": "string"
+                        },
+                        "union_id": {
+                          "format": "union_id",
+                          "type": "string"
+                        },
+                        "user_id": {
+                          "format": "user_id",
+                          "type": "string"
+                        }
+                      },
+                      "type": "object"
+                    }
+                  },
+                  "type": "object"
+                },
+                "type": "array"
+              }
+            },
+            "type": "object"
+          },
+          "operator_id": {
+            "properties": {
+              "open_id": {
+                "format": "open_id",
+                "type": "string"
+              },
+              "union_id": {
+                "format": "union_id",
+                "type": "string"
+              },
+              "user_id": {
+                "format": "user_id",
+                "type": "string"
+              }
+            },
+            "type": "object"
+          },
+          "operator_tenant_key": {
+            "type": "string"
+          }
+        },
+        "type": "object"
+      },
+      "header": {
+        "description": "事件头，所有事件结构一致",
+        "properties": {
+          "app_id": {
+            "description": "接收事件的应用 ID",
+            "type": "string"
+          },
+          "create_time": {
+            "description": "事件创建时间，毫秒时间戳字符串",
+            "type": "string"
+          },
+          "event_id": {
+            "description": "事件唯一 ID",
+            "type": "string"
+          },
+          "event_type": {
+            "description": "事件类型，用于路由",
+            "type": "string"
+          },
+          "tenant_key": {
+            "description": "租户唯一标识",
+            "type": "string"
+          },
+          "token": {
+            "description": "回调校验 token",
+            "type": "string"
+          }
+        },
+        "type": "object"
+      },
+      "schema": {
+        "description": "飞书事件协议版本",
+        "enum": [
+          "2.0"
+        ],
+        "type": "string"
+      }
+    },
+    "type": "object"
+  }

cmd/event/testdata/golden/schema_im_message_receive_json.golden

@@ -0,0 +1,130 @@
+{
+  "key": "im.message.receive_v1",
+  "display_name": "Receive message",
+  "description": "Receive IM messages",
+  "event_type": "im.message.receive_v1",
+  "subscription_type": "event",
+  "schema": {
+    "custom": {}
+  },
+  "scopes": [
+    "im:message.p2p_msg:readonly"
+  ],
+  "auth_types": [
+    "bot"
+  ],
+  "required_console_events": [
+    "im.message.receive_v1"
+  ],
+  "buffer_size": 100,
+  "workers": 1,
+  "resolved_output_schema": {
+    "type": "object",
+    "properties": {
+      "chat_id": {
+        "type": "string",
+        "description": "Chat/conversation ID; prefixed with oc_",
+        "format": "chat_id"
+      },
+      "chat_type": {
+        "type": "string",
+        "description": "Conversation type",
+        "enum": [
+          "p2p",
+          "group"
+        ]
+      },
+      "content": {
+        "type": "string",
+        "description": "Message content. For most types (text/post/image/file/audio, etc.) this is pre-rendered human-readable text."
+      },
+      "create_time": {
+        "type": "string",
+        "description": "Message creation time (ms timestamp string)",
+        "format": "timestamp_ms"
+      },
+      "event_id": {
+        "type": "string",
+        "description": "Event delivery ID. Do not use as the message deduplication key; use message_id instead."
+      },
+      "id": {
+        "type": "string",
+        "description": "Message ID (legacy alias of message_id, kept for compatibility)",
+        "format": "message_id"
+      },
+      "mentions": {
+        "type": "array",
+        "description": "Compact mentions aligned with im +messages-mget",
+        "items": {
+          "type": "object",
+          "properties": {
+            "id": {
+              "type": "string",
+              "description": "Mentioned user open_id; prefixed with ou_",
+              "format": "open_id"
+            },
+            "key": {
+              "type": "string",
+              "description": "Mention placeholder key, for example @_user_1"
+            },
+            "name": {
+              "type": "string",
+              "description": "Mentioned display name"
+            }
+          }
+        }
+      },
+      "message_id": {
+        "type": "string",
+        "description": "Message ID; prefixed with om_. Recommended idempotency key for im.message.receive_v1 consumers.",
+        "format": "message_id"
+      },
+      "message_type": {
+        "type": "string",
+        "description": "Message type"
+      },
+      "reply_to": {
+        "type": "string",
+        "description": "Parent message ID of the direct reply context, when present",
+        "format": "message_id"
+      },
+      "root_id": {
+        "type": "string",
+        "description": "Root message ID of the reply/thread context, when present",
+        "format": "message_id"
+      },
+      "sender_id": {
+        "type": "string",
+        "description": "Sender open_id; prefixed with ou_",
+        "format": "open_id"
+      },
+      "sender_type": {
+        "type": "string",
+        "description": "Sender type",
+        "enum": [
+          "user",
+          "bot"
+        ]
+      },
+      "thread_id": {
+        "type": "string",
+        "description": "Thread ID, when present"
+      },
+      "timestamp": {
+        "type": "string",
+        "description": "Event delivery time (ms timestamp string); prefers header.create_time",
+        "format": "timestamp_ms"
+      },
+      "type": {
+        "type": "string",
+        "description": "Event type; always im.message.receive_v1"
+      },
+      "update_time": {
+        "type": "string",
+        "description": "Message update time (ms timestamp string); emitted only when different from create_time",
+        "format": "timestamp_ms"
+      }
+    }
+  },
+  "jq_root_path": "."
+}

cmd/event/testdata/golden/schema_im_message_receive_text.golden

@@ -0,0 +1,119 @@
+Key:         im.message.receive_v1
+Description: Receive IM messages
+Event:       im.message.receive_v1
+
+Required Scopes:
+  - im:message.p2p_msg:readonly
+
+Required Console Events (must be enabled in developer console):
+  - im.message.receive_v1
+
+Output Schema:
+  {
+    "type": "object",
+    "properties": {
+      "chat_id": {
+        "type": "string",
+        "description": "Chat/conversation ID; prefixed with oc_",
+        "format": "chat_id"
+      },
+      "chat_type": {
+        "type": "string",
+        "description": "Conversation type",
+        "enum": [
+          "p2p",
+          "group"
+        ]
+      },
+      "content": {
+        "type": "string",
+        "description": "Message content. For most types (text/post/image/file/audio, etc.) this is pre-rendered human-readable text."
+      },
+      "create_time": {
+        "type": "string",
+        "description": "Message creation time (ms timestamp string)",
+        "format": "timestamp_ms"
+      },
+      "event_id": {
+        "type": "string",
+        "description": "Event delivery ID. Do not use as the message deduplication key; use message_id instead."
+      },
+      "id": {
+        "type": "string",
+        "description": "Message ID (legacy alias of message_id, kept for compatibility)",
+        "format": "message_id"
+      },
+      "mentions": {
+        "type": "array",
+        "description": "Compact mentions aligned with im +messages-mget",
+        "items": {
+          "type": "object",
+          "properties": {
+            "id": {
+              "type": "string",
+              "description": "Mentioned user open_id; prefixed with ou_",
+              "format": "open_id"
+            },
+            "key": {
+              "type": "string",
+              "description": "Mention placeholder key, for example @_user_1"
+            },
+            "name": {
+              "type": "string",
+              "description": "Mentioned display name"
+            }
+          }
+        }
+      },
+      "message_id": {
+        "type": "string",
+        "description": "Message ID; prefixed with om_. Recommended idempotency key for im.message.receive_v1 consumers.",
+        "format": "message_id"
+      },
+      "message_type": {
+        "type": "string",
+        "description": "Message type"
+      },
+      "reply_to": {
+        "type": "string",
+        "description": "Parent message ID of the direct reply context, when present",
+        "format": "message_id"
+      },
+      "root_id": {
+        "type": "string",
+        "description": "Root message ID of the reply/thread context, when present",
+        "format": "message_id"
+      },
+      "sender_id": {
+        "type": "string",
+        "description": "Sender open_id; prefixed with ou_",
+        "format": "open_id"
+      },
+      "sender_type": {
+        "type": "string",
+        "description": "Sender type",
+        "enum": [
+          "user",
+          "bot"
+        ]
+      },
+      "thread_id": {
+        "type": "string",
+        "description": "Thread ID, when present"
+      },
+      "timestamp": {
+        "type": "string",
+        "description": "Event delivery time (ms timestamp string); prefers header.create_time",
+        "format": "timestamp_ms"
+      },
+      "type": {
+        "type": "string",
+        "description": "Event type; always im.message.receive_v1"
+      },
+      "update_time": {
+        "type": "string",
+        "description": "Message update time (ms timestamp string); emitted only when different from create_time",
+        "format": "timestamp_ms"
+      }
+    }
+  }

cmd/event/wiring.go

@@ -0,0 +1,25 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package event
+
+import (
+	"fmt"
+
+	"github.com/larksuite/cli/events"
+	"github.com/larksuite/cli/internal/event/catalog"
+)
+
+// compileCatalog is the event command tree's single assembly point: it turns
+// the aggregated domain declarations into the immutable snapshot every
+// subcommand reads. A compile failure is a defect in declarations built into
+// this binary — there is nothing to recover at runtime, so it panics.
+func compileCatalog() *catalog.Snapshot {
+	// The strategy registry that validates references is the same one that
+	// executes them, so "compiled" implies "resolvable at run time".
+	snap, err := catalog.Compile(events.All(), consumeStrategies)
+	if err != nil {
+		panic(fmt.Sprintf("event catalog failed to compile: %v", err))
+	}
+	return snap
+}

events/all.go

@@ -1,7 +1,9 @@
 // Copyright (c) 2026 Lark Technologies Pte. Ltd.
 // SPDX-License-Identifier: MIT
 
-// Package events wires domain EventKey definitions into the global registry. Blank-import to populate.
+// Package events aggregates the domain EventKey declarations. All returns
+// them explicitly — whoever needs a catalog compiles one; nothing registers
+// itself through import side effects.
 package events
 
 import (
@@ -12,12 +14,14 @@ import (
 	"github.com/larksuite/cli/events/task"
 	"github.com/larksuite/cli/events/vc"
 	"github.com/larksuite/cli/events/whiteboard"
-	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
 )
 
+// All returns every domain's declarations, ready for catalog.Compile.
 // Mail is intentionally omitted in this phase.
-func init() {
-	all := [][]event.KeyDefinition{
+func All() []catalog.KeyDefinition {
+	var all []catalog.KeyDefinition
+	for _, keys := range [][]catalog.KeyDefinition{
 		application.Keys(),
 		approval.Keys(),
 		im.Keys(),
@@ -25,10 +29,8 @@ func init() {
 		task.Keys(),
 		vc.Keys(),
 		whiteboard.Keys(),
+	} {
+		all = append(all, keys...)
 	}
-	for _, keys := range all {
-		for _, k := range keys {
-			event.RegisterKey(k)
-		}
-	}
+	return all
 }

events/application/menu.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 // BotMenuOutput is the flattened shape for application.bot.menu_v6.
@@ -29,13 +30,6 @@ type BotMenuOutput struct {
 
 func processBotMenu(_ context.Context, _ event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
 	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-			AppID      string `json:"app_id"`
-			TenantKey  string `json:"tenant_key"`
-		} `json:"header"`
 		Event struct {
 			EventKey  string          `json:"event_key"`
 			Timestamp json.RawMessage `json:"timestamp"`
@@ -50,11 +44,11 @@ func processBotMenu(_ context.Context, _ event.APIClient, raw *event.RawEvent, _
 		} `json:"event"`
 	}
 	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload so consumers still see the event
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
 	menuTimestamp := timestampMillisString(envelope.Event.Timestamp)
-	timestamp := envelope.Header.CreateTime
+	timestamp := raw.SourceTime
 	if timestamp == "" {
 		timestamp = menuTimestamp
 	}
@@ -62,10 +56,10 @@ func processBotMenu(_ context.Context, _ event.APIClient, raw *event.RawEvent, _
 
 	out := &BotMenuOutput{
 		Type:            eventTypeBotMenuV6,
-		EventID:         envelope.Header.EventID,
+		EventID:         raw.EventID,
 		Timestamp:       timestamp,
-		AppID:           envelope.Header.AppID,
-		TenantKey:       envelope.Header.TenantKey,
+		AppID:           raw.AppID,
+		TenantKey:       raw.TenantKey,
 		EventKey:        envelope.Event.EventKey,
 		MenuTimestamp:   menuTimestamp,
 		OperatorID:      operatorID,

events/application/menu_test.go

@@ -11,6 +11,8 @@ import (
 	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 func TestKeysBotMenuMetadata(t *testing.T) {
@@ -51,14 +53,15 @@ func TestKeysBotMenuMetadata(t *testing.T) {
 
 func TestBotMenuRegistersCleanly(t *testing.T) {
 	const key = eventTypeBotMenuV6
-	event.UnregisterKeyForTest(key)
-	t.Cleanup(func() { event.UnregisterKeyForTest(key) })
-
-	for _, def := range Keys() {
-		event.RegisterKey(def)
+	snap, err := catalog.Compile(Keys(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("catalog.Compile(Keys()): %v", err)
 	}
-	if _, ok := event.Lookup(key); !ok {
-		t.Fatalf("event.Lookup(%q) not registered", key)
+	if _, ok := snap.Resolve(key); !ok {
+		t.Fatalf("snap.Resolve(%q): key missing from compiled catalog", key)
 	}
 }
 
@@ -199,14 +202,42 @@ func TestProcessBotMenuMalformedPayload(t *testing.T) {
 		Timestamp: time.Now(),
 	}
 	got, err := processBotMenu(context.Background(), nil, raw, nil)
-	if err != nil {
-		t.Fatalf("Process should swallow parse errors, got %v", err)
+	if !processing.IsDropMalformed(err) {
+		t.Fatalf("malformed payload must be dropped with a malformed marker, got err=%v", err)
 	}
-	if string(got) != "not json" {
-		t.Errorf("malformed fallback output = %q, want original bytes", string(got))
+	if got != nil {
+		t.Errorf("malformed payload must be dropped without output, got %q", string(got))
 	}
 }
 
+// fillCanonicalFromHeader copies the payload envelope header metadata onto
+// the RawEvent canonical fields. Process handlers read event_id, create_time,
+// app_id, and tenant_key from the RawEvent, which the consume pipeline fills
+// from the envelope header before dispatch; tests that hand-build a RawEvent
+// must mirror that so both views agree.
+func fillCanonicalFromHeader(t *testing.T, raw *event.RawEvent) {
+	t.Helper()
+	var envelope struct {
+		Header struct {
+			EventID    string `json:"event_id"`
+			EventType  string `json:"event_type"`
+			CreateTime string `json:"create_time"`
+			AppID      string `json:"app_id"`
+			TenantKey  string `json:"tenant_key"`
+		} `json:"header"`
+	}
+	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
+		t.Fatalf("parse envelope header: %v", err)
+	}
+	raw.EventID = envelope.Header.EventID
+	if envelope.Header.EventType != "" {
+		raw.EventType = envelope.Header.EventType
+	}
+	raw.SourceTime = envelope.Header.CreateTime
+	raw.AppID = envelope.Header.AppID
+	raw.TenantKey = envelope.Header.TenantKey
+}
+
 func runBotMenu(t *testing.T, payload string) BotMenuOutput {
 	t.Helper()
 	raw := &event.RawEvent{
@@ -215,6 +246,7 @@ func runBotMenu(t *testing.T, payload string) BotMenuOutput {
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := processBotMenu(context.Background(), nil, raw, nil)
 	if err != nil {
 		t.Fatalf("processBotMenu: %v", err)

events/approval/preconsume.go

@@ -13,23 +13,13 @@ import (
 	"github.com/larksuite/cli/internal/event"
 )
 
-type approvalEventType string
-type approvalSubscriptionPath string
-
-type approvalSubscriptionConfig struct {
-	eventType     approvalEventType
-	subscribePath approvalSubscriptionPath
-}
-
-func approvalSubscriptionPreConsume(cfg approvalSubscriptionConfig) func(context.Context, event.APIClient, map[string]string) (func() error, error) {
+func approvalSubscriptionPreConsume(eventType, subscribePath 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")
 		}
 
-		eventType := string(cfg.eventType)
-		subscribePath := string(cfg.subscribePath)
 		subscriptionTypes, err := approvalSubscriptionTypes(eventType, params)
 		if err != nil {
 			return nil, err

events/approval/register.go

@@ -10,6 +10,7 @@ import (
 	"reflect"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 const (
@@ -40,12 +41,9 @@ func Keys() []event.KeyDefinition {
 			Schema: event.SchemaDef{
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(ApprovalInstanceStatusChangedV4Output{})},
 			},
-			Process: processApprovalInstanceStatusChanged,
-			PreConsume: approvalSubscriptionPreConsume(approvalSubscriptionConfig{
-				eventType:     eventTypeApprovalInstanceStatusChangedV4,
-				subscribePath: pathApprovalInstancesSubscription,
-			}),
-			Scopes: []string{"approval:instance:read"},
+			Process:    processApprovalInstanceStatusChanged,
+			PreConsume: approvalSubscriptionPreConsume(eventTypeApprovalInstanceStatusChangedV4, pathApprovalInstancesSubscription),
+			Scopes:     []string{"approval:instance:read"},
 			AuthTypes: []string{
 				"user",
 			},
@@ -60,12 +58,9 @@ func Keys() []event.KeyDefinition {
 			Schema: event.SchemaDef{
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(ApprovalTaskStatusChangedV4Output{})},
 			},
-			Process: processApprovalTaskStatusChanged,
-			PreConsume: approvalSubscriptionPreConsume(approvalSubscriptionConfig{
-				eventType:     eventTypeApprovalTaskStatusChangedV4,
-				subscribePath: pathApprovalTasksSubscription,
-			}),
-			Scopes: []string{"approval:task:read"},
+			Process:    processApprovalTaskStatusChanged,
+			PreConsume: approvalSubscriptionPreConsume(eventTypeApprovalTaskStatusChangedV4, pathApprovalTasksSubscription),
+			Scopes:     []string{"approval:task:read"},
 			AuthTypes: []string{
 				"user",
 			},
@@ -99,11 +94,6 @@ func processApprovalInstanceStatusChanged(_ context.Context, _ event.APIClient,
 		return nil, nil
 	}
 	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-		} `json:"header"`
 		Event struct {
 			ApprovalCode string          `json:"approval_code"`
 			InstanceCode string          `json:"instance_code"`
@@ -114,13 +104,13 @@ func processApprovalInstanceStatusChanged(_ context.Context, _ event.APIClient,
 		} `json:"event"`
 	}
 	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload so consumers still see the event
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
 	out := &ApprovalInstanceStatusChangedV4Output{
-		Type:         envelope.Header.EventType,
-		EventID:      envelope.Header.EventID,
-		Timestamp:    envelope.Header.CreateTime,
+		Type:         raw.EventType,
+		EventID:      raw.EventID,
+		Timestamp:    raw.SourceTime,
 		ApprovalCode: envelope.Event.ApprovalCode,
 		InstanceCode: envelope.Event.InstanceCode,
 		ExternalID:   envelope.Event.ExternalID,
@@ -128,9 +118,6 @@ func processApprovalInstanceStatusChanged(_ context.Context, _ event.APIClient,
 		OperateTime:  envelope.Event.OperateTime,
 		StartUser:    envelope.Event.StartUser,
 	}
-	if out.Type == "" {
-		out.Type = raw.EventType
-	}
 	return json.Marshal(out)
 }
 
@@ -139,11 +126,6 @@ func processApprovalTaskStatusChanged(_ context.Context, _ event.APIClient, raw
 		return nil, nil
 	}
 	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-		} `json:"header"`
 		Event struct {
 			ApprovalCode   string          `json:"approval_code"`
 			InstanceCode   string          `json:"instance_code"`
@@ -156,13 +138,13 @@ func processApprovalTaskStatusChanged(_ context.Context, _ event.APIClient, raw
 		} `json:"event"`
 	}
 	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload so consumers still see the event
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
 	out := &ApprovalTaskStatusChangedV4Output{
-		Type:           envelope.Header.EventType,
-		EventID:        envelope.Header.EventID,
-		Timestamp:      envelope.Header.CreateTime,
+		Type:           raw.EventType,
+		EventID:        raw.EventID,
+		Timestamp:      raw.SourceTime,
 		ApprovalCode:   envelope.Event.ApprovalCode,
 		InstanceCode:   envelope.Event.InstanceCode,
 		TaskID:         envelope.Event.TaskID,
@@ -172,8 +154,5 @@ func processApprovalTaskStatusChanged(_ context.Context, _ event.APIClient, raw
 		Status:         envelope.Event.Status,
 		OperateTime:    envelope.Event.OperateTime,
 	}
-	if out.Type == "" {
-		out.Type = raw.EventType
-	}
 	return json.Marshal(out)
 }

events/approval/register_test.go

@@ -14,6 +14,8 @@ import (
 
 	"github.com/larksuite/cli/errs"
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
+	"github.com/larksuite/cli/internal/event/processing"
 	"github.com/larksuite/cli/internal/event/schemas"
 )
 
@@ -255,10 +257,7 @@ func TestApprovalPreConsumeRegistersSubscriptionTypesWithoutCleanup(t *testing.T
 
 	for _, tc := range tests {
 		t.Run(tc.name, func(t *testing.T) {
-			pc := approvalSubscriptionPreConsume(approvalSubscriptionConfig{
-				eventType:     approvalEventType(tc.eventType),
-				subscribePath: approvalSubscriptionPath(tc.subscribePath),
-			})
+			pc := approvalSubscriptionPreConsume(tc.eventType, tc.subscribePath)
 			rt := &fakeAPIClient{}
 			cleanup, err := pc(context.Background(), rt, tc.params)
 			if err != nil {
@@ -297,9 +296,7 @@ func assertCall(t *testing.T, got recordedCall, wantMethod, wantPath string, wan
 
 func TestApprovalPreConsumeValidationErrors(t *testing.T) {
 	t.Run("nil runtime", func(t *testing.T) {
-		pc := approvalSubscriptionPreConsume(approvalSubscriptionConfig{
-			eventType: eventTypeApprovalInstanceStatusChangedV4,
-		})
+		pc := approvalSubscriptionPreConsume(eventTypeApprovalInstanceStatusChangedV4, "")
 		_, err := pc(context.Background(), nil, map[string]string{"subscription_type": approvalSubscriptionTypeInvolved})
 		if err == nil {
 			t.Fatal("expected nil runtime error")
@@ -312,9 +309,7 @@ func TestApprovalPreConsumeValidationErrors(t *testing.T) {
 
 	for _, raw := range []string{"BAD", "[]", `["INVOLVED_APPROVAL",3]`} {
 		t.Run("invalid subscription type "+raw, func(t *testing.T) {
-			pc := approvalSubscriptionPreConsume(approvalSubscriptionConfig{
-				eventType: eventTypeApprovalInstanceStatusChangedV4,
-			})
+			pc := approvalSubscriptionPreConsume(eventTypeApprovalInstanceStatusChangedV4, "")
 			cleanup, err := pc(context.Background(), &fakeAPIClient{}, map[string]string{"subscription_type": raw})
 			if err == nil {
 				t.Fatal("expected invalid subscription_type error")
@@ -338,10 +333,7 @@ func TestApprovalPreConsumeValidationErrors(t *testing.T) {
 	t.Run("partial registration failure reports registered and failed relation types", func(t *testing.T) {
 		upstream := errs.NewAPIError(errs.SubtypeServerError, "approval subscription API failed")
 		rt := &fakeAPIClient{err: upstream, errOnCall: 2}
-		pc := approvalSubscriptionPreConsume(approvalSubscriptionConfig{
-			eventType:     eventTypeApprovalTaskStatusChangedV4,
-			subscribePath: pathApprovalTasksSubscription,
-		})
+		pc := approvalSubscriptionPreConsume(eventTypeApprovalTaskStatusChangedV4, pathApprovalTasksSubscription)
 
 		cleanup, err := pc(context.Background(), rt, map[string]string{})
 		if err == nil {
@@ -553,7 +545,7 @@ func TestProcessApprovalStatusChangedUsesRawEventTypeFallback(t *testing.T) {
 	}
 }
 
-func TestProcessApprovalStatusChangedMalformedPayloadPassthrough(t *testing.T) {
+func TestProcessApprovalStatusChangedMalformedPayloadDrop(t *testing.T) {
 	for _, tc := range []struct {
 		name      string
 		eventType string
@@ -569,11 +561,11 @@ func TestProcessApprovalStatusChangedMalformedPayloadPassthrough(t *testing.T) {
 				Timestamp: time.Now(),
 			}
 			got, err := tc.process(context.Background(), nil, raw, nil)
-			if err != nil {
-				t.Fatalf("Process should swallow parse errors, got %v", err)
+			if !processing.IsDropMalformed(err) {
+				t.Fatalf("malformed payload must be dropped with a malformed marker, got err=%v", err)
 			}
-			if string(got) != "not json" {
-				t.Errorf("malformed fallback output = %q, want original bytes", string(got))
+			if got != nil {
+				t.Errorf("malformed payload must be dropped without output, got %q", string(got))
 			}
 		})
 	}
@@ -599,6 +591,30 @@ func TestProcessApprovalStatusChangedNilRaw(t *testing.T) {
 	}
 }
 
+// fillCanonicalFromHeader copies the payload envelope header metadata onto
+// the RawEvent canonical fields. Process handlers read event_id and
+// create_time from the RawEvent, which the consume pipeline fills from the
+// envelope header before dispatch; tests that hand-build a RawEvent must
+// mirror that so both views agree.
+func fillCanonicalFromHeader(t *testing.T, raw *event.RawEvent) {
+	t.Helper()
+	var envelope struct {
+		Header struct {
+			EventID    string `json:"event_id"`
+			EventType  string `json:"event_type"`
+			CreateTime string `json:"create_time"`
+		} `json:"header"`
+	}
+	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
+		t.Fatalf("parse envelope header: %v", err)
+	}
+	raw.EventID = envelope.Header.EventID
+	if envelope.Header.EventType != "" {
+		raw.EventType = envelope.Header.EventType
+	}
+	raw.SourceTime = envelope.Header.CreateTime
+}
+
 func runApprovalInstanceStatusChanged(t *testing.T, payload string) ApprovalInstanceStatusChangedV4Output {
 	t.Helper()
 	raw := &event.RawEvent{
@@ -606,6 +622,7 @@ func runApprovalInstanceStatusChanged(t *testing.T, payload string) ApprovalInst
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := processApprovalInstanceStatusChanged(context.Background(), nil, raw, nil)
 	if err != nil {
 		t.Fatalf("Process returned error: %v", err)
@@ -624,6 +641,7 @@ func runApprovalTaskStatusChanged(t *testing.T, payload string) ApprovalTaskStat
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := processApprovalTaskStatusChanged(context.Background(), nil, raw, nil)
 	if err != nil {
 		t.Fatalf("Process returned error: %v", err)
@@ -636,17 +654,16 @@ func runApprovalTaskStatusChanged(t *testing.T, payload string) ApprovalTaskStat
 }
 
 func TestApprovalKeysRegisterCleanly(t *testing.T) {
-	for _, key := range []string{eventTypeApprovalInstanceStatusChangedV4, eventTypeApprovalTaskStatusChangedV4} {
-		event.UnregisterKeyForTest(key)
-		t.Cleanup(func() { event.UnregisterKeyForTest(key) })
-	}
-
-	for _, def := range Keys() {
-		event.RegisterKey(def)
+	snap, err := catalog.Compile(Keys(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("catalog.Compile(Keys()): %v", err)
 	}
 	for _, key := range []string{eventTypeApprovalInstanceStatusChangedV4, eventTypeApprovalTaskStatusChangedV4} {
-		if _, ok := event.Lookup(key); !ok {
-			t.Fatalf("event.Lookup(%q) not registered", key)
+		if _, ok := snap.Resolve(key); !ok {
+			t.Fatalf("snap.Resolve(%q): key missing from compiled catalog", key)
 		}
 	}
 }

events/arch_test.go

@@ -0,0 +1,356 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+// Architecture gates for the events declaration layer.
+//
+// events/<domain> packages are declarations: EventKeys, payload shapes, and
+// processing hooks. Two kinds of rot would quietly destroy that role:
+//
+//  1. Importing command wiring, a transport host, or a concrete adapter turns
+//     declarations into another place where process and transport concerns
+//     accumulate, and drags the whole adapter tree into every binary that
+//     only wanted the catalog.
+//  2. Re-parsing the envelope header inside a domain duplicates the kernel's
+//     single header decode; the copies then drift apart the day the envelope
+//     evolves.
+//
+// These tests turn both into build breaks.
+package events_test
+
+import (
+	"go/ast"
+	"go/parser"
+	"go/token"
+	"io/fs"
+	"path/filepath"
+	"reflect"
+	"slices"
+	"sort"
+	"strconv"
+	"strings"
+	"testing"
+)
+
+const (
+	archModulePath          = "github.com/larksuite/cli"
+	archAdapterImportPrefix = archModulePath + "/internal/event/adapter"
+)
+
+// archProductionGoFiles returns every non-test .go file under root,
+// skipping testdata directories. Paths are relative to root.
+func archProductionGoFiles(t *testing.T, root string) []string {
+	t.Helper()
+	var files []string
+	err := filepath.WalkDir(root, func(path string, d fs.DirEntry, err error) error {
+		if err != nil {
+			return err
+		}
+		if d.IsDir() {
+			if d.Name() == "testdata" {
+				return fs.SkipDir
+			}
+			return nil
+		}
+		if !strings.HasSuffix(path, ".go") || strings.HasSuffix(path, "_test.go") {
+			return nil
+		}
+		files = append(files, path)
+		return nil
+	})
+	if err != nil {
+		t.Fatalf("walk %s: %v", root, err)
+	}
+	sort.Strings(files)
+	return files
+}
+
+// archForbiddenDomainImport reports why importPath is banned in events/, if
+// it is. Domains may use the kernel (internal/event, model, catalog,
+// processing, ...); they must never see the layers that host or transport
+// them.
+func archForbiddenDomainImport(importPath string) (reason string, banned bool) {
+	switch importPath {
+	case "github.com/spf13/cobra":
+		return "CLI framework; command wiring lives in cmd, a declaration that needs cobra has stopped being a declaration", true
+	case archModulePath + "/internal/event/bus":
+		return "bus is a host process; a domain importing its host inverts the dependency direction", true
+	case archModulePath + "/internal/event/consume":
+		return "consume is a host process; a domain importing its host inverts the dependency direction", true
+	}
+	if importPath == archAdapterImportPrefix || strings.HasPrefix(importPath, archAdapterImportPrefix+"/") {
+		return "concrete adapter; domains must stay transport-agnostic so any host can serve them", true
+	}
+	return "", false
+}
+
+// TestArchEventsImportRedline fails when any production file under events/
+// imports command wiring, an event host, or a concrete adapter. It keeps the
+// declaration layer linkable everywhere without pulling in transports.
+func TestArchEventsImportRedline(t *testing.T) {
+	files := archProductionGoFiles(t, ".")
+	if len(files) == 0 {
+		t.Fatal("scanned zero production files under events/ — the gate is idling; fix the walker before trusting any green run")
+	}
+	fset := token.NewFileSet()
+	for _, file := range files {
+		f, err := parser.ParseFile(fset, file, nil, parser.ImportsOnly)
+		if err != nil {
+			t.Fatalf("parse %s: %v", file, err)
+		}
+		for _, imp := range f.Imports {
+			path, err := strconv.Unquote(imp.Path.Value)
+			if err != nil {
+				t.Fatalf("unquote import in %s: %v", file, err)
+			}
+			if reason, banned := archForbiddenDomainImport(path); banned {
+				t.Errorf("%s imports %q: %s", filepath.ToSlash(file), path, reason)
+			}
+		}
+	}
+}
+
+// envelopeHeaderTags are the metadata fields the kernel decodes exactly once
+// from the envelope header. A domain that re-declares any of them inside a
+// json:"header" block is re-parsing the envelope instead of consuming the
+// kernel's decode — the duplicate drifts silently when the envelope changes.
+var envelopeHeaderTags = map[string]bool{
+	"event_id":    true,
+	"event_type":  true,
+	"create_time": true,
+	"app_id":      true,
+	"tenant_key":  true,
+}
+
+// headerReparseBaseline is the ratchet of pinned pre-existing residue, keyed
+// by file (relative to events/) with the header metadata tags it re-parses.
+// It is empty: every domain consumes the kernel-decoded header, so the gate
+// runs at zero tolerance. Never add an entry — new code must read the
+// kernel-decoded header instead of unmarshalling the envelope again.
+var headerReparseBaseline = map[string][]string{}
+
+type archHeaderReparse struct {
+	file  string // slash path relative to events/
+	line  int
+	field string // Go field name inside the header block
+	tag   string // offending json tag
+}
+
+// archJSONTagName extracts the json name (first comma segment) from a struct
+// field tag, or "" when absent.
+func archJSONTagName(field *ast.Field) string {
+	if field.Tag == nil {
+		return ""
+	}
+	raw, err := strconv.Unquote(field.Tag.Value)
+	if err != nil {
+		return ""
+	}
+	name, _, _ := strings.Cut(reflect.StructTag(raw).Get("json"), ",")
+	return name
+}
+
+// archNamedStructIndex maps type names declared in the given files (one
+// package) to their struct bodies, so a json:"header" field with a named
+// type still resolves.
+func archNamedStructIndex(files []*ast.File) map[string]*ast.StructType {
+	index := make(map[string]*ast.StructType)
+	for _, f := range files {
+		for _, decl := range f.Decls {
+			gen, ok := decl.(*ast.GenDecl)
+			if !ok || gen.Tok != token.TYPE {
+				continue
+			}
+			for _, spec := range gen.Specs {
+				ts, ok := spec.(*ast.TypeSpec)
+				if !ok {
+					continue
+				}
+				if st, ok := ts.Type.(*ast.StructType); ok {
+					index[ts.Name.Name] = st
+				}
+			}
+		}
+	}
+	return index
+}
+
+// archStructBody resolves expr to a struct body: inline struct types,
+// pointers to them, and named types declared in the same package.
+func archStructBody(expr ast.Expr, named map[string]*ast.StructType) *ast.StructType {
+	switch v := expr.(type) {
+	case *ast.StructType:
+		return v
+	case *ast.StarExpr:
+		return archStructBody(v.X, named)
+	case *ast.Ident:
+		return named[v.Name]
+	}
+	return nil
+}
+
+// archFindHeaderReparses flags every field inside a json:"header" struct
+// block whose json tag re-declares envelope header metadata. Fields outside
+// header blocks are never flagged: a domain body owning its own create_time
+// (e.g. a message's own timestamps) is legitimate.
+func archFindHeaderReparses(fset *token.FileSet, file *ast.File, relPath string, named map[string]*ast.StructType) []archHeaderReparse {
+	var found []archHeaderReparse
+	ast.Inspect(file, func(n ast.Node) bool {
+		st, ok := n.(*ast.StructType)
+		if !ok {
+			return true
+		}
+		for _, field := range st.Fields.List {
+			if archJSONTagName(field) != "header" {
+				continue
+			}
+			body := archStructBody(field.Type, named)
+			if body == nil {
+				continue
+			}
+			for _, hf := range body.Fields.List {
+				tag := archJSONTagName(hf)
+				if !envelopeHeaderTags[tag] {
+					continue
+				}
+				name := "(embedded)"
+				if len(hf.Names) > 0 {
+					parts := make([]string, len(hf.Names))
+					for i, ident := range hf.Names {
+						parts[i] = ident.Name
+					}
+					name = strings.Join(parts, ",")
+				}
+				found = append(found, archHeaderReparse{
+					file:  relPath,
+					line:  fset.Position(hf.Pos()).Line,
+					field: name,
+					tag:   tag,
+				})
+			}
+		}
+		return true
+	})
+	return found
+}
+
+// TestArchEventsNoHeaderMetadataReparse fails when a production file under
+// events/ declares a json:"header" struct block that re-parses envelope
+// header metadata, except for the pinned pre-existing residue in
+// headerReparseBaseline (which may only shrink).
+func TestArchEventsNoHeaderMetadataReparse(t *testing.T) {
+	files := archProductionGoFiles(t, ".")
+	if len(files) == 0 {
+		t.Fatal("scanned zero production files under events/ — the gate is idling; fix the walker before trusting any green run")
+	}
+
+	// Parse per directory so named header types declared in a sibling file
+	// of the same package still resolve.
+	byDir := make(map[string][]string)
+	for _, file := range files {
+		dir := filepath.Dir(file)
+		byDir[dir] = append(byDir[dir], file)
+	}
+	dirs := make([]string, 0, len(byDir))
+	for dir := range byDir {
+		dirs = append(dirs, dir)
+	}
+	sort.Strings(dirs)
+
+	fset := token.NewFileSet()
+	var violations []archHeaderReparse
+	for _, dir := range dirs {
+		astFiles := make([]*ast.File, 0, len(byDir[dir]))
+		for _, file := range byDir[dir] {
+			f, err := parser.ParseFile(fset, file, nil, parser.SkipObjectResolution)
+			if err != nil {
+				t.Fatalf("parse %s: %v", file, err)
+			}
+			astFiles = append(astFiles, f)
+		}
+		named := archNamedStructIndex(astFiles)
+		for i, f := range astFiles {
+			rel := filepath.ToSlash(byDir[dir][i])
+			violations = append(violations, archFindHeaderReparses(fset, f, rel, named)...)
+		}
+	}
+
+	seen := make(map[string]bool)
+	for _, v := range violations {
+		seen[v.file+"\x00"+v.tag] = true
+		if slices.Contains(headerReparseBaseline[v.file], v.tag) {
+			continue
+		}
+		t.Errorf("%s:%d field %s re-parses envelope header metadata %q inside a json:\"header\" block — consume the kernel-decoded header instead of unmarshalling the envelope again", v.file, v.line, v.field, v.tag)
+	}
+
+	// Stale baseline entries: once a file stops re-parsing a tag, its entry
+	// must go, otherwise the ratchet is wider than reality and the cleanup
+	// can silently regress.
+	for file, tags := range headerReparseBaseline {
+		for _, tag := range tags {
+			if !seen[file+"\x00"+tag] {
+				t.Errorf("stale baseline entry %s / %q: no code matches it anymore — delete the entry so the cleanup is locked in", file, tag)
+			}
+		}
+	}
+}
+
+// TestArchEventsHeaderReparseDetectorSelfCheck runs the header-reparse
+// detector on synthetic sources with a known violation count. If the
+// detector rots (tag parsing, named-type resolution, header matching), the
+// main gate would report green on a violating tree; this test makes that
+// failure mode loud.
+func TestArchEventsHeaderReparseDetectorSelfCheck(t *testing.T) {
+	parse := func(src string) []archHeaderReparse {
+		t.Helper()
+		fset := token.NewFileSet()
+		f, err := parser.ParseFile(fset, "synthetic.go", src, parser.SkipObjectResolution)
+		if err != nil {
+			t.Fatalf("parse synthetic source: %v", err)
+		}
+		files := []*ast.File{f}
+		return archFindHeaderReparses(fset, f, "synthetic.go", archNamedStructIndex(files))
+	}
+
+	const violating = `package synth
+
+type namedHeader struct {
+	AppID string ` + "`json:\"app_id\"`" + `
+}
+
+type envelope struct {
+	Header struct {
+		EventID   string ` + "`json:\"event_id\"`" + `
+		TenantKey string ` + "`json:\"tenant_key,omitempty\"`" + `
+		Custom    string ` + "`json:\"custom\"`" + `
+	} ` + "`json:\"header,omitempty\"`" + `
+	Named *namedHeader ` + "`json:\"header\"`" + `
+	Body  struct {
+		CreateTime string ` + "`json:\"create_time\"`" + `
+	} ` + "`json:\"body\"`" + `
+}
+`
+	got := parse(violating)
+	gotIDs := make([]string, len(got))
+	for i, v := range got {
+		gotIDs[i] = v.field + ":" + v.tag
+	}
+	sort.Strings(gotIDs)
+	wantIDs := []string{"AppID:app_id", "EventID:event_id", "TenantKey:tenant_key"}
+	if !slices.Equal(gotIDs, wantIDs) {
+		t.Fatalf("detector self-check: flagged %v, want exactly %v — the detector has drifted and the main gate cannot be trusted", gotIDs, wantIDs)
+	}
+
+	const clean = `package synth
+
+type output struct {
+	EventID string ` + "`json:\"event_id\"`" + `
+	Header  struct {
+		Custom string ` + "`json:\"custom\"`" + `
+	} ` + "`json:\"header\"`" + `
+}
+`
+	if got := parse(clean); len(got) != 0 {
+		t.Fatalf("detector self-check: clean synthetic source flagged %+v — the detector over-triggers and will produce false reds", got)
+	}
+}

events/catalog_helper_test.go

@@ -0,0 +1,26 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package events_test
+
+import (
+	"testing"
+
+	"github.com/larksuite/cli/events"
+	"github.com/larksuite/cli/internal/event/catalog"
+)
+
+// compileRealCatalog compiles the full shipped declaration set exactly as the
+// runtime does. Tests that used to walk the global registry iterate this
+// snapshot instead.
+func compileRealCatalog(t *testing.T) *catalog.Snapshot {
+	t.Helper()
+	snap, err := catalog.Compile(events.All(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("compile catalog: %v", err)
+	}
+	return snap
+}

events/compile_test.go

@@ -0,0 +1,93 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package events_test
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/larksuite/cli/events"
+	"github.com/larksuite/cli/internal/event/catalog"
+)
+
+// This gate lives in the events package because the catalog package cannot
+// import the declarations it compiles (that would be an import cycle). It is
+// the acceptance half of the compiler's own rejection tests: the real catalog
+// must compile — a compiler that rejects everything would also pass those.
+func TestCompile_RealCatalogCompilesCleanly(t *testing.T) {
+	snap, err := catalog.Compile(events.All(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("the shipped declarations must compile: %v", err)
+	}
+	if snap.Len() == 0 {
+		t.Fatal("the compiled catalog is empty; the gate proved nothing")
+	}
+	if snap.Len() != len(expectedKeys) {
+		t.Fatalf("compiled %d keys, frozen baseline has %d", snap.Len(), len(expectedKeys))
+	}
+	for _, want := range expectedKeys {
+		if _, ok := snap.Resolve(want); !ok {
+			t.Errorf("baseline key missing from the compiled catalog: %s", want)
+		}
+	}
+}
+
+// Every shipped key must satisfy its compiled output contract: a resolvable
+// non-empty schema, a jq root that matches the output mode, and normalized
+// delivery values. Golden files pin a few representative keys byte-for-byte;
+// this covers the whole catalog structurally.
+func TestOutputContract_HoldsForEveryKey(t *testing.T) {
+	snap, err := catalog.Compile(events.All(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatal(err)
+	}
+
+	checked := 0
+	for _, entry := range snap.Entries() {
+		checked++
+		d := entry.Descriptor()
+		out := entry.Output()
+
+		var parsed map[string]json.RawMessage
+		if err := json.Unmarshal(out.SchemaJSON, &parsed); err != nil || len(parsed) == 0 {
+			t.Errorf("%s: resolved schema must be a non-empty JSON object (err=%v)", d.Key, err)
+		}
+
+		switch out.Mode {
+		case catalog.OutputNative:
+			if out.JQRootPath != ".event" {
+				t.Errorf("%s: native keys deliver the V2 envelope; jq root must be .event, got %q", d.Key, out.JQRootPath)
+			}
+			if entry.Binding().Process != nil {
+				t.Errorf("%s: native keys must not carry a processor", d.Key)
+			}
+		case catalog.OutputProcessed:
+			if out.JQRootPath != "." {
+				t.Errorf("%s: processed keys deliver a flat shape; jq root must be ., got %q", d.Key, out.JQRootPath)
+			}
+			if entry.Binding().Process == nil {
+				t.Errorf("%s: processed keys must carry a processor", d.Key)
+			}
+		default:
+			t.Errorf("%s: unknown output mode %q", d.Key, out.Mode)
+		}
+
+		cap := entry.Capability()
+		if cap.BufferSize <= 0 || cap.BufferSize > catalog.MaxBufferSize || cap.Workers <= 0 {
+			t.Errorf("%s: delivery values must be normalized, got buffer=%d workers=%d", d.Key, cap.BufferSize, cap.Workers)
+		}
+		if d.Domain == "" {
+			t.Errorf("%s: descriptor domain must always be resolved", d.Key)
+		}
+	}
+	if checked == 0 {
+		t.Fatal("no entries were checked; the gate proved nothing")
+	}
+}

events/expected_keys_test.go

@@ -0,0 +1,59 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package events_test
+
+import (
+	"testing"
+)
+
+// expectedKeys is the frozen catalog baseline. Adding, removing, or renaming
+// an EventKey is a deliberate contract change: update this list in the same
+// commit and call the change out in the changelog.
+var expectedKeys = []string{
+	"application.bot.menu_v6",
+	"approval.instance.status_changed_v4",
+	"approval.task.status_changed_v4",
+	"board.whiteboard.updated_v1",
+	"card.action.trigger",
+	"im.chat.disbanded_v1",
+	"im.chat.member.bot.added_v1",
+	"im.chat.member.bot.deleted_v1",
+	"im.chat.member.user.added_v1",
+	"im.chat.member.user.deleted_v1",
+	"im.chat.member.user.withdrawn_v1",
+	"im.chat.updated_v1",
+	"im.message.message_read_v1",
+	"im.message.reaction.created_v1",
+	"im.message.reaction.deleted_v1",
+	"im.message.receive_v1",
+	"minutes.minute.generated_v1",
+	"task.task.update_user_access_v2",
+	"vc.meeting.participant_meeting_ended_v1",
+	"vc.meeting.participant_meeting_joined_v1",
+	"vc.meeting.participant_meeting_started_v1",
+	"vc.note.generated_v1",
+	"vc.recording.recording_ended_v1",
+	"vc.recording.recording_started_v1",
+	"vc.recording.recording_transcript_generated_v1",
+}
+
+func TestRegisteredKeys_MatchFrozenBaseline(t *testing.T) {
+	all := compileRealCatalog(t).Definitions()
+	if len(all) == 0 {
+		t.Fatal("no EventKeys registered; the gate scanned nothing")
+	}
+	got := make(map[string]bool, len(all))
+	for _, def := range all {
+		got[def.Key] = true
+	}
+	for _, want := range expectedKeys {
+		if !got[want] {
+			t.Errorf("expected EventKey missing from registry: %s", want)
+		}
+		delete(got, want)
+	}
+	for extra := range got {
+		t.Errorf("EventKey not in frozen baseline (update expectedKeys deliberately): %s", extra)
+	}
+}

events/im/card_action.go

@@ -9,6 +9,7 @@ import (
 	"strings"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 // CardActionTriggerOutput is the flattened shape for card.action.trigger.
@@ -35,11 +36,6 @@ type CardActionTriggerOutput struct {
 
 func processCardAction(ctx context.Context, rt event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
 	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-		} `json:"header"`
 		Event struct {
 			Operator struct {
 				OpenID string `json:"open_id"`
@@ -64,7 +60,7 @@ func processCardAction(ctx context.Context, rt event.APIClient, raw *event.RawEv
 		} `json:"event"`
 	}
 	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
 	actionValue := marshalToString(envelope.Event.Action.Value)
@@ -72,9 +68,9 @@ func processCardAction(ctx context.Context, rt event.APIClient, raw *event.RawEv
 	options := strings.Join(envelope.Event.Action.Options, ",")
 
 	out := &CardActionTriggerOutput{
-		Type:        envelope.Header.EventType,
-		EventID:     envelope.Header.EventID,
-		Timestamp:   envelope.Header.CreateTime,
+		Type:        raw.EventType,
+		EventID:     raw.EventID,
+		Timestamp:   raw.SourceTime,
 		OperatorID:  envelope.Event.Operator.OpenID,
 		MessageID:   envelope.Event.Context.OpenMessageID,
 		ChatID:      envelope.Event.Context.OpenChatID,

events/im/card_action_test.go

@@ -10,10 +10,11 @@ import (
 	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 func TestCardActionTriggerRegistered(t *testing.T) {
-	def, ok := event.Lookup("card.action.trigger")
+	def, ok := lookupCompiledDef(t, "card.action.trigger")
 	if !ok {
 		t.Fatal("card.action.trigger should be registered via Keys()")
 	}
@@ -243,11 +244,11 @@ func TestProcessCardAction_MalformedPayload(t *testing.T) {
 		Timestamp: time.Now(),
 	}
 	got, err := processCardAction(context.Background(), nil, raw, nil)
-	if err != nil {
-		t.Fatalf("Process should swallow parse errors, got %v", err)
+	if !processing.IsDropMalformed(err) {
+		t.Fatalf("malformed payload must be dropped with a malformed marker, got err=%v", err)
 	}
-	if string(got) != "not json" {
-		t.Errorf("malformed fallback output = %q, want original bytes", string(got))
+	if got != nil {
+		t.Errorf("malformed payload must be dropped without output, got %q", string(got))
 	}
 }
 
@@ -415,6 +416,7 @@ func runCardAction(t *testing.T, payload string, rt event.APIClient) CardActionT
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := processCardAction(context.Background(), rt, raw, nil)
 	if err != nil {
 		t.Fatalf("Process error: %v", err)

events/im/catalog_helper_test.go

@@ -0,0 +1,54 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package im
+
+import (
+	"encoding/json"
+	"testing"
+
+	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
+)
+
+// fillCanonicalFromHeader copies the payload envelope header metadata onto
+// the RawEvent canonical fields. Process handlers read event_id and
+// create_time from the RawEvent, which the consume pipeline fills from the
+// envelope header before dispatch; tests that hand-build a RawEvent must
+// mirror that so both views agree.
+func fillCanonicalFromHeader(t *testing.T, raw *event.RawEvent) {
+	t.Helper()
+	var envelope struct {
+		Header struct {
+			EventID    string `json:"event_id"`
+			EventType  string `json:"event_type"`
+			CreateTime string `json:"create_time"`
+		} `json:"header"`
+	}
+	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
+		t.Fatalf("parse envelope header: %v", err)
+	}
+	raw.EventID = envelope.Header.EventID
+	if envelope.Header.EventType != "" {
+		raw.EventType = envelope.Header.EventType
+	}
+	raw.SourceTime = envelope.Header.CreateTime
+}
+
+// lookupCompiledDef compiles this domain's declarations and resolves one key,
+// exactly as the runtime catalog would for a consumer.
+func lookupCompiledDef(t *testing.T, key string) (*event.KeyDefinition, bool) {
+	t.Helper()
+	snap, err := catalog.Compile(Keys(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("catalog.Compile(Keys()): %v", err)
+	}
+	entry, ok := snap.Resolve(key)
+	if !ok {
+		return nil, false
+	}
+	return entry.Definition(), true
+}

events/im/message_receive.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 	convertlib "github.com/larksuite/cli/shortcuts/im/convert_lib"
 )
 
@@ -40,11 +41,6 @@ type MentionOutput struct {
 
 func processImMessageReceive(_ context.Context, _ event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
 	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-		} `json:"header"`
 		Event struct {
 			Message struct {
 				MessageID   string        `json:"message_id"`
@@ -68,7 +64,7 @@ func processImMessageReceive(_ context.Context, _ event.APIClient, raw *event.Ra
 		} `json:"event"`
 	}
 	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload so consumers still see the event
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
 	msg := envelope.Event.Message
@@ -82,14 +78,14 @@ func processImMessageReceive(_ context.Context, _ event.APIClient, raw *event.Ra
 		})
 	}
 
-	timestamp := envelope.Header.CreateTime
+	timestamp := raw.SourceTime
 	if timestamp == "" {
 		timestamp = msg.CreateTime
 	}
 
 	out := &ImMessageReceiveOutput{
-		Type:        envelope.Header.EventType,
-		EventID:     envelope.Header.EventID,
+		Type:        raw.EventType,
+		EventID:     raw.EventID,
 		Timestamp:   timestamp,
 		ID:          msg.MessageID,
 		MessageID:   msg.MessageID,

events/im/message_receive_test.go

@@ -6,22 +6,15 @@ package im
 import (
 	"context"
 	"encoding/json"
-	"os"
 	"testing"
 	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
-func TestMain(m *testing.M) {
-	for _, k := range Keys() {
-		event.RegisterKey(k)
-	}
-	os.Exit(m.Run())
-}
-
 func TestIMKeys_ProcessedReceiveRegistered(t *testing.T) {
-	def, ok := event.Lookup("im.message.receive_v1")
+	def, ok := lookupCompiledDef(t, "im.message.receive_v1")
 	if !ok {
 		t.Fatal("im.message.receive_v1 should be registered via Keys()")
 	}
@@ -53,7 +46,7 @@ func TestIMKeys_NativeEventsRegistered(t *testing.T) {
 		"im.chat.disbanded_v1",
 	}
 	for _, k := range want {
-		def, ok := event.Lookup(k)
+		def, ok := lookupCompiledDef(t, k)
 		if !ok {
 			t.Errorf("%s should be registered via Keys()", k)
 			continue
@@ -232,11 +225,11 @@ func TestProcessImMessageReceive_MalformedPayload(t *testing.T) {
 		Timestamp: time.Now(),
 	}
 	got, err := processImMessageReceive(context.Background(), nil, raw, nil)
-	if err != nil {
-		t.Fatalf("Process should swallow parse errors, got %v", err)
+	if !processing.IsDropMalformed(err) {
+		t.Fatalf("malformed payload must be dropped with a malformed marker, got err=%v", err)
 	}
-	if string(got) != "not json" {
-		t.Errorf("malformed fallback output = %q, want original bytes", string(got))
+	if got != nil {
+		t.Errorf("malformed payload must be dropped without output, got %q", string(got))
 	}
 }
 
@@ -248,6 +241,7 @@ func runReceive(t *testing.T, payload string) ImMessageReceiveOutput {
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := processImMessageReceive(context.Background(), nil, raw, nil)
 	if err != nil {
 		t.Fatalf("Process error: %v", err)
@@ -267,6 +261,7 @@ func runReceiveMap(t *testing.T, payload string) map[string]interface{} {
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := processImMessageReceive(context.Background(), nil, raw, nil)
 	if err != nil {
 		t.Fatalf("Process error: %v", err)

events/internal/subscribeprep/subscribeprep.go

@@ -0,0 +1,51 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+// Package subscribeprep provides the shared PreConsume hook for EventKeys
+// whose server-side subscription is a plain event_type register/unregister
+// pair against fixed OAPI paths.
+package subscribeprep
+
+import (
+	"context"
+	"time"
+
+	"github.com/larksuite/cli/errs"
+	"github.com/larksuite/cli/internal/event/processing"
+)
+
+// CleanupTimeout bounds how long the unsubscribe call has to finish during
+// PreConsume cleanup so a stuck OAPI cannot block process shutdown.
+const CleanupTimeout = 5 * time.Second
+
+// Hook returns a PreConsume that subscribes eventType via subscribePath and
+// hands back a cleanup that unsubscribes it via unsubscribePath.
+func Hook(eventType, subscribePath, unsubscribePath string) func(context.Context, processing.APIClient, map[string]string) (func() error, error) {
+	return func(ctx context.Context, rt processing.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")
+		}
+		return SubscribeWithCleanup(ctx, rt, eventType, subscribePath, unsubscribePath)
+	}
+}
+
+// SubscribeWithCleanup calls the subscribe OAPI for eventType and returns a
+// cleanup that invokes the matching unsubscribe, bounded by CleanupTimeout.
+// rt must be non-nil; callers that validate their own params (e.g. to build
+// per-resource paths) run those checks first and then delegate here.
+func SubscribeWithCleanup(ctx context.Context, rt processing.APIClient, eventType, subscribePath, unsubscribePath string) (func() error, error) {
+	body := map[string]string{"event_type": eventType}
+	if _, err := rt.CallAPI(ctx, "POST", subscribePath, body); err != nil {
+		return nil, err
+	}
+
+	return func() error {
+		cleanupCtx, cancel := context.WithTimeout(context.Background(), CleanupTimeout)
+		defer cancel()
+		if _, err := rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body); err != nil {
+			return err
+		}
+		return nil
+	}, nil
+}

events/lint_test.go

@@ -9,11 +9,19 @@ import (
 	"testing"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
 	"github.com/larksuite/cli/internal/event/schemas"
 )
 
 func TestAllKeys_FieldOverridePointersResolve(t *testing.T) {
-	for _, def := range event.ListAll() {
+	snap, err := catalog.Compile(All(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("compile catalog: %v", err)
+	}
+	for _, def := range snap.Definitions() {
 		if len(def.Schema.FieldOverrides) == 0 {
 			continue
 		}

events/minutes/catalog_helper_test.go

@@ -0,0 +1,29 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package minutes
+
+import (
+	"testing"
+
+	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
+)
+
+// lookupCompiledDef compiles this domain's declarations and resolves one key,
+// exactly as the runtime catalog would for a consumer.
+func lookupCompiledDef(t *testing.T, key string) (*event.KeyDefinition, bool) {
+	t.Helper()
+	snap, err := catalog.Compile(Keys(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("catalog.Compile(Keys()): %v", err)
+	}
+	entry, ok := snap.Resolve(key)
+	if !ok {
+		return nil, false
+	}
+	return entry.Definition(), true
+}

events/minutes/minute_generated.go

@@ -10,6 +10,7 @@ import (
 	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 	"github.com/larksuite/cli/internal/validate"
 )
 
@@ -36,11 +37,6 @@ type MinutesMinuteGeneratedOutput struct {
 
 func processMinutesMinuteGenerated(ctx context.Context, rt event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
 	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-		} `json:"header"`
 		Event struct {
 			MinuteToken  string `json:"minute_token"`
 			MinuteSource struct {
@@ -50,18 +46,15 @@ func processMinutesMinuteGenerated(ctx context.Context, rt event.APIClient, raw
 		} `json:"event"`
 	}
 	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload so consumers still see the event
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
 	out := &MinutesMinuteGeneratedOutput{
-		Type:        envelope.Header.EventType,
-		EventID:     envelope.Header.EventID,
-		Timestamp:   envelope.Header.CreateTime,
+		Type:        raw.EventType,
+		EventID:     raw.EventID,
+		Timestamp:   raw.SourceTime,
 		MinuteToken: envelope.Event.MinuteToken,
 	}
-	if out.Type == "" {
-		out.Type = raw.EventType
-	}
 	if src := envelope.Event.MinuteSource; src.SourceType != "" || src.SourceEntityID != "" {
 		out.MinuteSource = &MinutesMinuteSourceOutput{
 			SourceType:     src.SourceType,

events/minutes/minute_generated_test.go

@@ -7,12 +7,12 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"os"
 	"reflect"
 	"testing"
 	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 	"github.com/larksuite/cli/internal/validate"
 )
 
@@ -35,17 +35,10 @@ func assertSubscriptionRequest(t *testing.T, gotBody any, wantEventType string)
 	}
 }
 
-func TestMain(m *testing.M) {
-	for _, k := range Keys() {
-		event.RegisterKey(k)
-	}
-	os.Exit(m.Run())
-}
-
 func TestMinutesKeys_ProcessedMinuteGeneratedRegistered(t *testing.T) {
 	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
 
-	def, ok := event.Lookup(eventTypeMinuteGenerated)
+	def, ok := lookupCompiledDef(t, eventTypeMinuteGenerated)
 	if !ok {
 		t.Fatalf("%s should be registered via Keys()", eventTypeMinuteGenerated)
 	}
@@ -274,7 +267,7 @@ func TestProcessMinutesMinuteGenerated_EmptyTitleExhaustsRetries(t *testing.T) {
 func TestMinutesMinuteGenerated_PreConsumeSubscriptionLifecycle(t *testing.T) {
 	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
 
-	def, ok := event.Lookup(eventTypeMinuteGenerated)
+	def, ok := lookupCompiledDef(t, eventTypeMinuteGenerated)
 	if !ok {
 		t.Fatalf("%s should be registered via Keys()", eventTypeMinuteGenerated)
 	}
@@ -326,14 +319,38 @@ func TestProcessMinutesMinuteGenerated_MalformedPayload(t *testing.T) {
 		Timestamp: time.Now(),
 	}
 	got, err := processMinutesMinuteGenerated(context.Background(), nil, raw, nil)
-	if err != nil {
-		t.Fatalf("Process should swallow parse errors, got %v", err)
+	if !processing.IsDropMalformed(err) {
+		t.Fatalf("malformed payload must be dropped with a malformed marker, got err=%v", err)
 	}
-	if string(got) != "not json" {
-		t.Errorf("malformed fallback output = %q, want original bytes", string(got))
+	if got != nil {
+		t.Errorf("malformed payload must be dropped without output, got %q", string(got))
 	}
 }
 
+// fillCanonicalFromHeader copies the payload envelope header metadata onto
+// the RawEvent canonical fields. Process handlers read event_id and
+// create_time from the RawEvent, which the consume pipeline fills from the
+// envelope header before dispatch; tests that hand-build a RawEvent must
+// mirror that so both views agree.
+func fillCanonicalFromHeader(t *testing.T, raw *event.RawEvent) {
+	t.Helper()
+	var envelope struct {
+		Header struct {
+			EventID    string `json:"event_id"`
+			EventType  string `json:"event_type"`
+			CreateTime string `json:"create_time"`
+		} `json:"header"`
+	}
+	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
+		t.Fatalf("parse envelope header: %v", err)
+	}
+	raw.EventID = envelope.Header.EventID
+	if envelope.Header.EventType != "" {
+		raw.EventType = envelope.Header.EventType
+	}
+	raw.SourceTime = envelope.Header.CreateTime
+}
+
 func runMinuteGenerated(t *testing.T, rt event.APIClient, payload string) MinutesMinuteGeneratedOutput {
 	t.Helper()
 	raw := &event.RawEvent{
@@ -341,6 +358,7 @@ func runMinuteGenerated(t *testing.T, rt event.APIClient, payload string) Minute
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := processMinutesMinuteGenerated(context.Background(), rt, raw, nil)
 	if err != nil {
 		t.Fatalf("Process error: %v", err)

events/minutes/preconsume.go

@@ -1,37 +0,0 @@
-// Copyright (c) 2026 Lark Technologies Pte. Ltd.
-// SPDX-License-Identifier: MIT
-
-package minutes
-
-import (
-	"context"
-	"time"
-
-	"github.com/larksuite/cli/errs"
-	"github.com/larksuite/cli/internal/event"
-)
-
-const cleanupTimeout = 5 * time.Second
-
-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")
-		}
-
-		body := map[string]string{"event_type": eventType}
-		if _, err := rt.CallAPI(ctx, "POST", subscribePath, body); err != nil {
-			return nil, err
-		}
-
-		return func() error {
-			cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
-			defer cancel()
-			if _, err := rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body); err != nil {
-				return err
-			}
-			return nil
-		}, nil
-	}
-}

events/minutes/register.go

@@ -7,6 +7,7 @@ package minutes
 import (
 	"reflect"
 
+	"github.com/larksuite/cli/events/internal/subscribeprep"
 	"github.com/larksuite/cli/internal/event"
 )
 
@@ -31,7 +32,7 @@ func Keys() []event.KeyDefinition {
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(MinutesMinuteGeneratedOutput{})},
 			},
 			Process:    processMinutesMinuteGenerated,
-			PreConsume: subscriptionPreConsume(eventTypeMinuteGenerated, pathMinuteSubscribe, pathMinuteUnsubscribe),
+			PreConsume: subscribeprep.Hook(eventTypeMinuteGenerated, pathMinuteSubscribe, pathMinuteUnsubscribe),
 			Scopes:     []string{"minutes:minutes.basic:read"},
 			AuthTypes: []string{
 				"user",

events/output_baseline_test.go

@@ -0,0 +1,461 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package events_test
+
+import (
+	"bytes"
+	"context"
+	"encoding/json"
+	"flag"
+	"maps"
+	"os"
+	"path/filepath"
+	"sort"
+	"testing"
+	"time"
+
+	event "github.com/larksuite/cli/internal/event"
+)
+
+var updateBaseline = flag.Bool("update-baseline", false,
+	"rewrite testdata/output_baseline.json with the current Processed EventKey outputs")
+
+// TestMain pins the process timezone to UTC before any test runs. Several
+// Process handlers format timestamps in the machine's local timezone
+// (e.g. meeting start/end times, recording event times), so without the pin
+// the snapshot would drift between machines in different timezones.
+func TestMain(m *testing.M) {
+	time.Local = time.UTC
+	os.Exit(m.Run())
+}
+
+const baselineSnapshotPath = "testdata/output_baseline.json"
+
+// wantProcessedKeys freezes how many registered EventKeys define Process
+// (im 2, vc 7, minutes 1, application 1, approval 2). The count assertion
+// keeps this test honest: if a Processed key is added or removed, the covered
+// output surface changed and the baseline would silently widen or narrow
+// without it. Update the count, the fixtures, and the snapshot together,
+// deliberately.
+const wantProcessedKeys = 13
+
+const (
+	baselineEventID    = "evt-baseline-001"
+	baselineCreateTime = "1700000000000" // 2023-11-14T22:13:20Z in milliseconds
+)
+
+// baselineFixture holds the minimal well-formed inputs for one Processed
+// EventKey: the business body placed under "event" in the V2 envelope, plus
+// any extra header fields the handler reads beyond event_id / event_type /
+// create_time. Every fixture must drive Process down its success path — no
+// drop, no malformed-payload passthrough.
+type baselineFixture struct {
+	extraHeader map[string]string
+	eventBody   string
+}
+
+// baselineFixtures maps every Processed EventKey to its synthetic input.
+// Field values are fixed constants so the resulting output is byte-stable.
+var baselineFixtures = map[string]baselineFixture{
+	"application.bot.menu_v6": {
+		extraHeader: map[string]string{
+			"app_id":     "cli-baseline-app",
+			"tenant_key": "tenant-baseline",
+		},
+		// 10-digit seconds timestamp: the handler normalizes it to milliseconds.
+		eventBody: `{
+			"event_key": "baseline_menu_key",
+			"timestamp": 1700000000,
+			"operator": {
+				"operator_id": {
+					"open_id": "ou-baseline-operator",
+					"union_id": "on-baseline-operator",
+					"user_id": "user-baseline-operator"
+				},
+				"operator_name": "Baseline Operator"
+			}
+		}`,
+	},
+	"approval.instance.status_changed_v4": {
+		eventBody: `{
+			"approval_code": "approval-code-baseline",
+			"instance_code": "instance-code-baseline",
+			"external_id": "external-id-baseline",
+			"status": "APPROVED",
+			"operate_time": "1700000000000",
+			"start_user": {
+				"open_id": "ou-baseline-starter",
+				"union_id": "on-baseline-starter",
+				"user_id": "user-baseline-starter"
+			}
+		}`,
+	},
+	"approval.task.status_changed_v4": {
+		eventBody: `{
+			"approval_code": "approval-code-baseline",
+			"instance_code": "instance-code-baseline",
+			"task_id": "task-id-baseline",
+			"external_id": "external-id-baseline",
+			"task_external_id": "task-external-id-baseline",
+			"status": "APPROVED",
+			"operate_time": "1700000000000",
+			"assigned_user": {
+				"open_id": "ou-baseline-assignee",
+				"union_id": "on-baseline-assignee",
+				"user_id": "user-baseline-assignee"
+			}
+		}`,
+	},
+	// The card handler fetches the card content through the API client using
+	// context.open_message_id; the fake client below serves that request.
+	"card.action.trigger": {
+		eventBody: `{
+			"operator": {"open_id": "ou-baseline-operator"},
+			"token": "card-token-baseline",
+			"host": "im_message",
+			"action": {
+				"tag": "button",
+				"value": {"key": "baseline"},
+				"name": "baseline_button",
+				"form_value": {"field": "value"},
+				"input_value": "baseline input",
+				"option": "opt-1",
+				"options": ["opt-1", "opt-2"],
+				"checked": true,
+				"timezone": "Asia/Shanghai"
+			},
+			"context": {
+				"open_message_id": "om-baseline-card",
+				"open_chat_id": "oc-baseline-chat"
+			}
+		}`,
+	},
+	// update_time differs from create_time so the handler emits both; the
+	// mention placeholder in content exercises mention rendering.
+	"im.message.receive_v1": {
+		eventBody: `{
+			"sender": {
+				"sender_type": "user",
+				"sender_id": {"open_id": "ou-baseline-sender"}
+			},
+			"message": {
+				"message_id": "om-baseline-msg",
+				"root_id": "om-baseline-root",
+				"parent_id": "om-baseline-parent",
+				"thread_id": "omt-baseline-thread",
+				"chat_id": "oc-baseline-chat",
+				"chat_type": "p2p",
+				"message_type": "text",
+				"create_time": "1699999999000",
+				"update_time": "1700000000500",
+				"content": "{\"text\":\"hello @_user_1\"}",
+				"mentions": [
+					{
+						"key": "@_user_1",
+						"id": {"open_id": "ou-baseline-mention"},
+						"name": "Baseline User"
+					}
+				]
+			}
+		}`,
+	},
+	// The minutes handler enriches the output with the minute title via the
+	// API client; the fake client answers with a non-empty title on the first
+	// call so no retry attempt is made.
+	"minutes.minute.generated_v1": {
+		eventBody: `{
+			"minute_token": "minute-token-baseline",
+			"minute_source": {
+				"source_type": "meeting",
+				"source_entity_id": "meeting-entity-baseline"
+			}
+		}`,
+	},
+	"vc.meeting.participant_meeting_started_v1": {
+		eventBody: `{
+			"meeting": {
+				"id": "meeting-id-baseline",
+				"topic": "Baseline meeting",
+				"meeting_no": "123456789",
+				"start_time": "1700000000",
+				"calendar_event_id": "calendar-event-baseline"
+			}
+		}`,
+	},
+	"vc.meeting.participant_meeting_joined_v1": {
+		eventBody: `{
+			"meeting": {
+				"id": "meeting-id-baseline",
+				"topic": "Baseline meeting",
+				"meeting_no": "123456789",
+				"start_time": "1700000000",
+				"calendar_event_id": "calendar-event-baseline"
+			}
+		}`,
+	},
+	"vc.meeting.participant_meeting_ended_v1": {
+		eventBody: `{
+			"meeting": {
+				"id": "meeting-id-baseline",
+				"topic": "Baseline meeting",
+				"meeting_no": "123456789",
+				"start_time": "1700000000",
+				"end_time": "1700000600",
+				"calendar_event_id": "calendar-event-baseline"
+			}
+		}`,
+	},
+	// The note handler enriches the output with document tokens via the API
+	// client; the fake client answers with both artifacts on the first call
+	// so no retry attempt is made.
+	"vc.note.generated_v1": {
+		eventBody: `{"note_id": "note-id-baseline"}`,
+	},
+	// Recording handlers only emit events whose source is recording_bean;
+	// anything else is dropped, which would break the success-path contract.
+	"vc.recording.recording_started_v1": {
+		eventBody: `{
+			"unique_key": "recording-key-baseline",
+			"source": "recording_bean"
+		}`,
+	},
+	"vc.recording.recording_transcript_generated_v1": {
+		eventBody: `{
+			"unique_key": "recording-key-baseline",
+			"source": "recording_bean",
+			"transcript_items": [
+				{
+					"speaker": {"user_name": "Baseline Speaker"},
+					"text": "baseline transcript text",
+					"start_time_ms": "1700000000000",
+					"end_time_ms": "1700000001000",
+					"sentence_id": "sentence-baseline-1"
+				}
+			]
+		}`,
+	},
+	"vc.recording.recording_ended_v1": {
+		eventBody: `{
+			"unique_key": "recording-key-baseline",
+			"source": "recording_bean"
+		}`,
+	},
+}
+
+// baselineAPIResponses maps request paths to canned success responses for the
+// handlers that call the API during Process. Every response satisfies the
+// handler on the first call, so retry loops never engage and no real network
+// or credentials are involved.
+var baselineAPIResponses = map[string]string{
+	"/open-apis/im/v1/messages/om-baseline-card?card_msg_content_type=user_card_content": `{
+		"code": 0,
+		"msg": "success",
+		"data": {
+			"items": [
+				{"body": {"content": "{\"header\":{\"title\":{\"tag\":\"plain_text\",\"content\":\"Baseline card\"}}}"}}
+			]
+		}
+	}`,
+	"/open-apis/vc/v1/notes/note-id-baseline": `{
+		"code": 0,
+		"msg": "success",
+		"data": {
+			"note": {
+				"artifacts": [
+					{"artifact_type": 1, "doc_token": "note-doc-token-baseline"},
+					{"artifact_type": 2, "doc_token": "verbatim-doc-token-baseline"}
+				],
+				"note_source": {
+					"source_type": "meeting",
+					"source_entity_id": "meeting-entity-baseline"
+				}
+			}
+		}
+	}`,
+	"/open-apis/minutes/v1/minutes/minute-token-baseline": `{
+		"code": 0,
+		"msg": "success",
+		"data": {
+			"minute": {"title": "Baseline minute title"}
+		}
+	}`,
+}
+
+// baselineAPIClient serves the canned responses above. An unexpected request
+// path fails the test immediately instead of returning an error, because
+// several handlers swallow API errors (or retry with delays) and would
+// silently produce a degraded output that gets frozen into the baseline.
+type baselineAPIClient struct {
+	t *testing.T
+}
+
+func (c *baselineAPIClient) CallAPI(_ context.Context, method, path string, _ any) (json.RawMessage, error) {
+	c.t.Helper()
+	resp, ok := baselineAPIResponses[path]
+	if !ok {
+		c.t.Fatalf("unexpected API call during Process: %s %s — add a canned response to baselineAPIResponses", method, path)
+	}
+	return json.RawMessage(resp), nil
+}
+
+// TestProcessedOutputBaseline runs every Processed EventKey against a fixed
+// well-formed synthetic payload and compares the outputs with the frozen
+// snapshot in testdata/output_baseline.json. Any change to what a Processed
+// key writes to stdout for a well-formed event shows up here as a named,
+// per-key diff. Run with -update-baseline to accept an intentional change.
+func TestProcessedOutputBaseline(t *testing.T) {
+	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
+
+	rt := &baselineAPIClient{t: t}
+	got := map[string]json.RawMessage{}
+	seenFixtures := map[string]bool{}
+
+	for _, def := range compileRealCatalog(t).Definitions() {
+		if def.Process == nil {
+			continue
+		}
+		fx, ok := baselineFixtures[def.Key]
+		if !ok {
+			t.Fatalf("Processed EventKey %q has no baseline fixture; add one to baselineFixtures, bump wantProcessedKeys, and regenerate with -update-baseline", def.Key)
+		}
+		seenFixtures[def.Key] = true
+
+		payload := buildBaselineEnvelope(t, def.EventType, fx)
+		// The canonical fields mirror the synthetic envelope header exactly,
+		// including any extra header fields, just as the consume pipeline
+		// guarantees for real events before Process runs.
+		raw := &event.RawEvent{
+			EventID:    baselineEventID,
+			EventType:  def.EventType,
+			SourceTime: baselineCreateTime,
+			AppID:      fx.extraHeader["app_id"],
+			TenantKey:  fx.extraHeader["tenant_key"],
+			Payload:    payload,
+			Timestamp:  time.Unix(1700000000, 0).UTC(),
+		}
+
+		out, err := def.Process(context.Background(), rt, raw, nil)
+		if err != nil {
+			t.Fatalf("%s: Process returned error on well-formed payload: %v", def.Key, err)
+		}
+		if out == nil {
+			t.Fatalf("%s: Process dropped a well-formed payload; the fixture must exercise the success path", def.Key)
+		}
+		if bytes.Equal(compactJSON(t, def.Key, out), compactJSON(t, def.Key, payload)) {
+			t.Fatalf("%s: Process returned the input unchanged; the fixture must exercise the success path, not the malformed-payload passthrough", def.Key)
+		}
+		got[def.Key] = out
+	}
+
+	if len(got) != wantProcessedKeys {
+		t.Fatalf("processed %d EventKeys, want exactly %d; a Processed key was added or removed — update baselineFixtures, wantProcessedKeys, and the snapshot together (keys run: %v)",
+			len(got), wantProcessedKeys, sortedKeys(got))
+	}
+	for key := range baselineFixtures {
+		if !seenFixtures[key] {
+			t.Fatalf("baseline fixture %q matches no registered Processed EventKey; remove it or fix the key name", key)
+		}
+	}
+
+	if *updateBaseline {
+		writeBaselineSnapshot(t, got)
+		return
+	}
+	compareBaselineSnapshot(t, got)
+}
+
+// buildBaselineEnvelope wraps a fixture body in the standard V2 event
+// envelope with fixed header values.
+func buildBaselineEnvelope(t *testing.T, eventType string, fx baselineFixture) json.RawMessage {
+	t.Helper()
+	header := map[string]string{
+		"event_id":    baselineEventID,
+		"event_type":  eventType,
+		"create_time": baselineCreateTime,
+	}
+	maps.Copy(header, fx.extraHeader)
+	headerJSON, err := json.Marshal(header)
+	if err != nil {
+		t.Fatalf("marshal envelope header: %v", err)
+	}
+	envelope := map[string]json.RawMessage{
+		"schema": json.RawMessage(`"2.0"`),
+		"header": headerJSON,
+		"event":  json.RawMessage(fx.eventBody),
+	}
+	payload, err := json.Marshal(envelope)
+	if err != nil {
+		t.Fatalf("marshal envelope for %s: %v", eventType, err)
+	}
+	return payload
+}
+
+func writeBaselineSnapshot(t *testing.T, got map[string]json.RawMessage) {
+	t.Helper()
+	// MarshalIndent sorts map keys, so the snapshot is deterministic.
+	data, err := json.MarshalIndent(got, "", "  ")
+	if err != nil {
+		t.Fatalf("marshal snapshot: %v", err)
+	}
+	data = append(data, '\n')
+	if err := os.MkdirAll(filepath.Dir(baselineSnapshotPath), 0o755); err != nil {
+		t.Fatalf("create testdata dir: %v", err)
+	}
+	if err := os.WriteFile(baselineSnapshotPath, data, 0o644); err != nil {
+		t.Fatalf("write snapshot: %v", err)
+	}
+	t.Logf("baseline snapshot rewritten: %s (%d keys)", baselineSnapshotPath, len(got))
+}
+
+func compareBaselineSnapshot(t *testing.T, got map[string]json.RawMessage) {
+	t.Helper()
+	data, err := os.ReadFile(baselineSnapshotPath)
+	if os.IsNotExist(err) {
+		t.Fatalf("baseline snapshot %s not found; generate it with: go test ./events/ -run TestProcessedOutput -update-baseline", baselineSnapshotPath)
+	}
+	if err != nil {
+		t.Fatalf("read snapshot: %v", err)
+	}
+	var want map[string]json.RawMessage
+	if err := json.Unmarshal(data, &want); err != nil {
+		t.Fatalf("snapshot %s is not valid JSON: %v", baselineSnapshotPath, err)
+	}
+
+	for _, key := range sortedKeys(want) {
+		if _, ok := got[key]; !ok {
+			t.Errorf("%s: present in snapshot but produced no output this run; if the key was removed on purpose, regenerate with -update-baseline", key)
+		}
+	}
+	for _, key := range sortedKeys(got) {
+		wantOut, ok := want[key]
+		if !ok {
+			t.Errorf("%s: produced output but missing from snapshot; regenerate with -update-baseline", key)
+			continue
+		}
+		gotC := compactJSON(t, key, got[key])
+		wantC := compactJSON(t, key, wantOut)
+		if !bytes.Equal(gotC, wantC) {
+			t.Errorf("%s: Processed output drifted from baseline\n  got:  %s\n  want: %s\nIf this change is intentional, regenerate with -update-baseline", key, gotC, wantC)
+		}
+	}
+}
+
+// compactJSON canonicalizes whitespace so comparisons are content-only.
+func compactJSON(t *testing.T, key string, raw json.RawMessage) []byte {
+	t.Helper()
+	var buf bytes.Buffer
+	if err := json.Compact(&buf, raw); err != nil {
+		t.Fatalf("%s: output is not valid JSON: %v\nraw=%s", key, err, string(raw))
+	}
+	return buf.Bytes()
+}
+
+func sortedKeys(m map[string]json.RawMessage) []string {
+	keys := make([]string, 0, len(m))
+	for k := range m {
+		keys = append(keys, k)
+	}
+	sort.Strings(keys)
+	return keys
+}

events/schema_closure_test.go

@@ -0,0 +1,75 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package events_test
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"testing"
+	"time"
+
+	event "github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
+)
+
+// closureAPIClient answers any API call with a benign error: a handler facing
+// a malformed payload must decide to drop before it ever needs the API.
+type closureAPIClient struct{}
+
+func (closureAPIClient) CallAPI(context.Context, string, string, any) (json.RawMessage, error) {
+	return nil, errors.New("no API access for malformed input")
+}
+
+// Every Processed EventKey declares an output schema; its stdout must stay
+// inside that schema. A payload that cannot be decoded therefore has exactly
+// one legal outcome: a malformed drop. Passing the raw envelope through would
+// hand consumers a shape the schema never described.
+//
+// Native keys (Process == nil) are exempt by contract: their declared output
+// is the raw envelope itself.
+func TestAllKeys_MalformedPayloadStaysSchemaClosed(t *testing.T) {
+	const wantProcessedKeys = 13
+
+	processed := 0
+	for _, def := range compileRealCatalog(t).Definitions() {
+		if def.Process == nil {
+			continue
+		}
+		processed++
+		out, err := safeProcess(t, def, json.RawMessage(`this is definitely not valid json {{{`))
+		if out != nil {
+			t.Errorf("%s: malformed payload produced stdout output; it must be dropped", def.Key)
+		}
+		if !processing.IsDropMalformed(err) {
+			t.Errorf("%s: malformed payload must be dropped with a malformed marker, got err=%v", def.Key, err)
+		}
+	}
+	if processed == 0 {
+		t.Fatal("no processed keys were exercised; the gate scanned nothing")
+	}
+	if processed != wantProcessedKeys {
+		t.Fatalf("exercised %d processed keys, want exactly %d; update the count when keys are deliberately added or removed", processed, wantProcessedKeys)
+	}
+}
+
+// safeProcess isolates a panicking handler to a per-key finding instead of
+// aborting the whole gate: a handler that dereferences before decoding is a
+// bug in that key, not a reason to stop scanning the rest.
+func safeProcess(t *testing.T, def *event.KeyDefinition, payload json.RawMessage) (out json.RawMessage, err error) {
+	t.Helper()
+	defer func() {
+		if r := recover(); r != nil {
+			t.Errorf("%s: Process panicked on malformed payload: %v", def.Key, r)
+			out, err = nil, nil
+		}
+	}()
+	raw := &event.RawEvent{
+		EventID:   "evt-closure-1",
+		EventType: def.EventType,
+		Payload:   payload,
+		Timestamp: time.Unix(0, 0),
+	}
+	return def.Process(context.Background(), closureAPIClient{}, raw, map[string]string{})
+}

events/schema_instance_test.go

@@ -0,0 +1,259 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package events_test
+
+import (
+	"bytes"
+	"encoding/json"
+	"fmt"
+	"os"
+	"sort"
+	"strconv"
+	"testing"
+
+	"github.com/larksuite/cli/internal/event/catalog"
+)
+
+// The output baseline freezes what every Processed key writes to stdout; the
+// compiled catalog promises a schema for the same bytes. This test closes the
+// loop between the two: every frozen output must be an instance of its key's
+// resolved schema, so a schema and its real output can never drift apart with
+// both sides individually green.
+//
+// The repository deliberately carries no JSON Schema validation dependency,
+// so validation is done by a minimal in-repo checker that covers exactly the
+// subset the catalog compiler emits (see validateValue). Any schema construct
+// outside that subset is a loud failure, never a silent pass.
+func TestProcessedBaselineOutputs_ConformToDeclaredSchemas(t *testing.T) {
+	snap := compileRealCatalog(t)
+	baseline := readBaselineSnapshot(t)
+
+	validated := 0
+	for _, entry := range snap.Entries() {
+		out := entry.Output()
+		if out.Mode != catalog.OutputProcessed {
+			continue
+		}
+		key := entry.Descriptor().Key
+		frozen, ok := baseline[key]
+		if !ok {
+			t.Errorf("%s: Processed key has no entry in %s; regenerate the baseline first", key, baselineSnapshotPath)
+			continue
+		}
+		schema := decodeSchemaNode(t, key, out.SchemaJSON)
+		instance := decodeInstance(t, key, frozen)
+		for _, problem := range validateValue("$", schema, instance) {
+			t.Errorf("%s: frozen output violates the declared schema: %s", key, problem)
+		}
+		validated++
+	}
+
+	// Idle detection, both directions: every Processed key was checked
+	// against a baseline entry, and no baseline entry escaped the check.
+	if validated == 0 {
+		t.Fatal("no Processed key was validated; the gate scanned nothing")
+	}
+	if validated != len(baseline) {
+		t.Fatalf("validated %d Processed keys but the baseline holds %d entries — a baseline entry matches no compiled Processed key (keys: %v)",
+			validated, len(baseline), sortedKeys(baseline))
+	}
+}
+
+// The validator itself must bite: an output tampered with in memory — an
+// undeclared field, a primitive type flip — has to produce findings,
+// otherwise a green conformance run proves nothing. The baseline file is
+// never modified.
+func TestSchemaInstanceValidator_BitesOnTamperedOutput(t *testing.T) {
+	const key = "im.message.receive_v1"
+	snap := compileRealCatalog(t)
+	entry, ok := snap.Resolve(key)
+	if !ok {
+		t.Fatalf("key %s is gone from the catalog; pick another Processed key for this self-check", key)
+	}
+	baseline := readBaselineSnapshot(t)
+	frozen, ok := baseline[key]
+	if !ok {
+		t.Fatalf("key %s has no baseline entry; the self-check needs a real frozen output", key)
+	}
+	schema := decodeSchemaNode(t, key, entry.Output().SchemaJSON)
+
+	// Control: the untampered output is conformant, so any finding below is
+	// caused by the tampering alone.
+	if problems := validateValue("$", schema, decodeInstance(t, key, frozen)); len(problems) != 0 {
+		t.Fatalf("control failed: the untampered output already has findings: %v", problems)
+	}
+
+	tampered, ok := decodeInstance(t, key, frozen).(map[string]any)
+	if !ok {
+		t.Fatalf("baseline output for %s is not a JSON object", key)
+	}
+	tampered["field_the_schema_never_declared"] = "smuggled"
+	if problems := validateValue("$", schema, tampered); len(problems) != 1 {
+		t.Errorf("an undeclared field must produce exactly one finding, got: %v", problems)
+	}
+
+	flipped, _ := decodeInstance(t, key, frozen).(map[string]any)
+	flipped["message_id"] = true // declared as a string
+	if problems := validateValue("$", schema, flipped); len(problems) != 1 {
+		t.Errorf("a primitive type flip must produce exactly one finding, got: %v", problems)
+	}
+}
+
+func readBaselineSnapshot(t *testing.T) map[string]json.RawMessage {
+	t.Helper()
+	data, err := os.ReadFile(baselineSnapshotPath)
+	if err != nil {
+		t.Fatalf("read %s: %v", baselineSnapshotPath, err)
+	}
+	var out map[string]json.RawMessage
+	if err := json.Unmarshal(data, &out); err != nil {
+		t.Fatalf("%s is not valid JSON: %v", baselineSnapshotPath, err)
+	}
+	return out
+}
+
+func decodeSchemaNode(t *testing.T, key string, raw json.RawMessage) map[string]any {
+	t.Helper()
+	var schema map[string]any
+	if err := json.Unmarshal(raw, &schema); err != nil {
+		t.Fatalf("%s: resolved schema is not a JSON object: %v", key, err)
+	}
+	return schema
+}
+
+// decodeInstance parses a frozen output with UseNumber so integer/number
+// checks see the literal digits instead of a lossy float64.
+func decodeInstance(t *testing.T, key string, raw json.RawMessage) any {
+	t.Helper()
+	dec := json.NewDecoder(bytes.NewReader(raw))
+	dec.UseNumber()
+	var v any
+	if err := dec.Decode(&v); err != nil {
+		t.Fatalf("%s: baseline output is not valid JSON: %v", key, err)
+	}
+	return v
+}
+
+// validateValue checks one instance value against one schema node and returns
+// the problems found. It implements only the subset the catalog compiler can
+// emit (schemas.FromType plus raw declarations shaped the same way):
+//
+//   - type object with properties: every instance field must be declared in
+//     properties and conform to its node; undeclared fields are errors.
+//     Absent declared fields are legal (handlers omit empty members).
+//   - type string / integer / number / boolean: the JSON value kind must
+//     match.
+//   - type array with items: every element must conform to items.
+//
+// description/format/enum annotations are metadata, not instance constraints
+// here. Any construct outside the subset — a missing or unknown type, an
+// object without properties, additionalProperties, an array without items —
+// is reported as a problem so the validator can only be extended
+// deliberately, never bypassed by a schema it does not understand.
+func validateValue(path string, schema map[string]any, value any) []string {
+	typ, ok := schema["type"].(string)
+	if !ok {
+		return []string{fmt.Sprintf("%s: schema node has no \"type\"; outside the minimal validator subset, extend the validator deliberately", path)}
+	}
+
+	switch typ {
+	case "object":
+		obj, ok := value.(map[string]any)
+		if !ok {
+			return []string{fmt.Sprintf("%s: schema declares object, output has %s", path, jsonKind(value))}
+		}
+		if _, has := schema["additionalProperties"]; has {
+			return []string{fmt.Sprintf("%s: schema uses additionalProperties; outside the minimal validator subset, extend the validator deliberately", path)}
+		}
+		props, ok := schema["properties"].(map[string]any)
+		if !ok {
+			return []string{fmt.Sprintf("%s: object schema without properties; outside the minimal validator subset, extend the validator deliberately", path)}
+		}
+		var problems []string
+		for _, field := range sortedFieldNames(obj) {
+			fieldPath := path + "." + field
+			node, declared := props[field]
+			if !declared {
+				problems = append(problems, fmt.Sprintf("%s: field is not declared in the schema properties", fieldPath))
+				continue
+			}
+			nodeObj, ok := node.(map[string]any)
+			if !ok {
+				problems = append(problems, fmt.Sprintf("%s: schema property is not an object", fieldPath))
+				continue
+			}
+			problems = append(problems, validateValue(fieldPath, nodeObj, obj[field])...)
+		}
+		return problems
+
+	case "string":
+		if _, ok := value.(string); !ok {
+			return []string{fmt.Sprintf("%s: schema declares string, output has %s", path, jsonKind(value))}
+		}
+	case "boolean":
+		if _, ok := value.(bool); !ok {
+			return []string{fmt.Sprintf("%s: schema declares boolean, output has %s", path, jsonKind(value))}
+		}
+	case "integer":
+		num, ok := value.(json.Number)
+		if !ok {
+			return []string{fmt.Sprintf("%s: schema declares integer, output has %s", path, jsonKind(value))}
+		}
+		if _, err := strconv.ParseInt(num.String(), 10, 64); err != nil {
+			return []string{fmt.Sprintf("%s: schema declares integer, output has non-integer number %s", path, num)}
+		}
+	case "number":
+		if _, ok := value.(json.Number); !ok {
+			return []string{fmt.Sprintf("%s: schema declares number, output has %s", path, jsonKind(value))}
+		}
+
+	case "array":
+		arr, ok := value.([]any)
+		if !ok {
+			return []string{fmt.Sprintf("%s: schema declares array, output has %s", path, jsonKind(value))}
+		}
+		items, ok := schema["items"].(map[string]any)
+		if !ok {
+			return []string{fmt.Sprintf("%s: array schema without items; outside the minimal validator subset, extend the validator deliberately", path)}
+		}
+		var problems []string
+		for i, elem := range arr {
+			problems = append(problems, validateValue(fmt.Sprintf("%s[%d]", path, i), items, elem)...)
+		}
+		return problems
+
+	default:
+		return []string{fmt.Sprintf("%s: schema type %q; outside the minimal validator subset, extend the validator deliberately", path, typ)}
+	}
+	return nil
+}
+
+// jsonKind names a decoded JSON value's kind for problem messages.
+func jsonKind(v any) string {
+	switch v.(type) {
+	case nil:
+		return "null"
+	case bool:
+		return "boolean"
+	case string:
+		return "string"
+	case json.Number:
+		return "number"
+	case []any:
+		return "array"
+	case map[string]any:
+		return "object"
+	default:
+		return fmt.Sprintf("%T", v)
+	}
+}
+
+func sortedFieldNames(obj map[string]any) []string {
+	names := make([]string, 0, len(obj))
+	for name := range obj {
+		names = append(names, name)
+	}
+	sort.Strings(names)
+	return names
+}

events/task/register_test.go

@@ -8,7 +8,7 @@ import (
 	"reflect"
 	"testing"
 
-	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
 	"github.com/larksuite/cli/internal/event/schemas"
 )
 
@@ -83,13 +83,14 @@ func TestTaskUpdateUserAccessSchemaAnnotations(t *testing.T) {
 
 func TestTaskUpdateUserAccessRegistersCleanly(t *testing.T) {
 	const key = eventTypeTaskUpdateUserAccessV2
-	event.UnregisterKeyForTest(key)
-	t.Cleanup(func() { event.UnregisterKeyForTest(key) })
-
-	for _, def := range Keys() {
-		event.RegisterKey(def)
+	snap, err := catalog.Compile(Keys(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("catalog.Compile(Keys()): %v", err)
 	}
-	if _, ok := event.Lookup(key); !ok {
-		t.Fatalf("event.Lookup(%q) not registered", key)
+	if _, ok := snap.Resolve(key); !ok {
+		t.Fatalf("snap.Resolve(%q): key missing from compiled catalog", key)
 	}
 }

events/testdata/output_baseline.json

@@ -0,0 +1,177 @@
+{
+  "application.bot.menu_v6": {
+    "type": "application.bot.menu_v6",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "app_id": "cli-baseline-app",
+    "tenant_key": "tenant-baseline",
+    "event_key": "baseline_menu_key",
+    "menu_timestamp": "1700000000000",
+    "operator_id": "ou-baseline-operator",
+    "operator_open_id": "ou-baseline-operator",
+    "operator_union_id": "on-baseline-operator",
+    "operator_user_id": "user-baseline-operator",
+    "operator_name": "Baseline Operator"
+  },
+  "approval.instance.status_changed_v4": {
+    "type": "approval.instance.status_changed_v4",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "approval_code": "approval-code-baseline",
+    "instance_code": "instance-code-baseline",
+    "external_id": "external-id-baseline",
+    "status": "APPROVED",
+    "operate_time": "1700000000000",
+    "start_user": {
+      "open_id": "ou-baseline-starter",
+      "union_id": "on-baseline-starter",
+      "user_id": "user-baseline-starter"
+    }
+  },
+  "approval.task.status_changed_v4": {
+    "type": "approval.task.status_changed_v4",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "approval_code": "approval-code-baseline",
+    "instance_code": "instance-code-baseline",
+    "task_id": "task-id-baseline",
+    "external_id": "external-id-baseline",
+    "task_external_id": "task-external-id-baseline",
+    "assigned_user": {
+      "open_id": "ou-baseline-assignee",
+      "union_id": "on-baseline-assignee",
+      "user_id": "user-baseline-assignee"
+    },
+    "status": "APPROVED",
+    "operate_time": "1700000000000"
+  },
+  "card.action.trigger": {
+    "type": "card.action.trigger",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "operator_id": "ou-baseline-operator",
+    "message_id": "om-baseline-card",
+    "chat_id": "oc-baseline-chat",
+    "host": "im_message",
+    "token": "card-token-baseline",
+    "action_tag": "button",
+    "action_value": "{\"key\":\"baseline\"}",
+    "action_name": "baseline_button",
+    "form_value": "{\"field\":\"value\"}",
+    "input_value": "baseline input",
+    "option": "opt-1",
+    "options": "opt-1,opt-2",
+    "checked": true,
+    "timezone": "Asia/Shanghai",
+    "card_content": "{\"header\":{\"title\":{\"tag\":\"plain_text\",\"content\":\"Baseline card\"}}}"
+  },
+  "im.message.receive_v1": {
+    "type": "im.message.receive_v1",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "id": "om-baseline-msg",
+    "message_id": "om-baseline-msg",
+    "create_time": "1699999999000",
+    "update_time": "1700000000500",
+    "chat_id": "oc-baseline-chat",
+    "chat_type": "p2p",
+    "message_type": "text",
+    "sender_id": "ou-baseline-sender",
+    "sender_type": "user",
+    "root_id": "om-baseline-root",
+    "thread_id": "omt-baseline-thread",
+    "reply_to": "om-baseline-parent",
+    "content": "hello @Baseline User",
+    "mentions": [
+      {
+        "key": "@_user_1",
+        "id": "ou-baseline-mention",
+        "name": "Baseline User"
+      }
+    ]
+  },
+  "minutes.minute.generated_v1": {
+    "type": "minutes.minute.generated_v1",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "minute_token": "minute-token-baseline",
+    "title": "Baseline minute title",
+    "minute_source": {
+      "source_type": "meeting",
+      "source_entity_id": "meeting-entity-baseline"
+    }
+  },
+  "vc.meeting.participant_meeting_ended_v1": {
+    "type": "vc.meeting.participant_meeting_ended_v1",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "meeting_id": "meeting-id-baseline",
+    "topic": "Baseline meeting",
+    "meeting_no": "123456789",
+    "start_time": "2023-11-14T22:13:20Z",
+    "end_time": "2023-11-14T22:23:20Z",
+    "calendar_event_id": "calendar-event-baseline"
+  },
+  "vc.meeting.participant_meeting_joined_v1": {
+    "type": "vc.meeting.participant_meeting_joined_v1",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "meeting_id": "meeting-id-baseline",
+    "topic": "Baseline meeting",
+    "meeting_no": "123456789",
+    "start_time": "2023-11-14T22:13:20Z",
+    "calendar_event_id": "calendar-event-baseline"
+  },
+  "vc.meeting.participant_meeting_started_v1": {
+    "type": "vc.meeting.participant_meeting_started_v1",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "meeting_id": "meeting-id-baseline",
+    "topic": "Baseline meeting",
+    "meeting_no": "123456789",
+    "start_time": "2023-11-14T22:13:20Z",
+    "calendar_event_id": "calendar-event-baseline"
+  },
+  "vc.note.generated_v1": {
+    "type": "vc.note.generated_v1",
+    "event_id": "evt-baseline-001",
+    "timestamp": "1700000000000",
+    "note_id": "note-id-baseline",
+    "note_token": "note-doc-token-baseline",
+    "verbatim_token": "verbatim-doc-token-baseline",
+    "note_source": {
+      "source_type": "meeting",
+      "source_entity_id": "meeting-entity-baseline"
+    }
+  },
+  "vc.recording.recording_ended_v1": {
+    "type": "vc.recording.recording_ended_v1",
+    "event_id": "evt-baseline-001",
+    "event_time": "2023-11-14T22:13:20Z",
+    "unique_key": "recording-key-baseline",
+    "source": "recording_bean"
+  },
+  "vc.recording.recording_started_v1": {
+    "type": "vc.recording.recording_started_v1",
+    "event_id": "evt-baseline-001",
+    "event_time": "2023-11-14T22:13:20Z",
+    "unique_key": "recording-key-baseline",
+    "source": "recording_bean"
+  },
+  "vc.recording.recording_transcript_generated_v1": {
+    "type": "vc.recording.recording_transcript_generated_v1",
+    "event_id": "evt-baseline-001",
+    "event_time": "2023-11-14T22:13:20Z",
+    "unique_key": "recording-key-baseline",
+    "source": "recording_bean",
+    "transcript_items": [
+      {
+        "speaker_name": "Baseline Speaker",
+        "text": "baseline transcript text",
+        "start_time": "2023-11-14T22:13:20Z",
+        "end_time": "2023-11-14T22:13:21Z",
+        "sentence_id": "sentence-baseline-1"
+      }
+    ]
+  }
+}

events/vc/catalog_helper_test.go

@@ -0,0 +1,29 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package vc
+
+import (
+	"testing"
+
+	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/catalog"
+)
+
+// lookupCompiledDef compiles this domain's declarations and resolves one key,
+// exactly as the runtime catalog would for a consumer.
+func lookupCompiledDef(t *testing.T, key string) (*event.KeyDefinition, bool) {
+	t.Helper()
+	snap, err := catalog.Compile(Keys(), catalog.StrategyRefs{
+		catalog.StrategyNone,
+		catalog.StrategyLegacyPreConsume,
+	})
+	if err != nil {
+		t.Fatalf("catalog.Compile(Keys()): %v", err)
+	}
+	entry, ok := snap.Resolve(key)
+	if !ok {
+		return nil, false
+	}
+	return entry.Definition(), true
+}

events/vc/internal_helpers.go

@@ -0,0 +1,62 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package vc
+
+import (
+	"encoding/json"
+	"strconv"
+	"time"
+
+	"github.com/larksuite/cli/internal/event"
+)
+
+// recordingBeanSource is the only recording source the vc.recording.* keys
+// emit; events carrying any other source are silently filtered out.
+const recordingBeanSource = "recording_bean"
+
+// recordingBeanEventBody is the shared {"event": ...} body for
+// recording_started and recording_ended, whose payloads carry identical fields.
+type recordingBeanEventBody struct {
+	UniqueKey string `json:"unique_key"`
+	Source    string `json:"source"`
+}
+
+// decodeEventBody unmarshals the {"event": ...} envelope of raw and returns
+// the decoded body; ok is false when the payload does not decode.
+func decodeEventBody[T any](raw *event.RawEvent) (T, bool) {
+	var envelope struct {
+		Event T `json:"event"`
+	}
+	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
+		var zero T
+		return zero, false
+	}
+	return envelope.Event, true
+}
+
+// millisToLocalRFC3339 converts a unix-millisecond timestamp string to
+// RFC3339 in the local timezone; empty or non-numeric input yields "".
+func millisToLocalRFC3339(raw string) string {
+	if raw == "" {
+		return ""
+	}
+	millis, err := strconv.ParseInt(raw, 10, 64)
+	if err != nil {
+		return ""
+	}
+	return time.UnixMilli(millis).Local().Format(time.RFC3339)
+}
+
+// unixSecondsToLocalRFC3339 converts a unix-second timestamp string to
+// RFC3339 in the local timezone; empty or non-numeric input yields "".
+func unixSecondsToLocalRFC3339(raw string) string {
+	if raw == "" {
+		return ""
+	}
+	secs, err := strconv.ParseInt(raw, 10, 64)
+	if err != nil {
+		return ""
+	}
+	return time.Unix(secs, 0).Local().Format(time.RFC3339)
+}

events/vc/note_generated.go

@@ -11,6 +11,7 @@ import (
 
 	"github.com/larksuite/cli/errs"
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 	"github.com/larksuite/cli/internal/validate"
 )
 
@@ -42,28 +43,20 @@ type VCNoteGeneratedOutput struct {
 
 func processVCNoteGenerated(ctx context.Context, rt event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
 	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-		} `json:"header"`
 		Event struct {
 			NoteID string `json:"note_id"`
 		} `json:"event"`
 	}
 	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload so consumers still see the event
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
 	out := &VCNoteGeneratedOutput{
-		Type:      envelope.Header.EventType,
-		EventID:   envelope.Header.EventID,
-		Timestamp: envelope.Header.CreateTime,
+		Type:      raw.EventType,
+		EventID:   raw.EventID,
+		Timestamp: raw.SourceTime,
 		NoteID:    envelope.Event.NoteID,
 	}
-	if out.Type == "" {
-		out.Type = raw.EventType
-	}
 
 	if rt != nil && out.NoteID != "" {
 		fillVCNoteGeneratedDetails(ctx, rt, out)

events/vc/note_generated_test.go

@@ -10,12 +10,13 @@ import (
 	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 func TestVCKeys_ProcessedNoteGeneratedRegistered(t *testing.T) {
 	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
 
-	def, ok := event.Lookup(eventTypeNoteGenerated)
+	def, ok := lookupCompiledDef(t, eventTypeNoteGenerated)
 	if !ok {
 		t.Fatalf("%s should be registered via Keys()", eventTypeNoteGenerated)
 	}
@@ -113,7 +114,7 @@ func TestProcessVCNoteGenerated(t *testing.T) {
 func TestVCNoteGenerated_PreConsumeSubscriptionLifecycle(t *testing.T) {
 	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
 
-	def, ok := event.Lookup(eventTypeNoteGenerated)
+	def, ok := lookupCompiledDef(t, eventTypeNoteGenerated)
 	if !ok {
 		t.Fatalf("%s should be registered via Keys()", eventTypeNoteGenerated)
 	}
@@ -301,11 +302,11 @@ func TestProcessVCNoteGenerated_MalformedPayload(t *testing.T) {
 		Timestamp: time.Now(),
 	}
 	got, err := processVCNoteGenerated(context.Background(), nil, raw, nil)
-	if err != nil {
-		t.Fatalf("Process should swallow parse errors, got %v", err)
+	if !processing.IsDropMalformed(err) {
+		t.Fatalf("malformed payload must be dropped with a malformed marker, got err=%v", err)
 	}
-	if string(got) != "not json" {
-		t.Errorf("malformed fallback output = %q, want original bytes", string(got))
+	if got != nil {
+		t.Errorf("malformed payload must be dropped without output, got %q", string(got))
 	}
 }
 
@@ -316,6 +317,7 @@ func runNoteGenerated(t *testing.T, rt event.APIClient, payload string) VCNoteGe
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := processVCNoteGenerated(context.Background(), rt, raw, nil)
 	if err != nil {
 		t.Fatalf("Process error: %v", err)

events/vc/participant_meeting_ended.go

@@ -6,10 +6,9 @@ package vc
 import (
 	"context"
 	"encoding/json"
-	"strconv"
-	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 // VCParticipantMeetingEndedOutput is the flattened shape for vc.meeting.participant_meeting_ended_v1.
@@ -25,33 +24,28 @@ type VCParticipantMeetingEndedOutput struct {
 	CalendarEventID string `json:"calendar_event_id,omitempty" desc:"Calendar event ID associated with the meeting"`
 }
 
+type participantMeetingEndedEvent struct {
+	Meeting struct {
+		ID              string `json:"id"`
+		Topic           string `json:"topic"`
+		MeetingNo       string `json:"meeting_no"`
+		StartTime       string `json:"start_time"`
+		EndTime         string `json:"end_time"`
+		CalendarEventID string `json:"calendar_event_id"`
+	} `json:"meeting"`
+}
+
 func processVCParticipantMeetingEnded(_ context.Context, _ event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
-	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-		} `json:"header"`
-		Event struct {
-			Meeting struct {
-				ID              string `json:"id"`
-				Topic           string `json:"topic"`
-				MeetingNo       string `json:"meeting_no"`
-				StartTime       string `json:"start_time"`
-				EndTime         string `json:"end_time"`
-				CalendarEventID string `json:"calendar_event_id"`
-			} `json:"meeting"`
-		} `json:"event"`
-	}
-	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload so consumers still see the event
+	body, ok := decodeEventBody[participantMeetingEndedEvent](raw)
+	if !ok {
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
-	meeting := envelope.Event.Meeting
+	meeting := body.Meeting
 	out := &VCParticipantMeetingEndedOutput{
-		Type:            envelope.Header.EventType,
-		EventID:         envelope.Header.EventID,
-		Timestamp:       envelope.Header.CreateTime,
+		Type:            raw.EventType,
+		EventID:         raw.EventID,
+		Timestamp:       raw.SourceTime,
 		MeetingID:       meeting.ID,
 		Topic:           meeting.Topic,
 		MeetingNo:       meeting.MeetingNo,
@@ -59,19 +53,5 @@ func processVCParticipantMeetingEnded(_ context.Context, _ event.APIClient, raw
 		EndTime:         unixSecondsToLocalRFC3339(meeting.EndTime),
 		CalendarEventID: meeting.CalendarEventID,
 	}
-	if out.Type == "" {
-		out.Type = raw.EventType
-	}
 	return json.Marshal(out)
 }
-
-func unixSecondsToLocalRFC3339(raw string) string {
-	if raw == "" {
-		return ""
-	}
-	secs, err := strconv.ParseInt(raw, 10, 64)
-	if err != nil {
-		return ""
-	}
-	return time.Unix(secs, 0).Local().Format(time.RFC3339)
-}

events/vc/participant_meeting_ended_test.go

@@ -6,24 +6,17 @@ package vc
 import (
 	"context"
 	"encoding/json"
-	"os"
 	"testing"
 	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
-func TestMain(m *testing.M) {
-	for _, k := range Keys() {
-		event.RegisterKey(k)
-	}
-	os.Exit(m.Run())
-}
-
 func TestVCKeys_ProcessedMeetingEndedRegistered(t *testing.T) {
 	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
 
-	def, ok := event.Lookup(eventTypeMeetingEnded)
+	def, ok := lookupCompiledDef(t, eventTypeMeetingEnded)
 	if !ok {
 		t.Fatalf("%s should be registered via Keys()", eventTypeMeetingEnded)
 	}
@@ -130,18 +123,18 @@ func TestProcessVCParticipantMeetingEnded_MalformedPayload(t *testing.T) {
 		Timestamp: time.Now(),
 	}
 	got, err := processVCParticipantMeetingEnded(context.Background(), nil, raw, nil)
-	if err != nil {
-		t.Fatalf("Process should swallow parse errors, got %v", err)
+	if !processing.IsDropMalformed(err) {
+		t.Fatalf("malformed payload must be dropped with a malformed marker, got err=%v", err)
 	}
-	if string(got) != "not json" {
-		t.Errorf("malformed fallback output = %q, want original bytes", string(got))
+	if got != nil {
+		t.Errorf("malformed payload must be dropped without output, got %q", string(got))
 	}
 }
 
 func TestVCParticipantMeetingEnded_PreConsumeSubscriptionLifecycle(t *testing.T) {
 	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
 
-	def, ok := event.Lookup("vc.meeting.participant_meeting_ended_v1")
+	def, ok := lookupCompiledDef(t, "vc.meeting.participant_meeting_ended_v1")
 	if !ok {
 		t.Fatal("vc.meeting.participant_meeting_ended_v1 should be registered via Keys()")
 	}
@@ -191,6 +184,7 @@ func runMeetingEnded(t *testing.T, payload string) VCParticipantMeetingEndedOutp
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := processVCParticipantMeetingEnded(context.Background(), nil, raw, nil)
 	if err != nil {
 		t.Fatalf("Process error: %v", err)

events/vc/participant_meeting_joined.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 // VCParticipantMeetingJoinedOutput is the flattened shape for vc.meeting.participant_meeting_joined_v1.
@@ -22,41 +23,33 @@ type VCParticipantMeetingJoinedOutput struct {
 	CalendarEventID string `json:"calendar_event_id,omitempty" desc:"Calendar event ID associated with the meeting"`
 }
 
+type participantMeetingJoinedEvent struct {
+	Meeting struct {
+		ID              string `json:"id"`
+		Topic           string `json:"topic"`
+		MeetingNo       string `json:"meeting_no"`
+		StartTime       string `json:"start_time"`
+		EndTime         string `json:"end_time"`
+		CalendarEventID string `json:"calendar_event_id"`
+	} `json:"meeting"`
+}
+
 func processVCParticipantMeetingJoined(_ context.Context, _ event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
-	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-		} `json:"header"`
-		Event struct {
-			Meeting struct {
-				ID              string `json:"id"`
-				Topic           string `json:"topic"`
-				MeetingNo       string `json:"meeting_no"`
-				StartTime       string `json:"start_time"`
-				EndTime         string `json:"end_time"`
-				CalendarEventID string `json:"calendar_event_id"`
-			} `json:"meeting"`
-		} `json:"event"`
-	}
-	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload so consumers still see the event
+	body, ok := decodeEventBody[participantMeetingJoinedEvent](raw)
+	if !ok {
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
-	meeting := envelope.Event.Meeting
+	meeting := body.Meeting
 	out := &VCParticipantMeetingJoinedOutput{
-		Type:            envelope.Header.EventType,
-		EventID:         envelope.Header.EventID,
-		Timestamp:       envelope.Header.CreateTime,
+		Type:            raw.EventType,
+		EventID:         raw.EventID,
+		Timestamp:       raw.SourceTime,
 		MeetingID:       meeting.ID,
 		Topic:           meeting.Topic,
 		MeetingNo:       meeting.MeetingNo,
 		StartTime:       unixSecondsToLocalRFC3339(meeting.StartTime),
 		CalendarEventID: meeting.CalendarEventID,
 	}
-	if out.Type == "" {
-		out.Type = raw.EventType
-	}
 	return json.Marshal(out)
 }

events/vc/participant_meeting_lifecycle_test.go

@@ -11,6 +11,7 @@ import (
 	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 func TestVCKeys_ProcessedMeetingLifecycleRegistered(t *testing.T) {
@@ -24,7 +25,7 @@ func TestVCKeys_ProcessedMeetingLifecycleRegistered(t *testing.T) {
 		{eventTypeMeetingJoined, reflect.TypeOf(VCParticipantMeetingJoinedOutput{})},
 	} {
 		t.Run(tc.eventType, func(t *testing.T) {
-			def, ok := event.Lookup(tc.eventType)
+			def, ok := lookupCompiledDef(t, tc.eventType)
 			if !ok {
 				t.Fatalf("%s should be registered via Keys()", tc.eventType)
 			}
@@ -193,11 +194,11 @@ func TestProcessVCParticipantMeetingLifecycle_MalformedPayload(t *testing.T) {
 				Timestamp: time.Now(),
 			}
 			got, err := tc.process(context.Background(), nil, raw, nil)
-			if err != nil {
-				t.Fatalf("Process should swallow parse errors, got %v", err)
+			if !processing.IsDropMalformed(err) {
+				t.Fatalf("malformed payload must be dropped with a malformed marker, got err=%v", err)
 			}
-			if string(got) != "not json" {
-				t.Errorf("malformed fallback output = %q, want original bytes", string(got))
+			if got != nil {
+				t.Errorf("malformed payload must be dropped without output, got %q", string(got))
 			}
 		})
 	}
@@ -208,7 +209,7 @@ func TestVCParticipantMeetingLifecycle_PreConsumeSubscriptionLifecycle(t *testin
 
 	for _, eventType := range []string{eventTypeMeetingStarted, eventTypeMeetingJoined} {
 		t.Run(eventType, func(t *testing.T) {
-			def, ok := event.Lookup(eventType)
+			def, ok := lookupCompiledDef(t, eventType)
 			if !ok {
 				t.Fatalf("%s should be registered via Keys()", eventType)
 			}
@@ -273,6 +274,7 @@ func runMeetingLifecycleRaw(t *testing.T, eventType string, process event.Proces
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := process(context.Background(), nil, raw, nil)
 	if err != nil {
 		t.Fatalf("Process error: %v", err)

events/vc/participant_meeting_started.go

@@ -8,6 +8,7 @@ import (
 	"encoding/json"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 // VCParticipantMeetingStartedOutput is the flattened shape for vc.meeting.participant_meeting_started_v1.
@@ -22,40 +23,32 @@ type VCParticipantMeetingStartedOutput struct {
 	CalendarEventID string `json:"calendar_event_id,omitempty" desc:"Calendar event ID associated with the meeting"`
 }
 
+type participantMeetingStartedEvent struct {
+	Meeting struct {
+		ID              string `json:"id"`
+		Topic           string `json:"topic"`
+		MeetingNo       string `json:"meeting_no"`
+		StartTime       string `json:"start_time"`
+		CalendarEventID string `json:"calendar_event_id"`
+	} `json:"meeting"`
+}
+
 func processVCParticipantMeetingStarted(_ context.Context, _ event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
-	var envelope struct {
-		Header struct {
-			EventID    string `json:"event_id"`
-			EventType  string `json:"event_type"`
-			CreateTime string `json:"create_time"`
-		} `json:"header"`
-		Event struct {
-			Meeting struct {
-				ID              string `json:"id"`
-				Topic           string `json:"topic"`
-				MeetingNo       string `json:"meeting_no"`
-				StartTime       string `json:"start_time"`
-				CalendarEventID string `json:"calendar_event_id"`
-			} `json:"meeting"`
-		} `json:"event"`
-	}
-	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return raw.Payload, nil //nolint:nilerr // passthrough on malformed payload so consumers still see the event
+	body, ok := decodeEventBody[participantMeetingStartedEvent](raw)
+	if !ok {
+		return nil, processing.DropMalformed(raw.EventType)
 	}
 
-	meeting := envelope.Event.Meeting
+	meeting := body.Meeting
 	out := &VCParticipantMeetingStartedOutput{
-		Type:            envelope.Header.EventType,
-		EventID:         envelope.Header.EventID,
-		Timestamp:       envelope.Header.CreateTime,
+		Type:            raw.EventType,
+		EventID:         raw.EventID,
+		Timestamp:       raw.SourceTime,
 		MeetingID:       meeting.ID,
 		Topic:           meeting.Topic,
 		MeetingNo:       meeting.MeetingNo,
 		StartTime:       unixSecondsToLocalRFC3339(meeting.StartTime),
 		CalendarEventID: meeting.CalendarEventID,
 	}
-	if out.Type == "" {
-		out.Type = raw.EventType
-	}
 	return json.Marshal(out)
 }

events/vc/preconsume.go

@@ -1,37 +0,0 @@
-// Copyright (c) 2026 Lark Technologies Pte. Ltd.
-// SPDX-License-Identifier: MIT
-
-package vc
-
-import (
-	"context"
-	"time"
-
-	"github.com/larksuite/cli/errs"
-	"github.com/larksuite/cli/internal/event"
-)
-
-const cleanupTimeout = 5 * time.Second
-
-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")
-		}
-
-		body := map[string]string{"event_type": eventType}
-		if _, err := rt.CallAPI(ctx, "POST", subscribePath, body); err != nil {
-			return nil, err
-		}
-
-		return func() error {
-			cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
-			defer cancel()
-			if _, err := rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body); err != nil {
-				return err
-			}
-			return nil
-		}, nil
-	}
-}

events/vc/recording_ended.go

@@ -6,10 +6,9 @@ package vc
 import (
 	"context"
 	"encoding/json"
-	"strconv"
-	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 // VCRecordingEndedOutput is the flattened shape for vc.recording.recording_ended_v1.
@@ -21,64 +20,20 @@ type VCRecordingEndedOutput struct {
 	Source    string `json:"source,omitempty"     desc:"Recording source; always recording_bean"`
 }
 
-type recordingEndedEnvelope struct {
-	Header struct {
-		EventID    string `json:"event_id"`
-		EventType  string `json:"event_type"`
-		CreateTime string `json:"create_time"`
-	} `json:"header"`
-	Event recordingEndedEvent `json:"event"`
-}
-
-type recordingEndedEvent struct {
-	UniqueKey string `json:"unique_key"`
-	Source    string `json:"source"`
-}
-
 func processVCRecordingEnded(_ context.Context, _ event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
-	envelope, ok := parseRecordingEndedEnvelope(raw)
+	body, ok := decodeEventBody[recordingBeanEventBody](raw)
 	if !ok {
-		return raw.Payload, nil
+		return nil, processing.DropMalformed(raw.EventType)
 	}
-	if !isRecordingEndedBeanEvent(envelope) {
+	if body.Source != recordingBeanSource {
 		return nil, nil
 	}
 	out := &VCRecordingEndedOutput{
-		Type:      recordingEndedEventType(envelope, raw),
-		EventID:   envelope.Header.EventID,
-		EventTime: recordingEndedEventTime(envelope.Header.CreateTime),
-		UniqueKey: envelope.Event.UniqueKey,
-		Source:    envelope.Event.Source,
+		Type:      raw.EventType,
+		EventID:   raw.EventID,
+		EventTime: millisToLocalRFC3339(raw.SourceTime),
+		UniqueKey: body.UniqueKey,
+		Source:    body.Source,
 	}
 	return json.Marshal(out)
 }
-
-func parseRecordingEndedEnvelope(raw *event.RawEvent) (*recordingEndedEnvelope, bool) {
-	var envelope recordingEndedEnvelope
-	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return nil, false
-	}
-	return &envelope, true
-}
-
-func isRecordingEndedBeanEvent(envelope *recordingEndedEnvelope) bool {
-	return envelope != nil && envelope.Event.Source == "recording_bean"
-}
-
-func recordingEndedEventType(envelope *recordingEndedEnvelope, raw *event.RawEvent) string {
-	if envelope != nil && envelope.Header.EventType != "" {
-		return envelope.Header.EventType
-	}
-	return raw.EventType
-}
-
-func recordingEndedEventTime(raw string) string {
-	if raw == "" {
-		return ""
-	}
-	millis, err := strconv.ParseInt(raw, 10, 64)
-	if err != nil {
-		return ""
-	}
-	return time.UnixMilli(millis).Local().Format(time.RFC3339)
-}

events/vc/recording_started.go

@@ -6,10 +6,9 @@ package vc
 import (
 	"context"
 	"encoding/json"
-	"strconv"
-	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 // VCRecordingStartedOutput is the flattened shape for vc.recording.recording_started_v1.
@@ -21,64 +20,20 @@ type VCRecordingStartedOutput struct {
 	Source    string `json:"source,omitempty"     desc:"Recording source; always recording_bean"`
 }
 
-type recordingStartedEnvelope struct {
-	Header struct {
-		EventID    string `json:"event_id"`
-		EventType  string `json:"event_type"`
-		CreateTime string `json:"create_time"`
-	} `json:"header"`
-	Event recordingStartedEvent `json:"event"`
-}
-
-type recordingStartedEvent struct {
-	UniqueKey string `json:"unique_key"`
-	Source    string `json:"source"`
-}
-
 func processVCRecordingStarted(_ context.Context, _ event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
-	envelope, ok := parseRecordingStartedEnvelope(raw)
+	body, ok := decodeEventBody[recordingBeanEventBody](raw)
 	if !ok {
-		return raw.Payload, nil
+		return nil, processing.DropMalformed(raw.EventType)
 	}
-	if !isRecordingStartedBeanEvent(envelope) {
+	if body.Source != recordingBeanSource {
 		return nil, nil
 	}
 	out := &VCRecordingStartedOutput{
-		Type:      recordingStartedEventType(envelope, raw),
-		EventID:   envelope.Header.EventID,
-		EventTime: recordingStartedEventTime(envelope.Header.CreateTime),
-		UniqueKey: envelope.Event.UniqueKey,
-		Source:    envelope.Event.Source,
+		Type:      raw.EventType,
+		EventID:   raw.EventID,
+		EventTime: millisToLocalRFC3339(raw.SourceTime),
+		UniqueKey: body.UniqueKey,
+		Source:    body.Source,
 	}
 	return json.Marshal(out)
 }
-
-func parseRecordingStartedEnvelope(raw *event.RawEvent) (*recordingStartedEnvelope, bool) {
-	var envelope recordingStartedEnvelope
-	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return nil, false
-	}
-	return &envelope, true
-}
-
-func isRecordingStartedBeanEvent(envelope *recordingStartedEnvelope) bool {
-	return envelope != nil && envelope.Event.Source == "recording_bean"
-}
-
-func recordingStartedEventType(envelope *recordingStartedEnvelope, raw *event.RawEvent) string {
-	if envelope != nil && envelope.Header.EventType != "" {
-		return envelope.Header.EventType
-	}
-	return raw.EventType
-}
-
-func recordingStartedEventTime(raw string) string {
-	if raw == "" {
-		return ""
-	}
-	millis, err := strconv.ParseInt(raw, 10, 64)
-	if err != nil {
-		return ""
-	}
-	return time.UnixMilli(millis).Local().Format(time.RFC3339)
-}

events/vc/recording_test.go

@@ -12,6 +12,7 @@ import (
 	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 func TestVCKeys_RecordingEventsRegistered(t *testing.T) {
@@ -25,7 +26,7 @@ func TestVCKeys_RecordingEventsRegistered(t *testing.T) {
 		{eventTypeRecordingEnded},
 	} {
 		t.Run(tc.eventType, func(t *testing.T) {
-			def, ok := event.Lookup(tc.eventType)
+			def, ok := lookupCompiledDef(t, tc.eventType)
 			if !ok {
 				t.Fatalf("%s should be registered via Keys()", tc.eventType)
 			}
@@ -351,7 +352,7 @@ func TestProcessVCRecording_NonRecordingBeanFiltered(t *testing.T) {
 	}
 }
 
-func TestProcessVCRecording_MalformedPayloadPassthrough(t *testing.T) {
+func TestProcessVCRecording_MalformedPayloadDrop(t *testing.T) {
 	t.Setenv("LARKSUITE_CLI_CONFIG_DIR", t.TempDir())
 
 	for _, tc := range []struct {
@@ -370,11 +371,11 @@ func TestProcessVCRecording_MalformedPayloadPassthrough(t *testing.T) {
 				Timestamp: time.Now(),
 			}
 			got, err := tc.process(context.Background(), nil, raw, nil)
-			if err != nil {
-				t.Fatalf("Process should swallow parse errors, got %v", err)
+			if !processing.IsDropMalformed(err) {
+				t.Fatalf("malformed payload must be dropped with a malformed marker, got err=%v", err)
 			}
-			if string(got) != "not json" {
-				t.Errorf("malformed fallback output = %q, want original bytes", string(got))
+			if got != nil {
+				t.Errorf("malformed payload must be dropped without output, got %q", string(got))
 			}
 		})
 	}
@@ -391,7 +392,7 @@ func TestVCRecording_PreConsumeSubscriptionLifecycle(t *testing.T) {
 		{eventTypeRecordingEnded},
 	} {
 		t.Run(tc.eventType, func(t *testing.T) {
-			def, ok := event.Lookup(tc.eventType)
+			def, ok := lookupCompiledDef(t, tc.eventType)
 			if !ok {
 				t.Fatalf("%s should be registered via Keys()", tc.eventType)
 			}
@@ -456,6 +457,7 @@ func runRecordingProcessRaw(t *testing.T, eventType string, process event.Proces
 		Payload:   json.RawMessage(payload),
 		Timestamp: time.Now(),
 	}
+	fillCanonicalFromHeader(t, raw)
 	got, err := process(context.Background(), nil, raw, nil)
 	if err != nil {
 		t.Fatalf("Process error: %v", err)

events/vc/recording_transcript_generated.go

@@ -6,10 +6,9 @@ package vc
 import (
 	"context"
 	"encoding/json"
-	"strconv"
-	"time"
 
 	"github.com/larksuite/cli/internal/event"
+	"github.com/larksuite/cli/internal/event/processing"
 )
 
 // VCRecordingTranscriptItemOutput is one flattened transcript item for recording events.
@@ -31,15 +30,6 @@ type VCRecordingTranscriptGeneratedOutput struct {
 	TranscriptItems []VCRecordingTranscriptItemOutput `json:"transcript_items,omitempty" desc:"Generated transcript items"`
 }
 
-type recordingTranscriptGeneratedEnvelope struct {
-	Header struct {
-		EventID    string `json:"event_id"`
-		EventType  string `json:"event_type"`
-		CreateTime string `json:"create_time"`
-	} `json:"header"`
-	Event recordingTranscriptGeneratedEvent `json:"event"`
-}
-
 type recordingTranscriptGeneratedEvent struct {
 	UniqueKey       string                               `json:"unique_key"`
 	Source          string                               `json:"source"`
@@ -61,58 +51,24 @@ type recordingTranscriptGeneratedSpeakerIn struct {
 type recordingTranscriptGeneratedString string
 
 func processVCRecordingTranscriptGenerated(_ context.Context, _ event.APIClient, raw *event.RawEvent, _ map[string]string) (json.RawMessage, error) {
-	envelope, ok := parseRecordingTranscriptGeneratedEnvelope(raw)
+	body, ok := decodeEventBody[recordingTranscriptGeneratedEvent](raw)
 	if !ok {
-		return raw.Payload, nil
+		return nil, processing.DropMalformed(raw.EventType)
 	}
-	if !isRecordingTranscriptGeneratedBeanEvent(envelope) {
+	if body.Source != recordingBeanSource {
 		return nil, nil
 	}
 	out := &VCRecordingTranscriptGeneratedOutput{
-		Type:            recordingTranscriptGeneratedEventType(envelope, raw),
-		EventID:         envelope.Header.EventID,
-		EventTime:       recordingTranscriptGeneratedEventTime(envelope.Header.CreateTime),
-		UniqueKey:       envelope.Event.UniqueKey,
-		Source:          envelope.Event.Source,
-		TranscriptItems: recordingTranscriptItems(envelope.Event.TranscriptItems),
+		Type:            raw.EventType,
+		EventID:         raw.EventID,
+		EventTime:       millisToLocalRFC3339(raw.SourceTime),
+		UniqueKey:       body.UniqueKey,
+		Source:          body.Source,
+		TranscriptItems: recordingTranscriptItems(body.TranscriptItems),
 	}
 	return json.Marshal(out)
 }
 
-func parseRecordingTranscriptGeneratedEnvelope(raw *event.RawEvent) (*recordingTranscriptGeneratedEnvelope, bool) {
-	var envelope recordingTranscriptGeneratedEnvelope
-	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
-		return nil, false
-	}
-	return &envelope, true
-}
-
-func isRecordingTranscriptGeneratedBeanEvent(envelope *recordingTranscriptGeneratedEnvelope) bool {
-	return envelope != nil && envelope.Event.Source == "recording_bean"
-}
-
-func recordingTranscriptGeneratedEventType(envelope *recordingTranscriptGeneratedEnvelope, raw *event.RawEvent) string {
-	if envelope != nil && envelope.Header.EventType != "" {
-		return envelope.Header.EventType
-	}
-	return raw.EventType
-}
-
-func recordingTranscriptGeneratedEventTime(raw string) string {
-	return recordingTranscriptGeneratedMillisToLocalRFC3339(raw)
-}
-
-func recordingTranscriptGeneratedMillisToLocalRFC3339(raw string) string {
-	if raw == "" {
-		return ""
-	}
-	millis, err := strconv.ParseInt(raw, 10, 64)
-	if err != nil {
-		return ""
-	}
-	return time.UnixMilli(millis).Local().Format(time.RFC3339)
-}
-
 func recordingTranscriptItems(items []recordingTranscriptGeneratedItemIn) []VCRecordingTranscriptItemOutput {
 	if len(items) == 0 {
 		return nil
@@ -128,8 +84,8 @@ func recordingTranscriptItem(item recordingTranscriptGeneratedItemIn) VCRecordin
 	return VCRecordingTranscriptItemOutput{
 		SpeakerName: recordingSpeakerName(item.Speaker),
 		Text:        item.Text,
-		StartTime:   recordingTranscriptGeneratedMillisToLocalRFC3339(item.StartTimeMs.String()),
-		EndTime:     recordingTranscriptGeneratedMillisToLocalRFC3339(item.EndTimeMs.String()),
+		StartTime:   millisToLocalRFC3339(item.StartTimeMs.String()),
+		EndTime:     millisToLocalRFC3339(item.EndTimeMs.String()),
 		SentenceID:  item.SentenceID,
 	}
 }

events/vc/register.go

@@ -7,6 +7,7 @@ package vc
 import (
 	"reflect"
 
+	"github.com/larksuite/cli/events/internal/subscribeprep"
 	"github.com/larksuite/cli/internal/event"
 )
 
@@ -41,7 +42,7 @@ func Keys() []event.KeyDefinition {
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(VCParticipantMeetingStartedOutput{})},
 			},
 			Process:    processVCParticipantMeetingStarted,
-			PreConsume: subscriptionPreConsume(eventTypeMeetingStarted, pathMeetingSubscribe, pathMeetingUnsubscribe),
+			PreConsume: subscribeprep.Hook(eventTypeMeetingStarted, pathMeetingSubscribe, pathMeetingUnsubscribe),
 			Scopes:     []string{"vc:meeting.meetingevent:read"},
 			AuthTypes: []string{
 				"user",
@@ -57,7 +58,7 @@ func Keys() []event.KeyDefinition {
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(VCParticipantMeetingJoinedOutput{})},
 			},
 			Process:    processVCParticipantMeetingJoined,
-			PreConsume: subscriptionPreConsume(eventTypeMeetingJoined, pathMeetingSubscribe, pathMeetingUnsubscribe),
+			PreConsume: subscribeprep.Hook(eventTypeMeetingJoined, pathMeetingSubscribe, pathMeetingUnsubscribe),
 			Scopes:     []string{"vc:meeting.meetingevent:read"},
 			AuthTypes: []string{
 				"user",
@@ -73,7 +74,7 @@ func Keys() []event.KeyDefinition {
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(VCParticipantMeetingEndedOutput{})},
 			},
 			Process:    processVCParticipantMeetingEnded,
-			PreConsume: subscriptionPreConsume(eventTypeMeetingEnded, pathMeetingSubscribe, pathMeetingUnsubscribe),
+			PreConsume: subscribeprep.Hook(eventTypeMeetingEnded, pathMeetingSubscribe, pathMeetingUnsubscribe),
 			Scopes:     []string{"vc:meeting.meetingevent:read"},
 			AuthTypes: []string{
 				"user",
@@ -89,7 +90,7 @@ func Keys() []event.KeyDefinition {
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(VCNoteGeneratedOutput{})},
 			},
 			Process:    processVCNoteGenerated,
-			PreConsume: subscriptionPreConsume(eventTypeNoteGenerated, pathNoteSubscribe, pathNoteUnsubscribe),
+			PreConsume: subscribeprep.Hook(eventTypeNoteGenerated, pathNoteSubscribe, pathNoteUnsubscribe),
 			Scopes:     []string{"vc:note:read"},
 			AuthTypes: []string{
 				"user",
@@ -105,7 +106,7 @@ func Keys() []event.KeyDefinition {
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(VCRecordingStartedOutput{})},
 			},
 			Process:    processVCRecordingStarted,
-			PreConsume: subscriptionPreConsume(eventTypeRecordingStarted, pathRecordingSubscribe, pathRecordingUnsubscribe),
+			PreConsume: subscribeprep.Hook(eventTypeRecordingStarted, pathRecordingSubscribe, pathRecordingUnsubscribe),
 			Scopes:     []string{"vc:recording:read"},
 			AuthTypes: []string{
 				"user",
@@ -121,7 +122,7 @@ func Keys() []event.KeyDefinition {
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(VCRecordingTranscriptGeneratedOutput{})},
 			},
 			Process:    processVCRecordingTranscriptGenerated,
-			PreConsume: subscriptionPreConsume(eventTypeRecordingTranscriptGenerated, pathRecordingSubscribe, pathRecordingUnsubscribe),
+			PreConsume: subscribeprep.Hook(eventTypeRecordingTranscriptGenerated, pathRecordingSubscribe, pathRecordingUnsubscribe),
 			Scopes:     []string{"vc:recording:read"},
 			AuthTypes: []string{
 				"user",
@@ -137,7 +138,7 @@ func Keys() []event.KeyDefinition {
 				Custom: &event.SchemaSpec{Type: reflect.TypeOf(VCRecordingEndedOutput{})},
 			},
 			Process:    processVCRecordingEnded,
-			PreConsume: subscriptionPreConsume(eventTypeRecordingEnded, pathRecordingSubscribe, pathRecordingUnsubscribe),
+			PreConsume: subscribeprep.Hook(eventTypeRecordingEnded, pathRecordingSubscribe, pathRecordingUnsubscribe),
 			Scopes:     []string{"vc:recording:read"},
 			AuthTypes: []string{
 				"user",

events/vc/test_helpers_test.go

@@ -8,8 +8,34 @@ import (
 	"encoding/json"
 	"reflect"
 	"testing"
+
+	"github.com/larksuite/cli/internal/event"
 )
 
+// fillCanonicalFromHeader copies the payload envelope header metadata onto
+// the RawEvent canonical fields. Process handlers read event_id and
+// create_time from the RawEvent, which the consume pipeline fills from the
+// envelope header before dispatch; tests that hand-build a RawEvent must
+// mirror that so both views agree.
+func fillCanonicalFromHeader(t *testing.T, raw *event.RawEvent) {
+	t.Helper()
+	var envelope struct {
+		Header struct {
+			EventID    string `json:"event_id"`
+			EventType  string `json:"event_type"`
+			CreateTime string `json:"create_time"`
+		} `json:"header"`
+	}
+	if err := json.Unmarshal(raw.Payload, &envelope); err != nil {
+		t.Fatalf("parse envelope header: %v", err)
+	}
+	raw.EventID = envelope.Header.EventID
+	if envelope.Header.EventType != "" {
+		raw.EventType = envelope.Header.EventType
+	}
+	raw.SourceTime = envelope.Header.CreateTime
+}
+
 type stubAPIClient struct {
 	callFn func(ctx context.Context, method, path string, body any) (json.RawMessage, error)
 }

events/whiteboard/preconsume.go

@@ -6,17 +6,13 @@ package whiteboard
 import (
 	"context"
 	"fmt"
-	"time"
 
 	"github.com/larksuite/cli/errs"
+	"github.com/larksuite/cli/events/internal/subscribeprep"
 	"github.com/larksuite/cli/internal/event"
 	"github.com/larksuite/cli/internal/validate"
 )
 
-// cleanupTimeout bounds how long the unsubscribe call has to finish during
-// PreConsume cleanup so a stuck OAPI cannot block process shutdown.
-const cleanupTimeout = 5 * time.Second
-
 // whiteboardSubscriptionPreConsume calls the whiteboard event subscribe OAPI
 // and returns a cleanup that invokes the matching unsubscribe.
 //
@@ -39,18 +35,6 @@ func whiteboardSubscriptionPreConsume(eventType string) func(context.Context, ev
 		subscribePath := fmt.Sprintf("/open-apis/board/v1/whiteboards/%s/subscribe", encoded)
 		unsubscribePath := fmt.Sprintf("/open-apis/board/v1/whiteboards/%s/unsubscribe", encoded)
 
-		body := map[string]string{"event_type": eventType}
-		if _, err := rt.CallAPI(ctx, "POST", subscribePath, body); err != nil {
-			return nil, err
-		}
-
-		return func() error {
-			cleanupCtx, cancel := context.WithTimeout(context.Background(), cleanupTimeout)
-			defer cancel()
-			if _, err := rt.CallAPI(cleanupCtx, "POST", unsubscribePath, body); err != nil {
-				return err
-			}
-			return nil
-		}, nil
+		return subscribeprep.SubscribeWithCleanup(ctx, rt, eventType, subscribePath, unsubscribePath)
 	}
 }

events/whiteboard/register.go

@@ -24,10 +24,15 @@ func Keys() []event.KeyDefinition {
 			EventType:   eventTypeWhiteboardUpdated,
 			Params: []event.ParamDef{
 				{
-					Name:        "whiteboard_id",
-					Type:        event.ParamString,
-					Required:    true,
-					Description: "Whiteboard id to subscribe; subscription is per-whiteboard.",
+					Name:     "whiteboard_id",
+					Type:     event.ParamString,
+					Required: true,
+					// The server-side subscription is keyed per whiteboard, so
+					// the id must be part of the consumer's subscription
+					// identity: consumers of different whiteboards get their
+					// own setup/cleanup lifecycle instead of sharing one.
+					SubscriptionKey: true,
+					Description:     "Whiteboard id to subscribe; subscription is per-whiteboard.",
 				},
 			},
 			Schema: event.SchemaDef{

events/whiteboard/subscription_scope_test.go

@@ -0,0 +1,36 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package whiteboard
+
+import (
+	"testing"
+
+	event "github.com/larksuite/cli/internal/event"
+)
+
+// The whiteboard subscription is registered per whiteboard on the server, so
+// the whiteboard id must take part in the consumer's subscription identity.
+// Without it, two consumers of different whiteboards share one scope: the
+// second consumer's setup never runs (its whiteboard is never subscribed) and
+// whichever exits last unsubscribes the other one's still-active board.
+func TestWhiteboardID_IsPartOfSubscriptionIdentity(t *testing.T) {
+	defs := Keys()
+	if len(defs) != 1 {
+		t.Fatalf("expected exactly one whiteboard key, got %d", len(defs))
+	}
+	def := defs[0]
+
+	var found *event.ParamDef
+	for i := range def.Params {
+		if def.Params[i].Name == "whiteboard_id" {
+			found = &def.Params[i]
+		}
+	}
+	if found == nil {
+		t.Fatal("whiteboard_id param is missing")
+	}
+	if !found.SubscriptionKey {
+		t.Error("whiteboard_id must be a subscription key: the server-side subscription is per-whiteboard")
+	}
+}

internal/event/adapter/lark/websocket/feishu.go

@@ -1,7 +1,7 @@
 // Copyright (c) 2026 Lark Technologies Pte. Ltd.
 // SPDX-License-Identifier: MIT
 
-package source
+package websocket
 
 import (
 	"context"
@@ -18,7 +18,6 @@ import (
 	larkws "github.com/larksuite/oapi-sdk-go/v3/ws"
 
 	"github.com/larksuite/cli/internal/event"
-	"github.com/larksuite/cli/internal/event/protocol"
 )
 
 const maxEventBodyBytes = 1 << 20 // bound per-subscriber sendCh memory under runaway payloads
@@ -51,7 +50,7 @@ func (s *FeishuSource) Start(ctx context.Context, eventTypes []string, emit func
 	}
 
 	if notify != nil {
-		notify(protocol.SourceStateConnecting, "")
+		notify(sourceStateConnecting, "")
 	}
 	cli := larkws.NewClient(s.AppID, s.AppSecret, opts...)
 
@@ -83,6 +82,8 @@ func (s *FeishuSource) buildRawHandler(emit func(*event.RawEvent)) func(context.
 				EventID    string `json:"event_id"`
 				EventType  string `json:"event_type"`
 				CreateTime string `json:"create_time"`
+				AppID      string `json:"app_id"`
+				TenantKey  string `json:"tenant_key"`
 			} `json:"header"`
 		}
 		if err := json.Unmarshal(e.Body, &envelope); err != nil {
@@ -106,6 +107,8 @@ func (s *FeishuSource) buildRawHandler(emit func(*event.RawEvent)) func(context.
 			EventID:    envelope.Header.EventID,
 			EventType:  envelope.Header.EventType,
 			SourceTime: envelope.Header.CreateTime,
+			AppID:      envelope.Header.AppID,
+			TenantKey:  envelope.Header.TenantKey,
 			Payload:    json.RawMessage(e.Body),
 			Timestamp:  time.Now(),
 		})
@@ -157,12 +160,22 @@ func (a *sdkLogger) tryNotify(msg, errDetail string) {
 		if m := reconnectAttemptRe.FindStringSubmatch(lower); len(m) == 2 {
 			detail = "attempt " + m[1]
 		}
-		a.notify(protocol.SourceStateReconnecting, detail)
+		a.notify(sourceStateReconnecting, detail)
 	case strings.HasPrefix(lower, sdkLogDisconnected):
-		a.notify(protocol.SourceStateDisconnected, errDetail)
+		a.notify(sourceStateDisconnected, errDetail)
 	case strings.HasPrefix(lower, sdkLogConnected):
-		a.notify(protocol.SourceStateConnected, "")
+		a.notify(sourceStateConnected, "")
 	}
 }
 
 var _ larkcore.Logger = (*sdkLogger)(nil)
+
+// Source lifecycle states as this adapter reports them. The values are the
+// wire vocabulary of the bus's source_status frames; a pinning test keeps
+// them equal to the IPC constants without importing the IPC package here.
+const (
+	sourceStateConnecting   = "connecting"
+	sourceStateConnected    = "connected"
+	sourceStateDisconnected = "disconnected"
+	sourceStateReconnecting = "reconnecting"
+)

internal/event/adapter/lark/websocket/feishu_ingress_test.go

@@ -0,0 +1,62 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"context"
+	"testing"
+
+	larkevent "github.com/larksuite/oapi-sdk-go/v3/event"
+
+	event "github.com/larksuite/cli/internal/event"
+)
+
+// The websocket ingress is the only place that parses the envelope header;
+// every canonical fact consumers rely on must be captured here, once.
+func TestBuildRawHandler_ParsesCanonicalHeaderOnce(t *testing.T) {
+	s := &FeishuSource{}
+	var got *event.RawEvent
+	handler := s.buildRawHandler(func(ev *event.RawEvent) { got = ev })
+
+	body := []byte(`{"schema":"2.0","header":{"event_id":"evt-1","event_type":"im.message.receive_v1",` +
+		`"create_time":"1700000000000","app_id":"cli_test_app","tenant_key":"tenant_test"},"event":{}}`)
+	if err := handler(context.Background(), &larkevent.EventReq{Body: body}); err != nil {
+		t.Fatalf("handler: %v", err)
+	}
+	if got == nil {
+		t.Fatal("event was not emitted")
+	}
+	if got.EventID != "evt-1" || got.EventType != "im.message.receive_v1" {
+		t.Errorf("identity facts wrong: id=%q type=%q", got.EventID, got.EventType)
+	}
+	if got.SourceTime != "1700000000000" {
+		t.Errorf("SourceTime = %q, want upstream create_time", got.SourceTime)
+	}
+	if got.AppID != "cli_test_app" || got.TenantKey != "tenant_test" {
+		t.Errorf("tenant identity not captured: app_id=%q tenant_key=%q", got.AppID, got.TenantKey)
+	}
+	if got.Timestamp.IsZero() {
+		t.Error("local observation Timestamp must be set at ingress")
+	}
+}
+
+// A header that omits optional facts leaves them visibly empty — the ingress
+// never substitutes local configuration for missing upstream facts.
+func TestBuildRawHandler_MissingOptionalFactsStayEmpty(t *testing.T) {
+	s := &FeishuSource{}
+	var got *event.RawEvent
+	handler := s.buildRawHandler(func(ev *event.RawEvent) { got = ev })
+
+	body := []byte(`{"schema":"2.0","header":{"event_id":"evt-2","event_type":"im.message.receive_v1"},"event":{}}`)
+	if err := handler(context.Background(), &larkevent.EventReq{Body: body}); err != nil {
+		t.Fatalf("handler: %v", err)
+	}
+	if got == nil {
+		t.Fatal("event was not emitted")
+	}
+	if got.SourceTime != "" || got.AppID != "" || got.TenantKey != "" {
+		t.Errorf("missing facts must stay empty: source_time=%q app_id=%q tenant_key=%q",
+			got.SourceTime, got.AppID, got.TenantKey)
+	}
+}

internal/event/adapter/lark/websocket/feishu_log_test.go

@@ -1,7 +1,7 @@
 // Copyright (c) 2026 Lark Technologies Pte. Ltd.
 // SPDX-License-Identifier: MIT
 
-package source
+package websocket
 
 import (
 	"bytes"

internal/event/adapter/lark/websocket/feishu_test.go

@@ -1,12 +1,12 @@
 // Copyright (c) 2026 Lark Technologies Pte. Ltd.
 // SPDX-License-Identifier: MIT
 
-package source
+package websocket
 
 import (
 	"testing"
 
-	"github.com/larksuite/cli/internal/event/protocol"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/protocol"
 )
 
 // "disconnected to <url>" contains "connected to ws" — must use HasPrefix to avoid misclassifying as connect.

internal/event/adapter/lark/websocket/sdk_log_patterns.go

@@ -1,7 +1,7 @@
 // Copyright (c) 2026 Lark Technologies Pte. Ltd.
 // SPDX-License-Identifier: MIT
 
-package source
+package websocket
 
 // DO NOT trim trailing spaces — the HasPrefix disambiguator depends on them.
 const (

internal/event/adapter/lark/websocket/sdk_log_patterns_test.go

@@ -1,7 +1,7 @@
 // Copyright (c) 2026 Lark Technologies Pte. Ltd.
 // SPDX-License-Identifier: MIT
 
-package source
+package websocket
 
 import (
 	"context"
@@ -9,7 +9,7 @@ import (
 	"sync"
 	"testing"
 
-	"github.com/larksuite/cli/internal/event/protocol"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/protocol"
 )
 
 // Samples preserve the real SDK shape ("<verb> to <url>[conn_id=...]" — no space before bracket).

internal/event/adapter/lark/websocket/source.go

@@ -0,0 +1,11 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+// Package source is a pluggable event source abstraction (separate package to keep
+// business registrations free of SDK transitive deps).
+package websocket
+
+// StatusNotifier surfaces source lifecycle states; detail is free-form
+// context. A function alias so implementations structurally satisfy the
+// bus-side Source port without importing it.
+type StatusNotifier = func(state, detail string)

internal/event/adapter/lark/websocket/source_test.go

@@ -1,7 +1,7 @@
 // Copyright (c) 2026 Lark Technologies Pte. Ltd.
 // SPDX-License-Identifier: MIT
 
-package source
+package websocket
 
 import (
 	"context"
@@ -25,18 +25,6 @@ func (s *mockSource) Start(ctx context.Context, _ []string, emit func(*event.Raw
 	return nil
 }
 
-func TestRegister(t *testing.T) {
-	ResetForTest()
-
-	src := &mockSource{name: "test-source"}
-	Register(src)
-
-	sources := All()
-	if len(sources) != 1 || sources[0].Name() != "test-source" {
-		t.Errorf("unexpected sources: %v", sources)
-	}
-}
-
 func TestMockSource_EmitsEvents(t *testing.T) {
 	src := &mockSource{
 		name: "test",

internal/event/adapter/lark/websocket/state_pinning_test.go

@@ -0,0 +1,28 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package websocket
+
+import (
+	"testing"
+
+	"github.com/larksuite/cli/internal/event/adapter/localbus/protocol"
+	"github.com/larksuite/cli/internal/event/bus"
+)
+
+// The adapter reports source states with its own constants so it never
+// imports the IPC package; this test pins all three vocabularies (adapter,
+// bus port, IPC frame) to the same wire values.
+func TestSourceStates_MatchTheWireVocabulary(t *testing.T) {
+	pins := []struct{ adapter, port, wire string }{
+		{sourceStateConnecting, bus.SourceStateConnecting, protocol.SourceStateConnecting},
+		{sourceStateConnected, bus.SourceStateConnected, protocol.SourceStateConnected},
+		{sourceStateDisconnected, bus.SourceStateDisconnected, protocol.SourceStateDisconnected},
+		{sourceStateReconnecting, bus.SourceStateReconnecting, protocol.SourceStateReconnecting},
+	}
+	for _, pin := range pins {
+		if pin.adapter != pin.port || pin.port != pin.wire {
+			t.Errorf("state vocabulary drifted: adapter=%q port=%q wire=%q", pin.adapter, pin.port, pin.wire)
+		}
+	}
+}

internal/event/adapter/localbus/busctl/busctl.go

@@ -10,8 +10,8 @@ import (
 	"fmt"
 	"time"
 
-	"github.com/larksuite/cli/internal/event/protocol"
-	"github.com/larksuite/cli/internal/event/transport"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/protocol"
+	"github.com/larksuite/cli/internal/event/adapter/localbus/transport"
 )
 
 const readTimeout = 5 * time.Second // matches protocol.WriteTimeout

internal/event/adapter/localbus/protocol/canonical_fields_test.go

@@ -0,0 +1,86 @@
+// Copyright (c) 2026 Lark Technologies Pte. Ltd.
+// SPDX-License-Identifier: MIT
+
+package protocol
+
+import (
+	"bufio"
+	"bytes"
+	"encoding/json"
+	"testing"
+	"time"
+
+	"github.com/larksuite/cli/internal/event/model"
+)
+
+// Every canonical fact the ingress parsed must survive the wire round trip
+// verbatim — the consumer restores the event from this frame instead of
+// re-deriving anything from the payload.
+func TestEventFrame_CarriesCanonicalFactsVerbatim(t *testing.T) {
+	observed := time.Date(2023, 11, 14, 22, 13, 20, 123456789, time.UTC)
+	ev := &model.Event{
+		EventID:    "evt-42",
+		EventType:  "im.message.receive_v1",
+		SourceTime: "1700000000000",
+		AppID:      "cli_test_app",
+		TenantKey:  "tenant_test",
+		Payload:    json.RawMessage(`{"schema":"2.0"}`),
+		Timestamp:  observed,
+	}
+
+	var buf bytes.Buffer
+	if err := Encode(&buf, NewEvent(ev, 7)); err != nil {
+		t.Fatalf("encode: %v", err)
+	}
+	line, err := ReadFrame(bufio.NewReader(&buf))
+	if err != nil {
+		t.Fatalf("read frame: %v", err)
+	}
+	decoded, err := Decode(line)
+	if err != nil {
+		t.Fatalf("decode: %v", err)
+	}
+	frame, ok := decoded.(*Event)
+	if !ok {
+		t.Fatalf("decoded %T, want *Event", decoded)
+	}
+
+	if frame.EventID != ev.EventID || frame.EventType != ev.EventType ||
+		frame.SourceTime != ev.SourceTime || frame.AppID != ev.AppID ||
+		frame.TenantKey != ev.TenantKey || frame.Seq != 7 {
+		t.Errorf("canonical facts drifted across the wire: %+v", frame)
+	}
+	// observed_at is a fixed RFC3339Nano string contract, not an incidental
+	// time.Time marshal shape.
+	parsed, err := time.Parse(time.RFC3339Nano, frame.ObservedAt)
+	if err != nil {
+		t.Fatalf("observed_at %q is not RFC3339Nano: %v", frame.ObservedAt, err)
+	}
+	if !parsed.Equal(observed) {
+		t.Errorf("observed_at: got %v, want %v", parsed, observed)
+	}
+}
+
+// Facts the upstream omitted stay omitted on the wire: the frame never invents
+// values, and absent facts must not even appear as empty strings.
+func TestEventFrame_MissingFactsStayAbsent(t *testing.T) {
+	ev := &model.Event{
+		EventType: "im.message.receive_v1",
+		EventID:   "evt-1",
+		Payload:   json.RawMessage(`{}`),
+	}
+
+	raw, err := json.Marshal(NewEvent(ev, 1))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var asMap map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &asMap); err != nil {
+		t.Fatal(err)
+	}
+	for _, absent := range []string{"source_time", "app_id", "tenant_key", "observed_at"} {
+		if _, present := asMap[absent]; present {
+			t.Errorf("field %q must be omitted when the fact is missing, frame: %s", absent, raw)
+		}
+	}
+}