Files
Agent-System/internal/server/vault.go
T
weichenandCursor ce7d1b8ac5 fix(server): 后台 ticker 纳入 Shutdown;vault Drain 告警;bizdev 心跳超时(R-6)
reminder/vault 可停止;Drain 失败 slog.Warn;心跳/释放 lease 使用 10s 超时。

Co-authored-by: Cursor <cursoragent@cursor.com>
2026-09-18 03:44:08 +00:00

118 lines
3.1 KiB
Go

package server
import (
"context"
"fmt"
"log/slog"
"net/http"
"sync/atomic"
"time"
"chat-agent-system/internal/artifactvault"
"chat-agent-system/internal/config"
)
// Vault HTTP + in-process publisher stubs for U7 incremental.
// Default: FAV_PUBLISHER_ENABLED=false, FAV_DEGRADED_READ=false.
// Full ingest hooks / enumerate backfill remain on feature/final-artifact-vault.
func (s *Server) initArtifactVault() {
if s == nil || s.cfg == nil {
return
}
// Publisher store stays nil unless explicitly enabled AND vault roots are set.
if !s.cfg.FAVPublisherEnabled {
return
}
if s.cfg.VaultStateRoot == "" || s.cfg.VaultMasterKey == "" {
return
}
if err := config.ValidateIndependentVault(s.cfg.VaultStateRoot, s.cfg.DataDir); err != nil {
return
}
ident := s.cfg.VaultStorageIdentity
if ident == "" {
env := s.cfg.Env
if env == "" {
env = "dev"
}
ident = "vault-" + env
}
store, err := artifactvault.NewFilesystemStore(s.cfg.VaultStateRoot, ident, []byte(s.cfg.VaultMasterKey))
if err != nil {
return
}
var reg artifactvault.Registry
if s.pool != nil {
reg = artifactvault.NewPGRegistry(s.pool)
} else {
reg = artifactvault.NewMemoryRegistry()
}
s.vaultStore = store
s.vaultReg = reg
s.vaultPub = &artifactvault.Publisher{Reg: reg, Store: store}
}
func (s *Server) startVaultPublisher() {
if s == nil || s.vaultPub == nil || s.cfg == nil || !s.cfg.FAVPublisherEnabled {
return
}
s.vaultStop = make(chan struct{})
go func() {
t := time.NewTicker(2 * time.Second)
defer t.Stop()
for {
select {
case <-s.vaultStop:
return
case <-t.C:
if err := s.vaultPub.Drain(context.Background()); err != nil {
slog.Warn("vault_drain_failed", "error", err)
}
}
}
}()
atomic.StoreInt32(&s.vaultPublisherStarted, 1)
}
func (s *Server) tryVaultDegraded(ctx context.Context, env, logical, projectKey string) ([]byte, string, error) {
if s == nil || s.vaultPub == nil || s.cfg == nil || !s.cfg.FAVDegradedRead {
return nil, "", fmt.Errorf("artifactvault: degraded read disabled")
}
return s.vaultPub.DegradedPayload(ctx, env, logical, projectKey)
}
func (s *Server) handleVaultStatus(w http.ResponseWriter, r *http.Request) {
_ = r
cfg := s.cfg
degraded := false
publisher := false
storeReady := false
if cfg != nil {
degraded = cfg.FAVDegradedRead
publisher = cfg.FAVPublisherEnabled
}
if s.vaultStore != nil {
storeReady = true
}
writeJSONResponse(w, http.StatusOK, map[string]any{
"status": "ok",
"consumer": "fav",
"consumer_mode": "off",
"publisher_enabled": publisher,
"publisher_running": atomic.LoadInt32(&s.vaultPublisherStarted) == 1,
"degraded_read": degraded,
"store_configured": storeReady,
"subscriber_default": false,
"shadow_observe": true,
"would_stage": false,
"would_publish": false,
"live_not_opened": true,
"note": "fav consumer remains OFF until separate approval; shadow computes intended outcome only; full FAV branch not merged",
})
}
func (s *Server) vaultPublisherRunning() bool {
return atomic.LoadInt32(&s.vaultPublisherStarted) == 1
}