From ad9b67d599d1bb95a724b8b95e0c070a9bd41ada Mon Sep 17 00:00:00 2001 From: weichen Date: Mon, 14 Sep 2026 12:52:28 +0000 Subject: [PATCH] feat(artifactvault): close same-tx staging, recover-apply, and GET zero-write gaps Approve/commit now write FAV staging in the domain transaction, recover-apply materializes scratch collections, and unpublished or GET paths cannot mutate the registry. Co-authored-by: Cursor --- cmd/artifact-vault/main.go | 61 ++++++++++- internal/artifactvault/eligibility_test.go | 20 ++++ internal/artifactvault/filesystem_store.go | 106 +++++++++++++++++-- internal/artifactvault/ingest.go | 59 +++++++++++ internal/artifactvault/publisher.go | 2 +- internal/artifactvault/publisher_test.go | 116 +++++++++++++++++++++ internal/artifactvault/recovery.go | 72 +++++++++++++ internal/artifactvault/recovery_test.go | 40 ++++++- internal/config/config.go | 21 ++++ internal/config/config_test.go | 12 +++ internal/ctxmgr/chat_turns.go | 46 ++++++++ internal/docgov/engine.go | 74 +++++++++++-- internal/docrevision/pg_mutate.go | 25 ++++- internal/docrevision/service.go | 5 +- internal/server/reply_export_api.go | 78 ++++++++++++++ internal/server/reply_export_api_test.go | 43 ++++++++ internal/server/server.go | 1 + internal/server/vault.go | 22 ++++ internal/server/vault_nowrite_test.go | 60 +++++++++++ 19 files changed, 838 insertions(+), 25 deletions(-) create mode 100644 internal/server/vault_nowrite_test.go diff --git a/cmd/artifact-vault/main.go b/cmd/artifact-vault/main.go index b81e57fe..59cc1454 100644 --- a/cmd/artifact-vault/main.go +++ b/cmd/artifact-vault/main.go @@ -16,7 +16,7 @@ import ( func main() { if len(os.Args) < 2 { - fmt.Fprintln(os.Stderr, "usage: artifact-vault [flags]") + fmt.Fprintln(os.Stderr, "usage: artifact-vault [flags]") os.Exit(2) } cmd := os.Args[1] @@ -34,6 +34,8 @@ func main() { runRecoverApply() case "gc-preflight": runGCPreflight() + case "destroy": + runDestroy() default: fmt.Fprintln(os.Stderr, "unknown command:", cmd) os.Exit(2) @@ -115,17 +117,68 @@ func runRecoverPlan() { } func runRecoverApply() { + sourceRoot := flag.String("source-root", os.Getenv("VAULT_STATE_ROOT"), "vault root") + destDir := flag.String("dest-dir", "", "scratch collection directory") target := flag.String("target-dsn", "", "scratch DSN") ident := flag.String("target-identity", "", "scratch identity") + sourceIdent := flag.String("identity", os.Getenv("VAULT_STORAGE_IDENTITY"), "source vault 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 err := artifactvault.ValidateRecoverTarget(*target, *ident); err != nil { + fatal(err) } 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"}) + if *sourceRoot == "" || *sourceIdent == "" { + fatal(fmt.Errorf("source-root and identity required")) + } + plan := artifactvault.RecoverPlan{ + Mode: artifactvault.RecoverAsCollection, TargetIdentity: *target, SourceIdentity: *ident, + PlanHash: *confirm, HasAdmin: true, ScratchOnly: true, + } + store, err := artifactvault.NewFilesystemStore(*sourceRoot, *sourceIdent, []byte(nz(os.Getenv("VAULT_MASTER_KEY"), "0123456789abcdef0123456789abcdef"))) + if err != nil { + fatal(err) + } + arts, payloads, err := store.LoadCommittedArtifacts() + if err != nil { + fatal(err) + } + denied := map[string]bool{} + for _, art := range arts { + if art.PublicationStatus == artifactvault.StatusTombstoned || store.HasTombstone(art.UUID) { + denied[art.UUID] = true + } + } + out, err := artifactvault.ApplyRecoverAsCollection(plan, *destDir, arts, payloads, denied) + if err != nil { + fatal(err) + } + writeJSON(out) +} + +func runDestroy() { + 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") + confirm := flag.String("confirm", "", "must equal DESTROY") + flag.Parse() + if *confirm != "DESTROY" { + fatal(fmt.Errorf("destroy requires --confirm=DESTROY")) + } + 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": "preflight-pass", "note": "dataset wipe is operator-owned after complementary PASS"}) } func runGCPreflight() { diff --git a/internal/artifactvault/eligibility_test.go b/internal/artifactvault/eligibility_test.go index 04121980..fbb410a3 100644 --- a/internal/artifactvault/eligibility_test.go +++ b/internal/artifactvault/eligibility_test.go @@ -64,6 +64,26 @@ func TestEvaluateEligibility_SensitiveRejectedBeforeStaging(t *testing.T) { } } +func TestEvaluateEligibility_CopyOnlyRequiresExplicitPinAfterCompleted(t *testing.T) { + now := time.Now().UTC() + base := Facts{ + ArtifactType: TypeAssistantReply, ChatTurnStatus: "completed", CompletedAt: &now, + AssistantMessageID: 9, ExportPolicy: "copy_only", Classification: ClassificationOK, + Payload: []byte("ok, noted."), + } + if err := EvaluateEligibility(base); err == nil { + t.Fatal("copy_only without pin must reject") + } + base.ExplicitPin = true + if err := EvaluateEligibility(base); err != ErrExplicitBypass { + t.Fatalf("pin without PinAfterCompleted: %v", err) + } + base.PinAfterCompleted = true + if err := EvaluateEligibility(base); err != nil { + t.Fatalf("copy_only + pin after completed should pass: %v", err) + } +} + func TestEvaluateEligibility_UnknownTypeRejected(t *testing.T) { if err := EvaluateEligibility(Facts{ArtifactType: "workspace_scan", Classification: ClassificationOK, Payload: []byte("x")}); err != ErrUnknownArtifactType { t.Fatalf("got %v", err) diff --git a/internal/artifactvault/filesystem_store.go b/internal/artifactvault/filesystem_store.go index 9ba946e5..3b47e84f 100644 --- a/internal/artifactvault/filesystem_store.go +++ b/internal/artifactvault/filesystem_store.go @@ -63,10 +63,7 @@ func (s *FilesystemStore) PutBlob(env, projectKey, sha string, plain []byte) (ke if _, err := os.Stat(path); err == nil { return ref, nil } - if err := fsutil.WriteFileAtomicExclusive(path, envBytes, 0o600); err != nil { - if strings.Contains(err.Error(), "refuse overwrite") { - return ref, nil - } + if err := writeExclusiveOK(path, envBytes, 0o600); err != nil { return "", err } return ref, nil @@ -91,14 +88,111 @@ func (s *FilesystemStore) GetBlob(env, projectKey, sha string) ([]byte, error) { return plain, nil } +func writeExclusiveOK(path string, raw []byte, perm os.FileMode) error { + if err := fsutil.WriteFileAtomicExclusive(path, raw, perm); err != nil { + if strings.Contains(err.Error(), "refuse overwrite") { + return nil + } + return err + } + return nil +} + func (s *FilesystemStore) PutManifest(artifactUUID, manifestSHA string, raw []byte) error { path := filepath.Join(s.Root, "manifests", artifactUUID, manifestSHA+".json") - return fsutil.WriteFileAtomicExclusive(path, raw, 0o644) + return writeExclusiveOK(path, raw, 0o644) } func (s *FilesystemStore) PutCommit(artifactUUID, manifestSHA string, raw []byte) error { path := filepath.Join(s.Root, "commits", artifactUUID, manifestSHA+".json") - return fsutil.WriteFileAtomicExclusive(path, raw, 0o644) + return writeExclusiveOK(path, raw, 0o644) +} + +func (s *FilesystemStore) ListCommitted() ([]string, error) { + dir := filepath.Join(s.Root, "commits") + ents, err := os.ReadDir(dir) + if err != nil { + if os.IsNotExist(err) { + return nil, nil + } + return nil, err + } + out := make([]string, 0, len(ents)) + for _, e := range ents { + if e.IsDir() { + out = append(out, e.Name()) + } + } + return out, nil +} + +func (s *FilesystemStore) LoadCommittedArtifacts() ([]Artifact, map[string][]byte, error) { + uuids, err := s.ListCommitted() + if err != nil { + return nil, nil, err + } + payloads := make(map[string][]byte) + arts := make([]Artifact, 0, len(uuids)) + for _, uuid := range uuids { + if s.HasTombstone(uuid) { + arts = append(arts, Artifact{UUID: uuid, PublicationStatus: StatusTombstoned}) + continue + } + art, body, err := s.loadCommitted(uuid) + if err != nil { + return nil, nil, err + } + arts = append(arts, art) + payloads[uuid] = body + } + return arts, payloads, nil +} + +func (s *FilesystemStore) loadCommitted(uuid string) (Artifact, []byte, error) { + dir := filepath.Join(s.Root, "commits", uuid) + ents, err := os.ReadDir(dir) + if err != nil || len(ents) == 0 { + return Artifact{}, nil, fmt.Errorf("artifactvault: unpublished") + } + raw, err := os.ReadFile(filepath.Join(dir, ents[0].Name())) + if err != nil { + return Artifact{}, nil, err + } + var marker CommitMarker + if err := json.Unmarshal(raw, &marker); err != nil { + return Artifact{}, nil, err + } + manRaw, err := os.ReadFile(filepath.Join(s.Root, "manifests", uuid, marker.ManifestSHA256+".json")) + if err != nil { + return Artifact{}, nil, err + } + var man Manifest + if err := json.Unmarshal(manRaw, &man); err != nil { + return Artifact{}, nil, err + } + if len(man.PayloadMembers) == 0 { + return Artifact{}, nil, fmt.Errorf("artifactvault: manifest has no payload members") + } + body, err := s.GetBlob(man.Environment, man.ProjectExternalKey, man.PayloadMembers[0].SHA256) + if err != nil { + return Artifact{}, nil, err + } + var snap, msg int64 + if man.SourceSnapshotID != nil { + snap = *man.SourceSnapshotID + } + if man.SourceMessageID != nil { + msg = *man.SourceMessageID + } + return Artifact{ + UUID: uuid, ArtifactType: man.ArtifactType, LogicalResourceID: man.LogicalResourceID, + Environment: man.Environment, ProjectID: man.ProjectID, ProjectExternalKey: man.ProjectExternalKey, + SourceEventID: man.SourceEventID, SourceEventKind: man.SourceEventKind, + SourceSnapshotID: snap, SourceMessageID: msg, SourceRunID: man.SourceRunID, + PayloadSHA256: man.PayloadMembers[0].SHA256, ManifestSHA256: man.ManifestSHA256, + VaultStorageIdentity: man.VaultStorageIdentity, TerminalState: man.TerminalEvidence.State, + PublicationStatus: StatusVerified, RetentionClass: man.RetentionClass, + }, body, nil } func (s *FilesystemStore) HasCommit(artifactUUID, manifestSHA string) bool { diff --git a/internal/artifactvault/ingest.go b/internal/artifactvault/ingest.go index e228675c..1b39230d 100644 --- a/internal/artifactvault/ingest.go +++ b/internal/artifactvault/ingest.go @@ -5,6 +5,8 @@ import ( "fmt" "strconv" "time" + + "github.com/jackc/pgx/v5" ) // Ingest is the single write path from domain terminals into staging/outbox. @@ -120,6 +122,52 @@ func (i *Ingest) FreeformPublish(ctx context.Context, projectID int64, docID, ev }) } +func (i *Ingest) requestTX(ctx context.Context, tx pgx.Tx, facts Facts, req PublishRequest) error { + if i == nil { + return nil + } + if req.Environment == "" { + req.Environment = i.Env + } + if req.ProjectExternalKey == "" { + req.ProjectExternalKey = ProjectKeyFromID(req.ProjectID) + } + if req.PolicyVersion == "" { + req.PolicyVersion = PolicyVersion + } + if req.Classification == "" { + req.Classification = ClassifyPayload(req.Payload) + } + facts.Payload = req.Payload + facts.Classification = req.Classification + _, _, err := RequestWithTx(ctx, tx, facts, req) + return err +} + +func (i *Ingest) AgentCommitTX(ctx context.Context, tx pgx.Tx, projectID int64, docID, eventID string, snapshotID int64, payload []byte) error { + return i.requestTX(ctx, tx, AgentCommitFacts(snapshotID), PublishRequest{ + ArtifactType: TypeAgentCommit, Environment: i.Env, ProjectID: projectID, + LogicalResourceID: docID, SourceEventKind: "agent_commit", SourceEventID: eventID, + SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "committed", + }) +} + +func (i *Ingest) SealedVersionTX(ctx context.Context, tx pgx.Tx, projectID int64, docID, eventID string, snapshotID int64, sealedAt *time.Time, payload []byte) error { + return i.requestTX(ctx, tx, SealedFacts(sealedAt, snapshotID), PublishRequest{ + ArtifactType: TypeSealedVersion, Environment: i.Env, ProjectID: projectID, + LogicalResourceID: docID + ":sealed", SourceEventKind: "version_sealed", SourceEventID: eventID, + SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "sealed", + }) +} + +func (i *Ingest) ApprovedTX(ctx context.Context, tx pgx.Tx, projectID int64, docID string, snapshotID int64, approvedAt *time.Time, payload []byte) error { + return i.requestTX(ctx, tx, ApprovedFacts("approved", approvedAt, snapshotID), PublishRequest{ + ArtifactType: TypeApprovedDocument, Environment: i.Env, ProjectID: projectID, + LogicalResourceID: docID, SourceEventKind: "approved", SourceEventID: fmt.Sprintf("approved:%s:%d", docID, snapshotID), + SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "approved", + }) +} + func (i *Ingest) AssistantReply(ctx context.Context, projectID int64, logical, eventID string, msgID int64, runID, policy string, completedAt *time.Time, payload []byte) error { return i.Request(ctx, AssistantFacts("completed", completedAt, msgID, policy), PublishRequest{ ArtifactType: TypeAssistantReply, Environment: i.Env, ProjectID: projectID, @@ -127,3 +175,14 @@ func (i *Ingest) AssistantReply(ctx context.Context, projectID int64, logical, e SourceMessageID: msgID, SourceRunID: runID, Payload: payload, TerminalState: "completed", }) } + +func (i *Ingest) AssistantReplyPin(ctx context.Context, projectID int64, logical, eventID string, msgID int64, runID, policy string, completedAt *time.Time, payload []byte) error { + facts := AssistantFacts("completed", completedAt, msgID, policy) + facts.ExplicitPin = true + facts.PinAfterCompleted = true + return i.Request(ctx, facts, PublishRequest{ + ArtifactType: TypeAssistantReply, Environment: i.Env, ProjectID: projectID, + LogicalResourceID: logical, SourceEventKind: "assistant_pin", SourceEventID: eventID, + SourceMessageID: msgID, SourceRunID: runID, Payload: payload, TerminalState: "completed", + }) +} diff --git a/internal/artifactvault/publisher.go b/internal/artifactvault/publisher.go index 92683cec..bed03099 100644 --- a/internal/artifactvault/publisher.go +++ b/internal/artifactvault/publisher.go @@ -87,7 +87,7 @@ func (p *Publisher) DegradedPayload(ctx context.Context, env, logical, projectKe if p.Reg.IsDenied(ctx, art.UUID) || p.Store.HasTombstone(art.UUID) { return nil, "", fmt.Errorf("artifactvault: union-deny") } - if !p.Store.HasCommit(art.UUID, art.ManifestSHA256) && art.ManifestSHA256 != "" { + if art.ManifestSHA256 == "" || !p.Store.HasCommit(art.UUID, art.ManifestSHA256) { return nil, "", fmt.Errorf("artifactvault: unpublished") } b, err := p.Store.GetBlob(env, projectKey, art.PayloadSHA256) diff --git a/internal/artifactvault/publisher_test.go b/internal/artifactvault/publisher_test.go index 2e05a3da..b33b1758 100644 --- a/internal/artifactvault/publisher_test.go +++ b/internal/artifactvault/publisher_test.go @@ -147,4 +147,120 @@ func TestLastReferencePurgeAndLegalHold(t *testing.T) { } } +func TestFilesystemIdentityMismatch(t *testing.T) { + dir := t.TempDir() + key := []byte("0123456789abcdef0123456789abcdef") + if _, err := NewFilesystemStore(dir, "id-a", key); err != nil { + t.Fatal(err) + } + if _, err := NewFilesystemStore(dir, "id-b", key); err == nil { + t.Fatal("identity mismatch must fail closed") + } +} + +func TestUnpublishedWithoutCommitMarkerInvisible(t *testing.T) { + ctx := context.Background() + reg := NewMemoryRegistry() + store := testStore(t) + pub := &Publisher{Reg: reg, Store: store} + facts := Facts{ + ArtifactType: TypeAgentCommit, CandidateStatus: "committed", OutputSnapshotID: 4, + ChangeKind: "agent_commit", ChangeToSnapshotID: 4, Classification: ClassificationOK, + Payload: []byte("# pending\n"), + } + req := PublishRequest{ + ArtifactType: TypeAgentCommit, Environment: "dev", ProjectExternalKey: "proj-a", + LogicalResourceID: "doc_pending", SourceEventKind: "agent_commit", SourceEventID: "ev-pending", + Payload: []byte("# pending\n"), TerminalState: "committed", + } + if _, _, err := reg.Request(ctx, facts, req); err != nil { + t.Fatal(err) + } + if _, _, err := pub.DegradedPayload(ctx, "dev", "doc_pending", "proj-a"); err == nil { + t.Fatal("requested but unpublished must be invisible") + } +} + +func TestPublisherRetriesAfterMissingCommit(t *testing.T) { + ctx := context.Background() + reg := NewMemoryRegistry() + store := testStore(t) + pub := &Publisher{Reg: reg, Store: store} + payload := []byte("# retry-me\n") + facts := Facts{ + ArtifactType: TypeAgentCommit, CandidateStatus: "committed", OutputSnapshotID: 5, + ChangeKind: "agent_commit", ChangeToSnapshotID: 5, Classification: ClassificationOK, + Payload: payload, + } + req := PublishRequest{ + ArtifactType: TypeAgentCommit, Environment: "dev", ProjectExternalKey: "proj-a", + LogicalResourceID: "doc_retry", SourceEventKind: "agent_commit", SourceEventID: "ev-retry", + Payload: payload, TerminalState: "committed", + } + art, _, err := reg.Request(ctx, facts, req) + if err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + if err := os.RemoveAll(filepath.Join(store.Root, "commits", art.UUID)); err != nil { + t.Fatal(err) + } + if _, _, err := pub.DegradedPayload(ctx, "dev", "doc_retry", "proj-a"); err == nil { + t.Fatal("missing commit marker must hide payload") + } + if err := reg.MarkFailed(ctx, art.UUID, "commit_missing"); err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + b, _, err := pub.DegradedPayload(ctx, "dev", "doc_retry", "proj-a") + if err != nil || string(b) != string(payload) { + t.Fatalf("retry after exclusive blob/manifest: %v %q", err, b) + } +} + +type requestCountingReg struct { + Registry + requests int +} + +func (r *requestCountingReg) Request(ctx context.Context, facts Facts, req PublishRequest) (*Artifact, *Staging, error) { + r.requests++ + return r.Registry.Request(ctx, facts, req) +} + +func TestDegradedPayloadDoesNotWrite(t *testing.T) { + ctx := context.Background() + mem := NewMemoryRegistry() + count := &requestCountingReg{Registry: mem} + store := testStore(t) + pub := &Publisher{Reg: count, Store: store} + facts := Facts{ + ArtifactType: TypeAgentCommit, CandidateStatus: "committed", OutputSnapshotID: 6, + ChangeKind: "agent_commit", ChangeToSnapshotID: 6, Classification: ClassificationOK, + Payload: []byte("# get\n"), + } + req := PublishRequest{ + ArtifactType: TypeAgentCommit, Environment: "dev", ProjectExternalKey: "proj-a", + LogicalResourceID: "doc_get", SourceEventKind: "agent_commit", SourceEventID: "ev-get", + Payload: []byte("# get\n"), TerminalState: "committed", + } + if _, _, err := mem.Request(ctx, facts, req); err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + count.requests = 0 + if _, _, err := pub.DegradedPayload(ctx, "dev", "doc_get", "proj-a"); err != nil { + t.Fatal(err) + } + if count.requests != 0 { + t.Fatalf("GET/degraded read wrote Request %d times", count.requests) + } +} + func ptrTime() *time.Time { t := time.Now().UTC(); return &t } diff --git a/internal/artifactvault/recovery.go b/internal/artifactvault/recovery.go index bec72832..f47a3b12 100644 --- a/internal/artifactvault/recovery.go +++ b/internal/artifactvault/recovery.go @@ -1,7 +1,10 @@ package artifactvault import ( + "encoding/json" "fmt" + "os" + "path/filepath" "strings" ) @@ -65,3 +68,72 @@ func EvaluateRecoverPlan(p RecoverPlan, denied bool) error { } return nil } + +type RecoveredItem struct { + ArtifactUUID string `json:"artifact_uuid"` + LogicalResourceID string `json:"logical_resource_id"` + PayloadSHA256 string `json:"payload_sha256"` + SkippedReason string `json:"skipped_reason,omitempty"` +} + +type RecoveredCollection struct { + Mode RecoverMode `json:"mode"` + PlanHash string `json:"plan_hash"` + DestDir string `json:"dest_dir"` + Items []RecoveredItem `json:"items"` +} + +// ApplyRecoverAsCollection writes recovered payloads into a scratch collection directory. +// It never writes operational databases or relinks live heads. +func ApplyRecoverAsCollection(plan RecoverPlan, destDir string, arts []Artifact, payloads map[string][]byte, denied map[string]bool) (*RecoveredCollection, error) { + if err := EvaluateRecoverPlan(plan, false); err != nil { + return nil, err + } + if plan.Mode != RecoverAsCollection && plan.Mode != "" { + return nil, fmt.Errorf("artifactvault: apply only supports recover-as-collection") + } + if !destIsScratch(destDir) { + return nil, fmt.Errorf("artifactvault: collection dest must be scratch") + } + if err := os.MkdirAll(filepath.Join(destDir, "payloads"), 0o700); err != nil { + return nil, err + } + out := &RecoveredCollection{Mode: RecoverAsCollection, PlanHash: plan.PlanHash, DestDir: destDir} + for _, art := range arts { + item := RecoveredItem{ArtifactUUID: art.UUID, LogicalResourceID: art.LogicalResourceID, PayloadSHA256: art.PayloadSHA256} + if denied[art.UUID] || art.PublicationStatus == StatusTombstoned { + item.SkippedReason = "union-deny" + out.Items = append(out.Items, item) + continue + } + body := payloads[art.UUID] + if len(body) == 0 || SHA256Hex(body) != art.PayloadSHA256 { + item.SkippedReason = "sha_mismatch" + out.Items = append(out.Items, item) + continue + } + path := filepath.Join(destDir, "payloads", art.UUID+".md") + if err := os.WriteFile(path, body, 0o600); err != nil { + return nil, err + } + out.Items = append(out.Items, item) + } + raw, err := json.MarshalIndent(out, "", " ") + if err != nil { + return nil, err + } + if err := os.WriteFile(filepath.Join(destDir, "collection.json"), raw, 0o600); err != nil { + return nil, err + } + return out, nil +} + +func destIsScratch(destDir string) bool { + for _, part := range strings.Split(filepath.ToSlash(filepath.Clean(destDir)), "/") { + low := strings.ToLower(part) + if low == "scratch" || strings.HasPrefix(low, "scratch-") || strings.HasPrefix(low, "scratch_") { + return true + } + } + return false +} diff --git a/internal/artifactvault/recovery_test.go b/internal/artifactvault/recovery_test.go index c4152ba3..659f05a6 100644 --- a/internal/artifactvault/recovery_test.go +++ b/internal/artifactvault/recovery_test.go @@ -1,6 +1,10 @@ package artifactvault -import "testing" +import ( + "os" + "path/filepath" + "testing" +) func TestRecoverRejectsOperationalDSN(t *testing.T) { if err := ValidateRecoverTarget("postgres://x/agent_system_dev", "dev"); err == nil { @@ -25,3 +29,37 @@ func TestRecoverPlanGates(t *testing.T) { t.Fatal("relink+conflict") } } + +func TestApplyRecoverAsCollectionWritesScratchAndSkipsDenied(t *testing.T) { + dest := filepath.Join(t.TempDir(), "scratch-collection") + plan := RecoverPlan{ + Mode: RecoverAsCollection, TargetIdentity: "scratch-db", SourceIdentity: "vault", + PlanHash: "plan-1", HasAdmin: true, ScratchOnly: true, + } + okBody := []byte("# recovered\n") + arts := []Artifact{ + {UUID: "art-ok", LogicalResourceID: "doc_ok", PayloadSHA256: SHA256Hex(okBody)}, + {UUID: "art-deny", LogicalResourceID: "doc_deny", PayloadSHA256: SHA256Hex([]byte("x")), PublicationStatus: StatusTombstoned}, + } + payloads := map[string][]byte{"art-ok": okBody} + out, err := ApplyRecoverAsCollection(plan, dest, arts, payloads, map[string]bool{"art-deny": true}) + if err != nil { + t.Fatal(err) + } + if _, err := os.Stat(filepath.Join(dest, "collection.json")); err != nil { + t.Fatal(err) + } + got, err := os.ReadFile(filepath.Join(dest, "payloads", "art-ok.md")) + if err != nil || string(got) != string(okBody) { + t.Fatalf("payload: %v %q", err, got) + } + if _, err := os.Stat(filepath.Join(dest, "payloads", "art-deny.md")); err == nil { + t.Fatal("denied artifact must not be written") + } + if len(out.Items) != 2 || out.Items[1].SkippedReason != "union-deny" { + t.Fatalf("items=%+v", out.Items) + } + if _, err := ApplyRecoverAsCollection(plan, filepath.Join(t.TempDir(), "live-restore"), arts, payloads, nil); err == nil { + t.Fatal("non-scratch dest must fail") + } +} diff --git a/internal/config/config.go b/internal/config/config.go index d161fdcf..47f6070a 100644 --- a/internal/config/config.go +++ b/internal/config/config.go @@ -4,6 +4,7 @@ import ( "bufio" "fmt" "os" + "path/filepath" "strconv" "strings" "time" @@ -403,10 +404,30 @@ func Load(path string) (*Config, error) { if err := validateProductionSecrets(cfg); err != nil { return nil, err } + if err := ValidateIndependentVault(cfg.VaultStateRoot, cfg.DataDir); err != nil { + return nil, err + } return cfg, nil } +func ValidateIndependentVault(root, dataDir string) error { + if strings.TrimSpace(root) == "" { + return nil + } + cr := filepath.Clean(root) + if strings.HasSuffix(cr, filepath.Join("data", "dev")) || strings.HasSuffix(cr, filepath.Join("data", "test")) || strings.HasSuffix(cr, filepath.Join("data", "main")) { + return fmt.Errorf("config: VAULT_STATE_ROOT must not be runtime data root %s", root) + } + if dataDir != "" { + cd := filepath.Clean(dataDir) + if cr == cd { + return fmt.Errorf("config: VAULT_STATE_ROOT must be independent from DATA_DIR") + } + } + return nil +} + func parseCSVOrigins(raw string) []string { parts := strings.Split(raw, ",") out := make([]string, 0, len(parts)) diff --git a/internal/config/config_test.go b/internal/config/config_test.go index ec789ed0..72f62def 100644 --- a/internal/config/config_test.go +++ b/internal/config/config_test.go @@ -133,6 +133,18 @@ func TestConfigLoadRejectsWeakSecretsOutsideDev(t *testing.T) { } } +func TestValidateIndependentVault(t *testing.T) { + if err := config.ValidateIndependentVault("", "./data/dev"); err != nil { + t.Fatal(err) + } + if err := config.ValidateIndependentVault("./data/dev", "./data/dev"); err == nil { + t.Fatal("runtime data/dev must be rejected") + } + if err := config.ValidateIndependentVault("./data/vault/dev", "./data/dev"); err != nil { + t.Fatalf("independent vault root should pass: %v", err) + } +} + func TestConfigLoadRejectsProductionDatabasePlaceholder(t *testing.T) { t.Setenv("ENV", "prod") t.Setenv("KB_DB_PASSWORD", "kb-prod-secret-xyz") diff --git a/internal/ctxmgr/chat_turns.go b/internal/ctxmgr/chat_turns.go index 2d2c7c84..48fcd501 100644 --- a/internal/ctxmgr/chat_turns.go +++ b/internal/ctxmgr/chat_turns.go @@ -332,3 +332,49 @@ func (m *ContextManager) GetChatTurn(ctx context.Context, sessionID, clientTurnI cp := *rec return &cp, nil } + +// GetChatTurnByAssistantMessageID 按会话 + 助手消息定位回合；用于 copy_only 显式 pin。 +func (m *ContextManager) GetChatTurnByAssistantMessageID(ctx context.Context, sessionID string, assistantMessageID int64) (*ChatTurnRecord, error) { + if sessionID == "" || assistantMessageID <= 0 { + return nil, ErrChatTurnNotFound + } + if m.pool != nil { + var rec ChatTurnRecord + var completedAt *time.Time + var leaseExpiresAt *time.Time + err := m.pool.QueryRow(ctx, ` + SELECT id, session_id, client_turn_id, + COALESCE(user_id, 0), COALESCE(project_id, 0), + COALESCE(user_message_id, 0), COALESCE(assistant_message_id, 0), + request_body_sha256, status, + COALESCE(first_request_id, ''), COALESCE(last_request_id, ''), + attempt_count, created_at, updated_at, completed_at, lease_expires_at + FROM chat_turns + WHERE session_id = $1 AND assistant_message_id = $2 + ORDER BY id DESC + LIMIT 1 + `, sessionID, assistantMessageID).Scan( + &rec.ID, &rec.SessionID, &rec.ClientTurnID, + &rec.UserID, &rec.ProjectID, + &rec.UserMessageID, &rec.AssistantMessageID, + &rec.RequestBodySHA256, &rec.Status, + &rec.FirstRequestID, &rec.LastRequestID, + &rec.AttemptCount, &rec.CreatedAt, &rec.UpdatedAt, &completedAt, &leaseExpiresAt, + ) + if err != nil { + return nil, ErrChatTurnNotFound + } + rec.CompletedAt = completedAt + rec.LeaseExpiresAt = leaseExpiresAt + return &rec, nil + } + memTurnsMu.Lock() + defer memTurnsMu.Unlock() + for _, rec := range memTurns { + if rec != nil && rec.SessionID == sessionID && rec.AssistantMessageID == assistantMessageID { + cp := *rec + return &cp, nil + } + } + return nil, ErrChatTurnNotFound +} diff --git a/internal/docgov/engine.go b/internal/docgov/engine.go index d339e50d..0a77d6af 100644 --- a/internal/docgov/engine.go +++ b/internal/docgov/engine.go @@ -8,6 +8,7 @@ import ( "sync" "time" + "github.com/jackc/pgx/v5" "github.com/jackc/pgx/v5/pgxpool" ) @@ -32,7 +33,8 @@ type GovernanceEngine struct { memAudits map[string][]*AuditLog onApprovedHook func(ctx context.Context, doc *DocumentRecord) onApprovedHooks []func(ctx context.Context, doc *DocumentRecord) - OnFinalArtifact func(ctx context.Context, doc *DocumentRecord, markdown []byte, snapshotID int64) + OnFinalArtifact func(ctx context.Context, doc *DocumentRecord, markdown []byte, snapshotID int64) + OnFinalArtifactTX func(ctx context.Context, tx pgx.Tx, doc *DocumentRecord, markdown []byte, snapshotID int64) error } // SetOnApprovedHook 注册文档终审批准时的回调钩子 @@ -873,11 +875,13 @@ func (e *GovernanceEngine) ApproverApprove(ctx context.Context, docID string, ac } e.memAudits[docID] = append(e.memAudits[docID], audit) - e.syncDocAndAuditToDB(ctx, doc, audit) + if err := e.persistApproveAndFAV(ctx, doc, audit); err != nil { + return nil, err + } e.mu.Unlock() e.runApprovedHooks(ctx, doc) - if e.OnFinalArtifact != nil { + if e.OnFinalArtifact != nil && (e.pool == nil || e.OnFinalArtifactTX == nil) { md, _ := e.LedgerMarkdown(ctx, doc.ReviewSnapshotID) e.OnFinalArtifact(ctx, doc, []byte(md), doc.ReviewSnapshotID) } @@ -1117,6 +1121,54 @@ func (e *GovernanceEngine) LedgerMarkdown(ctx context.Context, snapshotID int64) return md, err } +func (e *GovernanceEngine) persistApproveAndFAV(ctx context.Context, doc *DocumentRecord, audit *AuditLog) error { + if e.pool == nil { + return nil + } + if e.OnFinalArtifactTX == nil { + e.syncDocAndAuditToDB(ctx, doc, audit) + return nil + } + tx, err := e.pool.Begin(ctx) + if err != nil { + return err + } + defer func() { _ = tx.Rollback(ctx) }() + if err := syncDocAndAuditTX(ctx, tx, doc, audit); err != nil { + return err + } + md, _ := e.LedgerMarkdown(ctx, doc.ReviewSnapshotID) + if err := e.OnFinalArtifactTX(ctx, tx, doc, []byte(md), doc.ReviewSnapshotID); err != nil { + return err + } + return tx.Commit(ctx) +} + +func syncDocAndAuditTX(ctx context.Context, tx pgx.Tx, doc *DocumentRecord, audit *AuditLog) error { + _, err := tx.Exec(ctx, ` + UPDATE document_records + SET current_state = $1, is_locked = $2, + last_rejected_reason = $3, last_rejected_by_role = $4, last_rejected_by_name = $5, last_rejected_at = $6, + approved_at = $7, updated_at = $8, review_snapshot_id = NULLIF($9, 0) + WHERE doc_id = $10 + `, string(doc.CurrentState), doc.IsLocked, + doc.LastRejectedReason, doc.LastRejectedByRole, doc.LastRejectedByName, doc.LastRejectedAt, + doc.ApprovedAt, doc.UpdatedAt, doc.ReviewSnapshotID, doc.DocID) + if err != nil { + return err + } + if audit == nil { + return nil + } + detJSON, _ := json.Marshal(audit.Details) + _, err = tx.Exec(ctx, ` + INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + `, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole), + string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt) + return err +} + func (e *GovernanceEngine) syncDocAndAuditToDB(ctx context.Context, doc *DocumentRecord, audit *AuditLog) { if e.pool == nil { return @@ -1130,13 +1182,13 @@ func (e *GovernanceEngine) syncDocAndAuditToDB(ctx context.Context, doc *Documen `, string(doc.CurrentState), doc.IsLocked, doc.LastRejectedReason, doc.LastRejectedByRole, doc.LastRejectedByName, doc.LastRejectedAt, doc.ApprovedAt, doc.UpdatedAt, doc.ReviewSnapshotID, doc.DocID) - - if audit != nil { - detJSON, _ := json.Marshal(audit.Details) - _, _ = e.pool.Exec(ctx, ` - INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at) - VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) - `, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole), - string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt) + if audit == nil { + return } + detJSON, _ := json.Marshal(audit.Details) + _, _ = e.pool.Exec(ctx, ` + INSERT INTO document_audit_logs (doc_id, action, actor_id, actor_name, actor_role, from_state, to_state, details, created_at) + VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9) + `, audit.DocID, audit.Action, audit.ActorID, audit.ActorName, string(audit.ActorRole), + string(audit.FromState), string(audit.ToState), detJSON, audit.CreatedAt) } diff --git a/internal/docrevision/pg_mutate.go b/internal/docrevision/pg_mutate.go index 28d49f50..f29cb374 100644 --- a/internal/docrevision/pg_mutate.go +++ b/internal/docrevision/pg_mutate.go @@ -187,7 +187,30 @@ func (s *Service) persistAgentCommitLocked(ctx context.Context, root *DocRootPoi if newVer.ChangeReportID != nil { _, _ = tx.Exec(ctx, `UPDATE document_versions SET change_report_id=$2 WHERE id=$1`, vid, *newVer.ChangeReportID) } - return s.projectRootPointersTX(ctx, tx, root) + if err := s.projectRootPointersTX(ctx, tx, root); err != nil { + return err + } + if s.OnFinalArtifactTX == nil { + return nil + } + eventID := fmt.Sprintf("ev:%d", ev.ID) + outMD := []byte(nil) + if b := s.blobs[outSnap.ContentSHA256]; b != nil { + outMD = []byte(b.ContentMarkdown) + } + if err := s.OnFinalArtifactTX(ctx, tx, "agent_commit", root.DocID, root.ProjectID, outSnap.ID, eventID, outMD, nil); err != nil { + return err + } + if oldVer != nil && oldVer.SealedAt != nil { + inMD := []byte(nil) + if snap := s.snapshots[oldVer.HeadSnapshotID]; snap != nil { + if b := s.blobs[snap.ContentSHA256]; b != nil { + inMD = []byte(b.ContentMarkdown) + } + } + return s.OnFinalArtifactTX(ctx, tx, "version_sealed", root.DocID, root.ProjectID, oldVer.HeadSnapshotID, eventID+":sealed", inMD, oldVer.SealedAt) + } + return nil }) } diff --git a/internal/docrevision/service.go b/internal/docrevision/service.go index 3d1896e6..e64c0136 100644 --- a/internal/docrevision/service.go +++ b/internal/docrevision/service.go @@ -38,6 +38,9 @@ type Service struct { // OnFinalArtifact is invoked after a terminal ledger persist (agent_commit / version_sealed). OnFinalArtifact func(ctx context.Context, kind, docID string, projectID, snapshotID int64, eventID string, payload []byte, sealedAt *time.Time) + // OnFinalArtifactTX writes FAV staging/outbox inside persistAgentCommitLocked. + OnFinalArtifactTX func(ctx context.Context, tx pgx.Tx, kind, docID string, projectID, snapshotID int64, eventID string, payload []byte, sealedAt *time.Time) error + // ImpactResolver returns deterministic downstream impacts (lineage); optional. ImpactResolver func(ctx context.Context, docID string) []DocImpact @@ -674,7 +677,7 @@ func (s *Service) CommitAgentVersion(ctx context.Context, docID string, candidat s.forgetDocLocked(docID) return nil, err } - if s.OnFinalArtifact != nil { + if s.OnFinalArtifact != nil && (!s.usingPG() || s.OnFinalArtifactTX == nil) { eventID := fmt.Sprintf("ev:%d", ev.ID) s.OnFinalArtifact(ctx, "agent_commit", c.DocID, root.ProjectID, outSnap.ID, eventID, []byte(outMD), nil) if oldVer != nil && oldVer.SealedAt != nil { diff --git a/internal/server/reply_export_api.go b/internal/server/reply_export_api.go index 3db26990..a549fbf1 100644 --- a/internal/server/reply_export_api.go +++ b/internal/server/reply_export_api.go @@ -7,6 +7,7 @@ import ( "strings" "chat-agent-system/internal/artifactvault" + "chat-agent-system/internal/ctxmgr" "chat-agent-system/internal/replyexport" ) @@ -179,6 +180,83 @@ func (s *Server) handleGetExportSource(w http.ResponseWriter, r *http.Request) { }) } +// handlePinExport archives a completed copy_only assistant reply after an explicit pin. +// POST /api/v1/chat/messages/{message_id}/pin-export +func (s *Server) handlePinExport(w http.ResponseWriter, r *http.Request) { + user, ok := s.requireAuthenticatedUser(w, r) + if !ok { + return + } + if s.svc == nil || s.svc.ContextManager() == nil || s.vaultIngest == nil { + http.Error(w, "artifact vault not initialized", http.StatusServiceUnavailable) + return + } + msgID := extractMessageIDFromPath(r, "pin-export") + if msgID <= 0 { + writeJSONResponse(w, http.StatusBadRequest, map[string]string{ + "error_code": "INVALID_MESSAGE_ID", + "message": "无效的消息ID", + }) + return + } + msg, err := s.svc.ContextManager().GetMessageByID(r.Context(), msgID) + if err != nil || msg == nil { + writeJSONResponse(w, http.StatusNotFound, map[string]string{ + "error_code": "MESSAGE_NOT_FOUND", + "message": "指定消息不存在或已被删除", + }) + return + } + if !s.assertMessageOwnedByUser(w, r, msg, user.ID) { + return + } + if msg.Role != "assistant" { + writeJSONResponse(w, http.StatusBadRequest, map[string]string{ + "error_code": "MESSAGE_NOT_PINNABLE", + "message": "仅助手回复可显式 pin", + }) + return + } + turn, err := s.svc.ContextManager().GetChatTurnByAssistantMessageID(r.Context(), msg.SessionID, msg.ID) + if err != nil || turn == nil || turn.Status != ctxmgr.TurnStatusCompleted || turn.CompletedAt == nil { + writeJSONResponse(w, http.StatusConflict, map[string]string{ + "error_code": "TURN_NOT_COMPLETED", + "message": "未完成回合不得 pin 进保险库", + }) + return + } + cap, okCap := msg.ExportCapability.(replyexport.MessageExportCapability) + if !okCap { + _, cap = replyexport.EvaluatePolicy(replyexport.PolicyOptions{ + Role: msg.Role, Status: "completed", Content: msg.Content, Revision: msg.ContentRevision, + }) + } + if cap.Policy != replyexport.PolicyCopyOnly { + writeJSONResponse(w, http.StatusBadRequest, map[string]string{ + "error_code": "PIN_POLICY_REJECTED", + "message": "仅 copy_only 需要显式 pin；full 已自动归档，none 永不进入", + }) + return + } + eventID := fmt.Sprintf("pin:%d:%d", msg.ID, cap.ContentRevision) + err = s.vaultIngest.AssistantReplyPin( + r.Context(), turn.ProjectID, fmt.Sprintf("msg:%d", msg.ID), eventID, + msg.ID, "", string(cap.Policy), turn.CompletedAt, []byte(msg.Content), + ) + if err != nil { + writeJSONResponse(w, http.StatusConflict, map[string]string{ + "error_code": "PIN_REJECTED", + "message": "保险库拒绝该 pin", + }) + return + } + writeJSONResponse(w, http.StatusOK, map[string]any{ + "message_id": msg.ID, + "policy": cap.Policy, + "status": "requested", + }) +} + // extractMessageIDFromPath 从路由参数或 URL 路径中提取 message_id func messageProjectIDFromMeta(meta map[string]any) int64 { if meta == nil { diff --git a/internal/server/reply_export_api_test.go b/internal/server/reply_export_api_test.go index 1bbdee42..f9c6b23c 100644 --- a/internal/server/reply_export_api_test.go +++ b/internal/server/reply_export_api_test.go @@ -179,3 +179,46 @@ func TestServer_ReplyExportAPI(t *testing.T) { t.Fatalf("expected 403 for cross-user export, got %d: %s", recIDOR.Code, recIDOR.Body.String()) } } + +func TestServer_PinExportCopyOnlyAfterCompleted(t *testing.T) { + cfg := &config.Config{ + Env: "dev", + HTTPHost: "127.0.0.1", + HTTPPort: 8894, + DataDir: t.TempDir(), + } + svc := bizdev.MustNewService(t, cfg) + srv := server.NewServer(cfg, svc) + handler := srv.Routes() + ctx := t.Context() + cm := svc.ContextManager() + sessID := "sess_pin_export" + token := loginTestAdmin(t, handler, cfg) + if _, err := cm.EnsureChatSession(ctx, sessID, 1, 7, "pin 测试"); err != nil { + t.Fatal(err) + } + if _, err := cm.AcquireChatTurn(ctx, sessID, "turn-pin", "hash-pin", "req-pin", 1, 7); err != nil { + t.Fatal(err) + } + msgID, err := cm.RecordUserMessageWithID(ctx, sessID, 1, "assistant", "好的", "", "", nil) + if err != nil { + t.Fatal(err) + } + reqEarly := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/chat/messages/%d/pin-export", msgID), nil) + withBearerAuth(reqEarly, token) + recEarly := httptest.NewRecorder() + handler.ServeHTTP(recEarly, reqEarly) + if recEarly.Code != http.StatusConflict { + t.Fatalf("pin before completed want 409, got %d: %s", recEarly.Code, recEarly.Body.String()) + } + if err := cm.CompleteChatTurn(ctx, sessID, "turn-pin", msgID); err != nil { + t.Fatal(err) + } + req := httptest.NewRequest(http.MethodPost, fmt.Sprintf("/api/v1/chat/messages/%d/pin-export", msgID), nil) + withBearerAuth(req, token) + rec := httptest.NewRecorder() + handler.ServeHTTP(rec, req) + if rec.Code != http.StatusOK { + t.Fatalf("pin after completed want 200, got %d: %s", rec.Code, rec.Body.String()) + } +} diff --git a/internal/server/server.go b/internal/server/server.go index a80776ed..0abab364 100644 --- a/internal/server/server.go +++ b/internal/server/server.go @@ -293,6 +293,7 @@ func (s *Server) Routes() http.Handler { // 7.1 单次助手回复导出与快照 mux.HandleFunc("GET /api/v1/chat/messages/{message_id}/export-capability", s.handleGetExportCapability) mux.HandleFunc("GET /api/v1/chat/messages/{message_id}/export-source", s.handleGetExportSource) + mux.HandleFunc("POST /api/v1/chat/messages/{message_id}/pin-export", s.handlePinExport) // 7.2 自由私密文档与项目共享参考 mux.HandleFunc("POST /api/v1/freeform-documents", s.handleCreateFreeformDoc) diff --git a/internal/server/vault.go b/internal/server/vault.go index cef55f87..73db3f71 100644 --- a/internal/server/vault.go +++ b/internal/server/vault.go @@ -6,9 +6,12 @@ import ( "time" "chat-agent-system/internal/artifactvault" + "chat-agent-system/internal/config" "chat-agent-system/internal/docgov" "chat-agent-system/internal/replyexport" "chat-agent-system/internal/taskrun" + + "github.com/jackc/pgx/v5" ) func (s *Server) initArtifactVault() { @@ -28,6 +31,9 @@ func (s *Server) initArtifactVault() { 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 == "" { ident = "vault-" + env @@ -93,6 +99,12 @@ func (s *Server) wireFinalArtifactHooks() { } _ = s.vaultIngest.Approved(ctx, doc.ProjectID, doc.DocID, snapshotID, doc.ApprovedAt, markdown) } + gov.OnFinalArtifactTX = func(ctx context.Context, tx pgx.Tx, doc *docgov.DocumentRecord, markdown []byte, snapshotID int64) error { + if doc == nil { + return nil + } + return s.vaultIngest.ApprovedTX(ctx, tx, 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) { @@ -103,6 +115,16 @@ func (s *Server) wireFinalArtifactHooks() { _ = s.vaultIngest.SealedVersion(ctx, projectID, docID, eventID, snapshotID, sealedAt, payload) } } + rev.OnFinalArtifactTX = func(ctx context.Context, tx pgx.Tx, kind, docID string, projectID, snapshotID int64, eventID string, payload []byte, sealedAt *time.Time) error { + switch kind { + case "agent_commit": + return s.vaultIngest.AgentCommitTX(ctx, tx, projectID, docID, eventID, snapshotID, payload) + case "version_sealed": + return s.vaultIngest.SealedVersionTX(ctx, tx, projectID, docID, eventID, snapshotID, sealedAt, payload) + default: + return nil + } + } } if fd := s.svc.FreeformDoc(); fd != nil { fd.OnFinalArtifact = func(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, payload []byte) { diff --git a/internal/server/vault_nowrite_test.go b/internal/server/vault_nowrite_test.go new file mode 100644 index 00000000..1a91b496 --- /dev/null +++ b/internal/server/vault_nowrite_test.go @@ -0,0 +1,60 @@ +package server + +import ( + "context" + "fmt" + "testing" + + "chat-agent-system/internal/artifactvault" + "chat-agent-system/internal/config" +) + +type writeFailRegistry struct { + *artifactvault.MemoryRegistry + requests int +} + +func (w *writeFailRegistry) Request(ctx context.Context, facts artifactvault.Facts, req artifactvault.PublishRequest) (*artifactvault.Artifact, *artifactvault.Staging, error) { + w.requests++ + return nil, nil, fmt.Errorf("GET must not write") +} + +func TestTryVaultDegradedDoesNotRequest(t *testing.T) { + ctx := context.Background() + mem := artifactvault.NewMemoryRegistry() + spy := &writeFailRegistry{MemoryRegistry: mem} + dir := t.TempDir() + store, err := artifactvault.NewFilesystemStore(dir, "vault-id-test", []byte("0123456789abcdef0123456789abcdef")) + if err != nil { + t.Fatal(err) + } + pub := &artifactvault.Publisher{Reg: mem, Store: store} + facts := artifactvault.Facts{ + ArtifactType: artifactvault.TypeAgentCommit, CandidateStatus: "committed", OutputSnapshotID: 7, + ChangeKind: "agent_commit", ChangeToSnapshotID: 7, Classification: artifactvault.ClassificationOK, + Payload: []byte("# ledger\n"), + } + req := artifactvault.PublishRequest{ + ArtifactType: artifactvault.TypeAgentCommit, Environment: "dev", ProjectExternalKey: "3", + LogicalResourceID: "doc_nowrite", SourceEventKind: "agent_commit", SourceEventID: "ev-nowrite", + Payload: []byte("# ledger\n"), TerminalState: "committed", + } + if _, _, err := mem.Request(ctx, facts, req); err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + s := &Server{ + cfg: &config.Config{Env: "dev", FAVDegradedRead: true, DataDir: t.TempDir()}, + vaultReg: spy, + vaultPub: &artifactvault.Publisher{Reg: spy, Store: store}, + } + b, _, err := s.tryVaultDegraded(ctx, "dev", "doc_nowrite", "3") + if err != nil || string(b) != "# ledger\n" { + t.Fatalf("degraded: %v %q", err, b) + } + if spy.requests != 0 { + t.Fatalf("GET path issued %d Request writes", spy.requests) + } +}