Merge pull request 'T-020: 回填 Issue #71 映射' (#72) from docs/T-020-issue-map into main

Bell/README.md

@@ -1,15 +1,17 @@
-# Bell 事件存储与内部审计入口
+# Bell 事件存储、审计与 Brain 事件入口
 
-Bell 当前实现 M3 的事件域基础及 Sense 审计 relay 的最小内部 HTTP 服务：
+Bell 当前实现 M3 的事件域基础、Sense 审计 relay 与默认关闭的 Brain 事件 ingress：
 
 - Bell 在可信 ingress 内为不含 `id` 的候选事实生成 `evt_` ULID。
 - 最终事件同时通过冻结 v0.1 JSON Schema 与六项代码级断言。
 - PostgreSQL `bell.events` 保存不可变事实；后续 outcome 追加到 `bell.event_outcomes`。
-- `bell_runtime` 对三张不可变事实表只有 `SELECT/INSERT`，没有 `UPDATE/DELETE/TRUNCATE` 或 migration owner 权限；仅可在短期 `audit_relay_receipts` 表查询、插入和清理过期收据。
+- `bell_runtime` 对事件、outcome、全局审计和 Brain 来源收据只有 `SELECT/INSERT`，没有 `UPDATE/DELETE/TRUNCATE` 或 migration owner 权限；只可清理两张短期 nonce 收据表。
 - `cmd/bell-api` 默认只监听 `127.0.0.1:8081`，接收 HMAC 签名的 `/internal/v1/audit-events:batch`，把脱敏设备操作事实追加到 `bell.audit_events`。
 - `(key_id, nonce)` 收据保存 10 分钟；相同摘要重放原结果，不同摘要返回冲突。非回环监听必须配置 TLS 证书和私钥。
+- T-019 可选 `/internal/v1/event-candidates` 把 HMAC key 绑定到一个 `producer_id`，通过 `event_ingress_bindings` 解析数字事件身份并复查当前 Site/Area/Sense Device；缺失、删除、Area 不一致或 `non_imaging_only` 均失败关闭。
+- `(producer_id, source_event_id)` 永久收据、最终事件和成功 nonce 响应同事务提交；相同 canonical candidate 返回原 Bell ID，不同 candidate 返回 `source_event_conflict`。
 
-Brain→Bell transport、公共认证/事件 API、规则、Alert 和证据对象存储仍需后续任务冻结。审计 relay 只服务 Sense，不得把 `internal/event` 的 Go 类型或该 HMAC 适配器当成公共协议。
+公共认证/事件 API、规则、Alert 和证据对象存储仍需后续任务冻结。两条 HMAC ingress 都是内部适配器，不得当成 Bell 公共协议或共用 key。
 
 启动内部 receiver 前必须私下设置 `BELL_DB_DSN` 和仓库外绝对路径 `BELL_AUDIT_KEYS_FILE`。远端监听还必须设置 `BELL_TLS_CERT_FILE`、`BELL_TLS_KEY_FILE`；仓库不保存 DSN、key 或证书：
 
@@ -17,6 +19,33 @@ Brain→Bell transport、公共认证/事件 API、规则、Alert 和证据对
 go -C Bell run ./cmd/bell-api
 ```
 
+事件 ingress 默认关闭。启用前，管理员先在专用数据库执行 `001`～`017` migration，并用受控 SQL 创建与现有 Bell Site/Area、Sense Device 一致的绑定；运行角色不能写绑定。然后私下设置：
+
+```powershell
+$env:BELL_EVENT_INGRESS_ENABLED = 'true'
+$env:BELL_EVENT_INGRESS_KEYS_FILE = 'D:\private\brain-event-keys.json'
+$env:BELL_EVIDENCE_FORBIDDEN_NAMES_FILE = 'D:\private\forbidden-evidence-names.txt'
+go -C Bell run ./cmd/bell-api
+```
+
+forbidden-names 文件每行一个不得出现在证据 URI 的租户/客户标记，至少一行、最多 256 行。key 文件格式见 `docs/contracts/README.md`。仓库不提供真实 secret、DSN、绑定或客户名称；绑定只能引用已经存在且同 Area/modality 的设备。
+
+管理员在事务中核对逻辑资源后，可按下列列签名配置一条视频绑定；数字 ID 是冻结事件契约使用的稳定正整数，不是把文本 ID 强转为数字：
+
+```sql
+INSERT INTO bell.event_ingress_bindings(
+  producer_id, tenant_id, site_id, device_id,
+  logical_tenant_id, logical_site_id, logical_device_id, logical_area_id,
+  modality, enabled
+) VALUES (
+  'brain-main', 1, 1, 1,
+  'tenant-logical-id', 'site-logical-id', 'device-logical-id', 'area-logical-id',
+  'video', true
+);
+```
+
+外键只负责资源存在；runtime 还会复查 Area 属于同 Site、设备当前 Area/modality 一致、Site/Area 未删除且策略允许成像。换绑或停用由管理员显式更新/删除 binding，不能修改永久来源收据。
+
 ## 验证
 
 ```powershell

Bell/cmd/bell-api/main.go

@@ -11,24 +11,31 @@ import (
 	"os"
 	"os/signal"
 	"path/filepath"
+	"strings"
 	"syscall"
 	"time"
 
 	"github.com/jackc/pgx/v5"
 	"github.com/jackc/pgx/v5/stdlib"
 
+	"yovision/bell/contracts"
 	"yovision/bell/internal/audit"
+	"yovision/bell/internal/event"
+	"yovision/bell/internal/ingress"
 	"yovision/bell/internal/store"
 )
 
 var version = "dev"
 
 type configuration struct {
-	address string
-	dsn     string
-	keyFile string
-	tlsCert string
-	tlsKey  string
+	address             string
+	dsn                 string
+	keyFile             string
+	tlsCert             string
+	tlsKey              string
+	eventIngressEnabled bool
+	eventKeyFile        string
+	forbiddenNamesFile  string
 }
 
 func main() {
@@ -41,11 +48,20 @@ func main() {
 
 func loadConfiguration() (configuration, error) {
 	value := configuration{
-		address: envOr("BELL_HTTP_ADDR", "127.0.0.1:8081"),
-		dsn:     os.Getenv("BELL_DB_DSN"),
-		keyFile: os.Getenv("BELL_AUDIT_KEYS_FILE"),
-		tlsCert: os.Getenv("BELL_TLS_CERT_FILE"),
-		tlsKey:  os.Getenv("BELL_TLS_KEY_FILE"),
+		address:            envOr("BELL_HTTP_ADDR", "127.0.0.1:8081"),
+		dsn:                os.Getenv("BELL_DB_DSN"),
+		keyFile:            os.Getenv("BELL_AUDIT_KEYS_FILE"),
+		tlsCert:            os.Getenv("BELL_TLS_CERT_FILE"),
+		tlsKey:             os.Getenv("BELL_TLS_KEY_FILE"),
+		eventKeyFile:       os.Getenv("BELL_EVENT_INGRESS_KEYS_FILE"),
+		forbiddenNamesFile: os.Getenv("BELL_EVIDENCE_FORBIDDEN_NAMES_FILE"),
+	}
+	switch os.Getenv("BELL_EVENT_INGRESS_ENABLED") {
+	case "", "false":
+	case "true":
+		value.eventIngressEnabled = true
+	default:
+		return configuration{}, errors.New("BELL_EVENT_INGRESS_ENABLED must be true or false")
 	}
 	if value.dsn == "" {
 		return configuration{}, errors.New("BELL_DB_DSN is required")
@@ -65,6 +81,14 @@ func loadConfiguration() (configuration, error) {
 	if (value.tlsCert == "") != (value.tlsKey == "") {
 		return configuration{}, errors.New("Bell TLS certificate and key must be configured together")
 	}
+	if value.eventIngressEnabled {
+		if value.eventKeyFile == "" || !filepath.IsAbs(value.eventKeyFile) {
+			return configuration{}, errors.New("BELL_EVENT_INGRESS_KEYS_FILE must be an absolute external path when event ingress is enabled")
+		}
+		if value.forbiddenNamesFile == "" || !filepath.IsAbs(value.forbiddenNamesFile) {
+			return configuration{}, errors.New("BELL_EVIDENCE_FORBIDDEN_NAMES_FILE must be an absolute external path when event ingress is enabled")
+		}
+	}
 	return value, nil
 }
 
@@ -106,6 +130,32 @@ func run(logger *slog.Logger) error {
 	}
 	mux := http.NewServeMux()
 	mux.Handle(audit.RelayPath, handler)
+	if cfg.eventIngressEnabled {
+		if err := repository.EventIngressReady(ctx); err != nil {
+			return err
+		}
+		eventKeys, err := ingress.LoadKeys(cfg.eventKeyFile)
+		if err != nil {
+			return err
+		}
+		forbiddenNames, err := loadForbiddenNames(cfg.forbiddenNamesFile)
+		if err != nil {
+			return err
+		}
+		guard, err := event.NewEvidenceGuard(forbiddenNames...)
+		if err != nil {
+			return err
+		}
+		factory, err := event.NewFactory(contracts.EventV01Schema, event.ULIDGenerator{}, repository, guard)
+		if err != nil {
+			return err
+		}
+		eventHandler, err := ingress.NewHandler(repository, eventKeys, factory)
+		if err != nil {
+			return err
+		}
+		mux.Handle(ingress.Path, eventHandler)
+	}
 	mux.HandleFunc("GET /healthz", func(writer http.ResponseWriter, _ *http.Request) {
 		writeStatus(writer, http.StatusOK, "ok")
 	})
@@ -114,6 +164,12 @@ func run(logger *slog.Logger) error {
 			writeStatus(writer, http.StatusServiceUnavailable, "not_ready")
 			return
 		}
+		if cfg.eventIngressEnabled {
+			if err := repository.EventIngressReady(request.Context()); err != nil {
+				writeStatus(writer, http.StatusServiceUnavailable, "not_ready")
+				return
+			}
+		}
 		writeStatus(writer, http.StatusOK, "ready")
 	})
 	server := &http.Server{Addr: cfg.address, Handler: mux, ReadHeaderTimeout: 5 * time.Second, ReadTimeout: 15 * time.Second, WriteTimeout: 15 * time.Second, IdleTimeout: 60 * time.Second, TLSConfig: &tls.Config{MinVersion: tls.VersionTLS12}}
@@ -138,6 +194,25 @@ func run(logger *slog.Logger) error {
 	return server.Shutdown(shutdownContext)
 }
 
+func loadForbiddenNames(path string) ([]string, error) {
+	raw, err := os.ReadFile(path)
+	if err != nil || len(raw) > 64<<10 {
+		return nil, errors.New("read Bell evidence forbidden-names file")
+	}
+	var values []string
+	for _, line := range strings.Split(string(raw), "\n") {
+		line = strings.TrimSpace(line)
+		if line == "" {
+			continue
+		}
+		values = append(values, line)
+	}
+	if len(values) == 0 || len(values) > 256 {
+		return nil, errors.New("Bell evidence forbidden-names file must contain 1 to 256 names")
+	}
+	return values, nil
+}
+
 func writeStatus(writer http.ResponseWriter, status int, value string) {
 	writer.Header().Set("Content-Type", "application/json")
 	writer.WriteHeader(status)

Bell/cmd/bell-api/main_test.go

@@ -31,3 +31,23 @@ func TestConfigurationRequiresTLSOutsideLoopback(t *testing.T) {
 		t.Fatalf("remote TLS Bell bind rejected: %v", err)
 	}
 }
+
+func TestEventIngressIsDisabledByDefaultAndRequiresExternalPolicy(t *testing.T) {
+	t.Setenv("BELL_DB_DSN", "postgres://bell@127.0.0.1/yovision")
+	t.Setenv("BELL_AUDIT_KEYS_FILE", filepath.Join(t.TempDir(), "audit.json"))
+	t.Setenv("BELL_EVENT_INGRESS_ENABLED", "")
+	value, err := loadConfiguration()
+	if err != nil || value.eventIngressEnabled {
+		t.Fatalf("default event ingress configuration: %+v %v", value, err)
+	}
+	t.Setenv("BELL_EVENT_INGRESS_ENABLED", "true")
+	if _, err := loadConfiguration(); err == nil {
+		t.Fatal("event ingress without keys and evidence policy was accepted")
+	}
+	t.Setenv("BELL_EVENT_INGRESS_KEYS_FILE", filepath.Join(t.TempDir(), "event-keys.json"))
+	t.Setenv("BELL_EVIDENCE_FORBIDDEN_NAMES_FILE", filepath.Join(t.TempDir(), "names.txt"))
+	value, err = loadConfiguration()
+	if err != nil || !value.eventIngressEnabled {
+		t.Fatalf("valid event ingress configuration rejected: %+v %v", value, err)
+	}
+}

Bell/internal/event/event.go

@@ -163,6 +163,7 @@ func (e Event) TenantID() int64           { return e.shape.TenantID }
 func (e Event) SiteID() int64             { return e.shape.SiteID }
 func (e Event) DeviceID() int64           { return e.shape.DeviceID }
 func (e Event) SourceEventID() string     { return e.shape.SourceEventID }
+func (e Event) Sensors() []Sensor         { return append([]Sensor(nil), e.shape.Sensors...) }
 func (e Event) Kind() string              { return e.shape.Kind }
 func (e Event) Severity() string          { return e.shape.Severity }
 func (e Event) OccurredAt() time.Time     { return e.shape.OccurredAt }

Bell/internal/ingress/ingress.go

@@ -0,0 +1,270 @@
+// Package ingress authenticates Brain event candidates and delegates their
+// atomic persistence to Bell's repository.
+package ingress
+
+import (
+	"bytes"
+	"context"
+	"crypto/hmac"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/hex"
+	"encoding/json"
+	"errors"
+	"io"
+	"net/http"
+	"regexp"
+	"strconv"
+	"strings"
+	"time"
+
+	"yovision/bell/internal/event"
+)
+
+const (
+	Path            = "/internal/v1/event-candidates"
+	MaxBodyBytes    = 1 << 20
+	HeaderKeyID     = "X-YoVision-Key-Id"
+	HeaderTimestamp = "X-YoVision-Timestamp"
+	HeaderNonce     = "X-YoVision-Nonce"
+	HeaderSignature = "X-YoVision-Signature"
+)
+
+var (
+	ErrReplayConflict = errors.New("event ingress replay conflict")
+	ErrSourceConflict = errors.New("event ingress source event conflict")
+	ErrIdentityDenied = errors.New("event ingress identity denied")
+	keyIDPattern      = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$`)
+	producerIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$`)
+)
+
+type Result struct {
+	SchemaVersion int    `json:"schema_version"`
+	ProducerID    string `json:"producer_id"`
+	SourceEventID string `json:"source_event_id"`
+	EventID       string `json:"event_id"`
+	Status        string `json:"status"`
+	HTTPStatus    int    `json:"-"`
+}
+
+type Repository interface {
+	Replay(
+		context.Context,
+		string,
+		string,
+		[sha256.Size]byte,
+		string,
+		string,
+		[sha256.Size]byte,
+	) (Result, bool, error)
+	ProcessEvent(
+		context.Context,
+		string,
+		string,
+		[sha256.Size]byte,
+		string,
+		[sha256.Size]byte,
+		event.Event,
+	) (Result, error)
+}
+
+type Handler struct {
+	repository Repository
+	keys       map[string]Key
+	factory    *event.Factory
+	now        func() time.Time
+}
+
+func NewHandler(repository Repository, keys map[string]Key, factory *event.Factory) (*Handler, error) {
+	if repository == nil || factory == nil || len(keys) == 0 {
+		return nil, errors.New("event ingress handler dependencies are required")
+	}
+	copyKeys := make(map[string]Key, len(keys))
+	for id, key := range keys {
+		if !keyIDPattern.MatchString(id) || !producerIDPattern.MatchString(key.ProducerID) || len(key.Secret) < 32 {
+			return nil, errors.New("invalid event ingress handler key")
+		}
+		copyKeys[id] = Key{ProducerID: key.ProducerID, Secret: append([]byte(nil), key.Secret...)}
+	}
+	return &Handler{repository: repository, keys: copyKeys, factory: factory, now: time.Now}, nil
+}
+
+type envelope struct {
+	SchemaVersion int             `json:"schema_version"`
+	ProducerID    string          `json:"producer_id"`
+	Candidate     json.RawMessage `json:"candidate"`
+}
+
+func (h *Handler) ServeHTTP(writer http.ResponseWriter, request *http.Request) {
+	if request.Method != http.MethodPost || request.URL.Path != Path {
+		writeError(writer, http.StatusNotFound, "not_found", "")
+		return
+	}
+	body, err := io.ReadAll(io.LimitReader(request.Body, MaxBodyBytes+1))
+	if err != nil || len(body) > MaxBodyBytes {
+		writeError(writer, http.StatusRequestEntityTooLarge, "payload_too_large", "")
+		return
+	}
+	keyID := request.Header.Get(HeaderKeyID)
+	timestamp := request.Header.Get(HeaderTimestamp)
+	nonce := request.Header.Get(HeaderNonce)
+	provided := request.Header.Get(HeaderSignature)
+	key, ok := h.keys[keyID]
+	seconds, timestampErr := strconv.ParseInt(timestamp, 10, 64)
+	nonceBytes, nonceErr := base64.RawURLEncoding.DecodeString(nonce)
+	signatureBytes, signatureErr := base64.RawURLEncoding.DecodeString(provided)
+	if !ok || timestampErr != nil || len(timestamp) < 10 || nonceErr != nil || len(nonceBytes) < 16 || len(nonceBytes) > 48 ||
+		signatureErr != nil || len(signatureBytes) != sha256.Size || absDuration(h.now().UTC().Sub(time.Unix(seconds, 0).UTC())) > 300*time.Second {
+		writeError(writer, http.StatusUnauthorized, "unauthorized", "")
+		return
+	}
+	expected := signature(key.Secret, canonicalString(request.Method, request.URL.EscapedPath(), timestamp, nonce, body))
+	if !hmac.Equal(signatureBytes, expected) {
+		writeError(writer, http.StatusUnauthorized, "unauthorized", "")
+		return
+	}
+
+	var value envelope
+	decoder := json.NewDecoder(bytes.NewReader(body))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&value); err != nil || value.SchemaVersion != 1 || !producerIDPattern.MatchString(value.ProducerID) || len(value.Candidate) == 0 {
+		writeError(writer, http.StatusBadRequest, "invalid_envelope", "")
+		return
+	}
+	var trailing any
+	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
+		writeError(writer, http.StatusBadRequest, "invalid_envelope", "")
+		return
+	}
+	if value.ProducerID != key.ProducerID {
+		writeError(writer, http.StatusUnauthorized, "unauthorized", "")
+		return
+	}
+	canonicalCandidate, err := canonicalJSON(value.Candidate)
+	if err != nil {
+		writeError(writer, http.StatusBadRequest, "invalid_envelope", "")
+		return
+	}
+	sourceEventID, err := candidateSourceEventID(canonicalCandidate)
+	if err != nil {
+		writeError(writer, http.StatusUnprocessableEntity, "candidate_invalid", "schema_invalid")
+		return
+	}
+	requestHash := sha256.Sum256(body)
+	candidateHash := sha256.Sum256(canonicalCandidate)
+	replay, found, err := h.repository.Replay(
+		request.Context(), keyID, nonce, requestHash, value.ProducerID, sourceEventID, candidateHash,
+	)
+	switch {
+	case errors.Is(err, ErrReplayConflict):
+		writeError(writer, http.StatusConflict, "replay_conflict", "")
+		return
+	case errors.Is(err, ErrSourceConflict):
+		writeError(writer, http.StatusConflict, "source_event_conflict", "")
+		return
+	case err != nil:
+		writeError(writer, http.StatusServiceUnavailable, "temporarily_unavailable", "")
+		return
+	case found:
+		writeJSON(writer, replay.HTTPStatus, replay)
+		return
+	}
+	created, err := h.factory.Create(request.Context(), canonicalCandidate)
+	if err != nil {
+		handleFactoryError(writer, err)
+		return
+	}
+	result, err := h.repository.ProcessEvent(request.Context(), keyID, nonce, requestHash, value.ProducerID, candidateHash, created)
+	switch {
+	case errors.Is(err, ErrReplayConflict):
+		writeError(writer, http.StatusConflict, "replay_conflict", "")
+	case errors.Is(err, ErrSourceConflict):
+		writeError(writer, http.StatusConflict, "source_event_conflict", "")
+	case errors.Is(err, ErrIdentityDenied):
+		writeError(writer, http.StatusForbidden, "identity_denied", "")
+	case err != nil:
+		writeError(writer, http.StatusServiceUnavailable, "temporarily_unavailable", "")
+	default:
+		writeJSON(writer, result.HTTPStatus, result)
+	}
+}
+
+func handleFactoryError(writer http.ResponseWriter, err error) {
+	var validation *event.ValidationError
+	if !errors.As(err, &validation) {
+		writeError(writer, http.StatusServiceUnavailable, "temporarily_unavailable", "")
+		return
+	}
+	switch validation.Code {
+	case event.CodePayloadTooLarge:
+		writeError(writer, http.StatusRequestEntityTooLarge, "payload_too_large", "")
+	case event.CodePrivacyDenied:
+		writeError(writer, http.StatusForbidden, "privacy_denied", "")
+	case event.CodePrivacyUnavailable:
+		writeError(writer, http.StatusServiceUnavailable, "privacy_unavailable", "")
+	default:
+		writeError(writer, http.StatusUnprocessableEntity, "candidate_invalid", string(validation.Code))
+	}
+}
+
+func canonicalJSON(raw []byte) ([]byte, error) {
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	decoder.UseNumber()
+	var value any
+	if err := decoder.Decode(&value); err != nil || value == nil {
+		return nil, errors.New("invalid JSON")
+	}
+	var trailing any
+	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
+		return nil, errors.New("multiple JSON values")
+	}
+	return json.Marshal(value)
+}
+
+var sourceEventIDPattern = regexp.MustCompile(`^[A-Za-z0-9_-]{1,128}$`)
+
+func candidateSourceEventID(raw []byte) (string, error) {
+	var object map[string]json.RawMessage
+	if err := json.Unmarshal(raw, &object); err != nil {
+		return "", errors.New("candidate is not an object")
+	}
+	var value string
+	if err := json.Unmarshal(object["source_event_id"], &value); err != nil || !sourceEventIDPattern.MatchString(value) {
+		return "", errors.New("candidate source event ID is invalid")
+	}
+	return value, nil
+}
+
+func canonicalString(method, path, timestamp, nonce string, body []byte) string {
+	digest := sha256.Sum256(body)
+	return strings.Join([]string{method, path, timestamp, nonce, hex.EncodeToString(digest[:])}, "\n")
+}
+
+func signature(secret []byte, canonical string) []byte {
+	mac := hmac.New(sha256.New, secret)
+	_, _ = mac.Write([]byte(canonical))
+	return mac.Sum(nil)
+}
+
+func absDuration(value time.Duration) time.Duration {
+	if value < 0 {
+		return -value
+	}
+	return value
+}
+
+func writeError(writer http.ResponseWriter, status int, code, detail string) {
+	value := map[string]string{"error": code}
+	if detail != "" {
+		value["detail_code"] = detail
+	}
+	writeJSON(writer, status, value)
+}
+
+func writeJSON(writer http.ResponseWriter, status int, value any) {
+	writer.Header().Set("Content-Type", "application/json")
+	writer.Header().Set("Cache-Control", "no-store")
+	writer.Header().Set("X-Content-Type-Options", "nosniff")
+	writer.WriteHeader(status)
+	_ = json.NewEncoder(writer).Encode(value)
+}

Bell/internal/ingress/ingress_test.go

