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

Bell/server/app/bell/integration/event_ingress/evidence.go

@@ -0,0 +1,117 @@
+package event_ingress
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/tls"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"strings"
+	"time"
+
+	"gorm.io/gorm"
+
+	"go-admin/app/bell/integration/machine_identity"
+)
+
+type EvidenceClient struct {
+	Endpoint string
+	Signer   machine_identity.Signer
+	HTTP     interface {
+		Do(*http.Request) (*http.Response, error)
+	}
+}
+
+func NewEvidenceClient(endpoint string, signer machine_identity.Signer) (*EvidenceClient, 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("Sense evidence endpoint must be an HTTPS origin without userinfo")
+	}
+	transport := &http.Transport{TLSClientConfig: &tls.Config{MinVersion: tls.VersionTLS12}, TLSHandshakeTimeout: 5 * time.Second, ResponseHeaderTimeout: 5 * time.Second}
+	return &EvidenceClient{Endpoint: strings.TrimRight(endpoint, "/"), Signer: signer, HTTP: &http.Client{Transport: transport, Timeout: 8 * time.Second}}, nil
+}
+
+func (c EvidenceClient) Refresh(ctx context.Context, db *gorm.DB, status EvidenceStatus) error {
+	path := "/v1/evidence/" + status.EvidenceID
+	token, err := c.Signer.Mint("yovision-sense", []string{"evidence:read"}, http.MethodGet, path, nil)
+	if err != nil {
+		return c.degrade(db, status, "unavailable", "machine_identity_error")
+	}
+	request, err := http.NewRequestWithContext(ctx, http.MethodGet, c.Endpoint+path, nil)
+	if err != nil {
+		return c.degrade(db, status, "unavailable", "invalid_request")
+	}
+	request.Header.Set("Authorization", "Bearer "+token)
+	request.Header.Set("X-Request-ID", newCorrelationID())
+	response, err := c.HTTP.Do(request)
+	if err != nil {
+		code := "evidence_unavailable"
+		if errors.Is(err, context.DeadlineExceeded) || errors.Is(ctx.Err(), context.DeadlineExceeded) {
+			code = "evidence_timeout"
+		}
+		return c.degrade(db, status, "unavailable", code)
+	}
+	defer response.Body.Close()
+	body, readErr := io.ReadAll(io.LimitReader(response.Body, 64*1024+1))
+	if readErr != nil || len(body) > 64*1024 {
+		return c.degrade(db, status, "unavailable", "invalid_evidence_response")
+	}
+	if response.StatusCode == http.StatusNotFound {
+		return c.degrade(db, status, "unavailable", "evidence_not_found")
+	}
+	if response.StatusCode == http.StatusGone {
+		return c.degrade(db, status, "expired", "evidence_expired")
+	}
+	if response.StatusCode != http.StatusOK {
+		return c.degrade(db, status, "unavailable", "evidence_unavailable")
+	}
+	decoder := json.NewDecoder(bytes.NewReader(body))
+	decoder.DisallowUnknownFields()
+	var evidence Evidence
+	if err = decoder.Decode(&evidence); err != nil || evidence.EvidenceID != status.EvidenceID || evidence.OwnerID != status.OwnerID || validateEvidence(evidence) != nil {
+		return c.degrade(db, status, "unavailable", "invalid_evidence_response")
+	}
+	canonical, err := canonicalJSON(body)
+	if err != nil {
+		return c.degrade(db, status, "unavailable", "invalid_evidence_response")
+	}
+	now := time.Now().UTC()
+	var expiresAt *time.Time
+	if evidence.ExpiresAt != "" {
+		parsedExpiry, parseErr := time.Parse(time.RFC3339Nano, evidence.ExpiresAt)
+		if parseErr != nil {
+			return c.degrade(db, status, "unavailable", "invalid_evidence_response")
+		}
+		parsedExpiry = parsedExpiry.UTC()
+		expiresAt = &parsedExpiry
+	}
+	return db.Model(&EvidenceStatus{}).Where("event_id = ? AND evidence_id = ?", status.EventID, status.EvidenceID).Updates(map[string]any{"status": evidence.Status, "resolution": "current", "current_payload": canonical, "last_error": "", "expires_at": expiresAt, "checked_at": now, "updated_at": now}).Error
+}
+
+func (c EvidenceClient) degrade(db *gorm.DB, status EvidenceStatus, resolution, code string) error {
+	now := time.Now().UTC()
+	return db.Model(&EvidenceStatus{}).Where("event_id = ? AND evidence_id = ?", status.EventID, status.EvidenceID).Updates(map[string]any{"resolution": resolution, "last_error": code, "checked_at": now, "updated_at": now}).Error
+}
+
+func newCorrelationID() string {
+	raw := make([]byte, 16)
+	if _, err := rand.Read(raw); err != nil {
+		return "request-id-fallback"
+	}
+	return base64.RawURLEncoding.EncodeToString(raw)
+}
+
+func LoadEvidenceClient(getenv func(string) string) (*EvidenceClient, error) {
+	key, err := machine_identity.LoadPrivateKey(getenv("BELL_SENSE_PRIVATE_KEY_PATH"))
+	if err != nil {
+		return nil, fmt.Errorf("load Bell evidence key: %w", err)
+	}
+	signer := machine_identity.Signer{Principal: strings.TrimSpace(getenv("BELL_SENSE_PRINCIPAL_ID")), KeyID: strings.TrimSpace(getenv("BELL_SENSE_KEY_ID")), PrivateKey: key}
+	return NewEvidenceClient(strings.TrimSpace(getenv("BELL_SENSE_EVIDENCE_ENDPOINT")), signer)
+}

Bell/server/app/bell/integration/event_ingress/handler.go

@@ -0,0 +1,107 @@
+package event_ingress
+
+import (
+	"errors"
+	"io"
+	"net/http"
+	"regexp"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+
+	"go-admin/app/bell/integration/machine_identity"
+)
+
+const MaxRequestBytes = 64 * 1024
+
+var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$`)
+
+type Handler struct {
+	DB       *gorm.DB
+	Verifier machine_identity.Verifier
+	Enabled  bool
+	Resolver EvidenceRefresher
+}
+
+func (h Handler) Post(c *gin.Context) {
+	if !h.Enabled {
+		writeProblem(c, http.StatusServiceUnavailable, "connector_disabled", "event connector is disabled", "")
+		return
+	}
+	requestID := c.GetHeader("X-Request-ID")
+	if requestID == "" {
+		requestID = uuid.NewString()
+	} else if !requestIDPattern.MatchString(requestID) {
+		writeProblem(c, http.StatusBadRequest, "invalid_request_id", "X-Request-ID must be an opaque 16-128 character value", "")
+		return
+	}
+	c.Header("X-Request-ID", requestID)
+	if c.Request.URL.RawQuery != "" || c.Request.URL.Fragment != "" || c.Request.URL.EscapedPath() != "/v1/events" {
+		writeProblem(c, http.StatusBadRequest, "invalid_request_target", "event request target must be the normalized /v1/events path", "")
+		return
+	}
+	if relayHeader := c.GetHeader("X-YoVision-Relay-ID"); relayHeader != "" {
+		relayID := strings.TrimSpace(relayHeader)
+		if relayID != relayHeader || !validID(relayID) {
+			writeProblem(c, http.StatusBadRequest, "invalid_event", "relay identity header is invalid", "")
+			return
+		}
+	}
+	body, err := io.ReadAll(http.MaxBytesReader(c.Writer, c.Request.Body, MaxRequestBytes))
+	if err != nil {
+		writeProblem(c, http.StatusBadRequest, "invalid_event", "event payload is invalid or too large", "")
+		return
+	}
+	token, err := machine_identity.BearerToken(c.GetHeader("Authorization"))
+	if err != nil {
+		writeProblem(c, http.StatusUnauthorized, machineErrorCode(err), "machine identity was rejected", "")
+		return
+	}
+	if _, err = h.Verifier.Verify(token, "yovision-bell", "events:ingest", c.Request.Method, c.Request.URL.EscapedPath(), body); err != nil {
+		status := http.StatusUnauthorized
+		code := machineErrorCode(err)
+		if code == "machine_scope_denied" || code == "machine_audience_denied" {
+			status = http.StatusForbidden
+		}
+		writeProblem(c, status, code, "machine identity was rejected", "")
+		return
+	}
+	parsed, err := ParseEvent(body)
+	if err != nil {
+		if errors.Is(err, ErrUnsupportedSchema) {
+			writeProblem(c, http.StatusUnprocessableEntity, "unsupported_schema_version", "event schema version is unsupported", "")
+			return
+		}
+		writeProblem(c, http.StatusBadRequest, "invalid_event", "event payload failed validation", "")
+		return
+	}
+	result, err := (Service{DB: h.DB, Resolver: h.Resolver}).Ingest(c.Request.Context(), parsed)
+	if errors.Is(err, ErrIdempotencyConflict) {
+		writeProblem(c, http.StatusConflict, "idempotency_conflict", "idempotency key is already bound to another payload", result.EventID)
+		return
+	}
+	if err != nil {
+		writeProblem(c, http.StatusServiceUnavailable, "ingest_unavailable", "event ingest is temporarily unavailable", "")
+		return
+	}
+	status := http.StatusCreated
+	if result.Disposition == "duplicate" {
+		status = http.StatusOK
+	}
+	c.JSON(status, result)
+}
+
+func writeProblem(c *gin.Context, status int, code, message, existing string) {
+	c.Header("Content-Type", "application/problem+json")
+	c.JSON(status, Problem{Code: code, Message: message, ExistingEventID: existing})
+}
+
+func machineErrorCode(err error) string {
+	var machineErr *machine_identity.Error
+	if errors.As(err, &machineErr) {
+		return machineErr.Code
+	}
+	return "machine_token_invalid"
+}

Bell/server/app/bell/integration/event_ingress/model.go

@@ -0,0 +1,116 @@
+package event_ingress
+
+import (
+	"encoding/json"
+	"time"
+)
+
+const (
+	EventSchemaVersion    = "yovision.event/v1"
+	EvidenceSchemaVersion = "yovision.evidence-reference/v1"
+)
+
+type Event struct {
+	SchemaVersion string      `json:"schema_version"`
+	ProducerID    string      `json:"producer_id"`
+	SourceEventID string      `json:"source_event_id"`
+	SiteRef       string      `json:"site_ref"`
+	DeviceRef     string      `json:"device_ref"`
+	ProfileRef    string      `json:"profile_ref"`
+	EventType     string      `json:"event_type"`
+	OccurredAt    string      `json:"occurred_at"`
+	Severity      string      `json:"severity"`
+	Rule          Rule        `json:"rule"`
+	Model         Model       `json:"model"`
+	Observation   Observation `json:"observation"`
+	Region        Region      `json:"region"`
+	Evidence      []Evidence  `json:"evidence"`
+}
+
+type Rule struct {
+	RuleID  string `json:"rule_id"`
+	Version string `json:"version"`
+}
+
+type Model struct {
+	Name    string `json:"name"`
+	Version string `json:"version"`
+}
+
+type Observation struct {
+	TrackID        string    `json:"track_id"`
+	Category       string    `json:"category"`
+	Confidence     float64   `json:"confidence"`
+	BBoxNormalized []float64 `json:"bbox_normalized,omitempty"`
+}
+
+type Region struct {
+	RegionID          string `json:"region_id"`
+	Kind              string `json:"kind"`
+	CrossingDirection string `json:"crossing_direction,omitempty"`
+}
+
+type Evidence struct {
+	SchemaVersion   string             `json:"schema_version"`
+	EvidenceID      string             `json:"evidence_id"`
+	OwnerID         string             `json:"owner_id"`
+	Type            string             `json:"type"`
+	Status          string             `json:"status"`
+	CapturedAt      string             `json:"captured_at"`
+	StatusUpdatedAt string             `json:"status_updated_at"`
+	ExpiresAt       string             `json:"expires_at,omitempty"`
+	ContentType     string             `json:"content_type,omitempty"`
+	Integrity       *EvidenceIntegrity `json:"integrity,omitempty"`
+	Failure         *EvidenceFailure   `json:"failure,omitempty"`
+}
+
+type EvidenceIntegrity struct {
+	Algorithm string `json:"algorithm"`
+	Digest    string `json:"digest"`
+	SizeBytes int64  `json:"size_bytes"`
+}
+
+type EvidenceFailure struct {
+	Code      string `json:"code"`
+	Retryable bool   `json:"retryable"`
+}
+
+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"`
+	Field           string `json:"field,omitempty"`
+	ExistingEventID string `json:"existing_event_id,omitempty"`
+}
+
+type ParsedEvent struct {
+	Event     Event
+	Canonical json.RawMessage
+	Digest    string
+	Occurred  time.Time
+}
+
+// EvidenceStatus is mutable Bell-owned resolution metadata kept separately
+// from the immutable Event and from Alert acknowledgement/close facts.
+type EvidenceStatus struct {
+	EventID        string          `gorm:"type:uuid;primaryKey"`
+	EvidenceID     string          `gorm:"size:128;primaryKey"`
+	OwnerID        string          `gorm:"size:128;not null;index"`
+	Status         string          `gorm:"size:16;not null"`
+	Resolution     string          `gorm:"size:16;not null;index"`
+	CurrentPayload json.RawMessage `gorm:"column:current_payload;type:jsonb;not null"`
+	LastError      string          `gorm:"size:64;not null;default:''"`
+	ExpiresAt      *time.Time      `gorm:"index"`
+	CheckedAt      *time.Time
+	CreatedAt      time.Time `gorm:"not null"`
+	UpdatedAt      time.Time `gorm:"not null"`
+}
+
+func (EvidenceStatus) TableName() string { return "bell_evidence_status" }

Bell/server/app/bell/integration/event_ingress/replay.go

@@ -0,0 +1,42 @@
+package event_ingress
+
+import (
+	"time"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+// ReplayToken is Bell-owned security state. It is intentionally separate from
+// business Receipt idempotency and remains effective across process restarts.
+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 "bell_machine_token_replays" }
+
+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
+}

Bell/server/app/bell/integration/event_ingress/runtime.go

@@ -0,0 +1,40 @@
+package event_ingress
+
+import (
+	"fmt"
+	"os"
+	"strings"
+
+	"github.com/gin-gonic/gin"
+	"github.com/go-admin-team/go-admin-core/sdk"
+
+	"go-admin/app/bell/integration/machine_identity"
+)
+
+func RegisterRuntime(engine *gin.Engine) error {
+	enabled := strings.EqualFold(strings.TrimSpace(os.Getenv("BELL_EVENT_INGRESS_ENABLED")), "true")
+	if !enabled {
+		return nil
+	}
+	db := sdk.Runtime.GetDbByKey("")
+	if db == nil {
+		return fmt.Errorf("Bell event ingress database is unavailable")
+	}
+	if !db.Migrator().HasTable(&ReplayToken{}) || !db.Migrator().HasTable(&EvidenceStatus{}) {
+		return fmt.Errorf("Bell event ingress migration is required")
+	}
+	registry, err := machine_identity.LoadRegistry(os.Getenv("BELL_MACHINE_PRINCIPAL_REGISTRY"), "yovision-bell")
+	if err != nil {
+		return fmt.Errorf("load Bell machine identity registry: %w", err)
+	}
+	var resolver EvidenceRefresher
+	if strings.EqualFold(strings.TrimSpace(os.Getenv("BELL_EVIDENCE_RESOLVER_ENABLED")), "true") {
+		resolver, err = LoadEvidenceClient(os.Getenv)
+		if err != nil {
+			return fmt.Errorf("load Bell evidence resolver: %w", err)
+		}
+	}
+	handler := Handler{DB: db, Enabled: true, Resolver: resolver, Verifier: machine_identity.Verifier{Registry: registry, Replay: PersistentReplayStore{DB: db}}}
+	engine.POST("/v1/events", handler.Post)
+	return nil
+}

Bell/server/app/bell/integration/event_ingress/service.go

@@ -0,0 +1,125 @@
+package event_ingress
+
+import (
+	"context"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"time"
+
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+
+	"go-admin/app/bell/event"
+	"go-admin/app/bell/receipt"
+)
+
+var ErrIdempotencyConflict = errors.New("idempotency_conflict")
+
+type EvidenceRefresher interface {
+	Refresh(context.Context, *gorm.DB, EvidenceStatus) error
+}
+
+type Service struct {
+	DB       *gorm.DB
+	Resolver EvidenceRefresher
+}
+
+func (s Service) Ingest(ctx context.Context, parsed ParsedEvent) (IngestResult, error) {
+	if s.DB == nil {
+		return IngestResult{}, errors.New("event database is unavailable")
+	}
+	var output IngestResult
+	err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
+		if tx.Dialector.Name() == "postgres" {
+			key := fmt.Sprintf("%d:%s:%s", len(parsed.Event.ProducerID), parsed.Event.ProducerID, parsed.Event.SourceEventID)
+			if err := tx.Exec("SELECT pg_advisory_xact_lock(hashtextextended(?, 0))", key).Error; err != nil {
+				return err
+			}
+		}
+		var existing struct {
+			EventID       string
+			PayloadSHA256 string
+		}
+		err := tx.Model(&receipt.Receipt{}).Clauses(clause.Locking{Strength: "UPDATE"}).
+			Select("event_id", "payload_sha256").Where("producer_id = ? AND source_event_id = ?", parsed.Event.ProducerID, parsed.Event.SourceEventID).First(&existing).Error
+		if err == nil {
+			if existing.PayloadSHA256 != parsed.Digest {
+				output.EventID = existing.EventID
+				return ErrIdempotencyConflict
+			}
+			output = IngestResult{EventID: existing.EventID, ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, Disposition: "duplicate", PayloadSHA256: parsed.Digest}
+			return tx.Create(&receipt.IngestAudit{ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, PayloadSHA256: parsed.Digest, Outcome: receipt.OutcomeReplay, ActorID: 0, CreatedAt: time.Now().UTC()}).Error
+		}
+		if !errors.Is(err, gorm.ErrRecordNotFound) {
+			return err
+		}
+		now := time.Now().UTC()
+		eventID := uuid.NewString()
+		var evidenceRef *string
+		if len(parsed.Event.Evidence) > 0 {
+			value := parsed.Event.Evidence[0].EvidenceID
+			evidenceRef = &value
+		}
+		item := event.Event{ID: eventID, ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID,
+			EventType: parsed.Event.EventType, OccurredAt: parsed.Occurred, Location: parsed.Event.SiteRef + "/" + parsed.Event.DeviceRef,
+			Severity: parsed.Event.Severity, EvidenceRef: evidenceRef, NormalizedPayload: parsed.Canonical, PayloadSHA256: parsed.Digest, ReceivedAt: now}
+		receiptItem := receipt.Receipt{ID: uuid.NewString(), EventID: eventID, ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, PayloadSHA256: parsed.Digest, AcceptedAt: now}
+		if err := tx.Create(&item).Error; err != nil {
+			return err
+		}
+		if err := tx.Create(&receiptItem).Error; err != nil {
+			return err
+		}
+		for _, evidence := range parsed.Event.Evidence {
+			payload, marshalErr := json.Marshal(evidence)
+			if marshalErr != nil {
+				return marshalErr
+			}
+			canonical, canonicalErr := canonicalJSON(payload)
+			if canonicalErr != nil {
+				return canonicalErr
+			}
+			var expiresAt *time.Time
+			if evidence.ExpiresAt != "" {
+				parsedExpiry, parseErr := time.Parse(time.RFC3339Nano, evidence.ExpiresAt)
+				if parseErr != nil {
+					return parseErr
+				}
+				parsedExpiry = parsedExpiry.UTC()
+				expiresAt = &parsedExpiry
+			}
+			status := EvidenceStatus{EventID: eventID, EvidenceID: evidence.EvidenceID, OwnerID: evidence.OwnerID,
+				Status: evidence.Status, Resolution: "snapshot", CurrentPayload: canonical, ExpiresAt: expiresAt, CreatedAt: now, UpdatedAt: now}
+			if err := tx.Create(&status).Error; err != nil {
+				return err
+			}
+		}
+		if err := tx.Create(&receipt.IngestAudit{ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, PayloadSHA256: parsed.Digest, Outcome: receipt.OutcomeAccepted, ActorID: 0, CreatedAt: now}).Error; err != nil {
+			return err
+		}
+		output = IngestResult{EventID: eventID, ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, Disposition: "created", PayloadSHA256: parsed.Digest}
+		return nil
+	})
+	if errors.Is(err, ErrIdempotencyConflict) {
+		auditErr := s.DB.WithContext(ctx).Create(&receipt.IngestAudit{ProducerID: parsed.Event.ProducerID, SourceEventID: parsed.Event.SourceEventID, PayloadSHA256: parsed.Digest, Outcome: receipt.OutcomeConflict, ActorID: 0, CreatedAt: time.Now().UTC()}).Error
+		if auditErr != nil {
+			return IngestResult{}, fmt.Errorf("record conflict audit: %w", auditErr)
+		}
+		return output, ErrIdempotencyConflict
+	}
+	if err != nil || s.Resolver == nil {
+		return output, err
+	}
+	var statuses []EvidenceStatus
+	if err = s.DB.WithContext(ctx).Where("event_id = ?", output.EventID).Find(&statuses).Error; err != nil {
+		return IngestResult{}, err
+	}
+	for _, status := range statuses {
+		// Evidence lookup is supplementary. The immutable Event/Receipt boundary
+		// remains accepted even when Sense is unavailable.
+		_ = s.Resolver.Refresh(ctx, s.DB.WithContext(ctx), status)
+	}
+	return output, nil
+}

Bell/server/app/bell/integration/event_ingress/validation.go

