feat: 实现可靠事件与证据 connector (#153)

2026-08-31 14:57:48 +08:00
parent cf00d73436
commit 04c5deecfb
36 changed files with 4102 additions and 14 deletions
@@ -0,0 +1,187 @@
+package bell_connector
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"net/http"
+	"net/url"
+	"regexp"
+	"strconv"
+	"strings"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
+)
+
+type HTTPDoer interface {
+	Do(*http.Request) (*http.Response, error)
+}
+
+var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$`)
+
+type Client struct {
+	Endpoint        string
+	RelayID         string
+	Signer          machine_identity.Signer
+	HTTP            HTTPDoer
+	Enabled         bool
+	MaxRequestBytes int64
+}
+
+func NewClient(endpoint, relayID string, signer machine_identity.Signer, policy machine_identity.TransportPolicy) (*Client, error) {
+	parsed, err := url.Parse(endpoint)
+	if err != nil || parsed.Scheme != "https" || parsed.Host == "" || parsed.User != nil || parsed.Path != "" || parsed.RawQuery != "" || parsed.Fragment != "" {
+		return nil, errors.New("Bell connector endpoint must be an HTTPS origin")
+	}
+	httpClient, err := policy.HTTPClient()
+	if err != nil {
+		return nil, err
+	}
+	return &Client{Endpoint: strings.TrimRight(endpoint, "/"), RelayID: relayID, Signer: signer, HTTP: httpClient, Enabled: true, MaxRequestBytes: policy.MaxRequestBytes}, nil
+}
+
+func (c Client) Send(ctx context.Context, body []byte) (IngestResult, error) {
+	if !c.Enabled {
+		return IngestResult{}, &DeliveryError{Code: "connector_disabled", Detail: "Bell connector is disabled", Terminal: true}
+	}
+	if c.HTTP == nil || !json.Valid(body) || (strings.TrimSpace(c.RelayID) != "" && !safeIdentifier(c.RelayID)) {
+		return IngestResult{}, errors.New("Bell connector is not configured")
+	}
+	maximum := c.MaxRequestBytes
+	if maximum == 0 {
+		maximum = MaxInboundBytes
+	}
+	if maximum < 1 || int64(len(body)) > maximum {
+		return IngestResult{}, &DeliveryError{Code: "event_request_too_large", Detail: "event exceeds the configured request limit", Terminal: true}
+	}
+	requestID, err := newRequestID()
+	if err != nil {
+		return IngestResult{}, fmt.Errorf("generate request correlation id: %w", err)
+	}
+	token, err := c.Signer.Mint("yovision-bell", []string{"events:ingest"}, http.MethodPost, "/v1/events", body)
+	if err != nil {
+		return IngestResult{}, fmt.Errorf("mint Bell machine token: %w", err)
+	}
+	request, err := http.NewRequestWithContext(ctx, http.MethodPost, strings.TrimRight(c.Endpoint, "/")+"/v1/events", bytes.NewReader(body))
+	if err != nil {
+		return IngestResult{}, err
+	}
+	request.Header.Set("Authorization", "Bearer "+token)
+	request.Header.Set("Content-Type", "application/json")
+	request.Header.Set("X-Request-ID", requestID)
+	if strings.TrimSpace(c.RelayID) != "" {
+		request.Header.Set("X-YoVision-Relay-ID", c.RelayID)
+	}
+	response, err := c.HTTP.Do(request)
+	if err != nil {
+		return IngestResult{}, fmt.Errorf("deliver event to Bell: %w", err)
+	}
+	defer response.Body.Close()
+	responseBody, err := io.ReadAll(io.LimitReader(response.Body, 64*1024+1))
+	if err != nil || len(responseBody) > 64*1024 {
+		return IngestResult{}, errors.New("Bell response is invalid")
+	}
+	if response.StatusCode == http.StatusCreated || response.StatusCode == http.StatusOK {
+		if response.Header.Get("X-Request-ID") != requestID {
+			return IngestResult{}, errors.New("Bell response request id is invalid")
+		}
+		var result IngestResult
+		if json.Unmarshal(responseBody, &result) != nil || result.EventID == "" || result.PayloadSHA256 == "" || (result.Disposition != "created" && result.Disposition != "duplicate") {
+			return IngestResult{}, errors.New("Bell response is invalid")
+		}
+		identity, parseErr := parseEventIdentity(body)
+		if parseErr != nil || result.ProducerID != identity.ProducerID || result.SourceEventID != identity.SourceEventID {
+			return IngestResult{}, errors.New("Bell response changed event identity")
+		}
+		canonical, canonicalErr := canonicalPayload(body)
+		if canonicalErr != nil {
+			return IngestResult{}, errors.New("delivered event cannot be canonicalized")
+		}
+		digest := sha256.Sum256(canonical)
+		if result.PayloadSHA256 != hex.EncodeToString(digest[:]) {
+			return IngestResult{}, errors.New("Bell response payload digest does not match the delivered event")
+		}
+		return result, nil
+	}
+	var problem Problem
+	_ = json.Unmarshal(responseBody, &problem)
+	terminal := response.StatusCode >= 400 && response.StatusCode < 500 && response.StatusCode != http.StatusTooManyRequests
+	if problem.Code == "" {
+		problem.Code = "bell_unavailable"
+	}
+	return IngestResult{}, &DeliveryError{Code: problem.Code, Detail: problem.Message, Terminal: terminal}
+}
+
+func newRequestID() (string, error) {
+	raw := make([]byte, 16)
+	if _, err := rand.Read(raw); err != nil {
+		return "", err
+	}
+	value := base64.RawURLEncoding.EncodeToString(raw)
+	if !requestIDPattern.MatchString(value) {
+		return "", errors.New("generated request id is invalid")
+	}
+	return value, nil
+}
+
+func canonicalPayload(raw []byte) ([]byte, error) {
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	decoder.UseNumber()
+	var value any
+	if err := decoder.Decode(&value); err != nil {
+		return nil, err
+	}
+	value, err := normalizeJCSNumbers(value)
+	if err != nil {
+		return nil, err
+	}
+	var buffer bytes.Buffer
+	encoder := json.NewEncoder(&buffer)
+	encoder.SetEscapeHTML(false)
+	if err := encoder.Encode(value); err != nil {
+		return nil, err
+	}
+	canonical := bytes.TrimSuffix(buffer.Bytes(), []byte("\n"))
+	canonical = bytes.ReplaceAll(canonical, []byte(`\u2028`), []byte("\u2028"))
+	canonical = bytes.ReplaceAll(canonical, []byte(`\u2029`), []byte("\u2029"))
+	return canonical, nil
+}
+
+func normalizeJCSNumbers(value any) (any, error) {
+	switch typed := value.(type) {
+	case json.Number:
+		number, err := strconv.ParseFloat(string(typed), 64)
+		if err != nil || math.IsNaN(number) || math.IsInf(number, 0) {
+			return nil, errors.New("JSON number is outside the RFC 8785 domain")
+		}
+		if number == 0 {
+			return float64(0), nil
+		}
+		return number, nil
+	case []any:
+		for index, item := range typed {
+			normalized, err := normalizeJCSNumbers(item)
+			if err != nil {
+				return nil, err
+			}
+			typed[index] = normalized
+		}
+	case map[string]any:
+		for key, item := range typed {
+			normalized, err := normalizeJCSNumbers(item)
+			if err != nil {
+				return nil, err
+			}
+			typed[key] = normalized
+		}
+	}
+	return value, nil
+}
@@ -0,0 +1,144 @@
+package bell_connector
+
+import (
+	"errors"
+	"io"
+	"net/http"
+	"regexp"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
+)
+
+const MaxInboundBytes = 64 * 1024
+
+var inboundRequestIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$`)
+
+type IngressHandler struct {
+	DB              *gorm.DB
+	Verifier        machine_identity.Verifier
+	EvidenceOwnerID string
+	Now             func() time.Time
+}
+
+func (h IngressHandler) Post(c *gin.Context) {
+	if !prepareMachineRequest(c, "/v1/events") {
+		return
+	}
+	body, err := io.ReadAll(http.MaxBytesReader(c.Writer, c.Request.Body, MaxInboundBytes))
+	if err != nil {
+		problem(c, http.StatusBadRequest, "invalid_event", "event payload is invalid or too large")
+		return
+	}
+	token, err := machine_identity.BearerToken(c.GetHeader("Authorization"))
+	if err != nil {
+		problem(c, http.StatusUnauthorized, machineCode(err), "machine identity was rejected")
+		return
+	}
+	if _, err = h.Verifier.Verify(token, "yovision-sense", "events:ingest", c.Request.Method, c.Request.URL.EscapedPath(), body); err != nil {
+		status := http.StatusUnauthorized
+		if code := machineCode(err); code == "machine_scope_denied" || code == "machine_audience_denied" {
+			status = http.StatusForbidden
+		}
+		problem(c, status, machineCode(err), "machine identity was rejected")
+		return
+	}
+	now := time.Now().UTC()
+	if h.Now != nil {
+		now = h.Now().UTC()
+	}
+	result, err := AcceptEvent(c.Request.Context(), h.DB, body, h.EvidenceOwnerID, now)
+	if errors.Is(err, ErrInboundConflict) {
+		problem(c, http.StatusConflict, "idempotency_conflict", "event identity is bound to another payload")
+		return
+	}
+	if err != nil {
+		problem(c, http.StatusBadRequest, "invalid_event", "event payload was rejected")
+		return
+	}
+	status := http.StatusAccepted
+	if result.Disposition == "duplicate" {
+		status = http.StatusOK
+	}
+	c.JSON(status, result)
+}
+
+type EvidenceHandler struct {
+	DB       *gorm.DB
+	Verifier machine_identity.Verifier
+	Now      func() time.Time
+}
+
+func (h EvidenceHandler) Get(c *gin.Context) {
+	path := c.Request.URL.EscapedPath()
+	if !prepareMachineRequest(c, path) || !safeIdentifier(c.Param("evidence_id")) {
+		if !c.Writer.Written() {
+			problem(c, http.StatusBadRequest, "evidence_not_found", "evidence reference is invalid")
+		}
+		return
+	}
+	token, err := machine_identity.BearerToken(c.GetHeader("Authorization"))
+	if err != nil {
+		problem(c, http.StatusUnauthorized, machineCode(err), "machine identity was rejected")
+		return
+	}
+	if _, err = h.Verifier.Verify(token, "yovision-sense", "evidence:read", c.Request.Method, path, nil); err != nil {
+		status := http.StatusUnauthorized
+		if code := machineCode(err); code == "machine_scope_denied" || code == "machine_audience_denied" {
+			status = http.StatusForbidden
+		}
+		problem(c, status, machineCode(err), "machine identity was rejected")
+		return
+	}
+	var record EvidenceRecord
+	if err = h.DB.WithContext(c.Request.Context()).First(&record, "evidence_id = ?", c.Param("evidence_id")).Error; err != nil {
+		if errors.Is(err, gorm.ErrRecordNotFound) {
+			problem(c, http.StatusNotFound, "evidence_not_found", "evidence reference is unknown")
+			return
+		}
+		problem(c, http.StatusServiceUnavailable, "evidence_unavailable", "evidence metadata is temporarily unavailable")
+		return
+	}
+	now := time.Now().UTC()
+	if h.Now != nil {
+		now = h.Now().UTC()
+	}
+	if record.ExpiresAt != nil && !record.ExpiresAt.After(now) {
+		problem(c, http.StatusGone, "evidence_expired", "evidence reference has expired")
+		return
+	}
+	c.Data(http.StatusOK, "application/json", record.Payload)
+}
+
+func prepareMachineRequest(c *gin.Context, expectedPath string) bool {
+	requestID := c.GetHeader("X-Request-ID")
+	if requestID == "" {
+		requestID = uuid.NewString()
+	} else if !inboundRequestIDPattern.MatchString(requestID) {
+		problem(c, http.StatusBadRequest, "invalid_request_id", "X-Request-ID is invalid")
+		return false
+	}
+	c.Header("X-Request-ID", requestID)
+	if c.Request.URL.RawQuery != "" || c.Request.URL.Fragment != "" || c.Request.URL.EscapedPath() != expectedPath {
+		problem(c, http.StatusBadRequest, "invalid_request_target", "request target is invalid")
+		return false
+	}
+	return true
+}
+
+func machineCode(err error) string {
+	var machineErr *machine_identity.Error
+	if errors.As(err, &machineErr) {
+		return machineErr.Code
+	}
+	return "machine_token_invalid"
+}
+
+func problem(c *gin.Context, status int, code, message string) {
+	c.Header("Content-Type", "application/problem+json")
+	c.JSON(status, Problem{Code: code, Message: message})
+}
@@ -0,0 +1,130 @@
+package bell_connector
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+var ErrInboundConflict = errors.New("idempotency_conflict")
+
+func AcceptEvent(ctx context.Context, db *gorm.DB, payload []byte, evidenceOwnerID string, now time.Time) (AcceptResult, error) {
+	identity, err := parseEventIdentity(payload)
+	if err != nil {
+		return AcceptResult{}, err
+	}
+	canonical, err := canonicalPayload(payload)
+	if err != nil {
+		return AcceptResult{}, err
+	}
+	digestBytes := sha256.Sum256(canonical)
+	digest := hex.EncodeToString(digestBytes[:])
+	result := AcceptResult{ProducerID: identity.ProducerID, SourceEventID: identity.SourceEventID, PayloadSHA256: digest}
+	err = db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if tx.Dialector.Name() == "postgres" {
+			key := fmt.Sprintf("%d:%s:%s", len(identity.ProducerID), identity.ProducerID, identity.SourceEventID)
+			if lockErr := tx.Exec("SELECT pg_advisory_xact_lock(hashtextextended(?, 0))", key).Error; lockErr != nil {
+				return lockErr
+			}
+		}
+		var existing InboundEvent
+		lookup := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("producer_id = ? AND source_event_id = ?", identity.ProducerID, identity.SourceEventID).First(&existing).Error
+		if lookup == nil {
+			if existing.PayloadSHA256 != digest {
+				return ErrInboundConflict
+			}
+			result.Disposition = "duplicate"
+			return nil
+		}
+		if !errors.Is(lookup, gorm.ErrRecordNotFound) {
+			return lookup
+		}
+		fact := InboundEvent{ID: uuid.NewString(), ProducerID: identity.ProducerID, SourceEventID: identity.SourceEventID, Payload: append([]byte(nil), canonical...), PayloadSHA256: digest, ReceivedAt: now.UTC()}
+		if err := tx.Create(&fact).Error; err != nil {
+			return err
+		}
+		if _, err := EnqueueEvent(tx, canonical, now); err != nil {
+			return err
+		}
+		if err := retainEvidence(tx, canonical, evidenceOwnerID, now); err != nil {
+			return err
+		}
+		result.Disposition = "accepted"
+		return nil
+	})
+	return result, err
+}
+
+func retainEvidence(tx *gorm.DB, payload []byte, ownerID string, now time.Time) error {
+	var envelope struct {
+		Evidence []json.RawMessage `json:"evidence"`
+	}
+	decoder := json.NewDecoder(bytes.NewReader(payload))
+	if err := decoder.Decode(&envelope); err != nil {
+		return err
+	}
+	for _, raw := range envelope.Evidence {
+		var metadata struct {
+			EvidenceID string `json:"evidence_id"`
+			OwnerID    string `json:"owner_id"`
+			Status     string `json:"status"`
+			ExpiresAt  string `json:"expires_at"`
+		}
+		if err := json.Unmarshal(raw, &metadata); err != nil || !safeIdentifier(metadata.EvidenceID) || !safeIdentifier(metadata.OwnerID) {
+			return errors.New("invalid evidence metadata")
+		}
+		if ownerID != "" && metadata.OwnerID != ownerID {
+			continue
+		}
+		canonical, err := canonicalPayload(raw)
+		if err != nil {
+			return err
+		}
+		var expiresAt *time.Time
+		if metadata.ExpiresAt != "" {
+			parsed, parseErr := time.Parse(time.RFC3339Nano, metadata.ExpiresAt)
+			if parseErr != nil {
+				return parseErr
+			}
+			parsed = parsed.UTC()
+			expiresAt = &parsed
+		}
+		record := EvidenceRecord{EvidenceID: metadata.EvidenceID, OwnerID: metadata.OwnerID, Status: metadata.Status, Payload: canonical, ExpiresAt: expiresAt, UpdatedAt: now.UTC()}
+		var existing EvidenceRecord
+		lookup := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&existing, "evidence_id = ?", metadata.EvidenceID).Error
+		if lookup == nil {
+			if existing.OwnerID != metadata.OwnerID || !validEvidenceTransition(existing.Status, metadata.Status) {
+				return errors.New("evidence metadata transition is invalid")
+			}
+		} else if !errors.Is(lookup, gorm.ErrRecordNotFound) {
+			return lookup
+		}
+		if err := tx.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "evidence_id"}}, DoUpdates: clause.AssignmentColumns([]string{"status", "payload", "expires_at", "updated_at"})}).Create(&record).Error; err != nil {
+			return err
+		}
+	}
+	return nil
+}
+
+func validEvidenceTransition(from, to string) bool {
+	if from == to {
+		return true
+	}
+	switch from {
+	case "pending":
+		return to == "processing" || to == "success" || to == "failed"
+	case "processing":
+		return to == "success" || to == "failed"
+	default:
+		return false
+	}
+}
@@ -0,0 +1,79 @@
+package bell_connector
+
+import (
+	"encoding/json"
+	"time"
+)
+
+const OutboxType = "bell_event_v1"
+
+type eventIdentity struct {
+	SchemaVersion string `json:"schema_version"`
+	ProducerID    string `json:"producer_id"`
+	SourceEventID string `json:"source_event_id"`
+}
+
+type IngestResult struct {
+	EventID       string `json:"event_id"`
+	ProducerID    string `json:"producer_id"`
+	SourceEventID string `json:"source_event_id"`
+	Disposition   string `json:"disposition"`
+	PayloadSHA256 string `json:"payload_sha256"`
+}
+
+type Problem struct {
+	Code            string `json:"code"`
+	Message         string `json:"message"`
+	ExistingEventID string `json:"existing_event_id,omitempty"`
+}
+
+type DeliveryError struct {
+	Code       string
+	Detail     string
+	Terminal   bool
+	RetryAfter time.Duration
+}
+
+func (e *DeliveryError) Error() string { return e.Code + ": " + e.Detail }
+
+// ReplayToken is Sense-owned verification state for authenticated evidence
+// requests. It is never shared with Bell's replay table or business receipts.
+type ReplayToken struct {
+	Principal string    `gorm:"size:128;primaryKey"`
+	TokenID   string    `gorm:"size:64;primaryKey"`
+	ExpiresAt time.Time `gorm:"not null;index"`
+	CreatedAt time.Time `gorm:"not null"`
+}
+
+func (ReplayToken) TableName() string { return "sense_machine_token_replays" }
+
+// InboundEvent is the Sense-owned local fact for an event received from Brain.
+// The immutable payload and Bell Outbox row are created in one transaction.
+type InboundEvent struct {
+	ID            string          `gorm:"size:36;primaryKey"`
+	ProducerID    string          `gorm:"size:128;not null;uniqueIndex:sense_inbound_event_key"`
+	SourceEventID string          `gorm:"size:128;not null;uniqueIndex:sense_inbound_event_key"`
+	Payload       json.RawMessage `gorm:"column:payload;type:jsonb;not null"`
+	PayloadSHA256 string          `gorm:"type:char(64);not null"`
+	ReceivedAt    time.Time       `gorm:"not null"`
+}
+
+func (InboundEvent) TableName() string { return "sense_inbound_events" }
+
+type EvidenceRecord struct {
+	EvidenceID string          `gorm:"size:128;primaryKey"`
+	OwnerID    string          `gorm:"size:128;not null;index"`
+	Status     string          `gorm:"size:16;not null;index"`
+	Payload    json.RawMessage `gorm:"column:payload;type:jsonb;not null"`
+	ExpiresAt  *time.Time      `gorm:"index"`
+	UpdatedAt  time.Time       `gorm:"not null"`
+}
+
+func (EvidenceRecord) TableName() string { return "sense_evidence_metadata" }
+
+type AcceptResult struct {
+	ProducerID    string `json:"producer_id"`
+	SourceEventID string `json:"source_event_id"`
+	Disposition   string `json:"disposition"`
+	PayloadSHA256 string `json:"payload_sha256"`
+}
@@ -0,0 +1,166 @@
+package bell_connector
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"strings"
+	"time"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
+)
+
+func EnqueueEvent(tx *gorm.DB, payload []byte, now time.Time) (outbox.Message, error) {
+	identity, err := parseEventIdentity(payload)
+	if err != nil {
+		return outbox.Message{}, err
+	}
+	keyDigest := sha256.Sum256([]byte(identity.ProducerID + "\x00" + identity.SourceEventID))
+	return outbox.Enqueue(tx, outbox.EnqueueInput{
+		InternalType:   OutboxType,
+		BusinessRef:    identity.SourceEventID,
+		IdempotencyKey: "bell-event-v1:" + hex.EncodeToString(keyDigest[:]),
+		PayloadJSON:    append([]byte(nil), payload...),
+	}, now)
+}
+
+func parseEventIdentity(payload []byte) (eventIdentity, error) {
+	decoder := json.NewDecoder(bytes.NewReader(payload))
+	var identity eventIdentity
+	if err := decoder.Decode(&identity); err != nil || !json.Valid(payload) || identity.SchemaVersion != "yovision.event/v1" ||
+		!safeIdentifier(identity.ProducerID) || !safeIdentifier(identity.SourceEventID) {
+		return eventIdentity{}, errors.New("invalid yovision.event/v1 payload")
+	}
+	return identity, nil
+}
+
+func safeIdentifier(value string) bool {
+	if value == "" || len(value) > 128 || strings.ContainsAny(value, "\\/@\x00\r\n") {
+		return false
+	}
+	for index, r := range value {
+		allowed := r >= 'A' && r <= 'Z' || r >= 'a' && r <= 'z' || r >= '0' && r <= '9' || (index > 0 && strings.ContainsRune("._:-", r))
+		if !allowed {
+			return false
+		}
+	}
+	return true
+}
+
+type Relay struct {
+	DB      *gorm.DB
+	Client  *Client
+	Now     func() time.Time
+	Backoff func(int) time.Duration
+}
+
+func (r Relay) DeliverBatch(ctx context.Context, worker string, limit int) (int, error) {
+	if r.DB == nil || r.Client == nil || strings.TrimSpace(worker) == "" || limit < 1 || limit > 100 {
+		return 0, errors.New("invalid Bell relay configuration")
+	}
+	items, err := r.claim(worker, limit)
+	if err != nil {
+		return 0, err
+	}
+	delivered := 0
+	queueRelay := outbox.NewRelay(r.DB)
+	queueRelay.Now = r.now
+	if r.Backoff != nil {
+		queueRelay.Backoff = r.Backoff
+	}
+	for _, item := range items {
+		_, deliveryErr := r.Client.Send(ctx, []byte(item.PayloadJSON))
+		if deliveryErr == nil {
+			if err = queueRelay.MarkSuccess(item.ID, worker); err != nil {
+				return delivered, err
+			}
+			delivered++
+			continue
+		}
+		var classified *DeliveryError
+		if errors.As(deliveryErr, &classified) && classified.Terminal {
+			if err = r.markTerminal(item, worker, classified.Code); err != nil {
+				return delivered, err
+			}
+			continue
+		}
+		if err = queueRelay.MarkFailure(item.ID, worker, deliveryErr.Error()); err != nil {
+			return delivered, err
+		}
+	}
+	return delivered, nil
+}
+
+func (r Relay) claim(worker string, limit int) ([]outbox.Message, error) {
+	now := r.now()
+	leaseUntil := now.Add(30 * time.Second)
+	claimed := make([]outbox.Message, 0, limit)
+	err := r.DB.Transaction(func(tx *gorm.DB) error {
+		var candidates []outbox.Message
+		query := tx.Where("internal_type = ? AND (((state IN ?) AND available_at <= ?) OR (state = ? AND lease_until < ?))", OutboxType, []string{outbox.StatePending, outbox.StateRetry}, now, outbox.StateProcessing, now).Order("available_at, created_at").Limit(limit)
+		if tx.Dialector.Name() == "postgres" {
+			query = query.Clauses(clause.Locking{Strength: "UPDATE", Options: "SKIP LOCKED"})
+		}
+		if err := query.Find(&candidates).Error; err != nil {
+			return err
+		}
+		for _, item := range candidates {
+			result := tx.Model(&outbox.Message{}).Where("id = ? AND version = ?", item.ID, item.Version).Updates(map[string]any{"state": outbox.StateProcessing, "lease_owner": worker, "lease_until": leaseUntil, "version": gorm.Expr("version + 1"), "updated_at": now})
+			if result.Error != nil {
+				return result.Error
+			}
+			if result.RowsAffected == 1 {
+				item.State, item.LeaseOwner, item.LeaseUntil, item.Version = outbox.StateProcessing, worker, &leaseUntil, item.Version+1
+				claimed = append(claimed, item)
+			}
+		}
+		return nil
+	})
+	return claimed, err
+}
+
+func (r Relay) markTerminal(item outbox.Message, worker, detail string) error {
+	now := r.now()
+	return r.DB.Transaction(func(tx *gorm.DB) error {
+		result := tx.Model(&outbox.Message{}).Where("id = ? AND state = ? AND lease_owner = ?", item.ID, outbox.StateProcessing, worker).Updates(map[string]any{
+			"state": outbox.StateDead, "attempt_count": gorm.Expr("attempt_count + 1"), "last_error": detail,
+			"lease_owner": "", "lease_until": nil, "version": gorm.Expr("version + 1"), "updated_at": now,
+		})
+		if result.Error != nil || result.RowsAffected != 1 {
+			if result.Error != nil {
+				return result.Error
+			}
+			return errors.New("Bell outbox lease lost")
+		}
+		return tx.Create(&outbox.Attempt{MessageID: item.ID, Number: item.AttemptCount + 1, Outcome: outbox.StateDead, Detail: detail, Worker: worker, CreatedAt: now}).Error
+	})
+}
+
+func (r Relay) now() time.Time {
+	if r.Now != nil {
+		return r.Now().UTC()
+	}
+	return time.Now().UTC()
+}
+
+func PreserveIdentity(before, after []byte) error {
+	left, err := parseEventIdentity(before)
+	if err != nil {
+		return err
+	}
+	right, err := parseEventIdentity(after)
+	if err != nil {
+		return err
+	}
+	if left.ProducerID != right.ProducerID || left.SourceEventID != right.SourceEventID {
+		return fmt.Errorf("relay changed original event identity")
+	}
+	return nil
+}
@@ -0,0 +1,29 @@
+package bell_connector
+
+import (
+	"time"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+type PersistentReplayStore struct{ DB *gorm.DB }
+
+func (s PersistentReplayStore) Consume(principal, tokenID string, expiresAt, now time.Time) bool {
+	if s.DB == nil {
+		return false
+	}
+	accepted := false
+	err := s.DB.Transaction(func(tx *gorm.DB) error {
+		if err := tx.Where("expires_at <= ?", now.UTC()).Delete(&ReplayToken{}).Error; err != nil {
+			return err
+		}
+		result := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&ReplayToken{Principal: principal, TokenID: tokenID, ExpiresAt: expiresAt.UTC(), CreatedAt: now.UTC()})
+		if result.Error != nil {
+			return result.Error
+		}
+		accepted = result.RowsAffected == 1
+		return nil
+	})
+	return err == nil && accepted
+}
@@ -0,0 +1,89 @@
+package bell_connector
+
+import (
+	"context"
+	"crypto/tls"
+	"errors"
+	"fmt"
+	"log"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"gorm.io/gorm"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
+)
+
+type Environment func(string) string
+
+func StartRuntime(ctx context.Context, engine *gin.Engine, db *gorm.DB, getenv Environment) error {
+	if getenv == nil {
+		getenv = os.Getenv
+	}
+	ingressEnabled := enabled(getenv("SENSE_EVENT_INGRESS_ENABLED"))
+	relayEnabled := enabled(getenv("SENSE_BELL_CONNECTOR_ENABLED"))
+	if !ingressEnabled && !relayEnabled {
+		return nil
+	}
+	if engine == nil || db == nil {
+		return errors.New("Sense connector runtime requires engine and database")
+	}
+	for _, model := range []any{&InboundEvent{}, &EvidenceRecord{}, &ReplayToken{}, &outbox.Message{}, &outbox.DeliveryRecord{}, &outbox.Attempt{}} {
+		if !db.Migrator().HasTable(model) {
+			return fmt.Errorf("Sense connector migration is not applied for %T", model)
+		}
+	}
+	if ingressEnabled {
+		registry, err := machine_identity.LoadRegistry(getenv("SENSE_MACHINE_PRINCIPAL_REGISTRY"), "yovision-sense")
+		if err != nil {
+			return fmt.Errorf("load Sense machine identity registry: %w", err)
+		}
+		verifier := machine_identity.Verifier{Registry: registry, Replay: PersistentReplayStore{DB: db}}
+		engine.POST("/v1/events", (IngressHandler{DB: db, Verifier: verifier, EvidenceOwnerID: strings.TrimSpace(getenv("SENSE_EVIDENCE_OWNER_ID"))}).Post)
+		engine.GET("/v1/evidence/:evidence_id", (EvidenceHandler{DB: db, Verifier: verifier}).Get)
+	}
+	if !relayEnabled {
+		return nil
+	}
+	privateKey, err := machine_identity.LoadPrivateKey(getenv("SENSE_BELL_PRIVATE_KEY_PATH"))
+	if err != nil {
+		return err
+	}
+	signer := machine_identity.Signer{Principal: strings.TrimSpace(getenv("SENSE_BELL_PRINCIPAL_ID")), KeyID: strings.TrimSpace(getenv("SENSE_BELL_KEY_ID")), PrivateKey: privateKey}
+	policy := machine_identity.TransportPolicy{TLSMinVersion: tls.VersionTLS12, VerifyCertificate: true, VerifyHostname: true, ConnectTimeout: 5 * time.Second, ResponseHeaderTimeout: 10 * time.Second, RequestTimeout: 15 * time.Second, MaxRequestBytes: MaxInboundBytes}
+	client, err := NewClient(strings.TrimSpace(getenv("SENSE_BELL_ENDPOINT")), strings.TrimSpace(getenv("SENSE_RELAY_ID")), signer, policy)
+	if err != nil {
+		return err
+	}
+	interval := 2 * time.Second
+	if raw := strings.TrimSpace(getenv("SENSE_BELL_RELAY_INTERVAL_MS")); raw != "" {
+		milliseconds, parseErr := strconv.Atoi(raw)
+		if parseErr != nil || milliseconds < 100 || milliseconds > 60000 {
+			return errors.New("SENSE_BELL_RELAY_INTERVAL_MS must be between 100 and 60000")
+		}
+		interval = time.Duration(milliseconds) * time.Millisecond
+	}
+	go runRelay(ctx, Relay{DB: db, Client: client}, interval)
+	return nil
+}
+
+func runRelay(ctx context.Context, relay Relay, interval time.Duration) {
+	ticker := time.NewTicker(interval)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ctx.Done():
+			return
+		case <-ticker.C:
+			if _, err := relay.DeliverBatch(ctx, "sense-bell-runtime", 50); err != nil && ctx.Err() == nil {
+				log.Printf("Sense Bell connector delivery failed: %v", err)
+			}
+		}
+	}
+}
+
+func enabled(value string) bool { return strings.EqualFold(strings.TrimSpace(value), "true") }
@@ -8,12 +8,12 @@ import (

 	"gorm.io/gorm"

-	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/bell_connector"
 )

 // CreateWithOutbox commits the local candidate and its internal delivery