@@ -0,0 +1,176 @@
+package ingress
+
+import (
+	"bytes"
+	"context"
+	"crypto/rand"
+	"crypto/sha256"
+	"encoding/base64"
+	"encoding/json"
+	"net/http"
+	"net/http/httptest"
+	"os"
+	"path/filepath"
+	"strconv"
+	"testing"
+	"time"
+
+	"yovision/bell/contracts"
+	"yovision/bell/internal/event"
+)
+
+type ingressIDs struct{}
+
+func (ingressIDs) NewEventID() (string, error) {
+	return "evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2B", nil
+}
+
+type ingressPrivacy struct{}
+
+func (ingressPrivacy) VideoAllowed(context.Context, int64, int64, int64) (bool, error) {
+	return true, nil
+}
+
+type fakeRepository struct {
+	replayResult Result
+	replayFound  bool
+	replayErr    error
+	processErr   error
+	processed    int
+}
+
+func (f *fakeRepository) Replay(context.Context, string, string, [sha256.Size]byte, string, string, [sha256.Size]byte) (Result, bool, error) {
+	return f.replayResult, f.replayFound, f.replayErr
+}
+
+func (f *fakeRepository) ProcessEvent(_ context.Context, _, _ string, _ [sha256.Size]byte, producer string, _ [sha256.Size]byte, value event.Event) (Result, error) {
+	f.processed++
+	if f.processErr != nil {
+		return Result{}, f.processErr
+	}
+	return Result{SchemaVersion: 1, ProducerID: producer, SourceEventID: value.SourceEventID(), EventID: value.ID(), Status: "accepted", HTTPStatus: 201}, nil
+}
+
+func ingressCandidate(t *testing.T) []byte {
+	t.Helper()
+	raw, err := os.ReadFile(filepath.Join("..", "..", "..", "docs", "raw", "contracts", "event-v0.1.example-current.json"))
+	if err != nil {
+		t.Fatal(err)
+	}
+	var object map[string]any
+	if err := json.Unmarshal(raw, &object); err != nil {
+		t.Fatal(err)
+	}
+	delete(object, "id")
+	object["kind"] = "zone_entry"
+	object["severity"] = "medium"
+	object["evidence"] = map[string]any{"snapshot_uris": []any{}, "clip_uri": nil, "clip_range": nil}
+	encoded, err := json.Marshal(object)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return encoded
+}
+
+func newIngressHandler(t *testing.T, repository Repository) (*Handler, []byte) {
+	t.Helper()
+	guard, err := event.NewEvidenceGuard("private-customer")
+	if err != nil {
+		t.Fatal(err)
+	}
+	factory, err := event.NewFactory(contracts.EventV01Schema, ingressIDs{}, ingressPrivacy{}, guard)
+	if err != nil {
+		t.Fatal(err)
+	}
+	secret := make([]byte, 32)
+	if _, err := rand.Read(secret); err != nil {
+		t.Fatal(err)
+	}
+	handler, err := NewHandler(repository, map[string]Key{"brain-a": {ProducerID: "brain-main", Secret: secret}}, factory)
+	if err != nil {
+		t.Fatal(err)
+	}
+	handler.now = func() time.Time { return time.Unix(1_800_000_000, 0).UTC() }
+	return handler, secret
+}
+
+func signedRequest(t *testing.T, secret []byte, producer string, candidate []byte) *http.Request {
+	t.Helper()
+	body, err := json.Marshal(map[string]any{"schema_version": 1, "producer_id": producer, "candidate": json.RawMessage(candidate)})
+	if err != nil {
+		t.Fatal(err)
+	}
+	request := httptest.NewRequest(http.MethodPost, Path, bytes.NewReader(body))
+	timestamp := strconv.FormatInt(1_800_000_000, 10)
+	nonce := base64.RawURLEncoding.EncodeToString([]byte("0123456789abcdef"))
+	request.Header.Set(HeaderKeyID, "brain-a")
+	request.Header.Set(HeaderTimestamp, timestamp)
+	request.Header.Set(HeaderNonce, nonce)
+	request.Header.Set(HeaderSignature, base64.RawURLEncoding.EncodeToString(signature(secret, canonicalString(http.MethodPost, Path, timestamp, nonce, body))))
+	return request
+}
+
+func TestHandlerAcceptsSignedCandidateAndRejectsProducerSpoofing(t *testing.T) {
+	repository := &fakeRepository{}
+	handler, secret := newIngressHandler(t, repository)
+	response := httptest.NewRecorder()
+	handler.ServeHTTP(response, signedRequest(t, secret, "brain-main", ingressCandidate(t)))
+	if response.Code != http.StatusCreated || repository.processed != 1 {
+		t.Fatalf("signed candidate: status=%d body=%s processed=%d", response.Code, response.Body.String(), repository.processed)
+	}
+	var result Result
+	if err := json.Unmarshal(response.Body.Bytes(), &result); err != nil || result.Status != "accepted" || result.EventID == "" {
+		t.Fatalf("invalid accepted response: %+v %v", result, err)
+	}
+
+	response = httptest.NewRecorder()
+	handler.ServeHTTP(response, signedRequest(t, secret, "brain-spoofed", ingressCandidate(t)))
+	if response.Code != http.StatusUnauthorized || repository.processed != 1 {
+		t.Fatalf("producer spoofing was not rejected: %d %s", response.Code, response.Body.String())
+	}
+}
+
+func TestHandlerReturnsDurableReplayAndStableConflicts(t *testing.T) {
+	repository := &fakeRepository{replayFound: true, replayResult: Result{
+		SchemaVersion: 1, ProducerID: "brain-main", SourceEventID: "source-1",
+		EventID: "evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2B", Status: "duplicate", HTTPStatus: 200,
+	}}
+	handler, secret := newIngressHandler(t, repository)
+	response := httptest.NewRecorder()
+	handler.ServeHTTP(response, signedRequest(t, secret, "brain-main", ingressCandidate(t)))
+	if response.Code != http.StatusOK || repository.processed != 0 {
+		t.Fatalf("durable replay did not bypass factory persistence: %d", response.Code)
+	}
+
+	repository.replayFound = false
+	repository.replayErr = ErrSourceConflict
+	response = httptest.NewRecorder()
+	handler.ServeHTTP(response, signedRequest(t, secret, "brain-main", ingressCandidate(t)))
+	if response.Code != http.StatusConflict {
+		t.Fatalf("source conflict status=%d body=%s", response.Code, response.Body.String())
+	}
+}
+
+func TestHandlerRejectsInvalidSignatureAndUpstreamPlatformID(t *testing.T) {
+	repository := &fakeRepository{}
+	handler, secret := newIngressHandler(t, repository)
+	request := signedRequest(t, secret, "brain-main", ingressCandidate(t))
+	request.Header.Set(HeaderSignature, base64.RawURLEncoding.EncodeToString(make([]byte, 32)))
+	response := httptest.NewRecorder()
+	handler.ServeHTTP(response, request)
+	if response.Code != http.StatusUnauthorized {
+		t.Fatalf("invalid signature status=%d", response.Code)
+	}
+
+	var object map[string]any
+	if err := json.Unmarshal(ingressCandidate(t), &object); err != nil {
+		t.Fatal(err)
+	}
+	object["id"] = "evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2B"
+	withID, _ := json.Marshal(object)
+	response = httptest.NewRecorder()
+	handler.ServeHTTP(response, signedRequest(t, secret, "brain-main", withID))
+	if response.Code != http.StatusUnprocessableEntity || repository.processed != 0 {
+		t.Fatalf("upstream ID status=%d body=%s", response.Code, response.Body.String())
+	}
+}

Bell/internal/ingress/key.go

@@ -0,0 +1,53 @@
+package ingress
+
+import (
+	"bytes"
+	"encoding/base64"
+	"encoding/json"
+	"errors"
+	"io"
+	"os"
+)
+
+type keyDocument struct {
+	Version int `json:"version"`
+	Keys    []struct {
+		KeyID      string `json:"key_id"`
+		ProducerID string `json:"producer_id"`
+		Secret     string `json:"secret_base64url"`
+	} `json:"keys"`
+}
+
+type Key struct {
+	ProducerID string
+	Secret     []byte
+}
+
+func LoadKeys(path string) (map[string]Key, error) {
+	raw, err := os.ReadFile(path)
+	if err != nil {
+		return nil, errors.New("read Bell event ingress key file")
+	}
+	var document keyDocument
+	decoder := json.NewDecoder(bytes.NewReader(raw))
+	decoder.DisallowUnknownFields()
+	if err := decoder.Decode(&document); err != nil || document.Version != 1 || len(document.Keys) == 0 {
+		return nil, errors.New("invalid Bell event ingress key file")
+	}
+	var trailing any
+	if err := decoder.Decode(&trailing); !errors.Is(err, io.EOF) {
+		return nil, errors.New("invalid Bell event ingress key file")
+	}
+	values := make(map[string]Key, len(document.Keys))
+	for _, item := range document.Keys {
+		secret, err := base64.RawURLEncoding.DecodeString(item.Secret)
+		if err != nil || !keyIDPattern.MatchString(item.KeyID) || !producerIDPattern.MatchString(item.ProducerID) || len(secret) < 32 {
+			return nil, errors.New("invalid Bell event ingress key")
+		}
+		if _, exists := values[item.KeyID]; exists {
+			return nil, errors.New("duplicate Bell event ingress key ID")
+		}
+		values[item.KeyID] = Key{ProducerID: item.ProducerID, Secret: append([]byte(nil), secret...)}
+	}
+	return values, nil
+}

Bell/internal/store/event_ingress_postgres.go