@@ -0,0 +1,217 @@
+package event_ingress
+
+import (
+	"bytes"
+	"crypto/sha256"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"io"
+	"math"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+)
+
+var (
+	ErrInvalidEvent      = errors.New("invalid_event")
+	ErrUnsupportedSchema = errors.New("unsupported_schema_version")
+	identifierPattern    = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$`)
+	hexDigestPattern     = regexp.MustCompile(`^[a-f0-9]{64}$`)
+)
+
+func ParseEvent(raw []byte) (ParsedEvent, error) {
+	var event Event
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&event); err != nil {
+		return ParsedEvent{}, fmt.Errorf("%w: malformed or unknown member", ErrInvalidEvent)
+	}
+	if err := decoder.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
+		return ParsedEvent{}, fmt.Errorf("%w: trailing JSON value", ErrInvalidEvent)
+	}
+	if event.SchemaVersion == "" {
+		return ParsedEvent{}, ErrInvalidEvent
+	}
+	if event.SchemaVersion != EventSchemaVersion {
+		return ParsedEvent{}, ErrUnsupportedSchema
+	}
+	occurred, err := time.Parse("2006-01-02T15:04:05.000Z", event.OccurredAt)
+	if err != nil || !validID(event.ProducerID) || !validID(event.SourceEventID) || !validID(event.SiteRef) ||
+		!validID(event.DeviceRef) || !validID(event.ProfileRef) || !validID(event.Rule.RuleID) ||
+		!validID(event.Observation.TrackID) || !validID(event.Region.RegionID) || event.Rule.Version == "" ||
+		len(event.Rule.Version) > 64 || event.Model.Name == "" || len(event.Model.Name) > 128 ||
+		event.Model.Version == "" || len(event.Model.Version) > 64 || event.Observation.Confidence < 0 ||
+		event.Observation.Confidence > 1 || math.IsNaN(event.Observation.Confidence) || math.IsInf(event.Observation.Confidence, 0) {
+		return ParsedEvent{}, ErrInvalidEvent
+	}
+	if event.EventType != "dangerous_area_entered" && event.EventType != "directional_line_crossed" {
+		return ParsedEvent{}, ErrInvalidEvent
+	}
+	if event.Severity != "low" && event.Severity != "medium" && event.Severity != "high" && event.Severity != "critical" {
+		return ParsedEvent{}, ErrInvalidEvent
+	}
+	if event.Observation.Category != "person" && event.Observation.Category != "vehicle" && event.Observation.Category != "other" {
+		return ParsedEvent{}, ErrInvalidEvent
+	}
+	if len(event.Observation.BBoxNormalized) != 0 && len(event.Observation.BBoxNormalized) != 4 {
+		return ParsedEvent{}, ErrInvalidEvent
+	}
+	for _, value := range event.Observation.BBoxNormalized {
+		if value < 0 || value > 1 || math.IsNaN(value) || math.IsInf(value, 0) {
+			return ParsedEvent{}, ErrInvalidEvent
+		}
+	}
+	if (event.EventType == "dangerous_area_entered" && (event.Region.Kind != "area" || event.Region.CrossingDirection != "")) ||
+		(event.EventType == "directional_line_crossed" && (event.Region.Kind != "line" || (event.Region.CrossingDirection != "a_to_b" && event.Region.CrossingDirection != "b_to_a"))) {
+		return ParsedEvent{}, ErrInvalidEvent
+	}
+	if event.Evidence == nil || len(event.Evidence) > 8 {
+		return ParsedEvent{}, ErrInvalidEvent
+	}
+	seenEvidence := map[string]bool{}
+	for _, evidence := range event.Evidence {
+		evidenceJSON, marshalErr := json.Marshal(evidence)
+		canonicalEvidence, canonicalErr := canonicalJSON(evidenceJSON)
+		if err := validateEvidence(evidence); err != nil || marshalErr != nil || canonicalErr != nil || seenEvidence[string(canonicalEvidence)] {
+			return ParsedEvent{}, ErrInvalidEvent
+		}
+		seenEvidence[string(canonicalEvidence)] = true
+	}
+	canonical, err := canonicalJSON(raw)
+	if err != nil {
+		return ParsedEvent{}, ErrInvalidEvent
+	}
+	if containsExplicitNull(raw) {
+		return ParsedEvent{}, fmt.Errorf("%w: optional members must be omitted", ErrInvalidEvent)
+	}
+	digest := sha256.Sum256(canonical)
+	return ParsedEvent{Event: event, Canonical: canonical, Digest: hex.EncodeToString(digest[:]), Occurred: occurred}, nil
+}
+
+func validateEvidence(value Evidence) error {
+	if value.SchemaVersion != EvidenceSchemaVersion || !validID(value.EvidenceID) || !validID(value.OwnerID) ||
+		(value.Type != "snapshot" && value.Type != "clip") {
+		return ErrInvalidEvent
+	}
+	if _, err := time.Parse(time.RFC3339Nano, value.CapturedAt); err != nil {
+		return ErrInvalidEvent
+	}
+	if _, err := time.Parse(time.RFC3339Nano, value.StatusUpdatedAt); err != nil {
+		return ErrInvalidEvent
+	}
+	if value.ExpiresAt != "" {
+		if _, err := time.Parse(time.RFC3339Nano, value.ExpiresAt); err != nil {
+			return ErrInvalidEvent
+		}
+	}
+	switch value.Status {
+	case "pending", "processing":
+		if value.ContentType != "" || value.Integrity != nil || value.Failure != nil {
+			return ErrInvalidEvent
+		}
+	case "success":
+		if value.Integrity == nil || value.Failure != nil || (value.ContentType != "image/jpeg" && value.ContentType != "image/png" && value.ContentType != "video/mp4") ||
+			value.Integrity.Algorithm != "sha256" || !hexDigestPattern.MatchString(value.Integrity.Digest) || value.Integrity.SizeBytes < 0 {
+			return ErrInvalidEvent
+		}
+	case "failed":
+		if value.Failure == nil || value.ContentType != "" || value.Integrity != nil ||
+			(value.Failure.Code != "capture_failed" && value.Failure.Code != "processing_failed" && value.Failure.Code != "expired" && value.Failure.Code != "unavailable") {
+			return ErrInvalidEvent
+		}
+	default:
+		return ErrInvalidEvent
+	}
+	return nil
+}
+
+func canonicalJSON(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
+}
+
+func containsExplicitNull(raw []byte) bool {
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	decoder.UseNumber()
+	var value any
+	if decoder.Decode(&value) != nil {
+		return true
+	}
+	return hasNull(value)
+}
+
+func hasNull(value any) bool {
+	switch typed := value.(type) {
+	case nil:
+		return true
+	case []any:
+		for _, item := range typed {
+			if hasNull(item) {
+				return true
+			}
+		}
+	case map[string]any:
+		for _, item := range typed {
+			if hasNull(item) {
+				return true
+			}
+		}
+	}
+	return false
+}
+
+func validID(value string) bool {
+	return identifierPattern.MatchString(value) && !strings.ContainsAny(strings.ToLower(value), "\\/@")
+}

Bell/server/app/bell/router/registry.go

@@ -9,6 +9,7 @@ import (
 	"github.com/go-admin-team/go-admin-core/sdk/config"
 	jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
 
+	"go-admin/app/bell/integration/event_ingress"
 	"go-admin/app/bell/synthetic"
 	"go-admin/common/middleware"
 )
@@ -32,6 +33,9 @@ func InitRouter() {
 	for _, register := range registrars {
 		register(v1, authMiddleware)
 	}
+	if err := event_ingress.RegisterRuntime(engine); err != nil {
+		log.Errorf("Bell event ingress init error: %v", err)
+	}
 	if synthetic.Enabled(config.ApplicationConfig.Mode, os.Getenv) {
 		registerSyntheticRouter(v1, authMiddleware)
 	}

Bell/server/cmd/migrate/migration/version/2026083112000_event_ingress.go

@@ -0,0 +1,29 @@
+package version
+
+import (
+	"runtime"
+
+	"go-admin/app/bell/integration/event_ingress"
+	"go-admin/cmd/migrate/migration"
+	common "go-admin/common/models"
+
+	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
+)
+
+func init() {
+	_, fileName, _, _ := runtime.Caller(0)
+	migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateBellEventIngress)
+}
+
+func migrateBellEventIngress(db *gorm.DB, version string) error {
+	return db.Transaction(func(tx *gorm.DB) error {
+		if err := tx.AutoMigrate(
+			&event_ingress.ReplayToken{},
+			&event_ingress.EvidenceStatus{},
+		); err != nil {
+			return err
+		}
+		return tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&common.Migration{Version: version}).Error
+	})
+}

Bell/server/cmd/migrate/migration/version/2026083112000_event_ingress_test.go

@@ -0,0 +1,65 @@
+package version
+
+import (
+	"testing"
+	"time"
+
+	"go-admin/app/bell/integration/event_ingress"
+	common "go-admin/common/models"
+
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+)
+
+func TestBellEventIngressMigrationIsIdempotent(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 = migrateBellEventIngress(db, version); err != nil {
+			t.Fatalf("migration attempt %d: %v", attempt+1, err)
+		}
+	}
+
+	for name, model := range map[string]any{
+		"replay tokens":     &event_ingress.ReplayToken{},
+		"evidence statuses": &event_ingress.EvidenceStatus{},
+	} {
+		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(&event_ingress.ReplayToken{}, "ExpiresAt") {
+		t.Fatal("replay expiry index missing")
+	}
+
+	now := time.Now().UTC()
+	token := event_ingress.ReplayToken{Principal: "brain", 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)
+	}
+}

Bell/server/tests/integration/event_ingress/ingress_test.go

@@ -0,0 +1,404 @@
+package event_ingress_test
+
+import (
+	"bytes"
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/json"
+	"io"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"github.com/go-admin-team/go-admin-core/sdk"
+	"gorm.io/driver/postgres"
+	"gorm.io/driver/sqlite"
+	"gorm.io/gorm"
+
+	"go-admin/app/bell/event"
+	"go-admin/app/bell/integration/event_ingress"
+	"go-admin/app/bell/integration/machine_identity"
+	"go-admin/app/bell/receipt"
+)
+
+func TestPostgresConcurrentBusinessAndSecurityIdempotency(t *testing.T) {
+	dsn := os.Getenv("BELL_EVENT_INGRESS_TEST_DATABASE_URL")
+	if dsn == "" {
+		t.Skip("set BELL_EVENT_INGRESS_TEST_DATABASE_URL to run PostgreSQL concurrency verification")
+	}
+	db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&event.Event{}, &receipt.Receipt{}, &receipt.IngestAudit{}, &event_ingress.ReplayToken{}, &event_ingress.EvidenceStatus{}); err != nil {
+		t.Fatal(err)
+	}
+	parsed, err := event_ingress.ParseEvent(fixture(t, "dangerous-area.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	const workers = 12
+	var created, duplicate, failures atomic.Int32
+	var wait sync.WaitGroup
+	for range workers {
+		wait.Add(1)
+		go func() {
+			defer wait.Done()
+			result, ingestErr := (event_ingress.Service{DB: db}).Ingest(context.Background(), parsed)
+			if ingestErr != nil {
+				failures.Add(1)
+				return
+			}
+			if result.Disposition == "created" {
+				created.Add(1)
+			} else if result.Disposition == "duplicate" {
+				duplicate.Add(1)
+			}
+		}()
+	}
+	wait.Wait()
+	if created.Load() != 1 || duplicate.Load() != workers-1 || failures.Load() != 0 {
+		t.Fatalf("concurrent ingest created=%d duplicate=%d failures=%d", created.Load(), duplicate.Load(), failures.Load())
+	}
+
+	now := time.Now().UTC()
+	var consumed atomic.Int32
+	for range workers {
+		wait.Add(1)
+		go func() {
+			defer wait.Done()
+			if (event_ingress.PersistentReplayStore{DB: db}).Consume("yv:sense:school-a", "concurrent-token-id-0001", now.Add(time.Minute), now) {
+				consumed.Add(1)
+			}
+		}()
+	}
+	wait.Wait()
+	if consumed.Load() != 1 {
+		t.Fatalf("concurrent replay consume accepted %d requests", consumed.Load())
+	}
+}
+
+func TestRuntimeRegistrationIsOptionalAndMigrationGated(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	t.Setenv("BELL_EVENT_INGRESS_ENABLED", "")
+	disabled := gin.New()
+	if err := event_ingress.RegisterRuntime(disabled); err != nil || len(disabled.Routes()) != 0 {
+		t.Fatalf("disabled runtime err=%v routes=%#v", err, disabled.Routes())
+	}
+
+	db, err := gorm.Open(sqlite.Open("file:bell-runtime?mode=memory&cache=shared"), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	sdk.Runtime.SetDb("", db)
+	t.Cleanup(func() { sdk.Runtime.SetDb("", nil) })
+	t.Setenv("BELL_EVENT_INGRESS_ENABLED", "true")
+	if err = event_ingress.RegisterRuntime(gin.New()); err == nil {
+		t.Fatal("enabled runtime started without formal migration")
+	}
+	if err = db.AutoMigrate(&event_ingress.ReplayToken{}, &event_ingress.EvidenceStatus{}); err != nil {
+		t.Fatal(err)
+	}
+	registryPath := writeRegistry(t, "yovision-bell", "yv:sense:school-a", "sense-key-0001")
+	t.Setenv("BELL_MACHINE_PRINCIPAL_REGISTRY", registryPath)
+	registered := gin.New()
+	if err = event_ingress.RegisterRuntime(registered); err != nil {
+		t.Fatalf("enabled runtime did not register after migration: %v", err)
+	}
+	routes := registered.Routes()
+	if len(routes) != 1 || routes[0].Method != http.MethodPost || routes[0].Path != "/v1/events" {
+		t.Fatalf("unexpected ingress routes: %#v", routes)
+	}
+}
+
+func TestContractFixtureIdempotencyConflictAndReplayPersistence(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	databasePath := filepath.Join(t.TempDir(), "bell-ingress.sqlite")
+	db := openDatabasePath(t, databasePath)
+	body := fixture(t, "dangerous-area.json")
+	publicKey, privateKey, err := ed25519.GenerateKey(rand.Reader)
+	if err != nil {
+		t.Fatal(err)
+	}
+	registry, err := machine_identity.NewRegistry(machine_identity.KeyRecord{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PublicKey: publicKey, Audience: "yovision-bell", Scopes: []string{"events:ingest"}, Enabled: true})
+	if err != nil {
+		t.Fatal(err)
+	}
+	now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
+	signer := machine_identity.Signer{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PrivateKey: privateKey, Now: func() time.Time { return now }}
+	newHandler := func() event_ingress.Handler {
+		return event_ingress.Handler{DB: db, Enabled: true, Verifier: machine_identity.Verifier{Registry: registry, Replay: event_ingress.PersistentReplayStore{DB: db}, Now: func() time.Time { return now }}}
+	}
+
+	firstToken := mint(t, signer, body)
+	first := request(t, newHandler(), body, firstToken)
+	if first.Code != http.StatusCreated {
+		t.Fatalf("first ingest status=%d body=%s", first.Code, first.Body.String())
+	}
+	var created event_ingress.IngestResult
+	decode(t, first, &created)
+	if created.Disposition != "created" || created.PayloadSHA256 != "4cc1e93820195caf713ea675ff33f178c9d4997dd8a81cb61287e9fea0e3d5e1" {
+		t.Fatalf("unexpected created result: %+v", created)
+	}
+	sqlDatabase, err := db.DB()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = sqlDatabase.Close(); err != nil {
+		t.Fatal(err)
+	}
+	db = openDatabasePath(t, databasePath)
+
+	// A new process-local Handler and replay store still reject the old token,
+	// proving that security replay state is durable rather than in-memory.
+	replayedToken := request(t, newHandler(), body, firstToken)
+	if replayedToken.Code != http.StatusUnauthorized || !strings.Contains(replayedToken.Body.String(), "machine_token_replayed") {
+		t.Fatalf("token replay status=%d body=%s", replayedToken.Code, replayedToken.Body.String())
+	}
+
+	duplicate := request(t, newHandler(), body, mint(t, signer, body))
+	if duplicate.Code != http.StatusOK {
+		t.Fatalf("business duplicate status=%d body=%s", duplicate.Code, duplicate.Body.String())
+	}
+	var duplicateResult event_ingress.IngestResult
+	decode(t, duplicate, &duplicateResult)
+	if duplicateResult.Disposition != "duplicate" || duplicateResult.EventID != created.EventID {
+		t.Fatalf("duplicate did not retain event identity: %+v", duplicateResult)
+	}
+	numericVariant := bytes.Replace(body, []byte(`0.93`), []byte(`0.930`), 1)
+	numericDuplicate := request(t, newHandler(), numericVariant, mint(t, signer, numericVariant))
+	if numericDuplicate.Code != http.StatusOK || !strings.Contains(numericDuplicate.Body.String(), created.PayloadSHA256) {
+		t.Fatalf("JCS-equivalent numeric payload was not a duplicate: %d %s", numericDuplicate.Code, numericDuplicate.Body.String())
+	}
+
+	var changed map[string]any
+	if err = json.Unmarshal(body, &changed); err != nil {
+		t.Fatal(err)
+	}
+	changed["severity"] = "critical"
+	conflicting, _ := json.Marshal(changed)
+	conflict := request(t, newHandler(), conflicting, mint(t, signer, conflicting))
+	if conflict.Code != http.StatusConflict || !strings.Contains(conflict.Body.String(), "idempotency_conflict") || !strings.Contains(conflict.Body.String(), created.EventID) {
+		t.Fatalf("conflict status=%d body=%s", conflict.Code, conflict.Body.String())
+	}
+	assertCount(t, db, &event.Event{}, 1)
+	assertCount(t, db, &receipt.Receipt{}, 1)
+	assertCount(t, db, &receipt.IngestAudit{}, 4)
+}
+
+func TestEvidenceDegradationIdentityErrorsAndDisabledConnector(t *testing.T) {
+	db := openDatabase(t)
+	publicKey, privateKey, _ := ed25519.GenerateKey(rand.Reader)
+	registry, _ := machine_identity.NewRegistry(machine_identity.KeyRecord{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PublicKey: publicKey, Audience: "yovision-bell", Scopes: []string{"events:ingest"}, Enabled: true})
+	now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
+	signer := machine_identity.Signer{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PrivateKey: privateKey, Now: func() time.Time { return now }}
+	handler := event_ingress.Handler{DB: db, Enabled: true, Verifier: machine_identity.Verifier{Registry: registry, Replay: event_ingress.PersistentReplayStore{DB: db}, Now: func() time.Time { return now }}}
+
+	pending := fixture(t, "dangerous-area.json")
+	if response := requestWithID(t, handler, pending, mint(t, signer, pending), "short"); response.Code != http.StatusBadRequest || !strings.Contains(response.Body.String(), "invalid_request_id") {
+		t.Fatalf("invalid request id status=%d body=%s", response.Code, response.Body.String())
+	}
+	missingRequestID := requestWithID(t, handler, pending, mint(t, signer, pending), "")
+	if missingRequestID.Code != http.StatusCreated || !requestIDPatternForTest(missingRequestID.Header().Get("X-Request-ID")) {
+		t.Fatalf("trusted hop did not create a request id: %d %s", missingRequestID.Code, missingRequestID.Body.String())
+	}
+	queryResponse := requestTarget(t, handler, pending, mint(t, signer, pending), "/v1/events?debug=true")
+	if queryResponse.Code != http.StatusBadRequest || !strings.Contains(queryResponse.Body.String(), "invalid_request_target") {
+		t.Fatalf("query target was accepted: %d %s", queryResponse.Code, queryResponse.Body.String())
+	}
+	if response := request(t, handler, pending, mint(t, signer, pending)); response.Code != http.StatusOK {
+		t.Fatalf("pending evidence rejected: %d %s", response.Code, response.Body.String())
+	}
+	failed := fixture(t, "directional-line-crossed.json")
+	if response := request(t, handler, failed, mint(t, signer, failed)); response.Code != http.StatusCreated {
+		t.Fatalf("failed evidence rejected: %d %s", response.Code, response.Body.String())
+	}
+	assertCount(t, db, &event.Event{}, 2)
+
+	wrongAudienceToken, err := signer.Mint("yovision-sense", []string{"events:ingest"}, http.MethodPost, "/v1/events", pending)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if response := request(t, handler, pending, wrongAudienceToken); response.Code != http.StatusForbidden {
+		t.Fatalf("wrong audience was not forbidden: %d %s", response.Code, response.Body.String())
+	}
+	if response := request(t, event_ingress.Handler{Enabled: false}, pending, "none"); response.Code != http.StatusServiceUnavailable {
+		t.Fatalf("disabled connector status=%d", response.Code)
+	}
+}
+
+func TestEvidenceResolverCurrentMissingExpiredAndTimeout(t *testing.T) {
+	db := openDatabase(t)
+	now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
+	status := event_ingress.EvidenceStatus{EventID: "event-1", EvidenceID: "ev-school-east-0001", OwnerID: "sense-school-a", Status: "pending", Resolution: "snapshot", CurrentPayload: json.RawMessage(`{"schema_version":"yovision.evidence-reference/v1","evidence_id":"ev-school-east-0001","owner_id":"sense-school-a","type":"snapshot","status":"pending","captured_at":"2026-08-31T00:00:01.125Z","status_updated_at":"2026-08-31T00:00:01.125Z"}`), CreatedAt: now, UpdatedAt: now}
+	if err := db.Create(&status).Error; err != nil {
+		t.Fatal(err)
+	}
+	_, privateKey, _ := ed25519.GenerateKey(rand.Reader)
+	signer := machine_identity.Signer{Principal: "yv:bell:school-a", KeyID: "bell-key-0001", PrivateKey: privateKey, Now: func() time.Time { return now }}
+	response := func(code int, body string) *http.Response {
+		return &http.Response{StatusCode: code, Body: io.NopCloser(strings.NewReader(body)), Header: make(http.Header)}
+	}
+	client := event_ingress.EvidenceClient{Endpoint: "https://sense.example", Signer: signer, HTTP: doFunc(func(*http.Request) (*http.Response, error) {
+		return response(http.StatusNotFound, `{}`), nil
+	})}
+	if err := client.Refresh(context.Background(), db, status); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.First(&status, "event_id = ? AND evidence_id = ?", "event-1", "ev-school-east-0001").Error; err != nil || status.Resolution != "unavailable" || status.LastError != "evidence_not_found" {
+		t.Fatalf("missing resolution=%s error=%s db=%v", status.Resolution, status.LastError, err)
+	}
+	client.HTTP = doFunc(func(*http.Request) (*http.Response, error) { return response(http.StatusGone, `{}`), nil })
+	if err := client.Refresh(context.Background(), db, status); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.First(&status, "event_id = ? AND evidence_id = ?", "event-1", "ev-school-east-0001").Error; err != nil || status.Resolution != "expired" {
+		t.Fatalf("expired resolution=%s db=%v", status.Resolution, err)
+	}
+	current := `{"schema_version":"yovision.evidence-reference/v1","evidence_id":"ev-school-east-0001","owner_id":"sense-school-a","type":"snapshot","status":"success","captured_at":"2026-08-31T00:00:01.125Z","status_updated_at":"2026-08-31T00:00:02.125Z","content_type":"image/jpeg","integrity":{"algorithm":"sha256","digest":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","size_bytes":1}}`
+	client.HTTP = doFunc(func(*http.Request) (*http.Response, error) { return response(http.StatusOK, current), nil })
+	if err := client.Refresh(context.Background(), db, status); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.First(&status, "event_id = ? AND evidence_id = ?", "event-1", "ev-school-east-0001").Error; err != nil || status.Resolution != "current" || status.Status != "success" || status.LastError != "" {
+		t.Fatalf("current status=%s resolution=%s error=%s db=%v", status.Status, status.Resolution, status.LastError, err)
+	}
+	client.HTTP = doFunc(func(*http.Request) (*http.Response, error) { return nil, context.DeadlineExceeded })
+	if err := client.Refresh(context.Background(), db, status); err != nil {
+		t.Fatal(err)
+	}
+	if err := db.First(&status, "event_id = ? AND evidence_id = ?", "event-1", "ev-school-east-0001").Error; err != nil || status.Resolution != "unavailable" || status.LastError != "evidence_timeout" {
+		t.Fatalf("timeout resolution=%s error=%s db=%v", status.Resolution, status.LastError, err)
+	}
+}
+
+func openDatabase(t *testing.T) *gorm.DB {
+	return openDatabasePath(t, filepath.Join(t.TempDir(), "bell-ingress.sqlite"))
+}
+
+func openDatabasePath(t *testing.T, databasePath string) *gorm.DB {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open(databasePath), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&event.Event{}, &receipt.Receipt{}, &receipt.IngestAudit{}, &event_ingress.ReplayToken{}, &event_ingress.EvidenceStatus{}); err != nil {
+		t.Fatal(err)
+	}
+	sqlDatabase, err := db.DB()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = sqlDatabase.Close() })
+	return db
+}
+
+func fixture(t *testing.T, name string) []byte {
+	t.Helper()
+	path := filepath.Join("..", "..", "..", "..", "..", "contracts", "events", "v1", "examples", name)
+	body, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return body
+}
+
+func writeRegistry(t *testing.T, audience, principal, keyID string) 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": audience,
+		"principals": []any{map[string]any{
+			"principal_id": principal, "enabled": true,
+			"keys": []any{map[string]any{
+				"kid": keyID, "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(), "registry.json")
+	if err = os.WriteFile(path, encoded, 0o600); err != nil {
+		t.Fatal(err)
+	}
+	return path
+}
+
+func mint(t *testing.T, signer machine_identity.Signer, body []byte) string {
+	t.Helper()
+	token, err := signer.Mint("yovision-bell", []string{"events:ingest"}, http.MethodPost, "/v1/events", body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return token
+}
+
+func request(t *testing.T, handler event_ingress.Handler, body []byte, token string) *httptest.ResponseRecorder {
+	return requestWithID(t, handler, body, token, "request-id-0000001")
+}
+
+func requestWithID(t *testing.T, handler event_ingress.Handler, body []byte, token, requestID string) *httptest.ResponseRecorder {
+	return requestTargetWithID(t, handler, body, token, "/v1/events", requestID)
+}
+
+func requestTarget(t *testing.T, handler event_ingress.Handler, body []byte, token, target string) *httptest.ResponseRecorder {
+	return requestTargetWithID(t, handler, body, token, target, "request-id-0000001")
+}
+
+func requestTargetWithID(t *testing.T, handler event_ingress.Handler, body []byte, token, target, requestID string) *httptest.ResponseRecorder {
+	t.Helper()
+	request := httptest.NewRequest(http.MethodPost, target, bytes.NewReader(body))
+	request.Header.Set("Authorization", "Bearer "+token)
+	request.Header.Set("X-Request-ID", requestID)
+	response := httptest.NewRecorder()
+	context, _ := gin.CreateTestContext(response)
+	context.Request = request
+	handler.Post(context)
+	return response
+}
+
+func requestIDPatternForTest(value string) bool {
+	if len(value) < 16 || len(value) > 128 {
+		return false
+	}
+	for index, r := range value {
+		if !(r >= 'A' && r <= 'Z' || r >= 'a' && r <= 'z' || r >= '0' && r <= '9' || index > 0 && strings.ContainsRune("._:-", r)) {
+			return false
+		}
+	}
+	return true
+}
+
+func decode(t *testing.T, response *httptest.ResponseRecorder, target any) {
+	t.Helper()
+	if err := json.Unmarshal(response.Body.Bytes(), target); err != nil {
+		t.Fatal(err)
+	}
+}
+
+func assertCount(t *testing.T, db *gorm.DB, model any, expected int64) {
+	t.Helper()
+	var count int64
+	if err := db.Model(model).Count(&count).Error; err != nil || count != expected {
+		t.Fatalf("count %T=%d expected=%d err=%v", model, count, expected, err)
+	}
+}
+
+type doFunc func(*http.Request) (*http.Response, error)
+
+func (function doFunc) Do(request *http.Request) (*http.Response, error) { return function(request) }

Brain/src/yovision_brain/app/__main__.py

@@ -11,6 +11,7 @@ from yovision_brain.config import ConfigError
 from yovision_brain.decode import DecoderError
 from yovision_brain.events import JsonLinesSink
 from yovision_brain.input import InputError
+from yovision_brain.integration.event_export import build_event_export_sink
 from yovision_brain.rules import RuleConfigError
 
 from .runner import run_pipeline
@@ -38,14 +39,16 @@ def main(argv: list[str] | None = None) -> int:
     stream = sys.stdout
     owned_stream = None
     try:
-        if args.output != "-":
+        export_sink = build_event_export_sink(raw.get("event_export"), base_dir=config_path.parent)
+        if export_sink is None and args.output != "-":
             try:
                 owned_stream = Path(args.output).open("w", encoding="utf-8", newline="\n")
             except OSError:
                 print(json.dumps({"status": "error", "message": "event output cannot be opened"}), file=sys.stderr)
                 return 2
             stream = owned_stream
-        summary = run_pipeline(raw, JsonLinesSink(stream), base_dir=config_path.parent)
+        sink = export_sink if export_sink is not None else JsonLinesSink(stream)
+        summary = run_pipeline(raw, sink, base_dir=config_path.parent)
     except (ConfigError, DecoderError, InputError, RuleConfigError, RuntimeError, ValueError) as exc:
         print(json.dumps({"status": "error", "message": str(exc)}, ensure_ascii=False), file=sys.stderr)
         return 3

Brain/src/yovision_brain/integration/event_export/__init__.py

@@ -0,0 +1,30 @@
+"""Map Brain-internal candidates to the frozen anonymous event contract."""
+
+from .mapper import (
+    EVENT_SCHEMA_VERSION,
+    EVIDENCE_SCHEMA_VERSION,
+    EventExportError,
+    canonical_json,
+    canonical_json_bytes,
+    export_event,
+    payload_sha256,
+)
+from .replay import SQLiteReplayCache
+from .client import DeliveryResult, EventDeliveryError, HTTPSMachineIdentitySender
+from .runtime import EventExportSink, build_event_export_sink
+
+__all__ = [
+    "EVENT_SCHEMA_VERSION",
+    "EVIDENCE_SCHEMA_VERSION",
+    "EventExportError",
+    "EventDeliveryError",
+    "DeliveryResult",
+    "EventExportSink",
+    "HTTPSMachineIdentitySender",
+    "SQLiteReplayCache",
+    "canonical_json",
+    "canonical_json_bytes",
+    "export_event",
+    "payload_sha256",
+    "build_event_export_sink",
+]

Brain/src/yovision_brain/integration/event_export/client.py

@@ -0,0 +1,167 @@
+"""Synchronous, request-bound HTTPS delivery for Brain event exports."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import re
+import secrets
+import urllib.error
+import urllib.request
+from dataclasses import dataclass
+from typing import Protocol
+from urllib.parse import urlsplit
+
+from yovision_brain.integration.machine_identity import Signer, TransportPolicy
+
+EVENT_PATH = "/v1/events"
+_REQUEST_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$")
+
+
+class EventDeliveryError(RuntimeError):
+    """An event was not accepted; callers must retain or reproduce the fact."""
+
+    def __init__(self, code: str, *, terminal: bool) -> None:
+        super().__init__(code)
+        self.code = code
+        self.terminal = terminal
+
+
+@dataclass(frozen=True, slots=True)
+class DeliveryResult:
+    disposition: str
+    request_id: str
+
+
+class _Headers(Protocol):
+    def get(self, name: str, default: str | None = None) -> str | None: ...
+
+
+class _Response(Protocol):
+    status: int
+    headers: _Headers
+
+    def read(self, amount: int = -1) -> bytes: ...
+    def close(self) -> None: ...
+
+
+class _Opener(Protocol):
+    def open(self, request: urllib.request.Request, timeout: float) -> _Response: ...
+
+
+class HTTPSMachineIdentitySender:
+    def __init__(
+        self,
+        endpoint: str,
+        signer: Signer,
+        policy: TransportPolicy,
+        *,
+        opener: _Opener | None = None,
+    ) -> None:
+        parsed = urlsplit(endpoint)
+        if (
+            parsed.scheme != "https"
+            or not parsed.hostname
+            or parsed.username is not None
+            or parsed.password is not None
+            or parsed.path not in {"", "/"}
+            or parsed.query
+            or parsed.fragment
+        ):
+            raise ValueError("event export endpoint must be an HTTPS origin")
+        policy.validate()
+        self._endpoint = endpoint.rstrip("/")
+        self._signer = signer
+        self._policy = policy
+        self._opener = opener or urllib.request.build_opener(
+            urllib.request.HTTPSHandler(context=policy.ssl_context())
+        )
+
+    def send(self, body: bytes) -> DeliveryResult:
+        if len(body) > self._policy.max_request_bytes:
+            raise EventDeliveryError("event_request_too_large", terminal=True)
+        request_id = "req-" + secrets.token_urlsafe(16)
+        if not _REQUEST_ID.fullmatch(request_id):  # pragma: no cover - defensive invariant
+            raise RuntimeError("generated request id is invalid")
+        token = self._signer.mint(
+            "yovision-sense", ("events:ingest",), "POST", EVENT_PATH, body
+        )
+        request = urllib.request.Request(
+            self._endpoint + EVENT_PATH,
+            data=body,
+            method="POST",
+            headers={
+                "Authorization": "Bearer " + token,
+                "Content-Type": "application/json",
+                "X-Request-ID": request_id,
+            },
+        )
+        try:
+            response = self._opener.open(
+                request, timeout=self._policy.request_timeout_ms / 1000
+            )
+        except urllib.error.HTTPError as exc:
+            response_body = exc.read(64 * 1024 + 1)
+            code = _problem_code(response_body) or "event_delivery_rejected"
+            raise EventDeliveryError(
+                code,
+                terminal=400 <= exc.code < 500 and exc.code != 429,
+            ) from None
+        except (OSError, TimeoutError, urllib.error.URLError):
+            raise EventDeliveryError("event_delivery_unavailable", terminal=False) from None
+
+        try:
+            response_body = response.read(64 * 1024 + 1)
+            if len(response_body) > 64 * 1024:
+                raise EventDeliveryError("event_response_invalid", terminal=False)
+            if response.status not in {200, 201, 202}:
+                raise EventDeliveryError(
+                    "event_delivery_rejected",
+                    terminal=400 <= response.status < 500 and response.status != 429,
+                )
+            response_request_id = response.headers.get("X-Request-ID")
+            if response_request_id != request_id:
+                raise EventDeliveryError("event_response_invalid", terminal=False)
+            disposition = _disposition(response_body, response.status, body)
+            return DeliveryResult(disposition=disposition, request_id=request_id)
+        finally:
+            response.close()
+
+
+def _problem_code(body: bytes) -> str | None:
+    try:
+        value = json.loads(body)
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        return None
+    code = value.get("code") if isinstance(value, dict) else None
+    return code if isinstance(code, str) and re.fullmatch(r"[a-z][a-z0-9_]{0,63}", code) else None
+
+
+def _disposition(body: bytes, status: int, request_body: bytes) -> str:
+    try:
+        value = json.loads(body)
+    except (UnicodeDecodeError, json.JSONDecodeError):
+        raise EventDeliveryError("event_response_invalid", terminal=False) from None
+    disposition = value.get("disposition") if isinstance(value, dict) else None
+    allowed = {"accepted", "created", "duplicate"}
+    if disposition not in allowed:
+        raise EventDeliveryError("event_response_invalid", terminal=False)
+    if status == 202 and disposition != "accepted":
+        raise EventDeliveryError("event_response_invalid", terminal=False)
+    try:
+        sent = json.loads(request_body)
+        response_identity = (
+            value["producer_id"],
+            value["source_event_id"],
+            value["payload_sha256"],
+        )
+        expected_identity = (
+            sent["producer_id"],
+            sent["source_event_id"],
+            hashlib.sha256(request_body).hexdigest(),
+        )
+    except (KeyError, TypeError, UnicodeDecodeError, json.JSONDecodeError):
+        raise EventDeliveryError("event_response_invalid", terminal=False) from None
+    if response_identity != expected_identity:
+        raise EventDeliveryError("event_response_invalid", terminal=False)
+    return disposition

Brain/src/yovision_brain/integration/event_export/mapper.py

@@ -0,0 +1,378 @@
+"""Safe, deterministic Brain producer mapping for ``yovision.event/v1``."""
+
+from __future__ import annotations
+
+import hashlib
+import json
+import math
+import re
+from datetime import datetime, timedelta, timezone
+from decimal import Decimal
+from typing import Mapping, Sequence
+
+from yovision_brain.events import INTERNAL_EVENT_SCHEMA, InternalEventCandidate
+
+EVENT_SCHEMA_VERSION = "yovision.event/v1"
+EVIDENCE_SCHEMA_VERSION = "yovision.evidence-reference/v1"
+
+_REFERENCE = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
+_SHA256 = re.compile(r"^[a-f0-9]{64}$")
+_URL = re.compile(r"^[A-Za-z][A-Za-z0-9+.-]*://")
+_WINDOWS_PATH = re.compile(r"^[A-Za-z]:[\\/]")
+_SENSITIVE_NAMES = frozenset(
+    {
+        "path",
+        "url",
+        "uri",
+        "password",
+        "secret",
+        "token",
+        "credential",
+        "signed_url",
+        "camera_url",
+        "face",
+        "face_id",
+        "face_template",
+        "alert",
+        "ack",
+        "close",
+        "notification",
+    }
+)
+_EVIDENCE_FIELDS = frozenset(
+    {
+        "schema_version",
+        "evidence_id",
+        "owner_id",
+        "type",
+        "status",
+        "captured_at",
+        "status_updated_at",
+        "expires_at",
+        "content_type",
+        "integrity",
+        "failure",
+    }
+)
+
+
+class EventExportError(ValueError):
+    """The internal candidate cannot safely satisfy the frozen contract."""
+
+
+def export_event(
+    candidate: InternalEventCandidate,
+    *,
+    producer_id: str,
+    site_ref: str,
+    device_ref: str | None = None,
+    severity: str,
+    evidence: Sequence[Mapping[str, object]] = (),
+    region_ref: str | None = None,
+    crossing_direction: str | None = None,
+    category: str | None = None,
+) -> dict[str, object]:
+    """Return a new closed v1 payload without mutating the internal candidate.
+
+    The candidate's already stable ``event_id`` is the source identity. Callers
+    must persist and retry the returned payload unchanged; transport attempts do
+    not participate in either identity field.
+    """
+
+    if candidate.schema != INTERNAL_EVENT_SCHEMA:
+        raise EventExportError("unsupported internal event candidate schema")
+    producer_id = _reference("producer_id", producer_id)
+    source_event_id = _reference("source_event_id", candidate.event_id)
+    site_ref = _reference("site_ref", site_ref)
+    device_ref = _reference("device_ref", device_ref or candidate.logical_input_id)
+    profile_ref = _reference("profile_ref", candidate.profile_id)
+    rule_id = _reference("rule.rule_id", candidate.rule_id)
+    region_id = _reference("region.region_id", region_ref or candidate.rule_id)
+    track_id = _reference("observation.track_id", candidate.track_id)
+
+    event_type = {
+        "danger_area_entered": "dangerous_area_entered",
+        "dangerous_area_entered": "dangerous_area_entered",
+        "directional_line_crossed": "directional_line_crossed",
+    }.get(candidate.event_type)
+    if event_type is None:
+        raise EventExportError("unsupported event type")
+    if severity not in {"low", "medium", "high", "critical"}:
+        raise EventExportError("unsupported severity")
+
+    observation = _observation(candidate, track_id=track_id, category=category)
+    region: dict[str, object] = {
+        "region_id": region_id,
+        "kind": "area" if event_type == "dangerous_area_entered" else "line",
+    }
+    if event_type == "directional_line_crossed":
+        if crossing_direction not in {"a_to_b", "b_to_a"}:
+            raise EventExportError("line events require a contract crossing_direction")
+        region["crossing_direction"] = crossing_direction
+    elif crossing_direction is not None:
+        raise EventExportError("area events cannot carry crossing_direction")
+
+    if len(evidence) > 8:
+        raise EventExportError("at most eight evidence references are allowed")
+    mapped_evidence = [_evidence_reference(item) for item in evidence]
+    if len({canonical_json_bytes(item) for item in mapped_evidence}) != len(mapped_evidence):
+        raise EventExportError("duplicate evidence references are not allowed")
+
+    payload: dict[str, object] = {
+        "schema_version": EVENT_SCHEMA_VERSION,
+        "producer_id": producer_id,
+        "source_event_id": source_event_id,
+        "site_ref": site_ref,
+        "device_ref": device_ref,
+        "profile_ref": profile_ref,
+        "event_type": event_type,
+        "occurred_at": _event_timestamp(candidate.occurred_at_ns),
+        "severity": severity,
+        "rule": {"rule_id": rule_id, "version": _bounded_text("rule.version", candidate.rule_version, 64)},
+        "model": {
+            "name": _bounded_text("model.name", candidate.model_name, 128),
+            "version": _bounded_text("model.version", candidate.model_version, 64),
+        },
+        "observation": observation,
+        "region": region,
+        "evidence": mapped_evidence,
+    }
+    _reject_unsafe(payload)
+    canonical_json_bytes(payload)  # Reject non-finite or unsupported values now.
+    return payload
+
+
+def canonical_json(value: object) -> str:
+    """Serialize the closed event-domain JCS subset used by frozen fixtures.
+
+    Contract values use JSON strings, containers, booleans, integers and finite
+    ordinary decimals. Integer-valued floats are normalized to their JSON number
+    form; the checked-in RFC 8785 vector fixes cross-language digest behavior.
+    """
+
+    return _encode_jcs(_normalize_numbers(value))
+
+
+def canonical_json_bytes(value: object) -> bytes:
+    return canonical_json(value).encode("utf-8")
+
+
+def payload_sha256(event: Mapping[str, object]) -> str:
+    return hashlib.sha256(canonical_json_bytes(event)).hexdigest()
+
+
+def _observation(
+    candidate: InternalEventCandidate, *, track_id: str, category: str | None
+) -> dict[str, object]:
+    internal = candidate.observation
+    confidence = internal.get("confidence")
+    if isinstance(confidence, bool) or not isinstance(confidence, (int, float)):
+        raise EventExportError("observation confidence must be numeric")
+    confidence = float(confidence)
+    if not math.isfinite(confidence) or not 0 <= confidence <= 1:
+        raise EventExportError("observation confidence must be finite and between zero and one")
+
+    internal_category = internal.get("category")
+    exported_category = category or {
+        "anonymous_target": "person",
+        "person": "person",
+        "vehicle": "vehicle",
+        "other": "other",
+    }.get(internal_category)
+    if exported_category not in {"person", "vehicle", "other"}:
+        raise EventExportError("observation category requires an explicit anonymous contract mapping")
+
+    result: dict[str, object] = {
+        "track_id": track_id,
+        "category": exported_category,
+        "confidence": confidence,
+    }
+    box = internal.get("box")
+    if box is not None:
+        if not isinstance(box, Mapping) or set(box) != {"left", "top", "right", "bottom"}:
+            raise EventExportError("internal observation box is malformed")
+        if candidate.frame_width <= 0 or candidate.frame_height <= 0:
+            raise EventExportError("frame dimensions must be positive")
+        coordinates = (box["left"], box["top"], box["right"], box["bottom"])
+        if any(isinstance(value, bool) or not isinstance(value, (int, float)) for value in coordinates):
+            raise EventExportError("box coordinates must be numeric")
+        normalized = [
+            float(coordinates[0]) / candidate.frame_width,
+            float(coordinates[1]) / candidate.frame_height,
+            float(coordinates[2]) / candidate.frame_width,
+            float(coordinates[3]) / candidate.frame_height,
+        ]
+        if any(not math.isfinite(value) or not 0 <= value <= 1 for value in normalized):
+            raise EventExportError("normalized box coordinates must be finite and between zero and one")
+        result["bbox_normalized"] = normalized
+    return result
+
+
+def _evidence_reference(source: Mapping[str, object]) -> dict[str, object]:
+    if not isinstance(source, Mapping):
+        raise EventExportError("evidence reference must be an object")
+    unknown = set(source) - _EVIDENCE_FIELDS
+    if unknown:
+        raise EventExportError(f"evidence reference contains forbidden fields: {sorted(unknown)!r}")
+    required = {
+        "schema_version",
+        "evidence_id",
+        "owner_id",
+        "type",
+        "status",
+        "captured_at",
+        "status_updated_at",
+    }
+    missing = required - set(source)
+    if missing:
+        raise EventExportError(f"evidence reference is missing fields: {sorted(missing)!r}")
+    result = dict(source)
+    if result["schema_version"] != EVIDENCE_SCHEMA_VERSION:
+        raise EventExportError("unsupported evidence schema version")
+    _reference("evidence.evidence_id", result["evidence_id"])
+    _reference("evidence.owner_id", result["owner_id"])
+    if result["type"] not in {"snapshot", "clip"}:
+        raise EventExportError("unsupported evidence type")
+    status = result["status"]
+    if status not in {"pending", "processing", "success", "failed"}:
+        raise EventExportError("unsupported evidence status")
+    for field in ("captured_at", "status_updated_at", "expires_at"):
+        if field in result:
+            _date_time(field, result[field])
+
+    if status in {"pending", "processing"}:
+        if any(field in result for field in ("content_type", "integrity", "failure")):
+            raise EventExportError(f"{status} evidence cannot claim content or failure")
+    elif status == "success":
+        if "failure" in result or "content_type" not in result or "integrity" not in result:
+            raise EventExportError("successful evidence requires content metadata and no failure")
+        if result["content_type"] not in {"image/jpeg", "image/png", "video/mp4"}:
+            raise EventExportError("unsupported evidence content type")
+        integrity = result["integrity"]
+        if not isinstance(integrity, Mapping) or set(integrity) != {"algorithm", "digest", "size_bytes"}:
+            raise EventExportError("evidence integrity is malformed")
+        if integrity["algorithm"] != "sha256" or not isinstance(integrity["digest"], str) or not _SHA256.fullmatch(integrity["digest"]):
+            raise EventExportError("evidence integrity must contain a SHA-256 digest")
+        if isinstance(integrity["size_bytes"], bool) or not isinstance(integrity["size_bytes"], int) or integrity["size_bytes"] < 0:
+            raise EventExportError("evidence size must be a non-negative integer")
+    else:
+        if "content_type" in result or "integrity" in result or "failure" not in result:
+            raise EventExportError("failed evidence requires only failure metadata")
+        failure = result["failure"]
+        if not isinstance(failure, Mapping) or set(failure) != {"code", "retryable"}:
+            raise EventExportError("evidence failure is malformed")
+        if failure["code"] not in {"capture_failed", "processing_failed", "expired", "unavailable"}:
+            raise EventExportError("unsupported evidence failure code")
+        if not isinstance(failure["retryable"], bool):
+            raise EventExportError("evidence retryable must be boolean")
+    _reject_unsafe(result)
+    return result
+
+
+def _reference(name: str, value: object) -> str:
+    if not isinstance(value, str) or not _REFERENCE.fullmatch(value):
+        raise EventExportError(f"{name} is not a valid logical reference")
+    return value
+
+
+def _bounded_text(name: str, value: object, maximum: int) -> str:
+    if not isinstance(value, str) or not 1 <= len(value) <= maximum:
+        raise EventExportError(f"{name} must be 1..{maximum} characters")
+    return value
+
+
+def _event_timestamp(nanoseconds: int) -> str:
+    if isinstance(nanoseconds, bool) or not isinstance(nanoseconds, int) or nanoseconds < 0:
+        raise EventExportError("occurred_at_ns must be a non-negative integer")
+    seconds, remainder = divmod(nanoseconds, 1_000_000_000)
+    value = datetime(1970, 1, 1, tzinfo=timezone.utc) + timedelta(
+        seconds=seconds, milliseconds=remainder // 1_000_000
+    )
+    return value.strftime("%Y-%m-%dT%H:%M:%S.") + f"{value.microsecond // 1000:03d}Z"
+
+
+def _date_time(name: str, value: object) -> None:
+    if not isinstance(value, str):
+        raise EventExportError(f"evidence {name} must be a date-time string")
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise EventExportError(f"evidence {name} must be a valid date-time") from exc
+    if parsed.tzinfo is None:
+        raise EventExportError(f"evidence {name} must include a timezone")
+
+
+def _reject_unsafe(value: object, *, key: str = "") -> None:
+    if isinstance(value, Mapping):
+        for child_key, child in value.items():
+            lowered = str(child_key).lower()
+            if lowered in _SENSITIVE_NAMES or lowered.endswith("_path") or lowered.endswith("_url"):
+                raise EventExportError(f"sensitive field {child_key!r} is forbidden")
+            _reject_unsafe(child, key=lowered)
+    elif isinstance(value, (list, tuple)):
+        for child in value:
+            _reject_unsafe(child, key=key)
+    elif isinstance(value, str):
+        if _URL.match(value) or _WINDOWS_PATH.match(value) or value.startswith(("/", "\\\\")):
+            raise EventExportError(f"path or URL value in {key or 'payload'} is forbidden")
+
+
+def _normalize_numbers(value: object) -> object:
+    if value is None or isinstance(value, (str, bool, int)):
+        return value
+    if isinstance(value, float):
+        if not math.isfinite(value):
+            raise EventExportError("canonical JSON rejects non-finite numbers")
+        if value == 0:
+            return 0
+        return value
+    if isinstance(value, Mapping):
+        if any(not isinstance(key, str) for key in value):
+            raise EventExportError("canonical JSON object keys must be strings")
+        return {key: _normalize_numbers(child) for key, child in value.items()}
+    if isinstance(value, (list, tuple)):
+        return [_normalize_numbers(child) for child in value]
+    raise EventExportError(f"canonical JSON does not support {type(value).__name__}")
+
+
+def _encode_jcs(value: object) -> str:
+    if value is None:
+        return "null"
+    if value is True:
+        return "true"
+    if value is False:
+        return "false"
+    if isinstance(value, str):
+        return json.dumps(value, ensure_ascii=False, separators=(",", ":"))
+    if isinstance(value, int):
+        return str(value)
+    if isinstance(value, float):
+        return _jcs_float(value)
+    if isinstance(value, list):
+        return "[" + ",".join(_encode_jcs(item) for item in value) + "]"
+    if isinstance(value, Mapping):
+        # Frozen contract keys are ASCII. Sorting them is therefore identical
+        # to RFC 8785's UTF-16 code-unit ordering without accepting extensions.
+        return "{" + ",".join(
+            _encode_jcs(key) + ":" + _encode_jcs(value[key]) for key in sorted(value)
+        ) + "}"
+    raise EventExportError(f"canonical JSON does not support {type(value).__name__}")
+
+
+def _jcs_float(value: float) -> str:
+    if not math.isfinite(value):
+        raise EventExportError("canonical JSON rejects non-finite numbers")
+    if value == 0:
+        return "0"
+    rendered = repr(value).lower()
+    absolute = abs(value)
+    if 1e-6 <= absolute < 1e21 and "e" in rendered:
+        return format(Decimal(rendered), "f")
+    if "e" in rendered:
+        mantissa, exponent = rendered.split("e", 1)
+        sign = ""
+        if exponent.startswith(("+", "-")):
+            sign, exponent = exponent[0], exponent[1:]
+        exponent = exponent.lstrip("0") or "0"
+        rendered = mantissa + "e" + sign + exponent
+    return rendered[:-2] if rendered.endswith(".0") else rendered

Brain/src/yovision_brain/integration/event_export/replay.py

@@ -0,0 +1,69 @@
+"""Durable atomic replay protection owned by the Brain connector."""
+
+from __future__ import annotations
+
+import sqlite3
+from pathlib import Path
+
+
+class SQLiteReplayCache:
+    """SQLite implementation of the machine-identity ``ReplayCache`` protocol.
+
+    A primary key makes consumption atomic across threads and processes. Entries
+    remain durable across connector restarts until their verifier expiry passes.
+    """
+
+    def __init__(self, database: str | Path, *, timeout_seconds: float = 5.0) -> None:
+        self._database = str(Path(database))
+        self._timeout_seconds = timeout_seconds
+        if timeout_seconds <= 0:
+            raise ValueError("SQLite replay timeout must be positive")
+        with self._connect() as connection:
+            connection.execute("PRAGMA journal_mode=WAL")
+            connection.execute(
+                """
+                CREATE TABLE IF NOT EXISTS machine_token_replay (
+                    principal TEXT NOT NULL,
+                    token_id TEXT NOT NULL,
+                    expires_at INTEGER NOT NULL,
+                    PRIMARY KEY (principal, token_id)
+                ) WITHOUT ROWID
+                """
+            )
+
+    def consume(self, principal: str, token_id: str, expires_at: int, now: int) -> bool:
+        if not principal or not token_id:
+            raise ValueError("replay identity must be non-empty")
+        if any(isinstance(value, bool) or not isinstance(value, int) for value in (expires_at, now)):
+            raise ValueError("replay timestamps must be integers")
+        if expires_at <= now:
+            return False
+
+        connection = self._connect()
+        try:
+            connection.execute("BEGIN IMMEDIATE")
+            connection.execute("DELETE FROM machine_token_replay WHERE expires_at <= ?", (now,))
+            try:
+                connection.execute(
+                    "INSERT INTO machine_token_replay (principal, token_id, expires_at) VALUES (?, ?, ?)",
+                    (principal, token_id, expires_at),
+                )
+            except sqlite3.IntegrityError:
+                connection.rollback()
+                return False
+            connection.commit()
+            return True
+        except BaseException:
+            connection.rollback()
+            raise
+        finally:
+            connection.close()
+
+    def _connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(
+            self._database,
+            timeout=self._timeout_seconds,
+            isolation_level=None,
+        )
+        connection.execute(f"PRAGMA busy_timeout={int(self._timeout_seconds * 1000)}")
+        return connection

Brain/src/yovision_brain/integration/event_export/runtime.py

@@ -0,0 +1,134 @@
+"""Closed runtime configuration and event sink for Brain-to-Sense export."""
+
+from __future__ import annotations
+
+from pathlib import Path
+from typing import Mapping
+
+from yovision_brain.events import InternalEventCandidate
+from yovision_brain.integration.machine_identity import (
+    Signer,
+    TransportPolicy,
+    load_private_key,
+)
+
+from .client import HTTPSMachineIdentitySender
+from .mapper import canonical_json_bytes, export_event
+
+
+class EventExportSink:
+    def __init__(
+        self,
+        sender: HTTPSMachineIdentitySender,
+        *,
+        producer_id: str,
+        site_ref: str,
+        severity: str,
+        region_refs: Mapping[str, str],
+        crossing_directions: Mapping[str, str],
+    ) -> None:
+        self._sender = sender
+        self._producer_id = producer_id
+        self._site_ref = site_ref
+        self._severity = severity
+        self._region_refs = dict(region_refs)
+        self._crossing_directions = dict(crossing_directions)
+
+    def write(self, candidate: InternalEventCandidate) -> None:
+        event = export_event(
+            candidate,
+            producer_id=self._producer_id,
+            site_ref=self._site_ref,
+            severity=self._severity,
+            region_ref=self._region_refs.get(candidate.rule_id),
+            crossing_direction=self._crossing_directions.get(candidate.rule_id),
+        )
+        # Mapping and serialization happen before minting, so the exact bytes are
+        # bound to the machine token and remain unchanged for this delivery.
+        self._sender.send(canonical_json_bytes(event))
+
+
+def build_event_export_sink(raw: object, *, base_dir: Path) -> EventExportSink | None:
+    if raw is None:
+        return None
+    config = _object("event_export", raw)
+    _closed(
+        "event_export",
+        config,
+        {
+            "enabled", "endpoint", "producer_id", "site_ref", "severity",
+            "region_refs", "crossing_directions", "machine_identity", "transport",
+        },
+    )
+    enabled = config.get("enabled", False)
+    if not isinstance(enabled, bool):
+        raise ValueError("event_export.enabled must be a boolean")
+    if not enabled:
+        if set(config) != {"enabled"}:
+            raise ValueError("disabled event_export may only contain enabled")
+        return None
+
+    identity = _object("event_export.machine_identity", config.get("machine_identity"))
+    _closed(
+        "event_export.machine_identity",
+        identity,
+        {"principal", "key_id", "private_key_path"},
+    )
+    transport_raw = _object("event_export.transport", config.get("transport"))
+    _closed(
+        "event_export.transport",
+        transport_raw,
+        {
+            "tls_min_version", "verify_certificate", "verify_hostname",
+            "connect_timeout_ms", "response_header_timeout_ms", "request_timeout_ms",
+            "max_request_bytes",
+        },
+    )
+    policy = TransportPolicy(**transport_raw)  # type: ignore[arg-type]
+    policy.validate()
+    key_path = _string("private_key_path", identity.get("private_key_path"))
+    resolved_key_path = Path(key_path)
+    if not resolved_key_path.is_absolute():
+        resolved_key_path = base_dir / resolved_key_path
+    signer = Signer(
+        _string("principal", identity.get("principal")),
+        _string("key_id", identity.get("key_id")),
+        load_private_key(resolved_key_path),
+    )
+    return EventExportSink(
+        HTTPSMachineIdentitySender(
+            _string("endpoint", config.get("endpoint")), signer, policy
+        ),
+        producer_id=_string("producer_id", config.get("producer_id")),
+        site_ref=_string("site_ref", config.get("site_ref")),
+        severity=_string("severity", config.get("severity")),
+        region_refs=_string_map("region_refs", config.get("region_refs", {})),
+        crossing_directions=_string_map(
+            "crossing_directions", config.get("crossing_directions", {})
+        ),
+    )
+
+
+def _object(name: str, value: object) -> Mapping[str, object]:
+    if not isinstance(value, Mapping) or any(not isinstance(key, str) for key in value):
+        raise ValueError(f"{name} must be an object")
+    return value
+
+
+def _closed(name: str, value: Mapping[str, object], allowed: set[str]) -> None:
+    unknown = set(value) - allowed
+    if unknown:
+        raise ValueError(f"{name} contains unsupported fields")
+
+
+def _string(name: str, value: object) -> str:
+    if not isinstance(value, str) or not value.strip():
+        raise ValueError(f"event_export.{name} must be a non-empty string")
+    return value
+
+
+def _string_map(name: str, value: object) -> Mapping[str, str]:
+    mapping = _object(f"event_export.{name}", value)
+    if any(not isinstance(item, str) or not item for item in mapping.values()):
+        raise ValueError(f"event_export.{name} values must be non-empty strings")
+    return mapping  # type: ignore[return-value]

Brain/src/yovision_brain/integration/sense_control/__init__.py

@@ -1,7 +0,0 @@
-"""Credential-free Sense control-plane connector for Brain."""
-
-from .consumer import ApplyResult, SourceConfigConsumer, SourceConfigError
-from .replay import SQLiteReplayStore
-from .status import RuntimeStatusPublisher
-
-__all__ = ["ApplyResult", "SourceConfigConsumer", "SourceConfigError", "SQLiteReplayStore", "RuntimeStatusPublisher"]

Brain/src/yovision_brain/integration/sense_control/connector.py

@@ -1,57 +0,0 @@
-"""Machine-authenticated adapters with bounded timeout/backoff and a kill switch."""
-
-from __future__ import annotations
-
-import time
-import re
-from dataclasses import dataclass
-from typing import Callable
-
-from yovision_brain.integration.machine_identity.token import Signer, Verifier, bearer_token
-
-from .consumer import ApplyResult, SourceConfigConsumer
-
-_REQUEST_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$")
-
-
-@dataclass(frozen=True, slots=True)
-class ConnectorResponse:
-    status: int
-    body: bytes
-    correlation_id: str
-
-
-class SourceConfigEndpoint:
-    def __init__(self, consumer: SourceConfigConsumer, verifier: Verifier, *, enabled: bool = True, max_body_bytes: int = 10 * 1024 * 1024) -> None:
-        self._consumer, self._verifier, self._enabled, self._max = consumer, verifier, enabled, max_body_bytes
-
-    def receive(self, authorization: str, body: bytes, correlation_id: str) -> ApplyResult:
-        if not self._enabled: raise RuntimeError("CONNECTOR_DISABLED")
-        if not _REQUEST_ID.fullmatch(correlation_id): raise ValueError("INVALID_CORRELATION_ID")
-        if len(body) > self._max: raise ValueError("REQUEST_TOO_LARGE")
-        token=bearer_token(authorization)
-        self._verifier.verify(token,"yovision-brain","source-config:write","POST","/machine/v1/source-config",body)
-        return self._consumer.apply(body)
-
-
-class StatusSender:
-    def __init__(self, signer: Signer, send: Callable[[str, bytes, str, float], ConnectorResponse], *, enabled: bool = True, timeout_seconds: float = 5.0, max_attempts: int = 4, sleeper: Callable[[float], None] = time.sleep) -> None:
-        if timeout_seconds <= 0 or max_attempts < 1: raise ValueError("invalid connector retry policy")
-        self._signer,self._send,self._enabled,self._timeout,self._attempts,self._sleep=signer,send,enabled,timeout_seconds,max_attempts,sleeper
-
-    def publish(self, body: bytes, correlation_id: str) -> ConnectorResponse:
-        if not self._enabled: raise RuntimeError("CONNECTOR_DISABLED")
-        if not _REQUEST_ID.fullmatch(correlation_id): raise ValueError("INVALID_CORRELATION_ID")
-        last: Exception|None=None
-        for attempt in range(self._attempts):
-            try:
-                # A retry gets a fresh jti: the previous request may have been
-                # accepted even when its response was lost.
-                token=self._signer.mint("yovision-sense",("runtime-status:write",),"POST","/machine/v1/runtime-status",body)
-                response=self._send("Bearer "+token,body,correlation_id,self._timeout)
-                if 200<=response.status<300:return response
-                if response.status<500:raise RuntimeError(f"STATUS_REJECTED_{response.status}")
-                last=RuntimeError(f"STATUS_REMOTE_{response.status}")
-            except (TimeoutError,ConnectionError) as exc:last=exc
-            if attempt+1<self._attempts:self._sleep(min(2**attempt,30))
-        raise RuntimeError("STATUS_DELIVERY_EXHAUSTED") from last

Brain/src/yovision_brain/integration/sense_control/consumer.py

@@ -1,250 +0,0 @@
-"""Strict source-config/v1 validation and atomic last-known-good application."""
-
-from __future__ import annotations
-
-import hashlib
-import hmac
-import json
-import re
-import sqlite3
-import threading
-import time
-from contextlib import closing
-from dataclasses import dataclass
-from datetime import datetime, timezone
-from pathlib import Path
-from typing import Callable, Mapping
-from urllib.parse import urlsplit
-
-from yovision_brain.rules.models import AreaDefinition, DirectionalLineDefinition, NormalizedPoint, RuleSet
-
-VERSION = "yovision.source-config/v1"
-_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._~-]{0,127}$")
-_EXTENSION_NAMESPACE = re.compile(r"^[A-Za-z][A-Za-z0-9.-]{0,127}$")
-_SECRET = re.compile(r"password|secret|credential|cookie|jwt|username|stream_uri", re.I)
-
-
-class SourceConfigError(ValueError):
-    def __init__(self, code: str) -> None:
-        super().__init__(code)
-        self.code = code
-
-
-@dataclass(frozen=True, slots=True)
-class AppliedConfig:
-    config_id: str
-    revision: int
-    logical_device_id: str
-    media_ref: str
-    profile_encoding: str
-    frame_rate: float
-    rule_state: str
-    rules: RuleSet | None
-
-
-@dataclass(frozen=True, slots=True)
-class ApplyResult:
-    config_id: str
-    revision: int
-    state: str
-    config: AppliedConfig | None
-
-
-class SourceConfigConsumer:
-    """Persists validated snapshots before atomically changing the active pointer."""
-
-    def __init__(self, state_path: str | Path, *, clock: Callable[[], float] | None = None) -> None:
-        self._path = str(state_path)
-        self._clock = clock or time.time
-        self._lock = threading.RLock()
-        with closing(self._connect()) as connection:
-            connection.executescript(
-                """
-                CREATE TABLE IF NOT EXISTS source_snapshots (
-                    config_id TEXT NOT NULL, revision INTEGER NOT NULL, effective_at INTEGER NOT NULL,
-                    state TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(config_id, revision));
-                CREATE TABLE IF NOT EXISTS source_active (
-                    config_id TEXT PRIMARY KEY, revision INTEGER NOT NULL,
-                    FOREIGN KEY(config_id, revision) REFERENCES source_snapshots(config_id, revision));
-                """
-            )
-            connection.commit()
-
-    def _connect(self) -> sqlite3.Connection:
-        connection = sqlite3.connect(self._path, timeout=5)
-        connection.execute("PRAGMA foreign_keys=ON")
-        connection.execute("PRAGMA journal_mode=WAL")
-        return connection
-
-    def apply(self, body: bytes) -> ApplyResult:
-        document = _parse_and_validate(body)
-        config_id, revision = document["config_id"], document["revision"]
-        mapped = _map(document)
-        effective_at = int(_timestamp(document["effective_at"]))
-        with self._lock, closing(self._connect()) as connection:
-            connection.execute("BEGIN IMMEDIATE")
-            latest = connection.execute(
-                "SELECT revision, payload, state, effective_at FROM source_snapshots WHERE config_id=? ORDER BY revision DESC LIMIT 1",
-                (config_id,),
-            ).fetchone()
-            if latest and revision < latest[0]:
-                connection.rollback()
-                raise SourceConfigError("STALE_REVISION")
-            canonical = body.decode("utf-8")
-            if latest and revision == latest[0]:
-                if json.loads(latest[1]) != document:
-                    connection.rollback()
-                    raise SourceConfigError("REVISION_CONFLICT")
-                connection.rollback()
-                return ApplyResult(config_id, revision, "idempotent", self.get_active(config_id))
-            connection.execute(
-                "INSERT INTO source_snapshots(config_id, revision, effective_at, state, payload) VALUES (?, ?, ?, ?, ?)",
-                (config_id, revision, effective_at, document["rule_set"]["state"], canonical),
-            )
-            if effective_at <= int(self._clock()):
-                connection.execute(
-                    "INSERT INTO source_active(config_id, revision) VALUES (?, ?) ON CONFLICT(config_id) DO UPDATE SET revision=excluded.revision",
-                    (config_id, revision),
-                )
-                state = "applied"
-            else:
-                state = "scheduled"
-            connection.commit()
-        return ApplyResult(config_id, revision, state, mapped if state == "applied" else self.get_active(config_id))
-
-    def activate_due(self) -> tuple[AppliedConfig, ...]:
-        now = int(self._clock())
-        activated: list[AppliedConfig] = []
-        with self._lock, closing(self._connect()) as connection:
-            connection.execute("BEGIN IMMEDIATE")
-            rows = connection.execute(
-                "SELECT s.payload FROM source_snapshots s JOIN (SELECT config_id, MAX(revision) revision FROM source_snapshots WHERE effective_at<=? GROUP BY config_id) d ON d.config_id=s.config_id AND d.revision=s.revision",
-                (now,),
-            ).fetchall()
-            for (payload,) in rows:
-                document = json.loads(payload)
-                connection.execute(
-                    "INSERT INTO source_active(config_id, revision) VALUES (?, ?) ON CONFLICT(config_id) DO UPDATE SET revision=excluded.revision",
-                    (document["config_id"], document["revision"]),
-                )
-                activated.append(_map(document))
-            connection.commit()
-        return tuple(activated)
-
-    def get_active(self, config_id: str) -> AppliedConfig | None:
-        with closing(self._connect()) as connection:
-            row = connection.execute(
-                "SELECT s.payload FROM source_active a JOIN source_snapshots s ON s.config_id=a.config_id AND s.revision=a.revision WHERE a.config_id=?",
-                (config_id,),
-            ).fetchone()
-        return _map(json.loads(row[0])) if row else None
-
-
-def _parse_and_validate(body: bytes) -> dict[str, object]:
-    try:
-        document = json.loads(body.decode("utf-8"))
-    except (UnicodeDecodeError, json.JSONDecodeError):
-        raise SourceConfigError("CONFIG_INVALID") from None
-    if not isinstance(document, dict):
-        raise SourceConfigError("CONFIG_INVALID")
-    if document.get("schema_version") != VERSION:
-        raise SourceConfigError("UNSUPPORTED_SCHEMA_VERSION")
-    if _contains_secret(document):
-        raise SourceConfigError("CONFIG_INVALID")
-    required = {"schema_version", "config_id", "revision", "published_at", "effective_at", "site", "logical_device", "profile", "media", "rule_set", "integrity"}
-    if set(document) - (required | {"extensions"}) or not required <= set(document):
-        raise SourceConfigError("CONFIG_INVALID")
-    extensions = document.get("extensions", {})
-    if not isinstance(extensions, dict) or any(
-        not isinstance(namespace, str)
-        or not _EXTENSION_NAMESPACE.fullmatch(namespace)
-        or not isinstance(value, dict)
-        for namespace, value in extensions.items()
-    ):
-        raise SourceConfigError("CONFIG_INVALID")
-    integrity = document.get("integrity")
-    if not isinstance(integrity, dict) or set(integrity) != {"algorithm", "value"} or integrity.get("algorithm") != "sha256":
-        raise SourceConfigError("CONFIG_INVALID")
-    unsigned = dict(document); unsigned.pop("integrity")
-    digest = hashlib.sha256(json.dumps(unsigned, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode()).hexdigest()
-    if not hmac.compare_digest(digest, str(integrity.get("value", ""))):
-        raise SourceConfigError("CONFIG_INVALID")
-    try:
-        if not _ID.fullmatch(document["config_id"]) or isinstance(document["revision"], bool) or document["revision"] < 1:
-            raise ValueError
-        published, effective = _timestamp(document["published_at"]), _timestamp(document["effective_at"])
-        if effective < published:
-            raise ValueError
-        for field in ("site", "logical_device"):
-            if not isinstance(document[field], dict) or set(document[field]) != {"id"} or not _ID.fullmatch(document[field]["id"]): raise ValueError
-        _validate_profile(document["profile"])
-        media = document["media"]
-        if not isinstance(media, dict) or set(media) != {"ref", "transport"} or media["transport"] != "rtsp" or not isinstance(media["ref"], str) or not media["ref"].startswith("media:") or any(marker in media["ref"] for marker in ("?", "#", "@", "\\", "://")): raise ValueError
-        _validate_rules(document["rule_set"], document["profile"])
-    except (KeyError, TypeError, ValueError, AttributeError):
-        raise SourceConfigError("CONFIG_INVALID") from None
-    return document
-
-
-def _validate_profile(profile: object) -> None:
-    if not isinstance(profile, dict) or set(profile) != {"id", "width", "height", "encoding", "frame_rate"}: raise ValueError
-    if not _ID.fullmatch(profile["id"]) or profile["encoding"] not in {"H264", "H265", "MJPEG"}: raise ValueError
-    for field in ("width", "height"):
-        if isinstance(profile[field], bool) or not isinstance(profile[field], int) or profile[field] < 1: raise ValueError
-    if isinstance(profile["frame_rate"], bool) or not isinstance(profile["frame_rate"], (int, float)) or profile["frame_rate"] <= 0: raise ValueError
-
-
-def _validate_rules(rules: object, profile: Mapping[str, object]) -> None:
-    if not isinstance(rules, dict) or set(rules) != {"version", "state", "profile_binding", "areas", "directional_lines"}: raise ValueError
-    if not _ID.fullmatch(rules["version"]) or rules["state"] not in {"active", "disabled", "recalibration_required"}: raise ValueError
-    binding = rules["profile_binding"]
-    if binding != {"profile_id": profile["id"], "width": profile["width"], "height": profile["height"]}: raise ValueError
-    if not isinstance(rules["areas"], list) or not isinstance(rules["directional_lines"], list) or len(rules["areas"]) > 1024 or len(rules["directional_lines"]) > 1024: raise ValueError
-    identifiers: set[str] = set()
-    for area in rules["areas"]:
-        if not isinstance(area, dict) or set(area) != {"id", "version", "kind", "enabled", "points"} or area["kind"] != "danger_area" or not isinstance(area["enabled"], bool) or not 3 <= len(area["points"]) <= 256: raise ValueError
-        _rule_identity(area, identifiers); points = tuple(_point(value) for value in area["points"])
-        polygon = sum(a[0]*b[1]-b[0]*a[1] for a,b in zip(points, points[1:]+points[:1])) / 2
-        if abs(polygon) < 1e-9: raise ValueError
-    for line in rules["directional_lines"]:
-        if not isinstance(line, dict) or set(line) != {"id", "version", "kind", "enabled", "start", "end", "trigger_direction"} or line["kind"] != "directional_line" or not isinstance(line["enabled"], bool) or line["trigger_direction"] not in {"left_to_right", "right_to_left"}: raise ValueError
-        _rule_identity(line, identifiers)
-        if _point(line["start"]) == _point(line["end"]): raise ValueError
-
-
-def _rule_identity(rule: Mapping[str, object], identifiers: set[str]) -> None:
-    if not isinstance(rule["id"], str) or not _ID.fullmatch(rule["id"]) or rule["id"] in identifiers or isinstance(rule["version"], bool) or not isinstance(rule["version"], int) or rule["version"] < 1: raise ValueError
-    identifiers.add(rule["id"])
-
-
-def _point(value: object) -> tuple[float, float]:
-    if not isinstance(value, dict) or set(value) != {"x", "y"}: raise ValueError
-    x, y = value["x"], value["y"]
-    if isinstance(x, bool) or isinstance(y, bool) or not isinstance(x, (int,float)) or not isinstance(y,(int,float)) or not 0 <= x <= 1 or not 0 <= y <= 1: raise ValueError
-    return float(x), float(y)
-
-
-def _map(document: Mapping[str, object]) -> AppliedConfig:
-    profile, rules = document["profile"], document["rule_set"]
-    rule_set = None
-    if rules["state"] == "active":
-        rule_set = RuleSet(
-            version=rules["version"], profile_id=profile["id"], width=profile["width"], height=profile["height"],
-            areas=tuple(AreaDefinition(a["id"], tuple(NormalizedPoint(**p) for p in a["points"])) for a in rules["areas"] if a["enabled"]),
-            directional_lines=tuple(DirectionalLineDefinition(l["id"], NormalizedPoint(**l["start"]), NormalizedPoint(**l["end"]), l["trigger_direction"]) for l in rules["directional_lines"] if l["enabled"]),
-        )
-    return AppliedConfig(document["config_id"], document["revision"], document["logical_device"]["id"], document["media"]["ref"], profile["encoding"], float(profile["frame_rate"]), rules["state"], rule_set)
-
-
-def _timestamp(value: object) -> float:
-    if not isinstance(value, str) or not value.endswith("Z"): raise ValueError
-    return datetime.fromisoformat(value[:-1] + "+00:00").astimezone(timezone.utc).timestamp()
-
-
-def _contains_secret(value: object) -> bool:
-    if isinstance(value, dict): return any(_SECRET.search(str(k)) or _contains_secret(v) for k,v in value.items())
-    if isinstance(value, list): return any(_contains_secret(item) for item in value)
-    if isinstance(value, str):
-        split=urlsplit(value)
-        return bool(split.username or split.password or value.startswith("file:") or re.match(r"^[A-Za-z]:[\\/]", value))
-    return False

Brain/src/yovision_brain/integration/sense_control/replay.py

@@ -1,44 +0,0 @@
-"""Brain-owned durable replay storage; never shared with Sense or Bell."""
-
-from __future__ import annotations
-
-import sqlite3
-import threading
-from contextlib import closing
-from pathlib import Path
-
-
-class SQLiteReplayStore:
-    def __init__(self, path: str | Path) -> None:
-        self._path = str(path)
-        self._lock = threading.Lock()
-        with closing(self._connect()) as connection:
-            connection.execute("PRAGMA journal_mode=WAL")
-            connection.execute(
-                "CREATE TABLE IF NOT EXISTS machine_replay (principal TEXT NOT NULL, token_id TEXT NOT NULL, expires_at INTEGER NOT NULL, PRIMARY KEY(principal, token_id))"
-            )
-
-    def _connect(self) -> sqlite3.Connection:
-        connection = sqlite3.connect(self._path, timeout=5, isolation_level=None)
-        connection.execute("PRAGMA busy_timeout=5000")
-        return connection
-
-    def consume(self, principal: str, token_id: str, expires_at: int, now: int) -> bool:
-        if not principal or not token_id or expires_at <= now:
-            return False
-        with self._lock, closing(self._connect()) as connection:
-            try:
-                connection.execute("BEGIN IMMEDIATE")
-                connection.execute("DELETE FROM machine_replay WHERE expires_at <= ?", (now,))
-                connection.execute(
-                    "INSERT INTO machine_replay(principal, token_id, expires_at) VALUES (?, ?, ?)",
-                    (principal, token_id, expires_at),
-                )
-                connection.execute("COMMIT")
-                return True
-            except sqlite3.IntegrityError:
-                connection.execute("ROLLBACK")
-                return False
-            except Exception:
-                connection.execute("ROLLBACK")
-                return False

Brain/src/yovision_brain/integration/sense_control/status.py

@@ -1,77 +0,0 @@
-"""Persistent sequence allocation and safe runtime-status/v1 production."""
-
-from __future__ import annotations
-
-import json
-import re
-import sqlite3
-import threading
-import uuid
-from contextlib import closing
-from datetime import datetime, timezone
-from pathlib import Path
-from typing import Mapping, Sequence
-
-_LOGICAL_REF = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
-_VERSION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]{0,63}$")
-_ERROR_CODE = re.compile(r"^[A-Z][A-Z0-9_]{2,63}$")
-_RUNTIME_STATES = {"unconfigured", "starting", "running", "degraded", "failed", "stopped"}
-
-
-class RuntimeStatusPublisher:
-    def __init__(self, state_path: str | Path, brain_instance_ref: str) -> None:
-        self._path, self._instance, self._lock = str(state_path), brain_instance_ref, threading.Lock()
-        with closing(self._connect()) as connection:
-            connection.execute("CREATE TABLE IF NOT EXISTS runtime_sequence(instance_ref TEXT PRIMARY KEY, sequence INTEGER NOT NULL)")
-            connection.commit()
-
-    def _connect(self) -> sqlite3.Connection:
-        return sqlite3.connect(self._path, timeout=5)
-
-    def build(self, *, runtime_state: str, runtime_version: str, started_at: datetime | None, model_ref: str, model_version: str, configurations: Sequence[Mapping[str, object]], health: Mapping[str, object], inputs: Sequence[Mapping[str, object]], observed_at: datetime | None = None) -> bytes:
-        _validate(runtime_state, runtime_version, self._instance, model_ref, model_version, configurations, health, inputs)
-        with self._lock, closing(self._connect()) as connection:
-            connection.execute("BEGIN IMMEDIATE")
-            row=connection.execute("SELECT sequence FROM runtime_sequence WHERE instance_ref=?",(self._instance,)).fetchone();sequence=(row[0]+1) if row else 0
-            connection.execute("INSERT INTO runtime_sequence(instance_ref,sequence) VALUES(?,?) ON CONFLICT(instance_ref) DO UPDATE SET sequence=excluded.sequence",(self._instance,sequence));connection.commit()
-        observed=(observed_at or datetime.now(timezone.utc)).astimezone(timezone.utc)
-        document={"schema_version":"yovision.runtime-status/v1","status_id":str(uuid.uuid4()),"brain_instance_ref":self._instance,"sequence":sequence,"observed_at":_utc(observed),"runtime":{"state":runtime_state,"version":runtime_version,"started_at":_utc(started_at) if started_at else None},"model":{"model_ref":model_ref,"version":model_version},"configurations":list(configurations),"health":dict(health),"inputs":list(inputs)}
-        raw=json.dumps(document,separators=(",",":"),sort_keys=True).encode()
-        lowered=raw.lower();
-        for marker in (b"password",b"credential",b"stream_uri",b"cookie",b"jwt",b"file://"):
-            if marker in lowered: raise ValueError("runtime status contains sensitive field")
-        return raw
-
-
-def _utc(value: datetime) -> str:
-    if value.tzinfo is None: raise ValueError("runtime timestamp must be timezone-aware")
-    return value.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00","Z")
-
-
-def _validate(runtime_state: str, runtime_version: str, instance: str, model_ref: str, model_version: str, configurations: Sequence[Mapping[str, object]], health: Mapping[str, object], inputs: Sequence[Mapping[str, object]]) -> None:
-    if runtime_state not in _RUNTIME_STATES or not _VERSION.fullmatch(runtime_version) or not _LOGICAL_REF.fullmatch(instance) or not _LOGICAL_REF.fullmatch(model_ref) or not _VERSION.fullmatch(model_version):
-        raise ValueError("invalid runtime identity or version")
-    if len(configurations) > 4096 or len(inputs) > 4096:
-        raise ValueError("runtime status collection too large")
-    seen: set[str] = set()
-    for item in configurations:
-        if set(item) != {"config_id", "apply_state", "applied_revision", "error_code"} or not isinstance(item["config_id"], str) or not _LOGICAL_REF.fullmatch(item["config_id"]) or item["config_id"] in seen:
-            raise ValueError("invalid configuration status")
-        seen.add(item["config_id"]); state=item["apply_state"]; revision=item["applied_revision"]; error=item["error_code"]
-        if state not in {"not_configured","applying","applied","rejected"} or (state=="not_configured" and revision is not None) or (state=="applied" and (isinstance(revision,bool) or not isinstance(revision,int) or revision<1)) or (state=="rejected" and (not isinstance(error,str) or not _ERROR_CODE.fullmatch(error))):
-            raise ValueError("invalid configuration status")
-    if set(health) != {"overall","error_codes","metrics"} or health["overall"] not in {"healthy","degraded","unhealthy"} or not _codes(health["error_codes"],32) or not _metrics(health["metrics"]):
-        raise ValueError("invalid health status")
-    for item in inputs:
-        if set(item) != {"input_ref","state","error_codes","metrics"} or not isinstance(item["input_ref"],str) or not _LOGICAL_REF.fullmatch(item["input_ref"]) or item["state"] not in _RUNTIME_STATES or not _codes(item["error_codes"],16) or not _metrics(item["metrics"]):
-            raise ValueError("invalid input status")
-
-
-def _codes(value: object, limit: int) -> bool:
-    return isinstance(value,list) and len(value)<=limit and len(set(value))==len(value) and all(isinstance(code,str) and _ERROR_CODE.fullmatch(code) for code in value)
-
-
-def _metrics(value: object) -> bool:
-    if not isinstance(value,Mapping) or set(value)!={"load_percent","queue_depth","latency_ms"}: return False
-    load,queue,latency=value["load_percent"],value["queue_depth"],value["latency_ms"]
-    return not isinstance(load,bool) and isinstance(load,(int,float)) and 0<=load<=100 and not isinstance(queue,bool) and isinstance(queue,int) and queue>=0 and not isinstance(latency,bool) and isinstance(latency,(int,float)) and latency>=0

Brain/tests/app/test_event_export_connector.py

@@ -0,0 +1,254 @@
+from __future__ import annotations
+
+import io
+import hashlib
+import json
+import re
+import urllib.error
+from pathlib import Path
+
+import pytest
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+from cryptography.hazmat.primitives import serialization
+
+import yovision_brain.app.__main__ as cli
+import yovision_brain.integration.event_export.runtime as export_runtime
+from yovision_brain.integration.event_export import (
+    EventDeliveryError,
+    HTTPSMachineIdentitySender,
+)
+from yovision_brain.integration.machine_identity import (
+    KeyRecord,
+    Registry,
+    ReplayStore,
+    Signer,
+    TransportPolicy,
+    Verifier,
+    bearer_token,
+)
+
+FIXTURE = Path(__file__).parents[1] / "fixtures" / "events" / "area.json"
+
+
+def _policy() -> TransportPolicy:
+    return TransportPolicy(
+        tls_min_version="1.2",
+        verify_certificate=True,
+        verify_hostname=True,
+        connect_timeout_ms=1_000,
+        response_header_timeout_ms=1_000,
+        request_timeout_ms=2_000,
+        max_request_bytes=64 * 1024,
+    )
+
+
+class _Response:
+    status = 202
+
+    def __init__(self, body: bytes, request_id: str) -> None:
+        self.body = body
+        self.headers = {"X-Request-ID": request_id}
+        self.closed = False
+
+    def read(self, amount: int = -1) -> bytes:
+        return self.body[:amount] if amount >= 0 else self.body
+
+    def close(self) -> None:
+        self.closed = True
+
+
+class _VerifyingOpener:
+    def __init__(self, verifier: Verifier) -> None:
+        self.verifier = verifier
+        self.requests = []
+
+    def open(self, request, timeout: float) -> _Response:  # noqa: ANN001
+        self.requests.append((request, timeout))
+        body = request.data
+        self.verifier.verify(
+            bearer_token(request.get_header("Authorization")),
+            "yovision-sense",
+            "events:ingest",
+            request.method,
+            "/v1/events",
+            body,
+        )
+        event = json.loads(body)
+        return _Response(json.dumps({
+            "producer_id": event["producer_id"],
+            "source_event_id": event["source_event_id"],
+            "payload_sha256": hashlib.sha256(body).hexdigest(),
+            "disposition": "accepted",
+        }).encode(), request.get_header("X-request-id"))
+
+
+def test_sender_binds_exact_path_body_and_safe_request_id() -> None:
+    private_key = Ed25519PrivateKey.generate()
+    signer = Signer("yv:brain:school-a", "brain-key-0001", private_key, clock=lambda: 100)
+    registry = Registry(
+        [
+            KeyRecord(
+                principal="yv:brain:school-a",
+                key_id="brain-key-0001",
+                public_key=private_key.public_key(),
+                audience="yovision-sense",
+                scopes=frozenset({"events:ingest"}),
+            )
+        ]
+    )
+    opener = _VerifyingOpener(Verifier(registry, ReplayStore(), clock=lambda: 100))
+    sender = HTTPSMachineIdentitySender(
+        "https://sense.example:8443", signer, _policy(), opener=opener
+    )
+
+    result = sender.send(
+        b'{"producer_id":"brain-school-a","schema_version":"yovision.event/v1",'
+        b'"source_event_id":"evt-1"}'
+    )
+
+    request, timeout = opener.requests[0]
+    assert request.full_url == "https://sense.example:8443/v1/events"
+    assert request.method == "POST"
+    assert request.get_header("Content-type") == "application/json"
+    assert re.fullmatch(r"[A-Za-z0-9][A-Za-z0-9._:-]{15,127}", result.request_id)
+    assert request.get_header("X-request-id") == result.request_id
+    assert (result.disposition, timeout) == ("accepted", 2.0)
+
+
+def test_sender_rejects_plaintext_and_marks_conflict_terminal() -> None:
+    key = Ed25519PrivateKey.generate()
+    signer = Signer("yv:brain:school-a", "brain-key-0001", key)
+    with pytest.raises(ValueError, match="HTTPS origin"):
+        HTTPSMachineIdentitySender("http://sense.example", signer, _policy())
+
+    class ConflictOpener:
+        def open(self, request, timeout: float):  # noqa: ANN001, ARG002
+            raise urllib.error.HTTPError(
+                request.full_url,
+                409,
+                "Conflict",
+                {},
+                io.BytesIO(b'{"code":"event_identity_conflict"}'),
+            )
+
+    sender = HTTPSMachineIdentitySender(
+        "https://sense.example", signer, _policy(), opener=ConflictOpener()
+    )
+    with pytest.raises(EventDeliveryError) as caught:
+        sender.send(b"{}")
+    assert (caught.value.code, caught.value.terminal) == (
+        "event_identity_conflict",
+        True,
+    )
+
+
+def test_disabled_connector_keeps_existing_json_lines_output(tmp_path: Path) -> None:
+    raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
+    raw["event_export"] = {"enabled": False}
+    config = tmp_path / "brain.json"
+    config.write_text(json.dumps(raw), encoding="utf-8")
+    output = tmp_path / "events.jsonl"
+
+    assert cli.main(["--config", str(config), "--output", str(output)]) == 0
+    assert json.loads(output.read_text(encoding="utf-8"))["schema"] == (
+        "brain.internal.event-candidate/v1"
+    )
+
+
+def test_delivery_failure_is_nonzero_and_not_silently_reported_as_success(
+    monkeypatch: pytest.MonkeyPatch,
+    capsys: pytest.CaptureFixture[str],
+    tmp_path: Path,
+) -> None:
+    class FailingSink:
+        def write(self, candidate) -> None:  # noqa: ANN001, ARG002
+            raise EventDeliveryError("event_delivery_unavailable", terminal=False)
+
+    monkeypatch.setattr(cli, "build_event_export_sink", lambda raw, base_dir: FailingSink())
+    unused_output = tmp_path / "disabled-json-lines-target"
+    unused_output.write_text("must remain unchanged", encoding="utf-8")
+    result = cli.main(
+        ["--config", str(FIXTURE), "--output", str(unused_output)]
+    )
+
+    assert result == 3
+    error = json.loads(capsys.readouterr().err.splitlines()[0])
+    assert error == {"status": "error", "message": "event_delivery_unavailable"}
+    assert unused_output.read_text(encoding="utf-8") == "must remain unchanged"
+
+
+def test_inline_private_key_material_is_rejected_without_echo(
+    tmp_path: Path, capsys: pytest.CaptureFixture[str]
+) -> None:
+    raw = json.loads(FIXTURE.read_text(encoding="utf-8"))
+    marker = "INLINE-PRIVATE-MATERIAL-MUST-NOT-LEAK"
+    raw["event_export"] = {
+        "enabled": True,
+        "endpoint": "https://sense.example",
+        "producer_id": "brain-school-a",
+        "site_ref": "site-school-a",
+        "severity": "high",
+        "region_refs": {},
+        "crossing_directions": {},
+        "machine_identity": {
+            "principal": "yv:brain:school-a",
+            "key_id": "brain-key-0001",
+            "private_key_path": "external.pem",
+            "private_key": marker,
+        },
+        "transport": {},
+    }
+    config = tmp_path / "brain.json"
+    config.write_text(json.dumps(raw), encoding="utf-8")
+
+    assert cli.main(["--config", str(config)]) == 3
+    assert marker not in capsys.readouterr().err
+
+
+def test_enabled_config_loads_machine_key_only_from_external_path(
+    tmp_path: Path, monkeypatch: pytest.MonkeyPatch
+) -> None:
+    key_path = tmp_path / "brain-machine.pem"
+    key_path.write_bytes(
+        Ed25519PrivateKey.generate().private_bytes(
+            serialization.Encoding.PEM,
+            serialization.PrivateFormat.PKCS8,
+            serialization.NoEncryption(),
+        )
+    )
+    captured = {}
+
+    class Sender:
+        def __init__(self, endpoint, signer, policy) -> None:  # noqa: ANN001
+            captured.update(endpoint=endpoint, signer=signer, policy=policy)
+
+    monkeypatch.setattr(export_runtime, "HTTPSMachineIdentitySender", Sender)
+    sink = export_runtime.build_event_export_sink(
+        {
+            "enabled": True,
+            "endpoint": "https://sense.example",
+            "producer_id": "brain-school-a",
+            "site_ref": "site-school-a",
+            "severity": "high",
+            "region_refs": {},
+            "crossing_directions": {},
+            "machine_identity": {
+                "principal": "yv:brain:school-a",
+                "key_id": "brain-key-0001",
+                "private_key_path": key_path.name,
+            },
+            "transport": {
+                "tls_min_version": "1.2",
+                "verify_certificate": True,
+                "verify_hostname": True,
+                "connect_timeout_ms": 1_000,
+                "response_header_timeout_ms": 1_000,
+                "request_timeout_ms": 2_000,
+                "max_request_bytes": 64 * 1024,
+            },
+        },
+        base_dir=tmp_path,
+    )
+
+    assert sink is not None
+    assert captured["endpoint"] == "https://sense.example"

Brain/tests/integration/event_export/test_mapper.py

@@ -0,0 +1,148 @@
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import pytest
+
+from yovision_brain.events import INTERNAL_EVENT_SCHEMA, InternalEventCandidate
+from yovision_brain.integration.event_export import EventExportError, canonical_json, export_event, payload_sha256
+
+ROOT = Path(__file__).resolve().parents[4]
+
+
+def _candidate(**changes: object) -> InternalEventCandidate:
+    values: dict[str, object] = {
+        "schema": INTERNAL_EVENT_SCHEMA,
+        "event_id": "evt-area-20260831-0001",
+        "logical_input_id": "camera-east-gate",
+        "event_type": "danger_area_entered",
+        "occurred_at_ns": 1_788_134_401_125_000_000,
+        "rule_id": "rule-east-danger",
+        "rule_version": "3",
+        "model_name": "anonymous-detector",
+        "model_version": "2026.08",
+        "profile_id": "profile-main-stream",
+        "frame_width": 100,
+        "frame_height": 100,
+        "track_id": "track-0042",
+        "observation": {
+            "category": "anonymous_target",
+            "confidence": 0.93,
+            "box": {"left": 12, "top": 20, "right": 31, "bottom": 74},
+            "anchor": {"x": 0.21, "y": 0.74},
+        },
+        "reason": "entered polygon",
+    }
+    values.update(changes)
+    return InternalEventCandidate(**values)  # type: ignore[arg-type]
+
+
+def _fixture(relative: str) -> dict[str, object]:
+    return json.loads((ROOT / relative).read_text(encoding="utf-8"))
+
+
+def test_brain_mapper_matches_frozen_producer_fixture_and_jcs_digest() -> None:
+    pending = _fixture("contracts/evidence/v1/examples/pending.json")
+    expected = _fixture("contracts/events/v1/examples/dangerous-area.json")
+    event = export_event(
+        _candidate(),
+        producer_id="brain-school-a",
+        site_ref="site-school-a",
+        severity="high",
+        evidence=[pending],
+        region_ref="region-east-danger",
+    )
+
+    assert event == expected
+    assert payload_sha256(event) == "4cc1e93820195caf713ea675ff33f178c9d4997dd8a81cb61287e9fea0e3d5e1"
+
+
+def test_retry_mapping_preserves_original_identity_and_payload() -> None:
+    candidate = _candidate()
+    arguments = {
+        "producer_id": "brain-school-a",
+        "site_ref": "site-school-a",
+        "severity": "high",
+        "evidence": [_fixture("contracts/evidence/v1/examples/pending.json")],
+        "region_ref": "region-east-danger",
+    }
+    first = export_event(candidate, **arguments)
+    retry = export_event(candidate, **arguments)
+
+    assert (first["producer_id"], first["source_event_id"]) == (
+        "brain-school-a",
+        candidate.event_id,
+    )
+    assert retry == first
+    assert payload_sha256(retry) == payload_sha256(first)
+
+
+@pytest.mark.parametrize("fixture", ["pending.json", "failed.json"])
+def test_evidence_degradation_states_are_exported_unchanged(fixture: str) -> None:
+    evidence = _fixture(f"contracts/evidence/v1/examples/{fixture}")
+    candidate = _candidate()
+    if fixture == "failed.json":
+        candidate = _candidate(
+            event_id="evt-line-20260831-0002",
+            logical_input_id="camera-north-corridor",
+            event_type="directional_line_crossed",
+            occurred_at_ns=1_788_134_590_000_000_000,
+            rule_id="rule-north-one-way",
+            rule_version="1",
+            track_id="track-0088",
+            observation={"category": "anonymous_target", "confidence": 0.88},
+        )
+    event = export_event(
+        candidate,
+        producer_id="brain-school-a",
+        site_ref="site-school-a",
+        severity="medium" if fixture == "failed.json" else "high",
+        evidence=[evidence],
+        region_ref="line-north-one-way" if fixture == "failed.json" else "region-east-danger",
+        crossing_direction="b_to_a" if fixture == "failed.json" else None,
+    )
+
+    assert event["evidence"] == [evidence]
+
+
+@pytest.mark.parametrize(
+    "unsafe",
+    [
+        {"local_path": "D:/captures/frame.jpg"},
+        {"url": "https://example.invalid/signed"},
+        {"token": "not-a-real-token"},
+    ],
+)
+def test_evidence_rejects_paths_urls_and_sensitive_fields(unsafe: dict[str, object]) -> None:
+    evidence = _fixture("contracts/evidence/v1/examples/pending.json")
+    evidence.update(unsafe)
+    with pytest.raises(EventExportError, match="forbidden"):
+        export_event(
+            _candidate(),
+            producer_id="brain-school-a",
+            site_ref="site-school-a",
+            severity="high",
+            evidence=[evidence],
+            region_ref="region-east-danger",
+        )
+
+
+def test_failed_evidence_cannot_claim_success_content() -> None:
+    evidence = _fixture("contracts/evidence/v1/examples/failed.json")
+    evidence["content_type"] = "video/mp4"
+    with pytest.raises(EventExportError, match="failed evidence"):
+        export_event(
+            _candidate(),
+            producer_id="brain-school-a",
+            site_ref="site-school-a",
+            severity="high",
+            evidence=[evidence],
+            region_ref="region-east-danger",
+        )
+
+
+def test_jcs_normalizes_number_lexemes_and_negative_zero() -> None:
+    assert canonical_json({"small": 1e-7, "fixed": 1e20, "zero": -0.0}) == (
+        '{"fixed":100000000000000000000,"small":1e-7,"zero":0}'
+    )

Brain/tests/integration/event_export/test_replay.py

@@ -0,0 +1,56 @@
+from __future__ import annotations
+
+from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
+import pytest
+
+from yovision_brain.integration.event_export import SQLiteReplayCache
+from yovision_brain.integration.machine_identity import (
+    KeyRecord,
+    MachineIdentityError,
+    Registry,
+    Signer,
+    Verifier,
+)
+
+
+def test_replay_cache_rejects_same_jti_after_connector_restart(tmp_path) -> None:
+    now = 1_800_000_000
+    private_key = Ed25519PrivateKey.generate()
+    registry = Registry(
+        [
+            KeyRecord(
+                principal="yv:brain:node-a",
+                key_id="brain-key-0001",
+                public_key=private_key.public_key(),
+                audience="yovision-sense",
+                scopes=frozenset({"events:ingest"}),
+            )
+        ]
+    )
+    token = Signer(
+        "yv:brain:node-a",
+        "brain-key-0001",
+        private_key,
+        clock=lambda: now,
+    ).mint("yovision-sense", ["events:ingest"], "POST", "/v1/events", b"{}")
+    database = tmp_path / "machine-replay.sqlite3"
+
+    first_process = Verifier(registry, SQLiteReplayCache(database), clock=lambda: now)
+    claims = first_process.verify(
+        token, "yovision-sense", "events:ingest", "POST", "/v1/events", b"{}"
+    )
+    assert claims.iss == "yv:brain:node-a"
+
+    restarted_process = Verifier(registry, SQLiteReplayCache(database), clock=lambda: now)
+    with pytest.raises(MachineIdentityError, match="machine_token_replayed") as caught:
+        restarted_process.verify(
+            token, "yovision-sense", "events:ingest", "POST", "/v1/events", b"{}"
+        )
+    assert caught.value.code == "machine_token_replayed"
+
+
+def test_replay_cache_atomically_reuses_expired_identity(tmp_path) -> None:
+    cache = SQLiteReplayCache(tmp_path / "machine-replay.sqlite3")
+    assert cache.consume("yv:brain:node-a", "jti-one", expires_at=110, now=100)
+    assert not cache.consume("yv:brain:node-a", "jti-one", expires_at=110, now=101)
+    assert cache.consume("yv:brain:node-a", "jti-one", expires_at=130, now=110)

Brain/tests/integration/sense_control/test_connector.py

@@ -1,101 +0,0 @@
-from __future__ import annotations
-
-import hashlib
-import json
-import threading
-from datetime import datetime, timedelta, timezone
-
-import pytest
-from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
-
-from yovision_brain.integration.machine_identity.token import KeyRecord, Registry, Signer, Verifier
-from yovision_brain.integration.sense_control.connector import ConnectorResponse, SourceConfigEndpoint, StatusSender
-from yovision_brain.integration.sense_control.consumer import SourceConfigConsumer, SourceConfigError
-from yovision_brain.integration.sense_control.replay import SQLiteReplayStore
-from yovision_brain.integration.sense_control.status import RuntimeStatusPublisher
-
-
-def source_document(revision: int = 1, *, state: str = "active", effective: int = 0) -> bytes:
-    now=datetime(2026,8,31,tzinfo=timezone.utc)
-    document={"schema_version":"yovision.source-config/v1","config_id":"gate-primary","revision":revision,"published_at":now.isoformat().replace("+00:00","Z"),"effective_at":(now+timedelta(seconds=effective)).isoformat().replace("+00:00","Z"),"site":{"id":"site-east"},"logical_device":{"id":"camera-1"},"profile":{"id":"main","width":1920,"height":1080,"encoding":"H264","frame_rate":25},"media":{"ref":"media:site-east/camera-1/main","transport":"rtsp"},"rule_set":{"version":f"rules-{revision}","state":state,"profile_binding":{"profile_id":"main","width":1920,"height":1080},"areas":[{"id":"danger","version":1,"kind":"danger_area","enabled":True,"points":[{"x":.1,"y":.1},{"x":.8,"y":.1},{"x":.5,"y":.8}]}],"directional_lines":[]}}
-    digest=hashlib.sha256(json.dumps(document,separators=(",",":"),sort_keys=True).encode()).hexdigest();document["integrity"]={"algorithm":"sha256","value":digest}
-    return json.dumps(document,separators=(",",":"),sort_keys=True).encode()
-
-
-def with_extension(body: bytes, extensions: object) -> bytes:
-    document=json.loads(body);document["extensions"]=extensions;unsigned=dict(document);unsigned.pop("integrity");document["integrity"]["value"]=hashlib.sha256(json.dumps(unsigned,separators=(",",":"),sort_keys=True).encode()).hexdigest();return json.dumps(document,separators=(",",":"),sort_keys=True).encode()
-
-
-def identity(tmp_path, now: int):
-    private=Ed25519PrivateKey.generate();signer=Signer("yv:sense:east","sense-key-01",private,clock=lambda:now)
-    registry=Registry([KeyRecord("yv:sense:east","sense-key-01",private.public_key(),"yovision-brain",frozenset({"source-config:write"}))])
-    replay=SQLiteReplayStore(tmp_path/"replay.sqlite")
-    return signer,Verifier(registry,replay,clock=lambda:now)
-
-
-def test_authenticated_apply_is_idempotent_and_replay_survives_restart(tmp_path):
-    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());signer,verifier=identity(tmp_path,now);body=source_document();consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now);endpoint=SourceConfigEndpoint(consumer,verifier)
-    token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body)
-    result=endpoint.receive("Bearer "+token,body,"corr-request-0001");assert result.state=="applied" and result.config.rules is not None
-    with pytest.raises(ValueError,match="machine_token_replayed"):endpoint.receive("Bearer "+token,body,"corr-request-0001")
-    restarted=SourceConfigEndpoint(SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now),Verifier(verifier._registry,SQLiteReplayStore(tmp_path/"replay.sqlite"),clock=lambda:now))
-    with pytest.raises(ValueError,match="machine_token_replayed"):restarted.receive("Bearer "+token,body,"corr-request-0002")
-    new_token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body);assert restarted.receive("Bearer "+new_token,body,"corr-request-0003").state=="idempotent"
-
-
-def test_atomic_replay_accepts_once_under_concurrency(tmp_path):
-    store=SQLiteReplayStore(tmp_path/"atomic.sqlite");results=[]
-    threads=[threading.Thread(target=lambda:results.append(store.consume("yv:sense:east","token-id",200,100))) for _ in range(12)]
-    for thread in threads:thread.start()
-    for thread in threads:thread.join()
-    assert results.count(True)==1
-
-
-def test_stale_unknown_profile_and_recalibration_are_safe(tmp_path):
-    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
-    assert consumer.apply(source_document(2)).config.rules is not None
-    with pytest.raises(SourceConfigError,match="STALE_REVISION"):consumer.apply(source_document(1))
-    invalid=json.loads(source_document(3));invalid["profile"]["width"]=1280;invalid["integrity"]["value"]="0"*64
-    with pytest.raises(SourceConfigError,match="CONFIG_INVALID"):consumer.apply(json.dumps(invalid).encode())
-    unknown=json.loads(source_document(3));unknown["schema_version"]="yovision.source-config/v2"
-    with pytest.raises(SourceConfigError,match="UNSUPPORTED_SCHEMA_VERSION"):consumer.apply(json.dumps(unknown).encode())
-    safe=consumer.apply(source_document(3,state="recalibration_required"));assert safe.config.rule_state=="recalibration_required" and safe.config.rules is None
-
-
-def test_future_effective_snapshot_activates_atomically_after_restart(tmp_path):
-    base=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());current=[base];path=tmp_path/"state.sqlite";consumer=SourceConfigConsumer(path,clock=lambda:current[0])
-    assert consumer.apply(source_document(1)).state=="applied";scheduled=consumer.apply(source_document(2,effective=60));assert scheduled.state=="scheduled" and scheduled.config.revision==1
-    current[0]+=61;restarted=SourceConfigConsumer(path,clock=lambda:current[0]);activated=restarted.activate_due();assert activated[0].revision==2 and restarted.get_active("gate-primary").revision==2
-
-
-def test_unknown_valid_extension_namespace_is_ignored(tmp_path):
-    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
-    result=consumer.apply(with_extension(source_document(),{"vendor.example":{"feature":"safe"}}));assert result.state=="applied" and result.config.revision==1
-
-
-@pytest.mark.parametrize("extensions", [[], {"1invalid":{}}, {"vendor_ok":{}}, {"vendor.example":"not-an-object"}])
-def test_invalid_extensions_are_rejected(tmp_path, extensions):
-    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
-    with pytest.raises(SourceConfigError,match="CONFIG_INVALID"):consumer.apply(with_extension(source_document(),extensions))
-
-
-def test_status_sequence_restart_retry_timeout_and_disable(tmp_path):
-    path=tmp_path/"status.sqlite";publisher=RuntimeStatusPublisher(path,"brain-east-01");health={"overall":"healthy","error_codes":[],"metrics":{"load_percent":1.0,"queue_depth":0,"latency_ms":2.0}}
-    one=json.loads(publisher.build(runtime_state="running",runtime_version="1.0.0",started_at=datetime.now(timezone.utc),model_ref="people-detection",model_version="1",configurations=[],health=health,inputs=[]));two=json.loads(RuntimeStatusPublisher(path,"brain-east-01").build(runtime_state="running",runtime_version="1.0.0",started_at=None,model_ref="people-detection",model_version="1",configurations=[],health=health,inputs=[]));assert (one["sequence"],two["sequence"])==(0,1)
-    private=Ed25519PrivateKey.generate();signer=Signer("yv:brain:east","brain-key-01",private,clock=lambda:1_787_000_000);attempts=[]
-    def send(_auth,_body,_corr,timeout):attempts.append(timeout);raise TimeoutError
-    sender=StatusSender(signer,send,max_attempts=3,sleeper=lambda _:None)
-    with pytest.raises(RuntimeError,match="STATUS_DELIVERY_EXHAUSTED"):sender.publish(b"{}","corr-request-0001")
-    assert attempts==[5.0,5.0,5.0]
-    disabled=StatusSender(signer,lambda *_:ConnectorResponse(204,b"","corr-request-0001"),enabled=False)
-    with pytest.raises(RuntimeError,match="CONNECTOR_DISABLED"):disabled.publish(b"{}","corr-request-0001")
-
-
-@pytest.mark.parametrize("request_id", ["short", "0123456789abcde\n", "0123456789abcde!", "a"*129])
-def test_connector_rejects_unsafe_request_ids(tmp_path, request_id):
-    now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());signer,verifier=identity(tmp_path,now);body=source_document();endpoint=SourceConfigEndpoint(SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now),verifier)
-    token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body)
-    with pytest.raises(ValueError,match="INVALID_CORRELATION_ID"):endpoint.receive("Bearer "+token,body,request_id)
-    status_signer=Signer("yv:brain:east","brain-key-01",Ed25519PrivateKey.generate(),clock=lambda:now)
-    sender=StatusSender(status_signer,lambda *_:ConnectorResponse(204,b"",request_id))
-    with pytest.raises(ValueError,match="INVALID_CORRELATION_ID"):sender.publish(b"{}",request_id)

Sense/server/app/sense/integration/bell_connector/client.go

@@ -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
+}

Sense/server/app/sense/integration/bell_connector/handlers.go

@@ -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})
+}

Sense/server/app/sense/integration/bell_connector/ingress.go

@@ -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
+	}
+}

Sense/server/app/sense/integration/bell_connector/model.go

@@ -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"`
+}

Sense/server/app/sense/integration/bell_connector/outbox.go

@@ -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
+}

Sense/server/app/sense/integration/bell_connector/replay.go

@@ -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
+}

Sense/server/app/sense/integration/bell_connector/runtime.go

@@ -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") }

Sense/server/app/sense/integration/brain_control/connector.go

@@ -1,118 +0,0 @@
-package brain_control
-
-import (
-	"errors"
-	"fmt"
-	"regexp"
-	"time"
-
-	mi "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
-)
-
-const (
-	sourceConfigPath  = "/machine/v1/source-config"
-	runtimeStatusPath = "/machine/v1/runtime-status"
-)
-
-var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$`)
-
-type SendResponse struct {
-	StatusCode    int
-	Body          []byte
-	CorrelationID string
-}
-type SendFunc func(authorization string, body []byte, correlationID string, timeout time.Duration) (SendResponse, error)
-
-type ConfigSender struct {
-	Signer      mi.Signer
-	Send        SendFunc
-	Enabled     bool
-	Timeout     time.Duration
-	MaxAttempts int
-	Sleep       func(time.Duration)
-}
-
-func (s ConfigSender) Publish(config SourceConfig, correlationID string) (SendResponse, error) {
-	if !s.Enabled {
-		return SendResponse{}, errors.New("CONNECTOR_DISABLED")
-	}
-	if s.Send == nil || !requestIDPattern.MatchString(correlationID) {
-		return SendResponse{}, errors.New("invalid connector configuration")
-	}
-	if s.Timeout <= 0 {
-		s.Timeout = 5 * time.Second
-	}
-	if s.MaxAttempts == 0 {
-		s.MaxAttempts = 4
-	}
-	if s.MaxAttempts < 1 {
-		return SendResponse{}, errors.New("invalid connector retry policy")
-	}
-	if s.Sleep == nil {
-		s.Sleep = time.Sleep
-	}
-	if err := ValidateSourceConfig(config); err != nil {
-		return SendResponse{}, err
-	}
-	body, err := MarshalSourceConfig(config)
-	if err != nil {
-		return SendResponse{}, err
-	}
-	var last error
-	for attempt := 0; attempt < s.MaxAttempts; attempt++ {
-		token, mintErr := s.Signer.Mint("yovision-brain", []string{"source-config:write"}, "POST", sourceConfigPath, body)
-		if mintErr != nil {
-			return SendResponse{}, mintErr
-		}
-		response, sendErr := s.Send("Bearer "+token, body, correlationID, s.Timeout)
-		if sendErr == nil && response.StatusCode >= 200 && response.StatusCode < 300 {
-			return response, nil
-		}
-		if sendErr == nil && response.StatusCode < 500 {
-			return SendResponse{}, fmt.Errorf("source config rejected: %d", response.StatusCode)
-		}
-		if sendErr != nil {
-			last = sendErr
-		} else {
-			last = fmt.Errorf("source config remote status: %d", response.StatusCode)
-		}
-		if attempt+1 < s.MaxAttempts {
-			delay := time.Second << attempt
-			if delay > 30*time.Second {
-				delay = 30 * time.Second
-			}
-			s.Sleep(delay)
-		}
-	}
-	return SendResponse{}, fmt.Errorf("source config delivery exhausted: %w", last)
-}
-
-type RuntimeStatusEndpoint struct {
-	Verifier     mi.Verifier
-	Store        ProjectionStore
-	Enabled      bool
-	MaxBodyBytes int
-}
-
-func (e RuntimeStatusEndpoint) Receive(authorization string, body []byte, correlationID string, expected map[string]int64) (ProjectionView, error) {
-	if !e.Enabled {
-		return ProjectionView{}, errors.New("CONNECTOR_DISABLED")
-	}
-	if !requestIDPattern.MatchString(correlationID) {
-		return ProjectionView{}, errors.New("INVALID_CORRELATION_ID")
-	}
-	if e.MaxBodyBytes == 0 {
-		e.MaxBodyBytes = 10 * 1024 * 1024
-	}
-	if len(body) > e.MaxBodyBytes {
-		return ProjectionView{}, errors.New("REQUEST_TOO_LARGE")
-	}
-	token, err := mi.BearerToken(authorization)
-	if err != nil {
-		return ProjectionView{}, err
-	}
-	if _, err = e.Verifier.Verify(token, "yovision-sense", "runtime-status:write", "POST", runtimeStatusPath, body); err != nil {
-		return ProjectionView{}, err
-	}
-	return e.Store.Ingest(body, expected)
-}

Sense/server/app/sense/integration/brain_control/connector_test.go

@@ -1,124 +0,0 @@
-package brain_control
-
-import (
-	"crypto/ed25519"
-	"crypto/rand"
-	"encoding/json"
-	"strings"
-	"sync"
-	"testing"
-	"time"
-
-	mi "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
-	"gorm.io/driver/sqlite"
-	"gorm.io/gorm"
-)
-
-func memoryDB(t *testing.T, name string) *gorm.DB {
-	t.Helper()
-	db, err := gorm.Open(sqlite.Open("file:"+name+"?mode=memory&cache=shared"), &gorm.Config{})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err = db.AutoMigrate(&ReplayToken{}, &RuntimeProjection{}, &SourceRevision{}); err != nil {
-		t.Fatal(err)
-	}
-	return db
-}
-
-func TestSourceMapperAndRecalibration(t *testing.T) {
-	now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
-	facts := SourceFacts{ConfigID: "gate-primary", SiteID: "site-east", LogicalDeviceID: "camera-1", MediaPath: "site/camera/main", Revision: 1, PublishedAt: now, EffectiveAt: now, Profile: Profile{ID: "main", Width: 1920, Height: 1080, Encoding: "h264", FrameRate: 25}, RuleSetVersion: "rules-1", Areas: []AreaRule{{ID: "danger", Version: 1, Kind: "danger_area", Enabled: true, Points: []Point{{.1, .1}, {.8, .1}, {.5, .8}}}}}
-	config, err := MapSourceConfig(facts)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err = ValidateSourceConfig(config); err != nil {
-		t.Fatal(err)
-	}
-	facts.Revision = 2
-	facts.NeedsRecalibration = true
-	config, err = MapSourceConfig(facts)
-	if err != nil {
-		t.Fatal(err)
-	}
-	if config.RuleSet.State != "recalibration_required" || config.RuleSet.Areas[0].Enabled {
-		t.Fatal("recalibration did not disable rules")
-	}
-}
-
-func TestReplayAtomicAndRestart(t *testing.T) {
-	db := memoryDB(t, "replay-package")
-	pub, priv, _ := ed25519.GenerateKey(rand.Reader)
-	now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
-	registry, _ := mi.NewRegistry(mi.KeyRecord{Principal: "yv:brain:east", KeyID: "brain-key-01", PublicKey: pub, Audience: "yovision-sense", Scopes: []string{"runtime-status:write"}, Enabled: true})
-	signer := mi.Signer{Principal: "yv:brain:east", KeyID: "brain-key-01", PrivateKey: priv, Now: func() time.Time { return now }}
-	body := []byte("{}")
-	token, _ := signer.Mint("yovision-sense", []string{"runtime-status:write"}, "POST", "/machine/v1/runtime-status", body)
-	accepted := 0
-	var mu sync.Mutex
-	var wg sync.WaitGroup
-	for range 8 {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			v := mi.Verifier{Registry: registry, Replay: GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
-			if _, err := v.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body); err == nil {
-				mu.Lock()
-				accepted++
-				mu.Unlock()
-			}
-		}()
-	}
-	wg.Wait()
-	if accepted != 1 {
-		t.Fatalf("accepted %d", accepted)
-	}
-	v := mi.Verifier{Registry: registry, Replay: GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
-	if _, err := v.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body); err == nil {
-		t.Fatal("restart replay accepted")
-	}
-}
-
-func TestProjectionStaleRecoveryAndMismatch(t *testing.T) {
-	db := memoryDB(t, "projection-package")
-	now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
-	store := ProjectionStore{DB: db, Clock: func() time.Time { return now }, StaleAfter: 90 * time.Second}
-	raw := runtimeFixture("018f4d6a-8d1b-4a25-8b37-9085f9c0d101", 1, now, 2)
-	view, err := store.Ingest(raw, map[string]int64{"gate": 3})
-	if err != nil || !view.RevisionMismatch {
-		t.Fatalf("view %+v err %v", view, err)
-	}
-	now = now.Add(91 * time.Second)
-	view, _ = store.View("brain-east-01")
-	if !view.Offline {
-		t.Fatal("not offline")
-	}
-	raw = runtimeFixture("018f4d6a-8d1b-4a25-8b37-9085f9c0d102", 2, now, 3)
-	view, err = store.Ingest(raw, map[string]int64{"gate": 3})
-	if err != nil || !view.Recovered {
-		t.Fatalf("recovery %+v err %v", view, err)
-	}
-}
-
-func TestConnectorRejectsUnsafeRequestIDs(t *testing.T) {
-	for _, value := range []string{"short", "0123456789abcde\n", "0123456789abcde!", strings.Repeat("a", 129)} {
-		sender := ConfigSender{Enabled: true, Send: func(string, []byte, string, time.Duration) (SendResponse, error) {
-			t.Fatal("unsafe request id reached transport")
-			return SendResponse{}, nil
-		}}
-		if _, err := sender.Publish(SourceConfig{}, value); err == nil {
-			t.Fatalf("sender accepted request id %q", value)
-		}
-		endpoint := RuntimeStatusEndpoint{Enabled: true}
-		if _, err := endpoint.Receive("Bearer ignored", nil, value, nil); err == nil || err.Error() != "INVALID_CORRELATION_ID" {
-			t.Fatalf("endpoint accepted request id %q: %v", value, err)
-		}
-	}
-}
-
-func runtimeFixture(id string, seq int64, observed time.Time, revision int64) []byte {
-	value := map[string]any{"schema_version": RuntimeStatusVersion, "status_id": id, "brain_instance_ref": "brain-east-01", "sequence": seq, "observed_at": observed.Format(time.RFC3339), "runtime": map[string]any{"state": "running", "version": "1.0.0", "started_at": observed.Format(time.RFC3339)}, "model": map[string]any{"model_ref": "people", "version": "1"}, "configurations": []any{map[string]any{"config_id": "gate", "apply_state": "applied", "applied_revision": revision, "error_code": nil}}, "health": map[string]any{"overall": "healthy", "error_codes": []any{}, "metrics": map[string]any{"load_percent": 1.0, "queue_depth": 0, "latency_ms": 1.0}}, "inputs": []any{}}
-	raw, _ := json.Marshal(value)
-	return raw
-}

Sense/server/app/sense/integration/brain_control/mapper.go

@@ -1,135 +0,0 @@
-package brain_control
-
-import (
-	"crypto/sha256"
-	"encoding/hex"
-	"encoding/json"
-	"errors"
-	"fmt"
-	"regexp"
-	"strings"
-	"time"
-)
-
-var stableID = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._~-]{0,127}$`)
-
-func MapSourceConfig(f SourceFacts) (SourceConfig, error) {
-	if !stableID.MatchString(f.ConfigID) || !stableID.MatchString(f.SiteID) || !stableID.MatchString(f.LogicalDeviceID) || !stableID.MatchString(f.Profile.ID) || f.Revision < 1 || f.Profile.Width < 1 || f.Profile.Height < 1 || f.Profile.FrameRate <= 0 {
-		return SourceConfig{}, errors.New("invalid source configuration facts")
-	}
-	encoding := strings.ToUpper(f.Profile.Encoding)
-	if encoding != "H264" && encoding != "H265" && encoding != "MJPEG" {
-		return SourceConfig{}, errors.New("unsupported profile encoding")
-	}
-	if f.PublishedAt.IsZero() || f.EffectiveAt.Before(f.PublishedAt) {
-		return SourceConfig{}, errors.New("invalid source configuration time")
-	}
-	if !stableID.MatchString(f.RuleSetVersion) {
-		return SourceConfig{}, errors.New("invalid rule set version")
-	}
-	if strings.ContainsAny(f.MediaPath, "?#@\\") || strings.Contains(f.MediaPath, "://") || f.MediaPath == "" {
-		return SourceConfig{}, errors.New("media path must be opaque and credential-free")
-	}
-	if len(f.Areas) > 1024 || len(f.DirectionalLines) > 1024 {
-		return SourceConfig{}, errors.New("too many rules")
-	}
-	seen := map[string]bool{}
-	for _, a := range f.Areas {
-		if !stableID.MatchString(a.ID) || seen[a.ID] || a.Version < 1 || a.Kind != "danger_area" || len(a.Points) < 3 || len(a.Points) > 256 || !validPoints(a.Points) || polygonArea(a.Points) == 0 {
-			return SourceConfig{}, errors.New("invalid area rule")
-		}
-		seen[a.ID] = true
-	}
-	for _, l := range f.DirectionalLines {
-		if !stableID.MatchString(l.ID) || seen[l.ID] || l.Version < 1 || l.Kind != "directional_line" || (l.TriggerDirection != "left_to_right" && l.TriggerDirection != "right_to_left") || !validPoints([]Point{l.Start, l.End}) || l.Start == l.End {
-			return SourceConfig{}, errors.New("invalid directional line rule")
-		}
-		seen[l.ID] = true
-	}
-	var out SourceConfig
-	out.SchemaVersion, out.ConfigID, out.Revision = SourceConfigVersion, f.ConfigID, f.Revision
-	out.PublishedAt, out.EffectiveAt = f.PublishedAt.UTC(), f.EffectiveAt.UTC()
-	out.Site.ID, out.LogicalDevice.ID = f.SiteID, f.LogicalDeviceID
-	out.Profile = f.Profile
-	out.Profile.Encoding = encoding
-	out.Media.Ref, out.Media.Transport = "media:"+strings.TrimPrefix(f.MediaPath, "/"), "rtsp"
-	out.RuleSet.Version = f.RuleSetVersion
-	out.RuleSet.State = "active"
-	if f.Disabled {
-		out.RuleSet.State = "disabled"
-	}
-	if f.NeedsRecalibration {
-		out.RuleSet.State = "recalibration_required"
-	}
-	out.RuleSet.ProfileBinding.ProfileID, out.RuleSet.ProfileBinding.Width, out.RuleSet.ProfileBinding.Height = f.Profile.ID, f.Profile.Width, f.Profile.Height
-	out.RuleSet.Areas = append([]AreaRule(nil), f.Areas...)
-	out.RuleSet.DirectionalLines = append([]DirectionalLineRule(nil), f.DirectionalLines...)
-	if out.RuleSet.State != "active" {
-		for i := range out.RuleSet.Areas {
-			out.RuleSet.Areas[i].Enabled = false
-		}
-		for i := range out.RuleSet.DirectionalLines {
-			out.RuleSet.DirectionalLines[i].Enabled = false
-		}
-	}
-	digest, err := sourceDigest(out)
-	if err != nil {
-		return SourceConfig{}, err
-	}
-	out.Integrity.Algorithm, out.Integrity.Value = "sha256", digest
-	return out, nil
-}
-
-func MarshalSourceConfig(config SourceConfig) ([]byte, error) { return json.Marshal(config) }
-
-func sourceDigest(config SourceConfig) (string, error) {
-	raw, err := json.Marshal(config)
-	if err != nil {
-		return "", err
-	}
-	var value map[string]any
-	if err = json.Unmarshal(raw, &value); err != nil {
-		return "", err
-	}
-	delete(value, "integrity")
-	canonical, err := json.Marshal(value)
-	if err != nil {
-		return "", err
-	}
-	sum := sha256.Sum256(canonical)
-	return hex.EncodeToString(sum[:]), nil
-}
-
-func validPoints(points []Point) bool {
-	for _, p := range points {
-		if p.X < 0 || p.X > 1 || p.Y < 0 || p.Y > 1 {
-			return false
-		}
-	}
-	return true
-}
-func polygonArea(p []Point) float64 {
-	var a float64
-	for i := range p {
-		n := p[(i+1)%len(p)]
-		a += p[i].X*n.Y - n.X*p[i].Y
-	}
-	if a < 0 {
-		a = -a
-	}
-	return a / 2
-}
-
-func ValidateSourceConfig(config SourceConfig) error {
-	if config.SchemaVersion != SourceConfigVersion {
-		return fmt.Errorf("unsupported source config version")
-	}
-	digest, err := sourceDigest(config)
-	if err != nil || config.Integrity.Algorithm != "sha256" || digest != config.Integrity.Value {
-		return errors.New("source config integrity mismatch")
-	}
-	_, err = MapSourceConfig(SourceFacts{ConfigID: config.ConfigID, SiteID: config.Site.ID, LogicalDeviceID: config.LogicalDevice.ID, MediaPath: strings.TrimPrefix(config.Media.Ref, "media:"), Revision: config.Revision, PublishedAt: config.PublishedAt, EffectiveAt: config.EffectiveAt, Profile: config.Profile, RuleSetVersion: config.RuleSet.Version, Disabled: config.RuleSet.State == "disabled", NeedsRecalibration: config.RuleSet.State == "recalibration_required", Areas: config.RuleSet.Areas, DirectionalLines: config.RuleSet.DirectionalLines})
-	return err
-}
-
-func UTCNow() time.Time { return time.Now().UTC() }

Sense/server/app/sense/integration/brain_control/models.go

@@ -1,122 +0,0 @@
-package brain_control
-
-import "time"
-
-const (
-	SourceConfigVersion  = "yovision.source-config/v1"
-	RuntimeStatusVersion = "yovision.runtime-status/v1"
-)
-
-type Point struct {
-	X float64 `json:"x"`
-	Y float64 `json:"y"`
-}
-type Profile struct {
-	ID        string  `json:"id"`
-	Width     int     `json:"width"`
-	Height    int     `json:"height"`
-	Encoding  string  `json:"encoding"`
-	FrameRate float64 `json:"frame_rate"`
-}
-type AreaRule struct {
-	ID      string  `json:"id"`
-	Version int64   `json:"version"`
-	Kind    string  `json:"kind"`
-	Enabled bool    `json:"enabled"`
-	Points  []Point `json:"points"`
-}
-type DirectionalLineRule struct {
-	ID               string `json:"id"`
-	Version          int64  `json:"version"`
-	Kind             string `json:"kind"`
-	Enabled          bool   `json:"enabled"`
-	Start            Point  `json:"start"`
-	End              Point  `json:"end"`
-	TriggerDirection string `json:"trigger_direction"`
-}
-
-type SourceConfig struct {
-	SchemaVersion string    `json:"schema_version"`
-	ConfigID      string    `json:"config_id"`
-	Revision      int64     `json:"revision"`
-	PublishedAt   time.Time `json:"published_at"`
-	EffectiveAt   time.Time `json:"effective_at"`
-	Site          struct {
-		ID string `json:"id"`
-	} `json:"site"`
-	LogicalDevice struct {
-		ID string `json:"id"`
-	} `json:"logical_device"`
-	Profile Profile `json:"profile"`
-	Media   struct {
-		Ref       string `json:"ref"`
-		Transport string `json:"transport"`
-	} `json:"media"`
-	RuleSet struct {
-		Version        string `json:"version"`
-		State          string `json:"state"`
-		ProfileBinding struct {
-			ProfileID string `json:"profile_id"`
-			Width     int    `json:"width"`
-			Height    int    `json:"height"`
-		} `json:"profile_binding"`
-		Areas            []AreaRule            `json:"areas"`
-		DirectionalLines []DirectionalLineRule `json:"directional_lines"`
-	} `json:"rule_set"`
-	Integrity struct {
-		Algorithm string `json:"algorithm"`
-		Value     string `json:"value"`
-	} `json:"integrity"`
-}
-
-// SourceFacts is an explicit, credential-free boundary DTO. Callers map their
-// GORM entities into it; database models are never serialized as a contract.
-type SourceFacts struct {
-	ConfigID, SiteID, LogicalDeviceID, MediaRouteID, MediaPath string
-	Revision                                                   int64
-	PublishedAt, EffectiveAt                                   time.Time
-	Profile                                                    Profile
-	RuleSetVersion                                             string
-	Disabled, NeedsRecalibration                               bool
-	Areas                                                      []AreaRule
-	DirectionalLines                                           []DirectionalLineRule
-}
-
-type ReplayToken struct {
-	Principal string    `gorm:"size:128;primaryKey"`
-	TokenID   string    `gorm:"size:96;primaryKey"`
-	ExpiresAt time.Time `gorm:"not null;index"`
-	CreatedAt time.Time `gorm:"not null"`
-}
-
-func (ReplayToken) TableName() string { return "sense_brain_runtime_replay_tokens" }
-
-type RuntimeProjection struct {
-	BrainInstanceRef      string    `gorm:"size:128;primaryKey"`
-	StatusID              string    `gorm:"size:36;not null;uniqueIndex"`
-	Sequence              int64     `gorm:"not null"`
-	ObservedAt            time.Time `gorm:"not null;index"`
-	ReceivedAt            time.Time `gorm:"not null"`
-	RuntimeState          string    `gorm:"size:32;not null"`
-	RuntimeVersion        string    `gorm:"size:64;not null"`
-	ModelRef              string    `gorm:"size:128;not null"`
-	ModelVersion          string    `gorm:"size:64;not null"`
-	HealthOverall         string    `gorm:"size:32;not null"`
-	ExpectedRevisionsJSON string    `gorm:"type:jsonb;not null"`
-	ConfigurationsJSON    string    `gorm:"type:jsonb;not null"`
-	HealthJSON            string    `gorm:"type:jsonb;not null"`
-	InputsJSON            string    `gorm:"type:jsonb;not null"`
-	WasOffline            bool      `gorm:"not null;default:false"`
-	CreatedAt             time.Time
-	UpdatedAt             time.Time
-}
-
-func (RuntimeProjection) TableName() string { return "sense_brain_runtime_projections" }
-
-type SourceRevision struct {
-	ConfigID  string `gorm:"size:128;primaryKey"`
-	Revision  int64  `gorm:"not null"`
-	UpdatedAt time.Time
-}
-
-func (SourceRevision) TableName() string { return "sense_brain_source_revisions" }

Sense/server/app/sense/integration/brain_control/runtime.go

@@ -1,145 +0,0 @@
-package brain_control
-
-import (
-	"bytes"
-	"encoding/json"
-	"errors"
-	"io"
-	"regexp"
-	"time"
-)
-
-type runtimeStatus struct {
-	SchemaVersion    string    `json:"schema_version"`
-	StatusID         string    `json:"status_id"`
-	BrainInstanceRef string    `json:"brain_instance_ref"`
-	Sequence         int64     `json:"sequence"`
-	ObservedAt       time.Time `json:"observed_at"`
-	Runtime          struct {
-		State     string     `json:"state"`
-		Version   string     `json:"version"`
-		StartedAt *time.Time `json:"started_at"`
-	} `json:"runtime"`
-	Model struct {
-		ModelRef string `json:"model_ref"`
-		Version  string `json:"version"`
-	} `json:"model"`
-	Configurations []configurationStatus `json:"configurations"`
-	Health         healthStatus          `json:"health"`
-	Inputs         []inputStatus         `json:"inputs"`
-}
-type configurationStatus struct {
-	ConfigID        string  `json:"config_id"`
-	ApplyState      string  `json:"apply_state"`
-	AppliedRevision *int64  `json:"applied_revision"`
-	ErrorCode       *string `json:"error_code"`
-}
-type metrics struct {
-	LoadPercent float64 `json:"load_percent"`
-	QueueDepth  int64   `json:"queue_depth"`
-	LatencyMS   float64 `json:"latency_ms"`
-}
-type healthStatus struct {
-	Overall    string   `json:"overall"`
-	ErrorCodes []string `json:"error_codes"`
-	Metrics    metrics  `json:"metrics"`
-}
-type inputStatus struct {
-	InputRef   string   `json:"input_ref"`
-	State      string   `json:"state"`
-	ErrorCodes []string `json:"error_codes"`
-	Metrics    metrics  `json:"metrics"`
-}
-
-var uuid4 = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$`)
-var errCode = regexp.MustCompile(`^[A-Z][A-Z0-9_]{2,63}$`)
-
-func parseRuntimeStatus(raw []byte) (runtimeStatus, error) {
-	var s runtimeStatus
-	d := json.NewDecoder(bytes.NewReader(raw))
-	d.DisallowUnknownFields()
-	if err := d.Decode(&s); err != nil {
-		return s, errors.New("CONFIG_INVALID")
-	}
-	if err := d.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
-		return s, errors.New("CONFIG_INVALID")
-	}
-	if s.SchemaVersion != RuntimeStatusVersion {
-		return s, errors.New("UNSUPPORTED_SCHEMA_VERSION")
-	}
-	if !uuid4.MatchString(s.StatusID) || !stableID.MatchString(s.BrainInstanceRef) || s.Sequence < 0 || s.ObservedAt.IsZero() || !validState(s.Runtime.State) || s.Runtime.Version == "" || !stableID.MatchString(s.Model.ModelRef) || s.Model.Version == "" || len(s.Configurations) > 4096 || len(s.Inputs) > 4096 || !validHealth(s.Health) {
-		return s, errors.New("CONFIG_INVALID")
-	}
-	seen := map[string]bool{}
-	for _, c := range s.Configurations {
-		if !stableID.MatchString(c.ConfigID) || seen[c.ConfigID] || !validApply(c) {
-			return s, errors.New("CONFIG_INVALID")
-		}
-		seen[c.ConfigID] = true
-	}
-	for _, i := range s.Inputs {
-		if !stableID.MatchString(i.InputRef) || !validState(i.State) || !validMetrics(i.Metrics) || !validCodes(i.ErrorCodes, 16) {
-			return s, errors.New("CONFIG_INVALID")
-		}
-	}
-	return s, nil
-}
-func validState(v string) bool {
-	switch v {
-	case "unconfigured", "starting", "running", "degraded", "failed", "stopped":
-		return true
-	}
-	return false
-}
-func validApply(c configurationStatus) bool {
-	switch c.ApplyState {
-	case "not_configured":
-		return c.AppliedRevision == nil
-	case "applying":
-		return true
-	case "applied":
-		return c.AppliedRevision != nil && *c.AppliedRevision >= 1
-	case "rejected":
-		return c.ErrorCode != nil && errCode.MatchString(*c.ErrorCode)
-	}
-	return false
-}
-func validMetrics(m metrics) bool {
-	return m.LoadPercent >= 0 && m.LoadPercent <= 100 && m.QueueDepth >= 0 && m.LatencyMS >= 0
-}
-func validCodes(v []string, n int) bool {
-	if len(v) > n {
-		return false
-	}
-	seen := map[string]bool{}
-	for _, x := range v {
-		if seen[x] || !errCode.MatchString(x) {
-			return false
-		}
-		seen[x] = true
-	}
-	return true
-}
-func validHealth(h healthStatus) bool {
-	return (h.Overall == "healthy" || h.Overall == "degraded" || h.Overall == "unhealthy") && validCodes(h.ErrorCodes, 32) && validMetrics(h.Metrics)
-}
-func validRuntimeTransition(from, to string) bool {
-	if from == to {
-		return true
-	}
-	switch from {
-	case "unconfigured":
-		return to == "starting" || to == "stopped"
-	case "starting":
-		return to == "running" || to == "degraded" || to == "failed" || to == "stopped"
-	case "running":
-		return to == "degraded" || to == "failed" || to == "stopped"
-	case "degraded":
-		return to == "running" || to == "failed" || to == "stopped"
-	case "failed":
-		return to == "starting" || to == "stopped"
-	case "stopped":
-		return to == "starting"
-	}
-	return false
-}

Sense/server/app/sense/integration/brain_control/store.go

@@ -1,159 +0,0 @@
-package brain_control
-
-import (
-	"encoding/json"
-	"errors"
-	"time"
-
-	"gorm.io/gorm"
-	"gorm.io/gorm/clause"
-)
-
-type GORMReplayStore struct{ DB *gorm.DB }
-
-func (s GORMReplayStore) Consume(principal, tokenID string, expiresAt, now time.Time) bool {
-	if s.DB == nil || principal == "" || tokenID == "" || !expiresAt.After(now) {
-		return false
-	}
-	return s.DB.Transaction(func(tx *gorm.DB) error {
-		if err := tx.Where("expires_at <= ?", now).Delete(&ReplayToken{}).Error; err != nil {
-			return err
-		}
-		result := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&ReplayToken{Principal: principal, TokenID: tokenID, ExpiresAt: expiresAt, CreatedAt: now})
-		if result.Error != nil {
-			return result.Error
-		}
-		if result.RowsAffected != 1 {
-			return errors.New("replayed")
-		}
-		return nil
-	}) == nil
-}
-
-type RevisionStore struct{ DB *gorm.DB }
-
-func (s RevisionStore) Next(configID string) (int64, error) {
-	if s.DB == nil || !stableID.MatchString(configID) {
-		return 0, errors.New("invalid revision store")
-	}
-	var next int64
-	err := s.DB.Transaction(func(tx *gorm.DB) error {
-		var row SourceRevision
-		err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("config_id = ?", configID).First(&row).Error
-		if errors.Is(err, gorm.ErrRecordNotFound) {
-			row = SourceRevision{ConfigID: configID, Revision: 1}
-			if err = tx.Create(&row).Error; err != nil {
-				return err
-			}
-			next = 1
-			return nil
-		}
-		if err != nil {
-			return err
-		}
-		row.Revision++
-		next = row.Revision
-		return tx.Save(&row).Error
-	})
-	return next, err
-}
-
-type ProjectionStore struct {
-	DB         *gorm.DB
-	Clock      func() time.Time
-	StaleAfter time.Duration
-	FutureSkew time.Duration
-}
-type ProjectionView struct {
-	Projection                                  RuntimeProjection
-	Offline, Stale, Recovered, RevisionMismatch bool
-}
-
-func (s ProjectionStore) Ingest(raw []byte, expected map[string]int64) (ProjectionView, error) {
-	if s.DB == nil {
-		return ProjectionView{}, errors.New("projection database required")
-	}
-	now := time.Now().UTC()
-	if s.Clock != nil {
-		now = s.Clock().UTC()
-	}
-	if s.StaleAfter == 0 {
-		s.StaleAfter = 90 * time.Second
-	}
-	if s.FutureSkew == 0 {
-		s.FutureSkew = 30 * time.Second
-	}
-	status, err := parseRuntimeStatus(raw)
-	if err != nil {
-		return ProjectionView{}, err
-	}
-	if status.ObservedAt.After(now.Add(s.FutureSkew)) {
-		return ProjectionView{}, errors.New("FUTURE_OBSERVATION")
-	}
-	var view ProjectionView
-	err = s.DB.Transaction(func(tx *gorm.DB) error {
-		var old RuntimeProjection
-		find := tx.Where("brain_instance_ref = ?", status.BrainInstanceRef).First(&old).Error
-		if find == nil {
-			if old.StatusID == status.StatusID {
-				view.Projection = old
-				return nil
-			}
-			if status.Sequence <= old.Sequence {
-				return errors.New("OUT_OF_ORDER_STATUS")
-			}
-			if !validRuntimeTransition(old.RuntimeState, status.Runtime.State) {
-				return errors.New("INVALID_STATUS_TRANSITION")
-			}
-		}
-		if find != nil && !errors.Is(find, gorm.ErrRecordNotFound) {
-			return find
-		}
-		expectedJSON, _ := json.Marshal(expected)
-		configs, _ := json.Marshal(status.Configurations)
-		health, _ := json.Marshal(status.Health)
-		inputs, _ := json.Marshal(status.Inputs)
-		p := RuntimeProjection{BrainInstanceRef: status.BrainInstanceRef, StatusID: status.StatusID, Sequence: status.Sequence, ObservedAt: status.ObservedAt, ReceivedAt: now, RuntimeState: status.Runtime.State, RuntimeVersion: status.Runtime.Version, ModelRef: status.Model.ModelRef, ModelVersion: status.Model.Version, HealthOverall: status.Health.Overall, ExpectedRevisionsJSON: string(expectedJSON), ConfigurationsJSON: string(configs), HealthJSON: string(health), InputsJSON: string(inputs), WasOffline: find == nil && now.Sub(old.ObservedAt) > s.StaleAfter}
-		if find == nil {
-			p.CreatedAt = old.CreatedAt
-			view.Recovered = p.WasOffline
-		}
-		if err := tx.Save(&p).Error; err != nil {
-			return err
-		}
-		view.Projection = p
-		return nil
-	})
-	if err != nil {
-		return ProjectionView{}, err
-	}
-	view.Stale = now.Sub(view.Projection.ObservedAt) > s.StaleAfter
-	view.Offline = view.Stale
-	applied := make(map[string]*int64, len(status.Configurations))
-	for _, c := range status.Configurations {
-		applied[c.ConfigID] = c.AppliedRevision
-	}
-	for configID, want := range expected {
-		got, ok := applied[configID]
-		if !ok || got == nil || *got != want {
-			view.RevisionMismatch = true
-		}
-	}
-	return view, nil
-}
-
-func (s ProjectionStore) View(instance string) (ProjectionView, error) {
-	var p RuntimeProjection
-	if err := s.DB.First(&p, "brain_instance_ref = ?", instance).Error; err != nil {
-		return ProjectionView{}, err
-	}
-	now := time.Now().UTC()
-	if s.Clock != nil {
-		now = s.Clock().UTC()
-	}
-	stale := s.StaleAfter
-	if stale == 0 {
-		stale = 90 * time.Second
-	}
-	return ProjectionView{Projection: p, Offline: now.Sub(p.ObservedAt) > stale, Stale: now.Sub(p.ObservedAt) > stale}, nil
-}

Sense/server/app/sense/local_event/outbox.go

@@ -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
 	})
 }

Sense/server/app/sense/local_event/outbox_test.go

@@ -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)

Sense/server/cmd/api/bell_connector_test.go

@@ -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
+}

Sense/server/cmd/api/server.go

@@ -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() {

Sense/server/cmd/migrate/migration/version/2026083112000_bell_connector.go

@@ -4,22 +4,27 @@ import (
 	"runtime"
 
 	"gorm.io/gorm"
+	"gorm.io/gorm/clause"
 
-	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/brain_control"
+	"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), migrateBrainRuntime)
+	migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateSenseBellConnector)
 }
 
-func migrateBrainRuntime(db *gorm.DB, version string) error {
+func migrateSenseBellConnector(db *gorm.DB, version string) error {
 	return db.Transaction(func(tx *gorm.DB) error {
-		if err := tx.AutoMigrate(&brain_control.ReplayToken{}, &brain_control.RuntimeProjection{}, &brain_control.SourceRevision{}); err != nil {
+		if err := tx.AutoMigrate(
+			&bell_connector.InboundEvent{},
+			&bell_connector.EvidenceRecord{},
+			&bell_connector.ReplayToken{},
+		); err != nil {
 			return err
 		}
-		return tx.Create(&common.Migration{Version: version}).Error
+		return tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&common.Migration{Version: version}).Error
 	})
 }

Sense/server/cmd/migrate/migration/version/2026083112000_bell_connector_test.go

@@ -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)
+	}
+}

Sense/server/cmd/migrate/migration/version/2026083112000_brain_runtime_test.go

@@ -1,28 +0,0 @@
-package version
-
-import (
-	"testing"
-
-	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/brain_control"
-	common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
-	"gorm.io/driver/sqlite"
-	"gorm.io/gorm"
-)
-
-func TestMigrateBrainRuntime(t *testing.T) {
-	db, err := gorm.Open(sqlite.Open("file:brain-runtime-migration?mode=memory&cache=shared"), &gorm.Config{})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err = db.AutoMigrate(&common.Migration{}); err != nil {
-		t.Fatal(err)
-	}
-	if err = migrateBrainRuntime(db, "2026083112000"); err != nil {
-		t.Fatal(err)
-	}
-	for _, model := range []any{&brain_control.ReplayToken{}, &brain_control.RuntimeProjection{}, &brain_control.SourceRevision{}} {
-		if !db.Migrator().HasTable(model) {
-			t.Fatalf("missing table for %T", model)
-		}
-	}
-}

Sense/tests/integration/bell_connector/connector_test.go

@@ -0,0 +1,235 @@
+package bell_connector_test
+
+import (
+	"bytes"
+	"context"
+	"crypto/ed25519"
+	"crypto/rand"
+	"crypto/tls"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strings"
+	"sync/atomic"
+	"testing"
+	"time"
+
+	"github.com/gin-gonic/gin"
+	"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/integration/machine_identity"
+	"git.ilapage.cn/ila/yovision/Sense/server/app/sense/outbox"
+)
+
+func TestPersistentOutboxRetriesWithoutChangingOriginalIdentity(t *testing.T) {
+	db, databasePath := openDatabase(t)
+	body := fixture(t)
+	_, privateKey, _ := ed25519.GenerateKey(rand.Reader)
+	signer := machine_identity.Signer{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PrivateKey: privateKey, Now: func() time.Time { return time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC) }}
+	var attempts atomic.Int32
+	var received [][]byte
+	server := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		payload := make([]byte, request.ContentLength)
+		_, _ = request.Body.Read(payload)
+		received = append(received, payload)
+		requestID := request.Header.Get("X-Request-ID")
+		writer.Header().Set("X-Request-ID", requestID)
+		if request.Header.Get("X-YoVision-Relay-ID") != "sense-school-a" || !strings.HasPrefix(request.Header.Get("Authorization"), "Bearer ") || len(requestID) < 16 || len(requestID) > 128 {
+			http.Error(writer, "missing relay identity", http.StatusUnauthorized)
+			return
+		}
+		if attempts.Add(1) == 1 {
+			writer.WriteHeader(http.StatusServiceUnavailable)
+			_, _ = writer.Write([]byte(`{"code":"ingest_unavailable","message":"temporarily unavailable"}`))
+			return
+		}
+		writer.Header().Set("Content-Type", "application/json")
+		writer.WriteHeader(http.StatusCreated)
+		_, _ = writer.Write([]byte(`{"event_id":"bell-event-1","producer_id":"brain-school-a","source_event_id":"evt-area-20260831-0001","disposition":"created","payload_sha256":"4cc1e93820195caf713ea675ff33f178c9d4997dd8a81cb61287e9fea0e3d5e1"}`))
+	}))
+	defer server.Close()
+
+	now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
+	if _, err := bell_connector.EnqueueEvent(db, body, now); err != nil {
+		t.Fatal(err)
+	}
+	client := &bell_connector.Client{Endpoint: server.URL, RelayID: "sense-school-a", Signer: signer, HTTP: server.Client(), Enabled: true}
+	relay := bell_connector.Relay{DB: db, Client: client, Now: func() time.Time { return now }, Backoff: func(int) time.Duration { return 0 }}
+	if delivered, err := relay.DeliverBatch(context.Background(), "worker-1", 10); err != nil || delivered != 0 {
+		t.Fatalf("unavailable delivery=%d err=%v", delivered, err)
+	}
+	var queued outbox.Message
+	if err := db.First(&queued).Error; err != nil || queued.State != outbox.StateRetry {
+		t.Fatalf("outbox was not retained for retry: state=%s err=%v", queued.State, err)
+	}
+	sqlDatabase, err := db.DB()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = sqlDatabase.Close(); err != nil {
+		t.Fatal(err)
+	}
+	db = openDatabasePath(t, databasePath)
+	if err = db.First(&queued).Error; err != nil {
+		t.Fatal(err)
+	}
+
+	// Constructing a new relay simulates restart; the durable row is claimed
+	// and delivered with its original bytes and business identity.
+	restarted := bell_connector.Relay{DB: db, Client: client, Now: func() time.Time { return now }, Backoff: func(int) time.Duration { return 0 }}
+	if delivered, err := restarted.DeliverBatch(context.Background(), "worker-2", 10); err != nil || delivered != 1 {
+		t.Fatalf("recovery delivery=%d err=%v", delivered, err)
+	}
+	if len(received) != 2 || string(received[0]) != string(body) || string(received[1]) != string(body) || bell_connector.PreserveIdentity(received[0], received[1]) != nil {
+		t.Fatal("relay changed the frozen payload or original identity")
+	}
+	if err := db.First(&queued).Error; err != nil || queued.State != outbox.StateDelivered {
+		t.Fatalf("outbox not delivered: state=%s err=%v", queued.State, err)
+	}
+}
+
+func TestTerminalConflictDisabledTimeoutAndReplayRestart(t *testing.T) {
+	db, databasePath := openDatabase(t)
+	body := fixture(t)
+	_, privateKey, _ := ed25519.GenerateKey(rand.Reader)
+	signer := machine_identity.Signer{Principal: "yv:sense:school-a", KeyID: "sense-key-0001", PrivateKey: privateKey}
+	policy := machine_identity.TransportPolicy{TLSMinVersion: tls.VersionTLS12, VerifyCertificate: true, VerifyHostname: true, ConnectTimeout: time.Second, ResponseHeaderTimeout: time.Second, RequestTimeout: time.Second, MaxRequestBytes: 64 * 1024}
+	if _, err := bell_connector.NewClient("https://user:pass@bell.example", "sense-school-a", signer, policy); err == nil || !strings.Contains(err.Error(), "HTTPS origin") {
+		t.Fatalf("endpoint userinfo was not rejected: %v", err)
+	}
+	conflictServer := httptest.NewServer(http.HandlerFunc(func(writer http.ResponseWriter, request *http.Request) {
+		writer.WriteHeader(http.StatusConflict)
+		_, _ = writer.Write([]byte(`{"code":"idempotency_conflict","message":"conflict","existing_event_id":"bell-1"}`))
+	}))
+	defer conflictServer.Close()
+	if _, err := bell_connector.EnqueueEvent(db, body, time.Now()); err != nil {
+		t.Fatal(err)
+	}
+	client := &bell_connector.Client{Endpoint: conflictServer.URL, Signer: signer, HTTP: conflictServer.Client(), Enabled: true}
+	if _, err := (bell_connector.Relay{DB: db, Client: client}).DeliverBatch(context.Background(), "worker", 1); err != nil {
+		t.Fatal(err)
+	}
+	var message outbox.Message
+	if err := db.First(&message).Error; err != nil || message.State != outbox.StateDead {
+		t.Fatalf("terminal conflict was retried: state=%s err=%v", message.State, err)
+	}
+	if _, err := (bell_connector.Client{Enabled: false}).Send(context.Background(), body); err == nil || !strings.Contains(err.Error(), "connector_disabled") {
+		t.Fatalf("disabled connector error=%v", err)
+	}
+	cancelled, cancel := context.WithCancel(context.Background())
+	cancel()
+	if _, err := client.Send(cancelled, body); err == nil {
+		t.Fatal("cancelled/timeout request unexpectedly succeeded")
+	}
+
+	now := time.Now().UTC()
+	firstStore := bell_connector.PersistentReplayStore{DB: db}
+	if !firstStore.Consume("yv:bell:school-a", "abcdefghijklmnopqrstuv", now.Add(time.Minute), now) {
+		t.Fatal("first replay consume failed")
+	}
+	sqlDatabase, err := db.DB()
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = sqlDatabase.Close(); err != nil {
+		t.Fatal(err)
+	}
+	db = openDatabasePath(t, databasePath)
+	restartedStore := bell_connector.PersistentReplayStore{DB: db}
+	if restartedStore.Consume("yv:bell:school-a", "abcdefghijklmnopqrstuv", now.Add(time.Minute), now) {
+		t.Fatal("replay was accepted after store restart")
+	}
+}
+
+func TestBrainIngressAndBellEvidenceEndpoint(t *testing.T) {
+	gin.SetMode(gin.TestMode)
+	db, _ := openDatabase(t)
+	brainPublic, brainPrivate, _ := ed25519.GenerateKey(rand.Reader)
+	bellPublic, bellPrivate, _ := ed25519.GenerateKey(rand.Reader)
+	registry, err := machine_identity.NewRegistry(
+		machine_identity.KeyRecord{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PublicKey: brainPublic, Audience: "yovision-sense", Scopes: []string{"events:ingest"}, Enabled: true},
+		machine_identity.KeyRecord{Principal: "yv:bell:school-a", KeyID: "bell-key-0001", PublicKey: bellPublic, Audience: "yovision-sense", Scopes: []string{"evidence:read"}, Enabled: true},
+	)
+	if err != nil {
+		t.Fatal(err)
+	}
+	now := time.Date(2026, 8, 31, 1, 0, 0, 0, time.UTC)
+	verifier := machine_identity.Verifier{Registry: registry, Replay: bell_connector.PersistentReplayStore{DB: db}, Now: func() time.Time { return now }}
+	body := fixture(t)
+	brainSigner := machine_identity.Signer{Principal: "yv:brain:school-a", KeyID: "brain-key-0001", PrivateKey: brainPrivate, Now: func() time.Time { return now }}
+	token, err := brainSigner.Mint("yovision-sense", []string{"events:ingest"}, http.MethodPost, "/v1/events", body)
+	if err != nil {
+		t.Fatal(err)
+	}
+	request := httptest.NewRequest(http.MethodPost, "/v1/events", bytes.NewReader(body))
+	request.Header.Set("Authorization", "Bearer "+token)
+	request.Header.Set("X-Request-ID", "request-id-0000001")
+	response := httptest.NewRecorder()
+	ginContext, _ := gin.CreateTestContext(response)
+	ginContext.Request = request
+	(bell_connector.IngressHandler{DB: db, Verifier: verifier, EvidenceOwnerID: "sense-school-a", Now: func() time.Time { return now }}).Post(ginContext)
+	if response.Code != http.StatusAccepted {
+		t.Fatalf("Brain ingress status=%d body=%s", response.Code, response.Body.String())
+	}
+	var evidence bell_connector.EvidenceRecord
+	if err = db.First(&evidence, "evidence_id = ?", "ev-school-east-0001").Error; err != nil {
+		t.Fatal(err)
+	}
+	bellSigner := machine_identity.Signer{Principal: "yv:bell:school-a", KeyID: "bell-key-0001", PrivateKey: bellPrivate, Now: func() time.Time { return now }}
+	path := "/v1/evidence/ev-school-east-0001"
+	token, err = bellSigner.Mint("yovision-sense", []string{"evidence:read"}, http.MethodGet, path, nil)
+	if err != nil {
+		t.Fatal(err)
+	}
+	getRequest := httptest.NewRequest(http.MethodGet, path, nil)
+	getRequest.Header.Set("Authorization", "Bearer "+token)
+	getRequest.Header.Set("X-Request-ID", "request-id-0000002")
+	getResponse := httptest.NewRecorder()
+	getContext, _ := gin.CreateTestContext(getResponse)
+	getContext.Request = getRequest
+	getContext.Params = gin.Params{{Key: "evidence_id", Value: "ev-school-east-0001"}}
+	(bell_connector.EvidenceHandler{DB: db, Verifier: verifier, Now: func() time.Time { return now }}).Get(getContext)
+	if getResponse.Code != http.StatusOK || !bytes.Equal(getResponse.Body.Bytes(), evidence.Payload) {
+		t.Fatalf("evidence lookup status=%d body=%s", getResponse.Code, getResponse.Body.String())
+	}
+}
+
+func openDatabase(t *testing.T) (*gorm.DB, string) {
+	databasePath := filepath.Join(t.TempDir(), "sense-bell-connector.sqlite")
+	return openDatabasePath(t, databasePath), databasePath
+}
+
+func openDatabasePath(t *testing.T, databasePath string) *gorm.DB {
+	t.Helper()
+	db, err := gorm.Open(sqlite.Open(databasePath), &gorm.Config{})
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err = db.AutoMigrate(&outbox.Message{}, &outbox.DeliveryRecord{}, &outbox.Attempt{}, &bell_connector.ReplayToken{}, &bell_connector.InboundEvent{}, &bell_connector.EvidenceRecord{}); err != nil {
+		t.Fatal(err)
+	}
+	sqlDatabase, err := db.DB()
+	if err != nil {
+		t.Fatal(err)
+	}
+	t.Cleanup(func() { _ = sqlDatabase.Close() })
+	return db
+}
+
+func fixture(t *testing.T) []byte {
+	t.Helper()
+	path := filepath.Join("..", "..", "..", "..", "contracts", "events", "v1", "examples", "dangerous-area.json")
+	body, err := os.ReadFile(path)
+	if err != nil {
+		t.Fatal(err)
+	}
+	var event map[string]any
+	if json.Unmarshal(body, &event) != nil {
+		t.Fatal("invalid fixture")
+	}
+	return body
+}

Sense/tests/integration/bell_connector/go.mod

@@ -0,0 +1,72 @@
+module git.ilapage.cn/ila/yovision/Sense/tests/integration/bell_connector
+
+go 1.26.5
+
+require (
+	git.ilapage.cn/ila/yovision/Sense/server v0.0.0
+	gorm.io/driver/sqlite v1.6.0
+	gorm.io/gorm v1.31.2
+)
+
+require (
+	dario.cat/mergo v1.0.1 // indirect
+	github.com/BurntSushi/toml v1.5.0 // indirect
+	github.com/andeya/ameda v1.5.3 // indirect
+	github.com/andeya/goutil v1.1.2 // indirect
+	github.com/bitly/go-simplejson v0.5.1 // indirect
+	github.com/bmatcuk/doublestar/v4 v4.10.0 // indirect
+	github.com/bytedance/go-tagexpr/v2 v2.9.11 // indirect
+	github.com/bytedance/gopkg v0.1.4 // indirect
+	github.com/bytedance/sonic v1.15.2 // indirect
+	github.com/bytedance/sonic/loader v0.5.2 // indirect
+	github.com/casbin/casbin/v2 v2.135.0 // indirect
+	github.com/casbin/govaluate v1.10.0 // indirect
+	github.com/chanxuehong/rand v0.0.0-20211009035549-2f07823e8e99 // indirect
+	github.com/chanxuehong/wechat v0.0.0-20230222024006-36f0325263cd // indirect
+	github.com/cloudwego/base64x v0.1.7 // indirect
+	github.com/fsnotify/fsnotify v1.9.0 // indirect
+	github.com/gabriel-vasile/mimetype v1.4.15 // indirect
+	github.com/ghodss/yaml v1.0.0 // indirect
+	github.com/gin-contrib/sse v1.1.1 // indirect
+	github.com/gin-gonic/gin v1.12.0 // indirect
+	github.com/go-admin-team/go-admin-core v1.5.3-rc.3.0.20250408121721-2763de5dcdf4 // indirect
+	github.com/go-admin-team/go-admin-core/plugins/logger/zap v1.5.2 // indirect
+	github.com/go-admin-team/go-admin-core/sdk v1.5.3-rc.3.0.20250408121721-2763de5dcdf4 // indirect
+	github.com/go-playground/locales v0.14.1 // indirect
+	github.com/go-playground/universal-translator v0.18.1 // indirect
+	github.com/go-playground/validator/v10 v10.30.3 // indirect
+	github.com/goccy/go-json v0.10.6 // indirect
+	github.com/goccy/go-yaml v1.19.2 // indirect
+	github.com/golang-jwt/jwt/v5 v5.3.1 // indirect
+	github.com/google/uuid v1.6.0 // indirect
+	github.com/jinzhu/inflection v1.0.0 // indirect
+	github.com/jinzhu/now v1.1.5 // indirect
+	github.com/json-iterator/go v1.1.12 // indirect
+	github.com/klauspost/cpuid/v2 v2.4.0 // indirect
+	github.com/leodido/go-urn v1.5.0 // indirect
+	github.com/mattn/go-isatty v0.0.24 // indirect
+	github.com/mattn/go-sqlite3 v1.14.49 // indirect
+	github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd // indirect
+	github.com/modern-go/reflect2 v1.0.2 // indirect
+	github.com/nyaruka/phonenumbers v1.2.2 // indirect
+	github.com/pelletier/go-toml/v2 v2.4.3 // indirect
+	github.com/pkg/errors v0.9.1 // indirect
+	github.com/quic-go/qpack v0.6.0 // indirect
+	github.com/quic-go/quic-go v0.61.0 // indirect
+	github.com/robfig/cron/v3 v3.0.1 // indirect
+	github.com/spf13/cast v1.7.1 // indirect
+	github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
+	github.com/ugorji/go/codec v1.3.2 // indirect
+	go.mongodb.org/mongo-driver/v2 v2.8.0 // indirect
+	go.uber.org/multierr v1.11.0 // indirect
+	go.uber.org/zap v1.27.0 // indirect
+	golang.org/x/arch v0.30.0 // indirect
+	golang.org/x/crypto v0.54.0 // indirect
+	golang.org/x/net v0.57.0 // indirect
+	golang.org/x/sys v0.47.0 // indirect
+	golang.org/x/text v0.40.0 // indirect
+	google.golang.org/protobuf v1.36.11 // indirect
+	gopkg.in/yaml.v2 v2.4.0 // indirect
+)
+
+replace git.ilapage.cn/ila/yovision/Sense/server => ../../../server

Sense/tests/integration/bell_connector/go.sum

@@ -0,0 +1,204 @@
+dario.cat/mergo v1.0.1 h1:Ra4+bf83h2ztPIQYNP99R6m+Y7KfnARDfID+a+vLl4s=
+dario.cat/mergo v1.0.1/go.mod h1:uNxQE+84aUszobStD9th8a29P2fMDhsBdgRYvZOxGmk=
+github.com/BurntSushi/toml v1.5.0 h1:W5quZX/G/csjUnuI8SUYlsHs9M38FC7znL0lIO+DvMg=
+github.com/BurntSushi/toml v1.5.0/go.mod h1:ukJfTF/6rtPPRCnwkur4qwRxa8vTRFBF0uk2lLoLwho=
+github.com/andeya/ameda v1.5.3 h1:SvqnhQPZwwabS8HQTRGfJwWPl2w9ZIPInHAw9aE1Wlk=
+github.com/andeya/ameda v1.5.3/go.mod h1:FQDHRe1I995v6GG+8aJ7UIUToEmbdTJn/U26NCPIgXQ=
+github.com/andeya/goutil v1.0.1/go.mod h1:jEG5/QnnhG7yGxwFUX6Q+JGMif7sjdHmmNVjn7nhJDo=
+github.com/andeya/goutil v1.1.2 h1:RiFWFkL/9yXh2SjQkNWOHqErU1x+RauHmeR23eNUzSg=
+github.com/andeya/goutil v1.1.2/go.mod h1:jEG5/QnnhG7yGxwFUX6Q+JGMif7sjdHmmNVjn7nhJDo=
+github.com/bitly/go-simplejson v0.5.1 h1:xgwPbetQScXt1gh9BmoJ6j9JMr3TElvuIyjR8pgdoow=
+github.com/bitly/go-simplejson v0.5.1/go.mod h1:YOPVLzCfwK14b4Sff3oP1AmGhI9T9Vsg84etUnlyp+Q=
+github.com/bmatcuk/doublestar/v4 v4.6.1/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc=
+github.com/bmatcuk/doublestar/v4 v4.10.0 h1:zU9WiOla1YA122oLM6i4EXvGW62DvKZVxIe6TYWexEs=
+github.com/bmatcuk/doublestar/v4 v4.10.0/go.mod h1:xBQ8jztBU6kakFMg+8WGxn0c6z1fTSPVIjEY1Wr7jzc=
+github.com/bytedance/go-tagexpr/v2 v2.9.11 h1:jJgmoDKPKacGl0llPYbYL/+/2N+Ng0vV0ipbnVssXHY=
+github.com/bytedance/go-tagexpr/v2 v2.9.11/go.mod h1:UAyKh4ZRLBPGsyTRFZoPqTni1TlojMdOJXQnEIPCX84=
+github.com/bytedance/gopkg v0.1.4 h1:oZnQwnX82KAIWb7033bEwtxvTqXcYMxDBaQxo5JJHWM=
+github.com/bytedance/gopkg v0.1.4/go.mod h1:v1zWfPm21Fb+OsyXN2VAHdL6TBb2L88anLQgdyje6R4=
+github.com/bytedance/sonic v1.15.2 h1:90H+rcF/FwLXwfB1cudOLq/je83n683Utf4Cbp0xHCo=
+github.com/bytedance/sonic v1.15.2/go.mod h1:mT2NbXunuaEbnZ+mRIX/vYqKISmgEuHFDI4UzmKx2SA=
+github.com/bytedance/sonic/loader v0.5.2 h1:0QtP1gevc1OZ6/H8Lb9BRZiCXd1Ftjd3OKuj1T1lBIo=
+github.com/bytedance/sonic/loader v0.5.2/go.mod h1:AR4NYCk5DdzZizZ5djGqQ92eEhCCcdf5x77udYiSJRo=
+github.com/casbin/casbin/v2 v2.135.0 h1:6BLkMQiGotYyS5yYeWgW19vxqugUlvHFkFiLnLR/bxk=
+github.com/casbin/casbin/v2 v2.135.0/go.mod h1:FmcfntdXLTcYXv/hxgNntcRPqAbwOG9xsism0yXT+18=
+github.com/casbin/govaluate v1.3.0/go.mod h1:G/UnbIjZk/0uMNaLwZZmFQrR72tYRZWQkO70si/iR7A=
+github.com/casbin/govaluate v1.10.0 h1:ffGw51/hYH3w3rZcxO/KcaUIDOLP84w7nsidMVgaDG0=
+github.com/casbin/govaluate v1.10.0/go.mod h1:G/UnbIjZk/0uMNaLwZZmFQrR72tYRZWQkO70si/iR7A=
+github.com/chanxuehong/rand v0.0.0-20211009035549-2f07823e8e99 h1:K62Lb6bsgLOB++z/VAvRvtiEBdNCuMfmQGTGGWMdPpM=
+github.com/chanxuehong/rand v0.0.0-20211009035549-2f07823e8e99/go.mod h1:9+sJ9zvvkXC5sPjPEZM3Jpb9n2Q2VtcrGZly0UHYF5I=
+github.com/chanxuehong/util v0.0.0-20200304121633-ca8141845b13/go.mod h1:XEYt99iTxMqkv+gW85JX/DdUINHUe43Sbe5AtqSaDAQ=
+github.com/chanxuehong/wechat v0.0.0-20230222024006-36f0325263cd h1:v3JNsFZmplLO/Cmiyr/rGvR7lW1ld9lB+d5h4yR0MTI=
+github.com/chanxuehong/wechat v0.0.0-20230222024006-36f0325263cd/go.mod h1:mysjrtCs9MmN8hqDf4/mc4eQ26Rt9s1p5oO+fhJlLB4=
+github.com/cloudwego/base64x v0.1.7 h1:NppS+Fgzg5ovhn4NkUXaDT3x9jldgH5ToMCqzBSi2zI=
+github.com/cloudwego/base64x v0.1.7/go.mod h1:Cu1PV9zfrSf7ET2tIbWbbEy7jO7HHJ13q4X2SQ8aWYg=
+github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/frankban/quicktest v1.14.6 h1:7Xjx+VpznH+oBnejlPUj8oUpdxnVs4f8XU8WnHkI4W8=
+github.com/frankban/quicktest v1.14.6/go.mod h1:4ptaffx2x8+WTWXmUCuVU6aPUX1/Mz7zb5vbUoiM6w0=
+github.com/fsnotify/fsnotify v1.9.0 h1:2Ml+OJNzbYCTzsxtv8vKSFD9PbJjmhYF14k/jKC7S9k=
+github.com/fsnotify/fsnotify v1.9.0/go.mod h1:8jBTzvmWwFyi3Pb8djgCCO5IBqzKJ/Jwo8TRcHyHii0=
+github.com/gabriel-vasile/mimetype v1.4.15 h1:05iP/CYtZ/w455R/KZM6rZ5ieAdh99UPtd+d3YzLmaI=
+github.com/gabriel-vasile/mimetype v1.4.15/go.mod h1:azpTcoLcDZRNgFou5j+APrqQx9HqVPWa6ijYQIIVswQ=
+github.com/ghodss/yaml v1.0.0 h1:wQHKEahhL6wmXdzwWG11gIVCkOv05bNOh+Rxn0yngAk=
+github.com/ghodss/yaml v1.0.0/go.mod h1:4dBDuWmgqj2HViK6kFavaiC9ZROes6MMH2rRYeMEF04=
+github.com/gin-contrib/sse v1.1.1 h1:uGYpNwTacv5R68bSGMapo62iLTRa9l5zxGCps4hK6ko=
+github.com/gin-contrib/sse v1.1.1/go.mod h1:QXzuVkA0YO7o/gun03UI1Q+FTI8ZV/n5t03kIQAI89s=
+github.com/gin-gonic/gin v1.12.0 h1:b3YAbrZtnf8N//yjKeU2+MQsh2mY5htkZidOM7O0wG8=
+github.com/gin-gonic/gin v1.12.0/go.mod h1:VxccKfsSllpKshkBWgVgRniFFAzFb9csfngsqANjnLc=
+github.com/go-admin-team/go-admin-core v1.5.3-rc.3.0.20250408121721-2763de5dcdf4 h1:UI9ppj+iWl/0Y6kp9w06dwGUdLsdxWRC3p0J0gLHApI=
+github.com/go-admin-team/go-admin-core v1.5.3-rc.3.0.20250408121721-2763de5dcdf4/go.mod h1:uWX7fPisJ6DluUP9vR3m3818RkDpZb/4dnwbZdmZN6Q=
+github.com/go-admin-team/go-admin-core/plugins/logger/zap v1.5.2 h1:cPTLzpvvyh8kyB24jblB+2W0QZBugP+8VYN3R34Pb4s=
+github.com/go-admin-team/go-admin-core/plugins/logger/zap v1.5.2/go.mod h1:ejtJ3aohd6EznZ9Q+KZVA3NwPU/2qIm0gayIGM3tIXw=
+github.com/go-admin-team/go-admin-core/sdk v1.5.3-rc.3.0.20250408121721-2763de5dcdf4 h1:gU2OBSCsfSrqWId2monoEQhPKXZqhZ8W9ol9f7A4DAE=
+github.com/go-admin-team/go-admin-core/sdk v1.5.3-rc.3.0.20250408121721-2763de5dcdf4/go.mod h1:va1lNEXHGnV161Avr0lzi5gnT8OazJ/wmN9xnsY9N/s=
+github.com/go-playground/assert/v2 v2.2.0 h1:JvknZsQTYeFEAhQwI4qEt9cyV5ONwRHC+lYKSsYSR8s=
+github.com/go-playground/assert/v2 v2.2.0/go.mod h1:VDjEfimB/XKnb+ZQfWdccd7VUvScMdVu0Titje2rxJ4=
+github.com/go-playground/locales v0.14.1 h1:EWaQ/wswjilfKLTECiXz7Rh+3BjFhfDFKv/oXslEjJA=
+github.com/go-playground/locales v0.14.1/go.mod h1:hxrqLVvrK65+Rwrd5Fc6F2O76J/NuW9t0sjnWqG1slY=
+github.com/go-playground/universal-translator v0.18.1 h1:Bcnm0ZwsGyWbCzImXv+pAJnYK9S473LQFuzCbDbfSFY=
+github.com/go-playground/universal-translator v0.18.1/go.mod h1:xekY+UJKNuX9WP91TpwSH2VMlDf28Uj24BCp08ZFTUY=
+github.com/go-playground/validator/v10 v10.30.3 h1:4MU6YkEwx7GbcPJOZxrtbu+QfF3pJLJuaYTeAH0DYy8=
+github.com/go-playground/validator/v10 v10.30.3/go.mod h1:4Axh7oCNGcoGkqLoE4YWt6n20mcEIsPRlB7vPk3lpyc=
+github.com/goccy/go-json v0.10.6 h1:p8HrPJzOakx/mn/bQtjgNjdTcN+/S6FcG2CTtQOrHVU=
+github.com/goccy/go-json v0.10.6/go.mod h1:oq7eo15ShAhp70Anwd5lgX2pLfOS3QCiwU/PULtXL6M=
+github.com/goccy/go-yaml v1.19.2 h1:PmFC1S6h8ljIz6gMRBopkjP1TVT7xuwrButHID66PoM=
+github.com/goccy/go-yaml v1.19.2/go.mod h1:XBurs7gK8ATbW4ZPGKgcbrY1Br56PdM69F7LkFRi1kA=
+github.com/golang-jwt/jwt/v5 v5.3.1 h1:kYf81DTWFe7t+1VvL7eS+jKFVWaUnK9cB1qbwn63YCY=
+github.com/golang-jwt/jwt/v5 v5.3.1/go.mod h1:fxCRLWMO43lRc8nhHWY6LGqRcf+1gQWArsqaEUEa5bE=
+github.com/golang/mock v1.4.4 h1:l75CXGRSwbaYNpl/Z2X1XIIAMSCquvXgpVZDhwEIJsc=
+github.com/golang/mock v1.4.4/go.mod h1:l3mdAwkq5BuhzHwde/uurv3sEJeZMXNpwsxVWU71h+4=
+github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
+github.com/golang/protobuf v1.5.0/go.mod h1:FsONVRAS9T7sI+LIUmWTfcYkHO4aIWwzhcaSAoJOfIk=
+github.com/google/go-cmp v0.5.5/go.mod h1:v8dTdLbMG2kIc/vJvl+f65V22dbkXbowE6jgT/gNBxE=
+github.com/google/go-cmp v0.7.0 h1:wk8382ETsv4JYUZwIsn6YpYiWiBsYLSJiTsyBybVuN8=
+github.com/google/go-cmp v0.7.0/go.mod h1:pXiqmnSA92OHEEa9HXL2W4E7lf9JzCmGVUdgjX3N/iU=
+github.com/google/gofuzz v1.0.0/go.mod h1:dBl0BpW6vV/+mYPU4Po3pmUjxk6FQPldtuIdl/M65Eg=
+github.com/google/uuid v1.6.0 h1:NIvaJDMOsjHA8n1jAhLSgzrAzy1Hgr+hNrb57e+94F0=
+github.com/google/uuid v1.6.0/go.mod h1:TIyPZe4MgqvfeYDBFedMoGGpEw/LqOeaOT+nhxU+yHo=
+github.com/gopherjs/gopherjs v1.17.2 h1:fQnZVsXk8uxXIStYb0N4bGk7jeyTalG/wsZjQ25dO0g=
+github.com/gopherjs/gopherjs v1.17.2/go.mod h1:pRRIvn/QzFLrKfvEz3qUuEhtE/zLCWfreZ6J5gM2i+k=
+github.com/jackc/pgpassfile v1.0.0 h1:/6Hmqy13Ss2zCq62VdNG8tM1wchn8zjSGOBJ6icpsIM=
+github.com/jackc/pgpassfile v1.0.0/go.mod h1:CEx0iS5ambNFdcRtxPj5JhEz+xB6uRky5eyVu/W2HEg=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761 h1:iCEnooe7UlwOQYpKFhBabPMi4aNAfoODPEFNiAnClxo=
+github.com/jackc/pgservicefile v0.0.0-20240606120523-5a60cdf6a761/go.mod h1:5TJZWKEWniPve33vlWYSoGYefn3gLQRzjfDlhSJ9ZKM=
+github.com/jackc/pgx/v5 v5.10.0 h1:VhSvgU2jSli8o3AqIEOTJr7rZwAEUVo4E4XhR94Zfr0=
+github.com/jackc/pgx/v5 v5.10.0/go.mod h1:mal1tBGAFfLHvZzaYh77YS/eC6IX9OWbRV1QIIM0Jn4=
+github.com/jackc/puddle/v2 v2.2.2 h1:PR8nw+E/1w0GLuRFSmiioY6UooMp6KJv0/61nB7icHo=
+github.com/jackc/puddle/v2 v2.2.2/go.mod h1:vriiEXHvEE654aYKXXjOvZM39qJ0q+azkZFrfEOc3H4=
+github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
+github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
+github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
+github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
+github.com/json-iterator/go v1.1.12 h1:PV8peI4a0ysnczrg+LtxykD8LfKY9ML6u2jnxaEnrnM=
+github.com/json-iterator/go v1.1.12/go.mod h1:e30LSqwooZae/UwlEbR2852Gd8hjQvJoHmT4TnhNGBo=
+github.com/jtolds/gls v4.20.0+incompatible h1:xdiiI2gbIgH/gLH7ADydsJ1uDOEzR8yvV7C0MuV77Wo=
+github.com/jtolds/gls v4.20.0+incompatible/go.mod h1:QJZ7F/aHp+rZTRtaJ1ow/lLfFfVYBRgL+9YlvaHOwJU=
+github.com/klauspost/cpuid/v2 v2.4.0 h1:S6Hrbc7+ywsr0r+RLapfGBHfyefhCTwEh3A0tV913Dw=
+github.com/klauspost/cpuid/v2 v2.4.0/go.mod h1:19jmZ9mjzoF//ddRSUsv0zfBTJWh3QJh9FNxZTMrGxU=
+github.com/kr/pretty v0.3.1 h1:flRD4NNwYAUpkphVc1HcthR4KEIFJ65n8Mw5qdRn3LE=
+github.com/kr/pretty v0.3.1/go.mod h1:hoEshYVHaxMs3cyo3Yncou5ZscifuDolrwPKZanG3xk=
+github.com/kr/text v0.2.0 h1:5Nx0Ya0ZqY2ygV366QzturHI13Jq95ApcVaJBhpS+AY=
+github.com/kr/text v0.2.0/go.mod h1:eLer722TekiGuMkidMxC/pM04lWEeraHUUmBw8l2grE=
+github.com/leodido/go-urn v1.5.0 h1:pLqT2kq1zpHW/1D18QMjMpdtX7cekxqtJJjg5ANyWw0=
+github.com/leodido/go-urn v1.5.0/go.mod h1:9BORnCDhdPBJNDEX+w1bJisa8yOKYi116VeO96s4ifE=
+github.com/mattn/go-isatty v0.0.24 h1:tGZZoVgT/KiqK1c8ocVLeDS8BSWMRd47J3Lbz7vsReI=
+github.com/mattn/go-isatty v0.0.24/go.mod h1:nMCL3Zebbrt45jsMDgnfIwz6ydEQApk5oEI3HqDio6A=
+github.com/mattn/go-sqlite3 v1.14.49 h1:B8jBHC3xhxZgxztrgruTuLucebnULQnx4W7cF7SAE9w=
+github.com/mattn/go-sqlite3 v1.14.49/go.mod h1:6JTjA44L93a0QCyJef5YvlPoKXntQPjzWv5gtm9sB6w=
+github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
+github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
+github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M=
+github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk=
+github.com/nyaruka/phonenumbers v1.0.55/go.mod h1:sDaTZ/KPX5f8qyV9qN+hIm+4ZBARJrupC6LuhshJq1U=
+github.com/nyaruka/phonenumbers v1.2.2 h1:OwVjf7Y4uHoK9VJUrA8ebR0ha2yc6sEYbfrwkq0asCY=
+github.com/nyaruka/phonenumbers v1.2.2/go.mod h1:wzk2qq7qwsaBKrfbkWKdgHYOOH+QFTesSpIq53ELw8M=
+github.com/pelletier/go-toml/v2 v2.4.3 h1:GTRvJQutkOSftxIFD5xw9aepkYNuPWmVJpffdDPYVpY=
+github.com/pelletier/go-toml/v2 v2.4.3/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY=
+github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
+github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/quic-go/go-ossfuzz-seeds v0.1.0 h1:APacT+iIaNF6fd8AGEiN3bT/Jtkd2jz4v4TzM7MFjy0=
+github.com/quic-go/go-ossfuzz-seeds v0.1.0/go.mod h1:3IOHRbJIc+L6YKMwfDtJAM9Vj9k0YY4muhuyUYk5tbk=
+github.com/quic-go/qpack v0.6.0 h1:g7W+BMYynC1LbYLSqRt8PBg5Tgwxn214ZZR34VIOjz8=
+github.com/quic-go/qpack v0.6.0/go.mod h1:lUpLKChi8njB4ty2bFLX2x4gzDqXwUpaO1DP9qMDZII=
+github.com/quic-go/quic-go v0.61.0 h1:ui88A53s8MSVYLC56en0KQ17HARk+9986Dn0SBfKNvA=
+github.com/quic-go/quic-go v0.61.0/go.mod h1:9So2anK4Tp22URSQq00k+Vo2PNkle96ycDPDHL4s9vs=
+github.com/robfig/cron/v3 v3.0.1 h1:WdRxkvbJztn8LMz/QEvLN5sBU+xKpSqwwUO1Pjr4qDs=
+github.com/robfig/cron/v3 v3.0.1/go.mod h1:eQICP3HwyT7UooqI/z+Ov+PtYAWygg1TEWWzGIFLtro=
+github.com/rogpeppe/go-internal v1.10.0 h1:TMyTOH3F/DB16zRVcYyreMH6GnZZrwQVAoYjRBZyWFQ=
+github.com/rogpeppe/go-internal v1.10.0/go.mod h1:UQnix2H7Ngw/k4C5ijL5+65zddjncjaFoBhdsK/akog=
+github.com/smarty/assertions v1.15.0 h1:cR//PqUBUiQRakZWqBiFFQ9wb8emQGDb0HeGdqGByCY=
+github.com/smarty/assertions v1.15.0/go.mod h1:yABtdzeQs6l1brC900WlRNwj6ZR55d7B+E8C6HtKdec=
+github.com/smartystreets/goconvey v1.8.1 h1:qGjIddxOk4grTu9JPOU31tVfq3cNdBlNa5sSznIX1xY=
+github.com/smartystreets/goconvey v1.8.1/go.mod h1:+/u4qLyY6x1jReYOp7GOM2FSt8aP9CzCZL03bI28W60=
+github.com/spf13/cast v1.7.1 h1:cuNEagBQEHWN1FnbGEjCXL2szYEXqfJPbP2HNUaca9Y=
+github.com/spf13/cast v1.7.1/go.mod h1:ancEpBxwJDODSW/UG4rDrAqiKolqNNh2DX3mk86cAdo=
+github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
+github.com/stretchr/objx v0.4.0/go.mod h1:YvHI0jy2hoMjB+UWwv71VJQ9isScKT/TqJzVSSt89Yw=
+github.com/stretchr/objx v0.5.0/go.mod h1:Yh+to48EsGEfYuaHDzXPcE3xhTkx73EhmCGUpEOglKo=
+github.com/stretchr/objx v0.5.2/go.mod h1:FRsXN1f5AsAjCGJKqEizvkpNtU+EGNCLh3NxZ/8L+MA=
+github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
+github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
+github.com/stretchr/testify v1.7.5/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
+github.com/stretchr/testify v1.8.4/go.mod h1:sz/lmYIOXD/1dqDmKjjqLyZ2RngseejIcXlSw2iwfAo=
+github.com/stretchr/testify v1.10.0/go.mod h1:r2ic/lqez/lEtzL7wO/rwa5dbSLXVDPFyf8C91i36aY=
+github.com/stretchr/testify v1.11.1 h1:7s2iGBzp5EwR7/aIZr8ao5+dra3wiQyKjjFuvgVKu7U=
+github.com/stretchr/testify v1.11.1/go.mod h1:wZwfW3scLgRK+23gO65QZefKpKQRnfz6sD981Nm4B6U=
+github.com/tidwall/match v1.1.1/go.mod h1:eRSPERbgtNPcGhD8UCthc6PmLEQXEWd3PRB5JTxsfmM=
+github.com/tidwall/pretty v1.2.0/go.mod h1:ITEVvHYasfjBbM0u2Pg8T2nJnzm8xPwvNhhsoaGGjNU=
+github.com/twitchyliquid64/golang-asm v0.15.1 h1:SU5vSMR7hnwNxj24w34ZyCi/FmDZTkS4MhqMhdFk5YI=
+github.com/twitchyliquid64/golang-asm v0.15.1/go.mod h1:a1lVb/DtPvCB8fslRZhAngC2+aY1QWCk3Cedj/Gdt08=
+github.com/ugorji/go/codec v1.3.2 h1:zkEASHHyEClGeURfgNT9PJZVfAbs9oEX9QXggwWNJbc=
+github.com/ugorji/go/codec v1.3.2/go.mod h1:pRBVtBSKl77K30Bv8R2P+cLSGaTtex6fsA2Wjqmfxj4=
+go.mongodb.org/mongo-driver/v2 v2.8.0 h1:CxWDGQYY8QQwNjAl/aq2sfWakdnWZynnqJ9F4DhHbP8=
+go.mongodb.org/mongo-driver/v2 v2.8.0/go.mod h1:yOI9kBsufol30iFsl1slpdq1I0eHPzybRWdyYUs8K/0=
+go.uber.org/goleak v1.3.0 h1:2K3zAYmnTNqV73imy9J1T3WC+gmCePx2hEGkimedGto=
+go.uber.org/goleak v1.3.0/go.mod h1:CoHD4mav9JJNrW/WLlf7HGZPjdw8EucARQHekz1X6bE=
+go.uber.org/mock v0.6.0 h1:hyF9dfmbgIX5EfOdasqLsWD6xqpNZlXblLB/Dbnwv3Y=
+go.uber.org/mock v0.6.0/go.mod h1:KiVJ4BqZJaMj4svdfmHM0AUx4NJYO8ZNpPnZn1Z+BBU=
+go.uber.org/multierr v1.11.0 h1:blXXJkSxSSfBVBlC76pxqeO+LN3aDfLQo+309xJstO0=
+go.uber.org/multierr v1.11.0/go.mod h1:20+QtiLqy0Nd6FdQB9TLXag12DsQkrbs3htMFfDN80Y=
+go.uber.org/zap v1.27.0 h1:aJMhYGrd5QSmlpLMr2MftRKl7t8J8PTZPA732ud/XR8=
+go.uber.org/zap v1.27.0/go.mod h1:GB2qFLM7cTU87MWRP2mPIjqfIDnGu+VIO4V/SdhGo2E=
+golang.org/x/arch v0.30.0 h1:sB9h+1gRGa2+LauFSV0tm8bK1J2yo1bx6/Uyi/P6DTU=
+golang.org/x/arch v0.30.0/go.mod h1:0X+GdSIP+kL5wPmpK7sdkEVTt2XoYP0cSjQSbZBwOi8=
+golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
+golang.org/x/crypto v0.54.0 h1:YLIA59K4fiNzHzjnZt2tUJQjQtUWfWbeHBqKtk3eScw=
+golang.org/x/crypto v0.54.0/go.mod h1:KWL8ny2AZdGR2cWmzeHrp2azQPGogOv+HeQaVEXC2dk=
+golang.org/x/net v0.0.0-20190311183353-d8887717615a/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
+golang.org/x/net v0.57.0 h1:K5+3DljvIuDG9/Jv9rvyMywYNFCQ9RSUY6OOTTkT+tE=
+golang.org/x/net v0.57.0/go.mod h1:KpXc8iv+r3XplLAG/f7Jsf9RPszJzdR0f58q9vGOuEU=
+golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
+golang.org/x/sync v0.22.0 h1:SZjpbeLmrCk4xhRSZFNZW5gFUeCeFgjekvI/+gfScek=
+golang.org/x/sync v0.22.0/go.mod h1:9xrNwdLfx4jkKbNva9FpL6vEN7evnE43NNNJQ2LF3+0=
+golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
+golang.org/x/sys v0.47.0 h1:o7XGOvZQCADBQQ4Y7VNq2dRWQR7JmOUW8Kxx4ZsNgWs=
+golang.org/x/sys v0.47.0/go.mod h1:4GL1E5IUh+htKOUEOaiffhrAeqysfVGipDYzABqnCmw=
+golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
+golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
+golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
+golang.org/x/tools v0.0.0-20190425150028-36563e24a262/go.mod h1:RgjU9mgBXZiqYHBnxXauZ1Gv1EHHAz9KjViQ78xBX0Q=
+golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
+google.golang.org/protobuf v1.26.0-rc.1/go.mod h1:jlhhOSvTdKEhbULTjvd4ARK9grFBp09yW+WbY/TyQbw=
+google.golang.org/protobuf v1.27.1/go.mod h1:9q0QmTI4eRPtz6boOQmLYwt+qCgq0jsYwAQnmE0givc=
+google.golang.org/protobuf v1.36.11 h1:fV6ZwhNocDyBLK0dj+fg8ektcVegBBuEolpbTQyBNVE=
+google.golang.org/protobuf v1.36.11/go.mod h1:HTf+CrKn2C3g5S8VImy6tdcUvCska2kB7j23XfzDpco=
+gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c h1:Hei/4ADfdWqJk1ZMxUNpqntNwaWcugrBjAiHlqqRiVk=
+gopkg.in/check.v1 v1.0.0-20201130134442-10cb98267c6c/go.mod h1:JHkPIbrfpd72SG/EVd6muEfDQjcINNoR0C8j2r3qZ4Q=
+gopkg.in/yaml.v2 v2.4.0 h1:D8xgwECY7CYvx+Y2n4sBz93Jn9JRvxdiyyo8CTfuKaY=
+gopkg.in/yaml.v2 v2.4.0/go.mod h1:RDklbk79AGWmwhnvt/jBztapEOGDOx6ZbXqjP6csGnQ=
+gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
+gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
+gorm.io/driver/postgres v1.6.2 h1:BvXQ/cNUg63q5TFNg672DmDcowZSFrNLkkA3Xe6GXq4=
+gorm.io/driver/postgres v1.6.2/go.mod h1:0c4fQA44XhOklXDkgtuKqysHCycTa5i9e3EIpDGCwXk=
+gorm.io/driver/sqlite v1.6.0 h1:WHRRrIiulaPiPFmDcod6prc4l2VGVWHz80KspNsxSfQ=
+gorm.io/driver/sqlite v1.6.0/go.mod h1:AO9V1qIQddBESngQUKWL9yoH93HIeA1X6V633rBwyT8=
+gorm.io/gorm v1.31.2 h1:3o8FXNo9v9S858gil+3LlZA1LkCOzgb4g5BL64FgaCo=
+gorm.io/gorm v1.31.2/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs=

Sense/tests/integration/brain_control/connector_test.go

@@ -1,138 +0,0 @@
-package brain_control_test
-
-import (
-	"crypto/ed25519"
-	"crypto/rand"
-	"encoding/json"
-	"strings"
-	"sync"
-	"testing"
-	"time"
-
-	bc "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/brain_control"
-	mi "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
-	"gorm.io/driver/sqlite"
-	"gorm.io/gorm"
-)
-
-func testDB(t *testing.T, name string) *gorm.DB {
-	t.Helper()
-	db, err := gorm.Open(sqlite.Open("file:"+name+"?mode=memory&cache=shared"), &gorm.Config{})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err = db.AutoMigrate(&bc.ReplayToken{}, &bc.RuntimeProjection{}, &bc.SourceRevision{}); err != nil {
-		t.Fatal(err)
-	}
-	return db
-}
-
-func TestMapperProducesCredentialFreeFrozenContract(t *testing.T) {
-	now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
-	c, err := bc.MapSourceConfig(bc.SourceFacts{ConfigID: "gate-primary", SiteID: "site-east", LogicalDeviceID: "camera-1", MediaPath: "site-east/camera-1/main", Revision: 1, PublishedAt: now, EffectiveAt: now, Profile: bc.Profile{ID: "main", Width: 1920, Height: 1080, Encoding: "h264", FrameRate: 25}, RuleSetVersion: "rules-1", Areas: []bc.AreaRule{{ID: "danger", Version: 1, Kind: "danger_area", Enabled: true, Points: []bc.Point{{X: .1, Y: .1}, {X: .8, Y: .1}, {X: .5, Y: .8}}}}})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if err = bc.ValidateSourceConfig(c); err != nil {
-		t.Fatal(err)
-	}
-	raw, _ := json.Marshal(c)
-	text := strings.ToLower(string(raw))
-	for _, secret := range []string{"password", "username", "rtsp://", "stream_uri", "credential"} {
-		if strings.Contains(text, secret) {
-			t.Fatalf("leaked %q", secret)
-		}
-	}
-	c2, err := bc.MapSourceConfig(bc.SourceFacts{ConfigID: "gate-primary", SiteID: "site-east", LogicalDeviceID: "camera-1", MediaPath: "site-east/camera-1/main", Revision: 2, PublishedAt: now, EffectiveAt: now, Profile: bc.Profile{ID: "main-v2", Width: 1280, Height: 720, Encoding: "H265", FrameRate: 20}, RuleSetVersion: "rules-2", NeedsRecalibration: true, Areas: []bc.AreaRule{{ID: "danger", Version: 2, Kind: "danger_area", Enabled: true, Points: []bc.Point{{X: .1, Y: .1}, {X: .8, Y: .1}, {X: .5, Y: .8}}}}})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if c2.RuleSet.State != "recalibration_required" || c2.RuleSet.Areas[0].Enabled {
-		t.Fatal("recalibration must disable rules")
-	}
-}
-
-func TestDurableReplayIsAtomicAndSurvivesVerifierRestart(t *testing.T) {
-	db := testDB(t, "sense-replay")
-	pub, priv, _ := ed25519.GenerateKey(rand.Reader)
-	now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
-	registry, _ := mi.NewRegistry(mi.KeyRecord{Principal: "yv:brain:east", KeyID: "brain-key-01", PublicKey: pub, Audience: "yovision-sense", Scopes: []string{"runtime-status:write"}, Enabled: true})
-	signer := mi.Signer{Principal: "yv:brain:east", KeyID: "brain-key-01", PrivateKey: priv, Now: func() time.Time { return now }}
-	body := []byte(`{"ok":true}`)
-	token, _ := signer.Mint("yovision-sense", []string{"runtime-status:write"}, "POST", "/machine/v1/runtime-status", body)
-	results := make(chan bool, 8)
-	var wg sync.WaitGroup
-	for i := 0; i < 8; i++ {
-		wg.Add(1)
-		go func() {
-			defer wg.Done()
-			v := mi.Verifier{Registry: registry, Replay: bc.GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
-			_, err := v.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body)
-			results <- err == nil
-		}()
-	}
-	wg.Wait()
-	close(results)
-	accepted := 0
-	for ok := range results {
-		if ok {
-			accepted++
-		}
-	}
-	if accepted != 1 {
-		t.Fatalf("accepted=%d", accepted)
-	}
-	v2 := mi.Verifier{Registry: registry, Replay: bc.GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
-	if _, err := v2.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body); err == nil {
-		t.Fatal("replay accepted after verifier restart")
-	}
-}
-
-func TestRevisionStoreConcurrentAndRestart(t *testing.T) {
-	db := testDB(t, "sense-revisions")
-	store := bc.RevisionStore{DB: db}
-	for want := int64(1); want <= 3; want++ {
-		got, err := store.Next("gate-primary")
-		if err != nil || got != want {
-			t.Fatalf("got %d err %v", got, err)
-		}
-	}
-	restarted := bc.RevisionStore{DB: db}
-	got, err := restarted.Next("gate-primary")
-	if err != nil || got != 4 {
-		t.Fatalf("restart got %d err %v", got, err)
-	}
-}
-
-func TestRuntimeProjectionStaleRecoveryMismatchAndOrdering(t *testing.T) {
-	db := testDB(t, "sense-projection")
-	now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
-	store := bc.ProjectionStore{DB: db, Clock: func() time.Time { return now }, StaleAfter: 90 * time.Second}
-	raw := statusJSON("018f4d6a-8d1b-4a25-8b37-9085f9c0d101", 41, now, "running", 20)
-	view, err := store.Ingest(raw, map[string]int64{"gate-primary": 21})
-	if err != nil {
-		t.Fatal(err)
-	}
-	if !view.RevisionMismatch || view.Stale {
-		t.Fatalf("bad initial view %+v", view)
-	}
-	now = now.Add(91 * time.Second)
-	view, err = store.View("brain-east-01")
-	if err != nil || !view.Offline || !view.Stale {
-		t.Fatalf("offline %+v %v", view, err)
-	}
-	raw = statusJSON("018f4d6a-8d1b-4a25-8b37-9085f9c0d102", 42, now, "running", 21)
-	view, err = store.Ingest(raw, map[string]int64{"gate-primary": 21})
-	if err != nil || !view.Recovered || view.RevisionMismatch {
-		t.Fatalf("recovery %+v %v", view, err)
-	}
-	if _, err = store.Ingest(statusJSON("018f4d6a-8d1b-4a25-8b37-9085f9c0d103", 41, now, "running", 21), nil); err == nil || err.Error() != "OUT_OF_ORDER_STATUS" {
-		t.Fatalf("expected ordering rejection: %v", err)
-	}
-}
-
-func statusJSON(id string, seq int64, observed time.Time, state string, revision int64) []byte {
-	v := map[string]any{"schema_version": bc.RuntimeStatusVersion, "status_id": id, "brain_instance_ref": "brain-east-01", "sequence": seq, "observed_at": observed.Format(time.RFC3339), "runtime": map[string]any{"state": state, "version": "1.0.0", "started_at": observed.Add(-time.Minute).Format(time.RFC3339)}, "model": map[string]any{"model_ref": "people-detection", "version": "2026.08.1"}, "configurations": []any{map[string]any{"config_id": "gate-primary", "apply_state": "applied", "applied_revision": revision, "error_code": nil}}, "health": map[string]any{"overall": "healthy", "error_codes": []any{}, "metrics": map[string]any{"load_percent": 1.0, "queue_depth": 0, "latency_ms": 2.0}}, "inputs": []any{}}
-	raw, _ := json.Marshal(v)
-	return raw
-}

Sense/tests/integration/brain_control/go.mod

@@ -1,18 +0,0 @@
-module git.ilapage.cn/ila/yovision/Sense/tests/integration/brain_control
-
-go 1.26.5
-
-require (
-	git.ilapage.cn/ila/yovision/Sense/server v0.0.0
-	gorm.io/driver/sqlite v1.6.0
-	gorm.io/gorm v1.31.2
-)
-
-require (
-	github.com/jinzhu/inflection v1.0.0 // indirect
-	github.com/jinzhu/now v1.1.5 // indirect
-	github.com/mattn/go-sqlite3 v1.14.49 // indirect
-	golang.org/x/text v0.40.0 // indirect
-)
-
-replace git.ilapage.cn/ila/yovision/Sense/server => ../../../server

Sense/tests/integration/brain_control/go.sum

@@ -1,12 +0,0 @@
-github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
-github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
-github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
-github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
-github.com/mattn/go-sqlite3 v1.14.49 h1:B8jBHC3xhxZgxztrgruTuLucebnULQnx4W7cF7SAE9w=
-github.com/mattn/go-sqlite3 v1.14.49/go.mod h1:6JTjA44L93a0QCyJef5YvlPoKXntQPjzWv5gtm9sB6w=
-golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
-golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
-gorm.io/driver/sqlite v1.6.0 h1:WHRRrIiulaPiPFmDcod6prc4l2VGVWHz80KspNsxSfQ=
-gorm.io/driver/sqlite v1.6.0/go.mod h1:AO9V1qIQddBESngQUKWL9yoH93HIeA1X6V633rBwyT8=
-gorm.io/gorm v1.31.2 h1:3o8FXNo9v9S858gil+3LlZA1LkCOzgb4g5BL64FgaCo=
-gorm.io/gorm v1.31.2/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs=