Compare commits

..
Author SHA1 Message Date
QiuSW e81f00e9aa feat: 实现 Bell 联系人与值班排班 (#183) 2026-09-01 12:16:52 +08:00
ila 5adee5c3b4 merge: 同步三项目协调 E2E 文档 (#171)
用户于 2026-08-31 明确验收通过 #171。
2026-08-31 21:42:47 +08:00
QiuSW a69ef627c7 docs: 修正协调 E2E 清理边界措辞 (#171) 2026-08-31 21:12:33 +08:00
QiuSW d4de462d44 docs: 同步协调 E2E 运行与排障说明 (#171) 2026-08-31 21:08:32 +08:00
ila 0276bceab5 merge: 完成三项目协调隔离 E2E (#155)
用户已于 2026-08-31 明确验收通过 #155。
2026-08-31 20:49:20 +08:00
QiuSW c2b2943a3a test: 增加三项目协调隔离 E2E (#155) 2026-08-31 17:34:37 +08:00
ila 55b12df373 docs: 同步根级部署编排说明 (#168)
用户于 2026-08-31 明确验收通过 #168。
2026-08-31 16:57:34 +08:00
QiuSW 27d465c250 docs: 同步根级部署编排说明 (#168) 2026-08-31 16:32:48 +08:00
ila 4a2c4aa638 feat: 建立三项目可选部署编排 (#154)
用户于 2026-08-31 明确验收通过 #154。
2026-08-31 16:16:49 +08:00
QiuSW 504dd1a2e9 test: 补充三端独立编排验证 (#154) 2026-08-31 16:12:14 +08:00
QiuSW e2f7183ecf feat: 建立三项目可选部署编排 (#154) 2026-08-31 16:08:05 +08:00
ila eb4e1a9ea1 merge: 同步 connector 验收文档 (#152 #153)
同步已验收 #152/#153 的 Wiki 核心镜像。
2026-08-31 15:40:54 +08:00
QiuSW 9055f2522c docs: 记录 connector 验收 (#152 #153) 2026-08-31 15:36:19 +08:00
ila 130087a1ba merge: 完成可靠事件与证据 connector (#153)
用户已于 2026-08-31 验收通过 #153。
2026-08-31 15:16:18 +08:00
ila 0e53e04e95 merge: 完成 Sense 与 Brain connector (#152)
用户已于 2026-08-31 验收通过 #152。
2026-08-31 15:15:48 +08:00
QiuSW 82afce1f81 feat: 实现 Sense Brain 控制连接器 (#152) 2026-08-31 11:51:58 +08:00
59 changed files with 4912 additions and 53 deletions
+59
View File
@@ -0,0 +1,59 @@
package contact
import (
"crypto/aes"
"crypto/cipher"
"crypto/rand"
"crypto/sha256"
"encoding/base64"
"errors"
"io"
"strings"
)
var ErrChannelKeyUnavailable = errors.New("联系人通道加密密钥未配置或格式错误")
func ParseChannelKey(value string) ([]byte, error) {
key, err := base64.StdEncoding.DecodeString(strings.TrimSpace(value))
if err != nil || len(key) != 32 {
return nil, ErrChannelKeyUnavailable
}
return key, nil
}
func encryptAddress(key []byte, value string) ([]byte, error) {
block, err := aes.NewCipher(key)
if err != nil {
return nil, ErrChannelKeyUnavailable
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return nil, err
}
nonce := make([]byte, gcm.NonceSize())
if _, err = io.ReadFull(rand.Reader, nonce); err != nil {
return nil, err
}
return gcm.Seal(nonce, nonce, []byte(value), nil), nil
}
func decryptAddress(key, encoded []byte) (string, error) {
block, err := aes.NewCipher(key)
if err != nil {
return "", ErrChannelKeyUnavailable
}
gcm, err := cipher.NewGCM(block)
if err != nil {
return "", err
}
if len(encoded) < gcm.NonceSize() {
return "", errors.New("通道密文已损坏")
}
plain, err := gcm.Open(nil, encoded[:gcm.NonceSize()], encoded[gcm.NonceSize():], nil)
return string(plain), err
}
func fingerprint(value string) string {
sum := sha256.Sum256([]byte(value))
return base64.RawURLEncoding.EncodeToString(sum[:])
}
+146
View File
@@ -0,0 +1,146 @@
package contact
import (
"errors"
"net/http"
"os"
"github.com/gin-gonic/gin"
"github.com/gin-gonic/gin/binding"
"github.com/go-admin-team/go-admin-core/sdk/api"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth/user"
)
type Handler struct{ api.Api }
type enabledInput struct {
Enabled *bool `json:"enabled" binding:"required"`
ExpectedVersion int `json:"expectedVersion" binding:"required"`
}
type validationInput struct {
Status string `json:"status"`
Detail string `json:"detail"`
}
func (h Handler) service() (Service, error) {
key, err := ParseChannelKey(os.Getenv("BELL_CONTACT_CHANNEL_KEY"))
return NewService(h.Orm, key), err
}
func (h Handler) List(c *gin.Context) {
var q PageQuery
h.MakeContext(c).MakeOrm().Bind(&q, binding.Form)
if h.Errors != nil {
h.Error(400, ErrInvalid, ErrInvalid.Error())
return
}
items, count, err := NewService(h.Orm, nil).List(c.Request.Context(), q)
if err != nil {
h.Error(500, errors.New("读取联系人失败"), "读取联系人失败")
return
}
p, s := pageValues(q.PageIndex, q.PageSize)
h.PageOK(items, int(count), p, s, "查询成功")
}
func (h Handler) Create(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可维护联系人"), "仅管理员可维护联系人")
return
}
var input WriteInput
h.MakeContext(c).MakeOrm().Bind(&input, binding.JSON)
if h.Errors != nil {
h.Error(400, ErrInvalid, ErrInvalid.Error())
return
}
item, err := NewService(h.Orm, nil).Create(c.Request.Context(), input, user.GetUserId(c))
h.result(item, err)
}
func (h Handler) Update(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可维护联系人"), "仅管理员可维护联系人")
return
}
var input WriteInput
h.MakeContext(c).MakeOrm().Bind(&input, binding.JSON)
if h.Errors != nil {
h.Error(400, ErrInvalid, ErrInvalid.Error())
return
}
item, err := NewService(h.Orm, nil).Update(c.Request.Context(), c.Param("id"), input, user.GetUserId(c))
h.result(item, err)
}
func (h Handler) SetEnabled(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可维护联系人"), "仅管理员可维护联系人")
return
}
var input enabledInput
h.MakeContext(c).MakeOrm().Bind(&input, binding.JSON)
if h.Errors != nil || input.Enabled == nil {
h.Error(400, ErrInvalid, ErrInvalid.Error())
return
}
item, err := NewService(h.Orm, nil).SetEnabled(c.Request.Context(), c.Param("id"), *input.Enabled, input.ExpectedVersion, user.GetUserId(c))
h.result(item, err)
}
func (h Handler) AddChannel(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可维护通道"), "仅管理员可维护通道")
return
}
if err := restoreChannelBody(c); err != nil {
h.MakeContext(c).Error(http.StatusBadRequest, ErrInvalid, ErrInvalid.Error())
return
}
var input ChannelInput
h.MakeContext(c).MakeOrm().Bind(&input, binding.JSON)
if h.Errors != nil {
h.Error(400, ErrInvalid, ErrInvalid.Error())
return
}
service, keyErr := h.service()
if keyErr != nil {
h.Error(503, keyErr, keyErr.Error())
return
}
item, err := service.AddChannel(c.Request.Context(), c.Param("id"), input, user.GetUserId(c))
h.result(item, err)
}
func (h Handler) ValidateChannel(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可验证通道"), "仅管理员可验证通道")
return
}
var input validationInput
h.MakeContext(c).MakeOrm().Bind(&input, binding.JSON)
if h.Errors != nil {
h.Error(400, ErrInvalid, ErrInvalid.Error())
return
}
item, err := NewService(h.Orm, nil).RecordValidation(c.Request.Context(), c.Param("id"), input.Status, input.Detail, user.GetUserId(c))
h.result(item, err)
}
func (h Handler) result(item any, err error) {
switch {
case err == nil:
h.OK(item, "保存成功")
case errors.Is(err, ErrInvalid):
h.Error(400, err, err.Error())
case errors.Is(err, ErrNotFound):
h.Error(404, err, err.Error())
case errors.Is(err, ErrConflict):
h.Error(409, err, err.Error())
case errors.Is(err, ErrChannelKeyUnavailable):
h.Error(503, err, err.Error())
default:
h.Logger.Errorf("write Bell contact failed: %v", err)
h.Error(409, errors.New("联系人保存失败"), "联系人保存失败")
}
}
func admin(c *gin.Context) bool {
claims := jwt.ExtractClaims(c)
role, _ := claims[jwt.RoleKey].(string)
return role == "admin"
}
var _ = http.StatusOK
+68
View File
@@ -0,0 +1,68 @@
package contact
import (
"encoding/json"
"time"
)
type Contact struct {
ID string `json:"id" gorm:"type:uuid;primaryKey"`
Name string `json:"name" gorm:"size:128;not null"`
Role string `json:"role" gorm:"size:128;not null"`
Enabled bool `json:"enabled" gorm:"not null;default:true;index"`
Version int `json:"version" gorm:"not null;default:1"`
CreatedBy int `json:"createdBy" gorm:"not null"`
UpdatedBy int `json:"updatedBy" gorm:"not null"`
CreatedAt time.Time `json:"createdAt" gorm:"type:timestamptz;not null"`
UpdatedAt time.Time `json:"updatedAt" gorm:"type:timestamptz;not null"`
}
func (Contact) TableName() string { return "bell_contacts" }
type Channel struct {
ID string `json:"id" gorm:"type:uuid;primaryKey"`
ContactID string `json:"contactId" gorm:"type:uuid;not null;index"`
Kind string `json:"kind" gorm:"size:16;not null"`
AddressCiphertext []byte `json:"-" gorm:"type:bytea;not null"`
AddressFingerprint string `json:"-" gorm:"size:64;not null;index"`
AddressMasked string `json:"addressMasked" gorm:"size:64;not null"`
CreatedBy int `json:"createdBy" gorm:"not null"`
CreatedAt time.Time `json:"createdAt" gorm:"type:timestamptz;not null"`
}
func (Channel) TableName() string { return "bell_contact_channels" }
type ChannelValidation struct {
ID string `json:"id" gorm:"type:uuid;primaryKey"`
ChannelID string `json:"channelId" gorm:"type:uuid;not null;index"`
Status string `json:"status" gorm:"size:16;not null"`
Detail string `json:"detail" gorm:"size:256;not null"`
ActorID int `json:"actorId" gorm:"not null"`
CreatedAt time.Time `json:"createdAt" gorm:"type:timestamptz;not null;index"`
}
func (ChannelValidation) TableName() string { return "bell_contact_channel_validations" }
type AuditFact struct {
ID string `json:"id" gorm:"type:uuid;primaryKey"`
ContactID string `json:"contactId" gorm:"type:uuid;not null;index"`
Action string `json:"action" gorm:"size:32;not null"`
Snapshot json.RawMessage `json:"snapshot" gorm:"type:jsonb;not null"`
ActorID int `json:"actorId" gorm:"not null"`
CreatedAt time.Time `json:"createdAt" gorm:"type:timestamptz;not null"`
}
func (AuditFact) TableName() string { return "bell_contact_audit_facts" }
type ChannelView struct {
ID string `json:"id"`
Kind string `json:"kind"`
AddressMasked string `json:"addressMasked"`
Status string `json:"status"`
ValidatedAt *time.Time `json:"validatedAt,omitempty"`
}
type ContactView struct {
Contact
Channels []ChannelView `json:"channels"`
}
+62
View File
@@ -0,0 +1,62 @@
package contact
import (
"bytes"
"errors"
"io"
"net/http"
"strings"
"github.com/gin-gonic/gin"
)
const maxChannelRequestBytes = 8 * 1024
const channelBodyKey = "bell.contact.channel-body"
const channelBodyErrorKey = "bell.contact.channel-body-error"
var redactedChannelBody = []byte(`{"redacted":true}`)
// RedactRequestBody must run before GoAdmin's LoggerToFile middleware. The
// handler restores the original body from Gin context, while sys_opera_log
// only sees a fixed marker and never the contact address.
func RedactRequestBody() gin.HandlerFunc {
return func(c *gin.Context) {
if c.Request.Method != http.MethodPost || !isChannelCreatePath(c.Request.URL.Path) {
c.Next()
return
}
body, err := io.ReadAll(io.LimitReader(c.Request.Body, maxChannelRequestBytes+1))
if err != nil {
c.Set(channelBodyErrorKey, err)
} else if len(body) > maxChannelRequestBytes {
c.Set(channelBodyErrorKey, errors.New("request body too large"))
} else {
c.Set(channelBodyKey, body)
}
_ = c.Request.Body.Close()
c.Request.Body = io.NopCloser(bytes.NewReader(redactedChannelBody))
c.Next()
}
}
func restoreChannelBody(c *gin.Context) error {
if value, ok := c.Get(channelBodyErrorKey); ok {
return value.(error)
}
value, ok := c.Get(channelBodyKey)
if !ok {
return errors.New("channel request body was not captured")
}
c.Request.Body = io.NopCloser(bytes.NewReader(value.([]byte)))
return nil
}
func isChannelCreatePath(path string) bool {
const prefix = "/api/v1/bell/contacts/"
const suffix = "/channels"
if !strings.HasPrefix(path, prefix) || !strings.HasSuffix(path, suffix) {
return false
}
id := strings.TrimSuffix(strings.TrimPrefix(path, prefix), suffix)
return id != "" && !strings.Contains(id, "/")
}
+251
View File
@@ -0,0 +1,251 @@
package contact
import (
"context"
"encoding/json"
"errors"
"strings"
"time"
"github.com/google/uuid"
"gorm.io/gorm"
)
type PageQuery struct {
PageIndex int `form:"pageIndex"`
PageSize int `form:"pageSize"`
Name string `form:"name"`
Enabled *bool `form:"enabled"`
}
type Service struct {
DB *gorm.DB
Key []byte
}
func NewService(db *gorm.DB, key []byte) Service { return Service{DB: db, Key: key} }
func (s Service) List(ctx context.Context, query PageQuery) ([]ContactView, int64, error) {
page, size := pageValues(query.PageIndex, query.PageSize)
db := s.DB.WithContext(ctx).Model(&Contact{})
if name := strings.TrimSpace(query.Name); name != "" {
db = db.Where("name ILIKE ? OR role ILIKE ?", "%"+name+"%", "%"+name+"%")
}
if query.Enabled != nil {
db = db.Where("enabled = ?", *query.Enabled)
}
var count int64
if err := db.Count(&count).Error; err != nil {
return nil, 0, err
}
var contacts []Contact
if err := db.Order("created_at DESC,id DESC").Offset((page - 1) * size).Limit(size).Find(&contacts).Error; err != nil {
return nil, 0, err
}
views := make([]ContactView, 0, len(contacts))
for _, item := range contacts {
view, err := s.view(ctx, item)
if err != nil {
return nil, 0, err
}
views = append(views, view)
}
return views, count, nil
}
func (s Service) Create(ctx context.Context, input WriteInput, actor int) (ContactView, error) {
input, err := normalizeContact(input)
if err != nil {
return ContactView{}, err
}
now := time.Now().UTC()
item := Contact{ID: uuid.NewString(), Name: input.Name, Role: input.Role, Enabled: true, Version: 1, CreatedBy: actor, UpdatedBy: actor, CreatedAt: now, UpdatedAt: now}
err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Create(&item).Error; err != nil {
return err
}
return appendAudit(tx, item, "created", actor)
})
if err != nil {
return ContactView{}, err
}
return ContactView{Contact: item, Channels: []ChannelView{}}, nil
}
func (s Service) Update(ctx context.Context, id string, input WriteInput, actor int) (ContactView, error) {
if _, err := uuid.Parse(id); err != nil {
return ContactView{}, ErrNotFound
}
input, err := normalizeContact(input)
if err != nil {
return ContactView{}, err
}
if input.ExpectedVersion < 1 {
return ContactView{}, ErrInvalid
}
var item Contact
err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
result := tx.Model(&Contact{}).Where("id = ? AND version = ?", id, input.ExpectedVersion).Updates(map[string]any{"name": input.Name, "role": input.Role, "version": gorm.Expr("version + 1"), "updated_by": actor, "updated_at": time.Now().UTC()})
if result.Error != nil {
return result.Error
}
if result.RowsAffected != 1 {
var count int64
_ = tx.Model(&Contact{}).Where("id = ?", id).Count(&count).Error
if count == 0 {
return ErrNotFound
}
return ErrConflict
}
if err := tx.First(&item, "id = ?", id).Error; err != nil {
return err
}
return appendAudit(tx, item, "updated", actor)
})
if err != nil {
return ContactView{}, err
}
return s.view(ctx, item)
}
func (s Service) SetEnabled(ctx context.Context, id string, enabled bool, expectedVersion, actor int) (ContactView, error) {
if _, err := uuid.Parse(id); err != nil {
return ContactView{}, ErrNotFound
}
if expectedVersion < 1 {
return ContactView{}, ErrInvalid
}
var item Contact
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
result := tx.Model(&Contact{}).Where("id = ? AND version = ?", id, expectedVersion).Updates(map[string]any{"enabled": enabled, "version": gorm.Expr("version + 1"), "updated_by": actor, "updated_at": time.Now().UTC()})
if result.Error != nil {
return result.Error
}
if result.RowsAffected != 1 {
return ErrConflict
}
if err := tx.First(&item, "id = ?", id).Error; err != nil {
return err
}
return appendAudit(tx, item, map[bool]string{true: "enabled", false: "disabled"}[enabled], actor)
})
if err != nil {
return ContactView{}, err
}
return s.view(ctx, item)
}
func (s Service) AddChannel(ctx context.Context, contactID string, input ChannelInput, actor int) (ChannelView, error) {
if len(s.Key) != 32 {
return ChannelView{}, ErrChannelKeyUnavailable
}
if _, err := uuid.Parse(contactID); err != nil {
return ChannelView{}, ErrNotFound
}
input, err := normalizeChannel(input)
if err != nil {
return ChannelView{}, err
}
ciphertext, err := encryptAddress(s.Key, input.Address)
if err != nil {
return ChannelView{}, err
}
now := time.Now().UTC()
item := Channel{ID: uuid.NewString(), ContactID: contactID, Kind: input.Kind, AddressCiphertext: ciphertext, AddressFingerprint: fingerprint(input.Address), AddressMasked: maskAddress(input.Address), CreatedBy: actor, CreatedAt: now}
err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var contact Contact
if err := tx.First(&contact, "id = ?", contactID).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return ErrNotFound
}
return err
}
if err := tx.Create(&item).Error; err != nil {
return err
}
return appendAudit(tx, contact, "channel_added", actor)
})
if err != nil {
return ChannelView{}, err
}
return ChannelView{ID: item.ID, Kind: item.Kind, AddressMasked: item.AddressMasked, Status: "pending"}, nil
}
func (s Service) RecordValidation(ctx context.Context, channelID, status, detail string, actor int) (ChannelView, error) {
status = strings.ToLower(strings.TrimSpace(status))
detail = strings.TrimSpace(detail)
if status != "verified" && status != "failed" {
return ChannelView{}, ErrInvalid
}
if len([]rune(detail)) > 256 || hasControl(detail) {
return ChannelView{}, ErrInvalid
}
var channel Channel
fact := ChannelValidation{ID: uuid.NewString(), ChannelID: channelID, Status: status, Detail: detail, ActorID: actor, CreatedAt: time.Now().UTC()}
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.First(&channel, "id = ?", channelID).Error; err != nil {
if errors.Is(err, gorm.ErrRecordNotFound) {
return ErrNotFound
}
return err
}
if err := tx.Create(&fact).Error; err != nil {
return err
}
var c Contact
if err := tx.First(&c, "id = ?", channel.ContactID).Error; err != nil {
return err
}
return appendAudit(tx, c, "channel_validation_"+status, actor)
})
if err != nil {
return ChannelView{}, err
}
return ChannelView{ID: channel.ID, Kind: channel.Kind, AddressMasked: channel.AddressMasked, Status: status, ValidatedAt: &fact.CreatedAt}, nil
}
// DecryptChannelAddress is intentionally server-only. API responses never expose this value.
func (s Service) DecryptChannelAddress(ctx context.Context, channelID string) (string, error) {
var item Channel
if err := s.DB.WithContext(ctx).First(&item, "id = ?", channelID).Error; err != nil {
return "", err
}
return decryptAddress(s.Key, item.AddressCiphertext)
}
func (s Service) view(ctx context.Context, item Contact) (ContactView, error) {
var channels []Channel
if err := s.DB.WithContext(ctx).Where("contact_id = ?", item.ID).Order("created_at,id").Find(&channels).Error; err != nil {
return ContactView{}, err
}
views := make([]ChannelView, 0, len(channels))
for _, ch := range channels {
view := ChannelView{ID: ch.ID, Kind: ch.Kind, AddressMasked: ch.AddressMasked, Status: "pending"}
var fact ChannelValidation
err := s.DB.WithContext(ctx).Where("channel_id = ?", ch.ID).Order("created_at DESC,id DESC").Take(&fact).Error
if err == nil {
view.Status = fact.Status
view.ValidatedAt = &fact.CreatedAt
} else if !errors.Is(err, gorm.ErrRecordNotFound) {
return ContactView{}, err
}
views = append(views, view)
}
return ContactView{Contact: item, Channels: views}, nil
}
func appendAudit(tx *gorm.DB, item Contact, action string, actor int) error {
snapshot, err := json.Marshal(map[string]any{"id": item.ID, "name": item.Name, "role": item.Role, "enabled": item.Enabled, "version": item.Version})
if err != nil {
return err
}
return tx.Create(&AuditFact{ID: uuid.NewString(), ContactID: item.ID, Action: action, Snapshot: snapshot, ActorID: actor, CreatedAt: time.Now().UTC()}).Error
}
func pageValues(page, size int) (int, int) {
if page < 1 {
page = 1
}
if size < 1 || size > 100 {
size = 20
}
return page, size
}
@@ -0,0 +1,60 @@
package contact
import (
"errors"
"regexp"
"strings"
)
var (
ErrInvalid = errors.New("联系人或通道信息不符合要求")
ErrNotFound = errors.New("联系人或通道不存在")
ErrConflict = errors.New("数据已被其他人员更新,请刷新后重试")
phonePattern = regexp.MustCompile(`^\+?[0-9]{6,20}$`)
)
type WriteInput struct {
Name string `json:"name"`
Role string `json:"role"`
ExpectedVersion int `json:"expectedVersion"`
}
type ChannelInput struct {
Kind string `json:"kind"`
Address string `json:"address"`
}
func normalizeContact(input WriteInput) (WriteInput, error) {
input.Name = strings.TrimSpace(input.Name)
input.Role = strings.TrimSpace(input.Role)
if input.Name == "" || len([]rune(input.Name)) > 128 || input.Role == "" || len([]rune(input.Role)) > 128 || hasControl(input.Name+input.Role) {
return WriteInput{}, ErrInvalid
}
return input, nil
}
func normalizeChannel(input ChannelInput) (ChannelInput, error) {
input.Kind = strings.ToLower(strings.TrimSpace(input.Kind))
input.Address = strings.ReplaceAll(strings.ReplaceAll(strings.TrimSpace(input.Address), " ", ""), "-", "")
if (input.Kind != "sms" && input.Kind != "voice") || !phonePattern.MatchString(input.Address) {
return ChannelInput{}, ErrInvalid
}
return input, nil
}
func maskAddress(value string) string {
runes := []rune(value)
if len(runes) <= 4 {
return "****"
}
return strings.Repeat("*", min(8, len(runes)-4)) + string(runes[len(runes)-4:])
}
func hasControl(value string) bool {
for _, r := range value {
if r < 32 || r == 127 {
return true
}
}
return false
}
@@ -0,0 +1,127 @@
package duty_schedule
import (
"errors"
"github.com/gin-gonic/gin"
"github.com/gin-gonic/gin/binding"
"github.com/go-admin-team/go-admin-core/sdk/api"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth/user"
)
type Handler struct{ api.Api }
func (h Handler) List(c *gin.Context) {
var q PageQuery
h.MakeContext(c).MakeOrm().Bind(&q, binding.Form)
if h.Errors != nil {
h.Error(400, ErrInvalid, ErrInvalid.Error())
return
}
items, count, err := NewService(h.Orm).List(c.Request.Context(), q)
if err != nil {
h.Error(500, errors.New("读取排班失败"), "读取排班失败")
return
}
p, s := pageValues(q.PageIndex, q.PageSize)
h.PageOK(items, int(count), p, s, "查询成功")
}
func (h Handler) CreateGroup(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可维护排班"), "仅管理员可维护排班")
return
}
var input GroupInput
h.bind(c, &input)
if h.Errors != nil {
return
}
item, err := NewService(h.Orm).CreateGroup(c.Request.Context(), input, user.GetUserId(c))
h.result(item, err)
}
func (h Handler) UpdateGroup(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可维护排班"), "仅管理员可维护排班")
return
}
var input GroupInput
h.bind(c, &input)
if h.Errors != nil {
return
}
item, err := NewService(h.Orm).UpdateGroup(c.Request.Context(), c.Param("id"), input, user.GetUserId(c))
h.result(item, err)
}
func (h Handler) AddMember(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可维护排班"), "仅管理员可维护排班")
return
}
var input MemberInput
h.bind(c, &input)
if h.Errors != nil {
return
}
item, err := NewService(h.Orm).AddMember(c.Request.Context(), c.Param("id"), input, user.GetUserId(c))
h.result(item, err)
}
func (h Handler) CreateSchedule(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可维护排班"), "仅管理员可维护排班")
return
}
var input ScheduleInput
h.bind(c, &input)
if h.Errors != nil {
return
}
item, err := NewService(h.Orm).CreateSchedule(c.Request.Context(), c.Param("id"), input, user.GetUserId(c))
h.result(item, err)
}
func (h Handler) Publish(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可发布排班"), "仅管理员可发布排班")
return
}
item, err := NewService(h.Orm).Publish(c.Request.Context(), c.Param("id"), user.GetUserId(c))
h.result(item, err)
}
func (h Handler) CreateOverride(c *gin.Context) {
if !admin(c) {
h.MakeContext(c).Error(403, errors.New("仅管理员可维护替班"), "仅管理员可维护替班")
return
}
var input OverrideInput
h.bind(c, &input)
if h.Errors != nil {
return
}
item, err := NewService(h.Orm).CreateOverride(c.Request.Context(), c.Param("id"), input, user.GetUserId(c))
h.result(item, err)
}
func (h *Handler) bind(c *gin.Context, value any) {
h.MakeContext(c).MakeOrm().Bind(value, binding.JSON)
if h.Errors != nil {
h.Error(400, ErrInvalid, ErrInvalid.Error())
}
}
func (h Handler) result(item any, err error) {
switch {
case err == nil:
h.OK(item, "保存成功")
case errors.Is(err, ErrInvalid) || errors.Is(err, ErrCoverage):
h.Error(400, err, err.Error())
case errors.Is(err, ErrNotFound):
h.Error(404, err, err.Error())
case errors.Is(err, ErrConflict):
h.Error(409, err, err.Error())
default:
h.Logger.Errorf("write Bell duty schedule failed: %v", err)
h.Error(409, errors.New("排班保存失败"), "排班保存失败")
}
}
func admin(c *gin.Context) bool {
claims := jwt.ExtractClaims(c)
role, _ := claims[jwt.RoleKey].(string)
return role == "admin"
}
@@ -0,0 +1,94 @@
package duty_schedule
import (
"encoding/json"
"time"
)
type Group struct {
ID string `json:"id" gorm:"type:uuid;primaryKey"`
Name string `json:"name" gorm:"size:128;not null;uniqueIndex"`
Timezone string `json:"timezone" gorm:"size:64;not null"`
Enabled bool `json:"enabled" gorm:"not null;default:true;index"`
Version int `json:"version" gorm:"not null;default:1"`
CreatedBy int `json:"createdBy" gorm:"not null"`
UpdatedBy int `json:"updatedBy" gorm:"not null"`
CreatedAt time.Time `json:"createdAt" gorm:"type:timestamptz;not null"`
UpdatedAt time.Time `json:"updatedAt" gorm:"type:timestamptz;not null"`
}
func (Group) TableName() string { return "bell_duty_groups" }
type Member struct {
GroupID string `json:"groupId" gorm:"type:uuid;primaryKey"`
ContactID string `json:"contactId" gorm:"type:uuid;primaryKey"`
Role string `json:"role" gorm:"size:16;not null"`
CreatedBy int `json:"createdBy" gorm:"not null"`
CreatedAt time.Time `json:"createdAt" gorm:"type:timestamptz;not null"`
}
func (Member) TableName() string { return "bell_duty_members" }
type ScheduleVersion struct {
ID string `json:"id" gorm:"type:uuid;primaryKey"`
GroupID string `json:"groupId" gorm:"type:uuid;not null;index"`
Version int `json:"version" gorm:"not null"`
Timezone string `json:"timezone" gorm:"size:64;not null"`
EffectiveFrom time.Time `json:"effectiveFrom" gorm:"type:timestamptz;not null"`
EffectiveTo *time.Time `json:"effectiveTo,omitempty" gorm:"type:timestamptz"`
Status string `json:"status" gorm:"size:16;not null"`
CreatedBy int `json:"createdBy" gorm:"not null"`
CreatedAt time.Time `json:"createdAt" gorm:"type:timestamptz;not null"`
PublishedBy *int `json:"publishedBy,omitempty"`
PublishedAt *time.Time `json:"publishedAt,omitempty" gorm:"type:timestamptz"`
}
func (ScheduleVersion) TableName() string { return "bell_duty_schedule_versions" }
type RotationSlot struct {
ID string `json:"id" gorm:"type:uuid;primaryKey"`
ScheduleVersionID string `json:"scheduleVersionId" gorm:"type:uuid;not null;index"`
Weekday int `json:"weekday" gorm:"not null"`
StartMinute int `json:"startMinute" gorm:"not null"`
EndMinute int `json:"endMinute" gorm:"not null"`
PrimaryContactID string `json:"primaryContactId" gorm:"type:uuid;not null"`
BackupContactID string `json:"backupContactId" gorm:"type:uuid;not null"`
}
func (RotationSlot) TableName() string { return "bell_duty_rotation_slots" }
type Override struct {
ID string `json:"id" gorm:"type:uuid;primaryKey"`
GroupID string `json:"groupId" gorm:"type:uuid;not null;index"`
OriginalContactID string `json:"originalContactId" gorm:"type:uuid;not null"`
ReplacementContactID string `json:"replacementContactId" gorm:"type:uuid;not null"`
StartsAt time.Time `json:"startsAt" gorm:"type:timestamptz;not null;index"`
EndsAt time.Time `json:"endsAt" gorm:"type:timestamptz;not null"`
Reason string `json:"reason" gorm:"size:256;not null"`
CreatedBy int `json:"createdBy" gorm:"not null"`
CreatedAt time.Time `json:"createdAt" gorm:"type:timestamptz;not null"`
}
func (Override) TableName() string { return "bell_duty_overrides" }
type AuditFact struct {
ID string `json:"id" gorm:"type:uuid;primaryKey"`
GroupID string `json:"groupId" gorm:"type:uuid;not null;index"`
Action string `json:"action" gorm:"size:32;not null"`
Snapshot json.RawMessage `json:"snapshot" gorm:"type:jsonb;not null"`
ActorID int `json:"actorId" gorm:"not null"`
CreatedAt time.Time `json:"createdAt" gorm:"type:timestamptz;not null"`
}
func (AuditFact) TableName() string { return "bell_duty_audit_facts" }
type GroupView struct {
Group
Members []Member `json:"members"`
Schedules []ScheduleView `json:"schedules"`
Overrides []Override `json:"overrides"`
}
type ScheduleView struct {
ScheduleVersion
Slots []RotationSlot `json:"slots"`
}
@@ -0,0 +1,299 @@
package duty_schedule
import (
"context"
"encoding/json"
"strings"
"time"
"github.com/google/uuid"
"gorm.io/gorm"
"gorm.io/gorm/clause"
"go-admin/app/bell/contact"
)
type PageQuery struct {
PageIndex int `form:"pageIndex"`
PageSize int `form:"pageSize"`
Name string `form:"name"`
Enabled *bool `form:"enabled"`
}
type Service struct{ DB *gorm.DB }
func NewService(db *gorm.DB) Service { return Service{DB: db} }
func (s Service) List(ctx context.Context, q PageQuery) ([]GroupView, int64, error) {
p, z := pageValues(q.PageIndex, q.PageSize)
db := s.DB.WithContext(ctx).Model(&Group{})
if name := strings.TrimSpace(q.Name); name != "" {
db = db.Where("name ILIKE ?", "%"+name+"%")
}
if q.Enabled != nil {
db = db.Where("enabled = ?", *q.Enabled)
}
var count int64
if err := db.Count(&count).Error; err != nil {
return nil, 0, err
}
var groups []Group
if err := db.Order("created_at DESC,id DESC").Offset((p - 1) * z).Limit(z).Find(&groups).Error; err != nil {
return nil, 0, err
}
views := make([]GroupView, 0, len(groups))
for _, g := range groups {
v, err := s.view(ctx, g)
if err != nil {
return nil, 0, err
}
views = append(views, v)
}
return views, count, nil
}
func (s Service) CreateGroup(ctx context.Context, input GroupInput, actor int) (GroupView, error) {
input, err := normalizeGroup(input)
if err != nil {
return GroupView{}, err
}
now := time.Now().UTC()
item := Group{ID: uuid.NewString(), Name: input.Name, Timezone: input.Timezone, Enabled: true, Version: 1, CreatedBy: actor, UpdatedBy: actor, CreatedAt: now, UpdatedAt: now}
err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := tx.Create(&item).Error; err != nil {
return err
}
return audit(tx, item.ID, "group_created", item, actor)
})
return GroupView{Group: item, Members: []Member{}, Schedules: []ScheduleView{}, Overrides: []Override{}}, err
}
func (s Service) UpdateGroup(ctx context.Context, id string, input GroupInput, actor int) (GroupView, error) {
input, err := normalizeGroup(input)
if err != nil {
return GroupView{}, err
}
if input.ExpectedVersion < 1 {
return GroupView{}, ErrInvalid
}
var item Group
err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
r := tx.Model(&Group{}).Where("id=? AND version=?", id, input.ExpectedVersion).Updates(map[string]any{"name": input.Name, "timezone": input.Timezone, "version": gorm.Expr("version+1"), "updated_by": actor, "updated_at": time.Now().UTC()})
if r.Error != nil {
return r.Error
}
if r.RowsAffected != 1 {
return ErrConflict
}
if err := tx.First(&item, "id=?", id).Error; err != nil {
return err
}
return audit(tx, id, "group_updated", item, actor)
})
if err != nil {
return GroupView{}, err
}
return s.view(ctx, item)
}
func (s Service) AddMember(ctx context.Context, groupID string, input MemberInput, actor int) (Member, error) {
input.Role = strings.ToLower(strings.TrimSpace(input.Role))
if input.Role != "primary" && input.Role != "backup" {
return Member{}, ErrInvalid
}
now := time.Now().UTC()
item := Member{GroupID: groupID, ContactID: input.ContactID, Role: input.Role, CreatedBy: actor, CreatedAt: now}
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := assertEnabledContact(tx, input.ContactID); err != nil {
return err
}
if err := tx.First(&Group{}, "id=?", groupID).Error; err != nil {
return ErrNotFound
}
if err := tx.Clauses(clause.OnConflict{Columns: []clause.Column{{Name: "group_id"}, {Name: "contact_id"}}, DoUpdates: clause.AssignmentColumns([]string{"role", "created_by", "created_at"})}).Create(&item).Error; err != nil {
return err
}
return audit(tx, groupID, "member_saved", item, actor)
})
return item, err
}
func (s Service) CreateSchedule(ctx context.Context, groupID string, input ScheduleInput, actor int) (ScheduleView, error) {
if input.EffectiveFrom.IsZero() || (input.EffectiveTo != nil && !input.EffectiveTo.After(input.EffectiveFrom)) {
return ScheduleView{}, ErrInvalid
}
if err := validateSlots(input.Slots); err != nil {
return ScheduleView{}, err
}
var result ScheduleView
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var group Group
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&group, "id=?", groupID).Error; err != nil {
return ErrNotFound
}
for _, slot := range input.Slots {
if err := assertGroupMember(tx, groupID, slot.PrimaryContactID); err != nil {
return err
}
if err := assertGroupMember(tx, groupID, slot.BackupContactID); err != nil {
return err
}
}
var latest int
tx.Model(&ScheduleVersion{}).Where("group_id=?", groupID).Select("coalesce(max(version),0)").Scan(&latest)
now := time.Now().UTC()
version := ScheduleVersion{ID: uuid.NewString(), GroupID: groupID, Version: latest + 1, Timezone: group.Timezone, EffectiveFrom: input.EffectiveFrom.UTC(), EffectiveTo: input.EffectiveTo, Status: "draft", CreatedBy: actor, CreatedAt: now}
if err := tx.Create(&version).Error; err != nil {
return err
}
slots := make([]RotationSlot, 0, len(input.Slots))
for _, in := range input.Slots {
slots = append(slots, RotationSlot{ID: uuid.NewString(), ScheduleVersionID: version.ID, Weekday: in.Weekday, StartMinute: in.StartMinute, EndMinute: in.EndMinute, PrimaryContactID: in.PrimaryContactID, BackupContactID: in.BackupContactID})
}
if err := tx.Create(&slots).Error; err != nil {
return err
}
if err := audit(tx, groupID, "schedule_created", version, actor); err != nil {
return err
}
result = ScheduleView{ScheduleVersion: version, Slots: slots}
return nil
})
return result, err
}
func (s Service) Publish(ctx context.Context, id string, actor int) (ScheduleView, error) {
var result ScheduleView
err := s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
var item ScheduleVersion
if err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).First(&item, "id=?", id).Error; err != nil {
return ErrNotFound
}
if item.Status != "draft" {
return ErrConflict
}
var slots []RotationSlot
if err := tx.Where("schedule_version_id=?", id).Find(&slots).Error; err != nil {
return err
}
inputs := make([]SlotInput, 0, len(slots))
for _, v := range slots {
inputs = append(inputs, SlotInput{Weekday: v.Weekday, StartMinute: v.StartMinute, EndMinute: v.EndMinute, PrimaryContactID: v.PrimaryContactID, BackupContactID: v.BackupContactID})
}
if err := validateSlots(inputs); err != nil {
return err
}
for _, v := range slots {
if err := assertVerifiedContact(tx, v.PrimaryContactID); err != nil {
return err
}
if err := assertVerifiedContact(tx, v.BackupContactID); err != nil {
return err
}
}
now := time.Now().UTC()
if err := tx.Model(&item).Updates(map[string]any{"status": "published", "published_by": actor, "published_at": now}).Error; err != nil {
return err
}
item.Status = "published"
item.PublishedBy = &actor
item.PublishedAt = &now
if err := audit(tx, item.GroupID, "schedule_published", item, actor); err != nil {
return err
}
result = ScheduleView{ScheduleVersion: item, Slots: slots}
return nil
})
return result, err
}
func (s Service) CreateOverride(ctx context.Context, groupID string, input OverrideInput, actor int) (Override, error) {
input, err := normalizeOverride(input)
if err != nil {
return Override{}, err
}
item := Override{ID: uuid.NewString(), GroupID: groupID, OriginalContactID: input.OriginalContactID, ReplacementContactID: input.ReplacementContactID, StartsAt: input.StartsAt.UTC(), EndsAt: input.EndsAt.UTC(), Reason: input.Reason, CreatedBy: actor, CreatedAt: time.Now().UTC()}
err = s.DB.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
if err := assertGroupMember(tx, groupID, input.OriginalContactID); err != nil {
return err
}
if err := assertGroupMember(tx, groupID, input.ReplacementContactID); err != nil {
return err
}
var overlaps int64
if err := tx.Model(&Override{}).Where("group_id=? AND original_contact_id=? AND starts_at < ? AND ends_at > ?", groupID, input.OriginalContactID, item.EndsAt, item.StartsAt).Count(&overlaps).Error; err != nil {
return err
}
if overlaps > 0 {
return ErrConflict
}
if err := tx.Create(&item).Error; err != nil {
return err
}
return audit(tx, groupID, "override_created", item, actor)
})
return item, err
}
func (s Service) view(ctx context.Context, g Group) (GroupView, error) {
v := GroupView{Group: g, Members: []Member{}, Schedules: []ScheduleView{}, Overrides: []Override{}}
if err := s.DB.WithContext(ctx).Where("group_id=?", g.ID).Order("role,contact_id").Find(&v.Members).Error; err != nil {
return v, err
}
var versions []ScheduleVersion
if err := s.DB.WithContext(ctx).Where("group_id=?", g.ID).Order("version DESC").Find(&versions).Error; err != nil {
return v, err
}
for _, sv := range versions {
var slots []RotationSlot
if err := s.DB.WithContext(ctx).Where("schedule_version_id=?", sv.ID).Order("weekday,start_minute").Find(&slots).Error; err != nil {
return v, err
}
v.Schedules = append(v.Schedules, ScheduleView{ScheduleVersion: sv, Slots: slots})
}
if err := s.DB.WithContext(ctx).Where("group_id=?", g.ID).Order("starts_at DESC").Limit(50).Find(&v.Overrides).Error; err != nil {
return v, err
}
return v, nil
}
func assertEnabledContact(tx *gorm.DB, id string) error {
var c contact.Contact
if err := tx.Where("id=? AND enabled=true", id).First(&c).Error; err != nil {
return ErrInvalid
}
return nil
}
func assertGroupMember(tx *gorm.DB, groupID, contactID string) error {
if err := assertEnabledContact(tx, contactID); err != nil {
return err
}
var count int64
if err := tx.Model(&Member{}).Where("group_id=? AND contact_id=?", groupID, contactID).Count(&count).Error; err != nil {
return err
}
if count != 1 {
return ErrInvalid
}
return nil
}
func assertVerifiedContact(tx *gorm.DB, contactID string) error {
if err := assertEnabledContact(tx, contactID); err != nil {
return err
}
var count int64
err := tx.Raw(`SELECT count(*) FROM bell_contact_channels c WHERE c.contact_id=? AND (SELECT v.status FROM bell_contact_channel_validations v WHERE v.channel_id=c.id ORDER BY v.created_at DESC,v.id DESC LIMIT 1)='verified'`, contactID).Scan(&count).Error
if err != nil {
return err
}
if count == 0 {
return ErrInvalid
}
return nil
}
func audit(tx *gorm.DB, groupID, action string, value any, actor int) error {
data, err := json.Marshal(value)
if err != nil {
return err
}
return tx.Create(&AuditFact{ID: uuid.NewString(), GroupID: groupID, Action: action, Snapshot: data, ActorID: actor, CreatedAt: time.Now().UTC()}).Error
}
func pageValues(p, s int) (int, int) {
if p < 1 {
p = 1
}
if s < 1 || s > 100 {
s = 20
}
return p, s
}
@@ -0,0 +1,90 @@
package duty_schedule
import (
"errors"
"sort"
"strings"
"time"
)
var (
ErrInvalid = errors.New("值班排班信息不符合要求")
ErrNotFound = errors.New("值班组或排班不存在")
ErrConflict = errors.New("数据已被其他人员更新,请刷新后重试")
ErrCoverage = errors.New("周排班存在空档或重叠")
)
type GroupInput struct {
Name string `json:"name"`
Timezone string `json:"timezone"`
ExpectedVersion int `json:"expectedVersion"`
}
type MemberInput struct {
ContactID string `json:"contactId"`
Role string `json:"role"`
}
type SlotInput struct {
Weekday int `json:"weekday"`
StartMinute int `json:"startMinute"`
EndMinute int `json:"endMinute"`
PrimaryContactID string `json:"primaryContactId"`
BackupContactID string `json:"backupContactId"`
}
type ScheduleInput struct {
EffectiveFrom time.Time `json:"effectiveFrom"`
EffectiveTo *time.Time `json:"effectiveTo"`
Slots []SlotInput `json:"slots"`
}
type OverrideInput struct {
OriginalContactID string `json:"originalContactId"`
ReplacementContactID string `json:"replacementContactId"`
StartsAt time.Time `json:"startsAt"`
EndsAt time.Time `json:"endsAt"`
Reason string `json:"reason"`
}
func normalizeGroup(input GroupInput) (GroupInput, error) {
input.Name = strings.TrimSpace(input.Name)
input.Timezone = strings.TrimSpace(input.Timezone)
if input.Name == "" || len([]rune(input.Name)) > 128 {
return GroupInput{}, ErrInvalid
}
if _, err := time.LoadLocation(input.Timezone); err != nil {
return GroupInput{}, ErrInvalid
}
return input, nil
}
func validateSlots(slots []SlotInput) error {
if len(slots) == 0 {
return ErrCoverage
}
byDay := map[int][]SlotInput{}
for _, slot := range slots {
if slot.Weekday < 0 || slot.Weekday > 6 || slot.StartMinute < 0 || slot.EndMinute > 1440 || slot.StartMinute >= slot.EndMinute || slot.PrimaryContactID == "" || slot.BackupContactID == "" || slot.PrimaryContactID == slot.BackupContactID {
return ErrInvalid
}
byDay[slot.Weekday] = append(byDay[slot.Weekday], slot)
}
for day := 0; day < 7; day++ {
daySlots := byDay[day]
sort.Slice(daySlots, func(i, j int) bool { return daySlots[i].StartMinute < daySlots[j].StartMinute })
cursor := 0
for _, slot := range daySlots {
if slot.StartMinute != cursor {
return ErrCoverage
}
cursor = slot.EndMinute
}
if cursor != 1440 {
return ErrCoverage
}
}
return nil
}
func normalizeOverride(input OverrideInput) (OverrideInput, error) {
input.Reason = strings.TrimSpace(input.Reason)
if input.OriginalContactID == "" || input.ReplacementContactID == "" || input.OriginalContactID == input.ReplacementContactID || input.StartsAt.IsZero() || !input.EndsAt.After(input.StartsAt) || input.Reason == "" || len([]rune(input.Reason)) > 256 {
return OverrideInput{}, ErrInvalid
}
return input, nil
}
+21
View File
@@ -0,0 +1,21 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"go-admin/app/bell/contact"
"go-admin/common/middleware"
)
func init() { registrars = append(registrars, registerContactRouter) }
func registerContactRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
h := contact.Handler{}
secured := v1.Group("").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole())
secured.GET("/contacts", h.List)
secured.POST("/contacts", h.Create)
secured.PUT("/contacts/:id", h.Update)
secured.PUT("/contacts/:id/enabled", h.SetEnabled)
secured.POST("/contacts/:id/channels", h.AddChannel)
secured.POST("/contact-channels/:id/validations", h.ValidateChannel)
}
@@ -0,0 +1,22 @@
package router
import (
"github.com/gin-gonic/gin"
jwt "github.com/go-admin-team/go-admin-core/sdk/pkg/jwtauth"
"go-admin/app/bell/duty_schedule"
"go-admin/common/middleware"
)
func init() { registrars = append(registrars, registerDutyScheduleRouter) }
func registerDutyScheduleRouter(v1 *gin.RouterGroup, auth *jwt.GinJWTMiddleware) {
h := duty_schedule.Handler{}
secured := v1.Group("").Use(auth.MiddlewareFunc()).Use(middleware.AuthCheckRole())
secured.GET("/duty-groups", h.List)
secured.POST("/duty-groups", h.CreateGroup)
secured.PUT("/duty-groups/:id", h.UpdateGroup)
secured.POST("/duty-groups/:id/members", h.AddMember)
secured.POST("/duty-groups/:id/schedules", h.CreateSchedule)
secured.POST("/duty-schedules/:id/publish", h.Publish)
secured.POST("/duty-groups/:id/overrides", h.CreateOverride)
}
+3 -1
View File
@@ -21,6 +21,7 @@ import (
"go-admin/app/admin/models"
"go-admin/app/admin/router"
"go-admin/app/bell/alert_lifecycle"
"go-admin/app/bell/contact"
bellrouter "go-admin/app/bell/router"
"go-admin/app/bell/synthetic"
"go-admin/common/bellconfig"
@@ -184,7 +185,8 @@ func initRouter() {
Use(common.RequestId(pkg.TrafficKey)).
Use(api.SetRequestLogger).
Use(synthetic.RedactRequestBody()).
Use(alert_lifecycle.RedactRequestBody())
Use(alert_lifecycle.RedactRequestBody()).
Use(contact.RedactRequestBody())
common.InitMiddleware(r)
@@ -0,0 +1,160 @@
package version
import (
"fmt"
"runtime"
"go-admin/app/bell/contact"
duty "go-admin/app/bell/duty_schedule"
"go-admin/cmd/migrate/migration"
common "go-admin/common/models"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
func init() {
_, fileName, _, _ := runtime.Caller(0)
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateBellContactSchedule)
}
func migrateBellContactSchedule(db *gorm.DB, version string) error {
return db.Transaction(func(tx *gorm.DB) error {
if err := tx.AutoMigrate(&contact.Contact{}, &contact.Channel{}, &contact.ChannelValidation{}, &contact.AuditFact{}, &duty.Group{}, &duty.Member{}, &duty.ScheduleVersion{}, &duty.RotationSlot{}, &duty.Override{}, &duty.AuditFact{}); err != nil {
return err
}
if tx.Dialector.Name() == "postgres" {
for _, sql := range contactScheduleSQL {
if err := tx.Exec(sql).Error; err != nil {
return err
}
}
if err := seedContactScheduleAccess(tx); err != nil {
return err
}
}
return tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&common.Migration{Version: version}).Error
})
}
var contactScheduleSQL = []string{
`ALTER TABLE bell_contacts ADD CONSTRAINT bell_contacts_version_check CHECK (version > 0)`,
`ALTER TABLE bell_contact_channels ADD CONSTRAINT bell_contact_channels_kind_check CHECK (kind IN ('sms','voice'))`,
`ALTER TABLE bell_contact_channels ADD CONSTRAINT bell_contact_channels_contact_fk FOREIGN KEY (contact_id) REFERENCES bell_contacts(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`CREATE UNIQUE INDEX bell_contact_channel_identity_idx ON bell_contact_channels(contact_id,kind,address_fingerprint)`,
`CREATE TRIGGER bell_contact_channels_immutable BEFORE UPDATE OR DELETE ON bell_contact_channels FOR EACH ROW EXECUTE FUNCTION bell_reject_immutable_fact()`,
`ALTER TABLE bell_contact_channel_validations ADD CONSTRAINT bell_contact_validation_status_check CHECK (status IN ('verified','failed'))`,
`ALTER TABLE bell_contact_channel_validations ADD CONSTRAINT bell_contact_validation_channel_fk FOREIGN KEY (channel_id) REFERENCES bell_contact_channels(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`ALTER TABLE bell_contact_audit_facts ADD CONSTRAINT bell_contact_audit_contact_fk FOREIGN KEY (contact_id) REFERENCES bell_contacts(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`ALTER TABLE bell_duty_groups ADD CONSTRAINT bell_duty_groups_version_check CHECK (version > 0)`,
`ALTER TABLE bell_duty_members ADD CONSTRAINT bell_duty_member_role_check CHECK (role IN ('primary','backup'))`,
`ALTER TABLE bell_duty_members ADD CONSTRAINT bell_duty_member_group_fk FOREIGN KEY (group_id) REFERENCES bell_duty_groups(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`ALTER TABLE bell_duty_members ADD CONSTRAINT bell_duty_member_contact_fk FOREIGN KEY (contact_id) REFERENCES bell_contacts(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`CREATE UNIQUE INDEX bell_duty_schedule_group_version_idx ON bell_duty_schedule_versions(group_id,version)`,
`ALTER TABLE bell_duty_schedule_versions ADD CONSTRAINT bell_duty_schedule_status_check CHECK (status IN ('draft','published'))`,
`ALTER TABLE bell_duty_schedule_versions ADD CONSTRAINT bell_duty_schedule_group_fk FOREIGN KEY (group_id) REFERENCES bell_duty_groups(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`CREATE OR REPLACE FUNCTION bell_guard_schedule_version() RETURNS trigger LANGUAGE plpgsql AS $$ BEGIN IF TG_OP = 'DELETE' OR OLD.status = 'published' THEN RAISE EXCEPTION 'Bell published schedule cannot be changed' USING ERRCODE = '55000'; END IF; IF NEW.status <> 'published' OR OLD.status <> 'draft' THEN RAISE EXCEPTION 'Bell schedule transition is invalid' USING ERRCODE = '55000'; END IF; RETURN NEW; END $$`,
`CREATE TRIGGER bell_duty_schedule_version_guard BEFORE UPDATE OR DELETE ON bell_duty_schedule_versions FOR EACH ROW EXECUTE FUNCTION bell_guard_schedule_version()`,
`ALTER TABLE bell_duty_rotation_slots ADD CONSTRAINT bell_duty_slot_range_check CHECK (weekday BETWEEN 0 AND 6 AND start_minute >= 0 AND end_minute <= 1440 AND start_minute < end_minute AND primary_contact_id <> backup_contact_id)`,
`ALTER TABLE bell_duty_rotation_slots ADD CONSTRAINT bell_duty_slot_version_fk FOREIGN KEY (schedule_version_id) REFERENCES bell_duty_schedule_versions(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`ALTER TABLE bell_duty_rotation_slots ADD CONSTRAINT bell_duty_slot_primary_fk FOREIGN KEY (primary_contact_id) REFERENCES bell_contacts(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`ALTER TABLE bell_duty_rotation_slots ADD CONSTRAINT bell_duty_slot_backup_fk FOREIGN KEY (backup_contact_id) REFERENCES bell_contacts(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`ALTER TABLE bell_duty_overrides ADD CONSTRAINT bell_duty_override_range_check CHECK (starts_at < ends_at AND original_contact_id <> replacement_contact_id)`,
`ALTER TABLE bell_duty_overrides ADD CONSTRAINT bell_duty_override_group_fk FOREIGN KEY (group_id) REFERENCES bell_duty_groups(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`ALTER TABLE bell_duty_overrides ADD CONSTRAINT bell_duty_override_original_fk FOREIGN KEY (original_contact_id) REFERENCES bell_contacts(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`ALTER TABLE bell_duty_overrides ADD CONSTRAINT bell_duty_override_replacement_fk FOREIGN KEY (replacement_contact_id) REFERENCES bell_contacts(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`ALTER TABLE bell_duty_audit_facts ADD CONSTRAINT bell_duty_audit_group_fk FOREIGN KEY (group_id) REFERENCES bell_duty_groups(id) ON UPDATE RESTRICT ON DELETE RESTRICT`,
`CREATE TRIGGER bell_contact_validations_immutable BEFORE UPDATE OR DELETE ON bell_contact_channel_validations FOR EACH ROW EXECUTE FUNCTION bell_reject_immutable_fact()`,
`CREATE TRIGGER bell_contact_audit_immutable BEFORE UPDATE OR DELETE ON bell_contact_audit_facts FOR EACH ROW EXECUTE FUNCTION bell_reject_immutable_fact()`,
`CREATE TRIGGER bell_duty_slots_immutable BEFORE UPDATE OR DELETE ON bell_duty_rotation_slots FOR EACH ROW EXECUTE FUNCTION bell_reject_immutable_fact()`,
`CREATE TRIGGER bell_duty_overrides_immutable BEFORE UPDATE OR DELETE ON bell_duty_overrides FOR EACH ROW EXECUTE FUNCTION bell_reject_immutable_fact()`,
`CREATE TRIGGER bell_duty_audit_immutable BEFORE UPDATE OR DELETE ON bell_duty_audit_facts FOR EACH ROW EXECUTE FUNCTION bell_reject_immutable_fact()`,
}
type contactScheduleSeed struct {
ID int
Path string
Action string
}
func seedContactScheduleAccess(tx *gorm.DB) error {
if err := tx.Exec(`SELECT setval(pg_get_serial_sequence('sys_menu','menu_id'),GREATEST((SELECT max(menu_id) FROM sys_menu),1));SELECT setval(pg_get_serial_sequence('sys_api','id'),GREATEST((SELECT max(id) FROM sys_api),1))`).Error; err != nil {
return err
}
var rootID int
if err := tx.Raw("SELECT menu_id FROM sys_menu WHERE path='/bell' AND parent_id=0 ORDER BY menu_id LIMIT 1").Scan(&rootID).Error; err != nil || rootID == 0 {
return fmt.Errorf("Bell menu root missing")
}
contacts, err := insertContactScheduleMenu(tx, rootID, "BellContacts", "联系人与通道", "user", "contacts", "C", "bell:contact:list", "", "/bell/contacts/index", 4)
if err != nil {
return err
}
dutyMenu, err := insertContactScheduleMenu(tx, rootID, "BellDutySchedules", "值班排班", "time", "duty-schedules", "C", "bell:duty:list", "", "/bell/duty-schedules/index", 5)
if err != nil {
return err
}
contactWrite, err := insertContactScheduleMenu(tx, contacts.ID, "", "维护联系人", "", "", "F", "bell:contact:write", "POST", "", 1)
if err != nil {
return err
}
dutyWrite, err := insertContactScheduleMenu(tx, dutyMenu.ID, "", "维护排班", "", "", "F", "bell:duty:write", "POST", "", 1)
if err != nil {
return err
}
specs := []struct{ title, path, action string }{
{"联系人列表", "/api/v1/bell/contacts", "GET"}, {"新增联系人", "/api/v1/bell/contacts", "POST"}, {"修改联系人", "/api/v1/bell/contacts/:id", "PUT"}, {"启停联系人", "/api/v1/bell/contacts/:id/enabled", "PUT"}, {"新增联系通道", "/api/v1/bell/contacts/:id/channels", "POST"}, {"记录通道验证", "/api/v1/bell/contact-channels/:id/validations", "POST"},
{"值班组列表", "/api/v1/bell/duty-groups", "GET"}, {"新增值班组", "/api/v1/bell/duty-groups", "POST"}, {"修改值班组", "/api/v1/bell/duty-groups/:id", "PUT"}, {"保存值班成员", "/api/v1/bell/duty-groups/:id/members", "POST"}, {"新增排班版本", "/api/v1/bell/duty-groups/:id/schedules", "POST"}, {"发布排班版本", "/api/v1/bell/duty-schedules/:id/publish", "POST"}, {"新增临时替班", "/api/v1/bell/duty-groups/:id/overrides", "POST"},
}
apis := make([]contactScheduleSeed, 0, len(specs))
for _, s := range specs {
v, e := insertContactScheduleAPI(tx, s.title, s.path, s.action)
if e != nil {
return e
}
apis = append(apis, v)
}
links := map[int][]contactScheduleSeed{contacts.ID: {apis[0]}, contactWrite.ID: apis[1:6], dutyMenu.ID: {apis[6]}, dutyWrite.ID: apis[7:]}
for menu, items := range links {
for _, item := range items {
if err := tx.Exec("INSERT INTO sys_menu_api_rule(sys_menu_menu_id,sys_api_id) VALUES(?,?) ON CONFLICT DO NOTHING", menu, item.ID).Error; err != nil {
return err
}
}
}
var operator int
if err := tx.Raw("SELECT role_id FROM sys_role WHERE role_key='operator' AND deleted_at IS NULL ORDER BY role_id LIMIT 1").Scan(&operator).Error; err != nil {
return err
}
if operator != 0 {
for _, menu := range []contactScheduleSeed{contacts, dutyMenu} {
if err := tx.Exec("INSERT INTO sys_role_menu(role_id,menu_id) VALUES(?,?) ON CONFLICT DO NOTHING", operator, menu.ID).Error; err != nil {
return err
}
}
for _, item := range []contactScheduleSeed{apis[0], apis[6]} {
if err := tx.Exec("INSERT INTO casbin_rule(ptype,v0,v1,v2,v3,v4,v5) VALUES('p','operator',?,?, '', '', '') ON CONFLICT DO NOTHING", item.Path, item.Action).Error; err != nil {
return err
}
}
}
return nil
}
func insertContactScheduleMenu(tx *gorm.DB, parent int, name, title, icon, path, menuType, permission, action, component string, sort int) (contactScheduleSeed, error) {
var id int
err := tx.Raw(`INSERT INTO sys_menu(menu_name,title,icon,path,paths,menu_type,action,permission,parent_id,no_cache,breadcrumb,component,sort,visible,is_frame,create_by,update_by,created_at,updated_at) VALUES(?,?,?,?, '',?,?,?,?,false,'',?,?, '0','1',1,1,now(),now()) RETURNING menu_id`, name, title, icon, path, menuType, action, permission, parent, component, sort).Scan(&id).Error
if err != nil {
return contactScheduleSeed{}, err
}
var parentPaths string
if err = tx.Raw("SELECT paths FROM sys_menu WHERE menu_id=?", parent).Scan(&parentPaths).Error; err != nil {
return contactScheduleSeed{}, err
}
if err = tx.Exec("UPDATE sys_menu SET paths=? WHERE menu_id=?", fmt.Sprintf("%s/%d", parentPaths, id), id).Error; err != nil {
return contactScheduleSeed{}, err
}
return contactScheduleSeed{ID: id}, nil
}
func insertContactScheduleAPI(tx *gorm.DB, title, path, action string) (contactScheduleSeed, error) {
var id int
err := tx.Raw(`INSERT INTO sys_api(handle,title,path,type,action,created_at,updated_at,create_by,update_by) VALUES('',?,?, 'BUS',?,now(),now(),1,1) RETURNING id`, title, path, action).Scan(&id).Error
return contactScheduleSeed{ID: id, Path: path, Action: action}, err
}
@@ -0,0 +1,36 @@
package version
import (
"testing"
"go-admin/app/bell/contact"
duty "go-admin/app/bell/duty_schedule"
common "go-admin/common/models"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func TestBellContactScheduleMigrationIsIdempotent(t *testing.T) {
db, err := gorm.Open(sqlite.Open(":memory:"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&common.Migration{}); err != nil {
t.Fatal(err)
}
for i := 0; i < 2; i++ {
if err = migrateBellContactSchedule(db, "2026090110000"); err != nil {
t.Fatalf("attempt %d: %v", i+1, err)
}
}
for name, model := range map[string]any{"contacts": &contact.Contact{}, "channels": &contact.Channel{}, "validations": &contact.ChannelValidation{}, "groups": &duty.Group{}, "members": &duty.Member{}, "versions": &duty.ScheduleVersion{}, "slots": &duty.RotationSlot{}, "overrides": &duty.Override{}} {
if !db.Migrator().HasTable(model) {
t.Fatalf("%s table missing", name)
}
}
var count int64
if err = db.Model(&common.Migration{}).Where("version=?", "2026090110000").Count(&count).Error; err != nil || count != 1 {
t.Fatalf("migration records=%d err=%v", count, err)
}
}
@@ -0,0 +1,169 @@
package bell_contact_schedule_test
import (
"context"
"encoding/json"
"errors"
"os"
"strings"
"testing"
"time"
adminmodels "go-admin/app/admin/models"
"go-admin/app/bell/contact"
duty "go-admin/app/bell/duty_schedule"
"gorm.io/driver/postgres"
"gorm.io/gorm"
)
func TestContactSchedulePostgres(t *testing.T) {
dsn := os.Getenv("BELL_CONTACT_SCHEDULE_TEST_DATABASE_URL")
if dsn == "" {
t.Skip("set BELL_CONTACT_SCHEDULE_TEST_DATABASE_URL to run PostgreSQL verification")
}
db, err := gorm.Open(postgres.Open(dsn), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
ctx := context.Background()
key := []byte("0123456789abcdef0123456789abcdef")
contacts := contact.NewService(db, key)
primary, err := contacts.Create(ctx, contact.WriteInput{Name: "联系人甲", Role: "主值班"}, 1)
if err != nil {
t.Fatal(err)
}
backup, err := contacts.Create(ctx, contact.WriteInput{Name: "联系人乙", Role: "备值班"}, 1)
if err != nil {
t.Fatal(err)
}
primaryChannel, err := contacts.AddChannel(ctx, primary.ID, contact.ChannelInput{Kind: "sms", Address: "+8613800000001"}, 1)
if err != nil {
t.Fatal(err)
}
backupChannel, err := contacts.AddChannel(ctx, backup.ID, contact.ChannelInput{Kind: "voice", Address: "+8613800000002"}, 1)
if err != nil {
t.Fatal(err)
}
if strings.Contains(primaryChannel.AddressMasked, "13800000001") {
t.Fatal("channel response leaked address")
}
plain, err := contacts.DecryptChannelAddress(ctx, primaryChannel.ID)
if err != nil || plain != "+8613800000001" {
t.Fatalf("server-only decrypt failed: %q %v", plain, err)
}
encoded, _ := json.Marshal(primaryChannel)
if strings.Contains(string(encoded), plain) {
t.Fatal("serialized channel leaked plaintext")
}
if _, err = contacts.RecordValidation(ctx, primaryChannel.ID, "verified", "合成验证", 1); err != nil {
t.Fatal(err)
}
if _, err = contacts.RecordValidation(ctx, backupChannel.ID, "verified", "合成验证", 1); err != nil {
t.Fatal(err)
}
if _, err = contacts.Update(ctx, primary.ID, contact.WriteInput{Name: "联系人甲", Role: "主值班", ExpectedVersion: 99}, 1); !errors.Is(err, contact.ErrConflict) {
t.Fatalf("stale contact update err=%v", err)
}
if _, err = contacts.SetEnabled(ctx, primary.ID, false, primary.Version, 1); err != nil {
t.Fatal(err)
}
listed, _, err := contacts.List(ctx, contact.PageQuery{PageIndex: 1, PageSize: 20})
if err != nil || len(listed) != 2 {
t.Fatalf("contact list len=%d err=%v", len(listed), err)
}
var primaryView *contact.ContactView
for index := range listed {
if listed[index].ID == primary.ID {
primaryView = &listed[index]
}
}
if primaryView == nil || primaryView.Enabled || len(primaryView.Channels) != 1 || primaryView.Channels[0].Status != "verified" {
t.Fatalf("contact enabled state was coupled to validation: %#v", primaryView)
}
// Re-enable with the new version before assigning duty.
var disabled contact.Contact
if err = db.First(&disabled, "id=?", primary.ID).Error; err != nil {
t.Fatal(err)
}
if _, err = contacts.SetEnabled(ctx, primary.ID, true, disabled.Version, 1); err != nil {
t.Fatal(err)
}
duties := duty.NewService(db)
group, err := duties.CreateGroup(ctx, duty.GroupInput{Name: "夜间值班组", Timezone: "Asia/Shanghai"}, 1)
if err != nil {
t.Fatal(err)
}
if _, err = duties.AddMember(ctx, group.ID, duty.MemberInput{ContactID: primary.ID, Role: "primary"}, 1); err != nil {
t.Fatal(err)
}
if _, err = duties.AddMember(ctx, group.ID, duty.MemberInput{ContactID: backup.ID, Role: "backup"}, 1); err != nil {
t.Fatal(err)
}
if _, err = duties.CreateSchedule(ctx, group.ID, duty.ScheduleInput{EffectiveFrom: time.Now().UTC().Add(time.Hour), Slots: []duty.SlotInput{{Weekday: 0, StartMinute: 0, EndMinute: 720, PrimaryContactID: primary.ID, BackupContactID: backup.ID}}}, 1); !errors.Is(err, duty.ErrCoverage) {
t.Fatalf("schedule gap was accepted: %v", err)
}
slots := make([]duty.SlotInput, 0, 7)
for day := 0; day < 7; day++ {
slots = append(slots, duty.SlotInput{Weekday: day, StartMinute: 0, EndMinute: 1440, PrimaryContactID: primary.ID, BackupContactID: backup.ID})
}
v1, err := duties.CreateSchedule(ctx, group.ID, duty.ScheduleInput{EffectiveFrom: time.Now().UTC().Add(time.Hour), Slots: slots}, 1)
if err != nil {
t.Fatal(err)
}
published, err := duties.Publish(ctx, v1.ID, 1)
if err != nil || published.Status != "published" {
t.Fatalf("publish status=%s err=%v", published.Status, err)
}
v2, err := duties.CreateSchedule(ctx, group.ID, duty.ScheduleInput{EffectiveFrom: time.Now().UTC().Add(24 * time.Hour), Slots: slots}, 1)
if err != nil || v2.Version != 2 {
t.Fatalf("second version=%d err=%v", v2.Version, err)
}
var persisted duty.ScheduleVersion
if err = db.First(&persisted, "id=?", v1.ID).Error; err != nil || persisted.Version != 1 || persisted.Status != "published" {
t.Fatalf("historical version changed: %#v err=%v", persisted, err)
}
now := time.Now().UTC().Add(2 * time.Hour)
if _, err = duties.CreateOverride(ctx, group.ID, duty.OverrideInput{OriginalContactID: primary.ID, ReplacementContactID: backup.ID, StartsAt: now, EndsAt: now.Add(time.Hour), Reason: "合成替班"}, 1); err != nil {
t.Fatal(err)
}
if _, err = duties.CreateOverride(ctx, group.ID, duty.OverrideInput{OriginalContactID: primary.ID, ReplacementContactID: backup.ID, StartsAt: now.Add(30 * time.Minute), EndsAt: now.Add(90 * time.Minute), Reason: "重叠替班"}, 1); !errors.Is(err, duty.ErrConflict) {
t.Fatalf("overlap err=%v", err)
}
if err = db.Model(&contact.ChannelValidation{}).Where("channel_id=?", primaryChannel.ID).Update("detail", "tampered").Error; err == nil {
t.Fatal("validation fact was mutable")
}
if err = db.Model(&duty.RotationSlot{}).Where("schedule_version_id=?", v1.ID).Update("start_minute", 1).Error; err == nil {
t.Fatal("published rotation slot was mutable")
}
if err = db.Model(&duty.ScheduleVersion{}).Where("id=?", v1.ID).Update("effective_from", time.Now().UTC()).Error; err == nil {
t.Fatal("published schedule version was mutable")
}
var menus, reads, writes int64
if err = db.Table("sys_role_menu rm").Joins("JOIN sys_role r ON r.role_id=rm.role_id").Joins("JOIN sys_menu m ON m.menu_id=rm.menu_id").Where("r.role_key=? AND m.path IN ?", "operator", []string{"contacts", "duty-schedules"}).Count(&menus).Error; err != nil {
t.Fatal(err)
}
if err = db.Table("casbin_rule").Where("v0=? AND v2=? AND v1 IN ?", "operator", "GET", []string{"/api/v1/bell/contacts", "/api/v1/bell/duty-groups"}).Count(&reads).Error; err != nil {
t.Fatal(err)
}
if err = db.Table("casbin_rule").Where("v0=? AND v2<>? AND (v1 LIKE ? OR v1 LIKE ?)", "operator", "GET", "/api/v1/bell/contacts%", "/api/v1/bell/duty-%").Count(&writes).Error; err != nil {
t.Fatal(err)
}
if menus != 2 || reads != 2 || writes != 0 {
t.Fatalf("operator access escaped scope: menus=%d reads=%d writes=%d", menus, reads, writes)
}
password := os.Getenv("BELL_RULE_ALERT_OPERATOR_PASSWORD")
if password != "" {
var roleID int
if err = db.Table("sys_role").Select("role_id").Where("role_key=?", "operator").Scan(&roleID).Error; err != nil || roleID == 0 {
t.Fatalf("operator role id=%d err=%v", roleID, err)
}
user := adminmodels.SysUser{Username: "bell_132_operator", Password: password, NickName: "Bell 处置员", RoleId: roleID, DeptId: 1, PostId: 1, Status: "2"}
if err = db.Create(&user).Error; err != nil {
t.Fatal(err)
}
}
}
@@ -0,0 +1,124 @@
[CmdletBinding()]
param([string]$PostgresBin = 'D:\pgsql17\bin')
Set-StrictMode -Version 3.0
$ErrorActionPreference = 'Stop'
$pgStarted = $false
$server = $null
$testRoot = Join-Path ([IO.Path]::GetTempPath()) ('yovision-bell-183-' + [guid]::NewGuid().ToString('N'))
$data = Join-Path $testRoot 'postgres'
$log = Join-Path $testRoot 'postgres.log'
$pgOut = Join-Path $testRoot 'pg.out'
$pgErr = Join-Path $testRoot 'pg.err'
$serverOut = Join-Path $testRoot 'bell.out.log'
$serverErr = Join-Path $testRoot 'bell.err.log'
$serverExe = Join-Path $testRoot 'bell-server.exe'
$serverRoot = (Resolve-Path (Join-Path $PSScriptRoot '..\..')).Path
function Get-FreePort {
$listener = [Net.Sockets.TcpListener]::new([Net.IPAddress]::Loopback, 0)
try { $listener.Start(); return ([Net.IPEndPoint]$listener.LocalEndpoint).Port } finally { $listener.Stop() }
}
function Wait-Port([int]$Port) {
for ($attempt = 0; $attempt -lt 120; $attempt++) {
try {
$client = [Net.Sockets.TcpClient]::new()
$open = $client.ConnectAsync('127.0.0.1', $Port).Wait(250) -and $client.Connected
$client.Dispose()
if ($open) { return }
} catch {}
Start-Sleep -Milliseconds 250
}
throw "PostgreSQL port $Port did not open"
}
function Wait-Health([string]$BaseUrl) {
for ($attempt = 0; $attempt -lt 100; $attempt++) {
try {
$health = Invoke-RestMethod -Uri "$BaseUrl/healthz" -TimeoutSec 2 -NoProxy
if ($health.status -eq 'ok') { return }
} catch {}
Start-Sleep -Milliseconds 300
}
throw 'Bell health endpoint did not become ready'
}
New-Item -ItemType Directory -Path $testRoot | Out-Null
$pgPort = Get-FreePort
$bellPort = Get-FreePort
$baseUrl = "http://127.0.0.1:$bellPort"
try {
foreach ($name in @('initdb.exe', 'pg_ctl.exe', 'createdb.exe', 'psql.exe')) {
if (-not (Test-Path -LiteralPath (Join-Path $PostgresBin $name) -PathType Leaf)) { throw "Missing PostgreSQL tool: $name" }
}
& (Join-Path $PostgresBin 'initdb.exe') -D $data -U postgres -A trust --encoding=UTF8 --no-locale | Out-Null
if ($LASTEXITCODE -ne 0) { throw 'initdb failed' }
$arguments = "-D `"$data`" -l `"$log`" -o `"-p $pgPort -h 127.0.0.1`" start"
Start-Process -FilePath (Join-Path $PostgresBin 'pg_ctl.exe') -ArgumentList $arguments -RedirectStandardOutput $pgOut -RedirectStandardError $pgErr -WindowStyle Hidden | Out-Null
Wait-Port $pgPort
$pgStarted = $true
& (Join-Path $PostgresBin 'createdb.exe') -h 127.0.0.1 -p $pgPort -U postgres bell_183
if ($LASTEXITCODE -ne 0) { throw 'createdb failed' }
$env:GOTOOLCHAIN = 'go1.26.5'
$env:BELL_DATABASE_URL = "host=127.0.0.1 port=$pgPort user=postgres dbname=bell_183 sslmode=disable"
$env:BELL_CONTACT_SCHEDULE_TEST_DATABASE_URL = $env:BELL_DATABASE_URL
$env:BELL_JWT_SECRET = [guid]::NewGuid().ToString('N') + [guid]::NewGuid().ToString('N')
$env:BELL_BOOTSTRAP_USERNAME = 'bell_183_admin'
$env:BELL_BOOTSTRAP_PASSWORD = [guid]::NewGuid().ToString('N')
$env:BELL_RULE_ALERT_OPERATOR_PASSWORD = [guid]::NewGuid().ToString('N')
$env:BELL_CONTACT_CHANNEL_KEY = [Convert]::ToBase64String([Text.Encoding]::UTF8.GetBytes('0123456789abcdef0123456789abcdef'))
$env:BELL_HOST = '127.0.0.1'
$env:BELL_PORT = $bellPort.ToString()
Push-Location $serverRoot
try {
go run . migrate -c config/settings.demo.yml *> (Join-Path $testRoot 'migrate.log')
if ($LASTEXITCODE -ne 0) { throw "migration failed: $(Join-Path $testRoot 'migrate.log')" }
go test ./tests/bell_contact_schedule -count=1 -v
if ($LASTEXITCODE -ne 0) { throw 'contact schedule tests failed' }
go build -o $serverExe .
if ($LASTEXITCODE -ne 0) { throw 'Bell build failed' }
} finally { Pop-Location }
$server = Start-Process -FilePath $serverExe -ArgumentList @('server', '-c', 'config/settings.demo.yml') -WorkingDirectory $serverRoot -RedirectStandardOutput $serverOut -RedirectStandardError $serverErr -WindowStyle Hidden -PassThru
Wait-Health $baseUrl
$adminBody = @{ username = $env:BELL_BOOTSTRAP_USERNAME; password = $env:BELL_BOOTSTRAP_PASSWORD; code = '0'; uuid = '0' } | ConvertTo-Json -Compress
$admin = Invoke-RestMethod -Method Post -Uri "$baseUrl/api/v1/login" -ContentType 'application/json' -Body $adminBody -NoProxy
$adminHeaders = @{ Authorization = "Bearer $($admin.token)" }
$contactBody = @{ name = 'HTTP联系人'; role = '测试值班' } | ConvertTo-Json -Compress
$created = Invoke-RestMethod -Method Post -Uri "$baseUrl/api/v1/bell/contacts" -Headers $adminHeaders -ContentType 'application/json; charset=utf-8' -Body $contactBody -NoProxy
if ([int]$created.code -ne 200) { throw 'administrator contact create failed' }
$address = '+8613900000003'
$channelBody = @{ kind = 'sms'; address = $address } | ConvertTo-Json -Compress
$channel = Invoke-RestMethod -Method Post -Uri "$baseUrl/api/v1/bell/contacts/$($created.data.id)/channels" -Headers $adminHeaders -ContentType 'application/json' -Body $channelBody -NoProxy
if ([int]$channel.code -ne 200 -or ($channel | ConvertTo-Json -Depth 10 -Compress).Contains($address)) { throw 'write-only channel HTTP boundary failed' }
$logRow = ''
for ($attempt = 0; $attempt -lt 40; $attempt++) {
$logRow = & (Join-Path $PostgresBin 'psql.exe') -h 127.0.0.1 -p $pgPort -U postgres -d bell_183 -Atc "SELECT id::text || '|' || coalesce(oper_param,'') FROM sys_opera_log WHERE oper_url LIKE '/api/v1/bell/contacts/%/channels' ORDER BY id DESC LIMIT 1"
if ($LASTEXITCODE -ne 0) { throw 'operation log query failed' }
if ($logRow) { break }
Start-Sleep -Milliseconds 100
}
if (-not $logRow -or $logRow.Contains($address)) { throw "operation log redaction failed: $logRow" }
$loggedBody = ($logRow -split '\|', 2)[1]
if ($loggedBody -and -not $loggedBody.Contains('"redacted":true')) { throw "unexpected operation log marker: $loggedBody" }
$operatorBody = @{ username = 'bell_132_operator'; password = $env:BELL_RULE_ALERT_OPERATOR_PASSWORD; code = '0'; uuid = '0' } | ConvertTo-Json -Compress
$operator = Invoke-RestMethod -Method Post -Uri "$baseUrl/api/v1/login" -ContentType 'application/json' -Body $operatorBody -NoProxy
$operatorHeaders = @{ Authorization = "Bearer $($operator.token)" }
foreach ($path in @('/api/v1/bell/contacts', '/api/v1/bell/duty-groups')) {
$read = Invoke-RestMethod -Uri "$baseUrl$path" -Headers $operatorHeaders -NoProxy
if ([int]$read.code -ne 200) { throw "operator read failed: $path" }
}
$denied = Invoke-RestMethod -Method Post -Uri "$baseUrl/api/v1/bell/contacts" -Headers $operatorHeaders -ContentType 'application/json' -Body $contactBody -NoProxy
if ([int]$denied.code -ne 403) { throw "operator write returned $($denied.code)" }
$menu = Invoke-RestMethod -Uri "$baseUrl/api/v1/menurole" -Headers $operatorHeaders -NoProxy
$menuJson = $menu.data | ConvertTo-Json -Depth 20 -Compress
foreach ($title in @('联系人与通道', '值班排班')) { if (-not $menuJson.Contains($title)) { throw "operator menu missing $title" } }
Write-Output 'BELL_183_HTTP admin_contact=200 channel_write_only=true operator_reads=200 operator_write=403 menus=true'
} finally {
if ($null -ne $server -and -not $server.HasExited) { Stop-Process -Id $server.Id -Force; $server.WaitForExit(5000) | Out-Null }
if ($pgStarted) { & (Join-Path $PostgresBin 'pg_ctl.exe') -D $data -m fast stop *> (Join-Path $testRoot 'stop.log') }
foreach ($name in @('BELL_DATABASE_URL', 'BELL_CONTACT_SCHEDULE_TEST_DATABASE_URL', 'BELL_JWT_SECRET', 'BELL_BOOTSTRAP_USERNAME', 'BELL_BOOTSTRAP_PASSWORD', 'BELL_RULE_ALERT_OPERATOR_PASSWORD', 'BELL_CONTACT_CHANNEL_KEY', 'BELL_HOST', 'BELL_PORT')) { Remove-Item "Env:$name" -ErrorAction SilentlyContinue }
Write-Verbose "Bell #183 artifacts: $testRoot"
}
@@ -0,0 +1,22 @@
package bell_contact_schedule_test
import (
"encoding/base64"
"errors"
"testing"
"go-admin/app/bell/contact"
)
func TestChannelKeyAndWriteOnlyRoundTrip(t *testing.T) {
encoded := base64.StdEncoding.EncodeToString([]byte("0123456789abcdef0123456789abcdef"))
key, err := contact.ParseChannelKey(encoded)
if err != nil || len(key) != 32 {
t.Fatalf("key parse failed: len=%d err=%v", len(key), err)
}
for _, value := range []string{"", "short", base64.StdEncoding.EncodeToString([]byte("0123456789abcdef"))} {
if _, err = contact.ParseChannelKey(value); !errors.Is(err, contact.ErrChannelKeyUnavailable) {
t.Fatalf("invalid key accepted: %q err=%v", value, err)
}
}
}
+7
View File
@@ -0,0 +1,7 @@
import request from '@/utils/request'
export function listContacts(query) { return request({ url: '/api/v1/bell/contacts', method: 'get', params: query }) }
export function createContact(data) { return request({ url: '/api/v1/bell/contacts', method: 'post', data }) }
export function updateContact(id, data) { return request({ url: `/api/v1/bell/contacts/${id}`, method: 'put', data }) }
export function setContactEnabled(id, enabled, expectedVersion) { return request({ url: `/api/v1/bell/contacts/${id}/enabled`, method: 'put', data: { enabled, expectedVersion }}) }
export function addContactChannel(id, data) { return request({ url: `/api/v1/bell/contacts/${id}/channels`, method: 'post', data }) }
export function validateContactChannel(id, data) { return request({ url: `/api/v1/bell/contact-channels/${id}/validations`, method: 'post', data }) }
+8
View File
@@ -0,0 +1,8 @@
import request from '@/utils/request'
export function listDutyGroups(query) { return request({ url: '/api/v1/bell/duty-groups', method: 'get', params: query }) }
export function createDutyGroup(data) { return request({ url: '/api/v1/bell/duty-groups', method: 'post', data }) }
export function updateDutyGroup(id, data) { return request({ url: `/api/v1/bell/duty-groups/${id}`, method: 'put', data }) }
export function saveDutyMember(id, data) { return request({ url: `/api/v1/bell/duty-groups/${id}/members`, method: 'post', data }) }
export function createDutySchedule(id, data) { return request({ url: `/api/v1/bell/duty-groups/${id}/schedules`, method: 'post', data }) }
export function publishDutySchedule(id) { return request({ url: `/api/v1/bell/duty-schedules/${id}/publish`, method: 'post' }) }
export function createDutyOverride(id, data) { return request({ url: `/api/v1/bell/duty-groups/${id}/overrides`, method: 'post', data }) }
+21
View File
@@ -0,0 +1,21 @@
<template>
<BasicLayout><template #wrapper><el-card>
<template #header><div class="heading"><div><h2>联系人与通道</h2><p>通道地址保存后只显示脱敏值,验证状态与联系人启用状态相互独立。</p></div><el-button v-permisaction="['bell:contact:write']" type="primary" @click="openCreate">新增联系人</el-button></div></template>
<el-form ref="queryForm" :model="query" :inline="true"><el-form-item label="联系人" prop="name"><el-input v-model="query.name" clearable placeholder="姓名或岗位" @keyup.enter="search" /></el-form-item><el-form-item label="状态" prop="enabled"><el-select v-model="query.enabled" clearable placeholder="全部" style="width:120px"><el-option label="启用" :value="true" /><el-option label="停用" :value="false" /></el-select></el-form-item><el-form-item><el-button type="primary" @click="search">搜索</el-button><el-button @click="reset">重置</el-button></el-form-item></el-form>
<el-alert v-if="error" :title="error" type="error" show-icon :closable="false" class="state" />
<el-table v-loading="loading" :data="items" border row-key="id">
<el-table-column prop="name" label="联系人" min-width="130" /><el-table-column prop="role" label="岗位" min-width="130" />
<el-table-column label="通道" min-width="260"><template #default="scope"><div v-if="scope.row.channels.length"><el-tag v-for="ch in scope.row.channels" :key="ch.id" :type="statusType(ch.status)" class="channel">{{ kindName(ch.kind) }} {{ ch.addressMasked }} · {{ statusName(ch.status) }}</el-tag></div><span v-else class="muted">未配置</span></template></el-table-column>
<el-table-column label="状态" width="100"><template #default="scope"><el-switch v-model="scope.row.enabled" :disabled="!canWrite" inline-prompt active-text="启" inactive-text="停" @change="toggle(scope.row)" /></template></el-table-column><el-table-column prop="version" label="版本" width="70" />
<el-table-column label="操作" width="210" fixed="right"><template #default="scope"><el-button v-permisaction="['bell:contact:write']" link type="primary" @click="openEdit(scope.row)">编辑</el-button><el-button v-permisaction="['bell:contact:write']" link type="primary" @click="openChannel(scope.row)">新增通道</el-button><el-dropdown v-if="scope.row.channels.length && canWrite" @command="command => validate(scope.row, command)"><el-button link type="primary">验证通道</el-button><template #dropdown><el-dropdown-menu><el-dropdown-item v-for="ch in scope.row.channels" :key="ch.id" :command="ch">{{ kindName(ch.kind) }} {{ ch.addressMasked }}</el-dropdown-item></el-dropdown-menu></template></el-dropdown></template></el-table-column>
<template #empty><el-empty description="暂无联系人" /></template>
</el-table><pagination v-show="total>0" v-model:current-page="query.pageIndex" v-model:page-size="query.pageSize" :total="total" @pagination="load" />
<el-dialog v-model="contactDialog" :title="editing?'编辑联系人':'新增联系人'" width="min(520px, calc(100vw - 32px))" :close-on-click-modal="false"><el-form ref="contactForm" :model="form" :rules="rules" label-position="top"><el-form-item label="称呼" prop="name"><el-input v-model.trim="form.name" maxlength="128" /></el-form-item><el-form-item label="岗位" prop="role"><el-input v-model.trim="form.role" maxlength="128" /></el-form-item></el-form><template #footer><el-button @click="contactDialog=false">取消</el-button><el-button type="primary" :loading="saving" @click="saveContact">保存</el-button></template></el-dialog>
<el-dialog v-model="channelDialog" title="新增联系通道" width="min(520px, calc(100vw - 32px))" :close-on-click-modal="false"><el-alert title="号码仅在本次填写时可见,保存后只返回脱敏值。" type="info" :closable="false" class="state" /><el-form ref="channelForm" :model="channel" :rules="channelRules" label-position="top"><el-form-item label="通道" prop="kind"><el-select v-model="channel.kind" style="width:100%"><el-option label="短信" value="sms" /><el-option label="语音" value="voice" /></el-select></el-form-item><el-form-item label="号码" prop="address"><el-input v-model.trim="channel.address" autocomplete="off" placeholder="请输入合法测试号码" /></el-form-item></el-form><template #footer><el-button @click="channelDialog=false">取消</el-button><el-button type="primary" :loading="saving" @click="saveChannel">保存</el-button></template></el-dialog>
</el-card></template></BasicLayout>
</template>
<script>
import { addContactChannel, createContact, listContacts, setContactEnabled, updateContact, validateContactChannel } from '@/api/bell/contact'
export default { name: 'BellContacts', data() { return { loading: false, saving: false, error: '', items: [], total: 0, contactDialog: false, channelDialog: false, editing: false, editingId: '', channelContactId: '', query: { pageIndex: 1, pageSize: 10, name: '', enabled: null }, form: { name: '', role: '', expectedVersion: 0 }, channel: { kind: 'sms', address: '' }, rules: { name: [{ required: true, message: '请输入称呼', trigger: 'blur' }], role: [{ required: true, message: '请输入岗位', trigger: 'blur' }] }, channelRules: { kind: [{ required: true, message: '请选择通道', trigger: 'change' }], address: [{ required: true, pattern: /^\+?[0-9 -]{6,24}$/, message: '请输入有效号码', trigger: 'blur' }] }} }, computed: { canWrite() { const p = this.$store.getters.permisaction || []; return p.includes('*:*:*') || p.includes('bell:contact:write') } }, created() { this.load() }, methods: { async load() { this.loading = true; this.error = ''; try { const r = await listContacts(this.query); this.items = r.data.list || []; this.total = r.data.count || 0 } catch (e) { this.error = e.message || '联系人加载失败' } finally { this.loading = false } }, search() { this.query.pageIndex = 1; this.load() }, reset() { this.$refs.queryForm.resetFields(); this.query.enabled = null; this.search() }, openCreate() { this.editing = false; this.editingId = ''; this.form = { name: '', role: '', expectedVersion: 0 }; this.contactDialog = true }, openEdit(row) { this.editing = true; this.editingId = row.id; this.form = { name: row.name, role: row.role, expectedVersion: row.version }; this.contactDialog = true }, openChannel(row) { this.channelContactId = row.id; this.channel = { kind: 'sms', address: '' }; this.channelDialog = true }, async saveContact() { try { await this.$refs.contactForm.validate(); this.saving = true; if (this.editing) await updateContact(this.editingId, this.form); else await createContact(this.form); this.msgSuccess('联系人已保存'); this.contactDialog = false; await this.load() } catch (e) { if (e && e.message) this.error = e.message } finally { this.saving = false } }, async saveChannel() { try { await this.$refs.channelForm.validate(); this.saving = true; await addContactChannel(this.channelContactId, this.channel); this.msgSuccess('通道已加密保存,等待验证'); this.channelDialog = false; await this.load() } catch (e) { if (e && e.message) this.error = e.message } finally { this.saving = false } }, async toggle(row) { try { await setContactEnabled(row.id, row.enabled, row.version); this.msgSuccess(row.enabled ? '联系人已启用' : '联系人已停用'); await this.load() } catch (e) { row.enabled = !row.enabled; this.error = e.message || '状态更新失败' } }, async validate(row, ch) { try { await this.$confirm(`确认合成验证 ${ch.addressMasked} 成功?本操作不会发送外部消息。`, '记录验证结果', { type: 'warning' }); await validateContactChannel(ch.id, { status: 'verified', detail: '人工合成验证' }); this.msgSuccess('验证事实已记录'); await this.load() } catch (e) { if (e !== 'cancel' && e !== 'close' && e && e.message) this.error = e.message } }, kindName(v) { return { sms: '短信', voice: '语音' }[v] || v }, statusName(v) { return { pending: '待验证', verified: '已验证', failed: '验证失败' }[v] || v }, statusType(v) { return { pending: 'warning', verified: 'success', failed: 'danger' }[v] || 'info' } }}
</script>
<style scoped>.heading{display:flex;align-items:center;justify-content:space-between;gap:16px}.heading h2{margin:0}.heading p{margin:6px 0 0;color:var(--el-text-color-secondary)}.state{margin-bottom:16px}.channel{margin:2px 6px 2px 0}.muted{color:var(--el-text-color-secondary)}</style>
File diff suppressed because one or more lines are too long
@@ -0,0 +1,7 @@
"""Credential-free Sense control-plane connector for Brain."""
from .consumer import ApplyResult, SourceConfigConsumer, SourceConfigError
from .replay import SQLiteReplayStore
from .status import RuntimeStatusPublisher
__all__ = ["ApplyResult", "SourceConfigConsumer", "SourceConfigError", "SQLiteReplayStore", "RuntimeStatusPublisher"]
@@ -0,0 +1,57 @@
"""Machine-authenticated adapters with bounded timeout/backoff and a kill switch."""
from __future__ import annotations
import time
import re
from dataclasses import dataclass
from typing import Callable
from yovision_brain.integration.machine_identity.token import Signer, Verifier, bearer_token
from .consumer import ApplyResult, SourceConfigConsumer
_REQUEST_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$")
@dataclass(frozen=True, slots=True)
class ConnectorResponse:
status: int
body: bytes
correlation_id: str
class SourceConfigEndpoint:
def __init__(self, consumer: SourceConfigConsumer, verifier: Verifier, *, enabled: bool = True, max_body_bytes: int = 10 * 1024 * 1024) -> None:
self._consumer, self._verifier, self._enabled, self._max = consumer, verifier, enabled, max_body_bytes
def receive(self, authorization: str, body: bytes, correlation_id: str) -> ApplyResult:
if not self._enabled: raise RuntimeError("CONNECTOR_DISABLED")
if not _REQUEST_ID.fullmatch(correlation_id): raise ValueError("INVALID_CORRELATION_ID")
if len(body) > self._max: raise ValueError("REQUEST_TOO_LARGE")
token=bearer_token(authorization)
self._verifier.verify(token,"yovision-brain","source-config:write","POST","/machine/v1/source-config",body)
return self._consumer.apply(body)
class StatusSender:
def __init__(self, signer: Signer, send: Callable[[str, bytes, str, float], ConnectorResponse], *, enabled: bool = True, timeout_seconds: float = 5.0, max_attempts: int = 4, sleeper: Callable[[float], None] = time.sleep) -> None:
if timeout_seconds <= 0 or max_attempts < 1: raise ValueError("invalid connector retry policy")
self._signer,self._send,self._enabled,self._timeout,self._attempts,self._sleep=signer,send,enabled,timeout_seconds,max_attempts,sleeper
def publish(self, body: bytes, correlation_id: str) -> ConnectorResponse:
if not self._enabled: raise RuntimeError("CONNECTOR_DISABLED")
if not _REQUEST_ID.fullmatch(correlation_id): raise ValueError("INVALID_CORRELATION_ID")
last: Exception|None=None
for attempt in range(self._attempts):
try:
# A retry gets a fresh jti: the previous request may have been
# accepted even when its response was lost.
token=self._signer.mint("yovision-sense",("runtime-status:write",),"POST","/machine/v1/runtime-status",body)
response=self._send("Bearer "+token,body,correlation_id,self._timeout)
if 200<=response.status<300:return response
if response.status<500:raise RuntimeError(f"STATUS_REJECTED_{response.status}")
last=RuntimeError(f"STATUS_REMOTE_{response.status}")
except (TimeoutError,ConnectionError) as exc:last=exc
if attempt+1<self._attempts:self._sleep(min(2**attempt,30))
raise RuntimeError("STATUS_DELIVERY_EXHAUSTED") from last
@@ -0,0 +1,250 @@
"""Strict source-config/v1 validation and atomic last-known-good application."""
from __future__ import annotations
import hashlib
import hmac
import json
import re
import sqlite3
import threading
import time
from contextlib import closing
from dataclasses import dataclass
from datetime import datetime, timezone
from pathlib import Path
from typing import Callable, Mapping
from urllib.parse import urlsplit
from yovision_brain.rules.models import AreaDefinition, DirectionalLineDefinition, NormalizedPoint, RuleSet
VERSION = "yovision.source-config/v1"
_ID = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._~-]{0,127}$")
_EXTENSION_NAMESPACE = re.compile(r"^[A-Za-z][A-Za-z0-9.-]{0,127}$")
_SECRET = re.compile(r"password|secret|credential|cookie|jwt|username|stream_uri", re.I)
class SourceConfigError(ValueError):
def __init__(self, code: str) -> None:
super().__init__(code)
self.code = code
@dataclass(frozen=True, slots=True)
class AppliedConfig:
config_id: str
revision: int
logical_device_id: str
media_ref: str
profile_encoding: str
frame_rate: float
rule_state: str
rules: RuleSet | None
@dataclass(frozen=True, slots=True)
class ApplyResult:
config_id: str
revision: int
state: str
config: AppliedConfig | None
class SourceConfigConsumer:
"""Persists validated snapshots before atomically changing the active pointer."""
def __init__(self, state_path: str | Path, *, clock: Callable[[], float] | None = None) -> None:
self._path = str(state_path)
self._clock = clock or time.time
self._lock = threading.RLock()
with closing(self._connect()) as connection:
connection.executescript(
"""
CREATE TABLE IF NOT EXISTS source_snapshots (
config_id TEXT NOT NULL, revision INTEGER NOT NULL, effective_at INTEGER NOT NULL,
state TEXT NOT NULL, payload TEXT NOT NULL, PRIMARY KEY(config_id, revision));
CREATE TABLE IF NOT EXISTS source_active (
config_id TEXT PRIMARY KEY, revision INTEGER NOT NULL,
FOREIGN KEY(config_id, revision) REFERENCES source_snapshots(config_id, revision));
"""
)
connection.commit()
def _connect(self) -> sqlite3.Connection:
connection = sqlite3.connect(self._path, timeout=5)
connection.execute("PRAGMA foreign_keys=ON")
connection.execute("PRAGMA journal_mode=WAL")
return connection
def apply(self, body: bytes) -> ApplyResult:
document = _parse_and_validate(body)
config_id, revision = document["config_id"], document["revision"]
mapped = _map(document)
effective_at = int(_timestamp(document["effective_at"]))
with self._lock, closing(self._connect()) as connection:
connection.execute("BEGIN IMMEDIATE")
latest = connection.execute(
"SELECT revision, payload, state, effective_at FROM source_snapshots WHERE config_id=? ORDER BY revision DESC LIMIT 1",
(config_id,),
).fetchone()
if latest and revision < latest[0]:
connection.rollback()
raise SourceConfigError("STALE_REVISION")
canonical = body.decode("utf-8")
if latest and revision == latest[0]:
if json.loads(latest[1]) != document:
connection.rollback()
raise SourceConfigError("REVISION_CONFLICT")
connection.rollback()
return ApplyResult(config_id, revision, "idempotent", self.get_active(config_id))
connection.execute(
"INSERT INTO source_snapshots(config_id, revision, effective_at, state, payload) VALUES (?, ?, ?, ?, ?)",
(config_id, revision, effective_at, document["rule_set"]["state"], canonical),
)
if effective_at <= int(self._clock()):
connection.execute(
"INSERT INTO source_active(config_id, revision) VALUES (?, ?) ON CONFLICT(config_id) DO UPDATE SET revision=excluded.revision",
(config_id, revision),
)
state = "applied"
else:
state = "scheduled"
connection.commit()
return ApplyResult(config_id, revision, state, mapped if state == "applied" else self.get_active(config_id))
def activate_due(self) -> tuple[AppliedConfig, ...]:
now = int(self._clock())
activated: list[AppliedConfig] = []
with self._lock, closing(self._connect()) as connection:
connection.execute("BEGIN IMMEDIATE")
rows = connection.execute(
"SELECT s.payload FROM source_snapshots s JOIN (SELECT config_id, MAX(revision) revision FROM source_snapshots WHERE effective_at<=? GROUP BY config_id) d ON d.config_id=s.config_id AND d.revision=s.revision",
(now,),
).fetchall()
for (payload,) in rows:
document = json.loads(payload)
connection.execute(
"INSERT INTO source_active(config_id, revision) VALUES (?, ?) ON CONFLICT(config_id) DO UPDATE SET revision=excluded.revision",
(document["config_id"], document["revision"]),
)
activated.append(_map(document))
connection.commit()
return tuple(activated)
def get_active(self, config_id: str) -> AppliedConfig | None:
with closing(self._connect()) as connection:
row = connection.execute(
"SELECT s.payload FROM source_active a JOIN source_snapshots s ON s.config_id=a.config_id AND s.revision=a.revision WHERE a.config_id=?",
(config_id,),
).fetchone()
return _map(json.loads(row[0])) if row else None
def _parse_and_validate(body: bytes) -> dict[str, object]:
try:
document = json.loads(body.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError):
raise SourceConfigError("CONFIG_INVALID") from None
if not isinstance(document, dict):
raise SourceConfigError("CONFIG_INVALID")
if document.get("schema_version") != VERSION:
raise SourceConfigError("UNSUPPORTED_SCHEMA_VERSION")
if _contains_secret(document):
raise SourceConfigError("CONFIG_INVALID")
required = {"schema_version", "config_id", "revision", "published_at", "effective_at", "site", "logical_device", "profile", "media", "rule_set", "integrity"}
if set(document) - (required | {"extensions"}) or not required <= set(document):
raise SourceConfigError("CONFIG_INVALID")
extensions = document.get("extensions", {})
if not isinstance(extensions, dict) or any(
not isinstance(namespace, str)
or not _EXTENSION_NAMESPACE.fullmatch(namespace)
or not isinstance(value, dict)
for namespace, value in extensions.items()
):
raise SourceConfigError("CONFIG_INVALID")
integrity = document.get("integrity")
if not isinstance(integrity, dict) or set(integrity) != {"algorithm", "value"} or integrity.get("algorithm") != "sha256":
raise SourceConfigError("CONFIG_INVALID")
unsigned = dict(document); unsigned.pop("integrity")
digest = hashlib.sha256(json.dumps(unsigned, ensure_ascii=False, separators=(",", ":"), sort_keys=True).encode()).hexdigest()
if not hmac.compare_digest(digest, str(integrity.get("value", ""))):
raise SourceConfigError("CONFIG_INVALID")
try:
if not _ID.fullmatch(document["config_id"]) or isinstance(document["revision"], bool) or document["revision"] < 1:
raise ValueError
published, effective = _timestamp(document["published_at"]), _timestamp(document["effective_at"])
if effective < published:
raise ValueError
for field in ("site", "logical_device"):
if not isinstance(document[field], dict) or set(document[field]) != {"id"} or not _ID.fullmatch(document[field]["id"]): raise ValueError
_validate_profile(document["profile"])
media = document["media"]
if not isinstance(media, dict) or set(media) != {"ref", "transport"} or media["transport"] != "rtsp" or not isinstance(media["ref"], str) or not media["ref"].startswith("media:") or any(marker in media["ref"] for marker in ("?", "#", "@", "\\", "://")): raise ValueError
_validate_rules(document["rule_set"], document["profile"])
except (KeyError, TypeError, ValueError, AttributeError):
raise SourceConfigError("CONFIG_INVALID") from None
return document
def _validate_profile(profile: object) -> None:
if not isinstance(profile, dict) or set(profile) != {"id", "width", "height", "encoding", "frame_rate"}: raise ValueError
if not _ID.fullmatch(profile["id"]) or profile["encoding"] not in {"H264", "H265", "MJPEG"}: raise ValueError
for field in ("width", "height"):
if isinstance(profile[field], bool) or not isinstance(profile[field], int) or profile[field] < 1: raise ValueError
if isinstance(profile["frame_rate"], bool) or not isinstance(profile["frame_rate"], (int, float)) or profile["frame_rate"] <= 0: raise ValueError
def _validate_rules(rules: object, profile: Mapping[str, object]) -> None:
if not isinstance(rules, dict) or set(rules) != {"version", "state", "profile_binding", "areas", "directional_lines"}: raise ValueError
if not _ID.fullmatch(rules["version"]) or rules["state"] not in {"active", "disabled", "recalibration_required"}: raise ValueError
binding = rules["profile_binding"]
if binding != {"profile_id": profile["id"], "width": profile["width"], "height": profile["height"]}: raise ValueError
if not isinstance(rules["areas"], list) or not isinstance(rules["directional_lines"], list) or len(rules["areas"]) > 1024 or len(rules["directional_lines"]) > 1024: raise ValueError
identifiers: set[str] = set()
for area in rules["areas"]:
if not isinstance(area, dict) or set(area) != {"id", "version", "kind", "enabled", "points"} or area["kind"] != "danger_area" or not isinstance(area["enabled"], bool) or not 3 <= len(area["points"]) <= 256: raise ValueError
_rule_identity(area, identifiers); points = tuple(_point(value) for value in area["points"])
polygon = sum(a[0]*b[1]-b[0]*a[1] for a,b in zip(points, points[1:]+points[:1])) / 2
if abs(polygon) < 1e-9: raise ValueError
for line in rules["directional_lines"]:
if not isinstance(line, dict) or set(line) != {"id", "version", "kind", "enabled", "start", "end", "trigger_direction"} or line["kind"] != "directional_line" or not isinstance(line["enabled"], bool) or line["trigger_direction"] not in {"left_to_right", "right_to_left"}: raise ValueError
_rule_identity(line, identifiers)
if _point(line["start"]) == _point(line["end"]): raise ValueError
def _rule_identity(rule: Mapping[str, object], identifiers: set[str]) -> None:
if not isinstance(rule["id"], str) or not _ID.fullmatch(rule["id"]) or rule["id"] in identifiers or isinstance(rule["version"], bool) or not isinstance(rule["version"], int) or rule["version"] < 1: raise ValueError
identifiers.add(rule["id"])
def _point(value: object) -> tuple[float, float]:
if not isinstance(value, dict) or set(value) != {"x", "y"}: raise ValueError
x, y = value["x"], value["y"]
if isinstance(x, bool) or isinstance(y, bool) or not isinstance(x, (int,float)) or not isinstance(y,(int,float)) or not 0 <= x <= 1 or not 0 <= y <= 1: raise ValueError
return float(x), float(y)
def _map(document: Mapping[str, object]) -> AppliedConfig:
profile, rules = document["profile"], document["rule_set"]
rule_set = None
if rules["state"] == "active":
rule_set = RuleSet(
version=rules["version"], profile_id=profile["id"], width=profile["width"], height=profile["height"],
areas=tuple(AreaDefinition(a["id"], tuple(NormalizedPoint(**p) for p in a["points"])) for a in rules["areas"] if a["enabled"]),
directional_lines=tuple(DirectionalLineDefinition(l["id"], NormalizedPoint(**l["start"]), NormalizedPoint(**l["end"]), l["trigger_direction"]) for l in rules["directional_lines"] if l["enabled"]),
)
return AppliedConfig(document["config_id"], document["revision"], document["logical_device"]["id"], document["media"]["ref"], profile["encoding"], float(profile["frame_rate"]), rules["state"], rule_set)
def _timestamp(value: object) -> float:
if not isinstance(value, str) or not value.endswith("Z"): raise ValueError
return datetime.fromisoformat(value[:-1] + "+00:00").astimezone(timezone.utc).timestamp()
def _contains_secret(value: object) -> bool:
if isinstance(value, dict): return any(_SECRET.search(str(k)) or _contains_secret(v) for k,v in value.items())
if isinstance(value, list): return any(_contains_secret(item) for item in value)
if isinstance(value, str):
split=urlsplit(value)
return bool(split.username or split.password or value.startswith("file:") or re.match(r"^[A-Za-z]:[\\/]", value))
return False
@@ -0,0 +1,44 @@
"""Brain-owned durable replay storage; never shared with Sense or Bell."""
from __future__ import annotations
import sqlite3
import threading
from contextlib import closing
from pathlib import Path
class SQLiteReplayStore:
def __init__(self, path: str | Path) -> None:
self._path = str(path)
self._lock = threading.Lock()
with closing(self._connect()) as connection:
connection.execute("PRAGMA journal_mode=WAL")
connection.execute(
"CREATE TABLE IF NOT EXISTS machine_replay (principal TEXT NOT NULL, token_id TEXT NOT NULL, expires_at INTEGER NOT NULL, PRIMARY KEY(principal, token_id))"
)
def _connect(self) -> sqlite3.Connection:
connection = sqlite3.connect(self._path, timeout=5, isolation_level=None)
connection.execute("PRAGMA busy_timeout=5000")
return connection
def consume(self, principal: str, token_id: str, expires_at: int, now: int) -> bool:
if not principal or not token_id or expires_at <= now:
return False
with self._lock, closing(self._connect()) as connection:
try:
connection.execute("BEGIN IMMEDIATE")
connection.execute("DELETE FROM machine_replay WHERE expires_at <= ?", (now,))
connection.execute(
"INSERT INTO machine_replay(principal, token_id, expires_at) VALUES (?, ?, ?)",
(principal, token_id, expires_at),
)
connection.execute("COMMIT")
return True
except sqlite3.IntegrityError:
connection.execute("ROLLBACK")
return False
except Exception:
connection.execute("ROLLBACK")
return False
@@ -0,0 +1,77 @@
"""Persistent sequence allocation and safe runtime-status/v1 production."""
from __future__ import annotations
import json
import re
import sqlite3
import threading
import uuid
from contextlib import closing
from datetime import datetime, timezone
from pathlib import Path
from typing import Mapping, Sequence
_LOGICAL_REF = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._:-]{0,127}$")
_VERSION = re.compile(r"^[A-Za-z0-9][A-Za-z0-9._+-]{0,63}$")
_ERROR_CODE = re.compile(r"^[A-Z][A-Z0-9_]{2,63}$")
_RUNTIME_STATES = {"unconfigured", "starting", "running", "degraded", "failed", "stopped"}
class RuntimeStatusPublisher:
def __init__(self, state_path: str | Path, brain_instance_ref: str) -> None:
self._path, self._instance, self._lock = str(state_path), brain_instance_ref, threading.Lock()
with closing(self._connect()) as connection:
connection.execute("CREATE TABLE IF NOT EXISTS runtime_sequence(instance_ref TEXT PRIMARY KEY, sequence INTEGER NOT NULL)")
connection.commit()
def _connect(self) -> sqlite3.Connection:
return sqlite3.connect(self._path, timeout=5)
def build(self, *, runtime_state: str, runtime_version: str, started_at: datetime | None, model_ref: str, model_version: str, configurations: Sequence[Mapping[str, object]], health: Mapping[str, object], inputs: Sequence[Mapping[str, object]], observed_at: datetime | None = None) -> bytes:
_validate(runtime_state, runtime_version, self._instance, model_ref, model_version, configurations, health, inputs)
with self._lock, closing(self._connect()) as connection:
connection.execute("BEGIN IMMEDIATE")
row=connection.execute("SELECT sequence FROM runtime_sequence WHERE instance_ref=?",(self._instance,)).fetchone();sequence=(row[0]+1) if row else 0
connection.execute("INSERT INTO runtime_sequence(instance_ref,sequence) VALUES(?,?) ON CONFLICT(instance_ref) DO UPDATE SET sequence=excluded.sequence",(self._instance,sequence));connection.commit()
observed=(observed_at or datetime.now(timezone.utc)).astimezone(timezone.utc)
document={"schema_version":"yovision.runtime-status/v1","status_id":str(uuid.uuid4()),"brain_instance_ref":self._instance,"sequence":sequence,"observed_at":_utc(observed),"runtime":{"state":runtime_state,"version":runtime_version,"started_at":_utc(started_at) if started_at else None},"model":{"model_ref":model_ref,"version":model_version},"configurations":list(configurations),"health":dict(health),"inputs":list(inputs)}
raw=json.dumps(document,separators=(",",":"),sort_keys=True).encode()
lowered=raw.lower();
for marker in (b"password",b"credential",b"stream_uri",b"cookie",b"jwt",b"file://"):
if marker in lowered: raise ValueError("runtime status contains sensitive field")
return raw
def _utc(value: datetime) -> str:
if value.tzinfo is None: raise ValueError("runtime timestamp must be timezone-aware")
return value.astimezone(timezone.utc).isoformat(timespec="seconds").replace("+00:00","Z")
def _validate(runtime_state: str, runtime_version: str, instance: str, model_ref: str, model_version: str, configurations: Sequence[Mapping[str, object]], health: Mapping[str, object], inputs: Sequence[Mapping[str, object]]) -> None:
if runtime_state not in _RUNTIME_STATES or not _VERSION.fullmatch(runtime_version) or not _LOGICAL_REF.fullmatch(instance) or not _LOGICAL_REF.fullmatch(model_ref) or not _VERSION.fullmatch(model_version):
raise ValueError("invalid runtime identity or version")
if len(configurations) > 4096 or len(inputs) > 4096:
raise ValueError("runtime status collection too large")
seen: set[str] = set()
for item in configurations:
if set(item) != {"config_id", "apply_state", "applied_revision", "error_code"} or not isinstance(item["config_id"], str) or not _LOGICAL_REF.fullmatch(item["config_id"]) or item["config_id"] in seen:
raise ValueError("invalid configuration status")
seen.add(item["config_id"]); state=item["apply_state"]; revision=item["applied_revision"]; error=item["error_code"]
if state not in {"not_configured","applying","applied","rejected"} or (state=="not_configured" and revision is not None) or (state=="applied" and (isinstance(revision,bool) or not isinstance(revision,int) or revision<1)) or (state=="rejected" and (not isinstance(error,str) or not _ERROR_CODE.fullmatch(error))):
raise ValueError("invalid configuration status")
if set(health) != {"overall","error_codes","metrics"} or health["overall"] not in {"healthy","degraded","unhealthy"} or not _codes(health["error_codes"],32) or not _metrics(health["metrics"]):
raise ValueError("invalid health status")
for item in inputs:
if set(item) != {"input_ref","state","error_codes","metrics"} or not isinstance(item["input_ref"],str) or not _LOGICAL_REF.fullmatch(item["input_ref"]) or item["state"] not in _RUNTIME_STATES or not _codes(item["error_codes"],16) or not _metrics(item["metrics"]):
raise ValueError("invalid input status")
def _codes(value: object, limit: int) -> bool:
return isinstance(value,list) and len(value)<=limit and len(set(value))==len(value) and all(isinstance(code,str) and _ERROR_CODE.fullmatch(code) for code in value)
def _metrics(value: object) -> bool:
if not isinstance(value,Mapping) or set(value)!={"load_percent","queue_depth","latency_ms"}: return False
load,queue,latency=value["load_percent"],value["queue_depth"],value["latency_ms"]
return not isinstance(load,bool) and isinstance(load,(int,float)) and 0<=load<=100 and not isinstance(queue,bool) and isinstance(queue,int) and queue>=0 and not isinstance(latency,bool) and isinstance(latency,(int,float)) and latency>=0
@@ -0,0 +1,101 @@
from __future__ import annotations
import hashlib
import json
import threading
from datetime import datetime, timedelta, timezone
import pytest
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey
from yovision_brain.integration.machine_identity.token import KeyRecord, Registry, Signer, Verifier
from yovision_brain.integration.sense_control.connector import ConnectorResponse, SourceConfigEndpoint, StatusSender
from yovision_brain.integration.sense_control.consumer import SourceConfigConsumer, SourceConfigError
from yovision_brain.integration.sense_control.replay import SQLiteReplayStore
from yovision_brain.integration.sense_control.status import RuntimeStatusPublisher
def source_document(revision: int = 1, *, state: str = "active", effective: int = 0) -> bytes:
now=datetime(2026,8,31,tzinfo=timezone.utc)
document={"schema_version":"yovision.source-config/v1","config_id":"gate-primary","revision":revision,"published_at":now.isoformat().replace("+00:00","Z"),"effective_at":(now+timedelta(seconds=effective)).isoformat().replace("+00:00","Z"),"site":{"id":"site-east"},"logical_device":{"id":"camera-1"},"profile":{"id":"main","width":1920,"height":1080,"encoding":"H264","frame_rate":25},"media":{"ref":"media:site-east/camera-1/main","transport":"rtsp"},"rule_set":{"version":f"rules-{revision}","state":state,"profile_binding":{"profile_id":"main","width":1920,"height":1080},"areas":[{"id":"danger","version":1,"kind":"danger_area","enabled":True,"points":[{"x":.1,"y":.1},{"x":.8,"y":.1},{"x":.5,"y":.8}]}],"directional_lines":[]}}
digest=hashlib.sha256(json.dumps(document,separators=(",",":"),sort_keys=True).encode()).hexdigest();document["integrity"]={"algorithm":"sha256","value":digest}
return json.dumps(document,separators=(",",":"),sort_keys=True).encode()
def with_extension(body: bytes, extensions: object) -> bytes:
document=json.loads(body);document["extensions"]=extensions;unsigned=dict(document);unsigned.pop("integrity");document["integrity"]["value"]=hashlib.sha256(json.dumps(unsigned,separators=(",",":"),sort_keys=True).encode()).hexdigest();return json.dumps(document,separators=(",",":"),sort_keys=True).encode()
def identity(tmp_path, now: int):
private=Ed25519PrivateKey.generate();signer=Signer("yv:sense:east","sense-key-01",private,clock=lambda:now)
registry=Registry([KeyRecord("yv:sense:east","sense-key-01",private.public_key(),"yovision-brain",frozenset({"source-config:write"}))])
replay=SQLiteReplayStore(tmp_path/"replay.sqlite")
return signer,Verifier(registry,replay,clock=lambda:now)
def test_authenticated_apply_is_idempotent_and_replay_survives_restart(tmp_path):
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());signer,verifier=identity(tmp_path,now);body=source_document();consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now);endpoint=SourceConfigEndpoint(consumer,verifier)
token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body)
result=endpoint.receive("Bearer "+token,body,"corr-request-0001");assert result.state=="applied" and result.config.rules is not None
with pytest.raises(ValueError,match="machine_token_replayed"):endpoint.receive("Bearer "+token,body,"corr-request-0001")
restarted=SourceConfigEndpoint(SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now),Verifier(verifier._registry,SQLiteReplayStore(tmp_path/"replay.sqlite"),clock=lambda:now))
with pytest.raises(ValueError,match="machine_token_replayed"):restarted.receive("Bearer "+token,body,"corr-request-0002")
new_token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body);assert restarted.receive("Bearer "+new_token,body,"corr-request-0003").state=="idempotent"
def test_atomic_replay_accepts_once_under_concurrency(tmp_path):
store=SQLiteReplayStore(tmp_path/"atomic.sqlite");results=[]
threads=[threading.Thread(target=lambda:results.append(store.consume("yv:sense:east","token-id",200,100))) for _ in range(12)]
for thread in threads:thread.start()
for thread in threads:thread.join()
assert results.count(True)==1
def test_stale_unknown_profile_and_recalibration_are_safe(tmp_path):
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
assert consumer.apply(source_document(2)).config.rules is not None
with pytest.raises(SourceConfigError,match="STALE_REVISION"):consumer.apply(source_document(1))
invalid=json.loads(source_document(3));invalid["profile"]["width"]=1280;invalid["integrity"]["value"]="0"*64
with pytest.raises(SourceConfigError,match="CONFIG_INVALID"):consumer.apply(json.dumps(invalid).encode())
unknown=json.loads(source_document(3));unknown["schema_version"]="yovision.source-config/v2"
with pytest.raises(SourceConfigError,match="UNSUPPORTED_SCHEMA_VERSION"):consumer.apply(json.dumps(unknown).encode())
safe=consumer.apply(source_document(3,state="recalibration_required"));assert safe.config.rule_state=="recalibration_required" and safe.config.rules is None
def test_future_effective_snapshot_activates_atomically_after_restart(tmp_path):
base=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());current=[base];path=tmp_path/"state.sqlite";consumer=SourceConfigConsumer(path,clock=lambda:current[0])
assert consumer.apply(source_document(1)).state=="applied";scheduled=consumer.apply(source_document(2,effective=60));assert scheduled.state=="scheduled" and scheduled.config.revision==1
current[0]+=61;restarted=SourceConfigConsumer(path,clock=lambda:current[0]);activated=restarted.activate_due();assert activated[0].revision==2 and restarted.get_active("gate-primary").revision==2
def test_unknown_valid_extension_namespace_is_ignored(tmp_path):
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
result=consumer.apply(with_extension(source_document(),{"vendor.example":{"feature":"safe"}}));assert result.state=="applied" and result.config.revision==1
@pytest.mark.parametrize("extensions", [[], {"1invalid":{}}, {"vendor_ok":{}}, {"vendor.example":"not-an-object"}])
def test_invalid_extensions_are_rejected(tmp_path, extensions):
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());consumer=SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now)
with pytest.raises(SourceConfigError,match="CONFIG_INVALID"):consumer.apply(with_extension(source_document(),extensions))
def test_status_sequence_restart_retry_timeout_and_disable(tmp_path):
path=tmp_path/"status.sqlite";publisher=RuntimeStatusPublisher(path,"brain-east-01");health={"overall":"healthy","error_codes":[],"metrics":{"load_percent":1.0,"queue_depth":0,"latency_ms":2.0}}
one=json.loads(publisher.build(runtime_state="running",runtime_version="1.0.0",started_at=datetime.now(timezone.utc),model_ref="people-detection",model_version="1",configurations=[],health=health,inputs=[]));two=json.loads(RuntimeStatusPublisher(path,"brain-east-01").build(runtime_state="running",runtime_version="1.0.0",started_at=None,model_ref="people-detection",model_version="1",configurations=[],health=health,inputs=[]));assert (one["sequence"],two["sequence"])==(0,1)
private=Ed25519PrivateKey.generate();signer=Signer("yv:brain:east","brain-key-01",private,clock=lambda:1_787_000_000);attempts=[]
def send(_auth,_body,_corr,timeout):attempts.append(timeout);raise TimeoutError
sender=StatusSender(signer,send,max_attempts=3,sleeper=lambda _:None)
with pytest.raises(RuntimeError,match="STATUS_DELIVERY_EXHAUSTED"):sender.publish(b"{}","corr-request-0001")
assert attempts==[5.0,5.0,5.0]
disabled=StatusSender(signer,lambda *_:ConnectorResponse(204,b"","corr-request-0001"),enabled=False)
with pytest.raises(RuntimeError,match="CONNECTOR_DISABLED"):disabled.publish(b"{}","corr-request-0001")
@pytest.mark.parametrize("request_id", ["short", "0123456789abcde\n", "0123456789abcde!", "a"*129])
def test_connector_rejects_unsafe_request_ids(tmp_path, request_id):
now=int(datetime(2026,8,31,tzinfo=timezone.utc).timestamp());signer,verifier=identity(tmp_path,now);body=source_document();endpoint=SourceConfigEndpoint(SourceConfigConsumer(tmp_path/"state.sqlite",clock=lambda:now),verifier)
token=signer.mint("yovision-brain",("source-config:write",),"POST","/machine/v1/source-config",body)
with pytest.raises(ValueError,match="INVALID_CORRELATION_ID"):endpoint.receive("Bearer "+token,body,request_id)
status_signer=Signer("yv:brain:east","brain-key-01",Ed25519PrivateKey.generate(),clock=lambda:now)
sender=StatusSender(status_signer,lambda *_:ConnectorResponse(204,b"",request_id))
with pytest.raises(ValueError,match="INVALID_CORRELATION_ID"):sender.publish(b"{}",request_id)
@@ -0,0 +1,118 @@
package brain_control
import (
"errors"
"fmt"
"regexp"
"time"
mi "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
)
const (
sourceConfigPath = "/machine/v1/source-config"
runtimeStatusPath = "/machine/v1/runtime-status"
)
var requestIDPattern = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._:-]{15,127}$`)
type SendResponse struct {
StatusCode int
Body []byte
CorrelationID string
}
type SendFunc func(authorization string, body []byte, correlationID string, timeout time.Duration) (SendResponse, error)
type ConfigSender struct {
Signer mi.Signer
Send SendFunc
Enabled bool
Timeout time.Duration
MaxAttempts int
Sleep func(time.Duration)
}
func (s ConfigSender) Publish(config SourceConfig, correlationID string) (SendResponse, error) {
if !s.Enabled {
return SendResponse{}, errors.New("CONNECTOR_DISABLED")
}
if s.Send == nil || !requestIDPattern.MatchString(correlationID) {
return SendResponse{}, errors.New("invalid connector configuration")
}
if s.Timeout <= 0 {
s.Timeout = 5 * time.Second
}
if s.MaxAttempts == 0 {
s.MaxAttempts = 4
}
if s.MaxAttempts < 1 {
return SendResponse{}, errors.New("invalid connector retry policy")
}
if s.Sleep == nil {
s.Sleep = time.Sleep
}
if err := ValidateSourceConfig(config); err != nil {
return SendResponse{}, err
}
body, err := MarshalSourceConfig(config)
if err != nil {
return SendResponse{}, err
}
var last error
for attempt := 0; attempt < s.MaxAttempts; attempt++ {
token, mintErr := s.Signer.Mint("yovision-brain", []string{"source-config:write"}, "POST", sourceConfigPath, body)
if mintErr != nil {
return SendResponse{}, mintErr
}
response, sendErr := s.Send("Bearer "+token, body, correlationID, s.Timeout)
if sendErr == nil && response.StatusCode >= 200 && response.StatusCode < 300 {
return response, nil
}
if sendErr == nil && response.StatusCode < 500 {
return SendResponse{}, fmt.Errorf("source config rejected: %d", response.StatusCode)
}
if sendErr != nil {
last = sendErr
} else {
last = fmt.Errorf("source config remote status: %d", response.StatusCode)
}
if attempt+1 < s.MaxAttempts {
delay := time.Second << attempt
if delay > 30*time.Second {
delay = 30 * time.Second
}
s.Sleep(delay)
}
}
return SendResponse{}, fmt.Errorf("source config delivery exhausted: %w", last)
}
type RuntimeStatusEndpoint struct {
Verifier mi.Verifier
Store ProjectionStore
Enabled bool
MaxBodyBytes int
}
func (e RuntimeStatusEndpoint) Receive(authorization string, body []byte, correlationID string, expected map[string]int64) (ProjectionView, error) {
if !e.Enabled {
return ProjectionView{}, errors.New("CONNECTOR_DISABLED")
}
if !requestIDPattern.MatchString(correlationID) {
return ProjectionView{}, errors.New("INVALID_CORRELATION_ID")
}
if e.MaxBodyBytes == 0 {
e.MaxBodyBytes = 10 * 1024 * 1024
}
if len(body) > e.MaxBodyBytes {
return ProjectionView{}, errors.New("REQUEST_TOO_LARGE")
}
token, err := mi.BearerToken(authorization)
if err != nil {
return ProjectionView{}, err
}
if _, err = e.Verifier.Verify(token, "yovision-sense", "runtime-status:write", "POST", runtimeStatusPath, body); err != nil {
return ProjectionView{}, err
}
return e.Store.Ingest(body, expected)
}
@@ -0,0 +1,124 @@
package brain_control
import (
"crypto/ed25519"
"crypto/rand"
"encoding/json"
"strings"
"sync"
"testing"
"time"
mi "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func memoryDB(t *testing.T, name string) *gorm.DB {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+name+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&ReplayToken{}, &RuntimeProjection{}, &SourceRevision{}); err != nil {
t.Fatal(err)
}
return db
}
func TestSourceMapperAndRecalibration(t *testing.T) {
now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
facts := SourceFacts{ConfigID: "gate-primary", SiteID: "site-east", LogicalDeviceID: "camera-1", MediaPath: "site/camera/main", Revision: 1, PublishedAt: now, EffectiveAt: now, Profile: Profile{ID: "main", Width: 1920, Height: 1080, Encoding: "h264", FrameRate: 25}, RuleSetVersion: "rules-1", Areas: []AreaRule{{ID: "danger", Version: 1, Kind: "danger_area", Enabled: true, Points: []Point{{.1, .1}, {.8, .1}, {.5, .8}}}}}
config, err := MapSourceConfig(facts)
if err != nil {
t.Fatal(err)
}
if err = ValidateSourceConfig(config); err != nil {
t.Fatal(err)
}
facts.Revision = 2
facts.NeedsRecalibration = true
config, err = MapSourceConfig(facts)
if err != nil {
t.Fatal(err)
}
if config.RuleSet.State != "recalibration_required" || config.RuleSet.Areas[0].Enabled {
t.Fatal("recalibration did not disable rules")
}
}
func TestReplayAtomicAndRestart(t *testing.T) {
db := memoryDB(t, "replay-package")
pub, priv, _ := ed25519.GenerateKey(rand.Reader)
now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
registry, _ := mi.NewRegistry(mi.KeyRecord{Principal: "yv:brain:east", KeyID: "brain-key-01", PublicKey: pub, Audience: "yovision-sense", Scopes: []string{"runtime-status:write"}, Enabled: true})
signer := mi.Signer{Principal: "yv:brain:east", KeyID: "brain-key-01", PrivateKey: priv, Now: func() time.Time { return now }}
body := []byte("{}")
token, _ := signer.Mint("yovision-sense", []string{"runtime-status:write"}, "POST", "/machine/v1/runtime-status", body)
accepted := 0
var mu sync.Mutex
var wg sync.WaitGroup
for range 8 {
wg.Add(1)
go func() {
defer wg.Done()
v := mi.Verifier{Registry: registry, Replay: GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
if _, err := v.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body); err == nil {
mu.Lock()
accepted++
mu.Unlock()
}
}()
}
wg.Wait()
if accepted != 1 {
t.Fatalf("accepted %d", accepted)
}
v := mi.Verifier{Registry: registry, Replay: GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
if _, err := v.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body); err == nil {
t.Fatal("restart replay accepted")
}
}
func TestProjectionStaleRecoveryAndMismatch(t *testing.T) {
db := memoryDB(t, "projection-package")
now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
store := ProjectionStore{DB: db, Clock: func() time.Time { return now }, StaleAfter: 90 * time.Second}
raw := runtimeFixture("018f4d6a-8d1b-4a25-8b37-9085f9c0d101", 1, now, 2)
view, err := store.Ingest(raw, map[string]int64{"gate": 3})
if err != nil || !view.RevisionMismatch {
t.Fatalf("view %+v err %v", view, err)
}
now = now.Add(91 * time.Second)
view, _ = store.View("brain-east-01")
if !view.Offline {
t.Fatal("not offline")
}
raw = runtimeFixture("018f4d6a-8d1b-4a25-8b37-9085f9c0d102", 2, now, 3)
view, err = store.Ingest(raw, map[string]int64{"gate": 3})
if err != nil || !view.Recovered {
t.Fatalf("recovery %+v err %v", view, err)
}
}
func TestConnectorRejectsUnsafeRequestIDs(t *testing.T) {
for _, value := range []string{"short", "0123456789abcde\n", "0123456789abcde!", strings.Repeat("a", 129)} {
sender := ConfigSender{Enabled: true, Send: func(string, []byte, string, time.Duration) (SendResponse, error) {
t.Fatal("unsafe request id reached transport")
return SendResponse{}, nil
}}
if _, err := sender.Publish(SourceConfig{}, value); err == nil {
t.Fatalf("sender accepted request id %q", value)
}
endpoint := RuntimeStatusEndpoint{Enabled: true}
if _, err := endpoint.Receive("Bearer ignored", nil, value, nil); err == nil || err.Error() != "INVALID_CORRELATION_ID" {
t.Fatalf("endpoint accepted request id %q: %v", value, err)
}
}
}
func runtimeFixture(id string, seq int64, observed time.Time, revision int64) []byte {
value := map[string]any{"schema_version": RuntimeStatusVersion, "status_id": id, "brain_instance_ref": "brain-east-01", "sequence": seq, "observed_at": observed.Format(time.RFC3339), "runtime": map[string]any{"state": "running", "version": "1.0.0", "started_at": observed.Format(time.RFC3339)}, "model": map[string]any{"model_ref": "people", "version": "1"}, "configurations": []any{map[string]any{"config_id": "gate", "apply_state": "applied", "applied_revision": revision, "error_code": nil}}, "health": map[string]any{"overall": "healthy", "error_codes": []any{}, "metrics": map[string]any{"load_percent": 1.0, "queue_depth": 0, "latency_ms": 1.0}}, "inputs": []any{}}
raw, _ := json.Marshal(value)
return raw
}
@@ -0,0 +1,135 @@
package brain_control
import (
"crypto/sha256"
"encoding/hex"
"encoding/json"
"errors"
"fmt"
"regexp"
"strings"
"time"
)
var stableID = regexp.MustCompile(`^[A-Za-z0-9][A-Za-z0-9._~-]{0,127}$`)
func MapSourceConfig(f SourceFacts) (SourceConfig, error) {
if !stableID.MatchString(f.ConfigID) || !stableID.MatchString(f.SiteID) || !stableID.MatchString(f.LogicalDeviceID) || !stableID.MatchString(f.Profile.ID) || f.Revision < 1 || f.Profile.Width < 1 || f.Profile.Height < 1 || f.Profile.FrameRate <= 0 {
return SourceConfig{}, errors.New("invalid source configuration facts")
}
encoding := strings.ToUpper(f.Profile.Encoding)
if encoding != "H264" && encoding != "H265" && encoding != "MJPEG" {
return SourceConfig{}, errors.New("unsupported profile encoding")
}
if f.PublishedAt.IsZero() || f.EffectiveAt.Before(f.PublishedAt) {
return SourceConfig{}, errors.New("invalid source configuration time")
}
if !stableID.MatchString(f.RuleSetVersion) {
return SourceConfig{}, errors.New("invalid rule set version")
}
if strings.ContainsAny(f.MediaPath, "?#@\\") || strings.Contains(f.MediaPath, "://") || f.MediaPath == "" {
return SourceConfig{}, errors.New("media path must be opaque and credential-free")
}
if len(f.Areas) > 1024 || len(f.DirectionalLines) > 1024 {
return SourceConfig{}, errors.New("too many rules")
}
seen := map[string]bool{}
for _, a := range f.Areas {
if !stableID.MatchString(a.ID) || seen[a.ID] || a.Version < 1 || a.Kind != "danger_area" || len(a.Points) < 3 || len(a.Points) > 256 || !validPoints(a.Points) || polygonArea(a.Points) == 0 {
return SourceConfig{}, errors.New("invalid area rule")
}
seen[a.ID] = true
}
for _, l := range f.DirectionalLines {
if !stableID.MatchString(l.ID) || seen[l.ID] || l.Version < 1 || l.Kind != "directional_line" || (l.TriggerDirection != "left_to_right" && l.TriggerDirection != "right_to_left") || !validPoints([]Point{l.Start, l.End}) || l.Start == l.End {
return SourceConfig{}, errors.New("invalid directional line rule")
}
seen[l.ID] = true
}
var out SourceConfig
out.SchemaVersion, out.ConfigID, out.Revision = SourceConfigVersion, f.ConfigID, f.Revision
out.PublishedAt, out.EffectiveAt = f.PublishedAt.UTC(), f.EffectiveAt.UTC()
out.Site.ID, out.LogicalDevice.ID = f.SiteID, f.LogicalDeviceID
out.Profile = f.Profile
out.Profile.Encoding = encoding
out.Media.Ref, out.Media.Transport = "media:"+strings.TrimPrefix(f.MediaPath, "/"), "rtsp"
out.RuleSet.Version = f.RuleSetVersion
out.RuleSet.State = "active"
if f.Disabled {
out.RuleSet.State = "disabled"
}
if f.NeedsRecalibration {
out.RuleSet.State = "recalibration_required"
}
out.RuleSet.ProfileBinding.ProfileID, out.RuleSet.ProfileBinding.Width, out.RuleSet.ProfileBinding.Height = f.Profile.ID, f.Profile.Width, f.Profile.Height
out.RuleSet.Areas = append([]AreaRule(nil), f.Areas...)
out.RuleSet.DirectionalLines = append([]DirectionalLineRule(nil), f.DirectionalLines...)
if out.RuleSet.State != "active" {
for i := range out.RuleSet.Areas {
out.RuleSet.Areas[i].Enabled = false
}
for i := range out.RuleSet.DirectionalLines {
out.RuleSet.DirectionalLines[i].Enabled = false
}
}
digest, err := sourceDigest(out)
if err != nil {
return SourceConfig{}, err
}
out.Integrity.Algorithm, out.Integrity.Value = "sha256", digest
return out, nil
}
func MarshalSourceConfig(config SourceConfig) ([]byte, error) { return json.Marshal(config) }
func sourceDigest(config SourceConfig) (string, error) {
raw, err := json.Marshal(config)
if err != nil {
return "", err
}
var value map[string]any
if err = json.Unmarshal(raw, &value); err != nil {
return "", err
}
delete(value, "integrity")
canonical, err := json.Marshal(value)
if err != nil {
return "", err
}
sum := sha256.Sum256(canonical)
return hex.EncodeToString(sum[:]), nil
}
func validPoints(points []Point) bool {
for _, p := range points {
if p.X < 0 || p.X > 1 || p.Y < 0 || p.Y > 1 {
return false
}
}
return true
}
func polygonArea(p []Point) float64 {
var a float64
for i := range p {
n := p[(i+1)%len(p)]
a += p[i].X*n.Y - n.X*p[i].Y
}
if a < 0 {
a = -a
}
return a / 2
}
func ValidateSourceConfig(config SourceConfig) error {
if config.SchemaVersion != SourceConfigVersion {
return fmt.Errorf("unsupported source config version")
}
digest, err := sourceDigest(config)
if err != nil || config.Integrity.Algorithm != "sha256" || digest != config.Integrity.Value {
return errors.New("source config integrity mismatch")
}
_, err = MapSourceConfig(SourceFacts{ConfigID: config.ConfigID, SiteID: config.Site.ID, LogicalDeviceID: config.LogicalDevice.ID, MediaPath: strings.TrimPrefix(config.Media.Ref, "media:"), Revision: config.Revision, PublishedAt: config.PublishedAt, EffectiveAt: config.EffectiveAt, Profile: config.Profile, RuleSetVersion: config.RuleSet.Version, Disabled: config.RuleSet.State == "disabled", NeedsRecalibration: config.RuleSet.State == "recalibration_required", Areas: config.RuleSet.Areas, DirectionalLines: config.RuleSet.DirectionalLines})
return err
}
func UTCNow() time.Time { return time.Now().UTC() }
@@ -0,0 +1,122 @@
package brain_control
import "time"
const (
SourceConfigVersion = "yovision.source-config/v1"
RuntimeStatusVersion = "yovision.runtime-status/v1"
)
type Point struct {
X float64 `json:"x"`
Y float64 `json:"y"`
}
type Profile struct {
ID string `json:"id"`
Width int `json:"width"`
Height int `json:"height"`
Encoding string `json:"encoding"`
FrameRate float64 `json:"frame_rate"`
}
type AreaRule struct {
ID string `json:"id"`
Version int64 `json:"version"`
Kind string `json:"kind"`
Enabled bool `json:"enabled"`
Points []Point `json:"points"`
}
type DirectionalLineRule struct {
ID string `json:"id"`
Version int64 `json:"version"`
Kind string `json:"kind"`
Enabled bool `json:"enabled"`
Start Point `json:"start"`
End Point `json:"end"`
TriggerDirection string `json:"trigger_direction"`
}
type SourceConfig struct {
SchemaVersion string `json:"schema_version"`
ConfigID string `json:"config_id"`
Revision int64 `json:"revision"`
PublishedAt time.Time `json:"published_at"`
EffectiveAt time.Time `json:"effective_at"`
Site struct {
ID string `json:"id"`
} `json:"site"`
LogicalDevice struct {
ID string `json:"id"`
} `json:"logical_device"`
Profile Profile `json:"profile"`
Media struct {
Ref string `json:"ref"`
Transport string `json:"transport"`
} `json:"media"`
RuleSet struct {
Version string `json:"version"`
State string `json:"state"`
ProfileBinding struct {
ProfileID string `json:"profile_id"`
Width int `json:"width"`
Height int `json:"height"`
} `json:"profile_binding"`
Areas []AreaRule `json:"areas"`
DirectionalLines []DirectionalLineRule `json:"directional_lines"`
} `json:"rule_set"`
Integrity struct {
Algorithm string `json:"algorithm"`
Value string `json:"value"`
} `json:"integrity"`
}
// SourceFacts is an explicit, credential-free boundary DTO. Callers map their
// GORM entities into it; database models are never serialized as a contract.
type SourceFacts struct {
ConfigID, SiteID, LogicalDeviceID, MediaRouteID, MediaPath string
Revision int64
PublishedAt, EffectiveAt time.Time
Profile Profile
RuleSetVersion string
Disabled, NeedsRecalibration bool
Areas []AreaRule
DirectionalLines []DirectionalLineRule
}
type ReplayToken struct {
Principal string `gorm:"size:128;primaryKey"`
TokenID string `gorm:"size:96;primaryKey"`
ExpiresAt time.Time `gorm:"not null;index"`
CreatedAt time.Time `gorm:"not null"`
}
func (ReplayToken) TableName() string { return "sense_brain_runtime_replay_tokens" }
type RuntimeProjection struct {
BrainInstanceRef string `gorm:"size:128;primaryKey"`
StatusID string `gorm:"size:36;not null;uniqueIndex"`
Sequence int64 `gorm:"not null"`
ObservedAt time.Time `gorm:"not null;index"`
ReceivedAt time.Time `gorm:"not null"`
RuntimeState string `gorm:"size:32;not null"`
RuntimeVersion string `gorm:"size:64;not null"`
ModelRef string `gorm:"size:128;not null"`
ModelVersion string `gorm:"size:64;not null"`
HealthOverall string `gorm:"size:32;not null"`
ExpectedRevisionsJSON string `gorm:"type:jsonb;not null"`
ConfigurationsJSON string `gorm:"type:jsonb;not null"`
HealthJSON string `gorm:"type:jsonb;not null"`
InputsJSON string `gorm:"type:jsonb;not null"`
WasOffline bool `gorm:"not null;default:false"`
CreatedAt time.Time
UpdatedAt time.Time
}
func (RuntimeProjection) TableName() string { return "sense_brain_runtime_projections" }
type SourceRevision struct {
ConfigID string `gorm:"size:128;primaryKey"`
Revision int64 `gorm:"not null"`
UpdatedAt time.Time
}
func (SourceRevision) TableName() string { return "sense_brain_source_revisions" }
@@ -0,0 +1,145 @@
package brain_control
import (
"bytes"
"encoding/json"
"errors"
"io"
"regexp"
"time"
)
type runtimeStatus struct {
SchemaVersion string `json:"schema_version"`
StatusID string `json:"status_id"`
BrainInstanceRef string `json:"brain_instance_ref"`
Sequence int64 `json:"sequence"`
ObservedAt time.Time `json:"observed_at"`
Runtime struct {
State string `json:"state"`
Version string `json:"version"`
StartedAt *time.Time `json:"started_at"`
} `json:"runtime"`
Model struct {
ModelRef string `json:"model_ref"`
Version string `json:"version"`
} `json:"model"`
Configurations []configurationStatus `json:"configurations"`
Health healthStatus `json:"health"`
Inputs []inputStatus `json:"inputs"`
}
type configurationStatus struct {
ConfigID string `json:"config_id"`
ApplyState string `json:"apply_state"`
AppliedRevision *int64 `json:"applied_revision"`
ErrorCode *string `json:"error_code"`
}
type metrics struct {
LoadPercent float64 `json:"load_percent"`
QueueDepth int64 `json:"queue_depth"`
LatencyMS float64 `json:"latency_ms"`
}
type healthStatus struct {
Overall string `json:"overall"`
ErrorCodes []string `json:"error_codes"`
Metrics metrics `json:"metrics"`
}
type inputStatus struct {
InputRef string `json:"input_ref"`
State string `json:"state"`
ErrorCodes []string `json:"error_codes"`
Metrics metrics `json:"metrics"`
}
var uuid4 = regexp.MustCompile(`^[0-9a-f]{8}-[0-9a-f]{4}-4[0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$`)
var errCode = regexp.MustCompile(`^[A-Z][A-Z0-9_]{2,63}$`)
func parseRuntimeStatus(raw []byte) (runtimeStatus, error) {
var s runtimeStatus
d := json.NewDecoder(bytes.NewReader(raw))
d.DisallowUnknownFields()
if err := d.Decode(&s); err != nil {
return s, errors.New("CONFIG_INVALID")
}
if err := d.Decode(&struct{}{}); !errors.Is(err, io.EOF) {
return s, errors.New("CONFIG_INVALID")
}
if s.SchemaVersion != RuntimeStatusVersion {
return s, errors.New("UNSUPPORTED_SCHEMA_VERSION")
}
if !uuid4.MatchString(s.StatusID) || !stableID.MatchString(s.BrainInstanceRef) || s.Sequence < 0 || s.ObservedAt.IsZero() || !validState(s.Runtime.State) || s.Runtime.Version == "" || !stableID.MatchString(s.Model.ModelRef) || s.Model.Version == "" || len(s.Configurations) > 4096 || len(s.Inputs) > 4096 || !validHealth(s.Health) {
return s, errors.New("CONFIG_INVALID")
}
seen := map[string]bool{}
for _, c := range s.Configurations {
if !stableID.MatchString(c.ConfigID) || seen[c.ConfigID] || !validApply(c) {
return s, errors.New("CONFIG_INVALID")
}
seen[c.ConfigID] = true
}
for _, i := range s.Inputs {
if !stableID.MatchString(i.InputRef) || !validState(i.State) || !validMetrics(i.Metrics) || !validCodes(i.ErrorCodes, 16) {
return s, errors.New("CONFIG_INVALID")
}
}
return s, nil
}
func validState(v string) bool {
switch v {
case "unconfigured", "starting", "running", "degraded", "failed", "stopped":
return true
}
return false
}
func validApply(c configurationStatus) bool {
switch c.ApplyState {
case "not_configured":
return c.AppliedRevision == nil
case "applying":
return true
case "applied":
return c.AppliedRevision != nil && *c.AppliedRevision >= 1
case "rejected":
return c.ErrorCode != nil && errCode.MatchString(*c.ErrorCode)
}
return false
}
func validMetrics(m metrics) bool {
return m.LoadPercent >= 0 && m.LoadPercent <= 100 && m.QueueDepth >= 0 && m.LatencyMS >= 0
}
func validCodes(v []string, n int) bool {
if len(v) > n {
return false
}
seen := map[string]bool{}
for _, x := range v {
if seen[x] || !errCode.MatchString(x) {
return false
}
seen[x] = true
}
return true
}
func validHealth(h healthStatus) bool {
return (h.Overall == "healthy" || h.Overall == "degraded" || h.Overall == "unhealthy") && validCodes(h.ErrorCodes, 32) && validMetrics(h.Metrics)
}
func validRuntimeTransition(from, to string) bool {
if from == to {
return true
}
switch from {
case "unconfigured":
return to == "starting" || to == "stopped"
case "starting":
return to == "running" || to == "degraded" || to == "failed" || to == "stopped"
case "running":
return to == "degraded" || to == "failed" || to == "stopped"
case "degraded":
return to == "running" || to == "failed" || to == "stopped"
case "failed":
return to == "starting" || to == "stopped"
case "stopped":
return to == "starting"
}
return false
}
@@ -0,0 +1,159 @@
package brain_control
import (
"encoding/json"
"errors"
"time"
"gorm.io/gorm"
"gorm.io/gorm/clause"
)
type GORMReplayStore struct{ DB *gorm.DB }
func (s GORMReplayStore) Consume(principal, tokenID string, expiresAt, now time.Time) bool {
if s.DB == nil || principal == "" || tokenID == "" || !expiresAt.After(now) {
return false
}
return s.DB.Transaction(func(tx *gorm.DB) error {
if err := tx.Where("expires_at <= ?", now).Delete(&ReplayToken{}).Error; err != nil {
return err
}
result := tx.Clauses(clause.OnConflict{DoNothing: true}).Create(&ReplayToken{Principal: principal, TokenID: tokenID, ExpiresAt: expiresAt, CreatedAt: now})
if result.Error != nil {
return result.Error
}
if result.RowsAffected != 1 {
return errors.New("replayed")
}
return nil
}) == nil
}
type RevisionStore struct{ DB *gorm.DB }
func (s RevisionStore) Next(configID string) (int64, error) {
if s.DB == nil || !stableID.MatchString(configID) {
return 0, errors.New("invalid revision store")
}
var next int64
err := s.DB.Transaction(func(tx *gorm.DB) error {
var row SourceRevision
err := tx.Clauses(clause.Locking{Strength: "UPDATE"}).Where("config_id = ?", configID).First(&row).Error
if errors.Is(err, gorm.ErrRecordNotFound) {
row = SourceRevision{ConfigID: configID, Revision: 1}
if err = tx.Create(&row).Error; err != nil {
return err
}
next = 1
return nil
}
if err != nil {
return err
}
row.Revision++
next = row.Revision
return tx.Save(&row).Error
})
return next, err
}
type ProjectionStore struct {
DB *gorm.DB
Clock func() time.Time
StaleAfter time.Duration
FutureSkew time.Duration
}
type ProjectionView struct {
Projection RuntimeProjection
Offline, Stale, Recovered, RevisionMismatch bool
}
func (s ProjectionStore) Ingest(raw []byte, expected map[string]int64) (ProjectionView, error) {
if s.DB == nil {
return ProjectionView{}, errors.New("projection database required")
}
now := time.Now().UTC()
if s.Clock != nil {
now = s.Clock().UTC()
}
if s.StaleAfter == 0 {
s.StaleAfter = 90 * time.Second
}
if s.FutureSkew == 0 {
s.FutureSkew = 30 * time.Second
}
status, err := parseRuntimeStatus(raw)
if err != nil {
return ProjectionView{}, err
}
if status.ObservedAt.After(now.Add(s.FutureSkew)) {
return ProjectionView{}, errors.New("FUTURE_OBSERVATION")
}
var view ProjectionView
err = s.DB.Transaction(func(tx *gorm.DB) error {
var old RuntimeProjection
find := tx.Where("brain_instance_ref = ?", status.BrainInstanceRef).First(&old).Error
if find == nil {
if old.StatusID == status.StatusID {
view.Projection = old
return nil
}
if status.Sequence <= old.Sequence {
return errors.New("OUT_OF_ORDER_STATUS")
}
if !validRuntimeTransition(old.RuntimeState, status.Runtime.State) {
return errors.New("INVALID_STATUS_TRANSITION")
}
}
if find != nil && !errors.Is(find, gorm.ErrRecordNotFound) {
return find
}
expectedJSON, _ := json.Marshal(expected)
configs, _ := json.Marshal(status.Configurations)
health, _ := json.Marshal(status.Health)
inputs, _ := json.Marshal(status.Inputs)
p := RuntimeProjection{BrainInstanceRef: status.BrainInstanceRef, StatusID: status.StatusID, Sequence: status.Sequence, ObservedAt: status.ObservedAt, ReceivedAt: now, RuntimeState: status.Runtime.State, RuntimeVersion: status.Runtime.Version, ModelRef: status.Model.ModelRef, ModelVersion: status.Model.Version, HealthOverall: status.Health.Overall, ExpectedRevisionsJSON: string(expectedJSON), ConfigurationsJSON: string(configs), HealthJSON: string(health), InputsJSON: string(inputs), WasOffline: find == nil && now.Sub(old.ObservedAt) > s.StaleAfter}
if find == nil {
p.CreatedAt = old.CreatedAt
view.Recovered = p.WasOffline
}
if err := tx.Save(&p).Error; err != nil {
return err
}
view.Projection = p
return nil
})
if err != nil {
return ProjectionView{}, err
}
view.Stale = now.Sub(view.Projection.ObservedAt) > s.StaleAfter
view.Offline = view.Stale
applied := make(map[string]*int64, len(status.Configurations))
for _, c := range status.Configurations {
applied[c.ConfigID] = c.AppliedRevision
}
for configID, want := range expected {
got, ok := applied[configID]
if !ok || got == nil || *got != want {
view.RevisionMismatch = true
}
}
return view, nil
}
func (s ProjectionStore) View(instance string) (ProjectionView, error) {
var p RuntimeProjection
if err := s.DB.First(&p, "brain_instance_ref = ?", instance).Error; err != nil {
return ProjectionView{}, err
}
now := time.Now().UTC()
if s.Clock != nil {
now = s.Clock().UTC()
}
stale := s.StaleAfter
if stale == 0 {
stale = 90 * time.Second
}
return ProjectionView{Projection: p, Offline: now.Sub(p.ObservedAt) > stale, Stale: now.Sub(p.ObservedAt) > stale}, nil
}
@@ -0,0 +1,25 @@
package version
import (
"runtime"
"gorm.io/gorm"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/brain_control"
"git.ilapage.cn/ila/yovision/Sense/server/cmd/migrate/migration"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
)
func init() {
_, fileName, _, _ := runtime.Caller(0)
migration.Migrate.SetVersion(migration.GetFilename(fileName), migrateBrainRuntime)
}
func migrateBrainRuntime(db *gorm.DB, version string) error {
return db.Transaction(func(tx *gorm.DB) error {
if err := tx.AutoMigrate(&brain_control.ReplayToken{}, &brain_control.RuntimeProjection{}, &brain_control.SourceRevision{}); err != nil {
return err
}
return tx.Create(&common.Migration{Version: version}).Error
})
}
@@ -0,0 +1,28 @@
package version
import (
"testing"
"git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/brain_control"
common "git.ilapage.cn/ila/yovision/Sense/server/common/models"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func TestMigrateBrainRuntime(t *testing.T) {
db, err := gorm.Open(sqlite.Open("file:brain-runtime-migration?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&common.Migration{}); err != nil {
t.Fatal(err)
}
if err = migrateBrainRuntime(db, "2026083112000"); err != nil {
t.Fatal(err)
}
for _, model := range []any{&brain_control.ReplayToken{}, &brain_control.RuntimeProjection{}, &brain_control.SourceRevision{}} {
if !db.Migrator().HasTable(model) {
t.Fatalf("missing table for %T", model)
}
}
}
@@ -0,0 +1,138 @@
package brain_control_test
import (
"crypto/ed25519"
"crypto/rand"
"encoding/json"
"strings"
"sync"
"testing"
"time"
bc "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/brain_control"
mi "git.ilapage.cn/ila/yovision/Sense/server/app/sense/integration/machine_identity"
"gorm.io/driver/sqlite"
"gorm.io/gorm"
)
func testDB(t *testing.T, name string) *gorm.DB {
t.Helper()
db, err := gorm.Open(sqlite.Open("file:"+name+"?mode=memory&cache=shared"), &gorm.Config{})
if err != nil {
t.Fatal(err)
}
if err = db.AutoMigrate(&bc.ReplayToken{}, &bc.RuntimeProjection{}, &bc.SourceRevision{}); err != nil {
t.Fatal(err)
}
return db
}
func TestMapperProducesCredentialFreeFrozenContract(t *testing.T) {
now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
c, err := bc.MapSourceConfig(bc.SourceFacts{ConfigID: "gate-primary", SiteID: "site-east", LogicalDeviceID: "camera-1", MediaPath: "site-east/camera-1/main", Revision: 1, PublishedAt: now, EffectiveAt: now, Profile: bc.Profile{ID: "main", Width: 1920, Height: 1080, Encoding: "h264", FrameRate: 25}, RuleSetVersion: "rules-1", Areas: []bc.AreaRule{{ID: "danger", Version: 1, Kind: "danger_area", Enabled: true, Points: []bc.Point{{X: .1, Y: .1}, {X: .8, Y: .1}, {X: .5, Y: .8}}}}})
if err != nil {
t.Fatal(err)
}
if err = bc.ValidateSourceConfig(c); err != nil {
t.Fatal(err)
}
raw, _ := json.Marshal(c)
text := strings.ToLower(string(raw))
for _, secret := range []string{"password", "username", "rtsp://", "stream_uri", "credential"} {
if strings.Contains(text, secret) {
t.Fatalf("leaked %q", secret)
}
}
c2, err := bc.MapSourceConfig(bc.SourceFacts{ConfigID: "gate-primary", SiteID: "site-east", LogicalDeviceID: "camera-1", MediaPath: "site-east/camera-1/main", Revision: 2, PublishedAt: now, EffectiveAt: now, Profile: bc.Profile{ID: "main-v2", Width: 1280, Height: 720, Encoding: "H265", FrameRate: 20}, RuleSetVersion: "rules-2", NeedsRecalibration: true, Areas: []bc.AreaRule{{ID: "danger", Version: 2, Kind: "danger_area", Enabled: true, Points: []bc.Point{{X: .1, Y: .1}, {X: .8, Y: .1}, {X: .5, Y: .8}}}}})
if err != nil {
t.Fatal(err)
}
if c2.RuleSet.State != "recalibration_required" || c2.RuleSet.Areas[0].Enabled {
t.Fatal("recalibration must disable rules")
}
}
func TestDurableReplayIsAtomicAndSurvivesVerifierRestart(t *testing.T) {
db := testDB(t, "sense-replay")
pub, priv, _ := ed25519.GenerateKey(rand.Reader)
now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
registry, _ := mi.NewRegistry(mi.KeyRecord{Principal: "yv:brain:east", KeyID: "brain-key-01", PublicKey: pub, Audience: "yovision-sense", Scopes: []string{"runtime-status:write"}, Enabled: true})
signer := mi.Signer{Principal: "yv:brain:east", KeyID: "brain-key-01", PrivateKey: priv, Now: func() time.Time { return now }}
body := []byte(`{"ok":true}`)
token, _ := signer.Mint("yovision-sense", []string{"runtime-status:write"}, "POST", "/machine/v1/runtime-status", body)
results := make(chan bool, 8)
var wg sync.WaitGroup
for i := 0; i < 8; i++ {
wg.Add(1)
go func() {
defer wg.Done()
v := mi.Verifier{Registry: registry, Replay: bc.GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
_, err := v.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body)
results <- err == nil
}()
}
wg.Wait()
close(results)
accepted := 0
for ok := range results {
if ok {
accepted++
}
}
if accepted != 1 {
t.Fatalf("accepted=%d", accepted)
}
v2 := mi.Verifier{Registry: registry, Replay: bc.GORMReplayStore{DB: db}, Now: func() time.Time { return now }}
if _, err := v2.Verify(token, "yovision-sense", "runtime-status:write", "POST", "/machine/v1/runtime-status", body); err == nil {
t.Fatal("replay accepted after verifier restart")
}
}
func TestRevisionStoreConcurrentAndRestart(t *testing.T) {
db := testDB(t, "sense-revisions")
store := bc.RevisionStore{DB: db}
for want := int64(1); want <= 3; want++ {
got, err := store.Next("gate-primary")
if err != nil || got != want {
t.Fatalf("got %d err %v", got, err)
}
}
restarted := bc.RevisionStore{DB: db}
got, err := restarted.Next("gate-primary")
if err != nil || got != 4 {
t.Fatalf("restart got %d err %v", got, err)
}
}
func TestRuntimeProjectionStaleRecoveryMismatchAndOrdering(t *testing.T) {
db := testDB(t, "sense-projection")
now := time.Date(2026, 8, 31, 0, 0, 0, 0, time.UTC)
store := bc.ProjectionStore{DB: db, Clock: func() time.Time { return now }, StaleAfter: 90 * time.Second}
raw := statusJSON("018f4d6a-8d1b-4a25-8b37-9085f9c0d101", 41, now, "running", 20)
view, err := store.Ingest(raw, map[string]int64{"gate-primary": 21})
if err != nil {
t.Fatal(err)
}
if !view.RevisionMismatch || view.Stale {
t.Fatalf("bad initial view %+v", view)
}
now = now.Add(91 * time.Second)
view, err = store.View("brain-east-01")
if err != nil || !view.Offline || !view.Stale {
t.Fatalf("offline %+v %v", view, err)
}
raw = statusJSON("018f4d6a-8d1b-4a25-8b37-9085f9c0d102", 42, now, "running", 21)
view, err = store.Ingest(raw, map[string]int64{"gate-primary": 21})
if err != nil || !view.Recovered || view.RevisionMismatch {
t.Fatalf("recovery %+v %v", view, err)
}
if _, err = store.Ingest(statusJSON("018f4d6a-8d1b-4a25-8b37-9085f9c0d103", 41, now, "running", 21), nil); err == nil || err.Error() != "OUT_OF_ORDER_STATUS" {
t.Fatalf("expected ordering rejection: %v", err)
}
}
func statusJSON(id string, seq int64, observed time.Time, state string, revision int64) []byte {
v := map[string]any{"schema_version": bc.RuntimeStatusVersion, "status_id": id, "brain_instance_ref": "brain-east-01", "sequence": seq, "observed_at": observed.Format(time.RFC3339), "runtime": map[string]any{"state": state, "version": "1.0.0", "started_at": observed.Add(-time.Minute).Format(time.RFC3339)}, "model": map[string]any{"model_ref": "people-detection", "version": "2026.08.1"}, "configurations": []any{map[string]any{"config_id": "gate-primary", "apply_state": "applied", "applied_revision": revision, "error_code": nil}}, "health": map[string]any{"overall": "healthy", "error_codes": []any{}, "metrics": map[string]any{"load_percent": 1.0, "queue_depth": 0, "latency_ms": 2.0}}, "inputs": []any{}}
raw, _ := json.Marshal(v)
return raw
}
@@ -0,0 +1,18 @@
module git.ilapage.cn/ila/yovision/Sense/tests/integration/brain_control
go 1.26.5
require (
git.ilapage.cn/ila/yovision/Sense/server v0.0.0
gorm.io/driver/sqlite v1.6.0
gorm.io/gorm v1.31.2
)
require (
github.com/jinzhu/inflection v1.0.0 // indirect
github.com/jinzhu/now v1.1.5 // indirect
github.com/mattn/go-sqlite3 v1.14.49 // indirect
golang.org/x/text v0.40.0 // indirect
)
replace git.ilapage.cn/ila/yovision/Sense/server => ../../../server
@@ -0,0 +1,12 @@
github.com/jinzhu/inflection v1.0.0 h1:K317FqzuhWc8YvSVlFMCCUb36O/S9MCKRDI7QkRKD/E=
github.com/jinzhu/inflection v1.0.0/go.mod h1:h+uFLlag+Qp1Va5pdKtLDYj+kHp5pxUVkryuEj+Srlc=
github.com/jinzhu/now v1.1.5 h1:/o9tlHleP7gOFmsnYNz3RGnqzefHA47wQpKrrdTIwXQ=
github.com/jinzhu/now v1.1.5/go.mod h1:d3SSVoowX0Lcu0IBviAWJpolVfI5UJVZZ7cO71lE/z8=
github.com/mattn/go-sqlite3 v1.14.49 h1:B8jBHC3xhxZgxztrgruTuLucebnULQnx4W7cF7SAE9w=
github.com/mattn/go-sqlite3 v1.14.49/go.mod h1:6JTjA44L93a0QCyJef5YvlPoKXntQPjzWv5gtm9sB6w=
golang.org/x/text v0.40.0 h1:Ub2Z6/xjgF1WrYQz2nuITOEegKFtiIy+rieRJ5lHZKs=
golang.org/x/text v0.40.0/go.mod h1:hpnzDAfGV753zIKo+wk3u1bVKCGPbrnF7+7LBF/UHVY=
gorm.io/driver/sqlite v1.6.0 h1:WHRRrIiulaPiPFmDcod6prc4l2VGVWHz80KspNsxSfQ=
gorm.io/driver/sqlite v1.6.0/go.mod h1:AO9V1qIQddBESngQUKWL9yoH93HIeA1X6V633rBwyT8=
gorm.io/gorm v1.31.2 h1:3o8FXNo9v9S858gil+3LlZA1LkCOzgb4g5BL64FgaCo=
gorm.io/gorm v1.31.2/go.mod h1:XyQVbO2k6YkOis7C2437jSit3SsDK72s7n7rsSHd+Gs=
+40
View File
@@ -0,0 +1,40 @@
# YoVision 可选协调部署
本目录只组合已经验收的 Sense、Brain、Bell 交付入口,不复制产品实现,也不成为业务事实源。三个产品仍使用独立包、进程、端口、数据目录、日志、数据库角色、账户空间、浏览器 Cookie 和机器身份。
## 准备
1. 将 `coordination.example.json` 复制到仓库外受控配置目录,填写三个已审核交付包的精确版本与绝对路径。
2. 分别准备仓库外 `sense.env`、`brain.env`、`bell.env`。Sense 至少提供 `SENSE_DATABASE_URL`、`SENSE_JWT_SECRET`、`SENSE_PORT`,Bell 至少提供 `BELL_DATABASE_URL`、`BELL_JWT_SECRET`、`BELL_PORT`、`BELL_WEB_PORT`;其余配置(包括 connector 开关)仍遵循各产品自己的环境文件说明。不得在清单或仓库中填写密码、JWT、token 或私钥内容。
3. 为每个调用实例创建独立 Ed25519 私钥和消费者公钥注册表;清单只引用私钥路径。
4. 为 Sense、Bell 创建不同 PostgreSQL 数据库和角色;先使用各自迁移入口完成备份与迁移。
5. 确认清单声明的所有端口、包目录、数据目录和日志目录互不重叠。
编排器会创建并隔离清单中的数据、日志目录;各产品环境文件或产品配置还必须把自身持久化与业务日志指向对应目录。编排器不会猜测或改写产品内部配置。
示例清单中的地址、版本和标识都是不可直接投产的占位值。默认 16 路只是当前交付配额,不是编排器容量上限。
## 命令
从仓库根目录运行;不需要修改 PowerShell 执行策略:
```powershell
pwsh -NoProfile -File scripts/runtime/coordination/start-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -ValidateOnly
pwsh -NoProfile -File scripts/runtime/coordination/start-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -Product bell,sense,brain
pwsh -NoProfile -File scripts/runtime/coordination/status-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -Product all
pwsh -NoProfile -File scripts/runtime/coordination/stop-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -Product brain
```
`.bat` 包装器接受相同参数。启动时,`all` 只包含清单中 `enabled=true` 的产品;停止和查看状态时,`all` 会覆盖三个产品,避免产品被停用后遗留进程。启动顺序固定为 Bell → Sense → Brain,停止顺序固定为 Brain → Sense → Bell。单端失败只返回非零并清理该端新进程,不自动停止已经运行的其他端。
## 状态与日志
编排状态写入清单的 `runtime_root\state`,每端只保存 PID、启动时间、版本、命令路径和清单摘要,不保存环境变量值。协调启动日志分别写入各产品独立 `log_directory`。产品自己的日志仍由产品入口管理。
`status` 返回 `running`、`unhealthy`、`stopped` 或 `stale`;仅当所选产品全部健康运行时退出码为 0。PID 与命令归属不匹配时不停止进程,必须人工核对。
## 升级与回退
升级顺序为:备份 Sense/Bell → 迁移 Bell → 启动/检查 Bell → 迁移 Sense → 启动/检查 Sense → 更新 Brain → 启用 connector。每次只替换一个独立包并更新清单版本。
回退时先停用 Brain event export、Sense ingress/relay 与 Bell ingress/evidence connector,再按产品独立入口回退包或恢复各自数据库。不得删除 Sense Outbox、运行投影、Bell Receipt/Event、replay 或审计事实。根级编排故障时直接恢复三个产品原有独立入口。
@@ -0,0 +1,66 @@
{
"schema_version": "yovision.coordination/v1",
"deployment_id": "school-a-yovision",
"runtime_root": "C:\\YoVision\\runtime\\coordination",
"products": {
"sense": {
"enabled": true,
"version": "replace-with-reviewed-sense-artifact-version",
"package_root": "C:\\YoVision\\packages\\sense",
"environment_file": "C:\\YoVision\\secrets\\sense.env",
"data_directory": "C:\\YoVision\\data\\sense",
"log_directory": "C:\\YoVision\\logs\\sense",
"ports": [18080, 9997, 8889],
"browser_origin": "http://127.0.0.1:18080",
"cookie_name": "Sense-Admin-Token",
"account_namespace": "sense-users",
"database_id": "sense",
"database_role": "sense_app",
"start": { "executable": "scripts\\runtime\\start-sense.ps1", "arguments": ["-Mode", "production"] },
"stop": { "executable": "scripts\\runtime\\stop-sense.ps1", "arguments": ["-Mode", "production"] },
"health": { "kind": "http", "url": "http://127.0.0.1:18080/healthz", "timeout_seconds": 60 },
"identities": [
{ "principal": "yv:sense:school-a", "key_id": "sense-2026-01", "private_key_path": "C:\\YoVision\\secrets\\sense-to-bell.ed25519" }
]
},
"brain": {
"enabled": true,
"version": "replace-with-reviewed-brain-artifact-version",
"package_root": "C:\\YoVision\\packages\\brain",
"environment_file": "C:\\YoVision\\secrets\\brain.env",
"data_directory": "C:\\YoVision\\data\\brain",
"log_directory": "C:\\YoVision\\logs\\brain",
"ports": [18100],
"browser_origin": "",
"cookie_name": "",
"account_namespace": "brain-machine-only",
"database_id": "",
"database_role": "",
"start": { "executable": ".venv\\Scripts\\python.exe", "arguments": ["-m", "yovision_brain.app", "--config", "C:\\YoVision\\config\\brain.json", "--output", "-"] },
"health": { "kind": "process", "timeout_seconds": 15 },
"identities": [
{ "principal": "yv:brain:school-a", "key_id": "brain-2026-01", "private_key_path": "C:\\YoVision\\secrets\\brain-to-sense.ed25519" }
]
},
"bell": {
"enabled": true,
"version": "replace-with-reviewed-bell-artifact-version",
"package_root": "C:\\YoVision\\packages\\bell",
"environment_file": "C:\\YoVision\\secrets\\bell.env",
"data_directory": "C:\\YoVision\\data\\bell",
"log_directory": "C:\\YoVision\\logs\\bell",
"ports": [18090, 18091],
"browser_origin": "http://127.0.0.1:18091",
"cookie_name": "Bell-Admin-Token",
"account_namespace": "bell-users",
"database_id": "bell",
"database_role": "bell_app",
"start": { "executable": "scripts\\runtime\\start-bell.ps1", "arguments": [] },
"stop": { "executable": "scripts\\runtime\\stop-bell.ps1", "arguments": [] },
"health": { "kind": "http", "url": "http://127.0.0.1:18090/healthz", "timeout_seconds": 60 },
"identities": [
{ "principal": "yv:bell:school-a", "key_id": "bell-2026-01", "private_key_path": "C:\\YoVision\\secrets\\bell-to-sense.ed25519" }
]
}
}
}
@@ -0,0 +1,76 @@
{
"$schema": "https://json-schema.org/draft/2020-12/schema",
"$id": "https://yovision.local/schemas/coordination/v1",
"title": "YoVision coordination deployment manifest v1",
"type": "object",
"additionalProperties": false,
"required": ["schema_version", "deployment_id", "runtime_root", "products"],
"properties": {
"schema_version": { "const": "yovision.coordination/v1" },
"deployment_id": { "type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._-]{2,63}$" },
"runtime_root": { "type": "string", "minLength": 1 },
"products": {
"type": "object",
"additionalProperties": false,
"required": ["sense", "brain", "bell"],
"properties": {
"sense": { "$ref": "#/$defs/product" },
"brain": { "$ref": "#/$defs/product" },
"bell": { "$ref": "#/$defs/product" }
}
}
},
"$defs": {
"command": {
"type": "object",
"additionalProperties": false,
"required": ["executable", "arguments"],
"properties": {
"executable": { "type": "string", "minLength": 1 },
"arguments": { "type": "array", "items": { "type": "string" } }
}
},
"identity": {
"type": "object",
"additionalProperties": false,
"required": ["principal", "key_id", "private_key_path"],
"properties": {
"principal": { "type": "string", "pattern": "^yv:(sense|brain|bell):[A-Za-z0-9][A-Za-z0-9._-]{0,63}$" },
"key_id": { "type": "string", "pattern": "^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$" },
"private_key_path": { "type": "string", "minLength": 1 }
}
},
"product": {
"type": "object",
"additionalProperties": false,
"required": ["enabled", "version", "package_root", "environment_file", "data_directory", "log_directory", "ports", "browser_origin", "cookie_name", "account_namespace", "database_id", "database_role", "start", "health", "identities"],
"properties": {
"enabled": { "type": "boolean" },
"version": { "type": "string", "minLength": 1 },
"package_root": { "type": "string", "minLength": 1 },
"environment_file": { "type": "string", "minLength": 1 },
"data_directory": { "type": "string", "minLength": 1 },
"log_directory": { "type": "string", "minLength": 1 },
"ports": { "type": "array", "items": { "type": "integer", "minimum": 1, "maximum": 65535 }, "uniqueItems": true },
"browser_origin": { "type": "string" },
"cookie_name": { "type": "string" },
"account_namespace": { "type": "string", "minLength": 1 },
"database_id": { "type": "string" },
"database_role": { "type": "string" },
"start": { "$ref": "#/$defs/command" },
"stop": { "$ref": "#/$defs/command" },
"health": {
"type": "object",
"additionalProperties": false,
"required": ["kind", "timeout_seconds"],
"properties": {
"kind": { "enum": ["process", "http"] },
"url": { "type": "string" },
"timeout_seconds": { "type": "integer", "minimum": 1, "maximum": 300 }
}
},
"identities": { "type": "array", "items": { "$ref": "#/$defs/identity" } }
}
}
}
}
@@ -0,0 +1,175 @@
Set-StrictMode -Version 3.0
$ErrorActionPreference = 'Stop'
function Assert-True {
param([Parameter(Mandatory = $true)][bool]$Condition, [Parameter(Mandatory = $true)][string]$Message)
if (-not $Condition) { throw $Message }
}
function Get-FreePort {
$listener = [Net.Sockets.TcpListener]::new([Net.IPAddress]::Loopback, 0)
try { $listener.Start(); return ([Net.IPEndPoint]$listener.LocalEndpoint).Port } finally { $listener.Stop() }
}
function Write-Utf8File {
param([Parameter(Mandatory = $true)][string]$Path, [Parameter(Mandatory = $true)][AllowEmptyString()][string]$Content)
$directory = Split-Path -Parent $Path
if ($directory) { New-Item -ItemType Directory -Force -Path $directory | Out-Null }
[IO.File]::WriteAllText($Path, $Content, [Text.UTF8Encoding]::new($false))
}
$repositoryRoot = [IO.Path]::GetFullPath((Join-Path $PSScriptRoot '..\..\..'))
$scriptsRoot = Join-Path $repositoryRoot 'scripts\runtime\coordination'
$startScript = Join-Path $scriptsRoot 'start-yovision.ps1'
$stopScript = Join-Path $scriptsRoot 'stop-yovision.ps1'
$statusScript = Join-Path $scriptsRoot 'status-yovision.ps1'
$tempParent = [IO.Path]::GetFullPath([IO.Path]::GetTempPath())
$testRoot = Join-Path $tempParent ("yovision coordination-" + [guid]::NewGuid().ToString('N'))
$manifestPath = Join-Path $testRoot 'config\coordination.json'
$manifest = $null
try {
New-Item -ItemType Directory -Force -Path $testRoot,(Join-Path $testRoot 'markers') | Out-Null
$ports = @{ sense = Get-FreePort; brain = Get-FreePort; bell = Get-FreePort; bellWeb = Get-FreePort }
Assert-True (($ports.Values | Select-Object -Unique).Count -eq 4) 'Dynamic test ports are not unique.'
$products = [ordered]@{}
foreach ($name in @('sense', 'brain', 'bell')) {
$packageRoot = Join-Path $testRoot "packages\$name"
$start = Join-Path $packageRoot 'start.ps1'
$stop = Join-Path $packageRoot 'stop.ps1'
Write-Utf8File -Path $start -Content @'
Set-StrictMode -Version 3.0
$ErrorActionPreference = 'Stop'
if ($env:FAKE_STARTED_FILE) { [IO.File]::WriteAllText($env:FAKE_STARTED_FILE, $PID.ToString(), [Text.UTF8Encoding]::new($false)) }
while ($true) { Start-Sleep -Milliseconds 250 }
'@
Write-Utf8File -Path $stop -Content @'
Set-StrictMode -Version 3.0
$ErrorActionPreference = 'Stop'
if ($env:FAKE_STOPPED_FILE) { [IO.File]::WriteAllText($env:FAKE_STOPPED_FILE, 'stopped', [Text.UTF8Encoding]::new($false)) }
$ownedPID = 0
if (-not [int]::TryParse($env:YOVISION_COORDINATION_OWNED_PID, [ref]$ownedPID)) { exit 4 }
& taskkill.exe /PID $ownedPID /T /F 2>$null | Out-Null
exit 0
'@
$environmentPath = Join-Path $testRoot "secrets\$name.env"
$environment = "FAKE_STARTED_FILE=$(Join-Path $testRoot "markers\$name.started")`nFAKE_STOPPED_FILE=$(Join-Path $testRoot "markers\$name.stopped")`n"
if ($name -eq 'sense') { $environment += "SENSE_DATABASE_URL=postgres://sense_role@127.0.0.1/sense_db`nSENSE_JWT_SECRET=$(('s' * 40))`nSENSE_PORT=$($ports.sense)`n" }
if ($name -eq 'bell') { $environment += "BELL_DATABASE_URL=postgres://bell_role@127.0.0.1/bell_db`nBELL_JWT_SECRET=$(('b' * 40))`nBELL_PORT=$($ports.bell)`nBELL_WEB_PORT=$($ports.bellWeb)`n" }
Write-Utf8File -Path $environmentPath -Content $environment
$keyPath = Join-Path $testRoot "secrets\$name.ed25519"
Write-Utf8File -Path $keyPath -Content ([guid]::NewGuid().ToString('N'))
$productPorts = if ($name -eq 'sense') { @($ports.sense) } elseif ($name -eq 'bell') { @($ports.bell, $ports.bellWeb) } else { @($ports.brain) }
$products[$name] = [ordered]@{
enabled = $true; version = "test-$name-v1"; package_root = $packageRoot; environment_file = $environmentPath
data_directory = (Join-Path $testRoot "data\$name"); log_directory = (Join-Path $testRoot "logs\$name"); ports = $productPorts
browser_origin = $(if ($name -eq 'sense') { "http://127.0.0.1:$($ports.sense)" } elseif ($name -eq 'bell') { "http://127.0.0.1:$($ports.bellWeb)" } else { '' })
cookie_name = $(if ($name -eq 'sense') { 'Sense-Admin-Token' } elseif ($name -eq 'bell') { 'Bell-Admin-Token' } else { '' })
account_namespace = "$name-accounts"; database_id = $(if ($name -eq 'brain') { '' } else { "${name}_db" }); database_role = $(if ($name -eq 'brain') { '' } else { "${name}_role" })
start = [ordered]@{ executable = 'start.ps1'; arguments = @() }; stop = [ordered]@{ executable = 'stop.ps1'; arguments = @() }
health = [ordered]@{ kind = 'process'; timeout_seconds = 5 }
identities = @([ordered]@{ principal = "yv:${name}:test"; key_id = "${name}-test-key"; private_key_path = $keyPath })
}
}
$manifest = [ordered]@{ schema_version = 'yovision.coordination/v1'; deployment_id = 'test-coordination'; runtime_root = (Join-Path $testRoot 'runtime'); products = $products }
Write-Utf8File -Path $manifestPath -Content ($manifest | ConvertTo-Json -Depth 20)
& pwsh.exe -NoProfile -File $startScript -Manifest $manifestPath -ValidateOnly
Assert-True ($LASTEXITCODE -eq 0) 'Manifest validation failed.'
& pwsh.exe -NoProfile -File $startScript -Manifest $manifestPath -Product sense
Assert-True ($LASTEXITCODE -eq 0) 'Sense-only start failed.'
& pwsh.exe -NoProfile -File $statusScript -Manifest $manifestPath -Product sense -Json
Assert-True ($LASTEXITCODE -eq 0) 'Sense-only status is not healthy.'
& pwsh.exe -NoProfile -File $statusScript -Manifest $manifestPath -Product bell -Json
Assert-True ($LASTEXITCODE -eq 3) 'Stopped Bell status did not return exit code 3.'
& pwsh.exe -NoProfile -File $stopScript -Manifest $manifestPath -Product sense
Assert-True ($LASTEXITCODE -eq 0) 'Sense-only stop failed.'
$occupiedPort = [Net.Sockets.TcpListener]::new([Net.IPAddress]::Loopback, $ports.sense)
try {
$occupiedPort.Start()
& pwsh.exe -NoProfile -File $startScript -Manifest $manifestPath -Product sense 2>$null
Assert-True ($LASTEXITCODE -eq 1) 'An occupied Sense port was not rejected.'
Assert-True (-not (Test-Path -LiteralPath (Join-Path $testRoot 'runtime\state\sense.json'))) 'Occupied-port failure left Sense state behind.'
} finally {
$occupiedPort.Stop()
}
foreach ($standaloneProduct in @('brain', 'bell')) {
& pwsh.exe -NoProfile -File $startScript -Manifest $manifestPath -Product $standaloneProduct
Assert-True ($LASTEXITCODE -eq 0) "$standaloneProduct standalone start failed."
& pwsh.exe -NoProfile -File $statusScript -Manifest $manifestPath -Product $standaloneProduct -Json
Assert-True ($LASTEXITCODE -eq 0) "$standaloneProduct standalone status is not healthy."
& pwsh.exe -NoProfile -File $stopScript -Manifest $manifestPath -Product $standaloneProduct
Assert-True ($LASTEXITCODE -eq 0) "$standaloneProduct standalone stop failed."
}
& pwsh.exe -NoProfile -File $startScript -Manifest $manifestPath -Product brain,bell
Assert-True ($LASTEXITCODE -eq 0) 'Brain+Bell combination start failed.'
$bellStatePath = Join-Path $testRoot 'runtime\state\bell.json'
$bellStateText = Get-Content -LiteralPath $bellStatePath -Raw -Encoding UTF8
$foreignState = $bellStateText | ConvertFrom-Json
$foreignState.pid = $PID
Write-Utf8File -Path $bellStatePath -Content ($foreignState | ConvertTo-Json -Depth 10)
$ownershipStatus = & pwsh.exe -NoProfile -File $statusScript -Manifest $manifestPath -Product bell -Json
Assert-True ($LASTEXITCODE -eq 3 -and ($ownershipStatus -join "`n") -match 'ownership-mismatch') 'Foreign PID ownership was not diagnosed.'
& pwsh.exe -NoProfile -File $stopScript -Manifest $manifestPath -Product bell 2>$null
Assert-True ($LASTEXITCODE -eq 1 -and $null -ne (Get-Process -Id $PID -ErrorAction SilentlyContinue)) 'Stop did not protect an unrelated process.'
Write-Utf8File -Path $bellStatePath -Content $bellStateText
$manifestText = Get-Content -LiteralPath $manifestPath -Raw -Encoding UTF8
Write-Utf8File -Path $manifestPath -Content ($manifestText + "`n")
$driftStatus = & pwsh.exe -NoProfile -File $statusScript -Manifest $manifestPath -Product bell -Json
Assert-True ($LASTEXITCODE -eq 3 -and ($driftStatus -join "`n") -match 'manifest-drift') 'Manifest drift was not diagnosed.'
& pwsh.exe -NoProfile -File $stopScript -Manifest $manifestPath -Product bell
Assert-True ($LASTEXITCODE -eq 0) 'Owned Bell could not be stopped after manifest drift.'
Write-Utf8File -Path $manifestPath -Content $manifestText
& pwsh.exe -NoProfile -File $startScript -Manifest $manifestPath -Product bell
Assert-True ($LASTEXITCODE -eq 0) 'Bell restart after manifest restoration failed.'
& pwsh.exe -NoProfile -File $stopScript -Manifest $manifestPath -Product brain
Assert-True ($LASTEXITCODE -eq 0) 'Brain-only stop failed.'
& pwsh.exe -NoProfile -File $statusScript -Manifest $manifestPath -Product bell -Json
Assert-True ($LASTEXITCODE -eq 0) 'Stopping Brain damaged Bell.'
Write-Utf8File -Path (Join-Path $testRoot 'packages\sense\start.ps1') -Content 'exit 7'
& pwsh.exe -NoProfile -File $startScript -Manifest $manifestPath -Product sense 2>$null
Assert-True ($LASTEXITCODE -eq 1) 'A failing product start did not return exit code 1.'
& pwsh.exe -NoProfile -File $statusScript -Manifest $manifestPath -Product bell -Json
Assert-True ($LASTEXITCODE -eq 0) 'A Sense start failure damaged Bell.'
$badManifest = $manifest | ConvertTo-Json -Depth 20 | ConvertFrom-Json -Depth 20
$badManifest.products.bell.database_id = $badManifest.products.sense.database_id
$badPath = Join-Path $testRoot 'config\invalid-isolation.json'
Write-Utf8File -Path $badPath -Content ($badManifest | ConvertTo-Json -Depth 20)
& pwsh.exe -NoProfile -File $startScript -Manifest $badPath -ValidateOnly 2>$null
Assert-True ($LASTEXITCODE -eq 1) 'Shared database identity was not rejected.'
$badPortManifest = $manifest | ConvertTo-Json -Depth 20 | ConvertFrom-Json -Depth 20
$badPortManifest.products.bell.ports[0] = $badPortManifest.products.sense.ports[0]
$badBellEnvironmentPath = Join-Path $testRoot 'secrets\bell-duplicate-port.env'
$badBellEnvironment = Get-Content -LiteralPath $badPortManifest.products.bell.environment_file -Raw -Encoding UTF8
$badBellEnvironment = $badBellEnvironment -replace "(?m)^BELL_PORT=\d+$", "BELL_PORT=$($ports.sense)"
Write-Utf8File -Path $badBellEnvironmentPath -Content $badBellEnvironment
$badPortManifest.products.bell.environment_file = $badBellEnvironmentPath
$badPortPath = Join-Path $testRoot 'config\invalid-port-isolation.json'
Write-Utf8File -Path $badPortPath -Content ($badPortManifest | ConvertTo-Json -Depth 20)
& pwsh.exe -NoProfile -File $startScript -Manifest $badPortPath -ValidateOnly 2>$null
Assert-True ($LASTEXITCODE -eq 1) 'Shared product port was not rejected.'
& pwsh.exe -NoProfile -File $stopScript -Manifest $manifestPath -Product all
Assert-True ($LASTEXITCODE -eq 0) 'Final stop failed.'
& pwsh.exe -NoProfile -File $statusScript -Manifest $manifestPath -Product all -Json
Assert-True ($LASTEXITCODE -eq 3) 'Stopped combination did not report non-running status.'
foreach ($name in @('sense', 'brain', 'bell')) {
$statePath = Join-Path $testRoot "runtime\state\$name.json"
Assert-True (-not (Test-Path -LiteralPath $statePath)) "State was not cleaned for $name."
}
Write-Host 'Coordination smoke passed: validation, port isolation, independent start/stop, combination, failure isolation, ownership and cleanup.'
exit 0
} finally {
if ($manifest -and (Test-Path -LiteralPath $manifestPath)) { & pwsh.exe -NoProfile -File $stopScript -Manifest $manifestPath -Product all 2>$null | Out-Null }
$resolved = [IO.Path]::GetFullPath($testRoot)
if ($resolved.StartsWith($tempParent, [StringComparison]::OrdinalIgnoreCase) -and (Test-Path -LiteralPath $resolved)) { Remove-Item -LiteralPath $resolved -Recurse -Force }
}
+74 -29
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Architecture-and-Code-Map
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Architecture-and-Code-Map.-
wiki_revision: 41c2193b2f1edb37abe1e8994d65d02207549039
synchronized_at: 2026-08-31T03:18:12Z
wiki_revision: 0b760f2398ea4c1fd1d776b0cf869532b167c495
synchronized_at: 2026-09-01T04:08:05Z
<!-- gitea-wiki-mirror:end -->
# 架构与代码地图
@@ -298,51 +298,56 @@ Brain 解码层位于 `Brain/src/yovision_brain/decode/`,只依赖 #11 的内
<!-- brain-local-events-v1:end -->
<!-- sense-brain-contracts-v1:start -->
## Sense↔Brain v1 契约边界
## Sense↔Brain v1 connector 结构
- Sense→Brain 配置:`contracts/source-config/v1/source-config.schema.json`;版本 `yovision.source-config/v1`。
- Brain→Sense 状态:`contracts/runtime-status/v1/runtime-status.schema.json`;版本 `yovision.runtime-status/v1`。
- 共同测试:`contracts/tests/source-config-v1/`、`contracts/tests/runtime-status-v1/`。
- 生产者/消费者 mapper 责任分别记录在 `mapper-fields.md` 与 `mapping.md`;产品 adapter 后续由 #152 实现。
共享协议仍位于 `contracts/source-config/v1/**` 与 `contracts/runtime-status/v1/**`。#152 的产品实现位于:
数据流固定为:
- Sense:`Sense/server/app/sense/integration/brain_control/**`;运行投影迁移为 `2026083112000_brain_runtime.go`。
- Brain:`Brain/src/yovision_brain/integration/sense_control/**`。
- 双端集成测试:`Sense/tests/integration/brain_control/**`、`Brain/tests/integration/sense_control/**`。
数据流为:
```text
Sense Device/Profile/Area 内部事实
→ source-config/v1 mapper
→ Brain adapter(后续 #152)
→ Brain 内部配置与运行
→ runtime-status/v1 mapper
→ Sense 只读运维投影(后续 #152)
Sense Device/Profile/Area
→ source-config/v1 mapper + revision store
→ machine identity transport adapter
→ Brain strict consumer + last-known-good + persistent replay
→ runtime-status/v1 mapper + monotonic sequence
→ Sense read-only runtime projection + stale/offline/recovery
```
共享契约统一使用 snake_case 与 `schema_version: yovision.<contract>/v1`。源配置使用 `config_id + integer revision`;运行状态以 `configurations[]` 按 `config_id` 回报实际应用 revision。未知主版本、重复配置 ID、倒序状态、摘要失败或敏感字段必须拒绝,且不得覆盖最后已知有效配置/投影。
配置以 `config_id + integer revision` 唯一标识,状态以 Brain instance + sequence 保证时序。未知主版本、摘要失败、过期 revision、错误 Profile/几何或倒序状态拒绝且不覆盖最后有效事实。双端不共享数据库、用户会话、摄像头凭据或内部模型。
协议不得包含摄像头凭据、RTSP URL、query token、内部绝对路径、数据库模型、用户/JWT/Cookie 或 Bell Alert 语义。当前只冻结契约,没有新增网络端点、机器身份或跨端 connector。
#152 提供可注入传输 adapter 和持久恢复边界;根级进程编排及部署级 HTTP hosting 由 #154 绑定,#155 验证真实跨进程故障隔离。
<!-- sense-brain-contracts-v1:end -->
<!-- standard-event-evidence-v1:start -->
## 标准事件与证据 v1 契约边界
## 标准事件与证据 v1 connector 结构
- Event Schema:`contracts/events/v1/event.schema.json`,版本 `yovision.event/v1`。
- Bell 接入描述:`contracts/events/v1/openapi.json`,返回创建、重复、幂等冲突和不支持版本等明确结果。
- Evidence Schema/API:`contracts/evidence/v1/evidence-reference.schema.json`、`openapi.json`,版本 `yovision.evidence-reference/v1`。
- 共同测试:`contracts/tests/events-v1/`、`contracts/tests/evidence-v1/`。
共享协议仍位于 `contracts/events/v1/**`、`contracts/evidence/v1/**`,机器身份位于 `contracts/machine-identity/v1/**`。#153 的产品实现位于:
后续 #153 的映射流固定为:
- Brain producer:`Brain/src/yovision_brain/integration/event_export/**`,由 CLI `app/__main__.py` 按配置启用。
- Sense gateway/relay:`Sense/server/app/sense/integration/bell_connector/**`,由 API 启动链注册 `POST /v1/events`、`GET /v1/evidence/:evidence_id` 和 Bell Outbox Worker。
- Bell consumer:`Bell/server/app/bell/integration/event_ingress/**`,由 router registry 注册 `POST /v1/events`。
- 正式迁移:Sense/Bell 各自的 `2026083112000_*connector*.go` / `*event_ingress.go`。
默认数据流为:
```text
Brain internal candidate / Sense local event
→ yovision.event/v1 producer mapper
→ Sense Outbox relay(默认拓扑,保持原 producer/source ID)
→ Bell v1 ingress
→ Bell private immutable Event + permanent Receipt
Brain internal candidate
→ canonical yovision.event/v1
→ Sense authenticated ingress
→ Sense InboundEvent + EvidenceRecord + Bell Outbox(同事务)
→ HTTPS relay(原 payload/producer/source ID 不变)
→ Bell authenticated ingress
→ permanent Receipt + immutable Event + conflict audit
→ Bell private Rule / Alert / ack / close
```
规范载荷使用 RFC 8785 JCS 与 SHA-256 形成稳定摘要。同键同摘要返回原 Event;同键不同摘要返回冲突并审计,不覆盖原事实。Evidence 只提供逻辑引用与状态/完整性元数据,不授予访问权限,不包含本机路径、签名 URL 或凭据;取证授权由后续机器身份和 connector 工单实现。
规范载荷使用 RFC 8785 JCS 与 SHA-256。同键同摘要返回 duplicate;同键异摘要稳定冲突并审计。机器令牌 replay 与业务幂等分别持久化,传输重试签发新 jti,但不改变业务 ID。证据解析失败或不可用只更新独立降级状态,不允许上游写 Alert 语义。
Brain candidate、Sense candidate/Outbox 与 Bell Event/Receipt/Alert 继续是各自内部模型。当前没有新增可运行的跨端 ingress/relay,不能把冻结 Schema 解释为端到端链路已完成。
Sense 与 Bell 只使用各自默认数据库;连接器关闭时不注册相应入口/Worker,三端继续独立启动。根级部署、真实 PKI/网络和最终 E2E 属于 #154/#155。
<!-- standard-event-evidence-v1:end -->
<!-- machine-identity-v1:start -->
@@ -354,3 +359,43 @@ v1 使用 HTTPS 上的 Ed25519 短期请求绑定 JWS。每个部署实例拥有
Sense/Bell 使用 Go 标准库 Ed25519,Brain 冻结 `cryptography==50.0.1`。固定跨语言向量证明 Go/Python 可互相验签。#151 只提供身份、注册表、传输策略及可注入 replay 接口;#152/#153 才注册业务 endpoint,并必须使用各产品独立的持久原子 replay store验证重启,不能共享数据库。
<!-- machine-identity-v1:end -->
<!-- coordination-deployment-v1:start -->
## 可选协调部署层
工单 #154 已于 2026-08-31 验收。根级协调部署层位于 `deploy/coordination/**` 与 `scripts/runtime/coordination/**`,它只负责声明、校验和调用三个独立产品入口:
```text
仓库外 coordination.json
├─ Sense 独立包 / env / DB / 端口 / 数据 / 日志 / 机器身份
├─ Brain 独立包 / env / 端口 / 数据 / 日志 / 机器身份
└─ Bell 独立包 / env / DB / 端口 / 数据 / 日志 / 机器身份
↓
coordination-common.ps1
├─ 封闭清单与隔离断言
├─ 单端或选择性组合 start/stop/status
├─ 健康检查、版本和清单摘要
└─ PID + 启动器 + 命令令牌归属保护
```
`coordination.schema.json` 定义版本 `yovision.coordination/v1`;`coordination.example.json` 只提供不可投产占位。公共实现 `coordination-common.ps1` 解析外部 env 数据但不执行其内容,校验产品命令位于各自包内、秘密路径位于仓库和包外,并拒绝路径、端口、数据库身份、JWT、Cookie 或机器身份复用。三个薄入口脚本分别调用公共实现,BAT 只透传参数和退出码。
协调层不拥有业务数据或契约,不共享用户表、JWT、Cookie、数据库内部模型、摄像头凭据或产品实现。它不替代 `contracts/**`,也不让任一产品成为另一产品的启动前置;connector 可关闭,三端核心能力继续独立运行。根级运行状态位于清单指定的 `runtime_root\state`,产品日志仍归各自日志目录和产品入口管理。
<!-- coordination-deployment-v1:end -->
<!-- bell-contact-schedule:start -->
## Bell 联系人与值班排班代码路径
后端调用链:
- `Bell/server/router/contact.go` → `app/bell/contact.Handler` → `app/bell/contact.Service` → 联系人、通道、验证事实与审计表。
- `Bell/server/router/duty_schedule.go` → `app/bell/duty_schedule.Handler` → `app/bell/duty_schedule.Service` → 值班组、成员、排班版本、时段、发布与临时替班表。
- `Bell/server/cmd/migrate/migration/version/2026090110000_contact_schedule.go` 创建业务表、约束、不可变事实触发器,以及 GoAdmin 菜单、API 和 Casbin 权限种子。
- `Bell/ui/src/api/bell/contact.js`、`duty-schedule.js` 与对应 `views/bell/` 页面复用 GoAdmin 的请求封装、布局、表格、表单、弹窗和权限指令。
安全边界:
- `BELL_CONTACT_CHANNEL_KEY` 是 Base64 编码的 32 字节 AES-256 密钥,只允许从进程环境读取,不得写入仓库或日志。
- 通道地址采用 AES-GCM 加密,API 模型不暴露密文字段;服务端只在受控发送路径按需解密。
- `Bell/server/cmd/api/server.go` 在 GoAdmin `LoggerToFile` 之前注册联系人通道请求体脱敏中间件,使操作日志只能接触固定脱敏内容。
<!-- bell-contact-schedule:end -->
+34 -2
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Business-Rules-and-Glossary
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Business-Rules-and-Glossary.-
wiki_revision: 7a91edf3ca35ade3e254937c4a68b53d816a3eea
synchronized_at: 2026-08-31T03:18:24Z
wiki_revision: 9be486cd2d3fe424cf389f65c2c868eacab6a020
synchronized_at: 2026-09-01T04:08:16Z
<!-- gitea-wiki-mirror:end -->
# 业务规则与术语
@@ -274,3 +274,35 @@ synchronized_at: 2026-08-31T03:18:24Z
- 轮换先登记新 `kid`,最多并存 24 小时,切换后移除旧 key;禁用 principal 或吊销 `kid` 对每次请求即时生效。
- 回退只能关闭 connector 并恢复三端独立运行,不得降级为明文、共享管理员身份、共享 JWT 或跳过签名/TLS 验证。
<!-- machine-identity-v1:end -->
<!-- integration-connectors-v1:start -->
## Connector 持久性与故障隔离规则
- **配置事实与运行投影分离**:Sense 的期望 revision 不等于 Brain 已应用 revision;只以 Brain runtime-status 的实际值更新只读投影。
- **最后有效配置**:Brain 对过期、未知版本、摘要错误或不安全配置拒绝应用,并保留 last-known-good;不得用失败输入覆盖当前配置。
- **机器重放与业务幂等分离**:每次网络重试必须使用新 jti;事件仍沿用原 `producer_id/source_event_id` 与规范载荷。
- **Sense 同事务网关**:Brain 事件只有在 Sense 的 InboundEvent、EvidenceRecord 与 Bell Outbox 同一事务成功后才算被 Sense 接受。
- **Bell 永久收据**:同键同摘要只形成一个 Receipt/Event;同键异摘要是终止冲突并追加脱敏审计,不能覆盖或删除原事实。
- **证据降级独立**:pending、failed、timeout、expired 不改变 Event 不可变性,也不自动 ack/close Alert。
- **停用规则**:关闭 connector 只停止新接入或投递;不得清空未投递 Outbox、持久 replay、Receipt、Event、证据元数据或审计。
- **独立运行**:Sense、Brain、Bell 不因对端未安装、离线或 connector 关闭而停止各自核心能力;不得以共享数据库/JWT/Cookie 规避故障隔离。
<!-- integration-connectors-v1:end -->
<!-- bell-contact-schedule:start -->
## Bell 联系人与排班术语
- **联系人(Contact)**:可参与 Bell 告警接收和值班安排的人员记录。
- **通知通道(Channel)**:联系人使用的消息到达方式;通道地址密文保存,对外只提供脱敏摘要。
- **通道验证事实(Channel Validation)**:一次独立、只追加的验证结果;“已验证”不等于“已启用”。
- **值班组(Duty Group)**:参与同一排班规则的一组联系人。
- **排班版本(Schedule Version)**:值班组在指定 IANA 时区下的一套完整周排班;发布后冻结,不原位修改。
- **轮值时段(Rotation Slot)**:排班版本内的连续时间段,包含不同的主值班与备值班联系人。
- **临时替班(Override)**:在已发布排班上的只追加例外事实,不改写原排班历史。
规则:
1. 发布前必须验证整周无空档、无重叠,并验证主备不同、成员有效及通知通道已启用且已验证。
2. 已发布排班、通道验证事实、审计事实和临时替班记录保持不可变;更正通过新版本或新事实完成。
3. 同一排班版本中的临时替班不能重叠。
4. 普通操作员只读;创建、修改、启停、验证、发布和替班由管理员执行。
<!-- bell-contact-schedule:end -->
+145 -5
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Local-Development-and-Verification
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Local-Development-and-Verification.-
wiki_revision: 771fdd92eb03e9dff7d3ae20b7cb89f1ca288959
synchronized_at: 2026-08-31T03:18:34Z
wiki_revision: e90d408200d7189771eeec35ae76a00837e2eedf
synchronized_at: 2026-09-01T04:08:26Z
<!-- gitea-wiki-mirror:end -->
# 本地开发与验证
@@ -652,7 +652,7 @@ python dev_scripts/harness.py check --strict
git diff --check
```
这些命令只验证冻结契约,不验证 #152 产品 adapter、真实网络传输、机器身份、现场断网恢复或端到端链路。
这些命令本身只验证冻结契约;#152 产品 adapter、持久 replay、停用/超时/恢复由本页后续 connector 测试覆盖,真实部署网络与端到端链路仍由 #154/#155 验证。
<!-- sense-brain-contracts-v1:end -->
<!-- standard-event-evidence-v1:start -->
@@ -677,7 +677,7 @@ python dev_scripts/harness.py check --strict
git diff --check
```
这些测试只验证冻结契约,不验证 #153 产品 mapper/relay/ingress、#151 机器身份、实际证据存储/授权、网络断线补投或跨项目 E2E。
这些命令本身只验证冻结契约;#153 产品 mapper/relay/ingress、机器身份、持久 replay、断线补投与证据降级由本页后续 connector 测试覆盖,真实证据存储、生产网络和跨项目 E2E 仍由 #154/#155 验证。
<!-- standard-event-evidence-v1:end -->
<!-- machine-identity-v1:start -->
@@ -704,5 +704,145 @@ cd ../..
Brain\.venv\Scripts\python.exe contracts\tests\machine-identity-v1\test_contract.py
```
这些测试只证明 #151 身份和传输基础。真实客户 PKI/网络、现场时钟漂移、业务 endpoint、断网补投以及持久 replay 重启恢复由 #152/#153/#155 验证;不得用进程内 replay store替代生产结论。
这些测试只证明 #151 身份和传输基础。#152/#153 已覆盖业务 adapter/endpoint、断网补投和持久 replay 重启恢复;真实客户 PKI/网络、现场时钟漂移与最终故障隔离仍由 #154/#155 验证,不得用进程内 replay store替代生产结论。
<!-- machine-identity-v1:end -->
<!-- integration-connectors-v1:start -->
## #152/#153 connector 验证
使用冻结 CPython 3.11.15 与 Go 1.26.5。从仓库根目录按受影响范围执行:
```powershell
$env:PYTHONPATH = (Resolve-Path Brain/src)
Brain\.venv\Scripts\python.exe -m pytest Brain/tests/integration/sense_control Brain/tests/integration/event_export Brain/tests/app/test_event_export_connector.py -q
cd Sense/server
go test ./...
go vet ./...
go test -race ./app/sense/integration/brain_control ./app/sense/integration/bell_connector ./app/sense/local_event ./cmd/api ./cmd/migrate/migration/version
cd ../../Sense/tests/integration/brain_control
go test -race ./...
cd ../bell_connector
go test -race ./...
cd ../../../../Bell/server
go test ./...
go vet ./...
go test -race ./app/bell/integration/event_ingress ./cmd/api ./cmd/migrate/migration/version
cd tests/integration/event_ingress
go test -race ./...
```
冻结契约与仓库检查:
```powershell
pwsh -NoProfile -File contracts/tests/source-config-v1/run.ps1
python contracts/tests/runtime-status-v1/test_contract.py
python contracts/tests/events-v1/test_contract.py
python contracts/tests/evidence-v1/test_contract.py
pwsh -NoProfile -File contracts/tests/machine-identity-v1/run.ps1
python -m unittest discover -s tests -v
python dev_scripts/harness.py check --strict
git diff --check
```
#152 覆盖 revision 幂等、last-known-good、单调状态、陈旧/离线/恢复、持久 replay、停用、超时和退避。#153 覆盖 Brain mapper、Sense 同事务 Outbox、Bell 永久 Receipt/Event、重复/冲突、证据降级、持久 replay、重启、断线恢复和多数据库默认库选择。
工单验收未使用客户 PKI、生产 PostgreSQL、真实三端网络或生产流量;这些结果只能由 #154 部署和 #155 E2E 补充。Sense 本地候选原子 Outbox 当前没有生产创建 caller,也不得据此声明本地产生链已完整接通。
<!-- integration-connectors-v1:end -->
<!-- coordination-deployment-v1:start -->
## 根级编排本地验证
工单 #154 的编排验证只使用仓库外临时目录和假交付包,不需要真实密码、数据库或客户 PKI。先运行清单和脚本验证:
```powershell
pwsh -NoProfile -File scripts/runtime/coordination/start-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -ValidateOnly
pwsh -NoLogo -NoProfile -File deploy/coordination/tests/coordination-smoke.ps1
```
冒烟测试必须返回 0,并覆盖 Sense、Brain、Bell 分别启动/状态/停止、Brain+Bell 选择性组合、停止单端不影响另一端、单端启动失败隔离、端口占用与重复端口拒绝、重复数据库身份拒绝、无归属 PID 保护、清单漂移诊断、退出码和状态清理。测试创建的目录必须解析在系统临时目录内,结束时只清理该测试目录。
仓库闭环验证:
```powershell
python dev_scripts/harness.py check --strict
python -m unittest discover -s tests -v
python dev_scripts/harness.py sync --check
git diff --check
```
这些测试证明编排和隔离控制,不证明真实产品包、生产 PostgreSQL、客户 PKI、真实摄像头/GPU、供应商服务、容量或长稳表现。真实契约闭环与 Brain/Bell 离线、重启、重复/冲突及证据降级由后续 E2E 工单验证。
<!-- coordination-deployment-v1:end -->
<!-- coordination-e2e-v1:start -->
## 三项目协调 E2E 验证
从仓库根目录执行唯一默认验收入口:
```powershell
pwsh scripts/e2e/coordination/run-coordination-e2e.ps1
```
可复制的工具参数:
```powershell
pwsh scripts/e2e/coordination/run-coordination-e2e.ps1 -PostgresBin D:\pgsql17\bin -Python Brain\.venv\Scripts\python.exe
```
默认完整运行按以下阶段串行收敛:
1. 创建动态 loopback 端口的临时 PostgreSQL,为 Sense/Bell 建立不同的随机 owner/database,并执行 Bell 正式迁移。
2. 在一次性 Python venv 中验证 source-config、runtime-status、machine-identity、events、evidence 五类冻结 v1 契约。
3. 验证 Sense 配置/状态、Brain 事件接入/证据/Outbox 与 PostgreSQL 并发恢复。
4. 验证 Brain 配置/状态 connector、匿名事件导出和持久重放。
5. 串行验证 Bell ingress/evidence、Rule/Alert 投影和 ack/close 生命周期。共享有状态 Bell 数据库的包不得并行运行。
6. 默认重新运行 Sense isolated E2E、Brain 全套测试和 Bell isolated E2E。
7. 核对 Sense/Bell database owner/database 不同、随机秘密未进入临时日志、所属端口/进程已停止,并安全清理专属临时目录。
#155 的完整通过记录为:source-config 11 项、runtime-status 14 项、machine-identity 10 项、events 7 项、evidence 4 项;Brain connector/event export 25 项、Brain 全套 74 项;Sense/Bell connector 与独立 isolated E2E 均通过。不同提交必须以当次真实输出为准,不复用这些数量冒充新结果。
只有默认完整入口退出码为 0、末行出现 `COORDINATION_E2E passed`、三端独立回归均成功并且工作区无计划外改动,才满足协调 E2E 技术验收。`-SkipIndependentProductE2E` 仅供 harness 调试,不能作为 #155/MVP 验收。`-KeepTemporary` 仅用于失败诊断,保留内容不得提交或共享。
仓库级闭环继续执行:
```powershell
python dev_scripts/harness.py check --strict
python -m unittest discover -s tests -v
python dev_scripts/harness.py sync --check
git diff --check
```
已验证范围不包含真实 GPU/生产模型、真实摄像头、通知供应商、生产迁移、16 路长稳和客户现场效果;这些项目必须在对应环境和独立工单中验收。
<!-- coordination-e2e-v1:end -->
<!-- bell-contact-schedule:start -->
## 验证 Bell 联系人与值班排班
后端与 PostgreSQL 集成验证:
```powershell
Set-Location Bell/server
$env:GOTOOLCHAIN = 'go1.26.5'
go test ./... -count=1
pwsh ./tests/bell_contact_schedule/run-postgres.ps1
```
前端验证:
```powershell
Set-Location Bell/ui
corepack pnpm lint
corepack pnpm build:prod
```
联系人通道加密需要在运行进程中配置 `BELL_CONTACT_CHANNEL_KEY`。它必须是 Base64 编码的 32 字节随机值,不得提交到 Git、Wiki、工单或日志。仅为当前 PowerShell 进程生成测试密钥:
```powershell
$keyBytes = [Security.Cryptography.RandomNumberGenerator]::GetBytes(32)
$env:BELL_CONTACT_CHANNEL_KEY = [Convert]::ToBase64String($keyBytes)
```
PostgreSQL 脚本会验证密文保存、写接口不回显地址、验证与启用状态分离、过期版本冲突、完整周覆盖、已发布历史不可修改、临时替班不重叠、不可变事实约束、操作员只读权限,以及操作日志不含通道明文。当前范围不连接真实外部消息提供商。
<!-- bell-contact-schedule:end -->
+74 -2
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Troubleshooting
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Troubleshooting
wiki_revision: 75bf70f375f58356e968b45139b54fcae391d1e8
synchronized_at: 2026-08-31T03:18:53Z
wiki_revision: eec4beb8e5b9b8ff18401b272fd249aff7711509
synchronized_at: 2026-08-31T13:00:41Z
<!-- gitea-wiki-mirror:end -->
# 故障排查
@@ -176,3 +176,75 @@ synchronized_at: 2026-08-31T03:18:53Z
排错日志只记录稳定错误码、已认证 principal/kid 和 correlation ID;不得记录令牌、签名、密钥、完整 Authorization header 或秘密路径。无法安全恢复时关闭 connector,三端保持独立运行。
<!-- machine-identity-v1:end -->
<!-- integration-connectors-v1:start -->
## #152/#153 connector 排错
| 现象/错误 | 检查 | 安全处理 |
|---|---|---|
| connector 启用后提示迁移缺失 | 核对 Sense/Bell `2026083112000_*` 迁移记录和默认数据库 | 停止 connector,先备份并执行正式迁移;不依赖 AutoMigrate |
| 配置被 Brain 拒绝 | 核对 schema 主版本、JCS/SHA-256、config_id/revision、Profile/几何和 recalibration 状态 | 修复新 revision;保留 last-known-good,不回写旧 revision |
| Sense 显示 Brain 陈旧/离线或 revision mismatch | 核对 Brain sequence、observed_at、实际 applied_revision 和传输错误码 | 修复时钟/传输/配置;不手工改只读投影 |
| Sense Outbox 持续积压 | 核对 Bell ingress 开关、HTTPS、机器身份、available_at、lease、attempt 与脱敏错误 | 恢复 Bell 后等待幂等补投;不清空消息或改业务 ID |
| Bell 返回 duplicate | 同一 producer/source ID 与相同规范摘要已接收 | 视为成功;不得创建新 source_event_id |
| Bell 返回 idempotency conflict | 同一 producer/source ID 对应不同摘要 | 终止重试并调查 producer;保留原 Receipt/Event 与冲突审计 |
| evidence unavailable/timeout/expired | 核对 Sense evidence endpoint、`evidence:read` 权限、owner_id、HTTPS 和过期时间 | 保持降级状态;不把失败伪装 success,不改 Alert 生命周期 |
| 启动出现重复路由或错误数据库 | 检查 Sense 是否使用默认 DB、是否重复启动实例 | 每实例只注册一次;不让 connector 随机选择 secondary DB |
| Request-ID 或机器令牌拒绝 | 检查 16–128 字符 Request-ID、方法/路径/正文绑定、audience/scope/kid 和时钟 | 新签发 token/jti;保持原业务幂等键,不记录 Authorization |
| 对端离线导致本端无法启动 | connector 被错误配置成强依赖 | 关闭对应开关并恢复独立运行;保留持久事实后另行排查 |
日志只记录稳定错误码、request/correlation ID、已认证 principal/kid 和脱敏业务引用;不得记录私钥、令牌、完整 Authorization、摄像头凭据、内部证据路径或事件完整敏感载荷。
<!-- integration-connectors-v1:end -->
<!-- coordination-deployment-v1:start -->
## 根级协调编排排错
工单 #154 已于 2026-08-31 验收;本节适用于已验收的可选根级编排。
先使用与启动时相同的仓库外清单查询状态:
```powershell
pwsh -NoProfile -File scripts/runtime/coordination/status-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -Product all
```
| 现象/状态 | 检查 | 安全处理 |
|---|---|---|
| 清单校验失败 | 检查包、env、私钥路径是否存在且位于仓库和包外;检查三端端口、目录、数据库、Cookie、JWT、账户和身份是否独立 | 修正仓库外清单或环境文件;不得放宽隔离断言或把秘密写进仓库 |
| `port ... is already in use` | 用状态命令确认是否已有受管实例,再检查对应监听端口 | 先按归属停止旧实例或为产品分配独立端口;不得终止未知进程 |
| `stopped` | 没有该产品状态文件 | 按需单独启动;不代表其他产品异常 |
| `unhealthy` | 进程仍归属本实例,但 HTTP/进程健康检查失败 | 查看该产品独立 `coordination.err.log`、`coordination.out.log` 和产品日志;修复该端,不自动重启其他端 |
| `stale` / `ownership-mismatch` | PID 已复用、启动器或命令令牌与状态不符 | 不会停止该进程;人工核对进程与状态文件,确认归属后再处理 |
| `stale` / `manifest-drift` | 运行中的实例来自不同清单摘要 | 使用原清单安全停止,或确认归属后停止再以新清单启动;不得用新清单覆盖运行事实 |
| 单端启动失败 | 查看该端协调日志、产品日志和退出码 | 编排只清理该端新进程;确认其他端状态,修复失败端后单独重试 |
| connector 使对端成为启动强依赖 | connector 开关或产品配置错误 | 关闭对应 event export、ingress、relay 或 evidence connector,恢复三端独立运行;保留 Outbox/Receipt/Event 等持久事实 |
| 停止命令拒绝执行 | 状态归属不匹配,或产品停止入口返回非零 | 不使用无条件 taskkill;先核对 PID、启动器、命令令牌和产品停止日志 |
日志和状态不得包含环境变量值、密码、JWT、token、私钥、完整 Authorization、摄像头凭据或客户数据。协调层故障时可停止使用根级入口并恢复三个产品各自的已验收启动脚本,不删除数据或共享事实。
<!-- coordination-deployment-v1:end -->
<!-- coordination-e2e-v1:start -->
## 三项目协调 E2E 排错
先从完整输出定位第一个失败阶段,不同时修改多个猜测原因。默认入口是:
```powershell
pwsh scripts/e2e/coordination/run-coordination-e2e.ps1
```
| 现象 | 检查 | 安全处理 |
|---|---|---|
| 缺少 `initdb.exe`、`pg_ctl.exe`、`createdb.exe` 或 `psql.exe` | 检查 `-PostgresBin` 是否指向同一 PostgreSQL 安装的 `bin` | 修正工具路径;不要改用生产数据库或默认 5432 |
| Python 缺少契约依赖 | 确认所选 Python 可创建 venv 且能安装仓库冻结依赖 | 修复 Python/依赖源后重试;不要把依赖临时装进产品环境充当固定基线 |
| Bell 正式迁移失败 | 查看当轮临时目录的 `bell-migrate.log`,核对首个 SQLSTATE | 修复迁移或工具链;不得跳过迁移、手工补表或连接业务库 |
| Sense/Bell 数据库隔离断言失败 | 检查生成的 owner/database 是否不同 | 停止测试;不得共享数据库、角色、JWT、Cookie 或账户空间 |
| Bell Rule/Alert 评估数偶发增加 | 检查是否把 ingress、rule、lifecycle 多个有状态 Go 包放在同一数据库并行运行 | 恢复 root harness 的串行阶段;不要降低断言或清除不可变事实 |
| `machine_token_*`、duplicate 或 conflict 与预期不符 | 核对 audience/scope/kid、请求绑定、jti 和 producer/source 业务键 | 保持稳定错误和原业务键;不得记录 Authorization、复用网页登录态或为重试改 source ID |
| Outbox 在 Bell 离线后未补投 | 核对 retry/lease/available_at、Bell 恢复和新 relay 进程 | 保留消息和历史,恢复 Bell 后重试;不得清空 Outbox/Receipt/Event |
| evidence timeout/unavailable 导致整条事件失败 | 检查 evidence resolver 与降级状态 | Event/Receipt/Alert 应继续保留;不得把失败伪装为 success |
| 端口仍占用或测试结束后有进程 | 确认进程是否由本轮测试启动,查看所属临时目录和状态 | 只停止能证明归属的进程;不得无条件 taskkill 或终止其他实例 |
| 秘密扫描失败 | 在本轮临时日志中搜索该随机测试值,禁止把值粘贴到工单 | 修复日志输出后重新运行;不得仅关闭扫描 |
| 需要保留失败现场 | 使用 `-KeepTemporary` 并记录明确绝对临时路径 | 目录只用于本机受控诊断,完成后按已核对路径清理;不得提交或共享 |
| 使用 `-SkipIndependentProductE2E` 后通过 | 只证明协调 harness 主体,不证明三端独立回归 | 修复依赖后重新运行默认完整入口,不能据此通过 MVP 验收 |
若失败发生在 Sense、Brain 或 Bell 的独立 E2E,转到对应产品章节按该端首个错误排查;不得在协调文档工单中顺手修改产品代码。真实 GPU、真机、通知供应商、生产迁移、16 路长稳或客户现场问题不属于该隔离 E2E 的结论。
<!-- coordination-e2e-v1:end -->
+26 -10
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Product-Requirements
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Product-Requirements.-
wiki_revision: c9970b0ee8b677b6be13f31af67b3206a3faf956
synchronized_at: 2026-08-31T02:00:00Z
wiki_revision: a2aada9db75909dc5fc85cc5d91c8c8bcab82c45
synchronized_at: 2026-09-01T04:09:23Z
<!-- gitea-wiki-mirror:end -->
# 产品需求
@@ -274,21 +274,37 @@ BRN-002 的首个解码阶段已通过工单 #13 验收。Brain 通过可替换
<!-- brain-local-events-delivery:end -->
<!-- sense-brain-contracts-v1:start -->
## Sense↔Brain 首批冻结契约
## Sense↔Brain 配置与状态 connector
工单 #148、#149 已于 2026-08-31 通过用户验收并合入 `dev`。Sense→Brain 源/规则配置的唯一共享事实源为 `contracts/source-config/v1/`,版本标识为 `yovision.source-config/v1`;Brain→Sense 运行状态的唯一共享事实源为 `contracts/runtime-status/v1/`,版本标识为 `yovision.runtime-status/v1`。
工单 #148、#149 冻结的唯一共享事实源仍为 `contracts/source-config/v1/`(`yovision.source-config/v1`)和 `contracts/runtime-status/v1/`(`yovision.runtime-status/v1`)。工单 #152 已于 2026-08-31 通过用户验收并合入 `dev`。
源配置按 `config_id + integer revision` 形成不可复用的配置流,携带逻辑站点/设备/Profile、无凭据媒体引用、归一化区域/方向线、规则版本与完整性摘要。运行状态按同一 `config_id` 在 `configurations[]` 中报告实际应用 revision,并包含 Brain 实例、运行/模型版本、健康、输入和稳定错误码。
Sense 现在可从 Device/Profile/Area 内部事实生成无凭据完整快照并持久分配 revision;Brain 严格校验版本、JCS/SHA-256、Profile/几何/状态,幂等应用并保留 last-known-good。Brain 以持久单调 sequence 发布实际应用 revision、运行/模型、健康和输入状态;Sense 保存只读投影并区分在线、陈旧、离线、恢复和 revision mismatch。双端持久 replay store 在各自数据库内防止机器令牌重放,重启不清空安全状态。
这两项只冻结协议和测试,不表示 #152 connector 已实现。Sense 与 Brain 仍可独立运行;Brain 不读取 Sense 数据库,Sense 不读取 Brain 内部状态。既有 `brain.internal.*`、Sense GORM 模型和运维投影继续是项目内部实现,不得直接作为共享协议。
connector 可停用,断线或对端未安装不阻断 Sense/Brain 核心能力。#152 交付产品 adapter、持久状态和恢复原语;部署级 HTTP 托管、进程编排、真实双机 TLS 与最终 E2E 仍属于 #154/#155。
<!-- sense-brain-contracts-v1:end -->
<!-- standard-event-evidence-v1:start -->
## 标准事件与证据引用冻结契约
## 标准事件、证据与可靠 connector
工单 #150 已于 2026-08-31 通过用户验收并合入 `dev`。Sense/Brain→Bell 标准匿名安全事件的唯一共享事实源为 `contracts/events/v1/`,版本标识 `yovision.event/v1`;证据逻辑引用的唯一共享事实源为 `contracts/evidence/v1/`,版本标识 `yovision.evidence-reference/v1`。
工单 #150 冻结的唯一共享事实源仍为 `contracts/events/v1/`(`yovision.event/v1`)和 `contracts/evidence/v1/`(`yovision.evidence-reference/v1`)。工单 #153 已于 2026-08-31 通过用户验收并合入 `dev`。
事件以原始 `(producer_id, source_event_id)` 永久幂等,Sense relay 不改变原始身份或业务载荷。事件只携带逻辑站点/设备/Profile、事件类型、发生时间、规则/模型版本、匿名观测、区域和证据逻辑引用,不携带用户会话、摄像头凭据、内部路径、人脸特征或 Alert/ack/close 状态。
Brain 将内部匿名候选映射为标准事件;默认拓扑由 Sense 以机器身份接收并在同一事务中写入 InboundEvent、证据元数据和 Bell Outbox,再由后台 Worker 可靠投递到 Bell。Bell 以持久机器令牌 replay、永久 `(producer_id, source_event_id)` Receipt、JCS/SHA-256 摘要和冲突审计写入不可变 Event;Alert/ack/close 仍只由 Bell 私有规则与处置流程产生。
证据状态为 `pending/processing/success/failed`;`success` 必须包含内容类型和 SHA-256 完整性元数据,失败或过期只降级证据,不改写不可变 Event 或 Bell Alert 生命周期。此工单只冻结契约和测试,#153 可靠 connector、机器身份、证据存储与实际授权取证尚未实现。
证据继续使用逻辑引用,状态为 `pending/processing/success/failed`;失败、超时或过期只形成明确降级,不暴露内部路径、长期签名 URL 或凭据,也不改写 Event/Alert 生命周期。三端 connector 均可停用并保持独立运行;停用或故障时保留 Outbox、Receipt、Event、replay 与审计事实。
Sense 的 `local_event.CreateWithOutbox` 是本地候选与标准事件 Outbox 的正式原子写入口;当前仓库尚无生产本地候选创建调用链,不把不存在的上游路径声明为已接通。真实客户 PKI、生产 PostgreSQL、现场网络与最终故障隔离由 #154/#155 验证。
<!-- standard-event-evidence-v1:end -->
<!-- bell-contact-schedule:start -->
## Bell 联系人和值班排班
Bell 的告警接收对象和值班安排由工单 #183 建立,目标是让管理员维护可审计、可验证且不泄露通道地址的通知基础数据。
- 联系人与通知通道分离管理;通道地址只以密文保存,API、操作日志和页面均不得返回或记录明文,只展示脱敏摘要。
- 通道“已启用”和“已验证”是两个独立状态;验证结果以只追加事实记录保存,不能用启用状态替代验证。
- 值班组成员必须是已启用联系人,发布前主值班与备值班必须是不同联系人,且均具有已启用、已验证的通知通道。
- 周排班使用 IANA 时区,必须覆盖完整一周且不允许时间空档或重叠;排班以版本发布,已发布版本不可修改。
- 临时替班作为只追加事实记录保存,不能覆盖或改写已发布历史,且同一排班版本内不得时间重叠。
- 普通操作员只读,管理员可维护与发布。
- 本范围只提供联系人、通道验证事实和值班排班基础能力,不执行真实外部通知;实际发送属于后续工单 #185。
<!-- bell-contact-schedule:end -->
+84 -4
View File
@@ -2,8 +2,8 @@
generated: true (请先修改 Gitea Wiki,禁止直接编辑本文件)
wiki_page: Deployment-and-Operations
wiki_url: https://git.ilapage.cn/ila/yovision/wiki/Deployment-and-Operations.-
wiki_revision: 54d720aa07d2c40684da3fc04378393c7f318a04
synchronized_at: 2026-08-31T03:20:15Z
wiki_revision: c376d69227aa159a7839a829f5bfb5a7beaf89e6
synchronized_at: 2026-08-31T13:12:00Z
<!-- gitea-wiki-mirror:end -->
# YoVision 部署与运维
@@ -146,11 +146,91 @@ Sense\start_sense.bat
<!-- machine-identity-v1:start -->
## 机器身份部署与轮换边界
#151 已冻结机器身份和传输基础,但 #152/#153 尚未注册业务 connector,因此当前不得手工拼接 endpoint 或临时共享凭据提前打通。
#151 已冻结机器身份和传输基础,#152/#153 已验收业务 adapter/connector;运行时仍必须使用正式配置、迁移和独立机器身份,不得手工拼接临时共享凭据。
部署时为每个调用实例独立生成 Ed25519 私钥,保存到仓库外受 ACL/秘密存储保护的位置;产品配置只记录私钥路径、principal、kid、目标 audience 和最小 scope。消费者从仓库外公钥注册表读取受信 principal/kid。不得把私钥、完整令牌、Authorization header、管理员密码、浏览器 JWT/Cookie 或 query token写入配置样例、日志、工单和备份。
传输固定使用 HTTPS,TLS 最低 1.2并验证证书链与主机名。轮换按“先登记新公钥 → 调用方切换新 kid → 验证流量 → 24 小时内移除旧 key”执行;应急吊销直接禁用 principal 或 key,并同时停用相关 connector。回退保持 Sense、Brain、Bell 独立运行,保留 Outbox、Receipt、Event 和最后已知状态。
#152/#153 必须为各产品注入自己的持久原子 `(principal,jti)` replay store并验证重启;内存 replay store 只用于适配测试或不重启的单进程原语。
#152/#153 已为各产品接入独立持久原子 `(principal,jti)` replay store并覆盖重启恢复;内存 replay store 仍只用于适配测试或不重启的单进程原语。
<!-- machine-identity-v1:end -->
<!-- integration-connectors-v1:start -->
## #152/#153 connector 部署与回退
升级前先备份 Sense、Bell 各自 PostgreSQL,并在停服窗口执行正式迁移:
- Sense:`2026083112000_brain_runtime.go`、`2026083112000_bell_connector.go`。
- Bell:`2026083112000_event_ingress.go`。
迁移缺失时 connector 必须拒绝启动,不得依赖运行时 AutoMigrate 临时补表。Sense 只使用默认数据库注册 ingress 和 Worker,Bell 使用自己的默认数据库;不得指向共享数据库。
Brain 的标准事件出口在自身 JSON 配置的 `event_export` 对象中启用。必须配置 HTTPS origin、producer/site/severity、仓库外私钥路径、独立 principal/kid 和严格 transport policy;关闭时配置只保留 `{"enabled": false}`,CLI 恢复原 JSON Lines 独立输出。endpoint 只能是 HTTPS origin,固定追加 `/v1/events`。
Sense 运行变量:
- `SENSE_EVENT_INGRESS_ENABLED`:注册 Brain→Sense `POST /v1/events` 与证据读取入口。
- `SENSE_MACHINE_PRINCIPAL_REGISTRY`:仓库外 Brain/Bell 公钥注册表。
- `SENSE_EVIDENCE_OWNER_ID`:Sense 证据所有者逻辑标识。
- `SENSE_BELL_CONNECTOR_ENABLED`:启动 Bell Outbox Worker。
- `SENSE_BELL_ENDPOINT`、`SENSE_RELAY_ID`:Bell HTTPS origin 与 relay 实例标识。
- `SENSE_BELL_PRINCIPAL_ID`、`SENSE_BELL_KEY_ID`、`SENSE_BELL_PRIVATE_KEY_PATH`:Sense→Bell 独立机器身份。
- `SENSE_BELL_RELAY_INTERVAL_MS`:100–60000 ms;未配置时 2000 ms。
Bell 运行变量:
- `BELL_EVENT_INGRESS_ENABLED`:注册 `POST /v1/events`。
- `BELL_MACHINE_PRINCIPAL_REGISTRY`:仓库外 producer/relay 公钥注册表。
- `BELL_EVIDENCE_RESOLVER_ENABLED`:启用 Bell→Sense 证据解析。
- `BELL_SENSE_EVIDENCE_ENDPOINT`、`BELL_SENSE_PRINCIPAL_ID`、`BELL_SENSE_KEY_ID`、`BELL_SENSE_PRIVATE_KEY_PATH`:Sense HTTPS origin 与 Bell 独立证据读取身份。
推荐启动顺序:完成两端迁移 → 启动 Bell ingress → 启动 Sense ingress/relay → 启用 Brain event_export。#152 的配置/状态 adapter 已有持久状态、重放与恢复接口,部署级 HTTP 托管和根级进程编排在 #154 统一绑定,不能用临时共享文件或数据库替代。
回退时关闭 Brain `event_export.enabled`、Sense 两个 connector 开关和 Bell ingress/evidence 开关;保留 last-known-good、运行投影、InboundEvent、EvidenceRecord、Outbox、ReplayToken、Receipt、Event 与审计。不得删除事实、关闭 TLS/验签或改用网页登录态。
<!-- integration-connectors-v1:end -->
<!-- coordination-deployment-v1:start -->
## 三项目可选根级部署编排
工单 #154 已于 2026-08-31 验收。根级编排只组合 Sense、Brain、Bell 已审核交付包,不复制产品实现,也不改变三端独立交付边界。事实入口如下:
- 清单 Schema:`deploy/coordination/coordination.schema.json`
- 无秘密示例:`deploy/coordination/coordination.example.json`
- 操作说明:`deploy/coordination/README.md`
- 启动、停止、状态入口:`scripts/runtime/coordination/{start,stop,status}-yovision.ps1`,并提供同名 BAT 包装器
生产或验收环境必须先把示例清单复制到仓库外受控目录,并分别准备仓库外 Sense、Brain、Bell 环境文件和每实例独立私钥。Sense/Bell 使用不同数据库、数据库角色、账户空间、浏览器 origin、Cookie、JWT、端口、包、数据目录和日志目录;三端机器 principal、key id 和私钥文件也不得复用。清单只记录版本、路径和公开标识,不保存密码、JWT、token 或私钥内容。
从仓库根目录使用:
```powershell
pwsh -NoProfile -File scripts/runtime/coordination/start-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -ValidateOnly
pwsh -NoProfile -File scripts/runtime/coordination/start-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -Product bell,sense,brain
pwsh -NoProfile -File scripts/runtime/coordination/status-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -Product all
pwsh -NoProfile -File scripts/runtime/coordination/stop-yovision.ps1 -Manifest C:\YoVision\config\coordination.json -Product brain
```
启动顺序为 Bell → Sense → Brain,停止顺序反向。启动时 `all` 只包含 `enabled=true` 的产品;停止和状态查询时 `all` 覆盖三端,避免停用配置后遗留进程。编排状态只保存 PID、版本、清单摘要和命令归属元数据;停止前必须核对 PID、启动器与命令令牌,归属不匹配时拒绝终止。单端启动失败只清理该端新进程,不自动停止其他端。
升级时每次只替换一个独立包并更新精确版本,先备份 Sense/Bell,再按 Bell → Sense → Brain 验证,最后启用 connector。回退时先停用 Brain event export、Sense ingress/relay 与 Bell ingress/evidence connector,再使用各产品独立入口回退包或恢复数据库;不得删除 Outbox、Receipt、Event、运行投影、replay 或审计事实。16 路只是当前交付配额,不是编排器硬上限;真实生产包、PostgreSQL、客户 PKI、真机容量与长稳仍需部署环境验收。
<!-- coordination-deployment-v1:end -->
<!-- coordination-e2e-v1:start -->
## 三项目协调 E2E 验收入口
工单 #155 已于 2026-08-31 验收,并通过 PR #170 合入 `dev@0276bce`。该入口只用于隔离开发/验收,不是生产部署、生产迁移或现场容量测试:
```powershell
pwsh scripts/e2e/coordination/run-coordination-e2e.ps1
```
入口要求 PowerShell 7、冻结的 Go 1.26.5 工具链、Brain 可用 Python 环境、PostgreSQL 17 命令行工具,以及 Sense/Bell 各自独立 E2E 已记录的本机依赖。PostgreSQL 工具默认从 `D:\pgsql17\bin` 读取,可通过 `-PostgresBin` 指定其他安装目录;Python 默认优先使用 `Brain\.venv\Scripts\python.exe`,也可通过 `-Python` 指定。不得为通过测试而连接生产数据库、客户设备或生产服务。
每次运行会在系统临时目录创建专属 PostgreSQL cluster,使用非 5432 动态 loopback 端口,并为 Sense、Bell 创建随机且不同的 database owner 和 database。测试凭据只存在于当前进程和临时测试范围,不进入仓库。入口按顺序执行冻结契约、三端 connector/持久化故障验证,并默认继续执行 Sense、Brain、Bell 各自已有的独立 E2E。
`-SkipIndependentProductE2E` 只用于定位协调 harness 自身故障;使用该参数的结果不能作为 #155 或 MVP #156 验收证据。`-KeepTemporary` 只用于受控保留失败诊断,目录可能包含一次性测试数据,排查完成后按明确绝对路径清理,不得提交或共享。
入口在 `finally` 中只停止自己启动的临时 PostgreSQL 和下游测试拥有的进程,核对端口关闭,并扫描临时日志是否出现本轮随机秘密。成功的最终判据是全部阶段退出码为 0,输出末行包含 `COORDINATION_E2E passed`,且 `git status --short` 没有产品源码或配置改动。
该验收证明版本化契约、配置/状态、匿名事件、Sense Outbox、Bell Receipt/Event/Alert/ack/close、离线恢复、身份/重放/冲突/证据降级及三端独立运行;不证明真实 GPU/生产模型、真实摄像头、通知供应商、生产迁移、16 路长稳或客户现场效果。
<!-- coordination-e2e-v1:end -->
@@ -0,0 +1,4 @@
@echo off
setlocal
pwsh.exe -NoProfile -File "%~dp0run-coordination-e2e.ps1" %*
exit /b %ERRORLEVEL%
@@ -0,0 +1,191 @@
[CmdletBinding()]
param(
[string]$PostgresBin = 'D:\pgsql17\bin',
[string]$Python = '',
[switch]$SkipIndependentProductE2E,
[switch]$KeepTemporary
)
Set-StrictMode -Version 3.0
$ErrorActionPreference = 'Stop'
$repositoryRoot = [IO.Path]::GetFullPath((Join-Path $PSScriptRoot '..\..\..'))
$temporaryRoot = [IO.Path]::GetFullPath((Join-Path ([IO.Path]::GetTempPath()) ('yovision-coordination-e2e-' + [guid]::NewGuid().ToString('N'))))
$postgresData = Join-Path $temporaryRoot 'postgres'
$postgresLog = Join-Path $temporaryRoot 'postgres.log'
$postgresStarted = $false
$savedEnvironment = @{}
$sensitiveValues = [Collections.Generic.List[string]]::new()
function Get-FreeTcpPort {
$listener = [Net.Sockets.TcpListener]::new([Net.IPAddress]::Loopback, 0)
try { $listener.Start(); return ([Net.IPEndPoint]$listener.LocalEndpoint).Port } finally { $listener.Stop() }
}
function Wait-Tcp([int]$Port, [bool]$Open, [int]$Attempts = 120) {
for ($attempt = 0; $attempt -lt $Attempts; $attempt++) {
$client = [Net.Sockets.TcpClient]::new()
try { $connected = $client.ConnectAsync('127.0.0.1', $Port).Wait(250) -and $client.Connected } catch { $connected = $false } finally { $client.Dispose() }
if ($connected -eq $Open) { return }
Start-Sleep -Milliseconds 250
}
throw "TCP port $Port did not reach open=$Open"
}
function New-RandomName([string]$Prefix) {
return $Prefix + '_' + [guid]::NewGuid().ToString('N').Substring(0, 12)
}
function New-RandomSecret {
$buffer = New-Object byte[] 48
$generator = [Security.Cryptography.RandomNumberGenerator]::Create()
try { $generator.GetBytes($buffer) } finally { $generator.Dispose() }
return [Convert]::ToBase64String($buffer).Replace('+', 'A').Replace('/', 'B')
}
function Set-TestEnvironment([string]$Name, [string]$Value, [bool]$Sensitive = $false) {
if (-not $script:savedEnvironment.ContainsKey($Name)) {
$script:savedEnvironment[$Name] = [Environment]::GetEnvironmentVariable($Name, 'Process')
}
[Environment]::SetEnvironmentVariable($Name, $Value, 'Process')
if ($Sensitive) { $script:sensitiveValues.Add($Value) }
}
function Invoke-Checked {
param([string]$Name, [string]$WorkingDirectory, [scriptblock]$Command)
Write-Host "[coordination-e2e] $Name"
Push-Location $WorkingDirectory
try {
& $Command
if ($LASTEXITCODE -ne 0) { throw "$Name failed with exit code $LASTEXITCODE" }
} finally { Pop-Location }
}
function Assert-NoSecretInLogs {
$logs = @(Get-ChildItem -LiteralPath $temporaryRoot -File -Recurse -ErrorAction SilentlyContinue)
foreach ($log in $logs) {
$stream = [IO.File]::Open($log.FullName, [IO.FileMode]::Open, [IO.FileAccess]::Read, [IO.FileShare]::ReadWrite -bor [IO.FileShare]::Delete)
try {
$reader = [IO.StreamReader]::new($stream, [Text.Encoding]::UTF8, $true)
try { $content = $reader.ReadToEnd() } finally { $reader.Dispose() }
} finally { $stream.Dispose() }
foreach ($secret in $sensitiveValues) {
if ($secret.Length -ge 8 -and $content.Contains($secret)) {
throw "Temporary log exposed a generated E2E secret: $($log.Name)"
}
}
}
}
New-Item -ItemType Directory -Path $temporaryRoot | Out-Null
try {
foreach ($tool in @('initdb.exe', 'pg_ctl.exe', 'createdb.exe', 'psql.exe')) {
$path = Join-Path $PostgresBin $tool
if (-not (Test-Path -LiteralPath $path -PathType Leaf)) { throw "Required PostgreSQL tool not found: $path" }
}
if ([string]::IsNullOrWhiteSpace($Python)) {
$candidate = Join-Path $repositoryRoot 'Brain\.venv\Scripts\python.exe'
$Python = if (Test-Path -LiteralPath $candidate -PathType Leaf) { $candidate } else { 'python.exe' }
}
$contractEnvironment = Join-Path $temporaryRoot 'contract-venv'
& $Python -m venv $contractEnvironment
if ($LASTEXITCODE -ne 0) { throw 'Could not create the isolated contract-test environment.' }
$contractPython = Join-Path $contractEnvironment 'Scripts\python.exe'
$env:PIP_DISABLE_PIP_VERSION_CHECK = '1'
& $contractPython -m pip install --quiet -r (Join-Path $repositoryRoot 'contracts\tests\source-config-v1\requirements.txt') 'cryptography==50.0.1'
if ($LASTEXITCODE -ne 0) { throw 'Could not install the pinned contract-test dependencies.' }
$postgresPort = Get-FreeTcpPort
if ($postgresPort -eq 5432) { throw 'Coordination E2E refuses the default PostgreSQL port.' }
$clusterUser = New-RandomName 'yvcoord'
$senseRole = New-RandomName 'sense_owner'
$bellRole = New-RandomName 'bell_owner'
$senseDatabase = New-RandomName 'sense_e2e'
$bellDatabase = New-RandomName 'bell_e2e'
& (Join-Path $PostgresBin 'initdb.exe') -D $postgresData -U $clusterUser -A trust --encoding=UTF8 --no-locale | Out-Null
if ($LASTEXITCODE -ne 0) { throw 'Isolated PostgreSQL initdb failed.' }
$startArguments = "-D `"$postgresData`" -l `"$postgresLog`" -o `"-p $postgresPort -h 127.0.0.1`" start"
Start-Process -FilePath (Join-Path $PostgresBin 'pg_ctl.exe') -ArgumentList $startArguments -RedirectStandardOutput (Join-Path $temporaryRoot 'pg-ctl.out.log') -RedirectStandardError (Join-Path $temporaryRoot 'pg-ctl.err.log') -WindowStyle Hidden | Out-Null
Wait-Tcp -Port $postgresPort -Open $true
$postgresStarted = $true
$psql = Join-Path $PostgresBin 'psql.exe'
foreach ($role in @($senseRole, $bellRole)) {
& $psql -X -h 127.0.0.1 -p $postgresPort -U $clusterUser -d postgres -v ON_ERROR_STOP=1 -c "CREATE ROLE $role LOGIN;" | Out-Null
if ($LASTEXITCODE -ne 0) { throw "Could not create isolated role $role" }
}
& (Join-Path $PostgresBin 'createdb.exe') -h 127.0.0.1 -p $postgresPort -U $clusterUser -O $senseRole $senseDatabase
if ($LASTEXITCODE -ne 0) { throw 'Could not create isolated Sense database.' }
& (Join-Path $PostgresBin 'createdb.exe') -h 127.0.0.1 -p $postgresPort -U $clusterUser -O $bellRole $bellDatabase
if ($LASTEXITCODE -ne 0) { throw 'Could not create isolated Bell database.' }
$senseDsn = "host=127.0.0.1 port=$postgresPort user=$senseRole dbname=$senseDatabase sslmode=disable"
$bellDsn = "host=127.0.0.1 port=$postgresPort user=$bellRole dbname=$bellDatabase sslmode=disable"
if ($senseDsn -eq $bellDsn -or $senseRole -eq $bellRole -or $senseDatabase -eq $bellDatabase) { throw 'Sense and Bell isolation invariant failed.' }
Set-TestEnvironment 'GOTOOLCHAIN' 'go1.26.5'
Set-TestEnvironment 'PYTHONDONTWRITEBYTECODE' '1'
Set-TestEnvironment 'SENSE_OUTBOX_TEST_DATABASE_URL' $senseDsn
Set-TestEnvironment 'BELL_DATABASE_URL' $bellDsn
Set-TestEnvironment 'BELL_EVENT_INGRESS_TEST_DATABASE_URL' $bellDsn
Set-TestEnvironment 'BELL_RULE_ALERT_TEST_DATABASE_URL' $bellDsn
Set-TestEnvironment 'BELL_ALERT_LIFECYCLE_TEST_DATABASE_URL' $bellDsn
Set-TestEnvironment 'BELL_JWT_SECRET' (New-RandomSecret) $true
Set-TestEnvironment 'BELL_BOOTSTRAP_USERNAME' (New-RandomName 'coord_admin')
Set-TestEnvironment 'BELL_BOOTSTRAP_PASSWORD' (New-RandomSecret) $true
Set-TestEnvironment 'BELL_RULE_ALERT_OPERATOR_PASSWORD' (New-RandomSecret) $true
Set-TestEnvironment 'BELL_HOST' '127.0.0.1'
Set-TestEnvironment 'BELL_PORT' (Get-FreeTcpPort).ToString()
Invoke-Checked 'Bell formal migrations' (Join-Path $repositoryRoot 'Bell\server') { go run . migrate -c config/settings.demo.yml *> (Join-Path $temporaryRoot 'bell-migrate.log') }
Invoke-Checked 'source-config v1 contract' $repositoryRoot { & $contractPython -m unittest discover -s contracts/tests/source-config-v1 -p 'test_*.py' -v }
Invoke-Checked 'runtime-status v1 contract' $repositoryRoot { & $contractPython contracts/tests/runtime-status-v1/test_contract.py }
Invoke-Checked 'machine-identity v1 cross-language contract' $repositoryRoot { & $contractPython contracts/tests/machine-identity-v1/test_contract.py }
Invoke-Checked 'events v1 contract' $repositoryRoot { & $contractPython contracts/tests/events-v1/test_contract.py }
Invoke-Checked 'evidence v1 contract' $repositoryRoot { & $contractPython contracts/tests/evidence-v1/test_contract.py }
Invoke-Checked 'Sense source/status integration' (Join-Path $repositoryRoot 'Sense\tests\integration\brain_control') { go test . -count=1 -v }
Invoke-Checked 'Sense Brain-event/evidence/Outbox integration' (Join-Path $repositoryRoot 'Sense\tests\integration\bell_connector') { go test . -count=1 -v }
Invoke-Checked 'Sense PostgreSQL Outbox recovery' (Join-Path $repositoryRoot 'Sense\server') { go test ./app/sense/outbox -count=1 -v }
Invoke-Checked 'Brain source/status connector and anonymous event export' $repositoryRoot {
& $Python -m pytest Brain/tests/integration/sense_control Brain/tests/integration/event_export -q
}
Invoke-Checked 'Bell ingress and evidence degradation' (Join-Path $repositoryRoot 'Bell\server') { go test ./tests/integration/event_ingress -count=1 -v }
Invoke-Checked 'Bell rule and alert projection' (Join-Path $repositoryRoot 'Bell\server') { go test ./tests/bell_rule_alert -count=1 -v }
Invoke-Checked 'Bell alert lifecycle' (Join-Path $repositoryRoot 'Bell\server') { go test ./tests/bell_alert_lifecycle -count=1 -v }
$identityFacts = (& $psql -X -h 127.0.0.1 -p $postgresPort -U $clusterUser -d postgres -tAc "select datname||':'||pg_get_userbyid(datdba) from pg_database where datname in ('$senseDatabase','$bellDatabase') order by datname;")
if (@($identityFacts).Count -ne 2 -or ($identityFacts -join '|') -notmatch [regex]::Escape($senseRole) -or ($identityFacts -join '|') -notmatch [regex]::Escape($bellRole)) {
throw 'PostgreSQL ownership isolation evidence is incomplete.'
}
if (-not $SkipIndependentProductE2E) {
Invoke-Checked 'Sense independent isolated E2E regression' $repositoryRoot { & (Join-Path $repositoryRoot 'Sense\tests\e2e\run-isolated-e2e.ps1') -PostgresBin $PostgresBin }
Invoke-Checked 'Brain independent test regression' $repositoryRoot { & $Python -m pytest Brain/tests -q }
Invoke-Checked 'Bell independent isolated E2E regression' $repositoryRoot { & (Join-Path $repositoryRoot 'Bell\tests\e2e\run-isolated-e2e.ps1') -PostgresBin $PostgresBin }
}
Assert-NoSecretInLogs
Write-Host "COORDINATION_E2E passed: versioned contracts, source/status, anonymous event, durable Outbox recovery, Bell Receipt/Event/Alert lifecycle, identity/replay/conflict/evidence faults, separate Sense/Bell databases. postgres_port=$postgresPort"
} finally {
if ($postgresStarted) {
Start-Process -FilePath (Join-Path $PostgresBin 'pg_ctl.exe') -ArgumentList "-D `"$postgresData`" -m fast stop" -RedirectStandardOutput (Join-Path $temporaryRoot 'pg-stop.out.log') -RedirectStandardError (Join-Path $temporaryRoot 'pg-stop.err.log') -WindowStyle Hidden -Wait | Out-Null
try { Wait-Tcp -Port $postgresPort -Open $false -Attempts 40 } catch {}
}
foreach ($entry in $savedEnvironment.GetEnumerator()) {
[Environment]::SetEnvironmentVariable($entry.Key, $entry.Value, 'Process')
}
if ($KeepTemporary) {
Write-Host "Kept coordination E2E directory: $temporaryRoot"
} elseif (Test-Path -LiteralPath $temporaryRoot) {
$resolved = [IO.Path]::GetFullPath($temporaryRoot)
$tempPrefix = [IO.Path]::GetFullPath([IO.Path]::GetTempPath())
if (-not $resolved.StartsWith($tempPrefix, [StringComparison]::OrdinalIgnoreCase) -or -not ([IO.Path]::GetFileName($resolved)).StartsWith('yovision-coordination-e2e-')) {
throw "Refusing unsafe temporary cleanup: $resolved"
}
Remove-Item -LiteralPath $resolved -Recurse -Force
}
}
@@ -0,0 +1,414 @@
Set-StrictMode -Version 3.0
$ErrorActionPreference = 'Stop'
$script:CoordinationProductNames = @('sense', 'brain', 'bell')
$script:CoordinationStartOrder = @('bell', 'sense', 'brain')
$script:CoordinationStopOrder = @('brain', 'sense', 'bell')
function Get-CoordinationRepositoryRoot {
return [IO.Path]::GetFullPath((Join-Path $PSScriptRoot '..\..\..'))
}
function Resolve-CoordinationPath {
param([Parameter(Mandatory = $true)][string]$Base, [Parameter(Mandatory = $true)][string]$Value)
if ([string]::IsNullOrWhiteSpace($Value)) { throw 'A required path is empty.' }
if ([IO.Path]::IsPathRooted($Value)) { return [IO.Path]::GetFullPath($Value) }
return [IO.Path]::GetFullPath((Join-Path $Base $Value))
}
function Test-CoordinationPathWithin {
param([Parameter(Mandatory = $true)][string]$Child, [Parameter(Mandatory = $true)][string]$Parent)
$childPath = [IO.Path]::GetFullPath($Child).TrimEnd('\', '/')
$parentPath = [IO.Path]::GetFullPath($Parent).TrimEnd('\', '/')
return $childPath.Equals($parentPath, [StringComparison]::OrdinalIgnoreCase) -or
$childPath.StartsWith($parentPath + [IO.Path]::DirectorySeparatorChar, [StringComparison]::OrdinalIgnoreCase)
}
function Get-CoordinationProperty {
param([Parameter(Mandatory = $true)]$Object, [Parameter(Mandatory = $true)][string]$Name, [switch]$Optional)
$property = $Object.PSObject.Properties[$Name]
if (-not $property) {
if ($Optional) { return $null }
throw "Missing manifest property: $Name"
}
return $property.Value
}
function Assert-CoordinationProperties {
param([Parameter(Mandatory = $true)]$Object, [Parameter(Mandatory = $true)][string[]]$Allowed, [Parameter(Mandatory = $true)][string]$Context)
foreach ($property in $Object.PSObject.Properties.Name) {
if ($property -notin $Allowed) { throw "$Context contains unsupported property: $property" }
}
}
function Read-CoordinationEnvironment {
param([Parameter(Mandatory = $true)][string]$Path)
if (-not (Test-Path -LiteralPath $Path -PathType Leaf)) { throw "Environment file not found: $Path" }
$values = @{}
$lineNumber = 0
foreach ($rawLine in Get-Content -LiteralPath $Path -Encoding UTF8) {
$lineNumber++
$line = $rawLine.Trim()
if ($line.Length -eq 0 -or $line.StartsWith('#')) { continue }
$separator = $line.IndexOf('=')
if ($separator -lt 1) { throw "Invalid environment file at line $lineNumber. Expected NAME=value." }
$name = $line.Substring(0, $separator).Trim()
if ($name -notmatch '^[A-Z][A-Z0-9_]{1,127}$') { throw "Invalid environment variable name at line $lineNumber." }
if ($values.ContainsKey($name)) { throw "Duplicate environment variable at line ${lineNumber}: $name" }
$value = $line.Substring($separator + 1)
if ($value.Length -ge 2) {
$first, $last = $value[0], $value[$value.Length - 1]
if (($first -eq '"' -and $last -eq '"') -or ($first -eq "'" -and $last -eq "'")) { $value = $value.Substring(1, $value.Length - 2) }
}
$values[$name] = $value
}
return $values
}
function Assert-CoordinationExternalSecretPath {
param([Parameter(Mandatory = $true)][string]$Path, [Parameter(Mandatory = $true)][string]$RepositoryRoot, [Parameter(Mandatory = $true)][string[]]$PackageRoots)
if (Test-CoordinationPathWithin -Child $Path -Parent $RepositoryRoot) { throw 'Secret or environment files must be outside the repository.' }
foreach ($packageRoot in $PackageRoots) {
if (Test-CoordinationPathWithin -Child $Path -Parent $packageRoot) { throw 'Secret or environment files must be outside product packages.' }
}
}
function Assert-CoordinationDistinctPaths {
param([Parameter(Mandatory = $true)][object[]]$Entries)
for ($left = 0; $left -lt $Entries.Count; $left++) {
for ($right = $left + 1; $right -lt $Entries.Count; $right++) {
if ((Test-CoordinationPathWithin -Child $Entries[$left].Path -Parent $Entries[$right].Path) -or
(Test-CoordinationPathWithin -Child $Entries[$right].Path -Parent $Entries[$left].Path)) {
throw "Deployment paths overlap: $($Entries[$left].Label) and $($Entries[$right].Label)."
}
}
}
}
function Resolve-CoordinationCommand {
param([Parameter(Mandatory = $true)][string]$PackageRoot, [Parameter(Mandatory = $true)]$Command)
Assert-CoordinationProperties -Object $Command -Allowed @('executable', 'arguments') -Context 'command'
$path = Resolve-CoordinationPath -Base $PackageRoot -Value ([string](Get-CoordinationProperty -Object $Command -Name 'executable'))
if (-not (Test-CoordinationPathWithin -Child $path -Parent $PackageRoot)) { throw 'Product commands must be inside their package root.' }
if (-not (Test-Path -LiteralPath $path -PathType Leaf)) { throw "Product command not found: $path" }
$rawArguments = Get-CoordinationProperty -Object $Command -Name 'arguments'
if ($rawArguments -is [string]) { throw 'Command arguments must be an array.' }
$arguments = @($rawArguments) | ForEach-Object { [string]$_ }
return [pscustomobject]@{ Path = $path; Arguments = @($arguments) }
}
function Get-CoordinationLauncher {
param([Parameter(Mandatory = $true)]$Command)
$extension = [IO.Path]::GetExtension($Command.Path).ToLowerInvariant()
if ($extension -eq '.ps1') {
$pwsh = (Get-Command pwsh.exe -ErrorAction Stop).Source
return [pscustomobject]@{ Executable = $pwsh; Arguments = @('-NoProfile', '-File', $Command.Path) + @($Command.Arguments); CommandToken = $Command.Path }
}
if ($extension -in @('.bat', '.cmd')) {
$cmd = (Get-Command cmd.exe -ErrorAction Stop).Source
return [pscustomobject]@{ Executable = $cmd; Arguments = @('/d', '/c', $Command.Path) + @($Command.Arguments); CommandToken = $Command.Path }
}
return [pscustomobject]@{ Executable = $Command.Path; Arguments = @($Command.Arguments); CommandToken = $Command.Path }
}
function ConvertTo-CoordinationArgument {
param([AllowEmptyString()][string]$Value)
if ($Value -notmatch '[\s"]') { return $Value }
return '"' + ($Value -replace '(\\*)"', '$1$1\"' -replace '(\\+)$', '$1$1') + '"'
}
function Invoke-CoordinationEnvironment {
param([Parameter(Mandatory = $true)][hashtable]$Values, [Parameter(Mandatory = $true)][scriptblock]$Action)
$saved = @{}
try {
foreach ($name in $Values.Keys) {
$saved[$name] = [Environment]::GetEnvironmentVariable($name, 'Process')
[Environment]::SetEnvironmentVariable($name, [string]$Values[$name], 'Process')
}
return & $Action
} finally {
foreach ($name in $Values.Keys) { [Environment]::SetEnvironmentVariable($name, $saved[$name], 'Process') }
}
}
function Get-CoordinationFileDigest {
param([Parameter(Mandatory = $true)][string]$Path)
return (Get-FileHash -LiteralPath $Path -Algorithm SHA256).Hash.ToLowerInvariant()
}
function Get-CoordinationDatabaseIdentity {
param([Parameter(Mandatory = $true)][string]$Connection)
if ($Connection -match '^postgres(?:ql)?://') {
$uri = [Uri]$Connection
$role = if ($uri.UserInfo) { [Uri]::UnescapeDataString(($uri.UserInfo -split ':', 2)[0]) } else { '' }
return [pscustomobject]@{ Database = [Uri]::UnescapeDataString($uri.AbsolutePath.Trim('/')); Role = $role }
}
$database = if ($Connection -match '(?i)(?:^|\s)(?:dbname|database)\s*=\s*(?:''([^'']+)''|"([^"]+)"|([^\s]+))') { @($Matches[1], $Matches[2], $Matches[3]) | Where-Object { $_ } | Select-Object -First 1 } else { '' }
$role = if ($Connection -match '(?i)(?:^|\s)(?:user|username)\s*=\s*(?:''([^'']+)''|"([^"]+)"|([^\s]+))') { @($Matches[1], $Matches[2], $Matches[3]) | Where-Object { $_ } | Select-Object -First 1 } else { '' }
return [pscustomobject]@{ Database = [string]$database; Role = [string]$role }
}
function Read-CoordinationManifest {
param([Parameter(Mandatory = $true)][string]$Manifest)
$manifestPath = [IO.Path]::GetFullPath($Manifest)
if (-not (Test-Path -LiteralPath $manifestPath -PathType Leaf)) { throw "Coordination manifest not found: $manifestPath" }
try { $raw = Get-Content -LiteralPath $manifestPath -Raw -Encoding UTF8 | ConvertFrom-Json -Depth 64 } catch { throw "Coordination manifest is not valid JSON: $($_.Exception.Message)" }
Assert-CoordinationProperties -Object $raw -Allowed @('schema_version', 'deployment_id', 'runtime_root', 'products') -Context 'manifest'
if ((Get-CoordinationProperty -Object $raw -Name 'schema_version') -ne 'yovision.coordination/v1') { throw 'Unsupported coordination manifest schema version.' }
$deploymentID = [string](Get-CoordinationProperty -Object $raw -Name 'deployment_id')
if ($deploymentID -notmatch '^[A-Za-z0-9][A-Za-z0-9._-]{2,63}$') { throw 'Invalid deployment_id.' }
$manifestRoot = Split-Path -Parent $manifestPath
$runtimeRoot = Resolve-CoordinationPath -Base $manifestRoot -Value ([string](Get-CoordinationProperty -Object $raw -Name 'runtime_root'))
$rawProducts = Get-CoordinationProperty -Object $raw -Name 'products'
Assert-CoordinationProperties -Object $rawProducts -Allowed $script:CoordinationProductNames -Context 'products'
$products = @()
foreach ($name in $script:CoordinationProductNames) {
$item = Get-CoordinationProperty -Object $rawProducts -Name $name
Assert-CoordinationProperties -Object $item -Allowed @('enabled', 'version', 'package_root', 'environment_file', 'data_directory', 'log_directory', 'ports', 'browser_origin', 'cookie_name', 'account_namespace', 'database_id', 'database_role', 'start', 'stop', 'health', 'identities') -Context $name
$rawEnabled = Get-CoordinationProperty -Object $item -Name 'enabled'
if ($rawEnabled -isnot [bool]) { throw "enabled must be a boolean for ${name}." }
$version = [string](Get-CoordinationProperty -Object $item -Name 'version')
if ([string]::IsNullOrWhiteSpace($version)) { throw "version is required for ${name}." }
$packageRoot = Resolve-CoordinationPath -Base $manifestRoot -Value ([string](Get-CoordinationProperty -Object $item -Name 'package_root'))
if (-not (Test-Path -LiteralPath $packageRoot -PathType Container)) { throw "Package root not found for ${name}: $packageRoot" }
$environmentFile = Resolve-CoordinationPath -Base $manifestRoot -Value ([string](Get-CoordinationProperty -Object $item -Name 'environment_file'))
$dataDirectory = Resolve-CoordinationPath -Base $manifestRoot -Value ([string](Get-CoordinationProperty -Object $item -Name 'data_directory'))
$logDirectory = Resolve-CoordinationPath -Base $manifestRoot -Value ([string](Get-CoordinationProperty -Object $item -Name 'log_directory'))
$start = Resolve-CoordinationCommand -PackageRoot $packageRoot -Command (Get-CoordinationProperty -Object $item -Name 'start')
$rawStop = Get-CoordinationProperty -Object $item -Name 'stop' -Optional
$stop = if ($null -eq $rawStop) { $null } else { Resolve-CoordinationCommand -PackageRoot $packageRoot -Command $rawStop }
$ports = @((Get-CoordinationProperty -Object $item -Name 'ports')) | ForEach-Object { [int]$_ }
foreach ($port in $ports) { if ($port -lt 1 -or $port -gt 65535) { throw "Invalid port for ${name}." } }
$health = Get-CoordinationProperty -Object $item -Name 'health'
Assert-CoordinationProperties -Object $health -Allowed @('kind', 'url', 'timeout_seconds') -Context "$name health"
$healthKind = [string](Get-CoordinationProperty -Object $health -Name 'kind')
$healthURL = [string](Get-CoordinationProperty -Object $health -Name 'url' -Optional)
$healthTimeout = [int](Get-CoordinationProperty -Object $health -Name 'timeout_seconds')
if ($healthKind -notin @('process', 'http') -or $healthTimeout -lt 1 -or $healthTimeout -gt 300) { throw "Invalid health policy for ${name}." }
if ($healthKind -eq 'http') {
$parsedHealth = $null
if (-not [Uri]::TryCreate($healthURL, [UriKind]::Absolute, [ref]$parsedHealth) -or $parsedHealth.Scheme -notin @('http', 'https')) { throw "Invalid health URL for ${name}." }
if ($ports -notcontains $parsedHealth.Port) { throw "Health URL port is not declared for ${name}." }
}
$browserOrigin = [string](Get-CoordinationProperty -Object $item -Name 'browser_origin')
if (-not [string]::IsNullOrWhiteSpace($browserOrigin)) {
$parsedOrigin = $null
if (-not [Uri]::TryCreate($browserOrigin, [UriKind]::Absolute, [ref]$parsedOrigin) -or $parsedOrigin.Scheme -notin @('http', 'https') -or $parsedOrigin.AbsolutePath -ne '/' -or $parsedOrigin.Query -or $parsedOrigin.Fragment) { throw "Invalid browser origin for ${name}." }
if ($ports -notcontains $parsedOrigin.Port) { throw "Browser origin port is not declared for ${name}." }
}
$identities = @()
foreach ($identity in @((Get-CoordinationProperty -Object $item -Name 'identities'))) {
Assert-CoordinationProperties -Object $identity -Allowed @('principal', 'key_id', 'private_key_path') -Context "$name identity"
$principal = [string](Get-CoordinationProperty -Object $identity -Name 'principal')
$keyID = [string](Get-CoordinationProperty -Object $identity -Name 'key_id')
if ($principal -notmatch "^yv:${name}:[A-Za-z0-9][A-Za-z0-9._-]{0,63}$" -or $keyID -notmatch '^[A-Za-z0-9][A-Za-z0-9._-]{0,63}$') { throw "Invalid machine identity metadata for ${name}." }
$keyPath = Resolve-CoordinationPath -Base $manifestRoot -Value ([string](Get-CoordinationProperty -Object $identity -Name 'private_key_path'))
if (-not (Test-Path -LiteralPath $keyPath -PathType Leaf)) { throw "Machine identity key not found for ${name}." }
$identities += [pscustomobject]@{ Principal = $principal; KeyID = $keyID; PrivateKeyPath = $keyPath }
}
$products += [pscustomobject]@{
Name = $name; Enabled = [bool]$rawEnabled; Version = $version
PackageRoot = $packageRoot; EnvironmentFile = $environmentFile; Environment = Read-CoordinationEnvironment -Path $environmentFile
DataDirectory = $dataDirectory; LogDirectory = $logDirectory; Ports = @($ports)
BrowserOrigin = $browserOrigin; CookieName = [string](Get-CoordinationProperty -Object $item -Name 'cookie_name')
AccountNamespace = [string](Get-CoordinationProperty -Object $item -Name 'account_namespace'); DatabaseID = [string](Get-CoordinationProperty -Object $item -Name 'database_id'); DatabaseRole = [string](Get-CoordinationProperty -Object $item -Name 'database_role')
Start = $start; Stop = $stop; HealthKind = $healthKind; HealthURL = $healthURL; HealthTimeoutSeconds = $healthTimeout; Identities = @($identities)
}
}
$result = [pscustomobject]@{ Path = $manifestPath; Digest = Get-CoordinationFileDigest -Path $manifestPath; DeploymentID = $deploymentID; RuntimeRoot = $runtimeRoot; Products = @($products); RepositoryRoot = Get-CoordinationRepositoryRoot }
Assert-CoordinationIsolation -Configuration $result
return $result
}
function Assert-CoordinationIsolation {
param([Parameter(Mandatory = $true)]$Configuration)
$packages = @($Configuration.Products | ForEach-Object { $_.PackageRoot })
$paths = @([pscustomobject]@{ Label = 'coordination runtime'; Path = $Configuration.RuntimeRoot })
foreach ($product in $Configuration.Products) {
$paths += [pscustomobject]@{ Label = "$($product.Name) package"; Path = $product.PackageRoot }
$paths += [pscustomobject]@{ Label = "$($product.Name) data"; Path = $product.DataDirectory }
$paths += [pscustomobject]@{ Label = "$($product.Name) logs"; Path = $product.LogDirectory }
Assert-CoordinationExternalSecretPath -Path $product.EnvironmentFile -RepositoryRoot $Configuration.RepositoryRoot -PackageRoots $packages
foreach ($identity in $product.Identities) { Assert-CoordinationExternalSecretPath -Path $identity.PrivateKeyPath -RepositoryRoot $Configuration.RepositoryRoot -PackageRoots $packages }
}
Assert-CoordinationDistinctPaths -Entries $paths
$ports = @{}
$environmentFiles = @{}
$identityKeys = @{}
$privateKeyPaths = @{}
foreach ($product in $Configuration.Products) {
if ($environmentFiles.ContainsKey($product.EnvironmentFile.ToLowerInvariant())) { throw 'Products must not share an environment file.' }
$environmentFiles[$product.EnvironmentFile.ToLowerInvariant()] = $true
foreach ($port in $product.Ports) {
if ($ports.ContainsKey($port)) { throw "Products must not share port $port." }
$ports[$port] = $product.Name
}
foreach ($identity in $product.Identities) {
$identityID = ($identity.Principal + '/' + $identity.KeyID).ToLowerInvariant()
if ($identityKeys.ContainsKey($identityID)) { throw 'Machine principal/key pairs must be unique per product instance.' }
$identityKeys[$identityID] = $true
$privateKeyID = $identity.PrivateKeyPath.ToLowerInvariant()
if ($privateKeyPaths.ContainsKey($privateKeyID)) { throw 'Machine identities must not share a private key file.' }
$privateKeyPaths[$privateKeyID] = $true
}
}
$sense = $Configuration.Products | Where-Object Name -eq 'sense'
$bell = $Configuration.Products | Where-Object Name -eq 'bell'
if ($sense.CookieName -ne 'Sense-Admin-Token' -or $bell.CookieName -ne 'Bell-Admin-Token' -or $sense.CookieName -eq $bell.CookieName) { throw 'Sense and Bell browser Cookie names are not isolated.' }
if ([string]::IsNullOrWhiteSpace($sense.BrowserOrigin) -or [string]::IsNullOrWhiteSpace($bell.BrowserOrigin) -or $sense.BrowserOrigin -eq $bell.BrowserOrigin) { throw 'Sense and Bell browser origins must be distinct.' }
foreach ($field in @('DatabaseID', 'DatabaseRole', 'AccountNamespace')) {
if ([string]::IsNullOrWhiteSpace($sense.$field) -or [string]::IsNullOrWhiteSpace($bell.$field) -or $sense.$field -eq $bell.$field) { throw "Sense and Bell $field values must be non-empty and distinct." }
}
foreach ($required in @(@($sense, 'SENSE_DATABASE_URL', 'SENSE_JWT_SECRET'), @($bell, 'BELL_DATABASE_URL', 'BELL_JWT_SECRET'))) {
$product, $databaseKey, $jwtKey = $required
if (-not $product.Environment.ContainsKey($databaseKey) -or [string]::IsNullOrWhiteSpace([string]$product.Environment[$databaseKey])) { throw "$databaseKey is required in the external environment file." }
if (-not $product.Environment.ContainsKey($jwtKey) -or ([string]$product.Environment[$jwtKey]).Length -lt 32) { throw "$jwtKey must contain at least 32 characters in the external environment file." }
}
if ([string]$sense.Environment['SENSE_DATABASE_URL'] -eq [string]$bell.Environment['BELL_DATABASE_URL']) { throw 'Sense and Bell must not share a database URL.' }
if ([string]$sense.Environment['SENSE_JWT_SECRET'] -ceq [string]$bell.Environment['BELL_JWT_SECRET']) { throw 'Sense and Bell must not share a JWT secret.' }
$senseDatabase = Get-CoordinationDatabaseIdentity -Connection ([string]$sense.Environment['SENSE_DATABASE_URL'])
$bellDatabase = Get-CoordinationDatabaseIdentity -Connection ([string]$bell.Environment['BELL_DATABASE_URL'])
if ($senseDatabase.Database -ne $sense.DatabaseID -or $senseDatabase.Role -ne $sense.DatabaseRole) { throw 'Sense database URL does not match its declared database and role.' }
if ($bellDatabase.Database -ne $bell.DatabaseID -or $bellDatabase.Role -ne $bell.DatabaseRole) { throw 'Bell database URL does not match its declared database and role.' }
foreach ($portRule in @(@($sense, 'SENSE_PORT', 0), @($bell, 'BELL_PORT', 0), @($bell, 'BELL_WEB_PORT', 1))) {
$product, $key, $index = $portRule
if (-not $product.Environment.ContainsKey($key) -or [int]$product.Environment[$key] -ne $product.Ports[[int]$index]) { throw "$key must match the declared product port." }
}
}
function Resolve-CoordinationSelection {
param([Parameter(Mandatory = $true)]$Configuration, [string[]]$Product = @('all'), [ValidateSet('start', 'stop', 'status')][string]$Operation = 'status')
$requested = @()
foreach ($entry in @($Product)) { $requested += @($entry -split ',') | ForEach-Object { $_.Trim().ToLowerInvariant() } | Where-Object { $_ } }
if ($requested.Count -eq 0 -or $requested -contains 'all') {
$requested = if ($Operation -eq 'start') { @($Configuration.Products | Where-Object Enabled | ForEach-Object Name) } else { @($Configuration.Products | ForEach-Object Name) }
}
foreach ($name in $requested) {
if ($name -notin $script:CoordinationProductNames) { throw "Unknown product selection: $name" }
$target = $Configuration.Products | Where-Object Name -eq $name
if ($Operation -eq 'start' -and -not $target.Enabled) { throw "Product is disabled in the manifest: $name" }
}
$order = if ($Operation -eq 'stop') { $script:CoordinationStopOrder } else { $script:CoordinationStartOrder }
return @($order | Where-Object { $requested -contains $_ } | ForEach-Object { $name = $_; $Configuration.Products | Where-Object Name -eq $name })
}
function Get-CoordinationStatePath {
param([Parameter(Mandatory = $true)]$Configuration, [Parameter(Mandatory = $true)]$Product)
return Join-Path $Configuration.RuntimeRoot "state\$($Product.Name).json"
}
function Read-CoordinationState {
param([Parameter(Mandatory = $true)]$Configuration, [Parameter(Mandatory = $true)]$Product)
$path = Get-CoordinationStatePath -Configuration $Configuration -Product $Product
if (-not (Test-Path -LiteralPath $path -PathType Leaf)) { return $null }
try { return Get-Content -LiteralPath $path -Raw -Encoding UTF8 | ConvertFrom-Json } catch { throw "Invalid coordination state for $($Product.Name)." }
}
function Test-CoordinationOwnedProcess {
param([Parameter(Mandatory = $true)]$State)
$process = Get-CimInstance Win32_Process -Filter "ProcessId = $([int]$State.pid)" -ErrorAction SilentlyContinue
if (-not $process -or [string]::IsNullOrWhiteSpace([string]$process.ExecutablePath)) { return $false }
$expected = [IO.Path]::GetFullPath([string]$State.launcher_executable)
if (-not [IO.Path]::GetFullPath([string]$process.ExecutablePath).Equals($expected, [StringComparison]::OrdinalIgnoreCase)) { return $false }
return ([string]$process.CommandLine).IndexOf([string]$State.command_token, [StringComparison]::OrdinalIgnoreCase) -ge 0
}
function Test-CoordinationHealth {
param([Parameter(Mandatory = $true)]$Product, [Parameter(Mandatory = $true)]$State)
if (-not (Test-CoordinationOwnedProcess -State $State)) { return $false }
if ($Product.HealthKind -eq 'process') { return $true }
try {
$response = Invoke-WebRequest -Uri $Product.HealthURL -Method Get -TimeoutSec 3 -UseBasicParsing
return $response.StatusCode -ge 200 -and $response.StatusCode -lt 400
} catch { return $false }
}
function Wait-CoordinationHealth {
param([Parameter(Mandatory = $true)]$Product, [Parameter(Mandatory = $true)]$State)
if ($Product.HealthKind -eq 'process') {
Start-Sleep -Milliseconds 750
if (Test-CoordinationHealth -Product $Product -State $State) { return }
throw "$($Product.Name) exited during the process health grace period."
}
$deadline = [DateTime]::UtcNow.AddSeconds($Product.HealthTimeoutSeconds)
do {
if (Test-CoordinationHealth -Product $Product -State $State) { return }
if (-not (Get-Process -Id ([int]$State.pid) -ErrorAction SilentlyContinue)) { throw "$($Product.Name) exited before becoming healthy." }
Start-Sleep -Milliseconds 250
} while ([DateTime]::UtcNow -lt $deadline)
throw "$($Product.Name) did not become healthy before the timeout."
}
function Assert-CoordinationPortsAvailable {
param([Parameter(Mandatory = $true)]$Product)
foreach ($port in $Product.Ports) {
if (Get-NetTCPConnection -State Listen -LocalPort $port -ErrorAction SilentlyContinue) { throw "$($Product.Name) port $port is already in use." }
}
}
function Start-CoordinationProduct {
param([Parameter(Mandatory = $true)]$Configuration, [Parameter(Mandatory = $true)]$Product)
$existing = Read-CoordinationState -Configuration $Configuration -Product $Product
if ($existing -and (Test-CoordinationOwnedProcess -State $existing)) {
if ([string]$existing.manifest_sha256 -ne $Configuration.Digest) { throw "$($Product.Name) is running from a different manifest revision." }
if (Test-CoordinationHealth -Product $Product -State $existing) { Write-Host "$($Product.Name) is already running."; return }
throw "$($Product.Name) has an owned but unhealthy process. Stop it before restart."
}
Assert-CoordinationPortsAvailable -Product $Product
New-Item -ItemType Directory -Force -Path $Configuration.RuntimeRoot,(Join-Path $Configuration.RuntimeRoot 'state'),$Product.DataDirectory,$Product.LogDirectory | Out-Null
$launcher = Get-CoordinationLauncher -Command $Product.Start
$argumentLine = (@($launcher.Arguments) | ForEach-Object { ConvertTo-CoordinationArgument -Value ([string]$_) }) -join ' '
$stdout = Join-Path $Product.LogDirectory 'coordination.out.log'
$stderr = Join-Path $Product.LogDirectory 'coordination.err.log'
$process = Invoke-CoordinationEnvironment -Values $Product.Environment -Action {
Start-Process -FilePath $launcher.Executable -ArgumentList $argumentLine -WorkingDirectory $Product.PackageRoot -RedirectStandardOutput $stdout -RedirectStandardError $stderr -WindowStyle Hidden -PassThru
}
$state = [ordered]@{
schema_version = 'yovision.coordination-state/v1'; deployment_id = $Configuration.DeploymentID; product = $Product.Name
pid = $process.Id; started_at = [DateTime]::UtcNow.ToString('o'); version = $Product.Version; manifest_sha256 = $Configuration.Digest
launcher_executable = [IO.Path]::GetFullPath($launcher.Executable); command_token = $launcher.CommandToken; package_root = $Product.PackageRoot
}
$statePath = Get-CoordinationStatePath -Configuration $Configuration -Product $Product
[IO.File]::WriteAllText($statePath, ($state | ConvertTo-Json -Depth 8), [Text.UTF8Encoding]::new($false))
try {
Wait-CoordinationHealth -Product $Product -State ([pscustomobject]$state)
Write-Host "$($Product.Name) started (version $($Product.Version))."
} catch {
if (Test-CoordinationOwnedProcess -State ([pscustomobject]$state)) { & taskkill.exe /PID $process.Id /T /F 2>$null | Out-Null }
Remove-Item -LiteralPath $statePath -Force -ErrorAction SilentlyContinue
throw
}
}
function Stop-CoordinationProduct {
param([Parameter(Mandatory = $true)]$Configuration, [Parameter(Mandatory = $true)]$Product)
$statePath = Get-CoordinationStatePath -Configuration $Configuration -Product $Product
$state = Read-CoordinationState -Configuration $Configuration -Product $Product
if (-not $state) { Write-Host "$($Product.Name) is stopped."; return }
if (-not (Test-CoordinationOwnedProcess -State $state)) { throw "$($Product.Name) state is stale or belongs to another process; no process was stopped." }
if ($Product.Stop) {
$stopLauncher = Get-CoordinationLauncher -Command $Product.Stop
$stopArguments = (@($stopLauncher.Arguments) | ForEach-Object { ConvertTo-CoordinationArgument -Value ([string]$_) }) -join ' '
$stopEnvironment = @{}
foreach ($name in $Product.Environment.Keys) { $stopEnvironment[$name] = $Product.Environment[$name] }
$stopEnvironment['YOVISION_COORDINATION_OWNED_PID'] = [string]$state.pid
$stopProcess = Invoke-CoordinationEnvironment -Values $stopEnvironment -Action { Start-Process -FilePath $stopLauncher.Executable -ArgumentList $stopArguments -WorkingDirectory $Product.PackageRoot -WindowStyle Hidden -Wait -PassThru }
if ($stopProcess.ExitCode -ne 0) { throw "$($Product.Name) stop entrypoint failed with exit code $($stopProcess.ExitCode)." }
}
$deadline = [DateTime]::UtcNow.AddSeconds(10)
while ((Test-CoordinationOwnedProcess -State $state) -and [DateTime]::UtcNow -lt $deadline) { Start-Sleep -Milliseconds 200 }
if (Test-CoordinationOwnedProcess -State $state) { & taskkill.exe /PID ([int]$state.pid) /T /F | Out-Null }
if (Get-Process -Id ([int]$state.pid) -ErrorAction SilentlyContinue) { throw "$($Product.Name) owned process did not stop." }
Remove-Item -LiteralPath $statePath -Force
Write-Host "$($Product.Name) stopped."
}
function Get-CoordinationProductStatus {
param([Parameter(Mandatory = $true)]$Configuration, [Parameter(Mandatory = $true)]$Product)
$state = Read-CoordinationState -Configuration $Configuration -Product $Product
if (-not $state) { return [pscustomobject]@{ Product = $Product.Name; Status = 'stopped'; PID = ''; Version = $Product.Version; Health = 'not-running' } }
if (-not (Test-CoordinationOwnedProcess -State $state)) { return [pscustomobject]@{ Product = $Product.Name; Status = 'stale'; PID = $state.pid; Version = $state.version; Health = 'ownership-mismatch' } }
if ([string]$state.manifest_sha256 -ne $Configuration.Digest) { return [pscustomobject]@{ Product = $Product.Name; Status = 'stale'; PID = $state.pid; Version = $state.version; Health = 'manifest-drift' } }
$healthy = Test-CoordinationHealth -Product $Product -State $state
return [pscustomobject]@{ Product = $Product.Name; Status = $(if ($healthy) { 'running' } else { 'unhealthy' }); PID = $state.pid; Version = $state.version; Health = $(if ($healthy) { 'ok' } else { 'failed' }) }
}
@@ -0,0 +1,3 @@
@echo off
pwsh.exe -NoProfile -File "%~dp0start-yovision.ps1" %*
exit /b %ERRORLEVEL%
@@ -0,0 +1,17 @@
param(
[Parameter(Mandatory = $true)][string]$Manifest,
[string[]]$Product = @('all'),
[switch]$ValidateOnly
)
. (Join-Path $PSScriptRoot 'coordination-common.ps1')
try {
$configuration = Read-CoordinationManifest -Manifest $Manifest
$selection = Resolve-CoordinationSelection -Configuration $configuration -Product $Product -Operation start
if ($ValidateOnly) { Write-Host "Coordination manifest is valid for: $(($selection.Name) -join ', ')."; exit 0 }
foreach ($item in $selection) { Start-CoordinationProduct -Configuration $configuration -Product $item }
exit 0
} catch {
Write-Error $_.Exception.Message
exit 1
}
@@ -0,0 +1,3 @@
@echo off
pwsh.exe -NoProfile -File "%~dp0status-yovision.ps1" %*
exit /b %ERRORLEVEL%
@@ -0,0 +1,18 @@
param(
[Parameter(Mandatory = $true)][string]$Manifest,
[string[]]$Product = @('all'),
[switch]$Json
)
. (Join-Path $PSScriptRoot 'coordination-common.ps1')
try {
$configuration = Read-CoordinationManifest -Manifest $Manifest
$selection = Resolve-CoordinationSelection -Configuration $configuration -Product $Product -Operation status
$result = @($selection | ForEach-Object { Get-CoordinationProductStatus -Configuration $configuration -Product $_ })
if ($Json) { $result | ConvertTo-Json -Depth 4 } else { $result | Format-Table -AutoSize }
if (@($result | Where-Object Status -ne 'running').Count -gt 0) { exit 3 }
exit 0
} catch {
Write-Error $_.Exception.Message
exit 1
}
@@ -0,0 +1,3 @@
@echo off
pwsh.exe -NoProfile -File "%~dp0stop-yovision.ps1" %*
exit /b %ERRORLEVEL%
@@ -0,0 +1,15 @@
param(
[Parameter(Mandatory = $true)][string]$Manifest,
[string[]]$Product = @('all')
)
. (Join-Path $PSScriptRoot 'coordination-common.ps1')
try {
$configuration = Read-CoordinationManifest -Manifest $Manifest
$selection = Resolve-CoordinationSelection -Configuration $configuration -Product $Product -Operation stop
foreach ($item in $selection) { Stop-CoordinationProduct -Configuration $configuration -Product $item }
exit 0
} catch {
Write-Error $_.Exception.Message
exit 1
}
+18
View File
@@ -0,0 +1,18 @@
# Coordination E2E
Run the complete isolated coordination acceptance from the repository root:
```powershell
pwsh scripts/e2e/coordination/run-coordination-e2e.ps1
```
The runner creates a temporary PostgreSQL cluster on a dynamic loopback port,
uses distinct random owners and databases for Sense and Bell, exercises the
frozen contracts and the three connector chains, runs each product's existing
independent regression, checks generated secrets are absent from temporary
logs, and removes only the processes and directory it created.
`-SkipIndependentProductE2E` is intended only for local harness debugging and
does not satisfy issue #155 acceptance. `-KeepTemporary` preserves disposable
diagnostics after a failed run; the directory contains test-only generated
credentials and must not be committed or shared.