@@ -0,0 +1,289 @@
+package store
+
+import (
+	"bytes"
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+
+	"yovision/bell/internal/event"
+	"yovision/bell/internal/ingress"
+)
+
+func (p *Postgres) EventIngressReady(ctx context.Context) error {
+	var version int64
+	if err := p.db.QueryRowContext(ctx, `SELECT COALESCE(MAX(version), 0) FROM bell.schema_migrations`).Scan(&version); err != nil || version < 5 {
+		return errors.New("postgres Bell schema migration v5 is required for event ingress")
+	}
+	var bindingSelect bool
+	var receiptSelect, receiptInsert, receiptUpdate, receiptDelete, receiptTruncate bool
+	var nonceSelect, nonceInsert, nonceUpdate, nonceDelete, nonceTruncate bool
+	var deviceID, deviceArea, deviceModality bool
+	var deviceEndpoint, deviceCredential, deviceProfile, devicePath bool
+	var siteID, areaPolicy bool
+	err := p.db.QueryRowContext(ctx, `SELECT
+		has_table_privilege(current_user, 'bell.event_ingress_bindings', 'SELECT'),
+		has_table_privilege(current_user, 'bell.event_ingress_receipts', 'SELECT'),
+		has_table_privilege(current_user, 'bell.event_ingress_receipts', 'INSERT'),
+		has_table_privilege(current_user, 'bell.event_ingress_receipts', 'UPDATE'),
+		has_table_privilege(current_user, 'bell.event_ingress_receipts', 'DELETE'),
+		has_table_privilege(current_user, 'bell.event_ingress_receipts', 'TRUNCATE'),
+		has_table_privilege(current_user, 'bell.event_ingress_nonces', 'SELECT'),
+		has_table_privilege(current_user, 'bell.event_ingress_nonces', 'INSERT'),
+		has_table_privilege(current_user, 'bell.event_ingress_nonces', 'UPDATE'),
+		has_table_privilege(current_user, 'bell.event_ingress_nonces', 'DELETE'),
+		has_table_privilege(current_user, 'bell.event_ingress_nonces', 'TRUNCATE'),
+		has_column_privilege(current_user, 'sense.devices', 'id', 'SELECT'),
+		has_column_privilege(current_user, 'sense.devices', 'area_id', 'SELECT'),
+		has_column_privilege(current_user, 'sense.devices', 'modality', 'SELECT'),
+		has_column_privilege(current_user, 'sense.devices', 'endpoint_ref', 'SELECT'),
+		has_column_privilege(current_user, 'sense.devices', 'credential_ref', 'SELECT'),
+		has_column_privilege(current_user, 'sense.devices', 'profile_token', 'SELECT'),
+		has_column_privilege(current_user, 'sense.devices', 'path_name', 'SELECT'),
+		has_column_privilege(current_user, 'bell.sites', 'id', 'SELECT'),
+		has_column_privilege(current_user, 'bell.areas', 'capture_policy', 'SELECT')`).Scan(
+		&bindingSelect,
+		&receiptSelect, &receiptInsert, &receiptUpdate, &receiptDelete, &receiptTruncate,
+		&nonceSelect, &nonceInsert, &nonceUpdate, &nonceDelete, &nonceTruncate,
+		&deviceID, &deviceArea, &deviceModality,
+		&deviceEndpoint, &deviceCredential, &deviceProfile, &devicePath,
+		&siteID, &areaPolicy,
+	)
+	if err != nil {
+		return errors.New("verify Bell event ingress privileges")
+	}
+	if !bindingSelect || !receiptSelect || !receiptInsert || receiptUpdate || receiptDelete || receiptTruncate ||
+		!nonceSelect || !nonceInsert || nonceUpdate || !nonceDelete || nonceTruncate ||
+		!deviceID || !deviceArea || !deviceModality || deviceEndpoint || deviceCredential || deviceProfile || devicePath ||
+		!siteID || !areaPolicy {
+		return errors.New("Bell event ingress privileges violate append-only boundary")
+	}
+	return nil
+}
+
+// VideoAllowed implements event.PrivacyPolicy using current Bell/Sense facts.
+// Exact producer ownership is checked again in ProcessEvent.
+func (p *Postgres) VideoAllowed(ctx context.Context, tenantID, siteID, deviceID int64) (bool, error) {
+	return bindingAllowed(ctx, p.db, "", tenantID, siteID, deviceID, "video")
+}
+
+type rowQuerier interface {
+	QueryRowContext(context.Context, string, ...any) *sql.Row
+}
+
+func bindingAllowed(
+	ctx context.Context,
+	query rowQuerier,
+	producerID string,
+	tenantID, siteID, deviceID int64,
+	modality string,
+) (bool, error) {
+	producerClause := ""
+	arguments := []any{tenantID, siteID, deviceID, modality}
+	if producerID != "" {
+		producerClause = " AND binding.producer_id=$5"
+		arguments = append(arguments, producerID)
+	}
+	statement := `SELECT EXISTS (
+		SELECT 1
+		FROM bell.event_ingress_bindings AS binding
+		JOIN bell.sites AS site
+		  ON site.tenant_id=binding.logical_tenant_id
+		 AND site.id=binding.logical_site_id
+		JOIN bell.areas AS area
+		  ON area.tenant_id=binding.logical_tenant_id
+		 AND area.site_id=binding.logical_site_id
+		 AND area.id=binding.logical_area_id
+		JOIN sense.devices AS device
+		  ON device.tenant_id=binding.logical_tenant_id
+		 AND device.site_id=binding.logical_site_id
+		 AND device.id=binding.logical_device_id
+		 AND device.area_id=binding.logical_area_id
+		 AND device.modality=binding.modality
+		WHERE binding.tenant_id=$1 AND binding.site_id=$2 AND binding.device_id=$3
+		  AND binding.modality=$4 AND binding.enabled
+		  AND site.deleted_at IS NULL AND area.deleted_at IS NULL
+		  AND (binding.modality <> 'video' OR area.capture_policy='video_allowed')` + producerClause + `
+	)`
+	var allowed bool
+	if err := query.QueryRowContext(ctx, statement, arguments...).Scan(&allowed); err != nil {
+		return false, fmt.Errorf("resolve Bell event ingress binding: %w", err)
+	}
+	return allowed, nil
+}
+
+func (p *Postgres) Replay(
+	ctx context.Context,
+	keyID, nonce string,
+	requestHash [sha256.Size]byte,
+	producerID, sourceEventID string,
+	candidateHash [sha256.Size]byte,
+) (ingress.Result, bool, error) {
+	tx, err := p.db.BeginTx(ctx, nil)
+	if err != nil {
+		return ingress.Result{}, false, errors.New("begin Bell event replay check")
+	}
+	defer tx.Rollback()
+	if err := lockIngress(ctx, tx, keyID, nonce, producerID, sourceEventID); err != nil {
+		return ingress.Result{}, false, err
+	}
+	if _, err := tx.ExecContext(ctx, `DELETE FROM bell.event_ingress_nonces WHERE expires_at <= clock_timestamp()`); err != nil {
+		return ingress.Result{}, false, errors.New("expire Bell event ingress nonces")
+	}
+	if value, found, err := readNonce(ctx, tx, keyID, nonce, requestHash); err != nil || found {
+		if err == nil {
+			err = tx.Commit()
+		}
+		return value, found, err
+	}
+	var storedHash []byte
+	var eventID string
+	err = tx.QueryRowContext(ctx, `SELECT candidate_hash, event_id
+		FROM bell.event_ingress_receipts WHERE producer_id=$1 AND source_event_id=$2`, producerID, sourceEventID).Scan(&storedHash, &eventID)
+	if err == nil {
+		if !bytes.Equal(storedHash, candidateHash[:]) {
+			return ingress.Result{}, false, ingress.ErrSourceConflict
+		}
+		value := ingress.Result{SchemaVersion: 1, ProducerID: producerID, SourceEventID: sourceEventID, EventID: eventID, Status: "duplicate", HTTPStatus: 200}
+		if err := insertNonce(ctx, tx, keyID, nonce, requestHash, value); err != nil {
+			return ingress.Result{}, false, err
+		}
+		if err := tx.Commit(); err != nil {
+			return ingress.Result{}, false, errors.New("commit Bell source replay")
+		}
+		return value, true, nil
+	}
+	if !errors.Is(err, sql.ErrNoRows) {
+		return ingress.Result{}, false, errors.New("read Bell event source receipt")
+	}
+	if err := tx.Commit(); err != nil {
+		return ingress.Result{}, false, errors.New("commit Bell event replay miss")
+	}
+	return ingress.Result{}, false, nil
+}
+
+func (p *Postgres) ProcessEvent(
+	ctx context.Context,
+	keyID, nonce string,
+	requestHash [sha256.Size]byte,
+	producerID string,
+	candidateHash [sha256.Size]byte,
+	value event.Event,
+) (ingress.Result, error) {
+	tx, err := p.db.BeginTx(ctx, nil)
+	if err != nil {
+		return ingress.Result{}, errors.New("begin Bell event ingress")
+	}
+	defer tx.Rollback()
+	if err := lockIngress(ctx, tx, keyID, nonce, producerID, value.SourceEventID()); err != nil {
+		return ingress.Result{}, err
+	}
+	if _, err := tx.ExecContext(ctx, `DELETE FROM bell.event_ingress_nonces WHERE expires_at <= clock_timestamp()`); err != nil {
+		return ingress.Result{}, errors.New("expire Bell event ingress nonces")
+	}
+	if replay, found, err := readNonce(ctx, tx, keyID, nonce, requestHash); err != nil || found {
+		if err == nil {
+			err = tx.Commit()
+		}
+		return replay, err
+	}
+	var storedHash []byte
+	var storedEventID string
+	err = tx.QueryRowContext(ctx, `SELECT candidate_hash, event_id
+		FROM bell.event_ingress_receipts WHERE producer_id=$1 AND source_event_id=$2`, producerID, value.SourceEventID()).Scan(&storedHash, &storedEventID)
+	if err == nil {
+		if !bytes.Equal(storedHash, candidateHash[:]) {
+			return ingress.Result{}, ingress.ErrSourceConflict
+		}
+		result := ingress.Result{SchemaVersion: 1, ProducerID: producerID, SourceEventID: value.SourceEventID(), EventID: storedEventID, Status: "duplicate", HTTPStatus: 200}
+		if err := insertNonce(ctx, tx, keyID, nonce, requestHash, result); err != nil {
+			return ingress.Result{}, err
+		}
+		if err := tx.Commit(); err != nil {
+			return ingress.Result{}, errors.New("commit Bell source duplicate")
+		}
+		return result, nil
+	}
+	if !errors.Is(err, sql.ErrNoRows) {
+		return ingress.Result{}, errors.New("read Bell event source receipt")
+	}
+	for _, sensor := range value.Sensors() {
+		allowed, err := bindingAllowed(ctx, tx, producerID, value.TenantID(), value.SiteID(), sensor.DeviceID, sensor.Modality)
+		if err != nil {
+			return ingress.Result{}, err
+		}
+		if !allowed {
+			return ingress.Result{}, ingress.ErrIdentityDenied
+		}
+	}
+	digest := value.Digest()
+	if _, err := tx.ExecContext(ctx, `INSERT INTO bell.events(
+		id, tenant_id, site_id, device_id, source_event_id, kind, severity,
+		occurred_at, detected_at, payload_hash, payload
+	) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11::jsonb)`,
+		value.ID(), value.TenantID(), value.SiteID(), value.DeviceID(), value.SourceEventID(),
+		value.Kind(), value.Severity(), value.OccurredAt(), value.DetectedAt(), digest[:], value.JSON(),
+	); err != nil {
+		return ingress.Result{}, fmt.Errorf("insert Bell ingress event: %w", err)
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO bell.event_ingress_receipts(
+		producer_id, source_event_id, candidate_hash, event_id
+	) VALUES ($1,$2,$3,$4)`, producerID, value.SourceEventID(), candidateHash[:], value.ID()); err != nil {
+		return ingress.Result{}, fmt.Errorf("insert Bell event source receipt: %w", err)
+	}
+	result := ingress.Result{SchemaVersion: 1, ProducerID: producerID, SourceEventID: value.SourceEventID(), EventID: value.ID(), Status: "accepted", HTTPStatus: 201}
+	if err := insertNonce(ctx, tx, keyID, nonce, requestHash, result); err != nil {
+		return ingress.Result{}, err
+	}
+	if err := tx.Commit(); err != nil {
+		return ingress.Result{}, errors.New("commit Bell event ingress")
+	}
+	return result, nil
+}
+
+func lockIngress(ctx context.Context, tx *sql.Tx, keyID, nonce, producerID, sourceEventID string) error {
+	for _, value := range []string{"event-nonce:" + keyID + ":" + nonce, "event-source:" + producerID + ":" + sourceEventID} {
+		if _, err := tx.ExecContext(ctx, `SELECT pg_advisory_xact_lock(hashtextextended($1, 0))`, value); err != nil {
+			return errors.New("lock Bell event ingress identity")
+		}
+	}
+	return nil
+}
+
+func readNonce(ctx context.Context, tx *sql.Tx, keyID, nonce string, requestHash [sha256.Size]byte) (ingress.Result, bool, error) {
+	var storedHash, body []byte
+	var status int
+	err := tx.QueryRowContext(ctx, `SELECT request_hash, response_status, response_body::text
+		FROM bell.event_ingress_nonces WHERE key_id=$1 AND nonce=$2`, keyID, nonce).Scan(&storedHash, &status, &body)
+	if errors.Is(err, sql.ErrNoRows) {
+		return ingress.Result{}, false, nil
+	}
+	if err != nil {
+		return ingress.Result{}, false, errors.New("read Bell event ingress nonce")
+	}
+	if !bytes.Equal(storedHash, requestHash[:]) {
+		return ingress.Result{}, false, ingress.ErrReplayConflict
+	}
+	var value ingress.Result
+	if err := json.Unmarshal(body, &value); err != nil {
+		return ingress.Result{}, false, errors.New("decode Bell event ingress nonce")
+	}
+	value.HTTPStatus = status
+	return value, true, nil
+}
+
+func insertNonce(ctx context.Context, tx *sql.Tx, keyID, nonce string, requestHash [sha256.Size]byte, value ingress.Result) error {
+	body, err := json.Marshal(value)
+	if err != nil {
+		return errors.New("encode Bell event ingress response")
+	}
+	if _, err := tx.ExecContext(ctx, `INSERT INTO bell.event_ingress_nonces(
+		key_id, nonce, request_hash, response_status, response_body, expires_at
+	) VALUES ($1,$2,$3,$4,$5::jsonb,clock_timestamp() + interval '10 minutes')`, keyID, nonce, requestHash[:], value.HTTPStatus, body); err != nil {
+		return errors.New("insert Bell event ingress nonce")
+	}
+	return nil
+}

Bell/internal/store/event_ingress_postgres_test.go

@@ -0,0 +1,197 @@
+package store
+
+import (
+	"context"
+	"crypto/sha256"
+	"database/sql"
+	"encoding/json"
+	"errors"
+	"fmt"
+	"os"
+	"sync"
+	"testing"
+	"time"
+
+	_ "github.com/jackc/pgx/v5/stdlib"
+
+	"yovision/bell/contracts"
+	"yovision/bell/internal/event"
+	"yovision/bell/internal/ingress"
+)
+
+type ingressStoreIDs struct{ id string }
+
+func (value ingressStoreIDs) NewEventID() (string, error) { return value.id, nil }
+
+func ingressStoreCandidate(t *testing.T, sourceEventID string) []byte {
+	t.Helper()
+	var object map[string]any
+	if err := json.Unmarshal(testCandidate(t, "brain-demo-v1"), &object); err != nil {
+		t.Fatal(err)
+	}
+	object["source_event_id"] = sourceEventID
+	object["tenant_id"] = float64(101)
+	object["site_id"] = float64(201)
+	object["device_id"] = float64(301)
+	object["sensors"] = []any{map[string]any{"device_id": float64(301), "modality": "video", "role": "primary"}}
+	object["kind"] = "zone_entry"
+	object["severity"] = "medium"
+	object["evidence"] = map[string]any{"snapshot_uris": []any{}, "clip_uri": nil, "clip_range": nil}
+	encoded, err := json.Marshal(object)
+	if err != nil {
+		t.Fatal(err)
+	}
+	return encoded
+}
+
+func ingressStoreEvent(t *testing.T, repository *Postgres, id, sourceEventID string) event.Event {
+	t.Helper()
+	guard, err := event.NewEvidenceGuard("private-customer")
+	if err != nil {
+		t.Fatal(err)
+	}
+	factory, err := event.NewFactory(contracts.EventV01Schema, ingressStoreIDs{id}, repository, guard)
+	if err != nil {
+		t.Fatal(err)
+	}
+	value, err := factory.Create(context.Background(), ingressStoreCandidate(t, sourceEventID))
+	if err != nil {
+		t.Fatal(err)
+	}
+	return value
+}
+
+func TestPostgresEventIngressAtomicSourceReceipt(t *testing.T) {
+	dsn := os.Getenv("YOVISION_TEST_BELL_POSTGRES_DSN")
+	adminDSN := os.Getenv("YOVISION_TEST_POSTGRES_ADMIN_DSN")
+	if dsn == "" || adminDSN == "" {
+		t.Skip("Bell runtime and admin PostgreSQL DSNs are not set")
+	}
+	admin, err := sql.Open("pgx", adminDSN)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer admin.Close()
+	ctx := context.Background()
+	now := time.Now().UTC()
+	statements := []struct {
+		query string
+		args  []any
+	}{
+		{`INSERT INTO bell.sites(tenant_id,id,name) VALUES ('ingress-tenant','ingress-site','Ingress Site')`, nil},
+		{`INSERT INTO bell.areas(tenant_id,site_id,id,name,capture_policy) VALUES ('ingress-tenant','ingress-site','ingress-area','Ingress Area','video_allowed')`, nil},
+		{`INSERT INTO sense.devices(id,tenant_id,site_id,serial_number,name,modality,desired_state,actual_state,area_id,created_at,updated_at)
+		 VALUES ('ingress-device','ingress-tenant','ingress-site','INGRESS-SERIAL','Ingress Device','video','enabled','online','ingress-area',$1,$1)`, []any{now}},
+		{`INSERT INTO bell.event_ingress_bindings(
+		 producer_id,tenant_id,site_id,device_id,logical_tenant_id,logical_site_id,logical_device_id,logical_area_id,modality
+		 ) VALUES ('brain-main',101,201,301,'ingress-tenant','ingress-site','ingress-device','ingress-area','video')`, nil},
+	}
+	for _, statement := range statements {
+		if _, err := admin.ExecContext(ctx, statement.query, statement.args...); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	db, err := sql.Open("pgx", dsn)
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer db.Close()
+	repository, err := OpenPostgres(ctx, db)
+	if err != nil {
+		t.Fatal(err)
+	}
+	if err := repository.EventIngressReady(ctx); err != nil {
+		t.Fatal(err)
+	}
+	candidate := ingressStoreCandidate(t, "BRN-ingress-0001")
+	candidateHash := sha256.Sum256(candidate)
+	value := ingressStoreEvent(t, repository, "evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2C", "BRN-ingress-0001")
+	requestHash := sha256.Sum256([]byte("first-request"))
+	result, err := repository.ProcessEvent(ctx, "brain-a", "AAAAAAAAAAAAAAAAAAAAAA", requestHash, "brain-main", candidateHash, value)
+	if err != nil || result.Status != "accepted" || result.HTTPStatus != 201 {
+		t.Fatalf("first ingress: %+v %v", result, err)
+	}
+
+	// This is the crash-after-Bell-commit case: Brain uses a new nonce before
+	// it has locally recorded the first response.
+	replayed, found, err := repository.Replay(
+		ctx, "brain-a", "BBBBBBBBBBBBBBBBBBBBBB", sha256.Sum256([]byte("retry-request")),
+		"brain-main", "BRN-ingress-0001", candidateHash,
+	)
+	if err != nil || !found || replayed.Status != "duplicate" || replayed.EventID != result.EventID {
+		t.Fatalf("durable source replay: %+v found=%v err=%v", replayed, found, err)
+	}
+	conflictHash := sha256.Sum256([]byte("changed-candidate"))
+	if _, _, err := repository.Replay(
+		ctx, "brain-a", "CCCCCCCCCCCCCCCCCCCCCC", sha256.Sum256([]byte("conflict-request")),
+		"brain-main", "BRN-ingress-0001", conflictHash,
+	); !errors.Is(err, ingress.ErrSourceConflict) {
+		t.Fatalf("expected source conflict, got %v", err)
+	}
+	if _, _, err := repository.Replay(
+		ctx, "brain-a", "AAAAAAAAAAAAAAAAAAAAAA", sha256.Sum256([]byte("different-request")),
+		"brain-main", "BRN-ingress-0001", candidateHash,
+	); !errors.Is(err, ingress.ErrReplayConflict) {
+		t.Fatalf("expected nonce replay conflict, got %v", err)
+	}
+
+	var wait sync.WaitGroup
+	errorsSeen := make(chan error, 8)
+	concurrentValues := make([]event.Event, 8)
+	for index := range concurrentValues {
+		id := fmt.Sprintf("evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2%c", "DEFGHJKM"[index])
+		concurrentValues[index] = ingressStoreEvent(t, repository, id, "BRN-ingress-0001")
+	}
+	for index := 0; index < 8; index++ {
+		wait.Add(1)
+		go func(index int) {
+			defer wait.Done()
+			nonce := base64Nonce(index)
+			response, err := repository.ProcessEvent(
+				ctx, "brain-a", nonce, sha256.Sum256([]byte(nonce)), "brain-main", candidateHash, concurrentValues[index],
+			)
+			if err != nil || response.Status != "duplicate" || response.EventID != result.EventID {
+				errorsSeen <- fmt.Errorf("concurrent duplicate %d: %+v %w", index, response, err)
+			}
+		}(index)
+	}
+	wait.Wait()
+	close(errorsSeen)
+	for err := range errorsSeen {
+		t.Error(err)
+	}
+	var eventCount, receiptCount int
+	if err := db.QueryRowContext(ctx, `SELECT
+		(SELECT count(*) FROM bell.events WHERE tenant_id=101 AND site_id=201 AND source_event_id='BRN-ingress-0001'),
+		(SELECT count(*) FROM bell.event_ingress_receipts WHERE producer_id='brain-main' AND source_event_id='BRN-ingress-0001')`).Scan(&eventCount, &receiptCount); err != nil {
+		t.Fatal(err)
+	}
+	if eventCount != 1 || receiptCount != 1 {
+		t.Fatalf("concurrent ingress created events=%d receipts=%d", eventCount, receiptCount)
+	}
+
+	if _, err := admin.ExecContext(ctx, `UPDATE bell.areas SET capture_policy='non_imaging_only'
+		WHERE tenant_id='ingress-tenant' AND id='ingress-area'`); err != nil {
+		t.Fatal(err)
+	}
+	allowed, err := repository.VideoAllowed(ctx, 101, 201, 301)
+	if err != nil || allowed {
+		t.Fatalf("privacy change did not fail closed: allowed=%v err=%v", allowed, err)
+	}
+	guard, _ := event.NewEvidenceGuard("private-customer")
+	factory, _ := event.NewFactory(contracts.EventV01Schema, ingressStoreIDs{"evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2N"}, repository, guard)
+	_, err = factory.Create(ctx, ingressStoreCandidate(t, "BRN-ingress-0002"))
+	var validation *event.ValidationError
+	if !errors.As(err, &validation) || validation.Code != event.CodePrivacyDenied {
+		t.Fatalf("privacy denial code drift: %v", err)
+	}
+	if _, err := db.ExecContext(ctx, `UPDATE bell.event_ingress_receipts SET event_id=event_id
+		WHERE producer_id='brain-main' AND source_event_id='BRN-ingress-0001'`); err == nil {
+		t.Fatal("runtime updated immutable event source receipt")
+	}
+}
+
+func base64Nonce(index int) string {
+	return fmt.Sprintf("D%021d", index)
+}

Brain/README.md

@@ -1,6 +1,6 @@
 # YoVision Brain 单路工程原型
 
-T-017 提供一条可见、可重复的工程链路：单路 frame source → 匿名人员检测/fixture → track → 归一化多边形区域 → `zone_entry` 候选事实。它不是生产模型、不是 NVR，也没有 Brain→Bell transport。
+T-017 提供可见、可重复的单路工程链路；T-019 为 `zone_entry` 候选增加默认关闭的可靠 Brain→Bell transport。它仍不是生产模型或 NVR：推理演示保持单路，事件先进入仓库外 SQLite Outbox，再由内部 HMAC client 投递给 Bell。
 
 ## 环境
 
@@ -36,6 +36,39 @@ URL 文件只允许一行、最多 4096 字节，服务不会在状态、页面
 
 真实流使用 OpenCV 内置 HOG/SVM 人员检测和轻量 centroid tracker，只验证可替换端口与区域链路。它不是 M3 生产模型，不能据此承诺召回率、误报率、GPU 容量或 16/128 路能力。
 
+## 可选 Brain → Bell 事件投递
+
+默认不传 `--event-ingress-config`，工程原型行为与 T-017 相同。启用时，配置、key 和 SQLite 文件都必须位于仓库外绝对路径；HTTP 只允许显式回环，非回环必须 HTTPS。key 文件也供 Bell event ingress 读取，但不得与 Sense 审计 key 混用：
+
+```json
+{"version":1,"keys":[{"key_id":"brain-a","producer_id":"brain-main","secret_base64url":"<至少32字节随机值的无填充base64url>"}]}
+```
+
+Brain 配置示例（占位 ID 必须与 Bell 管理员创建的 `bell.event_ingress_bindings` 一致）：
+
+```json
+{
+  "version": 1,
+  "producer_id": "brain-main",
+  "tenant_id": 1,
+  "site_id": 1,
+  "device_id": 1,
+  "modality": "video",
+  "severity": "medium",
+  "config_version": "brain-demo-v1",
+  "bell_url": "http://127.0.0.1:8081/internal/v1/event-candidates",
+  "key_id": "brain-a",
+  "key_file": "D:\\private\\brain-event-keys.json",
+  "outbox_path": "D:\\private\\brain-event-outbox.sqlite3"
+}
+```
+
+```powershell
+python -m Brain.yovision_brain --source synthetic --event-ingress-config D:\private\brain-event-ingress.json
+```
+
+Outbox 使用 WAL 和 `synchronous=FULL`，首次打开即绑定 producer/tenant/site/device；更换身份必须使用新的 Outbox 路径，不能让旧队列借新 producer 发送。最多保留 10,000 条待投递；只有 Bell `accepted/duplicate` 才确认 delivered。网络、401 与 5xx 按 1～300 秒退避、最多 100 次；稳定 4xx 进入 dead letter。终态行不会自动删除。状态页只显示计数和稳定错误码，不显示 URL、key、payload 或数据库路径；client 显式忽略环境 HTTP proxy。
+
 ## 验证
 
 ```powershell
@@ -48,6 +81,6 @@ python -m compileall -q Brain
 ## 边界
 
 - `source_event_id` 由 Brain 产生，平台 `evt_` ULID 由 Bell 产生。
-- 事件只保留在最多 100 项的内存环中；重启即丢失是刻意边界。
-- T-018 冻结并实现身份映射、候选契约、持久 Outbox、HMAC、幂等和 Bell ingress。
+- 页面仍只保留最多 100 项的内存环；启用 T-019 后，可靠性由独立 SQLite Outbox 承担，不能从页面事件环推断投递状态。
+- Bell 平台 ID 只由 Bell 生成；相同 producer/source candidate 重投返回原 ID，Brain 不改写 source ID 规避冲突。
 - `D:\OPC\silver_pose` 保持独立，不是本模块的源码目录、运行依赖或模型来源路径。

Brain/tests/test_ingress.py

@@ -0,0 +1,291 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import json
+import tempfile
+import threading
+import unittest
+from datetime import datetime, timezone
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+
+from Brain.yovision_brain.domain import EventCandidate
+from Brain.yovision_brain.ingress import (
+    BrainEventIngress,
+    DeliveryResult,
+    EventIngressClient,
+    EventOutbox,
+    EventRelayWorker,
+    IngressConfig,
+    KeyMaterial,
+    PermanentDelivery,
+    RetryableDelivery,
+    load_config,
+    map_candidate,
+)
+
+
+EVENT_ID = "evt_01J8XQ2K7M3P5R9T0V4W6Y8Z2B"
+
+
+def candidate(source_event_id: str = "BRN-test-0001", fixture: bool = True) -> EventCandidate:
+    return EventCandidate(
+        source_event_id=source_event_id,
+        kind="zone_entry",
+        source_ref="must-not-leave-brain",
+        track_id="P-1",
+        zone_id="zone-1",
+        zone_version=3,
+        occurred_at=datetime(2026, 8, 11, tzinfo=timezone.utc).isoformat().replace("+00:00", "Z"),
+        confidence=None,
+        fixture=fixture,
+    )
+
+
+def config(root: Path, bell_url: str = "http://127.0.0.1:8081/internal/v1/event-candidates") -> IngressConfig:
+    return IngressConfig(
+        producer_id="brain-main",
+        tenant_id=1,
+        site_id=2,
+        device_id=3,
+        modality="video",
+        severity="medium",
+        config_version="brain-demo-v1",
+        bell_url=bell_url,
+        key_id="brain-a",
+        key_file=root / "keys.json",
+        outbox_path=root / "outbox.sqlite3",
+    )
+
+
+class ConfigAndMapperTests(unittest.TestCase):
+    def test_external_config_binds_key_and_rejects_remote_plaintext(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            secret = bytes(range(32))
+            key_file = root / "keys.json"
+            key_file.write_text(
+                json.dumps(
+                    {
+                        "version": 1,
+                        "keys": [
+                            {
+                                "key_id": "brain-a",
+                                "producer_id": "brain-main",
+                                "secret_base64url": base64.urlsafe_b64encode(secret).rstrip(b"=").decode("ascii"),
+                            }
+                        ],
+                    }
+                ),
+                encoding="utf-8",
+            )
+            config_file = root / "config.json"
+            document = {
+                "version": 1,
+                "producer_id": "brain-main",
+                "tenant_id": 1,
+                "site_id": 2,
+                "device_id": 3,
+                "modality": "video",
+                "severity": "medium",
+                "config_version": "brain-demo-v1",
+                "bell_url": "http://127.0.0.1:8081/internal/v1/event-candidates",
+                "key_id": "brain-a",
+                "key_file": str(key_file.resolve()),
+                "outbox_path": str((root / "outbox.sqlite3").resolve()),
+            }
+            config_file.write_text(json.dumps(document), encoding="utf-8")
+            loaded, key = load_config(str(config_file.resolve()))
+            self.assertEqual(("brain-main", secret), (loaded.producer_id, key.secret))
+
+            document["bell_url"] = "http://camera.example/internal/v1/event-candidates"
+            config_file.write_text(json.dumps(document), encoding="utf-8")
+            with self.assertRaisesRegex(ValueError, "HTTPS"):
+                load_config(str(config_file.resolve()))
+
+    def test_mapper_produces_complete_candidate_without_sensitive_source(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            payload = json.loads(map_candidate(candidate(), config(Path(directory))))
+        self.assertNotIn("id", payload)
+        self.assertNotIn("source_ref", json.dumps(payload))
+        self.assertEqual("test", payload["outcome"])
+        self.assertEqual("auto", payload["outcome_source"])
+        self.assertIsNone(payload["confidence"])
+        self.assertEqual(payload["occurred_at"], payload["detected_at"])
+        self.assertEqual(0.0, payload["latency_seconds"])
+        self.assertEqual([{"device_id": 3, "modality": "video", "role": "primary"}], payload["sensors"])
+        expected = {
+            "schema_version", "source_event_id", "tenant_id", "site_id", "device_id", "sensors", "kind",
+            "severity", "confidence", "occurred_at", "detected_at", "latency_seconds", "config_version",
+            "rule", "subject", "observation", "evidence", "dedup_key", "aggregated_into", "outcome",
+            "outcome_source", "outcome_reason", "diagnostics", "ext",
+        }
+        self.assertEqual(expected, set(payload))
+
+
+class OutboxTests(unittest.TestCase):
+    def test_crash_recovery_retry_and_terminal_records_are_durable(self) -> None:
+        clock = [1_000.0]
+        now = lambda: clock[0]
+        with tempfile.TemporaryDirectory() as directory:
+            path = Path(directory) / "outbox.sqlite3"
+            payload = map_candidate(candidate(), config(Path(directory)))
+            outbox = EventOutbox(path, now=now)
+            self.assertTrue(outbox.enqueue(payload))
+            self.assertFalse(outbox.enqueue(payload))
+            leased = outbox.lease_one()
+            self.assertIsNotNone(leased)
+            outbox.close()  # Simulate a process crash while the lease is held.
+
+            clock[0] += 31.0
+            outbox = EventOutbox(path, now=now)
+            recovered = outbox.lease_one()
+            self.assertIsNotNone(recovered)
+            self.assertEqual(2, recovered.attempt_count)
+            outbox.mark_delivered(recovered, DeliveryResult("duplicate", EVENT_ID))
+            self.assertEqual(1, outbox.status()["delivered"])
+            outbox.close()
+
+            outbox = EventOutbox(path, now=now)
+            status = outbox.status()
+            self.assertEqual((1, 0), (status["delivered"], status["queued"]))
+            outbox.close()
+
+    def test_source_id_conflict_is_not_silently_overwritten(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            outbox = EventOutbox(root / "outbox.sqlite3")
+            outbox.enqueue(map_candidate(candidate(), config(root)))
+            changed = json.loads(map_candidate(candidate(), config(root)))
+            changed["config_version"] = "different"
+            with self.assertRaisesRegex(ValueError, "different candidate"):
+                outbox.enqueue(json.dumps(changed, sort_keys=True, separators=(",", ":")).encode())
+            outbox.close()
+
+    def test_crash_on_final_attempt_becomes_dead_letter(self) -> None:
+        clock = [1_000.0]
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            path = root / "outbox.sqlite3"
+            outbox = EventOutbox(path, now=lambda: clock[0])
+            outbox.enqueue(map_candidate(candidate(), config(root)))
+            for _ in range(99):
+                item = outbox.lease_one()
+                self.assertIsNotNone(item)
+                outbox.mark_retry(item, "network_error")
+                clock[0] += 301.0
+            final = outbox.lease_one()
+            self.assertEqual(100, final.attempt_count)
+            outbox.close()
+            clock[0] += 31.0
+            outbox = EventOutbox(path, now=lambda: clock[0])
+            self.assertIsNone(outbox.lease_one())
+            self.assertEqual(1, outbox.status()["dead_letter"])
+            outbox.close()
+
+    def test_outbox_cannot_be_reused_for_another_producer_identity(self) -> None:
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            path = root / "outbox.sqlite3"
+            outbox = EventOutbox(path)
+            original = config(root)
+            outbox.bind_identity(original)
+            outbox.close()
+            outbox = EventOutbox(path)
+            changed = IngressConfig(**{**original.__dict__, "producer_id": "brain-other"})
+            with self.assertRaisesRegex(ValueError, "different producer"):
+                outbox.bind_identity(changed)
+            outbox.close()
+
+
+class _BellHandler(BaseHTTPRequestHandler):
+    secret = bytes(range(32))
+    verified = False
+    failure: str | None = None
+
+    def log_message(self, _format: str, *_args: object) -> None:
+        return
+
+    def do_POST(self) -> None:  # noqa: N802
+        try:
+            length = int(self.headers["Content-Length"])
+            body = self.rfile.read(length)
+            timestamp = self.headers["X-YoVision-Timestamp"]
+            nonce = self.headers["X-YoVision-Nonce"]
+            canonical = "\n".join(("POST", self.path, timestamp, nonce, hashlib.sha256(body).hexdigest())).encode()
+            encoded_signature = self.headers["X-YoVision-Signature"]
+            supplied = base64.urlsafe_b64decode(encoded_signature + "=" * (-len(encoded_signature) % 4))
+            type(self).verified = hmac.compare_digest(supplied, hmac.new(self.secret, canonical, hashlib.sha256).digest())
+            envelope = json.loads(body)
+            response = json.dumps(
+                {
+                    "schema_version": 1,
+                    "producer_id": envelope["producer_id"],
+                    "source_event_id": envelope["candidate"]["source_event_id"],
+                    "event_id": EVENT_ID,
+                    "status": "accepted",
+                }
+            ).encode()
+            self.send_response(201)
+            self.send_header("Content-Type", "application/json")
+            self.send_header("Content-Length", str(len(response)))
+            self.end_headers()
+            self.wfile.write(response)
+        except Exception as exc:  # pragma: no cover - only improves test diagnostics
+            type(self).failure = repr(exc)
+            self.send_response(500)
+            self.end_headers()
+
+
+class ClientAndWorkerTests(unittest.TestCase):
+    def test_client_signs_and_accepts_bell_response(self) -> None:
+        server = HTTPServer(("127.0.0.1", 0), _BellHandler)
+        thread = threading.Thread(target=server.serve_forever, daemon=True)
+        thread.start()
+        try:
+            with tempfile.TemporaryDirectory() as directory:
+                root = Path(directory)
+                client = EventIngressClient(
+                    config(root, f"http://127.0.0.1:{server.server_port}/internal/v1/event-candidates"),
+                    KeyMaterial("brain-a", "brain-main", _BellHandler.secret),
+                    now=lambda: 1_800_000_000.0,
+                )
+                try:
+                    result = client.deliver(map_candidate(candidate(), config(root)))
+                except RetryableDelivery as exc:
+                    self.fail(f"test Bell handler failed: {_BellHandler.failure}; client={exc.code}")
+            self.assertEqual(("accepted", EVENT_ID), (result.status, result.event_id))
+            self.assertTrue(_BellHandler.verified)
+        finally:
+            server.shutdown()
+            server.server_close()
+            thread.join(timeout=2.0)
+
+    def test_worker_distinguishes_retryable_and_permanent_failures(self) -> None:
+        class FailingClient:
+            def __init__(self, failure: Exception) -> None:
+                self.failure = failure
+
+            def deliver(self, _payload: bytes) -> DeliveryResult:
+                raise self.failure
+
+        with tempfile.TemporaryDirectory() as directory:
+            root = Path(directory)
+            payload = map_candidate(candidate(), config(root))
+            outbox = EventOutbox(root / "retry.sqlite3", now=lambda: 100.0)
+            outbox.enqueue(payload)
+            EventRelayWorker(outbox, FailingClient(RetryableDelivery("unauthorized"))).run_once()
+            self.assertEqual((1, 0), (outbox.status()["queued"], outbox.status()["dead_letter"]))
+            outbox.close()
+
+            outbox = EventOutbox(root / "dead.sqlite3", now=lambda: 100.0)
+            outbox.enqueue(payload)
+            EventRelayWorker(outbox, FailingClient(PermanentDelivery("source_event_conflict"))).run_once()
+            self.assertEqual(1, outbox.status()["dead_letter"])
+            outbox.close()
+
+
+if __name__ == "__main__":
+    unittest.main()

Brain/tests/test_runtime.py

@@ -2,8 +2,50 @@ from __future__ import annotations
 
 import unittest
 
-from Brain.yovision_brain.runtime import DemoEngine
-from Brain.yovision_brain.source import SyntheticSource, cv2
+from datetime import datetime, timezone
+
+from Brain.yovision_brain.domain import Box, Detection
+from Brain.yovision_brain.runtime import DemoEngine, EventIngressUnavailable
+from Brain.yovision_brain.source import FramePacket, SyntheticSource, cv2, np
+
+
+class _TwoFrameSource:
+    fixture = True
+    mode = "test"
+    label = "test"
+    source_ref = "safe-source"
+
+    def __init__(self) -> None:
+        self.index = 0
+
+    def read(self) -> FramePacket:
+        boxes = (Box(0.10, 0.4, 0.20, 0.8), Box(0.60, 0.4, 0.70, 0.8))
+        detection = Detection("P-1", "person", boxes[min(self.index, 1)])
+        self.index += 1
+        return FramePacket(np.zeros((180, 320, 3), dtype=np.uint8), datetime.now(timezone.utc), (detection,))
+
+    def close(self) -> None:
+        return
+
+
+class _Ingress:
+    def __init__(self, fail: bool = False) -> None:
+        self.items = []
+        self.fail = fail
+
+    def submit(self, value: object) -> None:
+        if self.fail:
+            raise RuntimeError("disk unavailable")
+        self.items.append(value)
+
+    def start(self) -> None:
+        return
+
+    def stop(self) -> None:
+        return
+
+    def status(self) -> dict[str, object]:
+        return {"enabled": True, "queued": len(self.items)}
 
 
 @unittest.skipIf(cv2 is None, "pinned OpenCV package is not installed")
@@ -28,6 +70,23 @@ class RuntimeTests(unittest.TestCase):
         self.assertEqual(updated.version, 2)
         self.assertEqual(engine.state()["zone"]["name"], "新区域")
 
+    def test_event_is_persisted_before_it_is_exposed(self) -> None:
+        ingress = _Ingress()
+        engine = DemoEngine(_TwoFrameSource(), event_ingress=ingress)
+        engine.step()
+        engine.step()
+        state = engine.state()
+        self.assertEqual(1, len(ingress.items))
+        self.assertEqual("outbox_persisted", state["events"][0]["delivery_status"])
+        self.assertEqual(1, state["event_ingress"]["queued"])
+
+    def test_outbox_failure_does_not_claim_delivery(self) -> None:
+        engine = DemoEngine(_TwoFrameSource(), event_ingress=_Ingress(fail=True))
+        engine.step()
+        with self.assertRaises(EventIngressUnavailable):
+            engine.step()
+        self.assertEqual([], engine.state()["events"])
+
 
 if __name__ == "__main__":
     unittest.main()

Brain/tests/test_ui_contract.py

@@ -40,6 +40,22 @@ class UIContractTests(unittest.TestCase):
         self.assertNotRegex(visible_without_script, r"[\U0001F300-\U0001FAFF]")
         self.assertEqual(self.html.count("__BRAIN_DEMO_TOKEN__"), 1)
 
+    def test_reliable_delivery_status_is_visible_without_configuration_details(self) -> None:
+        required = (
+            'id="event-ingress"',
+            'id="ingress-queued"',
+            'id="ingress-dead-letter"',
+            "Outbox 已持久化",
+            "Bell 返回 accepted 或 duplicate 后才计为已送达",
+            "event_ingress: {enabled: false}",
+        )
+        for marker in required:
+            with self.subTest(marker=marker):
+                self.assertIn(marker, self.html)
+        for forbidden in ("bell_url", "key_file", "outbox_path"):
+            with self.subTest(forbidden=forbidden):
+                self.assertNotIn(forbidden, self.html)
+
 
 if __name__ == "__main__":
     unittest.main()

Brain/yovision_brain/__main__.py

@@ -3,6 +3,7 @@ from __future__ import annotations
 import argparse
 import sys
 
+from .ingress import BrainEventIngress
 from .runtime import DemoEngine
 from .server import parse_bind, serve
 from .source import StreamSource, SyntheticSource, read_stream_url, require_opencv
@@ -14,6 +15,7 @@ def build_parser() -> argparse.ArgumentParser:
     parser.add_argument("--stream-url-file", help="absolute external file containing one RTSP URL")
     parser.add_argument("--bind", default="127.0.0.1:8090", help="loopback bind address")
     parser.add_argument("--fps", type=float, default=2.0, help="prototype processing FPS (0, 30]")
+    parser.add_argument("--event-ingress-config", help="absolute external Brain-to-Bell ingress configuration")
     return parser
 
 
@@ -30,7 +32,8 @@ def main(argv: list[str] | None = None) -> int:
             if args.stream_url_file:
                 raise ValueError("--stream-url-file is only valid for stream mode")
             source = SyntheticSource()
-        engine = DemoEngine(source, fps=args.fps)
+        event_ingress = BrainEventIngress.from_file(args.event_ingress_config) if args.event_ingress_config else None
+        engine = DemoEngine(source, fps=args.fps, event_ingress=event_ingress)
     except (RuntimeError, ValueError) as exc:
         print(f"Brain demo configuration error: {exc}", file=sys.stderr)
         return 2

Brain/yovision_brain/ingress.py

@@ -0,0 +1,645 @@
+from __future__ import annotations
+
+import base64
+import binascii
+import hashlib
+import hmac
+import ipaddress
+import json
+import re
+import secrets
+import sqlite3
+import threading
+import time
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Callable
+from urllib.error import HTTPError, URLError
+from urllib.parse import urlsplit
+from urllib.request import HTTPRedirectHandler, ProxyHandler, Request, build_opener
+
+from .domain import EventCandidate
+
+
+INGRESS_PATH = "/internal/v1/event-candidates"
+MAX_PAYLOAD_BYTES = 1 << 20
+MAX_QUEUED = 10_000
+MAX_ATTEMPTS = 100
+HTTP_TIMEOUT_SECONDS = 10.0
+_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$")
+_SOURCE_ID = re.compile(r"^[A-Za-z0-9_-]{1,128}$")
+_EVENT_ID = re.compile(r"^evt_[0-9A-HJKMNP-TV-Z]{26}$")
+_ERROR_CODE = re.compile(r"^[a-z][a-z0-9_]{0,63}$")
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+@dataclass(frozen=True)
+class IngressConfig:
+    producer_id: str
+    tenant_id: int
+    site_id: int
+    device_id: int
+    modality: str
+    severity: str
+    config_version: str
+    bell_url: str
+    key_id: str
+    key_file: Path
+    outbox_path: Path
+
+
+@dataclass(frozen=True)
+class KeyMaterial:
+    key_id: str
+    producer_id: str
+    secret: bytes
+
+
+@dataclass(frozen=True)
+class LeasedEvent:
+    source_event_id: str
+    payload: bytes
+    attempt_count: int
+
+
+@dataclass(frozen=True)
+class DeliveryResult:
+    status: str
+    event_id: str
+
+
+class RetryableDelivery(RuntimeError):
+    def __init__(self, code: str) -> None:
+        super().__init__(code)
+        self.code = _safe_error(code)
+
+
+class PermanentDelivery(RuntimeError):
+    def __init__(self, code: str) -> None:
+        super().__init__(code)
+        self.code = _safe_error(code)
+
+
+class _NoRedirect(HTTPRedirectHandler):
+    def redirect_request(self, _request: Request, _file_pointer: object, _code: int, _message: str, _headers: object, _new_url: str) -> None:
+        return None
+
+
+def _is_within(path: Path, parent: Path) -> bool:
+    try:
+        path.relative_to(parent)
+        return True
+    except ValueError:
+        return False
+
+
+def _external_absolute_path(raw: object, field: str, *, must_exist: bool) -> Path:
+    if not isinstance(raw, str) or not raw:
+        raise ValueError(f"{field} must be an absolute external path")
+    path = Path(raw)
+    if not path.is_absolute():
+        raise ValueError(f"{field} must be an absolute external path")
+    resolved = path.resolve(strict=must_exist)
+    if _is_within(resolved, _REPO_ROOT):
+        raise ValueError(f"{field} must stay outside the repository")
+    return resolved
+
+
+def _load_json_file(path: Path, maximum: int = 64 << 10) -> object:
+    raw = path.read_bytes()
+    if len(raw) > maximum:
+        raise ValueError("external configuration file is too large")
+    try:
+        return json.loads(raw.decode("utf-8"))
+    except (UnicodeError, json.JSONDecodeError) as exc:
+        raise ValueError("external configuration is not valid JSON") from exc
+
+
+def _positive_int(value: object, field: str) -> int:
+    if isinstance(value, bool) or not isinstance(value, int) or value < 1:
+        raise ValueError(f"{field} must be a positive integer")
+    return value
+
+
+def _validate_bell_url(raw: object) -> str:
+    if not isinstance(raw, str):
+        raise ValueError("bell_url must be a URL")
+    parsed = urlsplit(raw)
+    if parsed.scheme not in {"http", "https"} or parsed.hostname is None or parsed.path != INGRESS_PATH:
+        raise ValueError("bell_url must target the versioned event ingress path")
+    if parsed.username is not None or parsed.password is not None or parsed.query or parsed.fragment:
+        raise ValueError("bell_url cannot contain credentials, query or fragment")
+    try:
+        port = parsed.port
+    except ValueError as exc:
+        raise ValueError("bell_url has an invalid port") from exc
+    if port is not None and not 1 <= port <= 65535:
+        raise ValueError("bell_url has an invalid port")
+    if parsed.scheme == "http":
+        host = parsed.hostname
+        loopback = host == "localhost"
+        if not loopback:
+            try:
+                loopback = ipaddress.ip_address(host).is_loopback
+            except ValueError:
+                loopback = False
+        if not loopback:
+            raise ValueError("non-loopback event ingress requires HTTPS")
+    return raw
+
+
+def load_config(path_value: str) -> tuple[IngressConfig, KeyMaterial]:
+    config_path = _external_absolute_path(path_value, "event ingress config", must_exist=True)
+    value = _load_json_file(config_path)
+    expected = {
+        "version",
+        "producer_id",
+        "tenant_id",
+        "site_id",
+        "device_id",
+        "modality",
+        "severity",
+        "config_version",
+        "bell_url",
+        "key_id",
+        "key_file",
+        "outbox_path",
+    }
+    if not isinstance(value, dict) or set(value) != expected or value.get("version") != 1:
+        raise ValueError("invalid event ingress configuration shape")
+    producer_id = value["producer_id"]
+    key_id = value["key_id"]
+    config_version = value["config_version"]
+    if not isinstance(producer_id, str) or not _ID.fullmatch(producer_id):
+        raise ValueError("invalid producer_id")
+    if not isinstance(key_id, str) or not _ID.fullmatch(key_id):
+        raise ValueError("invalid key_id")
+    if not isinstance(config_version, str) or not 1 <= len(config_version) <= 128:
+        raise ValueError("config_version must contain 1 to 128 characters")
+    if value["modality"] != "video":
+        raise ValueError("T-019 Brain ingress supports only video primary sensors")
+    if value["severity"] not in {"low", "medium", "high", "critical"}:
+        raise ValueError("invalid severity")
+    key_file = _external_absolute_path(value["key_file"], "key_file", must_exist=True)
+    outbox_path = _external_absolute_path(value["outbox_path"], "outbox_path", must_exist=False)
+    if not outbox_path.parent.is_dir():
+        raise ValueError("outbox_path parent directory must already exist")
+    config = IngressConfig(
+        producer_id=producer_id,
+        tenant_id=_positive_int(value["tenant_id"], "tenant_id"),
+        site_id=_positive_int(value["site_id"], "site_id"),
+        device_id=_positive_int(value["device_id"], "device_id"),
+        modality="video",
+        severity=value["severity"],
+        config_version=config_version,
+        bell_url=_validate_bell_url(value["bell_url"]),
+        key_id=key_id,
+        key_file=key_file,
+        outbox_path=outbox_path,
+    )
+    document = _load_json_file(key_file)
+    if not isinstance(document, dict) or set(document) != {"version", "keys"} or document.get("version") != 1:
+        raise ValueError("invalid event ingress key document")
+    keys = document.get("keys")
+    if not isinstance(keys, list) or not keys:
+        raise ValueError("event ingress key document has no keys")
+    selected: KeyMaterial | None = None
+    seen: set[str] = set()
+    for item in keys:
+        if not isinstance(item, dict) or set(item) != {"key_id", "producer_id", "secret_base64url"}:
+            raise ValueError("invalid event ingress key entry")
+        item_key = item["key_id"]
+        item_producer = item["producer_id"]
+        encoded = item["secret_base64url"]
+        if not isinstance(item_key, str) or not _ID.fullmatch(item_key) or item_key in seen:
+            raise ValueError("invalid or duplicate event ingress key ID")
+        if not isinstance(item_producer, str) or not _ID.fullmatch(item_producer) or not isinstance(encoded, str):
+            raise ValueError("invalid event ingress key entry")
+        seen.add(item_key)
+        try:
+            secret = base64.b64decode(encoded + "=" * (-len(encoded) % 4), altchars=b"-_", validate=True)
+        except (ValueError, binascii.Error) as exc:
+            raise ValueError("invalid event ingress secret") from exc
+        if len(secret) < 32:
+            raise ValueError("event ingress secret must contain at least 32 bytes")
+        if item_key == key_id:
+            selected = KeyMaterial(item_key, item_producer, secret)
+    if selected is None or selected.producer_id != producer_id:
+        raise ValueError("selected key is not bound to the configured producer")
+    return config, selected
+
+
+def map_candidate(value: EventCandidate, config: IngressConfig) -> bytes:
+    if not _SOURCE_ID.fullmatch(value.source_event_id):
+        raise ValueError("invalid source_event_id")
+    payload = {
+        "schema_version": "0.1",
+        "source_event_id": value.source_event_id,
+        "tenant_id": config.tenant_id,
+        "site_id": config.site_id,
+        "device_id": config.device_id,
+        "sensors": [{"device_id": config.device_id, "modality": config.modality, "role": "primary"}],
+        "kind": value.kind,
+        "severity": config.severity,
+        "confidence": None,
+        "occurred_at": value.occurred_at,
+        "detected_at": value.occurred_at,
+        "latency_seconds": 0.0,
+        "config_version": config.config_version,
+        "rule": None,
+        "subject": {
+            "class": "person",
+            "track_id": value.track_id,
+            "attributes": {},
+            "anon_id": None,
+            "identity": None,
+            "identity_status": "not_enabled",
+        },
+        "observation": {
+            "zone": value.zone_id,
+            "dwell_sec": 0.0,
+            "bbox_seq_uri": None,
+            "keypoint_seq_uri": None,
+            "signal_seq_uri": None,
+        },
+        "evidence": {"snapshot_uris": [], "clip_uri": None, "clip_range": None},
+        "dedup_key": None,
+        "aggregated_into": None,
+        "outcome": "test" if value.fixture else "unknown",
+        "outcome_source": "auto" if value.fixture else None,
+        "outcome_reason": None,
+        "diagnostics": None,
+        "ext": {"brain_candidate_version": "brain-demo-v1", "zone_version": value.zone_version},
+    }
+    encoded = json.dumps(payload, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode("utf-8")
+    if len(encoded) > MAX_PAYLOAD_BYTES:
+        raise ValueError("event candidate exceeds 1 MiB")
+    return encoded
+
+
+class EventOutbox:
+    def __init__(self, path: Path, *, now: Callable[[], float] = time.time) -> None:
+        if not path.is_absolute() or _is_within(path.resolve(), _REPO_ROOT):
+            raise ValueError("outbox path must be absolute and external")
+        self._now = now
+        self._lock = threading.RLock()
+        self._connection = sqlite3.connect(str(path), timeout=5.0, isolation_level=None, check_same_thread=False)
+        self._connection.row_factory = sqlite3.Row
+        self._connection.execute("PRAGMA journal_mode=WAL")
+        self._connection.execute("PRAGMA synchronous=FULL")
+        self._connection.execute("PRAGMA foreign_keys=ON")
+        self._connection.executescript(
+            """
+            CREATE TABLE IF NOT EXISTS event_outbox (
+                source_event_id TEXT PRIMARY KEY,
+                candidate_hash BLOB NOT NULL,
+                payload BLOB NOT NULL,
+                state TEXT NOT NULL CHECK(state IN ('queued','delivering','delivered','dead_letter')),
+                attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count BETWEEN 0 AND 100),
+                available_at REAL NOT NULL,
+                lease_until REAL,
+                bell_event_id TEXT,
+                last_error_code TEXT,
+                created_at REAL NOT NULL,
+                updated_at REAL NOT NULL,
+                delivered_at REAL,
+                dead_lettered_at REAL,
+                CHECK(length(candidate_hash)=32),
+                CHECK(length(payload)<=1048576),
+                CHECK((state='delivering')=(lease_until IS NOT NULL)),
+                CHECK(delivered_at IS NULL OR dead_lettered_at IS NULL)
+            );
+            CREATE INDEX IF NOT EXISTS event_outbox_due_idx
+                ON event_outbox(available_at, created_at, source_event_id)
+                WHERE state='queued';
+            CREATE TABLE IF NOT EXISTS event_outbox_identity (
+                singleton INTEGER PRIMARY KEY CHECK(singleton=1),
+                producer_id TEXT NOT NULL,
+                tenant_id INTEGER NOT NULL CHECK(tenant_id>=1),
+                site_id INTEGER NOT NULL CHECK(site_id>=1),
+                device_id INTEGER NOT NULL CHECK(device_id>=1)
+            );
+            """
+        )
+
+    def bind_identity(self, config: IngressConfig) -> None:
+        identity = (config.producer_id, config.tenant_id, config.site_id, config.device_id)
+        with self._lock:
+            self._connection.execute("BEGIN IMMEDIATE")
+            try:
+                row = self._connection.execute(
+                    "SELECT producer_id,tenant_id,site_id,device_id FROM event_outbox_identity WHERE singleton=1"
+                ).fetchone()
+                if row is None:
+                    self._connection.execute(
+                        """INSERT INTO event_outbox_identity(
+                           singleton,producer_id,tenant_id,site_id,device_id
+                           ) VALUES (1,?,?,?,?)""",
+                        identity,
+                    )
+                elif tuple(row) != identity:
+                    raise ValueError("event outbox is bound to a different producer or device identity")
+                self._connection.execute("COMMIT")
+            except Exception:
+                self._connection.execute("ROLLBACK")
+                raise
+
+    def enqueue(self, payload: bytes) -> bool:
+        if not 0 < len(payload) <= MAX_PAYLOAD_BYTES:
+            raise ValueError("event candidate payload must contain at most 1 MiB")
+        try:
+            candidate = json.loads(payload)
+            source_event_id = candidate["source_event_id"]
+        except (UnicodeError, json.JSONDecodeError, KeyError, TypeError) as exc:
+            raise ValueError("invalid event candidate payload") from exc
+        if not isinstance(source_event_id, str) or not _SOURCE_ID.fullmatch(source_event_id):
+            raise ValueError("invalid source_event_id")
+        digest = hashlib.sha256(payload).digest()
+        now = self._now()
+        with self._lock:
+            self._connection.execute("BEGIN IMMEDIATE")
+            try:
+                existing = self._connection.execute(
+                    "SELECT candidate_hash FROM event_outbox WHERE source_event_id=?", (source_event_id,)
+                ).fetchone()
+                if existing is not None:
+                    if not hmac.compare_digest(existing["candidate_hash"], digest):
+                        raise ValueError("source_event_id already has a different candidate")
+                    self._connection.execute("COMMIT")
+                    return False
+                active = self._connection.execute(
+                    "SELECT count(*) FROM event_outbox WHERE state IN ('queued','delivering')"
+                ).fetchone()[0]
+                if active >= MAX_QUEUED:
+                    raise RuntimeError("event outbox capacity exceeded")
+                self._connection.execute(
+                    """INSERT INTO event_outbox(
+                        source_event_id,candidate_hash,payload,state,available_at,created_at,updated_at
+                    ) VALUES (?,?,?,'queued',?,?,?)""",
+                    (source_event_id, digest, payload, now, now, now),
+                )
+                self._connection.execute("COMMIT")
+                return True
+            except Exception:
+                self._connection.execute("ROLLBACK")
+                raise
+
+    def lease_one(self, lease_seconds: float = 30.0) -> LeasedEvent | None:
+        now = self._now()
+        with self._lock:
+            self._connection.execute("BEGIN IMMEDIATE")
+            try:
+                self._connection.execute(
+                    """UPDATE event_outbox SET state='dead_letter', lease_until=NULL,
+                       last_error_code='retry_exhausted', dead_lettered_at=?, updated_at=?
+                       WHERE attempt_count>=? AND (
+                         state='queued' OR (state='delivering' AND lease_until<=?)
+                       )""",
+                    (now, now, MAX_ATTEMPTS, now),
+                )
+                self._connection.execute(
+                    """UPDATE event_outbox SET state='queued', lease_until=NULL,
+                       available_at=?, last_error_code='lease_expired', updated_at=?
+                       WHERE state='delivering' AND lease_until<=? AND attempt_count<?""",
+                    (now, now, now, MAX_ATTEMPTS),
+                )
+                row = self._connection.execute(
+                    """SELECT source_event_id,payload,attempt_count FROM event_outbox
+                       WHERE state='queued' AND available_at<=? AND attempt_count<?
+                       ORDER BY available_at,created_at,source_event_id LIMIT 1""",
+                    (now, MAX_ATTEMPTS),
+                ).fetchone()
+                if row is None:
+                    self._connection.execute("COMMIT")
+                    return None
+                attempt = row["attempt_count"] + 1
+                self._connection.execute(
+                    """UPDATE event_outbox SET state='delivering',attempt_count=?,lease_until=?,updated_at=?
+                       WHERE source_event_id=? AND state='queued'""",
+                    (attempt, now + lease_seconds, now, row["source_event_id"]),
+                )
+                self._connection.execute("COMMIT")
+                return LeasedEvent(row["source_event_id"], bytes(row["payload"]), attempt)
+            except Exception:
+                self._connection.execute("ROLLBACK")
+                raise
+
+    def mark_delivered(self, item: LeasedEvent, result: DeliveryResult) -> None:
+        now = self._now()
+        with self._lock:
+            changed = self._connection.execute(
+                """UPDATE event_outbox SET state='delivered',lease_until=NULL,bell_event_id=?,
+                   last_error_code=NULL,delivered_at=?,updated_at=?
+                   WHERE source_event_id=? AND state='delivering' AND attempt_count=?""",
+                (result.event_id, now, now, item.source_event_id, item.attempt_count),
+            ).rowcount
+        if changed != 1:
+            raise RuntimeError("event outbox delivery lease was lost")
+
+    def mark_retry(self, item: LeasedEvent, code: str) -> None:
+        now = self._now()
+        if item.attempt_count >= MAX_ATTEMPTS:
+            self.mark_dead(item, "retry_exhausted")
+            return
+        delay = min(300.0, float(2 ** min(item.attempt_count - 1, 9)))
+        with self._lock:
+            changed = self._connection.execute(
+                """UPDATE event_outbox SET state='queued',lease_until=NULL,available_at=?,
+                   last_error_code=?,updated_at=?
+                   WHERE source_event_id=? AND state='delivering' AND attempt_count=?""",
+                (now + delay, _safe_error(code), now, item.source_event_id, item.attempt_count),
+            ).rowcount
+        if changed != 1:
+            raise RuntimeError("event outbox retry lease was lost")
+
+    def mark_dead(self, item: LeasedEvent, code: str) -> None:
+        now = self._now()
+        with self._lock:
+            changed = self._connection.execute(
+                """UPDATE event_outbox SET state='dead_letter',lease_until=NULL,
+                   last_error_code=?,dead_lettered_at=?,updated_at=?
+                   WHERE source_event_id=? AND state='delivering' AND attempt_count=?""",
+                (_safe_error(code), now, now, item.source_event_id, item.attempt_count),
+            ).rowcount
+        if changed != 1:
+            raise RuntimeError("event outbox dead-letter lease was lost")
+
+    def status(self) -> dict[str, object]:
+        with self._lock:
+            counts = {row["state"]: row["count"] for row in self._connection.execute(
+                "SELECT state,count(*) AS count FROM event_outbox GROUP BY state"
+            )}
+            error = self._connection.execute(
+                """SELECT last_error_code FROM event_outbox WHERE last_error_code IS NOT NULL
+                   ORDER BY updated_at DESC LIMIT 1"""
+            ).fetchone()
+        return {
+            "enabled": True,
+            "queued": counts.get("queued", 0),
+            "delivering": counts.get("delivering", 0),
+            "delivered": counts.get("delivered", 0),
+            "dead_letter": counts.get("dead_letter", 0),
+            "last_error_code": None if error is None else error["last_error_code"],
+        }
+
+    def close(self) -> None:
+        with self._lock:
+            self._connection.close()
+
+
+class EventIngressClient:
+    def __init__(self, config: IngressConfig, key: KeyMaterial, *, now: Callable[[], float] = time.time) -> None:
+        self._config = config
+        self._key = key
+        self._now = now
+        # Internal event payloads must never be redirected through ambient
+        # HTTP(S)_PROXY settings.
+        self._opener = build_opener(ProxyHandler({}), _NoRedirect())
+
+    def deliver(self, candidate: bytes) -> DeliveryResult:
+        try:
+            candidate_object = json.loads(candidate)
+        except (UnicodeError, json.JSONDecodeError) as exc:
+            raise PermanentDelivery("candidate_invalid") from exc
+        body = json.dumps(
+            {"schema_version": 1, "producer_id": self._config.producer_id, "candidate": candidate_object},
+            ensure_ascii=False,
+            sort_keys=True,
+            separators=(",", ":"),
+        ).encode("utf-8")
+        if len(body) > MAX_PAYLOAD_BYTES:
+            raise PermanentDelivery("payload_too_large")
+        timestamp = str(int(self._now()))
+        nonce = base64.urlsafe_b64encode(secrets.token_bytes(16)).rstrip(b"=").decode("ascii")
+        digest = hashlib.sha256(body).hexdigest()
+        canonical = "\n".join(("POST", INGRESS_PATH, timestamp, nonce, digest)).encode("utf-8")
+        signature_value = base64.urlsafe_b64encode(hmac.new(self._key.secret, canonical, hashlib.sha256).digest()).rstrip(b"=").decode("ascii")
+        request = Request(self._config.bell_url, data=body, method="POST")
+        request.add_header("Content-Type", "application/json")
+        request.add_header("X-YoVision-Key-Id", self._key.key_id)
+        request.add_header("X-YoVision-Timestamp", timestamp)
+        request.add_header("X-YoVision-Nonce", nonce)
+        request.add_header("X-YoVision-Signature", signature_value)
+        try:
+            with self._opener.open(request, timeout=HTTP_TIMEOUT_SECONDS) as response:
+                status = response.status
+                response_body = response.read(64 << 10)
+        except HTTPError as exc:
+            status = exc.code
+            response_body = exc.read(64 << 10)
+        except (URLError, TimeoutError, OSError) as exc:
+            raise RetryableDelivery("network_error") from exc
+        if status in {200, 201}:
+            try:
+                value = json.loads(response_body)
+                response_status = value["status"]
+                event_id = value["event_id"]
+                if (
+                    value["schema_version"] != 1
+                    or value["producer_id"] != self._config.producer_id
+                    or value["source_event_id"] != candidate_object["source_event_id"]
+                    or response_status not in {"accepted", "duplicate"}
+                    or not isinstance(event_id, str)
+                    or not _EVENT_ID.fullmatch(event_id)
+                ):
+                    raise ValueError
+            except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+                raise RetryableDelivery("invalid_response") from exc
+            return DeliveryResult(response_status, event_id)
+        code = "http_error"
+        try:
+            error_value = json.loads(response_body)
+            if isinstance(error_value, dict) and isinstance(error_value.get("error"), str):
+                code = error_value["error"]
+        except (UnicodeError, json.JSONDecodeError):
+            pass
+        if status == 401 or status >= 500:
+            raise RetryableDelivery(code)
+        raise PermanentDelivery(code)
+
+
+class EventRelayWorker:
+    def __init__(self, outbox: EventOutbox, client: EventIngressClient) -> None:
+        self._outbox = outbox
+        self._client = client
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._last_error_code: str | None = None
+
+    def run_once(self) -> bool:
+        item = self._outbox.lease_one()
+        if item is None:
+            return False
+        try:
+            result = self._client.deliver(item.payload)
+        except PermanentDelivery as exc:
+            self._outbox.mark_dead(item, exc.code)
+        except RetryableDelivery as exc:
+            self._outbox.mark_retry(item, exc.code)
+        except Exception:
+            self._outbox.mark_retry(item, "client_error")
+        else:
+            self._outbox.mark_delivered(item, result)
+        return True
+
+    def start(self) -> None:
+        if self._thread is not None:
+            return
+        self._thread = threading.Thread(target=self._run, name="brain-event-relay", daemon=True)
+        self._thread.start()
+
+    def _run(self) -> None:
+        while not self._stop.is_set():
+            try:
+                worked = self.run_once()
+                self._last_error_code = None
+            except Exception:
+                self._last_error_code = "event_outbox_unavailable"
+                self._stop.wait(1.0)
+                continue
+            if not worked:
+                self._stop.wait(0.25)
+
+    def stop(self) -> None:
+        self._stop.set()
+        if self._thread is not None:
+            self._thread.join(timeout=HTTP_TIMEOUT_SECONDS + 2.0)
+            self._thread = None
+
+    def last_error_code(self) -> str | None:
+        return self._last_error_code
+
+
+class BrainEventIngress:
+    def __init__(self, config: IngressConfig, key: KeyMaterial) -> None:
+        self._config = config
+        self._outbox = EventOutbox(config.outbox_path)
+        self._outbox.bind_identity(config)
+        self._worker = EventRelayWorker(self._outbox, EventIngressClient(config, key))
+
+    @classmethod
+    def from_file(cls, path: str) -> "BrainEventIngress":
+        config, key = load_config(path)
+        return cls(config, key)
+
+    def submit(self, value: EventCandidate) -> None:
+        self._outbox.enqueue(map_candidate(value, self._config))
+
+    def start(self) -> None:
+        self._worker.start()
+
+    def stop(self) -> None:
+        self._worker.stop()
+        self._outbox.close()
+
+    def status(self) -> dict[str, object]:
+        value = self._outbox.status()
+        if self._worker.last_error_code() is not None:
+            value["last_error_code"] = self._worker.last_error_code()
+        return value
+
+
+def _safe_error(value: str) -> str:
+    if not isinstance(value, str) or not _ERROR_CODE.fullmatch(value):
+        return "unknown_error"
+    return value

Brain/yovision_brain/runtime.py

@@ -24,8 +24,12 @@ DEFAULT_ZONE = Zone(
 )
 
 
+class EventIngressUnavailable(RuntimeError):
+    pass
+
+
 class DemoEngine:
-    def __init__(self, source: Any, fps: float = 2.0, event_limit: int = 100) -> None:
+    def __init__(self, source: Any, fps: float = 2.0, event_limit: int = 100, event_ingress: Any | None = None) -> None:
         if not math.isfinite(fps) or fps <= 0.0 or fps > 30.0:
             raise ValueError("fps must be within (0, 30]")
         if not 1 <= event_limit <= 100:
@@ -36,6 +40,7 @@ class DemoEngine:
         self._detector = None if source.fixture else HOGPersonDetector()
         self._tracker = CentroidTracker()
         self._evaluator = ZoneEntryEvaluator(source.source_ref, source.fixture)
+        self._event_ingress = event_ingress
         self._zone = DEFAULT_ZONE
         self._events: deque[dict[str, object]] = deque(maxlen=event_limit)
         self._lock = threading.RLock()
@@ -54,6 +59,8 @@ class DemoEngine:
     def start(self) -> None:
         if self._thread is not None:
             return
+        if self._event_ingress is not None:
+            self._event_ingress.start()
         self._thread = threading.Thread(target=self._run, name="brain-demo", daemon=True)
         self._thread.start()
 
@@ -62,6 +69,8 @@ class DemoEngine:
         if self._thread is not None:
             self._thread.join(timeout=3.0)
             self._thread = None
+        if self._event_ingress is not None:
+            self._event_ingress.stop()
         self._source.close()
 
     def _run(self) -> None:
@@ -70,6 +79,9 @@ class DemoEngine:
             started = time.perf_counter()
             try:
                 self.step()
+            except EventIngressUnavailable:
+                with self._lock:
+                    self._last_error_code = "event_outbox_unavailable"
             except RuntimeError:
                 with self._lock:
                     self._connected = False
@@ -89,6 +101,12 @@ class DemoEngine:
         with self._lock:
             zone = self._zone
         new_events, inside_by_track = self._evaluator.evaluate(self._sequence, packet.captured_at, zone, detections)
+        if self._event_ingress is not None:
+            try:
+                for item in new_events:
+                    self._event_ingress.submit(item)
+            except Exception as exc:
+                raise EventIngressUnavailable("persist event candidate") from exc
         ok, encoded = cv2.imencode(".jpg", packet.frame, [int(cv2.IMWRITE_JPEG_QUALITY), 82])
         if not ok:
             raise RuntimeError("frame encoding failed")
@@ -96,7 +114,12 @@ class DemoEngine:
         serialized_detections = [self._serialize_detection(item, inside_by_track.get(item.track_id, False)) for item in detections]
         with self._lock:
             for event in new_events:
-                self._events.appendleft(event.as_dict())
+                serialized_event = event.as_dict()
+                if self._event_ingress is not None:
+                    # This is an immutable handoff fact, not a live delivery
+                    # status. Current counts live under event_ingress.
+                    serialized_event["delivery_status"] = "outbox_persisted"
+                self._events.appendleft(serialized_event)
             self._frame_jpeg = encoded.tobytes()
             self._frame_width = int(width)
             self._frame_height = int(height)
@@ -130,6 +153,13 @@ class DemoEngine:
     def state(self) -> dict[str, object]:
         with self._lock:
             zone = self._zone
+            if self._event_ingress is None:
+                ingress_status: dict[str, object] = {"enabled": False}
+            else:
+                try:
+                    ingress_status = self._event_ingress.status()
+                except Exception:
+                    ingress_status = {"enabled": True, "last_error_code": "event_outbox_unavailable"}
             return {
                 "prototype": True,
                 "notice": "工程原型；合成回放不是模型输出，HOG 适配器不是生产检测模型。",
@@ -159,6 +189,7 @@ class DemoEngine:
                 },
                 "detections": list(self._detections),
                 "events": list(self._events),
+                "event_ingress": ingress_status,
                 "last_error_code": self._last_error_code,
                 "generated_at": datetime.now(timezone.utc).isoformat().replace("+00:00", "Z"),
             }

deploy/postgres/016_event_ingress.sql

@@ -0,0 +1,108 @@
+-- Bell v5 Brain event ingress identity bindings and durable receipts.
+
+CREATE TABLE IF NOT EXISTS bell.event_ingress_bindings (
+    producer_id text NOT NULL,
+    tenant_id bigint NOT NULL,
+    site_id bigint NOT NULL,
+    device_id bigint NOT NULL,
+    logical_tenant_id text NOT NULL,
+    logical_site_id text NOT NULL,
+    logical_device_id text NOT NULL,
+    logical_area_id text NOT NULL,
+    modality text NOT NULL,
+    enabled boolean NOT NULL DEFAULT true,
+    created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+    updated_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+    PRIMARY KEY (producer_id, tenant_id, site_id, device_id),
+    CONSTRAINT bell_ingress_binding_site_fk
+        FOREIGN KEY (logical_tenant_id, logical_site_id)
+        REFERENCES bell.sites(tenant_id, id),
+    CONSTRAINT bell_ingress_binding_area_fk
+        FOREIGN KEY (logical_tenant_id, logical_area_id)
+        REFERENCES bell.areas(tenant_id, id),
+    CONSTRAINT bell_ingress_binding_device_fk
+        FOREIGN KEY (logical_tenant_id, logical_site_id, logical_device_id)
+        REFERENCES sense.devices(tenant_id, site_id, id),
+    CONSTRAINT bell_ingress_binding_numeric_ids CHECK (
+        tenant_id >= 1 AND site_id >= 1 AND device_id >= 1
+    ),
+    CONSTRAINT bell_ingress_binding_producer CHECK (
+        char_length(producer_id) BETWEEN 1 AND 64
+        AND producer_id ~ '^[A-Za-z0-9][A-Za-z0-9._-]*$'
+    ),
+    CONSTRAINT bell_ingress_binding_logical_ids CHECK (
+        btrim(logical_tenant_id) <> '' AND btrim(logical_site_id) <> ''
+        AND btrim(logical_device_id) <> '' AND btrim(logical_area_id) <> ''
+    ),
+    CONSTRAINT bell_ingress_binding_modality CHECK (
+        modality IN ('video', 'radar', 'contact', 'button', 'wearable', 'other')
+    )
+);
+ALTER TABLE bell.event_ingress_bindings OWNER TO bell_app;
+
+CREATE INDEX IF NOT EXISTS bell_event_ingress_binding_numeric_idx
+    ON bell.event_ingress_bindings(tenant_id, site_id, device_id)
+    WHERE enabled;
+
+CREATE TABLE IF NOT EXISTS bell.event_ingress_receipts (
+    producer_id text NOT NULL,
+    source_event_id text NOT NULL,
+    candidate_hash bytea NOT NULL,
+    event_id text NOT NULL REFERENCES bell.events(id),
+    received_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+    PRIMARY KEY (producer_id, source_event_id),
+    CONSTRAINT bell_ingress_receipt_producer CHECK (
+        char_length(producer_id) BETWEEN 1 AND 64
+        AND producer_id ~ '^[A-Za-z0-9][A-Za-z0-9._-]*$'
+    ),
+    CONSTRAINT bell_ingress_receipt_source CHECK (
+        source_event_id ~ '^[A-Za-z0-9_-]{1,128}$'
+    ),
+    CONSTRAINT bell_ingress_receipt_hash CHECK (octet_length(candidate_hash) = 32),
+    CONSTRAINT bell_ingress_receipt_event_id CHECK (
+        event_id ~ '^evt_[0-9A-HJKMNP-TV-Z]{26}$'
+    )
+);
+ALTER TABLE bell.event_ingress_receipts OWNER TO bell_app;
+
+DROP TRIGGER IF EXISTS bell_event_ingress_receipts_immutable
+    ON bell.event_ingress_receipts;
+CREATE TRIGGER bell_event_ingress_receipts_immutable
+BEFORE UPDATE OR DELETE ON bell.event_ingress_receipts
+FOR EACH ROW EXECUTE FUNCTION bell.reject_immutable_change();
+
+CREATE INDEX IF NOT EXISTS bell_event_ingress_receipt_event_idx
+    ON bell.event_ingress_receipts(event_id);
+
+CREATE TABLE IF NOT EXISTS bell.event_ingress_nonces (
+    key_id text NOT NULL,
+    nonce text NOT NULL,
+    request_hash bytea NOT NULL,
+    response_status integer NOT NULL,
+    response_body jsonb NOT NULL,
+    received_at timestamptz NOT NULL DEFAULT clock_timestamp(),
+    expires_at timestamptz NOT NULL,
+    PRIMARY KEY (key_id, nonce),
+    CONSTRAINT bell_ingress_nonce_key CHECK (
+        char_length(key_id) BETWEEN 1 AND 64
+        AND key_id ~ '^[A-Za-z0-9][A-Za-z0-9._-]*$'
+    ),
+    CONSTRAINT bell_ingress_nonce_value CHECK (
+        char_length(nonce) BETWEEN 22 AND 64
+        AND nonce ~ '^[A-Za-z0-9_-]+$'
+    ),
+    CONSTRAINT bell_ingress_nonce_hash CHECK (octet_length(request_hash) = 32),
+    CONSTRAINT bell_ingress_nonce_status CHECK (response_status IN (200, 201)),
+    CONSTRAINT bell_ingress_nonce_body CHECK (jsonb_typeof(response_body) = 'object'),
+    CONSTRAINT bell_ingress_nonce_ttl CHECK (
+        expires_at >= received_at + interval '10 minutes'
+        AND expires_at <= received_at + interval '11 minutes'
+    )
+);
+ALTER TABLE bell.event_ingress_nonces OWNER TO bell_app;
+
+CREATE INDEX IF NOT EXISTS bell_event_ingress_nonce_expiry_idx
+    ON bell.event_ingress_nonces(expires_at, key_id, nonce);
+
+INSERT INTO bell.schema_migrations(version) VALUES (5)
+ON CONFLICT (version) DO NOTHING;

deploy/postgres/017_privileges_event_ingress.sql

@@ -0,0 +1,25 @@
+-- Bell runtime may read controlled identity bindings, append events/source
+-- receipts and maintain only short-lived nonce receipts.
+
+REVOKE ALL ON TABLE bell.event_ingress_bindings,
+    bell.event_ingress_receipts, bell.event_ingress_nonces FROM PUBLIC;
+REVOKE ALL ON TABLE bell.event_ingress_bindings,
+    bell.event_ingress_receipts, bell.event_ingress_nonces FROM bell_runtime;
+
+GRANT SELECT ON TABLE bell.event_ingress_bindings TO bell_runtime;
+GRANT SELECT, INSERT ON TABLE bell.event_ingress_receipts TO bell_runtime;
+GRANT SELECT, INSERT, DELETE ON TABLE bell.event_ingress_nonces TO bell_runtime;
+
+-- Column grants deliberately exclude endpoint_ref, credential_ref,
+-- profile_token and path_name.
+GRANT USAGE ON SCHEMA sense TO bell_runtime;
+REVOKE ALL ON TABLE sense.devices FROM bell_runtime;
+GRANT SELECT (id, tenant_id, site_id, area_id, modality)
+    ON TABLE sense.devices TO bell_runtime;
+GRANT SELECT (tenant_id, id, deleted_at)
+    ON TABLE bell.sites TO bell_runtime;
+GRANT SELECT (tenant_id, site_id, id, capture_policy, deleted_at)
+    ON TABLE bell.areas TO bell_runtime;
+
+REVOKE ALL ON TABLE bell.event_ingress_bindings,
+    bell.event_ingress_receipts, bell.event_ingress_nonces FROM sense_app;

deploy/postgres/README.md

@@ -1,6 +1,6 @@
 # YoVision PostgreSQL 初始化
 
-本目录实现 T-009～T-012、T-015～T-016 的 PostgreSQL `17.10` schema。SQL 必须按文件名前缀顺序执行：`001`～`004` 创建 NOLOGIN 权限角色、Bell/Sense 初始对象和配额权限；`005`～`007` 增量增加 Area 与本地审计；`008`～`009` 增加 Control API 状态；`010`～`011` 增加调和 fencing、MediaMTX Path 历史归属、孤儿报告/受控处置结果；`012`～`013` 增加 Bell 不可变事件、append-only outcome 和独立 `bell_runtime` 最小权限；`014`～`015` 增加审计 Outbox relay fencing、Bell 全局审计事实、短期防重收据和双方最小权限。全部 SQL 可重放。对象 owner/迁移角色为 `bell_app`/`sense_app`；应用登录角色及密码由部署环境或密钥系统创建，Sense 登录加入 `sense_app`，Bell 运行登录只加入 `bell_runtime`，仓库不保存登录凭据。
+本目录实现 T-009～T-012、T-015～T-016、T-019 的 PostgreSQL `17.10` schema。SQL 必须按文件名前缀顺序执行：`001`～`004` 创建 NOLOGIN 权限角色、Bell/Sense 初始对象和配额权限；`005`～`007` 增量增加 Area 与本地审计；`008`～`009` 增加 Control API 状态；`010`～`011` 增加调和 fencing、MediaMTX Path 历史归属、孤儿报告/受控处置结果；`012`～`013` 增加 Bell 不可变事件、append-only outcome 和独立 `bell_runtime` 最小权限；`014`～`015` 增加审计 relay；`016`～`017` 增加 Brain 事件身份绑定、永久来源收据、短期 nonce 和最小权限。全部 SQL 可重放。对象 owner/迁移角色为 `bell_app`/`sense_app`；应用登录角色及密码由部署环境或密钥系统创建，Sense 登录加入 `sense_app`，Bell 运行登录只加入 `bell_runtime`，仓库不保存登录凭据。
 
 生产/共享实例必须由管理员先备份并在 YoVision 专用数据库中执行。Sense 进程不会用高权限自动建库或建角色。示例只使用私有环境变量，不把实际 DSN 写入脚本或日志：
 
@@ -23,6 +23,7 @@ Get-ChildItem deploy/postgres/[0-9][0-9][0-9]_*.sql |
 - `sense.device_operation_outbox` 是本地持久化队列，Bell 全局审计真相只写入 `bell.audit_events`。Sense 只领取/确认本地行，不获得 Bell 表权限；Bell 只通过签名 HTTP ingress 收取，不读取 Outbox。
 - `bell.audit_events` 与事件事实一样不可更新/删除；`bell.audit_relay_receipts` 只为 10 分钟 nonce 幂等窗口保留，Bell runtime 仅可在这张限定表中查询、插入和清理过期记录。
 - `bell.events` 与 `bell.event_outcomes` 由 `bell_app` 拥有；`bell_runtime` 只获得 `SELECT/INSERT`，没有 owner、`UPDATE`、`DELETE` 或 `TRUNCATE` 权限，数据库 trigger 再拒绝 owner 路径的意外事实改写。
+- `bell.event_ingress_bindings` 由管理员维护并引用当前 Bell Site/Area 和 Sense Device；Bell runtime 只读绑定及 Sense 设备的逻辑 ID/Area/modality 列，不可读取 endpoint、credential、profile token 或 MediaMTX path。`event_ingress_receipts` 永久只追加；只有 `event_ingress_nonces` 可按 10 分钟 TTL 清理。
 - `sense.control_idempotency_receipts` 不保存原始 Idempotency-Key，只保存 scope/request SHA-256 和脱敏响应快照；`batch_operations`/items 只保存逻辑 ID、状态和稳定错误，不保存连接秘密。
 - 调和与孤儿租约使用 PostgreSQL `clock_timestamp()` 和 fencing token；过期 worker 不能提交完成/失败或扫描报告。`media_path_ownership`、扫描和处置表不保存 endpoint、credential 或 source URI；数据库约束禁止为 `unowned` finding 写删除结果。
 - PUBLIC 对两个业务 schema 的表和函数没有权限。

deploy/postgres/tests/assertions.sql

@@ -71,7 +71,7 @@ BEGIN
        OR NOT has_table_privilege('sense_app', 'sense.device_operation_outbox', 'DELETE') THEN
         RAISE EXCEPTION 'sense_app lacks access to its local audit Outbox';
     END IF;
-    IF (SELECT max(version) FROM bell.schema_migrations) <> 4
+    IF (SELECT max(version) FROM bell.schema_migrations) <> 5
        OR (SELECT max(version) FROM sense.schema_migrations) <> 6 THEN
         RAISE EXCEPTION 'schema migration version drift';
     END IF;
@@ -228,3 +228,29 @@ BEGIN
     END IF;
 END
 $audit_relay$;
+
+DO $event_ingress$
+BEGIN
+    IF NOT has_table_privilege('yovision_t015_bell', 'bell.event_ingress_bindings', 'SELECT')
+       OR has_table_privilege('yovision_t015_bell', 'bell.event_ingress_bindings', 'INSERT,UPDATE,DELETE,TRUNCATE')
+       OR NOT has_table_privilege('yovision_t015_bell', 'bell.event_ingress_receipts', 'SELECT,INSERT')
+       OR has_table_privilege('yovision_t015_bell', 'bell.event_ingress_receipts', 'UPDATE,DELETE,TRUNCATE')
+       OR NOT has_table_privilege('yovision_t015_bell', 'bell.event_ingress_nonces', 'SELECT,INSERT,DELETE')
+       OR has_table_privilege('yovision_t015_bell', 'bell.event_ingress_nonces', 'UPDATE,TRUNCATE') THEN
+        RAISE EXCEPTION 'Bell runtime violates event ingress privileges';
+    END IF;
+    IF NOT has_column_privilege('yovision_t015_bell', 'sense.devices', 'id', 'SELECT')
+       OR NOT has_column_privilege('yovision_t015_bell', 'sense.devices', 'area_id', 'SELECT')
+       OR has_column_privilege('yovision_t015_bell', 'sense.devices', 'endpoint_ref', 'SELECT')
+       OR has_column_privilege('yovision_t015_bell', 'sense.devices', 'credential_ref', 'SELECT')
+       OR has_column_privilege('yovision_t015_bell', 'sense.devices', 'profile_token', 'SELECT')
+       OR has_column_privilege('yovision_t015_bell', 'sense.devices', 'path_name', 'SELECT') THEN
+        RAISE EXCEPTION 'Bell runtime Sense device projection privilege drift';
+    END IF;
+    IF has_table_privilege('public', 'bell.event_ingress_bindings', 'SELECT,INSERT,UPDATE,DELETE,TRUNCATE')
+       OR has_table_privilege('public', 'bell.event_ingress_receipts', 'SELECT,INSERT,UPDATE,DELETE,TRUNCATE')
+       OR has_table_privilege('public', 'bell.event_ingress_nonces', 'SELECT,INSERT,UPDATE,DELETE,TRUNCATE') THEN
+        RAISE EXCEPTION 'Bell event ingress tables leaked to PUBLIC';
+    END IF;
+END
+$event_ingress$;

docs/00-ai-start-here.md

@@ -40,7 +40,7 @@ MVP 以默认 16 路跑通一个场景的端到端闭环；架构、数据和 UI
 
 ## 当前阶段
 
-当前为 **M0 指定型号实机准入、M1 Sense 五路混合源集成和 M2 本地 16 路软件基线均已完成，M3 已建立 Bell 不可变事件存储、Sense→Bell 全局审计 relay、Brain 单路可视化工程原型，并完成 T-018 Sense 回环 NVR 管理面纵切**。后续本地开发统一使用已准入的一台海康样机，多路软件闭环使用独立合成 RTSP 源补足；真实多设备证据延后到客户/借用/租赁条件具备时执行。客户网络尚未提供，T-013 WireGuard 继续后置；当前执行 T-019，建立 Brain→Bell 可靠业务事件 ingress。
+当前为 **M0 指定型号实机准入、M1 Sense 五路混合源集成和 M2 本地 16 路软件基线均已完成，M3 已建立 Bell 不可变事件存储、Sense→Bell 全局审计 relay、Brain 单路可视化工程原型、Sense 回环 NVR 管理面纵切和 Brain→Bell 可靠事件 ingress**。后续本地开发统一使用已准入的一台海康样机，多路软件闭环使用独立合成 RTSP 源补足；真实多设备证据延后到客户/借用/租赁条件具备时执行。客户网络尚未提供，T-013 WireGuard 继续后置；下一项建议把 Bell 规则→Alert→ack 最小纵切与证据切片拆成独立任务。
 
 优先路径：
 

docs/03-tech-stack.md

@@ -87,6 +87,12 @@ T-018 不增加生产或前端依赖：Go 使用标准库 `embed`/`net/http` 提
 
 该选择只适用于默认关闭的回环工程控制台，不冻结 Bell 最终前端框架。页面默认 16 项 cursor 分页、最多 4 路按需预览且不自动加载视频；Sense HTTP 与 WebRTC 基地址必须同时是显式回环地址。非回环 HTTPS/JWT/OIDC、MediaMTX 外部认证、录像/回放和正式客户会话均保持待冻结，不能从 T-018 的本地演示外推为生产安全或 NVR 存储能力。
 
+### 1.7 Brain → Bell 事件 ingress（T-019）
+
+T-019 不增加第三方依赖。Brain 继续使用 Python 3.10.11 标准库 `sqlite3`、`urllib`、HMAC/SHA-256 和线程；SQLite Outbox 开启 WAL 与 `synchronous=FULL`，文件、配置和 key 都必须是仓库外绝对路径。内部 HTTP client 显式禁用环境代理，避免业务事件被 `HTTP(S)_PROXY` 意外转发；回环可用 HTTP，非回环只接受 HTTPS。队列最多 10,000 条待投递、单 payload 1 MiB、1～300 秒退避、最多 100 次，终态记录保留供运维核查。
+
+Bell 复用已冻结 Go 1.26.5、PostgreSQL 17.10/pgx、JSON Schema 和 ULID，不新增消息总线或 SDK。`brain-event-ingress-v1` 是单事件、producer-bound HMAC 内部适配器；Bell 用永久来源收据和事务级 advisory lock 保证同 `(producer_id, source_event_id)` 只生成一个平台事件。该选择不冻结 Bell 公共认证、生产 Brain GPU/runtime 或未来高吞吐 transport；若容量基准证明单条 HTTP 不足，必须发布兼容迁移方案，不能绕过来源幂等键。
+
 ## 2. 外部项目边界
 
 - MiBeeNvr：只用于 M0 隔离实验室、ONVIF兼容性和交互参考，不作为生产依赖。
@@ -98,7 +104,7 @@ T-018 不增加生产或前端依赖：Go 使用标准库 `embed`/`net/http` 提
 
 - Python、Savant/DeepStream 的精确版本；Go、MediaMTX 与 PostgreSQL 已分别为 Sense M1/M2 冻结，后续阶段可按升级流程调整。
 - Bell 前端框架和组件库。
-- Brain→Bell 业务事件投递 transport；Sense→Bell 审计 relay 已独立冻结为内部 HTTP，不能据此默认 Brain transport。
+- Brain 生产推理 runtime 与更高吞吐 transport；T-019 已冻结首版 Brain→Bell 单事件内部 HTTP ingress，但不据此承诺 64/128 路吞吐或消息总线选型。
 - 目标 GPU/边缘硬件、解码能力和每 worker 的 `max_sources`。
 - MinIO/S3 的精确版本、加密实现，以及客户/法务确认后的最终生命周期策略。
 - 短信/语音供应商及生产双路径组合；是否开发原生 App 最早在 M4 根据试点反馈决定。

docs/04-architecture.md

@@ -47,15 +47,15 @@ Sense ── 视频流/触发信号 ──> Brain
 
 1. Bell 持有站点、Area、配额与 `capture_policy`；首期在同一 PostgreSQL 实例内发布 `bell.site_quota_v1` 和 `bell.area_policy_v1` 两个版本化只读视图。T-009/T-010 已实现 Bell 源表/视图、最小权限和 Sense PostgreSQL repository；Sense 按 Area→Site 的固定 advisory-lock 顺序执行策略与配额准入并记录所用版本。未来分库必须发布新版本契约，不能静默改变 v1 语义。
 2. Sense 维护设备期望态，通过 MediaMTX API 和对账器收敛实际态；PostgreSQL 多实例以数据库时钟短租约和 fencing token 领取 due row，过期 worker 不得提交结果。
-3. Brain 消费视频与触发信号，产生符合冻结契约的事件候选。T-017 已先建立单路 frame source、可替换 detector、轻量 track、多边形进入判定和回环可视化工程原型；合成 fixture 与 OpenCV HOG 均明确不是生产模型，候选只含 `source_event_id`，不自报平台 `id`。T-018 的 Sense 回环控制台只读取设备事实并按需嵌入 MediaMTX WebRTC 页面，不代理媒体、不参与推理或事件生成。
-4. Bell 做 schema 与代码级断言，生成平台 ULID，保存不可变事件。T-015 已实现内部 candidate→final event factory、append-only PostgreSQL repository 和独立 outcome 事实；Brain→Bell transport、认证和公共 API 仍未冻结。
+3. Brain 消费视频与触发信号，产生符合冻结契约的事件候选。T-017 已建立单路 frame source、可替换 detector、轻量 track、多边形进入判定和回环可视化工程原型；T-019 把候选先写入仓库外 SQLite Outbox，再经 producer-bound HMAC 内部 HTTP 投递。合成 fixture 与 OpenCV HOG 均不是生产模型，Brain 只生成 `source_event_id`，不自报平台 `id`。T-018 的 Sense 回环控制台不参与推理或事件生成。
+4. Bell 做身份/Area 隐私、schema 与代码级断言，生成平台 ULID，保存不可变事件。T-015 已实现内部 candidate→final event factory、append-only PostgreSQL repository 和独立 outcome 事实；T-019 增加数字事件身份到 Bell/Sense 逻辑身份的受控绑定、永久来源收据和短期 nonce 收据。该 ingress 是内部适配器，Bell 公共 API/JWT/OIDC 仍未冻结。
 5. 规则命中后创建独立 Alert，先落库再投递，等待 ack 并按策略升级。
 6. Bell 发起 pre-roll 证据回捞，Sense 提供切片接口。
 7. 用户标记 outcome，反馈进入 Brain 的数据闭环。
 
 首个 M3 数据流部署在 S2 民办寄宿学校的 16 路高风险点位，只运行越线、危险区域和聚集等匿名规则，不加载人脸底库。
 
-## 5. 十二条不可越界的决定
+## 5. 十三条不可越界的决定
 
 1. MediaMTX 独立运行，Sense 管配置与生命周期。
 2. 设备型触发源归 Sense；需要解码的像素级触发归 Brain。
@@ -69,6 +69,7 @@ Sense ── 视频流/触发信号 ──> Brain
 10. 设备领域模型使用 `modality + capabilities`，页面不以摄像头作为唯一根实体；未实现协议适配器明确为 `adapter_not_ready`，不得用模拟遥测伪装交付。
 11. Tenant/Site/Area/RBAC、配额、`capture_policy` 与全局审计属于 Bell；Sense Control API v1 只管理 Device 期望态与收敛查询，Sense 只读消费版本化投影并在设备写路径执行，投影不可用时只阻断相关新变更，不静默切断已有链路。
 12. Sense 的设备操作审计先写本地持久化 Outbox，再由幂等 relay 异步送入 Bell 全局审计；不得使用“先执行高风险操作、再尽力入队”的顺序。T-016 已实现 HMAC/nonce 内部 HTTP relay、数据库时钟 lease/fencing、逐项确认与 Bell 不可变全局事实；Sense 不获得 Bell schema 权限，Bell 不读取 Sense Outbox。
+13. Brain 的业务事件同样先写仓库外持久 Outbox，再异步投递；Bell 以 `(producer_id, source_event_id)` 永久收据而非短期 nonce 保证跨重启幂等。只有 `accepted/duplicate` 可确认本地 delivered；事件 ID 始终由 Bell 生成，稳定冲突不得自动改写来源 ID 或 payload。
 
 ## 6. 容量架构
 
@@ -93,8 +94,8 @@ T-014 已在单台 Windows 主机上用隔离 PostgreSQL、真实 Control API、
 - 设备创建和期望态受理在本地事务内同时写脱敏 `sense.device_operation_outbox`；Outbox 失败回滚业务写入，相同期望态不增加 generation 但仍审计。relay 最多领取 100 行，以 30 秒数据库 lease 和单调 fencing token 防止过期 worker 确认；成功和 dead letter 都保留本地事实。
 - Bell 先验证时间窗、nonce 和 constant-time HMAC，再逐项校验 v1/v2 事件；同 nonce/同摘要重放原结果，同 nonce/不同摘要拒绝。`bell.audit_events` 只追加且不自动清理，只有 10 分钟幂等收据允许 Bell runtime 删除过期行。
 - Bell 最终事件写入 `bell.events`；同平台 ID/同摘要仅视为幂等重放，同 ID/不同摘要拒绝。`bell_runtime` 只有 `SELECT/INSERT`，事件与 outcome 的 UPDATE/DELETE 另由数据库 trigger 拒绝；后续人工/自动 outcome 追加到独立表，不改写事件 payload。
-- Brain 投递失败落本地队列重试，不阻塞实时推理主链路。
-- T-017 的 100 项内存事件环只服务单路工程演示，重启可丢失且不等同于上述投递队列；Brain→Bell 后续建议 T-019 必须另行实现有界持久 Outbox、身份映射、认证和幂等确认，不能把 demo 内存状态升级为生产 transport。T-018 先实现 Sense 回环 NVR 管理面纵切，不改变该事件链边界。
+- T-019 的 Brain SQLite Outbox 与推理内存环分离：最多 10,000 条待投递，网络/5xx/认证故障按 1～300 秒重试，最多 100 次；稳定 4xx 进入 dead letter。Outbox 写入失败时不得在演示状态中伪装为已排队或已投递。
+- Bell 先按 HMAC key 绑定 producer，再解析候选；永久来源收据、最终 event 和成功 nonce 响应同事务提交。同来源/同 canonical hash 返回原 Bell ID，同来源/不同 hash 返回冲突。T-017 的 100 项内存环仍只服务页面显示，不承担可靠投递。
 - Alert 先落库再投递，进程重启恢复未完成升级链。
 - 值班排班发布前必须按 Site 时区校验班次空档、重叠、联系人停用和通道验证；排班以新版本和未来生效时间发布，不原地改写历史。交接班是进行中 Alert 的显式责任转移事件，不替代排班版本变更。
 - 事件证据技术默认保留 30 天并按生命周期删除；客户/法务在 M3 生产上线前确认法规适用性和最终期限，技术默认值不能覆盖其结论。
@@ -102,7 +103,7 @@ T-014 已在单台 Windows 主机上用隔离 PostgreSQL、真实 Control API、
 ## 8. 数据与契约
 
 - Bell 核心实体：Tenant → Site → Area（含 `capture_policy`）以及 Role/Binding/Quota/Audit；Sense 核心实体：Device（含 `modality + capabilities`）→ StreamBinding/Zone，以及只记录已观察版本的 SiteQuota/AreaPolicyProjection。两个 schema 以稳定逻辑 ID 关联，不跨 schema 写入；配额 v1 为五列，Area v1 固定为 `tenant_id/site_id/area_id/capture_policy/source_version/source_updated_at` 六列。
-- Sense Control API v1 使用站点作用域路径、认证上下文 tenant、HMAC cursor 分页、PostgreSQL 幂等收据与资源 ETag；敏感连接引用只写不读。T-011 已实现 7 个 handler，并以 feature flag 限定到 PostgreSQL 路径；首版外部静态 SHA-256 注册表只实现认证 port 的私有部署适配器。T-016 审计 relay 采用独立外部 HMAC key 文件和内部端点，不等同于 Bell 公共管理认证。正式签名和兼容规则以 [`contracts/`](contracts/) 为准。
+- Sense Control API v1 使用站点作用域路径、认证上下文 tenant、HMAC cursor 分页、PostgreSQL 幂等收据与资源 ETag；敏感连接引用只写不读。T-011 已实现 7 个 handler，并以 feature flag 限定到 PostgreSQL 路径。T-016 审计 relay 与 T-019 Brain 事件 ingress 各用独立外部 HMAC key/端点，均不等同于 Bell 公共管理认证。正式签名和兼容规则以 [`contracts/`](contracts/) 为准。
 - 业务实体：Rule → Event → Alert → DeliveryAttempt/Ack；Event 与 Alert 不合并。
 - Bell 通知域分为三个聚合：Contact/Team 保存身份、成员关系和已验证通道；OnCallSchedule/ScheduleVersion/ShiftException 保存时区、轮换与例外；EscalationPolicy/Step 通过 `person / team / on_call_schedule` 类型化 `target_ref` 引用目标。三者共享逻辑 ID，不复制手机号、班次或轮换字段。
 - 每个 DeliveryAttempt 创建时解析当时生效的排班版本，并保存实际收件人、通道、`schedule_version` 和解析时间快照；之后联系人或排班修改不得回写既有投递事实。
@@ -118,17 +119,17 @@ Sense/cmd + Sense/internal/{device,onvif,mtx,reconcile,orphan,metrics,probe,trig
 Brain/{pipeline,models,judge,emit,trigger,contracts}
 Bell/cmd + Bell/internal/{ingest,event,rule,alert,deliver,feedback,tenant,audit,store}
 Bell/{web,packs,contracts}
-deploy/postgres/{001_roles.sql,...,015_privileges_audit_relay.sql,tests}
+deploy/postgres/{001_roles.sql,...,017_privileges_event_ingress.sql,tests}
 ```
 
-Sense 脚手架和 PostgreSQL `001`～`015` 已实现；Bell 已有事件校验/不可变存储 Go 基础和只面向 Sense 审计 relay 的最小 `bell-api`，但没有公共管理 API。Brain 已有 T-017 单路工程原型，包括合成/RTSP source、HOG 演示 detector、track、zone entry、回环服务与自包含 UI；尚无生产模型、GPU pipeline、持久 Outbox 或 Bell ingress。
+Sense 脚手架和 PostgreSQL `001`～`017` 已实现；Bell 已有事件校验/不可变存储、Sense 审计 relay 与默认关闭的 Brain 事件 ingress，但没有公共管理 API。Brain 已有单路工程原型和仓库外 SQLite 可靠事件 Outbox；尚无生产模型、GPU pipeline、证据切片或规则/Alert 链。
 
 ## 10. 开发顺序
 
 - M0 不写生产代码。
 - M1 只动 Sense，以 1 路 T-001 准入实机 + 至少 4 路独立合成 RTSP 源完成五路接入骨架与 MediaMTX；设备模型从此时起保持模态/能力可扩展，但不提前实现非视频适配器。真实多设备现场门禁移到 T-007，阻塞生产试点但不阻塞本地开发。
 - M2 仍以 Sense 为主；Control API、对账、多租户投影和本地 16 路开通/停用基线已完成，隧道等待客户网络条件后补验。
-- M3 Brain 与 Bell 同时起步，事件契约首次被真实使用；按项目负责人调整，T-017 后先补 Sense 回环 NVR 管理面纵切，再继续 Brain→Bell 事件 ingress。
+- M3 Brain 与 Bell 同时起步；T-015～T-019 已打通不可变事件消费者、审计 relay、Brain 单路候选、Sense 回环管理纵切和 Brain→Bell 可靠事件 ingress。下一步仍应在证据切片与规则/Alert 之间按客户可见闭环拆分独立任务。
 - M4/M5 再做 64/128 路分片、完整管理端和多个场景包；M6 接入雷达、门磁、按钮和可穿戴等非视频适配器。
 
 M3 先执行不少于 2 周的 dry-run，冻结现场标注集，按规则报告召回率和每路每天误报数；现场基线评审后才把数值阈值写入站点验收附件。算法效果指标与系统 SLA 分开验收。

docs/api.md

@@ -1,6 +1,6 @@
 # API 与契约
 
-> Brain → Bell 事件契约 v0.1、Sense Control API v1、Bell 配额/Area 只读投影 v1、Sense 本地设备审计事件 v1/v2 与 Sense→Bell 审计 relay v1 已冻结；其他 API 仍在设计阶段。不得把本文的“待定”自行具体化为公共契约。
+> Brain → Bell 事件契约 v0.1 与内部 ingress v1、Sense Control API v1、Bell 配额/Area 只读投影 v1、Sense 本地设备审计事件 v1/v2 与 Sense→Bell 审计 relay v1 已冻结；其他 API 仍在设计阶段。不得把本文的“待定”自行具体化为公共契约。
 
 ## 1. 已冻结：Brain → Bell 事件契约
 
@@ -17,7 +17,7 @@
 - 证据文件名只含事件 ID 与日期目录，不含 IP、端口、凭据或客户名。
 - `sensors` 中恰有一个 primary，且其 `device_id` 与顶层一致。
 
-T-015 已实现 Bell 消费端的内部组装与存储边界：可信 ingress 先接收“不含平台 `id`”的候选事实，Bell 生成 `evt_` ULID 后形成最终 v0.1 对象，再执行 schema 与六项代码断言并不可变落库。该候选类型是 Bell 内部 port，不是 Brain 可依赖的 HTTP/消息总线协议；transport、认证和重放语义仍由后续任务冻结。
+T-015 已实现 Bell 消费端的内部组装与存储边界；T-019 在 [`contracts/brain-event-ingress-v1.openapi.json`](contracts/brain-event-ingress-v1.openapi.json) 冻结并实现网络适配器。Brain 发送“不含平台 `id`”的完整候选，Bell 生成 `evt_` ULID 后执行 schema、六项代码断言与不可变落库。内部 endpoint 使用独立 producer-bound HMAC key，不是 Bell 公共 API 或 JWT/OIDC 的替代品。
 
 ## 2. 跨系统接口状态
 
@@ -26,12 +26,13 @@ T-015 已实现 Bell 消费端的内部组装与存储边界：可信 ingress
 | Sense → Bell | 读取站点视频配额 | 同一 PostgreSQL 实例内只读 `bell.site_quota_v1`；默认 16、最大 128；失败时拒绝新增/启用但不影响已有流 | T-008 冻结，T-009 已实现数据路径 |
 | Sense → Bell | 读取 Area 成像准入 | 只读 `bell.area_policy_v1`；`video_allowed | non_imaging_only`；缺失/非法/回退失败关闭但不影响已有设备 | T-010 已冻结并实现数据路径 |
 | Sense → Bell | 汇入设备操作审计 | `POST /internal/v1/audit-events:batch`；1～100 项、1 MiB、10 秒 deadline、HMAC/nonce、逐项确认；非回环必须 HTTPS | T-016 已冻结并实现 |
+| Brain → Bell | 汇入业务事件候选 | `POST /internal/v1/event-candidates`；单条完整 v0.1-minus-id candidate、1 MiB、10 秒 deadline、producer-bound HMAC、永久来源收据；非回环必须 HTTPS | T-019 已冻结并实现 |
 | Bell → Sense | 请求事件证据/pre-roll 切片 | 幂等、按租户授权、异步结果、不得暴露原始凭据 | 待 M3 设计 |
 | Bell → Brain | outcome/误报反馈 | 原事件不可变；反馈可重试、去重、审计 | 待 M3 设计 |
 | Sense → Brain | 流绑定与设备型触发 | 分片可路由，触发入口与流控制解耦 | 待 M2/M3 设计 |
 | Worker → 控制面 | 注册、心跳、容量 | `max_sources` 来自 profile/压测，不固定为 16 | 待 M3 设计 |
 
-冻结签名和失败语义见 [`contracts/README.md`](contracts/README.md)、[`contracts/sense-audit-relay-v1.openapi.json`](contracts/sense-audit-relay-v1.openapi.json)、[`contracts/site-quota-v1.sql`](contracts/site-quota-v1.sql)、[`contracts/area-policy-v1.sql`](contracts/area-policy-v1.sql) 与 [`contracts/sense-device-audit-v1.schema.json`](contracts/sense-device-audit-v1.schema.json)。Bell 拥有投影源数据和视图，Sense 数据库角色只有 `SELECT`；未来分库必须发布新版本，不能在 v1 下把本地视图静默替换为网络调用。
+冻结签名和失败语义见 [`contracts/README.md`](contracts/README.md)、[`contracts/brain-event-ingress-v1.openapi.json`](contracts/brain-event-ingress-v1.openapi.json)、[`contracts/sense-audit-relay-v1.openapi.json`](contracts/sense-audit-relay-v1.openapi.json)、[`contracts/site-quota-v1.sql`](contracts/site-quota-v1.sql)、[`contracts/area-policy-v1.sql`](contracts/area-policy-v1.sql) 与 [`contracts/sense-device-audit-v1.schema.json`](contracts/sense-device-audit-v1.schema.json)。Bell 拥有投影源数据、视图、事件身份绑定与来源收据；未来分库必须发布新版本，不能在 v1 下静默改变一致性或身份语义。
 
 ## 3. 已冻结：Sense Control API v1
 

docs/contracts/README.md

@@ -1,6 +1,6 @@
-# Sense 控制面、准入投影与本地审计契约 v1
+# YoVision 控制面、审计与内部事件传输契约 v1
 
-> 冻结日期：2026-08-11。Control API 与审计 relay 契约版本：`1.0.0`。Sense 是设备期望态的提供方；Bell 是 Tenant、Site、Area、RBAC、配额与全局审计的所有者。T-016 已实现 Sense Outbox 到 Bell 的内部 relay；Bell 公共管理服务与 JWT/OIDC 仍未实现。
+> 冻结日期：2026-08-11。Control API、审计 relay 与 Brain 事件 ingress 契约版本：`1.0.0`。Sense 是设备期望态的提供方；Bell 是 Tenant、Site、Area、RBAC、配额、事件与全局审计的所有者。T-016 已实现 Sense Outbox 审计 relay，T-019 已实现 Brain 事件 Outbox ingress；Bell 公共管理服务与 JWT/OIDC 仍未实现。
 
 ## 契约文件
 
@@ -12,6 +12,7 @@
 | [`sense-device-audit-v1.schema.json`](sense-device-audit-v1.schema.json) | Sense | 本地 Outbox；Bell relay | 脱敏设备操作审计事实，不包含传输协议 |
 | [`sense-device-audit-v2.schema.json`](sense-device-audit-v2.schema.json) | Sense | 本地 Outbox；Bell relay | v1 后继，增加脱敏配置修改受理事实；v1 文件保持不变 |
 | [`sense-audit-relay-v1.openapi.json`](sense-audit-relay-v1.openapi.json) | Bell | Sense | 内部批量端点、HMAC、逐项确认、nonce 防重与重试边界 |
+| [`brain-event-ingress-v1.openapi.json`](brain-event-ingress-v1.openapi.json) | Bell | Brain | 单业务事件入站、producer/key 绑定、Bell ID 与跨重启幂等语义 |
 
 OpenAPI 的 `/api/v1` 路径是公共控制面边界；`/healthz`、`/readyz` 仍是非业务运维探针。v1 不提供设备删除：停用设备使用期望态接口，保留设备、操作和审计历史。Site、Area、配额、RBAC 和审计聚合不由 Sense 提供 CRUD。
 
@@ -72,6 +73,20 @@ Outbox 用 30 秒数据库时钟 lease、单调 fencing token 和 `FOR UPDATE SK
 {"version":1,"keys":[{"key_id":"sense-a","secret_base64url":"<external-secret>"}]}
 ```
 
+## Brain → Bell 业务事件 ingress
+
+Brain 向 `POST /internal/v1/event-candidates` 单条投递完整 event v0.1 candidate；candidate 与冻结最终事件的顶层字段完全一致，只省略 Bell 所有的 `id`。envelope 额外携带 `schema_version=1` 和 `producer_id`。Bell 校验 HMAC、key 与 producer 的绑定、当前逻辑设备/Area 映射、event schema 和六项语义断言后生成 `evt_` ULID；首次接受返回 `201 accepted`，同一来源事实重投返回 `200 duplicate` 和原 Bell ID。
+
+认证 canonical string 与审计 relay 使用相同五行算法，但 key 文件独立且每项增加 `producer_id`，不得混用审计 key。正文最多 1 MiB、deadline 10 秒、允许 300 秒时钟偏差，nonce 收据至少保留 600 秒；HTTP 只允许回环地址，非回环必须 HTTPS。key 文件形态为：
+
+```json
+{"version":1,"keys":[{"key_id":"brain-a","producer_id":"brain-main","secret_base64url":"<external-secret>"}]}
+```
+
+Bell 的永久 `event_ingress_receipts` 以 `(producer_id, source_event_id)` 唯一并保存 canonical candidate SHA-256。相同 hash 重投返回原 event ID，不同 hash 返回 `409 source_event_conflict`；事件、来源收据和成功 nonce 响应在一个 PostgreSQL 事务中提交。`event_ingress_nonces` 只是 10 分钟防重表，可清理；来源收据和事件不自动删除、不可更新。Brain 必须先把 candidate 写入仓库外 SQLite Outbox，只有 `accepted/duplicate` 可标记 delivered；401、网络和 5xx 以 1～300 秒退避重试，稳定 4xx 进入 dead letter，最多 100 次、最多 10,000 条待投递。
+
+数字 `tenant_id/site_id/device_id` 通过 Bell 所有的 `event_ingress_bindings` 映射到当前 Bell Site/Area 和 Sense Device 逻辑 ID。运行时只读取 Sense 设备的 ID、Area 与 modality 列，不获得 endpoint、credential、profile token 或 path；绑定缺失/禁用、设备或 Area 不一致、删除、非视频或 `capture_policy != video_allowed` 均失败关闭。首版绑定只由受控 migration/admin SQL 配置，没有公共 CRUD。
+
 ## 兼容与废弃
 
 - v1 可增加不改变已有语义的可选响应字段和新错误细节；客户端必须忽略未知响应字段。
@@ -87,9 +102,11 @@ Outbox 用 30 秒数据库时钟 lease、单调 fencing token 和 `FOR UPDATE SK
 python -m json.tool docs/contracts/sense-control-v1.openapi.json | Out-Null
 python -m json.tool docs/contracts/sense-device-audit-v2.schema.json | Out-Null
 python -m json.tool docs/contracts/sense-audit-relay-v1.openapi.json | Out-Null
+python -m json.tool docs/contracts/brain-event-ingress-v1.openapi.json | Out-Null
 python -m unittest discover -s tests -p "test_sense_control_contract.py"
 python -m unittest discover -s tests -p "test_sense_control_implementation.py"
 python -m unittest discover -s tests -p "test_sense_audit_relay_contract.py"
+python -m unittest discover -s tests -p "test_brain_event_ingress_contract.py"
 ```
 
 测试同时校验 OpenAPI 结构、生成 server glue、HTTP handler 与 PostgreSQL migration/事务；它不替代 Bell 消费方联合验收或客户现场容量验证。

docs/contracts/brain-event-ingress-v1.openapi.json

@@ -0,0 +1,147 @@
+{
+  "openapi": "3.1.0",
+  "info": {
+    "title": "YoVision Brain Event Ingress",
+    "version": "1.0.0",
+    "description": "Internal single-event transport. The candidate is the frozen event v0.1 object with only the Bell-owned id omitted."
+  },
+  "paths": {
+    "/internal/v1/event-candidates": {
+      "post": {
+        "operationId": "ingestBrainEventCandidate",
+        "summary": "Validate and persist one Brain event candidate",
+        "parameters": [
+          {"$ref": "#/components/parameters/KeyId"},
+          {"$ref": "#/components/parameters/Timestamp"},
+          {"$ref": "#/components/parameters/Nonce"},
+          {"$ref": "#/components/parameters/Signature"}
+        ],
+        "requestBody": {
+          "required": true,
+          "content": {
+            "application/json": {
+              "schema": {"$ref": "#/components/schemas/IngressRequest"}
+            }
+          }
+        },
+        "responses": {
+          "200": {"description": "The same producer/source candidate was already stored", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/IngressResponse"}}}},
+          "201": {"description": "Candidate accepted and stored", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/IngressResponse"}}}},
+          "400": {"$ref": "#/components/responses/Error"},
+          "401": {"$ref": "#/components/responses/Error"},
+          "403": {"$ref": "#/components/responses/Error"},
+          "409": {"$ref": "#/components/responses/Error"},
+          "413": {"$ref": "#/components/responses/Error"},
+          "422": {"$ref": "#/components/responses/Error"},
+          "503": {"$ref": "#/components/responses/Error"}
+        }
+      }
+    }
+  },
+  "components": {
+    "parameters": {
+      "KeyId": {"name": "X-YoVision-Key-Id", "in": "header", "required": true, "schema": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$"}},
+      "Timestamp": {"name": "X-YoVision-Timestamp", "in": "header", "required": true, "schema": {"type": "string", "pattern": "^[0-9]{10,}$"}},
+      "Nonce": {"name": "X-YoVision-Nonce", "in": "header", "required": true, "schema": {"type": "string", "minLength": 22, "maxLength": 64}},
+      "Signature": {"name": "X-YoVision-Signature", "in": "header", "required": true, "schema": {"type": "string", "minLength": 43, "maxLength": 43}}
+    },
+    "schemas": {
+      "IngressRequest": {
+        "type": "object",
+        "additionalProperties": false,
+        "required": ["schema_version", "producer_id", "candidate"],
+        "properties": {
+          "schema_version": {"const": 1},
+          "producer_id": {"type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$"},
+          "candidate": {"$ref": "#/components/schemas/EventV01Candidate"}
+        }
+      },
+      "EventV01Candidate": {
+        "description": "Exact docs/raw/contracts/event-v0.1.schema.json event object with the required Bell-owned id field removed; nested shapes and semantic assertions remain authoritative in that frozen contract.",
+        "type": "object",
+        "additionalProperties": false,
+        "required": [
+          "schema_version", "source_event_id", "tenant_id", "site_id", "device_id", "sensors",
+          "kind", "severity", "confidence", "occurred_at", "detected_at", "latency_seconds",
+          "config_version", "rule", "subject", "observation", "evidence", "dedup_key",
+          "aggregated_into", "outcome", "outcome_source", "outcome_reason", "diagnostics", "ext"
+        ],
+        "properties": {
+          "schema_version": {"const": "0.1"},
+          "source_event_id": {"type": "string", "pattern": "^[A-Za-z0-9_-]{1,128}$"},
+          "tenant_id": {"type": "integer", "minimum": 1},
+          "site_id": {"type": "integer", "minimum": 1},
+          "device_id": {"type": "integer", "minimum": 1},
+          "sensors": {"type": "array", "minItems": 1, "items": {"type": "object"}},
+          "kind": {"type": "string", "pattern": "^[a-z][a-z0-9_]{2,63}$"},
+          "severity": {"type": "string", "enum": ["low", "medium", "high", "critical"]},
+          "confidence": {"type": ["number", "null"], "minimum": 0, "maximum": 1},
+          "occurred_at": {"type": "string", "format": "date-time"},
+          "detected_at": {"type": "string", "format": "date-time"},
+          "latency_seconds": {"type": "number", "minimum": 0},
+          "config_version": {"type": "string", "minLength": 1, "maxLength": 128},
+          "rule": {"type": ["object", "null"]},
+          "subject": {"type": "object"},
+          "observation": {"type": ["object", "null"]},
+          "evidence": {"type": "object"},
+          "dedup_key": {"type": ["string", "null"]},
+          "aggregated_into": {"type": ["string", "null"]},
+          "outcome": {"type": "string"},
+          "outcome_source": {"type": ["string", "null"]},
+          "outcome_reason": {"type": ["string", "null"]},
+          "diagnostics": {"type": ["object", "null"]},
+          "ext": {"type": "object"}
+        }
+      },
+      "IngressResponse": {
+        "type": "object",
+        "additionalProperties": false,
+        "required": ["schema_version", "producer_id", "source_event_id", "event_id", "status"],
+        "properties": {
+          "schema_version": {"const": 1},
+          "producer_id": {"type": "string"},
+          "source_event_id": {"type": "string"},
+          "event_id": {"type": "string", "pattern": "^evt_[0-9A-HJKMNP-TV-Z]{26}$"},
+          "status": {"type": "string", "enum": ["accepted", "duplicate"]}
+        }
+      },
+      "Error": {
+        "type": "object",
+        "additionalProperties": false,
+        "required": ["error"],
+        "properties": {
+          "error": {"type": "string"},
+          "detail_code": {"type": "string"}
+        }
+      }
+    },
+    "responses": {
+      "Error": {"description": "Stable machine-readable failure", "content": {"application/json": {"schema": {"$ref": "#/components/schemas/Error"}}}}
+    }
+  },
+  "x-yovision-final-event-schema": "docs/raw/contracts/event-v0.1.schema.json",
+  "x-yovision-signature": {
+    "algorithm": "HMAC-SHA256",
+    "canonical_lines": ["method", "escaped_path", "unix_timestamp_seconds", "base64url_nonce", "lowercase_hex_sha256_body"],
+    "clock_skew_seconds": 300,
+    "nonce_receipt_ttl_seconds": 600,
+    "minimum_secret_bytes": 32,
+    "key_document_shape": {"version": 1, "keys": [{"key_id": "brain-a", "producer_id": "brain-main", "secret_base64url": "external-secret"}]}
+  },
+  "x-yovision-delivery": {
+    "maximum_body_bytes": 1048576,
+    "deadline_seconds": 10,
+    "initial_retry_seconds": 1,
+    "maximum_retry_seconds": 300,
+    "maximum_attempts": 100,
+    "maximum_queued": 10000,
+    "success_statuses": ["accepted", "duplicate"],
+    "retryable_http": [401, 500, 502, 503, 504]
+  },
+  "x-yovision-idempotency": {
+    "source_scope": ["producer_id", "source_event_id"],
+    "same_candidate": "return duplicate with the original Bell event_id",
+    "different_candidate": "409 source_event_conflict",
+    "bell_owns_event_id": true
+  }
+}

docs/current-state.md

@@ -4,28 +4,29 @@
 
 ## 当前阶段
 
-- 阶段：M0 指定摄像头型号准入、M1“一实机 + 四合成源”软件闭环和 M2 本地 16 路批量收敛/稳定基线已通过；M3 已建立 Bell 不可变事件存储、Sense→Bell 全局审计 relay 及 Brain 单路匿名区域事件工程原型。客户网络尚未提供，WireGuard T-013 后置，五条独立真实上游和生产 SLA 仍未验收。
-- 生产代码：Sense 已包含可构建进程、SQLite/PostgreSQL repository、Site/Area 准入、设备操作 Outbox、可选签名 relay、标准 ONVIF SOAP/WS-Security adapter、凭据引用、MediaMTX 生成客户端、Control API v1、对账/探活、数据库租约、孤儿只读扫描/受控命令、低基数指标和可重复 16 路容量脚本；Bell 已包含事件 v0.1 校验/不可变存储、append-only outcome，以及只服务 Sense 审计的最小 `bell-api` 和全局审计 repository。Brain 已包含单路 source/detector/track/zone-entry 流水线和回环可视化页，但仍没有生产模型/GPU pipeline、Brain 事件 ingress、公共管理服务/JWT、规则/Alert 或正式 Web/H5。
+- 阶段：M0 指定摄像头型号准入、M1“一实机 + 四合成源”软件闭环和 M2 本地 16 路批量收敛/稳定基线已通过；M3 已建立 Bell 不可变事件存储、Sense→Bell 全局审计 relay、Brain 单路匿名区域事件工程原型及 Brain→Bell 可靠事件 ingress。客户网络尚未提供，WireGuard T-013 后置，五条独立真实上游和生产 SLA 仍未验收。
+- 生产代码：Sense 已包含可构建进程、SQLite/PostgreSQL repository、Site/Area 准入、设备操作 Outbox、可选签名 relay、标准 ONVIF SOAP/WS-Security adapter、凭据引用、MediaMTX 生成客户端、Control API v1、对账/探活、数据库租约、孤儿只读扫描/受控命令、低基数指标和可重复 16 路容量脚本；Bell 已包含事件 v0.1 校验/不可变存储、append-only outcome、Sense 审计 relay 和默认关闭的 Brain 事件 ingress。Brain 已包含单路 source/detector/track/zone-entry、回环可视化页与仓库外 SQLite 事件 Outbox，但仍没有生产模型/GPU pipeline、公共管理服务/JWT、规则/Alert、证据切片或正式 Web/H5。
 - 默认容量：16 路；单站点本阶段上限 128 路，必须横向分片。
 
 ## 仓库现实
 
-- `Sense/` 已有 Go module 与 `cmd/sense-api`；`Bell/` 已有事件域 Go module 和最小 `cmd/bell-api` 内部审计 receiver；`Brain/` 已有 Python 3.10 单路工程原型、19 项单元/HTTP/UI 契约测试和 `docs/design/brain/index.html` 自包含页面。Bell/Sense migration 统一位于根目录 `deploy/postgres/`。
+- `Sense/` 已有 Go module 与 `cmd/sense-api`；`Bell/` 已有事件域 Go module 和最小 `cmd/bell-api` 两条内部 ingress；`Brain/` 已有 Python 3.10 单路工程原型、30 项单元/HTTP/UI/Outbox 测试和 `docs/design/brain/index.html` 自包含页面。Bell/Sense migration 统一位于根目录 `deploy/postgres/`。
 - Sense 设备模型使用 `modality + capabilities`，SQLite 执行 v1 migration；视频配额默认 16、允许 1～128，17/128/129、新增/启用和“降低配额不关闭已有流”均有测试。
 - T-009 冻结 PostgreSQL `17.10` 和 `pgx/v5 v5.10.0`，实现 `bell`/`sense` schema、NOLOGIN 权限角色、Bell Site 版本 trigger、`bell.site_quota_v1` 和 Sense PostgreSQL repository；同站点并发准入用事务级 advisory lock，配额缺失/越界/版本回退时失败关闭且不改变已有流。
 - T-010 增量实现 `bell.areas`、`bell.area_policy_v1`、Area 版本观察和 `sense.device_operation_outbox`；`non_imaging_only` 拒绝成像设备创建/启用，失败不改变已有设备。设备创建/期望态受理与脱敏 Outbox 同事务，相同期望态不增加 generation 但仍审计。
-- Windows 隔离测试使用 `D:\pgsql17\bin` 启动随机回环端口临时集群，`001`～`015` migration 可重放；Sense Outbox fencing、Bell event/global-audit repository、nonce 收据、权限、幂等冲突与不可变性测试通过后自动清理，现有 `D:\pgsql17\data` 和 5432 服务未被读取、停止或修改。
-- T-015 冻结 Bell Go 1.26.5、JSON Schema v6.0.2 和 ULID v2.1.2；Bell 拒绝上游自报平台 ID，在内部 candidate 组装后执行冻结 v0.1 schema 与六项语义断言。`bell_runtime` 只允许追加/读取 `bell.events`、`bell.event_outcomes`；Brain transport、整数事件 ID 与现有文本逻辑 ID 的跨系统映射、公共 API 和生产隐私 resolver 仍未冻结，不能把内部 factory 当成已上线入口。
+- Windows 隔离测试使用 `D:\pgsql17\bin` 启动随机回环端口临时集群，`001`～`017` migration 可重放；Sense Outbox fencing、Bell event/global-audit/Brain-ingress repository、nonce/永久来源收据、权限、并发幂等冲突与不可变性测试通过后自动清理，现有 `D:\pgsql17\data` 和 5432 服务未被读取、停止或修改。
+- T-015 冻结 Bell Go 1.26.5、JSON Schema v6.0.2 和 ULID v2.1.2；Bell 拒绝上游自报平台 ID，在内部 candidate 组装后执行冻结 v0.1 schema 与六项语义断言。T-019 已为该 factory 增加内部网络入口和生产隐私 resolver；公共 API、JWT/OIDC 仍未冻结。
 - T-016 冻结 `sense-audit-relay-v1`：每批 1～100 项、1 MiB、10 秒 deadline、300 秒时钟窗、600 秒 nonce 收据、30 秒数据库 lease、1～300 秒退避。Sense 使用 `FOR UPDATE SKIP LOCKED` 和 fencing token；Bell constant-time 校验 HMAC，逐项返回 accepted/duplicate/rejected，并把全局事实追加到不可变 `bell.audit_events`。relay 默认关闭，非回环两端必须 HTTPS/TLS，key 只从仓库外文件读取。
 - T-017 冻结的只是工程原型：Python 3.10.11、NumPy 1.26.4、OpenCV 4.9.0.80；默认 2 FPS 合成 fixture，可选从仓库外文件读取 MediaMTX/RTSP。合成框不是模型输出，HOG/SVM 不是生产 detector，100 项内存事件环不是可靠投递；候选不含 Bell 平台 ID。
 - T-018 Sense 回环工程控制台已通过项目负责人产品验收：默认关闭并强制 Sense/MediaMTX 播放端显式回环，直接读取 Control API v1 的设备、配额、期望态/实际态/收敛事实；设备 cursor 每页 16 项，只有成像、启用、在线且已收敛的设备可选，最多 4 路按需嵌入 MediaMTX WebRTC 页面。token 只驻留页面内存，刷新即丢失；该验收不包含录像/回放、非回环生产认证、真实 16 机或生产 SLA。
+- T-019 冻结 `brain-event-ingress-v1`：单事件 1 MiB/10 秒、producer-bound HMAC、300 秒时钟窗、600 秒 nonce、SQLite WAL Outbox、1～300 秒退避、100 次/10,000 条边界。Bell 以受控绑定复查当前 Site/Area/Sense Device，永久 `(producer_id,source_event_id)` 收据与事件同事务提交；同 candidate 跨重启返回原 Bell ID，不同 candidate 稳定冲突。
 - MediaMTX 固定为独立二进制 `v1.19.3`，官方 OpenAPI 已按 SHA-256 vendoring，并由固定 `oapi-codegen v2.8.0` 生成客户端；手写薄封装有 create/read/delete、幂等 ensure、探活和只返回名称的受限分页枚举测试。
 - T-003 对账进度与指数退避持久化，覆盖取消和 SQLite 重启恢复；T-006 增加真实 ONVIF adapter、RTSP router、实验室播种/状态工具、故障代理和五路自动验收。T-012 的普通调和不枚举孤儿；独立 PostgreSQL 扫描默认只报告，未知归属永不删除。
 - T-006 正式使用 1 台准入实机和 4 个独立合成 publisher 连续观察 `1806.6 s` / 180 次采样，四类恢复均通过，最大与最终 `unconverged` 均为 0；详细证据见 `docs/research/sense-5-stream-integration.md`。
 - T-014 正式使用隔离 PostgreSQL、真实 Control API、两套 MediaMTX 和 16 个独立低码率合成 publisher，完成 17 路配额拒绝、三轮 `16 → 0 → 16` 批量收敛和固定四路故障恢复；稳定观察 `1800.1 s` / 180 次采样，最大与最终 `unconverged` 均为 0、最终在线 Path 16、帧错误 0。证据见 `docs/research/sense-16-stream-capacity.md`；不外推到真实 16 机、网络、录像、AI/GPU、64/128 路或生产 SLA。
 - `docs/raw/01`～`08` 已记录需求、分析、方案、客户场景、事件比对和三系统职责。
 - `docs/raw/contracts/event-v0.1.schema.json` 已冻结，并有多份示例与语义说明。
-- `docs/contracts/sense-control-v1.openapi.json` 的 7 个站点作用域/operation endpoint 已由 T-011 实现；`sense-audit-relay-v1.openapi.json` 已由 T-016 实现。默认 SQLite 只暴露运维探针与低基数 `/metrics`，不注册业务路由或 relay；Bell 管理服务与 JWT/OIDC 尚未实现。
+- `docs/contracts/sense-control-v1.openapi.json` 的 7 个站点作用域/operation endpoint 已由 T-011 实现；`sense-audit-relay-v1.openapi.json` 和 `brain-event-ingress-v1.openapi.json` 已分别由 T-016/T-019 实现。两条内部 ingress 使用独立 key；Bell 管理服务与 JWT/OIDC 尚未实现。
 - T-012 把 PostgreSQL schema 提升到 v5：due row 用数据库时钟、`FOR UPDATE SKIP LOCKED`、逐项续租和 fencing token 协调；MediaMTX Path 历史归属、15 分钟孤儿快照、最多 10%/128 项安全闸、无 bypass 的本地处置命令及 `/metrics` 已实现。SQLite 明确保留单实例开发语义。
 - harness coding 文档、上下文清单、Gitea Issue/PR 模板和治理脚本已接入。
 - Gitea 已初始化 12 个协作标签；`status/waiting` 用于依赖或外部条件未满足的未领取任务，实时可领取状态必须从 Gitea 查询，不在本文复制。
@@ -47,7 +48,7 @@ Windows：
 go -C Sense run ./cmd/sense-api
 ```
 
-Bell 事件域和内部审计 receiver 验证：
+Bell 事件域与两条内部 ingress 验证：
 
 ```powershell
 go -C Bell test ./...
@@ -61,6 +62,7 @@ Brain 单路工程原型验证与启动：
 python -m unittest discover -s Brain/tests -p "test_*.py" -v
 python -m compileall -q Brain
 python -m Brain.yovision_brain --source synthetic
+# 可选：追加 --event-ingress-config <仓库外绝对路径>
 ```
 
 跨平台直接验证：
@@ -97,12 +99,12 @@ Sense 默认监听 `127.0.0.1:8080`，提供 `/healthz`、`/readyz` 运维探针
 - 人脸方向已延后至 M5 的 S4 成人园区候选试点；必要性/PIP 影响评估、单独同意与替代方式、合法底库来源和删除流程未完成，阻塞人脸能力上线。
 - 短信/语音具体供应商未选；生产前必须选定两条独立投递路径并验证故障切换。
 - 生产 Brain 的 Python/Savant/DeepStream 精确版本、目标硬件和 Bell 前端栈尚未冻结；T-017 的 Python/OpenCV 只适用于工程原型，不能外推为生产选择。Sense M1 的 Go、SQLite driver、MediaMTX、生成器及生成运行时版本已在 T-003 冻结，PostgreSQL/pgx 版本已在 T-009 冻结。
-- 本机现有 PostgreSQL 5432 实例使用 SCRAM 且当前开发进程没有管理员密码；T-009～T-016 不绕过认证，自动验收使用隔离临时集群。向共享/生产实例安装 migration 前仍需管理员私下提供专用数据库、最小权限登录角色、外部 Control API/relay key 文件、TLS 证书与备份方案。
+- 本机现有 PostgreSQL 5432 实例使用 SCRAM 且当前开发进程没有管理员密码；T-009～T-019 不绕过认证，自动验收使用隔离临时集群。向共享/生产实例安装 migration 前仍需管理员私下提供专用数据库、最小权限登录角色、外部 Control API/两类 ingress key、Brain Outbox 路径、TLS 证书与备份方案。
 - 代码知识图谱在无业务代码阶段可能为空；工具不可用时使用 `rg` 处理文档与配置。
 
 ## 下一步
 
-客户网络仍未提供，T-013 WireGuard 继续后置。T-018 Sense NVR 管理面已通过产品验收；当前执行 T-019，独立冻结并实现 Brain→Bell 业务事件身份、持久重试、认证和幂等 ingress。客户授权、借用或租赁条件具备后再执行 T-007 五条独立真实上游现场门禁；上述本地任务均不解除 T-007/T-013，也不形成真实多路、模型效果或生产 SLA 承诺。
+客户网络仍未提供，T-013 WireGuard 继续后置。T-019 已完成本地 Brain→Bell 可靠业务事件 ingress；下一项建议独立建立 Bell 规则→Alert→ack 的最小可见纵切，证据/pre-roll 切片另立任务且不与 Alert 状态机混交。客户授权、借用或租赁条件具备后再执行 T-007 五条独立真实上游现场门禁；上述本地任务均不解除 T-007/T-013，也不形成真实多路、模型效果或生产 SLA 承诺。
 
 ## 已知风险
 

docs/design/brain/index.html

@@ -90,11 +90,23 @@
     .form-actions { display: grid; grid-template-columns: 1fr 1fr; gap: 8px; }
     .inline-state { min-height: 42px; padding: 9px 10px; border-left: 3px solid var(--cyan); background: rgba(56, 189, 248, .07); color: var(--muted); font-size: 12px; }
     .event-list { max-height: 330px; padding: 8px; display: grid; gap: 7px; overflow-y: auto; }
+    .ingress-summary { padding: 11px 12px; display: grid; gap: 8px; border-bottom: 1px solid var(--line); background: rgba(56, 189, 248, .04); }
+    .ingress-head { display: flex; align-items: center; justify-content: space-between; gap: 10px; }
+    .ingress-head h3 { margin: 0; font-size: 13px; }
+    .ingress-state { min-height: 28px; padding: 3px 9px; display: inline-flex; align-items: center; gap: 7px; border: 1px solid var(--line); border-radius: 999px; color: var(--muted); font-size: 12px; font-weight: 750; }
+    .status-dot.error { background: var(--danger); box-shadow: 0 0 0 4px rgba(251, 113, 133, .12); }
+    .ingress-copy { margin: 0; color: var(--muted); font-size: 12px; overflow-wrap: anywhere; }
+    .ingress-counts { margin: 0; display: grid; grid-template-columns: repeat(4, minmax(0, 1fr)); gap: 6px; }
+    .ingress-counts[hidden] { display: none; }
+    .ingress-counts div { min-width: 0; padding: 7px 8px; border: 1px solid var(--line); border-radius: 8px; background: var(--surface-2); }
+    .ingress-counts dt { color: var(--muted); font-size: 11px; }
+    .ingress-counts dd { margin: 2px 0 0; color: var(--text); font-size: 15px; font-weight: 800; font-variant-numeric: tabular-nums; }
     .event { padding: 11px; border: 1px solid var(--line); border-radius: 10px; background: var(--surface-2); }
     .event-head { display: flex; align-items: center; justify-content: space-between; gap: 8px; }
     .event strong { font-size: 13px; }
     .event time, .event code { color: var(--muted); font-size: 12px; }
     .event code { display: block; margin-top: 6px; overflow-wrap: anywhere; }
+    .event-delivery { margin-top: 5px; color: var(--muted); font-size: 12px; }
     .event-tag { padding: 2px 7px; border: 1px solid rgba(251, 191, 36, .35); border-radius: 999px; background: rgba(251, 191, 36, .08); color: #ffe18a; font-size: 11px; }
     .empty { min-height: 120px; padding: 26px 16px; display: grid; place-items: center; color: var(--muted); text-align: center; }
     .architecture-note { margin-top: 14px; padding: 13px 15px; display: grid; grid-template-columns: auto 1fr; gap: 11px; border: 1px solid rgba(139, 124, 246, .30); border-radius: 12px; background: rgba(139, 124, 246, .07); color: var(--muted); }
@@ -127,6 +139,8 @@
       .metrics { grid-template-columns: 1fr 1fr; }
       .metric strong { font-size: 17px; }
       .form-actions { grid-template-columns: 1fr; }
+      .ingress-head { align-items: flex-start; flex-direction: column; }
+      .ingress-counts { grid-template-columns: 1fr 1fr; }
     }
     @media (prefers-reduced-motion: reduce) {
       *, *::before, *::after { scroll-behavior: auto !important; transition-duration: .01ms !important; animation-duration: .01ms !important; animation-iteration-count: 1 !important; }
@@ -155,7 +169,7 @@
       <div class="page-head">
         <div>
           <h1>单路检测与区域判定</h1>
-          <p>验证从画面、匿名人员框、track 到区域进入候选事实的工程闭环。平台事件 ID、持久投递和 Alert 由后续 Bell 链路负责。</p>
+          <p>验证从画面、匿名人员框、track 到区域进入候选事实的工程闭环。Brain 先持久化到 Outbox，再由 Bell 分配平台事件 ID；Alert 仍由后续 Bell 链路负责。</p>
         </div>
         <div id="mode-badge" class="mode-badge">加载运行状态</div>
       </div>
@@ -220,9 +234,22 @@
 
           <section class="card" aria-labelledby="event-title">
             <div class="card-head">
-              <div><h2 id="event-title">最近候选事实</h2><p>只产生 source_event_id；Bell 才生成平台 ULID</p></div>
+              <div><h2 id="event-title">最近候选事实</h2><p>内存环仅用于观察；可靠性以 Outbox 状态为准</p></div>
               <span id="event-count" class="chip"><strong>0</strong> 项</span>
             </div>
+            <section id="event-ingress" class="ingress-summary" aria-labelledby="event-ingress-title" role="status" aria-live="polite" aria-atomic="true">
+              <div class="ingress-head">
+                <h3 id="event-ingress-title">Bell 可靠投递</h3>
+                <span class="ingress-state"><i id="ingress-dot" class="status-dot" aria-hidden="true"></i><span id="ingress-label">未启用</span></span>
+              </div>
+              <p id="ingress-copy" class="ingress-copy">可靠投递未启用；候选事实不会被标记为 Bell 已接收。</p>
+              <dl id="ingress-counts" class="ingress-counts" aria-label="Outbox 投递计数" hidden>
+                <div><dt>待投递</dt><dd id="ingress-queued">0</dd></div>
+                <div><dt>投递中</dt><dd id="ingress-delivering">0</dd></div>
+                <div><dt>已送达</dt><dd id="ingress-delivered">0</dd></div>
+                <div><dt>死信</dt><dd id="ingress-dead-letter">0</dd></div>
+              </dl>
+            </section>
             <div id="event-list" class="event-list" aria-live="polite">
               <div class="empty">人员从区域外进入后，这里会出现一次候选事实。</div>
             </div>
@@ -252,6 +279,7 @@
       let state = null;
       let offlineSequence = 0;
       let offlineTriggered = false;
+      let ingressSignature = "";
 
       const offlineState = () => {
         offlineSequence += 1;
@@ -275,6 +303,7 @@
           zone: state && state.zone ? state.zone : {id: "zone-demo-01", name: "楼梯口危险区", version: 1, points: parsePoints(false)},
           detections: bbox[2] - bbox[0] > .01 ? [{track_id: "P-DEMO-001", class: "person", bbox, detector_score: null, inside_zone: inside}] : [],
           events,
+          event_ingress: {enabled: false},
           last_error_code: null
         };
       };
@@ -376,11 +405,48 @@
           const tag = document.createElement("span"); tag.className = "event-tag"; tag.textContent = event.fixture ? "夹具候选" : "检测候选";
           head.append(title, tag);
           const time = document.createElement("time"); time.dateTime = event.occurred_at; time.textContent = `${formatTime(event.occurred_at)} · ${event.kind}`;
+          const delivery = document.createElement("div"); delivery.className = "event-delivery"; delivery.textContent = event.delivery_status === "outbox_persisted" ? "交接：Outbox 已持久化" : "交接：仅内存观察";
           const code = document.createElement("code"); code.textContent = event.source_event_id;
-          article.append(head, time, code); return article;
+          article.append(head, time, delivery, code); return article;
         }));
       }
 
+      function renderIngress(value = {enabled: false}) {
+        const enabled = value.enabled === true;
+        const queued = Number(value.queued) || 0;
+        const delivering = Number(value.delivering) || 0;
+        const delivered = Number(value.delivered) || 0;
+        const deadLetter = Number(value.dead_letter) || 0;
+        const errorCode = typeof value.last_error_code === "string" ? value.last_error_code : "";
+        const nextSignature = JSON.stringify([enabled, queued, delivering, delivered, deadLetter, errorCode]);
+        if (nextSignature === ingressSignature) return;
+        ingressSignature = nextSignature;
+        const dot = $("ingress-dot");
+        const counts = $("ingress-counts");
+        counts.hidden = !enabled;
+        $("ingress-queued").textContent = String(queued);
+        $("ingress-delivering").textContent = String(delivering);
+        $("ingress-delivered").textContent = String(delivered);
+        $("ingress-dead-letter").textContent = String(deadLetter);
+        if (!enabled) {
+          dot.className = "status-dot";
+          $("ingress-label").textContent = "未启用";
+          $("ingress-copy").textContent = "可靠投递未启用；候选事实不会被标记为 Bell 已接收。";
+        } else if (deadLetter > 0) {
+          dot.className = "status-dot error";
+          $("ingress-label").textContent = "需要处理";
+          $("ingress-copy").textContent = `存在死信${errorCode ? `（${errorCode}）` : ""}；请检查 Bell 绑定、鉴权或候选字段后再处理。`;
+        } else if (errorCode || queued > 0 || delivering > 0) {
+          dot.className = "status-dot warning";
+          $("ingress-label").textContent = errorCode ? "正在重试" : "投递中";
+          $("ingress-copy").textContent = errorCode ? `Outbox 已保留候选并自动重试（${errorCode}）。` : "候选已持久化到本地 Outbox，正在等待 Bell 确认。";
+        } else {
+          dot.className = "status-dot online";
+          $("ingress-label").textContent = "投递正常";
+          $("ingress-copy").textContent = "Outbox 可用；Bell 返回 accepted 或 duplicate 后才计为已送达。";
+        }
+      }
+
       function render(next) {
         state = next;
         $("connection-dot").className = `status-dot ${next.source.connected ? "online" : "warning"}`;
@@ -398,6 +464,7 @@
         if (document.activeElement !== $("zone-name") && document.activeElement !== $("zone-points")) {
           $("zone-name").value = next.zone.name; setPoints(next.zone.points);
         }
+        renderIngress(next.event_ingress);
         renderEvents(next.events);
         drawOverlay();
       }

docs/raw/contracts/README.md

@@ -47,6 +47,7 @@
 | kind | 含义 | 默认 severity | 产出方 | 状态 |
 | --- | --- | --- | --- | --- |
 | `fall` | 人员摔倒确认 | `high` | silver_pose FSM 进入 CONFIRMED | ✅ 已上线 |
+| `zone_entry` | 匿名 track 从区域外进入区域内 | `medium` | Brain 区域进入判定 | ✅ T-019 ingress 已接入 |
 
 ## 4. silver_pose → 契约 映射表
 

docs/tasks/T-019.md

@@ -3,16 +3,18 @@ id: T-019
 title: 建立 Brain 到 Bell 的可靠事件入站链路
 phase: 3
 deps: [T-015, T-017]
-status: TODO
+status: DONE
 created: 2026-08-11
-issue: null
-context_ref: null
-claim_branch: null
-work_branch: null
+issue: 67
+context_ref: b7fe44eeb0caf8d41abef365ec2100c74373321f
+claim_branch: claims/T-019
+work_branch: agent/codex/T-019
 write_paths:
   - docs/tasks/T-019.md
   - docs/contracts/
+  - docs/raw/contracts/README.md
   - Brain/
+  - docs/design/brain/index.html
   - Bell/
   - deploy/postgres/
   - scripts/test_postgres.ps1
@@ -41,6 +43,7 @@ T-017 已能产生匿名区域事件候选，T-015 已能在 Bell 内部校验
 ## 方案
 
 1. 在 `docs/contracts/brain-event-ingress-v1.openapi.json` 冻结单事件 envelope、响应、稳定错误码和 HMAC 规则。请求包含 `schema_version=1`、`producer_id` 与不含 Bell `id` 的完整 event v0.1 candidate；Brain 提供 `source_event_id`，Bell 生成并返回平台 `evt_` ULID。正文上限 1 MiB，deadline 10 秒。
+   同步把 T-017 已产生的 `zone_entry` 登记到 event v0.1 kind 注册表；不修改冻结 JSON Schema 或既有 kind 语义。
 2. 复用 T-016 已验证的 HMAC-SHA256 canonical 形式：method、path、Unix 秒、随机 nonce、body SHA-256 以换行连接；使用独立的仓库外 key 文件，并把每个 key 绑定到一个 `producer_id`。允许 300 秒时钟偏差，nonce 防重收据保留 600 秒。回环可用 HTTP，非回环必须 HTTPS；事件 key 与审计 relay key 不混用。
 3. 新增 Bell 所有的 `event_ingress_bindings`，把 `(producer_id, tenant_id, site_id, device_id)` 数字事件身份绑定到现有 Bell tenant/site/device/area 逻辑身份和 `video` 模态。入站必须 fail closed：绑定缺失、禁用、站点/Area 已删除、设备或 Area 归属不一致、非视频或 `capture_policy != video_allowed` 均拒绝。首版只提供受控 SQL 配置方式，不新增公共绑定管理 API。
 4. 新增永久 append-only `event_ingress_receipts`，以 `(producer_id, source_event_id)` 唯一，保存 canonical candidate SHA-256 与 Bell event ID；另用短期 `event_ingress_nonces` 防请求重放。Bell 用单事务和事务级 advisory lock 完成 nonce、来源收据、事件创建与响应：同来源且同 payload 返回原 Bell ID 和 `duplicate`，同来源不同 payload 返回 `source_event_conflict`，不得产生第二条事件。
@@ -83,3 +86,20 @@ T-017 已能产生匿名区域事件候选，T-015 已能在 Bell 内部校验
 
 - T-019 独立冻结 Brain→Bell 业务事件身份、持久重试、认证和幂等 ingress；不复用 T-017 内存事件环，也不扩张到规则、Alert 或证据链。
 - 当前会话未暴露 codebase-memory MCP 图工具，代码发现按仓库规则降级为定向读取与 `rg`；任务定义前主分支 `./init.ps1` 基线通过：72 项根测试、19 项 Brain 测试，以及 Sense/Bell generate/test/vet/build 全绿。
+- 任务定义已合入默认分支并创建唯一 Gitea Issue #67；本映射提交合入后才添加 `status/todo` 并允许 dispatcher 分配。
+
+### 2026-08-11 领取与基线
+
+- dispatcher `ila` 检查依赖和开放工单后，从默认分支 `b7fe44eeb0caf8d41abef365ec2100c74373321f` 创建并读回 `claims/T-019` 与 `agent/codex/T-019`；Issue #67 已分配给 `ila`，标签为 `status/doing`，结构化 CLAIM 评论与本文件 `write_paths` 一致。
+- T-007 保持 `status/waiting`，不构成活跃写路径预留；本任务不启用委派。
+
+### 2026-08-11 实现与自动化验收
+
+- 冻结并实现 `brain-event-ingress-v1` 单事件 envelope。Bell event key 独立于审计 key，并把 `key_id` 绑定到唯一 `producer_id`；两端共享 method/path/timestamp/nonce/body-hash 五行 HMAC-SHA256，允许 300 秒时钟偏差，正文上限 1 MiB，非回环必须 HTTPS。Brain client 禁用环境代理和重定向，避免内部事件被旁路转发。
+- PostgreSQL 新增可重放 `016`～`017`：Bell 所有的数字→逻辑身份绑定引用现有 Site/Area/Sense Device；运行时只能读取设备 ID、Area 与 modality 列，不能读取 endpoint、credential、profile token 或 path。永久来源收据和事件不可更新/删除，只有 10 分钟 nonce 表允许清理。
+- Bell 在相同事务中提交最终 event、`(producer_id,source_event_id,candidate_hash)` 永久收据和成功 nonce 响应；同来源/同 canonical candidate 返回原 Bell ID，不同 candidate 返回稳定冲突。隔离 PostgreSQL 测试以 8 个并发重投确认只生成 1 条 event/1 条来源收据，并验证 Area 改为 `non_imaging_only` 后新视频事件失败关闭。
+- Brain 新增完整 v0.1-minus-id mapper、WAL/`synchronous=FULL` SQLite Outbox、HMAC client 和 worker。Outbox 首次打开绑定 producer/tenant/site/device，禁止换身份复用旧队列；候选持久化成功后才进入页面内存环。fixture 显式映射为 `outcome=test/outcome_source=auto`，URI/凭据/`source_ref` 不出 Brain。队列执行 1～300 秒退避、100 次/10,000 条边界，`accepted/duplicate` 才 delivered，稳定 4xx 进入 dead letter；租约崩溃恢复与 Bell 成功后本地未 ack 的重投均有测试。
+- Brain 工程原型在既有“最近候选事实”卡片内展示脱敏 Outbox 状态与待投递/投递中/已送达/死信计数；以文字和状态点共同区分未启用、正常、重试和死信，且明确单条候选的“仅内存观察”或“Outbox 已持久化”。状态无变化时不重复触发 `aria-live`，页面不暴露 Bell 地址、key 文件或 SQLite 路径。
+- `zone_entry` 已登记到 event v0.1 kind 注册表；冻结 JSON Schema 未修改。Bell/Brain/部署 README、API、架构、技术栈和当前状态均同步，未扩张到证据、规则、Alert、公共认证或生产模型。
+- 最终门禁通过：`./init.ps1`（77 项根测试、30 项 Brain 测试及 Sense/Bell generate/test/vet/build）；独立 `go -C Bell test/vet/build ./...`、Brain compileall、4 项跨语言 ingress 契约测试和三项治理校验；`./scripts/test_postgres.ps1` 在 PostgreSQL 17.10 临时集群将 `001`～`017` 重放两次并通过真实 repository、权限、并发幂等、隐私和不可变性验证，随机端口/临时目录已清理且现有 5432 listener 未改变；`git diff --check` 通过。
+- 本任务不需要真实摄像头或目标 GPU；synthetic fixture mapper、SQLite 崩溃恢复、真实本地 HTTP HMAC client、Bell handler 和真实 PostgreSQL consumer 的联合自动化证据满足本地验收。结果不构成算法效果、真实多路或生产 SLA。

docs/tasks/T-020.md

@@ -0,0 +1,88 @@
+---
+id: T-020
+title: 建立 Bell 规则到 Alert 与 ack 的最小可见纵切
+phase: 3
+deps: [T-015, T-019]
+status: TODO
+created: 2026-08-11
+issue: 71
+context_ref: null
+claim_branch: null
+work_branch: null
+write_paths:
+  - docs/tasks/T-020.md
+  - Bell/
+  - deploy/postgres/
+  - scripts/test_postgres.ps1
+  - tests/test_postgres_contract.py
+  - tests/test_bell_alert_contract.py
+  - docs/contracts/
+  - docs/00-ai-start-here.md
+  - docs/03-tech-stack.md
+  - docs/04-architecture.md
+  - docs/06-tasks.md
+  - docs/api.md
+  - docs/contracts/README.md
+  - docs/current-state.md
+  - docs/routes.md
+---
+
+## 问题 / 背景
+
+T-019 已把 Brain 候选可靠地写成 Bell 不可变 Event，但 Bell 尚未消费 Event、记录规则版本、创建独立 Alert 或接受值班员 ack。现有 Bell 原型能说明目标交互，却没有可运行数据/API；因此客户仍看不到从匿名区域事件到“有人确认处理”的业务闭环。
+
+本任务建立默认关闭、只允许回环联调的最小可见纵切：Bell 从不可变 Event 运行一个受限规则内核，为匹配事件幂等创建 Alert，使用 append-only 状态迁移实现首次 ack 竞争者获胜与确认后关闭，并通过自包含值班台展示真实 PostgreSQL 状态。它为后续升级链、双路径通知、证据切片和正式公共认证提供基础，但不提前伪造这些事实。
+
+## 关联需求与交互（如适用）
+
+- 用户故事：US-003（处置业务预警）、US-005（配置规则与升级链中的规则版本边界）。
+- 交互清单：IX-005（预警到达）、IX-006（ack 竞争）、IX-007（只展示实际已有的升级/投递事实）、IX-008（Event/Alert 双向关联与证据降级）、IX-013（权限与隐私）。
+- 相关页面 / 路由：复用已确认的 `docs/design/bell/index.html` 值班台信息架构；新增默认关闭的回环工程路由 `/bell-console/` 与 `/bell-console/api/v1/*`。这些路由不是待冻结的 Bell 公共 `/api/v1`，不冻结最终前端框架。
+
+## 方案
+
+1. 在 `docs/contracts/bell-alert-console-v1.openapi.json` 冻结回环工程 API：按状态分页列出 Alert、读取详情、`ack` 和 `close`。列表默认 16、最大 100，稳定按 `created_at DESC,id DESC`；tenant/Site/actor 只来自启动时的仓库外控制台上下文，不接受客户端自报。
+2. PostgreSQL 新增 append-only `rule_versions`、`event_rule_sweeps`、`rule_evaluations`、`alerts`、`alert_events`、`alert_transitions` 和 `alert_command_receipts`。Rule version、evaluation、Alert 身份、Event 关联、状态迁移与幂等收据均禁止 UPDATE/DELETE/TRUNCATE；当前状态由最新 transition 推导，不把 Event 和 Alert 合并。
+3. 首版规则内核只支持精确 `event_kind`、最小严重度、可选 Site 范围和生效时间。每个稳定 `rule_key` 的最新生效版本决定 enabled/disabled；配置从仓库外绝对 JSON 文件加载并按 canonical hash 幂等发布，不提供公共规则编辑 API，也不实现任意表达式或场景 DSL。
+4. 规则 worker 扫描尚无 sweep 的不可变 Event；在单个 PostgreSQL 事务内写 sweep、每条已考虑规则的 evaluation、匹配 Alert、初始 `open` transition 与 Event 关联。并发 worker 通过唯一键和事务级 advisory lock 收敛；无规则/未命中也写 durable sweep，避免无限重试。崩溃前未提交可重做，提交后不得生成重复 Alert。
+5. Alert ID 由 Bell 生成 `alt_` ULID。首版每个命中 Rule/Event 生成一个 Alert，但关系表按多对多建模；Alert 保存命中时的 rule version、标题与严重度快照，后续规则变化不改写历史。
+6. 状态机固定为 `open -> acknowledged -> closed`。`ack` 的首个成功事务固化 actor/时间；并发后到者返回 `409 already_acknowledged` 并带当前处置人和时间，不覆盖、不双成功。`close` 只允许从 acknowledged 进入 closed。写请求必须携带 8～128 字符 `Idempotency-Key`；同 key/同命令重放原响应，同 key/不同命令稳定冲突。
+7. `bell-api` 以 `BELL_ALERTS_ENABLED=true` 显式启用规则 worker；规则文件必须是仓库外绝对路径。值班台另以 `BELL_ALERT_CONSOLE_ENABLED=true` 启用，只允许整个 Bell HTTP 监听地址为显式回环，并要求仓库外 token 文件、正整数 tenant/Site 与受限 actor ref。两项默认关闭，控制台 token 只进入页面内存和 `no-store` 响应，不记录。
+8. `Bell/web/` 使用 Go `embed`、HTML/CSS/原生 JavaScript 实现，不增加前端依赖或 CDN。页面显示真实 Alert 队列、关联 Event、规则版本和 append-only 时间线；无证据时明确标注“证据切片尚未启用”，无升级/投递实现时不显示虚构倒计时或成功状态。ack/close 有 loading、成功、并发冲突、可重试错误与无权限反馈，颜色不是唯一状态表达；375px 与桌面均可操作。
+9. 增加 Go 域/handler/store 测试、静态 OpenAPI/UI 契约测试，以及隔离 PostgreSQL migration replay、最小权限、规则幂等、并发 evaluation、并发 ack、幂等重放和重启后状态恢复验证。当前会话无 codebase-memory 图工具，代码发现降级为定向读取和 `rg` 并记录在执行证据。
+
+## 不可变约束
+
+- 阈值 / 数值边界：Alert 列表默认 16、最大 100；16 不是业务上限，站点仍允许 1～128 路。规则文件和控制台 token 均限制大小；actor/rule/idempotency 字段有显式长度与字符集边界。worker 必须可取消且空闲轮询有界，不按设备或 tenant 创建 goroutine。
+- 判定式 / 状态转换：Rule 最新生效版本决定是否启用；严重度顺序固定为 `low < medium < high < critical`。每个 `(event_id, rule_version_id)` 最多一个 evaluation/Alert；`open -> acknowledged -> closed` 之外的迁移拒绝。首次 ack 获胜，后到者不得覆盖。Event、Rule version、Alert 事实和所有 transition 均 append-only。
+- 安全边界：功能默认关闭；工程控制台只允许显式回环绑定。token、规则配置、DSN 与客户数据均在仓库外；页面/API/日志不得返回 Event 原始 payload、流 URI、凭据、内部 DSN 或 token。tenant/Site/actor 不来自请求体、查询参数或可修改浏览器存储。
+- 既有契约：冻结 Event v0.1、T-019 ingress、Bell 事件 ULID/不可变性和 Sense/Bell schema 所有权保持不变。Alert 是独立实体并通过关系表关联 Event；本任务不修改 `docs/raw/contracts/event-v0.1.schema.json`。
+
+## 验收要点
+
+- 任务相关验证：`go -C Bell test ./...`、`go -C Bell vet ./...`、`go -C Bell build ./...`、`python -m unittest discover -s tests -p "test_bell_alert_contract.py" -v`、`python -m unittest discover -s tests -p "test_postgres_contract.py" -v`、`./scripts/test_postgres.ps1 -PgRoot D:\pgsql17`。
+- 完整门禁：运行 `./init.ps1`、三项治理基线、`git diff --check`。PostgreSQL migration 必须从空库执行两遍；8 个并发 evaluator 对同 Event/Rule 只创建 1 个 Alert，8 个并发 ack 只有 1 个成功且其余观察同一处置人；同幂等 key 重放稳定，重启后状态/时间线不丢失；运行角色不能改写或删除任一规则/Alert 历史事实。
+- 人工 / 设备验收：不需要摄像头、GPU、客户网络或通知供应商。页面沿用已由产品确认的 Bell 原型信息架构；任务责任人在本地以合成 Event 检查 375px/桌面、键盘、加载/空态/成功/冲突/错误和 reduced-motion。该工程验收不等于正式 Web/H5、JWT/RBAC、算法效果或生产 SLA 验收。
+- 构建产物：Bell `bell-api`、嵌入式 `/bell-console/`、OpenAPI 和 PostgreSQL migration；不提交运行时规则文件、token、数据库或事件样本。
+
+## 边界（不改什么）
+
+不实现证据/pre-roll 切片、MinIO/S3、升级计时/重启续跑、联系人/排班、通知 provider、短信/语音/Webhook、本地声光、静默、交接班、误报反馈、正式公共 JWT/OIDC/RBAC、正式 Bell 前端框架、规则编辑器/试运行/回滚、生产模型或多路容量压测；不修改 Sense、Brain、MediaMTX、`_reference/` 或冻结 Event v0.1。
+
+## 协作约束
+
+- 责任 Agent：codex。
+- 唯一写入者：codex。
+- 委派：不启用。
+- Gitea：任务定义先合入默认分支；创建唯一 Issue 并完成双向映射后，由 dispatcher 串行领取并回填 `context_ref`、claim / 工作分支。
+
+任何新增写路径先检查与其他活跃任务是否重叠；同一时刻只有一个 Agent 修改本任务的 `write_paths`。
+
+## 执行记录
+
+### 2026-08-11 任务定义
+
+- T-020 独立冻结规则扫描、Alert 身份/关系和首次 ack 竞争语义；证据、升级/投递和正式公共认证继续拆分，不用占位状态伪装完成。
+- 复用已确认 Bell 原型及 US-003/US-005、IX-005～IX-008/IX-013；工程值班台不冻结最终前端框架。
+- 当前会话未提供 codebase-memory MCP 图工具，按仓库规则降级为定向读取与 `rg`。任务定义前 `./init.ps1` 基线通过：77 项根测试、30 项 Brain 测试，以及 Sense/Bell generate/test/vet/build 全绿。
+- 任务定义已合入默认分支并创建唯一 Gitea Issue #71；本映射合入默认分支后才允许添加 `status/todo` 并由 dispatcher 分配。

scripts/test_postgres.ps1

@@ -93,7 +93,9 @@ try {
             '012_bell_events.sql',
             '013_privileges_bell_events.sql',
             '014_audit_relay.sql',
-            '015_privileges_audit_relay.sql'
+            '015_privileges_audit_relay.sql',
+            '016_event_ingress.sql',
+            '017_privileges_event_ingress.sql'
         )) {
             Invoke-Checked $psql '-X' '-v' 'ON_ERROR_STOP=1' '-d' $adminDatabaseDSN '-f' (Join-Path $repoRoot "deploy\postgres\$name")
         }

tests/test_brain_event_ingress_contract.py

@@ -0,0 +1,70 @@
+"""Cross-language invariants for Brain-to-Bell event ingress v1."""
+
+from __future__ import annotations
+
+import json
+import unittest
+from pathlib import Path
+
+
+ROOT = Path(__file__).resolve().parents[1]
+CONTRACT = ROOT / "docs" / "contracts" / "brain-event-ingress-v1.openapi.json"
+EVENT_SCHEMA = ROOT / "docs" / "raw" / "contracts" / "event-v0.1.schema.json"
+
+
+class BrainEventIngressContractTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.document = json.loads(CONTRACT.read_text(encoding="utf-8"))
+        cls.event_schema = json.loads(EVENT_SCHEMA.read_text(encoding="utf-8"))
+
+    def test_candidate_is_exact_final_top_level_shape_minus_bell_id(self) -> None:
+        candidate = self.document["components"]["schemas"]["EventV01Candidate"]
+        expected = set(self.event_schema["required"]) - {"id"}
+        self.assertEqual(expected, set(candidate["required"]))
+        self.assertEqual(expected, set(candidate["properties"]))
+        self.assertFalse(candidate["additionalProperties"])
+        self.assertIn("id", self.event_schema["required"])
+
+    def test_transport_auth_and_statuses_are_frozen(self) -> None:
+        operation = self.document["paths"]["/internal/v1/event-candidates"]["post"]
+        self.assertEqual({"200", "201", "400", "401", "403", "409", "413", "422", "503"}, set(operation["responses"]))
+        header_names = {
+            self.document["components"]["parameters"][parameter["$ref"].split("/")[-1]]["name"]
+            for parameter in operation["parameters"]
+        }
+        self.assertEqual(
+            {"X-YoVision-Key-Id", "X-YoVision-Timestamp", "X-YoVision-Nonce", "X-YoVision-Signature"},
+            header_names,
+        )
+
+    def test_numeric_safety_bounds_match_both_implementations(self) -> None:
+        signature = self.document["x-yovision-signature"]
+        delivery = self.document["x-yovision-delivery"]
+        self.assertEqual((300, 600, 32), (signature["clock_skew_seconds"], signature["nonce_receipt_ttl_seconds"], signature["minimum_secret_bytes"]))
+        self.assertEqual((1048576, 10, 1, 300, 100, 10000), (
+            delivery["maximum_body_bytes"], delivery["deadline_seconds"], delivery["initial_retry_seconds"],
+            delivery["maximum_retry_seconds"], delivery["maximum_attempts"], delivery["maximum_queued"],
+        ))
+        brain = (ROOT / "Brain" / "yovision_brain" / "ingress.py").read_text(encoding="utf-8")
+        bell = (ROOT / "Bell" / "internal" / "ingress" / "ingress.go").read_text(encoding="utf-8")
+        for marker in (
+            "/internal/v1/event-candidates", "X-YoVision-Key-Id", "X-YoVision-Timestamp",
+            "X-YoVision-Nonce", "X-YoVision-Signature",
+        ):
+            self.assertIn(marker, brain)
+            self.assertIn(marker, bell)
+        self.assertIn("MAX_QUEUED = 10_000", brain)
+        self.assertIn("MAX_ATTEMPTS = 100", brain)
+        self.assertIn("300*time.Second", bell)
+
+    def test_event_key_is_bound_to_producer_and_separate_from_audit(self) -> None:
+        key_shape = self.document["x-yovision-signature"]["key_document_shape"]["keys"][0]
+        self.assertEqual({"key_id", "producer_id", "secret_base64url"}, set(key_shape))
+        bell_main = (ROOT / "Bell" / "cmd" / "bell-api" / "main.go").read_text(encoding="utf-8")
+        self.assertIn("BELL_EVENT_INGRESS_KEYS_FILE", bell_main)
+        self.assertIn("BELL_AUDIT_KEYS_FILE", bell_main)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/test_postgres_contract.py

@@ -62,6 +62,8 @@ class PostgresContractTests(unittest.TestCase):
                 "013_privileges_bell_events.sql",
                 "014_audit_relay.sql",
                 "015_privileges_audit_relay.sql",
+                "016_event_ingress.sql",
+                "017_privileges_event_ingress.sql",
             ],
             names,
         )
