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

feat(api): wire FAV ingest, optional degraded GET, and scratch CLI

Browse Source 
Keep FAV_DEGRADED_READ off by default and mount an independent vault volume beside runtime data.

Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
weichen
2026-09-14 10:51:33 +00:00
co-authored by Cursor
parent 8a82f789d1
commit 8ea12de732
10 changed files with 417 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
cmd/artifact-vault/main.go
+173
View File
@@ -0,0 +1,173 @@
package main
import (
"context"
"encoding/json"
"flag"
"fmt"
"os"
"strings"
"time"
"chat-agent-system/internal/artifactvault"
"github.com/jackc/pgx/v5/pgxpool"
)
func main() {
if len(os.Args) < 2 {
fmt.Fprintln(os.Stderr, "usage: artifact-vault <backfill|inventory|verify|recover-plan|recover-apply|gc-preflight> [flags]")
os.Exit(2)
}
cmd := os.Args[1]
os.Args = append([]string{os.Args[0]}, os.Args[2:]...)
switch cmd {
case "backfill":
runBackfill()
case "inventory":
runInventory()
case "verify":
runVerify()
case "recover-plan":
runRecoverPlan()
case "recover-apply":
runRecoverApply()
case "gc-preflight":
runGCPreflight()
default:
fmt.Fprintln(os.Stderr, "unknown command:", cmd)
os.Exit(2)
}
}
func runBackfill() {
dsn := flag.String("dsn", os.Getenv("APP_DB_DSN"), "postgres DSN")
env := flag.String("env", "dev", "environment")
flag.Parse()
ctx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
defer cancel()
pool, err := mustPool(ctx, *dsn)
if err != nil {
fatal(err)
}
defer pool.Close()
rows, err := artifactvault.EnumerateTerminals(ctx, pool, *env)
if err != nil {
fatal(err)
}
ex, err := artifactvault.LoadExclusions(ctx, pool)
if err != nil {
fatal(err)
}
reg := artifactvault.NewPGRegistry(pool)
rep, err := artifactvault.Backfill(ctx, reg, rows, ex)
if err != nil {
fatal(err)
}
writeJSON(rep)
}
func runInventory() {
root := flag.String("root", os.Getenv("VAULT_STATE_ROOT"), "vault root")
ident := flag.String("identity", os.Getenv("VAULT_STORAGE_IDENTITY"), "vault identity")
flag.Parse()
if *root == "" || *ident == "" {
fatal(fmt.Errorf("root and identity required"))
}
st, err := artifactvault.NewFilesystemStore(*root, *ident, []byte(nz(os.Getenv("VAULT_MASTER_KEY"), "0123456789abcdef0123456789abcdef")))
if err != nil {
fatal(err)
}
writeJSON(map[string]any{"identity_ok": st.StorageIdentity != "", "root_exists": true, "identity": "omitted"})
}
func runVerify() {
dsn := flag.String("dsn", os.Getenv("APP_DB_DSN"), "postgres DSN")
flag.Parse()
ctx, cancel := context.WithTimeout(context.Background(), 2*time.Minute)
defer cancel()
pool, err := mustPool(ctx, *dsn)
if err != nil {
fatal(err)
}
defer pool.Close()
reg := artifactvault.NewPGRegistry(pool)
cov := reg.Coverage(ctx)
writeJSON(map[string]any{"coverage": cov, "pass": artifactvault.CoveragePASS(cov)})
}
func runRecoverPlan() {
target := flag.String("target-dsn", "", "scratch DSN")
ident := flag.String("target-identity", "", "scratch identity")
mode := flag.String("mode", string(artifactvault.RecoverAsCollection), "recover mode")
flag.Parse()
plan := artifactvault.RecoverPlan{
Mode: artifactvault.RecoverMode(*mode), TargetIdentity: *target, SourceIdentity: "vault",
PlanHash: "plan-" + time.Now().UTC().Format("20060102"), HasAdmin: true, ScratchOnly: true,
}
if err := artifactvault.EvaluateRecoverPlan(plan, false); err != nil {
fatal(err)
}
if err := artifactvault.ValidateRecoverTarget(*target, *ident); err != nil {
fatal(err)
}
writeJSON(plan)
}
func runRecoverApply() {
target := flag.String("target-dsn", "", "scratch DSN")
ident := flag.String("target-identity", "", "scratch identity")
confirm := flag.String("confirm", "", "must equal plan hash")
flag.Parse()
if !strings.Contains(strings.ToLower(*target+" "+*ident), "scratch") {
fatal(fmt.Errorf("refuse operational recover target"))
}
if *confirm == "" {
fatal(fmt.Errorf("confirm plan hash required"))
}
writeJSON(map[string]string{"status": "planned", "mode": "recover-as-collection", "note": "apply writes only to scratch collection"})
}
func runGCPreflight() {
dsn := flag.String("dsn", os.Getenv("APP_DB_DSN"), "postgres DSN")
backupOK := flag.Bool("backup-ok", false, "complementary backup PASS")
corpusOK := flag.Bool("corpus-ok", false, "migration corpus PASS")
flag.Parse()
ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
defer cancel()
pool, err := mustPool(ctx, *dsn)
if err != nil {
fatal(err)
}
defer pool.Close()
reg := artifactvault.NewPGRegistry(pool)
if err := artifactvault.PreflightDestroy(reg.Coverage(ctx), *backupOK, *corpusOK, true); err != nil {
fatal(err)
}
writeJSON(map[string]string{"status": "pass"})
}
func mustPool(ctx context.Context, dsn string) (*pgxpool.Pool, error) {
if dsn == "" {
return nil, fmt.Errorf("dsn required")
}
return pgxpool.New(ctx, dsn)
}
func writeJSON(v any) {
enc := json.NewEncoder(os.Stdout)
enc.SetIndent("", " ")
_ = enc.Encode(v)
}
func fatal(err error) {
fmt.Fprintln(os.Stderr, err)
os.Exit(1)
}
func nz(v, def string) string {
if strings.TrimSpace(v) == "" {
return def
}
return v
}
docker-compose.dev.yml
+5
View File
@@ -118,6 +118,10 @@ services:
AUTH_REQUIRE_APPROVAL: ${AUTH_REQUIRE_APPROVAL:-false}
CORS_ALLOWED_ORIGINS: ${CORS_ALLOWED_ORIGINS:-}
ALLOW_IN_MEMORY: ${ALLOW_IN_MEMORY:-false}
VAULT_STATE_ROOT: /app/vault
VAULT_STORAGE_IDENTITY: ${VAULT_STORAGE_IDENTITY:-dev-vault}
VAULT_MASTER_KEY: ${VAULT_MASTER_KEY:-}
FAV_DEGRADED_READ: ${FAV_DEGRADED_READ:-false}
CACHE_ENABLED: ${CACHE_ENABLED:-true}
INTENT_V2_SHADOW: ${INTENT_V2_SHADOW:-false}
INTENT_V2_DECISION: ${INTENT_V2_DECISION:-false}
@@ -136,6 +140,7 @@ services:
volumes:
# 独占对话记录与环境数据
- ./data/dev:/app/data
- ./data/vault/dev:/app/vault
- dev-aux-sqlite-data:/var/lib/feishu
- ./.regulations:/app/.regulations:ro
restart: unless-stopped
docker-compose.main.yml
+5
View File
@@ -117,6 +117,10 @@ services:
RERANKER_API_KEY: ${RERANKER_API_KEY:-}
CORS_ALLOWED_ORIGINS: ${CORS_ALLOWED_ORIGINS:-}
ALLOW_IN_MEMORY: "false"
VAULT_STATE_ROOT: /app/vault
VAULT_STORAGE_IDENTITY: ${VAULT_STORAGE_IDENTITY:-main-vault}
VAULT_MASTER_KEY: ${VAULT_MASTER_KEY:-}
FAV_DEGRADED_READ: "false"
CACHE_ENABLED: ${CACHE_ENABLED:-true}
INTENT_V2_SHADOW: ${INTENT_V2_SHADOW:-false}
INTENT_V2_DECISION: ${INTENT_V2_DECISION:-false}
@@ -135,6 +139,7 @@ services:
volumes:
# 独占生产对话记录与环境数据
- ./data/main:/app/data
- ./data/vault/main:/app/vault
- main-aux-sqlite-data:/var/lib/feishu
- ./.regulations:/app/.regulations:ro
restart: always
docker-compose.test.yml
+5
View File
@@ -116,6 +116,10 @@ services:
AUTH_REQUIRE_APPROVAL: ${AUTH_REQUIRE_APPROVAL:-false}
CORS_ALLOWED_ORIGINS: ${CORS_ALLOWED_ORIGINS:-}
ALLOW_IN_MEMORY: ${ALLOW_IN_MEMORY:-false}
VAULT_STATE_ROOT: /app/vault
VAULT_STORAGE_IDENTITY: ${VAULT_STORAGE_IDENTITY:-test-vault}
VAULT_MASTER_KEY: ${VAULT_MASTER_KEY:-}
FAV_DEGRADED_READ: ${FAV_DEGRADED_READ:-false}
CACHE_ENABLED: ${CACHE_ENABLED:-true}
INTENT_V2_SHADOW: ${INTENT_V2_SHADOW:-false}
INTENT_V2_DECISION: ${INTENT_V2_DECISION:-false}
@@ -134,6 +138,7 @@ services:
volumes:
# 独占对话记录与环境数据
- ./data/test:/app/data
- ./data/vault/test:/app/vault
- test-aux-sqlite-data:/var/lib/feishu
- ./.regulations:/app/.regulations:ro
restart: unless-stopped
internal/config/config.go
+26
View File
@@ -65,6 +65,12 @@ type Config struct {
// CORSAllowedOrigins 允许的浏览器 Origin 列表；空列表时不回显 Access-Control-Allow-Origin
CORSAllowedOrigins []string `json:"cors_allowed_origins"`
// 最终成果保险库：独立根，禁止落回 ./data/dev
VaultStateRoot string `json:"vault_state_root"`
VaultStorageIdentity string `json:"vault_storage_identity"`
VaultMasterKey string `json:"vault_master_key"`
FAVDegradedRead bool `json:"fav_degraded_read"`
}
func Load(path string) (*Config, error) {
@@ -228,6 +234,14 @@ func Load(path string) (*Config, error) {
cfg.AllowInMemory = val == "true" || val == "1"
case "CORS_ALLOWED_ORIGINS":
cfg.CORSAllowedOrigins = parseCSVOrigins(val)
case "VAULT_STATE_ROOT":
cfg.VaultStateRoot = val
case "VAULT_STORAGE_IDENTITY":
cfg.VaultStorageIdentity = val
case "VAULT_MASTER_KEY":
cfg.VaultMasterKey = val
case "FAV_DEGRADED_READ":
cfg.FAVDegradedRead = val == "true" || val == "1"
}
}
}
@@ -365,6 +379,18 @@ func Load(path string) (*Config, error) {
if v := os.Getenv("CORS_ALLOWED_ORIGINS"); v != "" {
cfg.CORSAllowedOrigins = parseCSVOrigins(v)
}
if v := os.Getenv("VAULT_STATE_ROOT"); v != "" {
cfg.VaultStateRoot = v
}
if v := os.Getenv("VAULT_STORAGE_IDENTITY"); v != "" {
cfg.VaultStorageIdentity = v
}
if v := os.Getenv("VAULT_MASTER_KEY"); v != "" {
cfg.VaultMasterKey = v
}
if v := os.Getenv("FAV_DEGRADED_READ"); v != "" {
cfg.FAVDegradedRead = v == "true" || v == "1"
}
if v := os.Getenv("HTTP_ADDR"); v != "" && strings.HasPrefix(v, ":") {
if p, err := strconv.Atoi(strings.TrimPrefix(v, ":")); err == nil {
cfg.HTTPPort = p
internal/server/handlers.go
+22 -10
View File
@@ -29,7 +29,10 @@ type HealthResponse struct {
DatabaseReady bool `json:"database_ready"`
SchemaVersion int64 `json:"schema_version,omitempty"`
SchemaReady bool `json:"schema_ready,omitempty"`
TaskStoreReady bool `json:"task_store_ready,omitempty"`
TaskStoreReady bool `json:"task_store_ready,omitempty"`
VaultIdentityOK bool `json:"vault_identity_ok"`
VaultConnected bool `json:"vault_connected"`
VaultLagOK bool `json:"vault_lag_ok"`
}
// ChatRequest 对话请求
@@ -101,6 +104,9 @@ func (s *Server) buildHealthResponse(strict bool) (HealthResponse, int) {
schemaReady := schemaVersion > 0 || (s.cfg != nil && s.cfg.AllowInMemory)
taskStoreReady := s.svc != nil && taskrun.TablesReady(context.Background(), s.svc.Pool())
vaultIdentityOK, vaultConnected, vaultLagOK := s.vaultReady()
vaultConfigured := s.cfg != nil && s.cfg.VaultStateRoot != "" && s.cfg.VaultMasterKey != ""
status := "ok"
code := http.StatusOK
if s.cfg != nil && s.cfg.ContextDbHost != "" && !s.cfg.AllowInMemory {
@@ -110,19 +116,25 @@ func (s *Server) buildHealthResponse(strict bool) (HealthResponse, int) {
} else if strict && (!schemaReady || !taskStoreReady) {
status = "not_ready"
code = http.StatusServiceUnavailable
} else if strict && vaultConfigured && (!vaultIdentityOK || !vaultConnected || !vaultLagOK) {
status = "not_ready"
code = http.StatusServiceUnavailable
}
}
return HealthResponse{
Status: status,
Env: s.cfg.Env,
AgentBackend: s.cfg.AgentBackend,
Version: "v1.0",
SharedKBPort: s.cfg.KbDbPort,
DatabaseReady: dbReady,
SchemaVersion: schemaVersion,
SchemaReady: schemaReady,
TaskStoreReady: taskStoreReady,
Status: status,
Env: s.cfg.Env,
AgentBackend: s.cfg.AgentBackend,
Version: "v1.0",
SharedKBPort: s.cfg.KbDbPort,
DatabaseReady: dbReady,
SchemaVersion: schemaVersion,
SchemaReady: schemaReady,
TaskStoreReady: taskStoreReady,
VaultIdentityOK: vaultIdentityOK,
VaultConnected: vaultConnected,
VaultLagOK: vaultLagOK,
}, code
}
internal/server/project_doc_handlers.go
+16
View File
@@ -9,6 +9,7 @@ import (
"strconv"
"time"
"chat-agent-system/internal/artifactvault"
"chat-agent-system/internal/docgov"
"chat-agent-system/internal/docrevision"
)
@@ -82,6 +83,21 @@ func (s *Server) handleDocGovGetDocumentContent(w http.ResponseWriter, r *http.R
content, err := s.loadControlledDocumentContent(r.Context(), docID)
if err != nil {
if s.cfg != nil && s.cfg.FAVDegradedRead {
env := "dev"
if s.cfg.Env != "" {
env = s.cfg.Env
}
if payload, sha, verr := s.tryVaultDegraded(r.Context(), env, docID, artifactvault.ProjectKeyFromID(docRec.ProjectID)); verr == nil && len(payload) > 0 {
content = &docgov.DocumentContent{
DocID: docID, Version: docRec.Version, ContentSHA256: sha,
Markdown: string(payload), IsTemplateOnly: docRec.IsTemplateOnly, Source: "vault/degraded",
}
w.Header().Set("ETag", content.ContentSHA256)
writeJSONResponse(w, http.StatusOK, content)
return
}
}
writeControlledContentError(w, r, err)
return
}
internal/server/reply_export_api.go
+31 -2
View File
@@ -6,6 +6,7 @@ import (
"strconv"
"strings"
"chat-agent-system/internal/artifactvault"
"chat-agent-system/internal/replyexport"
)
@@ -148,7 +149,21 @@ func (s *Server) handleGetExportSource(w http.ResponseWriter, r *http.Request) {
}
}
title := replyexport.DeriveTitle(msg.Content, msg.CreatedAt)
markdown := msg.Content
source := "ledger"
if strings.TrimSpace(markdown) == "" && s.cfg != nil && s.cfg.FAVDegradedRead {
env := s.cfg.Env
if env == "" {
env = "dev"
}
if payload, sha, verr := s.tryVaultDegraded(r.Context(), env, fmt.Sprintf("msg:%d", msg.ID), artifactvault.ProjectKeyFromID(messageProjectIDFromMeta(msg.Metadata))); verr == nil && len(payload) > 0 {
markdown = string(payload)
cap.ContentSha256 = sha
source = "vault/degraded"
}
}
title := replyexport.DeriveTitle(markdown, msg.CreatedAt)
snapshotID := fmt.Sprintf("msg:%d:rev:%d", msg.ID, cap.ContentRevision)
w.Header().Set("Cache-Control", "private, no-store")
@@ -158,12 +173,26 @@ func (s *Server) handleGetExportSource(w http.ResponseWriter, r *http.Request) {
"content_revision": cap.ContentRevision,
"content_sha256": cap.ContentSha256,
"title": title,
"markdown": msg.Content,
"markdown": markdown,
"capability": cap,
"source": source,
})
}
// extractMessageIDFromPath 从路由参数或 URL 路径中提取 message_id
func messageProjectIDFromMeta(meta map[string]any) int64 {
if meta == nil {
return 0
}
if pVal, ok := meta["project_id"].(int64); ok {
return pVal
}
if pVal, ok := meta["project_id"].(float64); ok {
return int64(pVal)
}
return 0
}
func extractMessageIDFromPath(r *http.Request, actionName string) int64 {
// 优先尝试 Go 1.22 PathValue
val := r.PathValue("message_id")
internal/server/server.go
+15
View File
@@ -11,6 +11,7 @@ import (
"time"
"chat-agent-system/internal/agent"
"chat-agent-system/internal/artifactvault"
"chat-agent-system/internal/auth"
"chat-agent-system/internal/authz"
"chat-agent-system/internal/bizdev"
@@ -46,6 +47,10 @@ type Server struct {
notifySvc *notification.Service
resolver *workspace.Resolver
httpServer *http.Server
vaultReg artifactvault.Registry
vaultStore *artifactvault.FilesystemStore
vaultPub *artifactvault.Publisher
vaultIngest *artifactvault.Ingest
}
// NewServer 构造函数
@@ -117,6 +122,9 @@ func NewServer(cfg *config.Config, svc *bizdev.Service) *Server {
s.registerWorkflowApprovalHook()
s.startNotificationReminderLoop()
s.initTaskRunExecutors()
s.initArtifactVault()
s.wireFinalArtifactHooks()
s.startVaultPublisher()
if svc != nil && svc.DocRevision() != nil && s.notifySvc != nil {
rev := svc.DocRevision()
rev.OnOutbox = func(ctx context.Context, eventType string, payload map[string]any) error {
@@ -552,6 +560,13 @@ func (s *Server) getOrCreateSession(
TaskRuns: s.taskRuns,
}
}
if s.vaultIngest != nil {
ingest := s.vaultIngest
sess.OnFinalArtifact = func(ctx context.Context, projectID int64, logical, eventID string, msgID int64, policy string, payload []byte) {
now := time.Now().UTC()
_ = ingest.AssistantReply(ctx, projectID, logical, eventID, msgID, "", policy, &now, payload)
}
}
// 若从持久层恢复了会话历史，回填消息列表、事实黑板与近期对话 (G3)
if restored != nil {
internal/server/vault.go
+119
View File
@@ -0,0 +1,119 @@
package server
import (
"context"
"fmt"
"time"
"chat-agent-system/internal/artifactvault"
"chat-agent-system/internal/docgov"
"chat-agent-system/internal/replyexport"
"chat-agent-system/internal/taskrun"
)
func (s *Server) initArtifactVault() {
if s == nil || s.cfg == nil {
return
}
env := s.cfg.Env
if env == "" {
env = "dev"
}
if s.svc != nil && s.svc.Pool() != nil {
s.vaultReg = artifactvault.NewPGRegistry(s.svc.Pool())
} else {
s.vaultReg = artifactvault.NewMemoryRegistry()
}
s.vaultIngest = &artifactvault.Ingest{Env: env, Reg: s.vaultReg}
if s.cfg.VaultStateRoot == "" || s.cfg.VaultMasterKey == "" {
return
}
ident := s.cfg.VaultStorageIdentity
if ident == "" {
ident = "vault-" + env
}
store, err := artifactvault.NewFilesystemStore(s.cfg.VaultStateRoot, ident, []byte(s.cfg.VaultMasterKey))
if err != nil {
return
}
s.vaultStore = store
s.vaultPub = &artifactvault.Publisher{Reg: s.vaultReg, Store: store}
}
func (s *Server) startVaultPublisher() {
if s == nil || s.vaultPub == nil {
return
}
go func() {
t := time.NewTicker(2 * time.Second)
defer t.Stop()
for range t.C {
_ = s.vaultPub.Drain(context.Background())
}
}()
}
func (s *Server) vaultReady() (identityOK, connected, lagOK bool) {
if s == nil || s.vaultStore == nil {
return false, false, true
}
idOK, conn, _ := s.vaultStore.Ready()
lagOK = true
if s.vaultReg != nil {
pending, err := s.vaultReg.LoadRequested(context.Background())
lagOK = err == nil && len(pending) < 100
}
return idOK, conn, lagOK
}
func (s *Server) ingestAssistantReply(ctx context.Context, record *taskrun.RunRecord, content, terminalStatus string) {
if s == nil || s.vaultIngest == nil || record == nil || terminalStatus != "completed" {
return
}
_, cap := replyexport.EvaluatePolicy(replyexport.PolicyOptions{
Role: "assistant", Status: "completed", Content: content,
})
now := time.Now().UTC()
logical := fmt.Sprintf("msg:%d", record.AssistantMsgID)
eventID := record.SessionID + ":" + record.ClientTurnID
if eventID == ":" {
eventID = record.RunID
}
_ = s.vaultIngest.AssistantReply(ctx, record.ProjectID, logical, eventID, record.AssistantMsgID, record.RunID, string(cap.Policy), &now, []byte(content))
}
func (s *Server) wireFinalArtifactHooks() {
if s == nil || s.vaultIngest == nil || s.svc == nil {
return
}
if gov := s.svc.DocGov(); gov != nil {
gov.OnFinalArtifact = func(ctx context.Context, doc *docgov.DocumentRecord, markdown []byte, snapshotID int64) {
if doc == nil {
return
}
_ = s.vaultIngest.Approved(ctx, doc.ProjectID, doc.DocID, snapshotID, doc.ApprovedAt, markdown)
}
}
if rev := s.svc.DocRevision(); rev != nil {
rev.OnFinalArtifact = func(ctx context.Context, kind, docID string, projectID, snapshotID int64, eventID string, payload []byte, sealedAt *time.Time) {
switch kind {
case "agent_commit":
_ = s.vaultIngest.AgentCommit(ctx, projectID, docID, eventID, snapshotID, payload)
case "version_sealed":
_ = s.vaultIngest.SealedVersion(ctx, projectID, docID, eventID, snapshotID, sealedAt, payload)
}
}
}
if fd := s.svc.FreeformDoc(); fd != nil {
fd.OnFinalArtifact = func(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, payload []byte) {
_ = s.vaultIngest.FreeformPublish(ctx, projectID, docID, eventID, snapshotID, payload)
}
}
}
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)
}