weichen/Agent-System
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(api): add FAV vault status stub with gated publisher

Browse Source 
Expose GET /api/v1/vault/status and only wire in-process Drain when
FAV_PUBLISHER_ENABLED is explicitly true; degraded read stays off by default.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-16 15:40:10 +00:00
co-authored by Cursor
parent 3478a5a05a
commit 92d7a3d117
5 changed files with 187 additions and 2 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/server/auth_middleware.go
+2 -1
View File
@@ -18,7 +18,8 @@ func (s *Server) allowLegacyTestPrincipal() bool {
func isPublicRoute(method, path string) bool {
if path == "/" || path == "/health" || path == "/api/v1/health" ||
path == "/ready" || path == "/api/v1/ready" {
path == "/ready" || path == "/api/v1/ready" ||
path == "/api/v1/vault/status" {
return true
}
if method == http.MethodPost && path == "/api/v1/auth/login" {
internal/server/project_doc_handlers.go
+1 -1
View File
@@ -25,7 +25,7 @@ func (s *Server) docflowQuery() *docflow.Query {
Authorize: func(ctx context.Context, p docquery.Principal, rec *docgov.DocumentRecord) error {
return s.authorizeControlledRead(ctx, p, rec)
},
FavDegradedRead: false,
FavDegradedRead: s.cfg != nil && s.cfg.FAVDegradedRead,
}}
}
internal/server/server.go
+11
View File
@@ -14,6 +14,7 @@ import (
"chat-agent-system/internal/app/chatflow"
"chat-agent-system/internal/app/intentflow"
"chat-agent-system/internal/app/stageflow"
"chat-agent-system/internal/artifactvault"
"chat-agent-system/internal/auth"
"chat-agent-system/internal/authz"
"chat-agent-system/internal/bizdev"
@@ -71,6 +72,12 @@ type Server struct {
intentPlans intentflow.Service
planAccessFn func(ctx context.Context, projectID, userID int64) error
planSessionOwnerFn func(ctx context.Context, sessionID string) (int64, error)
// FAV stubs (U7 incremental): nil / off unless FAV_PUBLISHER_ENABLED.
vaultReg artifactvault.Registry
vaultStore *artifactvault.FilesystemStore
vaultPub *artifactvault.Publisher
vaultPublisherStarted int32
}
// NewFromRuntime attaches the HTTP adapter to an already-wired composition root.
@@ -114,6 +121,8 @@ func NewFromRuntime(rt *app.Runtime) *Server {
s.svc.SetStageAccessGate(newBizdevStageGate(s.instSvc, s.pwSvc))
}
s.initTaskRunExecutors()
s.initArtifactVault()
s.startVaultPublisher()
return s
}
@@ -163,6 +172,8 @@ func (s *Server) Routes() http.Handler {
mux.HandleFunc("GET /api/v1/health", s.handleHealth)
mux.HandleFunc("GET /ready", s.handleReadiness)
mux.HandleFunc("GET /api/v1/ready", s.handleReadiness)
// FAV status stub (publisher/degraded default off; no ingest)
mux.HandleFunc("GET /api/v1/vault/status", s.handleVaultStatus)
// 2. 项目与多工作流控制面
mux.HandleFunc("POST /api/v1/projects", s.handleCreateProject)
internal/server/vault.go
+104
View File
@@ -0,0 +1,104 @@
package server
import (
"context"
"fmt"
"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
}
go func() {
t := time.NewTicker(2 * time.Second)
defer t.Stop()
for range t.C {
_ = s.vaultPub.Drain(context.Background())
}
}()
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,
"note": "fav consumer remains OFF until separate approval; full FAV branch not merged",
})
}
func (s *Server) vaultPublisherRunning() bool {
return atomic.LoadInt32(&s.vaultPublisherStarted) == 1
}
internal/server/vault_test.go
+69
View File
@@ -0,0 +1,69 @@
package server
import (
"context"
"encoding/json"
"net/http"
"net/http/httptest"
"testing"
"chat-agent-system/internal/config"
)
func TestVaultDefaultsStayDisabled(t *testing.T) {
cfg, err := config.Load("/dev/null")
if err != nil {
t.Fatal(err)
}
if cfg.FAVDegradedRead {
t.Fatal("FAV_DEGRADED_READ must default false")
}
if cfg.FAVPublisherEnabled {
t.Fatal("FAV_PUBLISHER_ENABLED must default false")
}
s := &Server{cfg: cfg}
s.initArtifactVault()
s.startVaultPublisher()
if s.vaultPub != nil {
t.Fatal("vault publisher must stay nil when flag off")
}
if s.vaultPublisherRunning() {
t.Fatal("vault publisher must not start when flag off")
}
if _, _, err := s.tryVaultDegraded(context.Background(), "dev", "doc", "1"); err == nil {
t.Fatal("degraded read must refuse when flag off")
}
}
func TestVaultStatusReportsOff(t *testing.T) {
s := &Server{cfg: &config.Config{Env: "dev"}}
req := httptest.NewRequest(http.MethodGet, "/api/v1/vault/status", nil)
rr := httptest.NewRecorder()
s.handleVaultStatus(rr, req)
if rr.Code != http.StatusOK {
t.Fatalf("status=%d", rr.Code)
}
var body map[string]any
if err := json.Unmarshal(rr.Body.Bytes(), &body); err != nil {
t.Fatal(err)
}
if body["consumer_mode"] != "off" {
t.Fatalf("consumer_mode=%v", body["consumer_mode"])
}
if body["publisher_enabled"] != false {
t.Fatalf("publisher_enabled=%v", body["publisher_enabled"])
}
if body["degraded_read"] != false {
t.Fatalf("degraded_read=%v", body["degraded_read"])
}
}
func TestValidateIndependentVaultRejectsDataRoot(t *testing.T) {
if err := config.ValidateIndependentVault("./data/dev", "./data/dev"); err == nil {
t.Fatal("expected vault/data collision error")
}
if err := config.ValidateIndependentVault("", "./data/dev"); err != nil {
t.Fatal(err)
}
}