-// record atomically. The payload remains Sense-internal and is not a Bell or
-// Brain contract.
+// record atomically. The payload is the frozen anonymous event contract; it
+// never carries the candidate's internal evidence path or delivery attempts.
 func CreateWithOutbox(ctx context.Context, db *gorm.DB, candidate EventCandidate, payload map[string]interface{}, now time.Time) error {
 	encoded, err := json.Marshal(payload)
 	if err != nil {
@@ -23,7 +23,7 @@ func CreateWithOutbox(ctx context.Context, db *gorm.DB, candidate EventCandidate
 		if err := tx.Create(&candidate).Error; err != nil {
 			return fmt.Errorf("create local event candidate: %w", err)
 		}
-		_, err = outbox.Enqueue(tx, outbox.EnqueueInput{InternalType: "local_event_candidate", BusinessRef: candidate.ID, IdempotencyKey: "local-event:" + candidate.ID + ":v1", PayloadJSON: encoded}, now)
+		_, err = bell_connector.EnqueueEvent(tx, encoded, now)
 		return err
 	})
 }
@@ -2,6 +2,9 @@ package local_event

 import (
 	"context"
+	"encoding/json"
+	"os"
+	"path/filepath"
 	"testing"
 	"time"

@@ -9,6 +12,7 @@ import (
 	"gorm.io/driver/sqlite"
 	"gorm.io/gorm"

+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/bell_connector"
 	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
 )

@@ -24,7 +28,18 @@ func TestCreateWithOutboxCommitsAndRollsBackAtomically(t *testing.T) {
 	}
 	now := time.Date(2026, 8, 28, 10, 0, 0, 0, time.UTC)
 	candidate := EventCandidate{ID: uuid.NewString(), OccurredAt: now, SourceRef: "SEN-CAM-01", RuleRef: "rule-1", RuleName: "区域闯入", CandidateState: CandidateStateCandidate, EvidenceState: EvidenceStatePending, RetainUntil: now.Add(24 * time.Hour)}
-	if err = CreateWithOutbox(context.Background(), db, candidate, map[string]interface{}{"eventId": candidate.ID}, now); err != nil {
+	fixturePath := filepath.Join("..", "..", "..", "..", "..", "contracts", "events", "v1", "examples", "dangerous-area.json")
+	fixture, err := os.ReadFile(fixturePath)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var payload map[string]interface{}
+	if err = json.Unmarshal(fixture, &payload); err != nil {
+		t.Fatal(err)
+	}
+	payload["producer_id"] = "sense-local"
+	payload["source_event_id"] = candidate.ID
+	if err = CreateWithOutbox(context.Background(), db, candidate, payload, now); err != nil {
 		t.Fatal(err)
 	}
 	var candidates, messages int64
@@ -33,9 +48,13 @@ func TestCreateWithOutboxCommitsAndRollsBackAtomically(t *testing.T) {
 	if candidates != 1 || messages != 1 {
 		t.Fatalf("candidates=%d messages=%d", candidates, messages)
 	}
+	var message outbox.Message
+	if err = db.First(&message).Error; err != nil || message.InternalType != bell_connector.OutboxType {
+		t.Fatalf("local event did not enqueue Bell contract delivery: type=%s err=%v", message.InternalType, err)
+	}
 	duplicate := candidate
 	duplicate.ID = candidate.ID
-	if err = CreateWithOutbox(context.Background(), db, duplicate, map[string]interface{}{"eventId": duplicate.ID}, now); err == nil {
+	if err = CreateWithOutbox(context.Background(), db, duplicate, payload, now); err == nil {
 		t.Fatal("expected duplicate transaction failure")
 	}
 	db.Model(&EventCandidate{}).Count(&candidates)
@@ -0,0 +1,119 @@
+package api
+
+import (
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/json"
+	"os"
+	"path/filepath"
+	"testing"
+
+	"github.com/gin-gonic/gin"
+	"github.com/go-admin-team/go-admin-core/sdk"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/bell_connector"
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
+)
+
+func TestBellConnectorRuntimeWiringIsOptionalAndMigrationGated(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	db, err := gorm.Open(sqlite.Open("file:sense-api-bell-connector?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	engine := gin.New()
+	disabled := func(string) string { return "" }
+	if err = startBellConnectorRuntime(context.Background(), engine, db, disabled); err != nil {
+		t.Fatalf("disabled connector prevented Sense startup: %v", err)
+	}
+	if len(engine.Routes()) != 0 {
+		t.Fatalf("disabled connector registered routes: %#v", engine.Routes())
+	}
+	enabled := func(key string) string {
+		if key == "SENSE_EVENT_INGRESS_ENABLED" {
+			return "true"
+		}
+		return ""
+	}
+	if err = startBellConnectorRuntime(context.Background(), gin.New(), db, enabled); err == nil {
+		t.Fatal("enabled connector started without its formal migration")
+	}
+
+	if err = db.AutoMigrate(
+		&bell_connector.InboundEvent{}, &bell_connector.EvidenceRecord{}, &bell_connector.ReplayToken{},
+		&outbox.Message{}, &outbox.DeliveryRecord{}, &outbox.Attempt{},
+	); err != nil {
+		t.Fatal(err)
+	}
+	registryPath := writeSenseRegistry(t)
+	runtimeEnvironment := func(key string) string {
+		switch key {
+		case "SENSE_EVENT_INGRESS_ENABLED":
+			return "true"
+		case "SENSE_MACHINE_PRINCIPAL_REGISTRY":
+			return registryPath
+		default:
+			return ""
+		}
+	}
+	registered := gin.New()
+	if err = startBellConnectorRuntime(context.Background(), registered, db, runtimeEnvironment); err != nil {
+		t.Fatalf("enabled connector did not register after migration: %v", err)
+	}
+	routes := registered.Routes()
+	if len(routes) != 2 || routes[0].Path != "/v1/events" || routes[1].Path != "/v1/evidence/:evidence_id" {
+		t.Fatalf("unexpected connector routes: %#v", routes)
+	}
+}
+
+func TestConnectorRuntimeUsesOnlyTheExplicitDefaultDatabase(t *testing.T) {
+	defaultDB, err := gorm.Open(sqlite.Open("file:sense-default-runtime?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	secondaryDB, err := gorm.Open(sqlite.Open("file:sense-secondary-runtime?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	sdk.Runtime.SetDb("", defaultDB)
+	sdk.Runtime.SetDb("analytics", secondaryDB)
+	t.Cleanup(func() {
+		sdk.Runtime.SetDb("", nil)
+		sdk.Runtime.SetDb("analytics", nil)
+	})
+	if selected := defaultRuntimeDatabase(); selected != defaultDB {
+		t.Fatalf("runtime selected a non-default database: %p", selected)
+	}
+}
+
+func writeSenseRegistry(t *testing.T) string {
+	t.Helper()
+	publicKey, _, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	document := map[string]any{
+		"version":  "yovision.machine-principal-registry/v1",
+		"audience": "yovision-sense",
+		"principals": []any{map[string]any{
+			"principal_id": "yv:brain:school-a", "enabled": true,
+			"keys": []any{map[string]any{
+				"kid": "brain-key-0001", "public_key_base64url": base64.RawURLEncoding.EncodeToString(publicKey),
+				"status": "active", "scopes": []string{"events:ingest"},
+			}},
+		}},
+	}
+	encoded, err := json.Marshal(document)
+	if err != nil {
+		t.Fatal(err)
+	}
+	path := filepath.Join(t.TempDir(), "sense-registry.json")
+	if err = os.WriteFile(path, encoded, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}
@@ -17,9 +17,11 @@ import (
 	"github.com/go-admin-team/go-admin-core/sdk/pkg"
 	"github.com/pkg/errors"
 	"github.com/spf13/cobra"
+	"gorm.io/gorm"

 	"git.ilapage.cn/ila/yovision/Sense/server/app/admin/models"
 	"git.ilapage.cn/ila/yovision/Sense/server/app/admin/router"
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/bell_connector"
 	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/media"
 	"git.ilapage.cn/ila/yovision/Sense/server/common/database"
 	"git.ilapage.cn/ila/yovision/Sense/server/common/global"
@@ -88,15 +90,19 @@ func run() error {
 	}
 	runtimeCtx, runtimeCancel := context.WithCancel(context.Background())
 	defer runtimeCancel()
-	var runtimeDBFound bool
-	for _, db := range sdk.Runtime.GetDb() {
-		runtimeDBFound = true
+	db := defaultRuntimeDatabase()
+	if db != nil {
+		engine, engineOK := sdk.Runtime.GetEngine().(*gin.Engine)
+		if !engineOK || engine == nil {
+			return errors.New("Sense connector runtime requires Gin engine")
+		}
+		if err := startBellConnectorRuntime(runtimeCtx, engine, db, os.Getenv); err != nil {
+			return fmt.Errorf("Bell connector runtime unavailable: %w", err)
+		}
 		if err := media.StartRuntime(runtimeCtx, db); err != nil {
 			return fmt.Errorf("MediaMTX runtime unavailable: %w", err)
 		}
-		break
-	}
-	if !runtimeDBFound {
+	} else {
 		log.Error("MediaMTX runtime unavailable: Sense database is not initialized")
 	}

@@ -166,6 +172,14 @@ func run() error {
 	return nil
 }

+func startBellConnectorRuntime(ctx context.Context, engine *gin.Engine, db *gorm.DB, getenv bell_connector.Environment) error {
+	return bell_connector.StartRuntime(ctx, engine, db, getenv)
+}
+
+func defaultRuntimeDatabase() *gorm.DB {
+	return sdk.Runtime.GetDbByKey("")
+}
+
 //var Router runtime.Router

 func tip() {
@@ -0,0 +1,30 @@
+package version
+
+import (
+	"runtime"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/bell_connector"
+	"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
+	common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
+)
+
+func init() {
+	_, fileName, _, _ := runtime.Caller(0)
+	migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSenseBellConnector)
+}
+
+func migrateSenseBellConnector(db *gorm.DB, version string) error {
+	return db.Transaction(func(tx *gorm.DB) error {
+		if err := tx.AutoMigrate(
+			&bell_connector.InboundEvent{},
+			&bell_connector.EvidenceRecord{},
+			&bell_connector.ReplayToken{},
+		); err != nil {
+			return err
+		}
+		return tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&common.Migration{Version: version}).Error
+	})
+}
@@ -0,0 +1,66 @@
+package version
+
+import (
+	"testing"
+	"time"
+
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/bell_connector"
+	common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
+)
+
+func TestSenseBellConnectorMigrationIsIdempotent(t *testing.T) {
+	db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&common.Migration{}); err != nil {
+		t.Fatal(err)
+	}
+
+	const version = "2026083112000"
+	for attempt := 0; attempt < 2; attempt++ {
+		if err = migrateSenseBellConnector(db, version); err != nil {
+			t.Fatalf("migration attempt %d: %v", attempt+1, err)
+		}
+	}
+
+	for name, model := range map[string]any{
+		"inbound events":   &bell_connector.InboundEvent{},
+		"evidence records": &bell_connector.EvidenceRecord{},
+		"replay tokens":    &bell_connector.ReplayToken{},
+	} {
+		if !db.Migrator().HasTable(model) {
+			t.Fatalf("%s table missing", name)
+		}
+		var count int64
+		if err = db.Model(model).Count(&count).Error; err != nil {
+			t.Fatalf("count %s: %v", name, err)
+		}
+		if count != 0 {
+			t.Fatalf("migration inserted %d %s fixtures", count, name)
+		}
+	}
+	if !db.Migrator().HasIndex(&bell_connector.ReplayToken{}, "ExpiresAt") {
+		t.Fatal("replay expiry index missing")
+	}
+
+	now := time.Now().UTC()
+	token := bell_connector.ReplayToken{Principal: "sense", TokenID: "token-1", ExpiresAt: now.Add(time.Minute), CreatedAt: now}
+	if err = db.Create(&token).Error; err != nil {
+		t.Fatal(err)
+	}
+	if err = db.Create(&token).Error; err == nil {
+		t.Fatal("duplicate replay token accepted")
+	}
+
+	var applied int64
+	if err = db.Model(&common.Migration{}).Where("version = ?", version).Count(&applied).Error; err != nil {
+		t.Fatal(err)
+	}
+	if applied != 1 {
+		t.Fatalf("migration records=%d, want 1", applied)
+	}
+}