test(events): 补授权拒绝与脱敏投递的模块契约

新增 TestPublishAsReturnsDeniedErrorAndBlocksDelivery
      策略拒绝 → 返回可诊断 DeniedError，订阅者零投递
    新增 TestSubscribersReceiveScrubbedPayload
      入向脱敏后订阅者看不到明文
    新增 TestHandlerPanicDoesNotBreakOtherSubscribers
      handler panic 不影响其他订阅者
    变异：绕过 AllowPublish 判断 → denied publish 返回 nil 错误
          跳过 ScrubInbound     → 明文 SECRET 到达订阅者

internal/events/test/bus_contract_test.go

@@ -4,6 +4,7 @@ package test
 // 只调导出方法，不碰 b.subs、不读源码。
 
 import (
+	"errors"
 	"testing"
 	"time"
 
@@ -96,3 +97,88 @@ func TestSetGuardRoundTrip(t *testing.T) {
 		t.Fatal("★ Guard() 没拿回 SetGuard 设置的守卫")
 	}
 }
+
+// denyGuard 拒绝来自指定发布者的一切发布。
+type denyGuard struct {
+	denyPublisher string
+	scrubTo       string
+}
+
+func (g *denyGuard) AllowPublish(p events.Publisher, et events.EventType) (bool, string) {
+	if p.Name == g.denyPublisher {
+		return false, "policy forbids " + p.Name
+	}
+	return true, ""
+}
+func (g *denyGuard) AllowSubscribe(events.Publisher, events.EventType) (bool, string) {
+	return true, ""
+}
+func (g *denyGuard) ScrubPayload(et events.EventType, m map[string]interface{}) map[string]interface{} {
+	return m
+}
+func (g *denyGuard) ScrubInbound(et events.EventType, m map[string]interface{}) map[string]interface{} {
+	if g.scrubTo != "" {
+		for k := range m {
+			m[k] = g.scrubTo
+		}
+	}
+	return m
+}
+
+// TestPublishAsReturnsDeniedErrorAndBlocksDelivery：被策略拒绝的发布必须
+// 返回可诊断错误，且订阅者收不到任何东西。
+func TestPublishAsReturnsDeniedErrorAndBlocksDelivery(t *testing.T) {
+	bus := events.NewBus()
+	bus.SetGuard(&denyGuard{denyPublisher: "bad-plugin"})
+
+	var delivered int
+	bus.Subscribe(events.EventType("contract.denied"), func(*events.Event) { delivered++ })
+
+	err := bus.PublishAs(events.Publisher{Name: "bad-plugin"},
+		&events.Event{Type: events.EventType("contract.denied"), Payload: map[string]interface{}{"x": 1}})
+	if err == nil {
+		t.Fatal("denied publish returned nil error")
+	}
+	var denied *events.DeniedError
+	if !errors.As(err, &denied) || denied.Publisher != "bad-plugin" {
+		t.Fatalf("error is not a diagnosable DeniedError: %v", err)
+	}
+	if delivered != 0 {
+		t.Fatalf("denied event still reached subscribers: %d", delivered)
+	}
+}
+
+func TestSubscribersReceiveScrubbedPayload(t *testing.T) {
+	bus := events.NewBus()
+	bus.SetGuard(&denyGuard{scrubTo: "[redacted]"})
+	got := make(chan map[string]interface{}, 1)
+	bus.Subscribe(events.EventType("contract.scrub"), func(e *events.Event) { got <- e.Payload })
+
+	if err := bus.PublishAs(events.Publisher{Name: "p"},
+		&events.Event{Type: events.EventType("contract.scrub"),
+			Payload: map[string]interface{}{"token": "SECRET", "x": 1}}); err != nil {
+		t.Fatalf("publish: %v", err)
+	}
+	select {
+	case p := <-got:
+		for k, v := range p {
+			if v == "SECRET" {
+				t.Fatalf("plaintext secret reached subscriber: %v", p)
+			}
+			_ = k
+		}
+	case <-time.After(2 * time.Second):
+		t.Fatal("subscriber never received the event")
+	}
+}
+
+func TestHandlerPanicDoesNotBreakOtherSubscribers(t *testing.T) {
+	bus := events.NewBus()
+	var ran int
+	bus.Subscribe(events.EventType("contract.panic"), func(*events.Event) { panic("boom") })
+	bus.Subscribe(events.EventType("contract.panic"), func(*events.Event) { ran++ })
+	bus.Publish(&events.Event{Type: events.EventType("contract.panic")})
+	if ran != 1 {
+		t.Fatalf("healthy subscriber skipped after panic: ran=%d", ran)
+	}
+}