diff --git a/cmd/artifact-vault/main.go b/cmd/artifact-vault/main.go new file mode 100644 index 00000000..b81e57fe --- /dev/null +++ b/cmd/artifact-vault/main.go @@ -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 [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 +} diff --git a/docker-compose.dev.yml b/docker-compose.dev.yml index fa6c6aa4..53ac9864 100644 --- a/docker-compose.dev.yml +++ b/docker-compose.dev.yml @@ -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 diff --git a/docker-compose.main.yml b/docker-compose.main.yml index 553c770d..9364d506 100644 --- a/docker-compose.main.yml +++ b/docker-compose.main.yml @@ -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 diff --git a/docker-compose.test.yml b/docker-compose.test.yml index d0d880e4..82dcc1d4 100644 --- a/docker-compose.test.yml +++ b/docker-compose.test.yml @@ -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 diff --git a/internal/config/config.go b/internal/config/config.go index c6d52240..d161fdcf 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -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 diff --git a/internal/server/handlers.go b/internal/server/handlers.go index 09bbe592..c73207a2 100644 --- a/internal/server/handlers.go +++ b/internal/server/handlers.go @@ -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 } diff --git a/internal/server/project_doc_handlers.go b/internal/server/project_doc_handlers.go index 30792e32..2b2930bc 100644 --- a/internal/server/project_doc_handlers.go +++ b/internal/server/project_doc_handlers.go @@ -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 } diff --git a/internal/server/reply_export_api.go b/internal/server/reply_export_api.go index 5165899c..3db26990 100644 --- a/internal/server/reply_export_api.go +++ b/internal/server/reply_export_api.go @@ -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") diff --git a/internal/server/server.go b/internal/server/server.go index e9731e9f..a80776ed 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -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 { diff --git a/internal/server/vault.go b/internal/server/vault.go new file mode 100644 index 00000000..cef55f87 --- /dev/null +++ b/internal/server/vault.go @@ -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) +}