@@ -146,6 +148,28 @@ class PostgresContractTests(unittest.TestCase):
         self.assertIn("grant select, insert on table bell.audit_events to bell_runtime", privileges)
         self.assertIn("revoke all on table bell.audit_events, bell.audit_relay_receipts from sense_app", privileges)
 
+    def test_event_ingress_has_permanent_source_receipts_and_minimal_privileges(self) -> None:
+        migration = normalized(migration_text("016_event_ingress.sql"))
+        privileges = normalized(migration_text("017_privileges_event_ingress.sql"))
+        for marker in (
+            "create table if not exists bell.event_ingress_bindings",
+            "create table if not exists bell.event_ingress_receipts",
+            "primary key (producer_id, source_event_id)",
+            "create table if not exists bell.event_ingress_nonces",
+            "expires_at >= received_at + interval '10 minutes'",
+            "insert into bell.schema_migrations(version) values (5)",
+        ):
+            self.assertIn(marker, migration)
+        self.assertIn(
+            "grant select, insert on table bell.event_ingress_receipts to bell_runtime",
+            privileges,
+        )
+        self.assertIn(
+            "grant select, insert, delete on table bell.event_ingress_nonces to bell_runtime",
+            privileges,
+        )
+        self.assertNotIn("endpoint_ref", privileges.split("revoke all on table sense.devices")[1])
+
 
 if __name__ == "__main__":
     unittest.main()