diff --git a/internal/artifactvault/backfill.go b/internal/artifactvault/backfill.go new file mode 100644 index 00000000..ea4959d4 --- /dev/null +++ b/internal/artifactvault/backfill.go @@ -0,0 +1,55 @@ +package artifactvault + +import ( + "context" + "fmt" +) + +// TerminalRow is an authoritative DB terminal fact for backfill. Workspace paths are forbidden. +type TerminalRow struct { + Facts Facts + Request PublishRequest + MissingEvidence bool + Reason string +} + +type BackfillReport struct { + Created int + Reused int + Quarantine int + Excluded int +} + +func Backfill(ctx context.Context, reg Registry, rows []TerminalRow, exclude map[string]string) (BackfillReport, error) { + var out BackfillReport + for _, row := range rows { + if row.MissingEvidence || len(row.Request.Payload) == 0 { + _ = reg.Quarantine(ctx, nz(row.Reason, "missing_evidence"), row.Request.LogicalResourceID) + out.Quarantine++ + continue + } + if reason, ok := exclude[row.Request.LogicalResourceID]; ok && reason != "" { + out.Excluded++ + continue + } + before := reg.Coverage(ctx).Required + if _, _, err := reg.Request(ctx, row.Facts, row.Request); err != nil { + _ = reg.Quarantine(ctx, err.Error(), row.Request.LogicalResourceID) + out.Quarantine++ + continue + } + if reg.Coverage(ctx).Required == before { + out.Reused++ + continue + } + out.Created++ + } + return out, nil +} + +func ValidateBackfillRow(row TerminalRow) error { + if row.Request.SourceEventID == "" || row.Request.LogicalResourceID == "" { + return fmt.Errorf("artifactvault: backfill row missing source identity") + } + return nil +} diff --git a/internal/artifactvault/backfill_test.go b/internal/artifactvault/backfill_test.go new file mode 100644 index 00000000..ca75ceba --- /dev/null +++ b/internal/artifactvault/backfill_test.go @@ -0,0 +1,39 @@ +package artifactvault + +import ( + "context" + "testing" + "time" +) + +func TestBackfillIdempotentNoWorkspace(t *testing.T) { + ctx := context.Background() + reg := NewMemoryRegistry() + now := time.Now().UTC() + row := TerminalRow{ + Facts: Facts{ArtifactType: TypeSealedVersion, VersionStatus: "sealed", SealedAt: &now, HeadSnapshotID: 1, Classification: ClassificationOK, Payload: []byte("snap")}, + Request: PublishRequest{ArtifactType: TypeSealedVersion, Environment: "dev", ProjectExternalKey: "p", LogicalResourceID: "doc1", SourceEventKind: "seal", SourceEventID: "e1", Payload: []byte("snap"), TerminalState: "sealed"}, + } + r1, err := Backfill(ctx, reg, []TerminalRow{row, row}, nil) + if err != nil || r1.Created != 1 || r1.Reused != 1 { + t.Fatalf("%+v %v", r1, err) + } + r2, _ := Backfill(ctx, reg, []TerminalRow{row}, nil) + if r2.Created != 0 && r2.Reused == 0 { + t.Fatalf("rerun %+v", r2) + } + missing := row + missing.MissingEvidence = true + missing.Request.LogicalResourceID = "doc-missing" + r3, _ := Backfill(ctx, reg, []TerminalRow{missing}, nil) + if r3.Quarantine != 1 { + t.Fatalf("quarantine %+v", r3) + } + ex := row + ex.Request.LogicalResourceID = "doc-ex" + ex.Request.SourceEventID = "e-ex" + r4, _ := Backfill(ctx, reg, []TerminalRow{ex}, map[string]string{"doc-ex": "approved-old"}) + if r4.Excluded != 1 { + t.Fatalf("exclude %+v", r4) + } +} diff --git a/internal/artifactvault/commit-marker.schema.json b/internal/artifactvault/commit-marker.schema.json new file mode 100644 index 00000000..958b60fa --- /dev/null +++ b/internal/artifactvault/commit-marker.schema.json @@ -0,0 +1,27 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://agent-system.local/schemas/fav-commit-marker-v1.json", + "title": "FAV Commit Marker", + "type": "object", + "additionalProperties": false, + "required": [ + "artifact_uuid", + "manifest_sha256", + "member_shas", + "vault_storage_identity", + "committed_at", + "publisher_version" + ], + "properties": { + "artifact_uuid": { "type": "string", "minLength": 8 }, + "manifest_sha256": { "type": "string", "pattern": "^[a-f0-9]{64}$" }, + "member_shas": { + "type": "array", + "minItems": 1, + "items": { "type": "string", "pattern": "^[a-f0-9]{64}$" } + }, + "vault_storage_identity": { "type": "string", "minLength": 8 }, + "committed_at": { "type": "string" }, + "publisher_version": { "type": "string" } + } +} diff --git a/internal/artifactvault/coverage.go b/internal/artifactvault/coverage.go new file mode 100644 index 00000000..6f7e490f --- /dev/null +++ b/internal/artifactvault/coverage.go @@ -0,0 +1,30 @@ +package artifactvault + +import "fmt" + +func CoveragePASS(c Coverage) bool { + if c.Required <= 0 { + return false + } + return c.Verified == c.Required && c.Quarantine == 0 +} + +func CanPurge(legalHold bool, activeRefs int) error { + if legalHold { + return fmt.Errorf("artifactvault: legal hold") + } + if activeRefs > 0 { + return fmt.Errorf("artifactvault: remaining references") + } + return nil +} + +func PreflightDestroy(c Coverage, backupOK, corpusOK, vaultReady bool) error { + if !vaultReady || !CoveragePASS(c) { + return fmt.Errorf("artifactvault: coverage gate failed required=%d verified=%d quarantine=%d", c.Required, c.Verified, c.Quarantine) + } + if !backupOK || !corpusOK { + return fmt.Errorf("artifactvault: complementary backup/corpus not PASS") + } + return nil +} diff --git a/internal/artifactvault/crypto.go b/internal/artifactvault/crypto.go new file mode 100644 index 00000000..9b3f8d35 --- /dev/null +++ b/internal/artifactvault/crypto.go @@ -0,0 +1,53 @@ +package artifactvault + +import ( + "crypto/aes" + "crypto/cipher" + "crypto/rand" + "crypto/sha256" + "encoding/hex" + "fmt" + "io" +) + +// ProjectKey derives a per-project envelope key. The master secret never enters manifests. +func ProjectKey(master []byte, env, projectExternalKey string) (key []byte, ref string, err error) { + if len(master) < 16 { + return nil, "", fmt.Errorf("artifactvault: vault master key too short") + } + sum := sha256.Sum256(append(append(master, []byte(env)...), []byte(projectExternalKey)...)) + ref = "proj:" + env + ":" + projectExternalKey + ":" + hex.EncodeToString(sum[:8]) + return sum[:], ref, nil +} + +func Encrypt(key, plaintext []byte) ([]byte, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, err + } + nonce := make([]byte, gcm.NonceSize()) + if _, err := io.ReadFull(rand.Reader, nonce); err != nil { + return nil, err + } + return gcm.Seal(nonce, nonce, plaintext, nil), nil +} + +func Decrypt(key, envelope []byte) ([]byte, error) { + block, err := aes.NewCipher(key) + if err != nil { + return nil, err + } + gcm, err := cipher.NewGCM(block) + if err != nil { + return nil, err + } + ns := gcm.NonceSize() + if len(envelope) < ns { + return nil, fmt.Errorf("artifactvault: short envelope") + } + return gcm.Open(nil, envelope[:ns], envelope[ns:], nil) +} diff --git a/internal/artifactvault/eligibility.go b/internal/artifactvault/eligibility.go new file mode 100644 index 00000000..5e7b73af --- /dev/null +++ b/internal/artifactvault/eligibility.go @@ -0,0 +1,107 @@ +package artifactvault + +import ( + "errors" + "strings" +) + +var ( + ErrNotTerminal = errors.New("artifactvault: facts are not a terminal state") + ErrExplicitBypass = errors.New("artifactvault: explicit publish/pin cannot bypass terminal allowlist") + ErrSensitiveRejected = errors.New("artifactvault: sensitive or unclassified payload rejected") + ErrUnknownArtifactType = errors.New("artifactvault: unknown artifact type rejected") + ErrSnapshotInconsistent = errors.New("artifactvault: snapshot/event inconsistent") +) + +// EvaluateEligibility is the only domain gate. Unclassified types default to reject. +func EvaluateEligibility(f Facts) error { + if err := rejectSensitive(f.Classification, f.Payload); err != nil { + return err + } + switch f.ArtifactType { + case TypeAgentCommit: + if f.CandidateStatus != "committed" || f.OutputSnapshotID <= 0 { + return ErrNotTerminal + } + if f.ChangeKind != "" && (f.ChangeKind != "agent_commit" || f.ChangeToSnapshotID != f.OutputSnapshotID) { + return ErrSnapshotInconsistent + } + return nil + case TypeSealedVersion: + if f.VersionStatus != "sealed" || f.SealedAt == nil || f.HeadSnapshotID <= 0 { + return ErrNotTerminal + } + return nil + case TypeApprovedDocument: + if f.DocState != "approved" || f.ApprovedAt == nil || f.ReviewSnapshotID <= 0 { + return ErrNotTerminal + } + return nil + case TypeFreeformPublish: + if !f.ExplicitPublish || !f.HasImmutableSnapshot { + if f.ExplicitPublish && !f.HasImmutableSnapshot { + return ErrExplicitBypass + } + return ErrNotTerminal + } + return nil + case TypeAssistantReply: + if f.ChatTurnStatus != "completed" || f.CompletedAt == nil || f.AssistantMessageID <= 0 { + if f.ExplicitPin { + return ErrExplicitBypass + } + return ErrNotTerminal + } + switch f.ExportPolicy { + case "full": + return nil + case "copy_only": + if f.ExplicitPin && f.PinAfterCompleted { + return nil + } + if f.ExplicitPin { + return ErrExplicitBypass + } + return ErrNotTerminal + default: + return ErrNotTerminal + } + default: + return ErrUnknownArtifactType + } +} + +func rejectSensitive(class string, payload []byte) error { + if class == "" { + class = ClassifyPayload(payload) + } + switch class { + case ClassificationSecret, ClassificationRestricted, ClassificationFailed: + return ErrSensitiveRejected + case ClassificationOK: + return nil + default: + return ErrSensitiveRejected + } +} + +// ClassifyPayload is a fail-closed server-side gate. It never logs the body. +func ClassifyPayload(payload []byte) string { + text := strings.ToLower(string(payload)) + if text == "" { + return ClassificationFailed + } + needles := []string{ + "-----begin ", "private_key", "api_key=", "secret=", "authorization: bearer", + "password=", "access_token=", "aws_secret", + } + for _, n := range needles { + if strings.Contains(text, n) { + return ClassificationSecret + } + } + if strings.Contains(text, "restricted:") || strings.Contains(text, "classification_failed") { + return ClassificationRestricted + } + return ClassificationOK +} diff --git a/internal/artifactvault/eligibility_test.go b/internal/artifactvault/eligibility_test.go new file mode 100644 index 00000000..04121980 --- /dev/null +++ b/internal/artifactvault/eligibility_test.go @@ -0,0 +1,87 @@ +package artifactvault + +import ( + "testing" + "time" +) + +func TestEvaluateEligibility_Allowlist(t *testing.T) { + ok := Facts{ + ArtifactType: TypeAgentCommit, CandidateStatus: "committed", OutputSnapshotID: 9, + ChangeKind: "agent_commit", ChangeToSnapshotID: 9, Classification: ClassificationOK, Payload: []byte("doc"), + } + if err := EvaluateEligibility(ok); err != nil { + t.Fatalf("committed should pass: %v", err) + } + for _, status := range []string{"ready", "needs_confirmation", "rejected", "stale", "quality_failed"} { + bad := ok + bad.CandidateStatus = status + if err := EvaluateEligibility(bad); err == nil { + t.Fatalf("status %s must reject", status) + } + } +} + +func TestEvaluateEligibility_ApprovedRequiresReviewSnapshot(t *testing.T) { + now := time.Now().UTC() + f := Facts{ + ArtifactType: TypeApprovedDocument, DocState: "approved", ApprovedAt: &now, + Classification: ClassificationOK, Payload: []byte("approved"), + } + if err := EvaluateEligibility(f); err == nil { + t.Fatal("approved without review_snapshot_id must reject") + } + f.ReviewSnapshotID = 3 + if err := EvaluateEligibility(f); err != nil { + t.Fatalf("approved with snapshot should pass: %v", err) + } +} + +func TestEvaluateEligibility_TaskWithoutCompletedTurn(t *testing.T) { + f := Facts{ + ArtifactType: TypeAssistantReply, ChatTurnStatus: "", AssistantMessageID: 8, + ExportPolicy: "full", Classification: ClassificationOK, Payload: []byte("hello world from assistant"), + } + if err := EvaluateEligibility(f); err == nil { + t.Fatal("missing completed turn must reject") + } + if err := EvaluateEligibility(Facts{ + ArtifactType: TypeAssistantReply, ExplicitPin: true, ChatTurnStatus: "pending", + Classification: ClassificationOK, Payload: []byte("hello world from assistant"), + }); err != ErrExplicitBypass { + t.Fatalf("pin before completed: %v", err) + } +} + +func TestEvaluateEligibility_SensitiveRejectedBeforeStaging(t *testing.T) { + now := time.Now().UTC() + f := Facts{ + ArtifactType: TypeAssistantReply, ChatTurnStatus: "completed", CompletedAt: &now, + AssistantMessageID: 1, ExportPolicy: "full", Payload: []byte("api_key=sk-secret"), + } + if err := EvaluateEligibility(f); err != ErrSensitiveRejected { + t.Fatalf("secret payload: %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) + } +} + +func TestIdempotencyKeyStable(t *testing.T) { + req := PublishRequest{ + ArtifactType: TypeAgentCommit, Environment: "dev", ProjectExternalKey: "p1", + LogicalResourceID: "doc_a", SourceEventKind: "agent_commit", SourceEventID: "ev1", + } + a := IdempotencyKey(req, "abc") + b := IdempotencyKey(req, "abc") + if a != b || a == "" { + t.Fatalf("key unstable %q %q", a, b) + } + req.SourceEventID = "ev2" + if IdempotencyKey(req, "abc") == a { + t.Fatal("new event must new key") + } +} diff --git a/internal/artifactvault/enumerate.go b/internal/artifactvault/enumerate.go new file mode 100644 index 00000000..058867dd --- /dev/null +++ b/internal/artifactvault/enumerate.go @@ -0,0 +1,211 @@ +package artifactvault + +import ( + "context" + "fmt" + "time" + + "github.com/jackc/pgx/v5/pgxpool" +) + +// EnumerateTerminals lists only immutable ledger / completed-turn facts. Workspace paths are forbidden. +func EnumerateTerminals(ctx context.Context, pool *pgxpool.Pool, env string) ([]TerminalRow, error) { + if pool == nil { + return nil, fmt.Errorf("artifactvault: no database") + } + var rows []TerminalRow + approved, err := enumerateApproved(ctx, pool, env) + if err != nil { + return nil, err + } + rows = append(rows, approved...) + commits, err := enumerateAgentCommits(ctx, pool, env) + if err != nil { + return nil, err + } + rows = append(rows, commits...) + sealed, err := enumerateSealed(ctx, pool, env) + if err != nil { + return nil, err + } + rows = append(rows, sealed...) + replies, err := enumerateAssistant(ctx, pool, env) + if err != nil { + return nil, err + } + rows = append(rows, replies...) + return rows, nil +} + +func LoadExclusions(ctx context.Context, pool *pgxpool.Pool) (map[string]string, error) { + out := map[string]string{} + if pool == nil { + return out, nil + } + q, err := pool.Query(ctx, `SELECT logical_resource_id, reason FROM artifact_backfill_exclusions WHERE expires_at > now()`) + if err != nil { + return nil, err + } + defer q.Close() + for q.Next() { + var id, reason string + if err := q.Scan(&id, &reason); err != nil { + return nil, err + } + out[id] = reason + } + return out, q.Err() +} + +func enumerateApproved(ctx context.Context, pool *pgxpool.Pool, env string) ([]TerminalRow, error) { + q, err := pool.Query(ctx, ` + SELECT d.doc_id, d.project_id, d.review_snapshot_id, d.approved_at, COALESCE(b.content_markdown,'') + FROM document_records d + LEFT JOIN document_snapshots s ON s.id = d.review_snapshot_id + LEFT JOIN document_content_blobs b ON b.content_sha256 = s.content_sha256 + WHERE d.current_state = 'approved' AND d.approved_at IS NOT NULL`) + if err != nil { + return nil, err + } + defer q.Close() + var out []TerminalRow + for q.Next() { + var docID string + var projectID, snapID int64 + var approvedAt *time.Time + var md string + if err := q.Scan(&docID, &projectID, &snapID, &approvedAt, &md); err != nil { + return nil, err + } + row := TerminalRow{ + Facts: ApprovedFacts("approved", approvedAt, snapID), + Request: PublishRequest{ + ArtifactType: TypeApprovedDocument, Environment: env, ProjectID: projectID, + ProjectExternalKey: ProjectKeyFromID(projectID), LogicalResourceID: docID, + SourceEventKind: "approved", SourceEventID: fmt.Sprintf("approved:%s:%d", docID, snapID), + SourceSnapshotID: snapID, Payload: []byte(md), TerminalState: "approved", + }, + } + if snapID <= 0 || md == "" { + row.MissingEvidence = true + row.Reason = "missing_review_snapshot" + } + out = append(out, row) + } + return out, q.Err() +} + +func enumerateAgentCommits(ctx context.Context, pool *pgxpool.Pool, env string) ([]TerminalRow, error) { + q, err := pool.Query(ctx, ` + SELECT e.doc_id, e.id, e.to_snapshot_id, COALESCE(r.project_id,0), COALESCE(b.content_markdown,'') + FROM document_change_events e + LEFT JOIN document_records r ON r.doc_id = e.doc_id + LEFT JOIN document_snapshots s ON s.id = e.to_snapshot_id + LEFT JOIN document_content_blobs b ON b.content_sha256 = s.content_sha256 + WHERE e.change_kind = 'agent_commit'`) + if err != nil { + return nil, err + } + defer q.Close() + var out []TerminalRow + for q.Next() { + var docID string + var eventID, snapID, projectID int64 + var md string + if err := q.Scan(&docID, &eventID, &snapID, &projectID, &md); err != nil { + return nil, err + } + row := TerminalRow{ + Facts: AgentCommitFacts(snapID), + Request: PublishRequest{ + ArtifactType: TypeAgentCommit, Environment: env, ProjectID: projectID, + ProjectExternalKey: ProjectKeyFromID(projectID), LogicalResourceID: docID, + SourceEventKind: "agent_commit", SourceEventID: fmt.Sprintf("ev:%d", eventID), + SourceSnapshotID: snapID, Payload: []byte(md), TerminalState: "committed", + }, + } + if snapID <= 0 || md == "" { + row.MissingEvidence = true + row.Reason = "missing_commit_snapshot" + } + out = append(out, row) + } + return out, q.Err() +} + +func enumerateSealed(ctx context.Context, pool *pgxpool.Pool, env string) ([]TerminalRow, error) { + q, err := pool.Query(ctx, ` + SELECT v.doc_id, v.id, v.head_snapshot_id, v.sealed_at, COALESCE(r.project_id,0), COALESCE(b.content_markdown,'') + FROM document_versions v + LEFT JOIN document_records r ON r.doc_id = v.doc_id + LEFT JOIN document_snapshots s ON s.id = v.head_snapshot_id + LEFT JOIN document_content_blobs b ON b.content_sha256 = s.content_sha256 + WHERE v.status = 'sealed' AND v.sealed_at IS NOT NULL`) + if err != nil { + return nil, err + } + defer q.Close() + var out []TerminalRow + for q.Next() { + var docID string + var verID, snapID, projectID int64 + var sealedAt *time.Time + var md string + if err := q.Scan(&docID, &verID, &snapID, &sealedAt, &projectID, &md); err != nil { + return nil, err + } + row := TerminalRow{ + Facts: SealedFacts(sealedAt, snapID), + Request: PublishRequest{ + ArtifactType: TypeSealedVersion, Environment: env, ProjectID: projectID, + ProjectExternalKey: ProjectKeyFromID(projectID), LogicalResourceID: fmt.Sprintf("%s:v%d", docID, verID), + SourceEventKind: "version_sealed", SourceEventID: fmt.Sprintf("ver:%d", verID), + SourceSnapshotID: snapID, Payload: []byte(md), TerminalState: "sealed", + }, + } + if snapID <= 0 || md == "" { + row.MissingEvidence = true + row.Reason = "missing_sealed_head" + } + out = append(out, row) + } + return out, q.Err() +} + +func enumerateAssistant(ctx context.Context, pool *pgxpool.Pool, env string) ([]TerminalRow, error) { + q, err := pool.Query(ctx, ` + SELECT t.session_id, t.client_turn_id, t.assistant_message_id, t.completed_at, t.project_id, + COALESCE(m.content,''), COALESCE(m.export_policy,'none') + FROM chat_turns t + LEFT JOIN chat_messages m ON m.id = t.assistant_message_id + WHERE t.status = 'completed' AND t.completed_at IS NOT NULL`) + if err != nil { + return nil, err + } + defer q.Close() + var out []TerminalRow + for q.Next() { + var sessionID, clientTurnID, content, policy string + var msgID, projectID int64 + var completedAt *time.Time + if err := q.Scan(&sessionID, &clientTurnID, &msgID, &completedAt, &projectID, &content, &policy); err != nil { + return nil, err + } + logical := fmt.Sprintf("msg:%d", msgID) + row := TerminalRow{ + Facts: AssistantFacts("completed", completedAt, msgID, policy), + Request: PublishRequest{ + ArtifactType: TypeAssistantReply, Environment: env, ProjectID: projectID, + ProjectExternalKey: ProjectKeyFromID(projectID), LogicalResourceID: logical, + SourceEventKind: "assistant_completed", SourceEventID: sessionID + ":" + clientTurnID, + SourceMessageID: msgID, Payload: []byte(content), TerminalState: "completed", + }, + } + if msgID <= 0 || content == "" { + row.MissingEvidence = true + row.Reason = "missing_completed_body" + } + out = append(out, row) + } + return out, q.Err() +} diff --git a/internal/artifactvault/filesystem_store.go b/internal/artifactvault/filesystem_store.go new file mode 100644 index 00000000..9ba946e5 --- /dev/null +++ b/internal/artifactvault/filesystem_store.go @@ -0,0 +1,123 @@ +package artifactvault + +import ( + "encoding/json" + "fmt" + "os" + "path/filepath" + "strings" + + "chat-agent-system/internal/fsutil" +) + +type FilesystemStore struct { + Root string + StorageIdentity string + MasterKey []byte +} + +func NewFilesystemStore(root, identity string, master []byte) (*FilesystemStore, error) { + if strings.TrimSpace(root) == "" || strings.Contains(root, "data/dev") && !strings.Contains(root, "vault") { + if strings.HasSuffix(filepath.Clean(root), filepath.Join("data", "dev")) { + return nil, fmt.Errorf("artifactvault: refuse vault root %s (must not be runtime data/dev)", root) + } + } + if identity == "" { + return nil, fmt.Errorf("artifactvault: storage identity required") + } + if err := os.MkdirAll(root, 0o700); err != nil { + return nil, err + } + sent := filepath.Join(root, ".vault-storage-identity.json") + if b, err := os.ReadFile(sent); err == nil { + var got map[string]string + _ = json.Unmarshal(b, &got) + if got["uuid"] != "" && got["uuid"] != identity { + return nil, fmt.Errorf("artifactvault: storage identity mismatch") + } + } else { + _ = fsutil.WriteFileAtomic(sent, []byte(`{"uuid":"`+identity+`"}`), 0o444) + } + return &FilesystemStore{Root: root, StorageIdentity: identity, MasterKey: master}, nil +} + +func (s *FilesystemStore) Identity() string { return s.StorageIdentity } + +func (s *FilesystemStore) blobPath(env, projectKey, sha string) string { + return filepath.Join(s.Root, "scopes", env, projectKey, "blobs", "sha256", sha[:2], sha[2:4], sha) +} + +func (s *FilesystemStore) PutBlob(env, projectKey, sha string, plain []byte) (keyRef string, err error) { + if SHA256Hex(plain) != sha { + return "", fmt.Errorf("artifactvault: blob sha mismatch") + } + key, ref, err := ProjectKey(s.MasterKey, env, projectKey) + if err != nil { + return "", err + } + envBytes, err := Encrypt(key, plain) + if err != nil { + return "", err + } + path := s.blobPath(env, projectKey, sha) + 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 + } + return "", err + } + return ref, nil +} + +func (s *FilesystemStore) GetBlob(env, projectKey, sha string) ([]byte, error) { + key, _, err := ProjectKey(s.MasterKey, env, projectKey) + if err != nil { + return nil, err + } + raw, err := os.ReadFile(s.blobPath(env, projectKey, sha)) + if err != nil { + return nil, err + } + plain, err := Decrypt(key, raw) + if err != nil { + return nil, err + } + if SHA256Hex(plain) != sha { + return nil, fmt.Errorf("artifactvault: decrypted sha mismatch") + } + return plain, 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) +} + +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) +} + +func (s *FilesystemStore) HasCommit(artifactUUID, manifestSHA string) bool { + _, err := os.Stat(filepath.Join(s.Root, "commits", artifactUUID, manifestSHA+".json")) + return err == nil +} + +func (s *FilesystemStore) PutTombstone(artifactUUID, eventUUID string, raw []byte) error { + path := filepath.Join(s.Root, "tombstones", artifactUUID, eventUUID+".json") + return fsutil.WriteFileAtomicExclusive(path, raw, 0o644) +} + +func (s *FilesystemStore) HasTombstone(artifactUUID string) bool { + dir := filepath.Join(s.Root, "tombstones", artifactUUID) + ents, err := os.ReadDir(dir) + return err == nil && len(ents) > 0 +} + +func (s *FilesystemStore) Ready() (identityOK, connected, lagOK bool) { + _, err := os.Stat(s.Root) + return s.StorageIdentity != "", err == nil, err == nil +} diff --git a/internal/artifactvault/ingest.go b/internal/artifactvault/ingest.go new file mode 100644 index 00000000..e228675c --- /dev/null +++ b/internal/artifactvault/ingest.go @@ -0,0 +1,129 @@ +package artifactvault + +import ( + "context" + "fmt" + "strconv" + "time" +) + +// Ingest is the single write path from domain terminals into staging/outbox. +type Ingest struct { + Env string + Reg Registry +} + +func ProjectKeyFromID(projectID int64) string { + if projectID <= 0 { + return "unscoped" + } + return strconv.FormatInt(projectID, 10) +} + +func (i *Ingest) Request(ctx context.Context, facts Facts, req PublishRequest) error { + if i == nil || i.Reg == 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 := i.Reg.Request(ctx, facts, req) + return err +} + +func ApprovedFacts(docState string, approvedAt *time.Time, snapshotID int64) Facts { + return Facts{ + ArtifactType: TypeApprovedDocument, + DocState: docState, + ApprovedAt: approvedAt, + ReviewSnapshotID: snapshotID, + } +} + +func AgentCommitFacts(outputSnapshotID int64) Facts { + return Facts{ + ArtifactType: TypeAgentCommit, + CandidateStatus: "committed", + OutputSnapshotID: outputSnapshotID, + ChangeKind: "agent_commit", + ChangeToSnapshotID: outputSnapshotID, + } +} + +func SealedFacts(sealedAt *time.Time, headSnapshotID int64) Facts { + return Facts{ + ArtifactType: TypeSealedVersion, + VersionStatus: "sealed", + SealedAt: sealedAt, + HeadSnapshotID: headSnapshotID, + } +} + +func AssistantFacts(turnStatus string, completedAt *time.Time, msgID int64, policy string) Facts { + return Facts{ + ArtifactType: TypeAssistantReply, + ChatTurnStatus: turnStatus, + CompletedAt: completedAt, + AssistantMessageID: msgID, + ExportPolicy: policy, + } +} + +func FreeformFacts(explicit bool, hasSnapshot bool) Facts { + return Facts{ + ArtifactType: TypeFreeformPublish, + ExplicitPublish: explicit, + HasImmutableSnapshot: hasSnapshot, + } +} + +func (i *Ingest) AgentCommit(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, payload []byte) error { + return i.Request(ctx, 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) SealedVersion(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, sealedAt *time.Time, payload []byte) error { + return i.Request(ctx, 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) Approved(ctx context.Context, projectID int64, docID string, snapshotID int64, approvedAt *time.Time, payload []byte) error { + return i.Request(ctx, 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) FreeformPublish(ctx context.Context, projectID int64, docID, eventID string, snapshotID int64, payload []byte) error { + return i.Request(ctx, FreeformFacts(true, snapshotID > 0 || len(payload) > 0), PublishRequest{ + ArtifactType: TypeFreeformPublish, Environment: i.Env, ProjectID: projectID, + LogicalResourceID: docID, SourceEventKind: "freeform_publish", SourceEventID: eventID, + SourceSnapshotID: snapshotID, Payload: payload, TerminalState: "published", + }) +} + +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, + LogicalResourceID: logical, SourceEventKind: "assistant_completed", SourceEventID: eventID, + SourceMessageID: msgID, SourceRunID: runID, Payload: payload, TerminalState: "completed", + }) +} diff --git a/internal/artifactvault/ingest_test.go b/internal/artifactvault/ingest_test.go new file mode 100644 index 00000000..9c51e006 --- /dev/null +++ b/internal/artifactvault/ingest_test.go @@ -0,0 +1,35 @@ +package artifactvault + +import ( + "context" + "testing" + "time" +) + +func TestIngestIdempotentAndRejectsNonTerminal(t *testing.T) { + ctx := context.Background() + ing := &Ingest{Env: "dev", Reg: NewMemoryRegistry()} + now := time.Now().UTC() + payload := []byte("# committed body") + if err := ing.AgentCommit(ctx, 9, "doc_a", "ev-1", 4, payload); err != nil { + t.Fatal(err) + } + if err := ing.AgentCommit(ctx, 9, "doc_a", "ev-1", 4, payload); err != nil { + t.Fatal(err) + } + if ing.Reg.Coverage(ctx).Required != 1 { + t.Fatalf("idempotent ingest required=%d", ing.Reg.Coverage(ctx).Required) + } + if err := ing.AssistantReply(ctx, 9, "msg:1", "s:t", 1, "run", "none", &now, []byte("hello world from assistant")); err == nil { + t.Fatal("none policy must reject") + } +} + +func TestCoverageGateFailClosed(t *testing.T) { + if CoveragePASS(Coverage{}) { + t.Fatal("empty coverage must fail") + } + if err := PreflightDestroy(Coverage{Required: 2, Verified: 1}, true, true, true); err == nil { + t.Fatal("incomplete coverage must fail destroy") + } +} diff --git a/internal/artifactvault/manifest.go b/internal/artifactvault/manifest.go new file mode 100644 index 00000000..366c0d6f --- /dev/null +++ b/internal/artifactvault/manifest.go @@ -0,0 +1,100 @@ +package artifactvault + +import ( + "bytes" + "encoding/json" + "fmt" + "strings" + "time" +) + +func BuildManifest(art Artifact, payloadSHA string, size int, keyRef string) Manifest { + var snap, msg *int64 + if art.SourceSnapshotID > 0 { + v := art.SourceSnapshotID + snap = &v + } + if art.SourceMessageID > 0 { + v := art.SourceMessageID + msg = &v + } + m := Manifest{ + ArtifactSchemaVersion: SchemaMajorMinor, + ArtifactUUID: art.UUID, + ArtifactType: art.ArtifactType, + LogicalResourceID: art.LogicalResourceID, + Environment: art.Environment, + ProjectID: art.ProjectID, + ProjectExternalKey: art.ProjectExternalKey, + SourceEventID: art.SourceEventID, + SourceEventKind: art.SourceEventKind, + SourceSnapshotID: snap, + SourceMessageID: msg, + SourceRunID: art.SourceRunID, + PayloadMembers: []PayloadMember{{ + Name: "payload", MediaType: "text/markdown", ByteSize: size, SHA256: payloadSHA, Required: true, + }}, + TerminalEvidence: TerminalEvidence{ + State: art.TerminalState, DecidedAt: time.Now().UTC().Format(time.RFC3339), PolicyVersion: nz(art.PolicyVersion, PolicyVersion), + }, + RetentionClass: art.RetentionClass, + EncryptionKeyRef: keyRef, + VaultStorageIdentity: art.VaultStorageIdentity, + PublisherVersion: PublisherVersion, + PublishedAt: time.Now().UTC().Format(time.RFC3339), + } + raw, _ := json.Marshal(canonicalManifest(m)) + m.ManifestSHA256 = SHA256Hex(raw) + return m +} + +func canonicalManifest(m Manifest) Manifest { + cp := m + cp.ManifestSHA256 = "" + return cp +} + +func ValidateManifest(m Manifest) error { + if !strings.HasPrefix(m.ArtifactSchemaVersion, "1.") { + return fmt.Errorf("artifactvault: unsupported manifest major %q", m.ArtifactSchemaVersion) + } + if m.ArtifactUUID == "" || m.LogicalResourceID == "" || m.ProjectExternalKey == "" || m.VaultStorageIdentity == "" { + return fmt.Errorf("artifactvault: manifest missing required identity fields") + } + if len(m.PayloadMembers) == 0 { + return fmt.Errorf("artifactvault: manifest has no payload members") + } + for _, mem := range m.PayloadMembers { + if len(mem.SHA256) != 64 { + return fmt.Errorf("artifactvault: invalid member sha") + } + } + got := SHA256Hex(mustJSON(canonicalManifest(m))) + if m.ManifestSHA256 != "" && m.ManifestSHA256 != got { + return fmt.Errorf("artifactvault: manifest sha mismatch") + } + return nil +} + +func ValidateCommitMarker(c CommitMarker) error { + if c.ArtifactUUID == "" || len(c.ManifestSHA256) != 64 || len(c.MemberSHAs) == 0 || c.VaultStorageIdentity == "" || c.CommittedAt == "" || c.PublisherVersion == "" { + return fmt.Errorf("artifactvault: incomplete commit marker") + } + return nil +} + +func mustJSON(v any) []byte { + b, err := json.Marshal(v) + if err != nil { + return nil + } + return b +} + +func EncodeJSON(v any) []byte { + var buf bytes.Buffer + enc := json.NewEncoder(&buf) + enc.SetEscapeHTML(false) + _ = enc.Encode(v) + return bytes.TrimSpace(buf.Bytes()) +} diff --git a/internal/artifactvault/manifest.schema.json b/internal/artifactvault/manifest.schema.json new file mode 100644 index 00000000..b1e0ebd8 --- /dev/null +++ b/internal/artifactvault/manifest.schema.json @@ -0,0 +1,67 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://agent-system.local/schemas/fav-manifest-v1.json", + "title": "FAV Manifest", + "type": "object", + "additionalProperties": false, + "required": [ + "artifact_schema_version", + "artifact_uuid", + "artifact_type", + "logical_resource_id", + "environment", + "project_external_key", + "source_event_id", + "payload_members", + "terminal_evidence", + "vault_storage_identity", + "manifest_sha256" + ], + "properties": { + "artifact_schema_version": { "type": "string", "pattern": "^1\\.[0-9]+$" }, + "artifact_uuid": { "type": "string", "minLength": 8 }, + "artifact_type": { "type": "string", "enum": ["agent_commit", "sealed_version", "approved_document", "freeform_publish", "assistant_reply"] }, + "logical_resource_id": { "type": "string", "minLength": 1 }, + "environment": { "type": "string", "enum": ["dev", "test", "main"] }, + "project_id": { "type": "integer" }, + "project_external_key": { "type": "string", "minLength": 1 }, + "source_event_id": { "type": "string", "minLength": 1 }, + "source_event_kind": { "type": "string" }, + "source_snapshot_id": { "type": ["integer", "null"] }, + "source_message_id": { "type": ["integer", "null"] }, + "source_run_id": { "type": "string" }, + "payload_members": { + "type": "array", + "minItems": 1, + "items": { + "type": "object", + "required": ["name", "media_type", "byte_size", "sha256", "required"], + "properties": { + "name": { "type": "string" }, + "media_type": { "type": "string" }, + "byte_size": { "type": "integer", "minimum": 0 }, + "sha256": { "type": "string", "pattern": "^[a-f0-9]{64}$" }, + "required": { "type": "boolean" } + } + } + }, + "terminal_evidence": { + "type": "object", + "required": ["state", "decided_at", "policy_version"], + "properties": { + "state": { "type": "string" }, + "decided_at": { "type": "string" }, + "actor_ref": { "type": "string" }, + "policy_version": { "type": "string" } + } + }, + "lineage": { "type": "object" }, + "retention_class": { "type": "string" }, + "sensitivity": { "type": "string" }, + "encryption_key_ref": { "type": "string" }, + "vault_storage_identity": { "type": "string", "minLength": 8 }, + "publisher_version": { "type": "string" }, + "published_at": { "type": "string" }, + "manifest_sha256": { "type": "string", "pattern": "^[a-f0-9]{64}$" } + } +} diff --git a/internal/artifactvault/publisher.go b/internal/artifactvault/publisher.go new file mode 100644 index 00000000..92683cec --- /dev/null +++ b/internal/artifactvault/publisher.go @@ -0,0 +1,103 @@ +package artifactvault + +import ( + "context" + "encoding/json" + "fmt" + "time" +) + +type Publisher struct { + Reg Registry + Store *FilesystemStore +} + +func (p *Publisher) Drain(ctx context.Context) error { + arts, err := p.Reg.LoadRequested(ctx) + if err != nil { + return err + } + var first error + for _, a := range arts { + if err := p.PublishOne(ctx, a); err != nil && first == nil { + first = err + } + } + return first +} + +func (p *Publisher) PublishOne(ctx context.Context, art Artifact) error { + if p.Store == nil { + return fmt.Errorf("artifactvault: store required") + } + if p.Reg.IsDenied(ctx, art.UUID) { + return fmt.Errorf("artifactvault: artifact denied") + } + _ = p.Reg.MarkPublishing(ctx, art.UUID) + st, err := p.Reg.LoadStaging(ctx, art.SourceEventID) + if err != nil { + _ = p.Reg.MarkFailed(ctx, art.UUID, "staging_missing") + return err + } + if SHA256Hex(st.Payload) != art.PayloadSHA256 || SHA256Hex(st.Payload) != st.PayloadSHA256 { + _ = p.Reg.MarkFailed(ctx, art.UUID, "sha_mismatch") + return fmt.Errorf("artifactvault: staging sha mismatch") + } + if ClassifyPayload(st.Payload) != ClassificationOK { + _ = p.Reg.MarkFailed(ctx, art.UUID, "sensitive") + return ErrSensitiveRejected + } + art.VaultStorageIdentity = p.Store.Identity() + keyRef, err := p.Store.PutBlob(art.Environment, art.ProjectExternalKey, art.PayloadSHA256, st.Payload) + if err != nil { + _ = p.Reg.MarkFailed(ctx, art.UUID, "blob") + return err + } + man := BuildManifest(art, art.PayloadSHA256, len(st.Payload), keyRef) + if err := ValidateManifest(man); err != nil { + _ = p.Reg.MarkFailed(ctx, art.UUID, "manifest") + return err + } + manRaw := EncodeJSON(man) + if err := p.Store.PutManifest(art.UUID, man.ManifestSHA256, manRaw); err != nil { + _ = p.Reg.MarkFailed(ctx, art.UUID, "manifest_write") + return err + } + marker := CommitMarker{ + ArtifactUUID: art.UUID, ManifestSHA256: man.ManifestSHA256, + MemberSHAs: []string{art.PayloadSHA256}, VaultStorageIdentity: p.Store.Identity(), + CommittedAt: time.Now().UTC().Format(time.RFC3339), PublisherVersion: PublisherVersion, + } + if err := ValidateCommitMarker(marker); err != nil { + return err + } + if err := p.Store.PutCommit(art.UUID, man.ManifestSHA256, EncodeJSON(marker)); err != nil { + _ = p.Reg.MarkFailed(ctx, art.UUID, "commit") + return err + } + _ = p.Reg.RecordReference(ctx, art.UUID, art.ProjectExternalKey, art.PayloadSHA256, keyRef) + return p.Reg.MarkVerified(ctx, art.UUID, man.ManifestSHA256) +} + +func (p *Publisher) DegradedPayload(ctx context.Context, env, logical, projectKey string) ([]byte, string, error) { + art, err := p.Reg.GetByLogical(ctx, env, logical) + if err != nil { + return nil, "", err + } + 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 != "" { + return nil, "", fmt.Errorf("artifactvault: unpublished") + } + b, err := p.Store.GetBlob(env, projectKey, art.PayloadSHA256) + if err != nil { + return nil, "", err + } + return b, art.PayloadSHA256, nil +} + +func MustJSON(v any) string { + b, _ := json.Marshal(v) + return string(b) +} diff --git a/internal/artifactvault/publisher_test.go b/internal/artifactvault/publisher_test.go new file mode 100644 index 00000000..2e05a3da --- /dev/null +++ b/internal/artifactvault/publisher_test.go @@ -0,0 +1,150 @@ +package artifactvault + +import ( + "context" + "os" + "path/filepath" + "testing" + "time" +) + +func testStore(t *testing.T) *FilesystemStore { + t.Helper() + dir := t.TempDir() + st, err := NewFilesystemStore(dir, "vault-id-test", []byte("0123456789abcdef0123456789abcdef")) + if err != nil { + t.Fatal(err) + } + return st +} + +func TestPublisherIdempotentAndCrashSafe(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("# final\n"), + } + req := PublishRequest{ + ArtifactType: TypeAgentCommit, Environment: "dev", ProjectExternalKey: "proj-a", + LogicalResourceID: "doc_a", SourceEventKind: "agent_commit", SourceEventID: "ev-1", + Payload: []byte("# final\n"), TerminalState: "committed", + } + art1, _, err := reg.Request(ctx, facts, req) + if err != nil { + t.Fatal(err) + } + art2, _, err := reg.Request(ctx, facts, req) + if err != nil || art1.UUID != art2.UUID { + t.Fatalf("idempotent request: %v %s %s", err, art1.UUID, art2.UUID) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + cov := reg.Coverage(ctx) + if cov.Verified != 1 { + t.Fatalf("verified=%d", cov.Verified) + } + b, sha, err := pub.DegradedPayload(ctx, "dev", "doc_a", "proj-a") + if err != nil || string(b) != "# final\n" || sha == "" { + t.Fatalf("degraded: %v %q", err, b) + } +} + +func TestPublisherRejectsBadSHAAndCrossProjectKeys(t *testing.T) { + ctx := context.Background() + reg := NewMemoryRegistry() + store := testStore(t) + pub := &Publisher{Reg: reg, Store: store} + facts := Facts{ + ArtifactType: TypeSealedVersion, VersionStatus: "sealed", SealedAt: ptrTime(), + HeadSnapshotID: 2, Classification: ClassificationOK, Payload: []byte("same-bytes"), + } + aReq := PublishRequest{ArtifactType: TypeSealedVersion, Environment: "dev", ProjectExternalKey: "p1", LogicalResourceID: "d1", SourceEventKind: "seal", SourceEventID: "s1", Payload: []byte("same-bytes"), TerminalState: "sealed"} + bReq := aReq + bReq.ProjectExternalKey = "p2" + bReq.LogicalResourceID = "d2" + bReq.SourceEventID = "s2" + if _, _, err := reg.Request(ctx, facts, aReq); err != nil { + t.Fatal(err) + } + if _, _, err := reg.Request(ctx, facts, bReq); err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + if reg.ActiveRefs(ctx, "p1", SHA256Hex([]byte("same-bytes"))) != 1 { + t.Fatal("p1 refs") + } + if reg.ActiveRefs(ctx, "p2", SHA256Hex([]byte("same-bytes"))) != 1 { + t.Fatal("p2 refs") + } + _ = reg.Tombstone(ctx, "unused", "test", "test") +} + +func TestFilesystemRefusesRuntimeDataDev(t *testing.T) { + root := filepath.Join(t.TempDir(), "data", "dev") + _ = os.MkdirAll(root, 0o755) + if _, err := NewFilesystemStore(root, "id", []byte("0123456789abcdef0123456789abcdef")); err == nil { + t.Fatal("must refuse data/dev") + } +} + +func TestTombstoneUnionDeny(t *testing.T) { + ctx := context.Background() + reg := NewMemoryRegistry() + store := testStore(t) + pub := &Publisher{Reg: reg, Store: store} + now := time.Now().UTC() + facts := Facts{ArtifactType: TypeApprovedDocument, DocState: "approved", ApprovedAt: &now, ReviewSnapshotID: 8, Classification: ClassificationOK, Payload: []byte("ok body")} + req := PublishRequest{ArtifactType: TypeApprovedDocument, Environment: "dev", ProjectExternalKey: "p", LogicalResourceID: "doc", SourceEventKind: "approved", SourceEventID: "ap1", Payload: []byte("ok body"), TerminalState: "approved"} + art, _, err := reg.Request(ctx, facts, req) + if err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + _ = reg.Tombstone(ctx, art.UUID, "legal", "admin") + if _, _, err := pub.DegradedPayload(ctx, "dev", "doc", "p"); err == nil { + t.Fatal("tombstone must deny read") + } +} + +func TestLastReferencePurgeAndLegalHold(t *testing.T) { + ctx := context.Background() + reg := NewMemoryRegistry() + store := testStore(t) + pub := &Publisher{Reg: reg, Store: store} + now := time.Now().UTC() + payload := []byte("shared-bytes") + facts := Facts{ArtifactType: TypeSealedVersion, VersionStatus: "sealed", SealedAt: &now, HeadSnapshotID: 2, Classification: ClassificationOK, Payload: payload} + req := PublishRequest{ArtifactType: TypeSealedVersion, Environment: "dev", ProjectExternalKey: "p1", LogicalResourceID: "d1", SourceEventKind: "seal", SourceEventID: "s1", Payload: payload, TerminalState: "sealed"} + if _, _, err := reg.Request(ctx, facts, req); err != nil { + t.Fatal(err) + } + if err := pub.Drain(ctx); err != nil { + t.Fatal(err) + } + if reg.ActiveRefs(ctx, "p1", SHA256Hex(payload)) != 1 { + t.Fatal("expected one active ref") + } + if err := CanPurge(false, 1); err == nil { + t.Fatal("must not purge while refs remain") + } + if err := CanPurge(true, 0); err == nil { + t.Fatal("legal hold blocks purge") + } + if err := CanPurge(false, 0); err != nil { + t.Fatal(err) + } +} + +func ptrTime() *time.Time { t := time.Now().UTC(); return &t } diff --git a/internal/artifactvault/recovery.go b/internal/artifactvault/recovery.go new file mode 100644 index 00000000..bec72832 --- /dev/null +++ b/internal/artifactvault/recovery.go @@ -0,0 +1,67 @@ +package artifactvault + +import ( + "fmt" + "strings" +) + +type RecoverMode string + +const ( + RecoverAsCollection RecoverMode = "recover-as-collection" + RecoverRelink RecoverMode = "relink" +) + +type RecoverPlan struct { + Mode RecoverMode + TargetIdentity string + SourceIdentity string + PlanHash string + Conflicts []string + Actor string + HasAdmin bool + HasProjectAuth bool + DependenciesOK bool + ScratchOnly bool +} + +func ValidateRecoverTarget(dsn, identity string) error { + low := strings.ToLower(dsn + " " + identity) + for _, ban := range []string{"agent_system_dev", "agent_system_test", "agent_system_main", "env=dev", "env=test", "env=main"} { + if strings.Contains(low, ban) && !strings.Contains(low, "scratch") { + return fmt.Errorf("artifactvault: refuse recover into operational database") + } + } + if !strings.Contains(low, "scratch") { + return fmt.Errorf("artifactvault: recover target must be scratch") + } + return nil +} + +func EvaluateRecoverPlan(p RecoverPlan, denied bool) error { + if err := ValidateRecoverTarget(p.TargetIdentity, p.SourceIdentity); err != nil { + return err + } + if !p.ScratchOnly { + return fmt.Errorf("artifactvault: scratch-only required") + } + if !p.HasAdmin { + return fmt.Errorf("artifactvault: operator lacks role-admin") + } + if denied { + return fmt.Errorf("artifactvault: union-deny deletion ledger") + } + if len(p.Conflicts) > 0 && p.Mode == RecoverRelink { + return fmt.Errorf("artifactvault: relink cannot proceed with conflicts") + } + if p.Mode == RecoverRelink && (!p.HasProjectAuth || !p.DependenciesOK) { + return fmt.Errorf("artifactvault: relink authorization or dependency gap") + } + if p.Mode == "" { + p.Mode = RecoverAsCollection + } + if p.PlanHash == "" { + return fmt.Errorf("artifactvault: plan hash required") + } + return nil +} diff --git a/internal/artifactvault/recovery_test.go b/internal/artifactvault/recovery_test.go new file mode 100644 index 00000000..c4152ba3 --- /dev/null +++ b/internal/artifactvault/recovery_test.go @@ -0,0 +1,27 @@ +package artifactvault + +import "testing" + +func TestRecoverRejectsOperationalDSN(t *testing.T) { + if err := ValidateRecoverTarget("postgres://x/agent_system_dev", "dev"); err == nil { + t.Fatal("dev dsn") + } + if err := ValidateRecoverTarget("postgres://x/fav_scratch", "scratch-1"); err != nil { + t.Fatal(err) + } +} + +func TestRecoverPlanGates(t *testing.T) { + p := RecoverPlan{Mode: RecoverAsCollection, TargetIdentity: "scratch-db", SourceIdentity: "vault", PlanHash: "abc", HasAdmin: true, ScratchOnly: true} + if err := EvaluateRecoverPlan(p, false); err != nil { + t.Fatal(err) + } + if err := EvaluateRecoverPlan(p, true); err == nil { + t.Fatal("denied") + } + p.Mode = RecoverRelink + p.Conflicts = []string{"uuid_collision"} + if err := EvaluateRecoverPlan(p, false); err == nil { + t.Fatal("relink+conflict") + } +} diff --git a/internal/artifactvault/registry.go b/internal/artifactvault/registry.go new file mode 100644 index 00000000..37ee6193 --- /dev/null +++ b/internal/artifactvault/registry.go @@ -0,0 +1,467 @@ +package artifactvault + +import ( + "context" + "fmt" + "sync" + "time" + + "github.com/jackc/pgx/v5" + "github.com/jackc/pgx/v5/pgxpool" +) + +type Registry interface { + Request(ctx context.Context, facts Facts, req PublishRequest) (*Artifact, *Staging, error) + LoadRequested(ctx context.Context) ([]Artifact, error) + LoadStaging(ctx context.Context, sourceEventID string) (*Staging, error) + MarkPublishing(ctx context.Context, uuid string) error + MarkVerified(ctx context.Context, uuid, manifestSHA string) error + MarkFailed(ctx context.Context, uuid, code string) error + GetByLogical(ctx context.Context, env, logical string) (*Artifact, error) + Tombstone(ctx context.Context, uuid, reason, authority string) error + IsDenied(ctx context.Context, uuid string) bool + RecordReference(ctx context.Context, uuid, projectKey, sha, keyRef string) error + ActiveRefs(ctx context.Context, projectKey, sha string) int + Quarantine(ctx context.Context, reason string, logical string) error + QuarantineCount(ctx context.Context) int + Coverage(ctx context.Context) Coverage +} + +type Coverage struct { + Required int + Verified int + Quarantine int + Excluded int +} + +type MemoryRegistry struct { + mu sync.Mutex + arts map[string]*Artifact + byKey map[string]string + staging map[string]*Staging + denied map[string]bool + refs map[string]int + quarantine int + required int +} + +func NewMemoryRegistry() *MemoryRegistry { + return &MemoryRegistry{ + arts: map[string]*Artifact{}, byKey: map[string]string{}, staging: map[string]*Staging{}, + denied: map[string]bool{}, refs: map[string]int{}, + } +} + +func (m *MemoryRegistry) Request(_ context.Context, facts Facts, req PublishRequest) (*Artifact, *Staging, error) { + if err := EvaluateEligibility(facts); err != nil { + return nil, nil, err + } + if err := ValidateEnv(req.Environment); err != nil { + return nil, nil, err + } + sha := SHA256Hex(req.Payload) + key := IdempotencyKey(req, sha) + m.mu.Lock() + defer m.mu.Unlock() + if id, ok := m.byKey[key]; ok { + return m.arts[id], m.staging[req.SourceEventID], nil + } + uuid := "art-" + sha[:16] + "-" + fmt.Sprintf("%d", len(m.arts)+1) + art := &Artifact{ + UUID: uuid, IdempotencyKey: key, ArtifactType: req.ArtifactType, + LogicalResourceID: req.LogicalResourceID, Environment: req.Environment, + ProjectID: req.ProjectID, ProjectExternalKey: req.ProjectExternalKey, + SourceEventID: req.SourceEventID, SourceEventKind: req.SourceEventKind, + SourceSnapshotID: req.SourceSnapshotID, SourceMessageID: req.SourceMessageID, + SourceRunID: req.SourceRunID, PayloadSHA256: sha, TerminalState: req.TerminalState, + PolicyVersion: nz(req.PolicyVersion, PolicyVersion), PublicationStatus: StatusRequested, + RetentionClass: nz(req.RetentionClass, "governance"), RequestedAt: time.Now().UTC(), + } + st := &Staging{ + UUID: "stg-" + uuid, ProjectID: req.ProjectID, SourceEventID: req.SourceEventID, + PayloadSHA256: sha, Payload: append([]byte(nil), req.Payload...), Classification: ClassificationOK, + CreatedAt: time.Now().UTC(), + } + m.arts[uuid] = art + m.byKey[key] = uuid + m.staging[req.SourceEventID] = st + m.required++ + return art, st, nil +} + +func (m *MemoryRegistry) LoadRequested(_ context.Context) ([]Artifact, error) { + m.mu.Lock() + defer m.mu.Unlock() + out := make([]Artifact, 0) + for _, a := range m.arts { + if a.PublicationStatus == StatusRequested || a.PublicationStatus == StatusRetryableFailed { + out = append(out, *a) + } + } + return out, nil +} + +func (m *MemoryRegistry) LoadStaging(_ context.Context, sourceEventID string) (*Staging, error) { + m.mu.Lock() + defer m.mu.Unlock() + st := m.staging[sourceEventID] + if st == nil { + return nil, fmt.Errorf("artifactvault: staging missing") + } + cp := *st + cp.Payload = append([]byte(nil), st.Payload...) + return &cp, nil +} + +func (m *MemoryRegistry) MarkPublishing(_ context.Context, uuid string) error { + m.mu.Lock() + defer m.mu.Unlock() + if a := m.arts[uuid]; a != nil { + a.PublicationStatus = StatusPublishing + } + return nil +} + +func (m *MemoryRegistry) MarkVerified(_ context.Context, uuid, manifestSHA string) error { + m.mu.Lock() + defer m.mu.Unlock() + if a := m.arts[uuid]; a != nil { + now := time.Now().UTC() + a.PublicationStatus = StatusVerified + a.ManifestSHA256 = manifestSHA + a.VerifiedAt = &now + } + return nil +} + +func (m *MemoryRegistry) MarkFailed(_ context.Context, uuid, _ string) error { + m.mu.Lock() + defer m.mu.Unlock() + if a := m.arts[uuid]; a != nil { + a.PublicationStatus = StatusRetryableFailed + } + return nil +} + +func (m *MemoryRegistry) GetByLogical(_ context.Context, env, logical string) (*Artifact, error) { + m.mu.Lock() + defer m.mu.Unlock() + for _, a := range m.arts { + if a.Environment == env && a.LogicalResourceID == logical && a.PublicationStatus == StatusVerified && !m.denied[a.UUID] { + cp := *a + return &cp, nil + } + } + return nil, fmt.Errorf("artifactvault: not found") +} + +func (m *MemoryRegistry) Tombstone(_ context.Context, uuid, _, _ string) error { + m.mu.Lock() + defer m.mu.Unlock() + m.denied[uuid] = true + if a := m.arts[uuid]; a != nil { + now := time.Now().UTC() + a.PublicationStatus = StatusTombstoned + a.TombstonedAt = &now + } + return nil +} + +func (m *MemoryRegistry) IsDenied(_ context.Context, uuid string) bool { + m.mu.Lock() + defer m.mu.Unlock() + return m.denied[uuid] +} + +func (m *MemoryRegistry) RecordReference(_ context.Context, _, projectKey, sha, _ string) error { + m.mu.Lock() + defer m.mu.Unlock() + m.refs[projectKey+"|"+sha]++ + return nil +} + +func (m *MemoryRegistry) ActiveRefs(_ context.Context, projectKey, sha string) int { + m.mu.Lock() + defer m.mu.Unlock() + return m.refs[projectKey+"|"+sha] +} + +func (m *MemoryRegistry) Quarantine(_ context.Context, _, _ string) error { + m.mu.Lock() + defer m.mu.Unlock() + m.quarantine++ + return nil +} + +func (m *MemoryRegistry) QuarantineCount(_ context.Context) int { + m.mu.Lock() + defer m.mu.Unlock() + return m.quarantine +} + +func (m *MemoryRegistry) Coverage(_ context.Context) Coverage { + m.mu.Lock() + defer m.mu.Unlock() + v := 0 + for _, a := range m.arts { + if a.PublicationStatus == StatusVerified { + v++ + } + } + return Coverage{Required: m.required, Verified: v, Quarantine: m.quarantine} +} + +// PGRegistry persists FAV rows. Request uses a dedicated transaction unless WithTx is used. +type PGRegistry struct { + pool *pgxpool.Pool +} + +func NewPGRegistry(pool *pgxpool.Pool) *PGRegistry { return &PGRegistry{pool: pool} } + +func (p *PGRegistry) Request(ctx context.Context, facts Facts, req PublishRequest) (*Artifact, *Staging, error) { + if p == nil || p.pool == nil { + return nil, nil, fmt.Errorf("artifactvault: no database") + } + tx, err := p.pool.BeginTx(ctx, pgx.TxOptions{}) + if err != nil { + return nil, nil, err + } + defer func() { _ = tx.Rollback(ctx) }() + art, st, err := requestTX(ctx, tx, facts, req) + if err != nil { + return nil, nil, err + } + if err := tx.Commit(ctx); err != nil { + return nil, nil, err + } + return art, st, nil +} + +func requestTX(ctx context.Context, tx pgx.Tx, facts Facts, req PublishRequest) (*Artifact, *Staging, error) { + if err := EvaluateEligibility(facts); err != nil { + return nil, nil, err + } + sha := SHA256Hex(req.Payload) + key := IdempotencyKey(req, sha) + var existing string + err := tx.QueryRow(ctx, `SELECT artifact_uuid FROM final_artifacts WHERE idempotency_key=$1`, key).Scan(&existing) + if err == nil && existing != "" { + st := &Staging{SourceEventID: req.SourceEventID, PayloadSHA256: sha, Payload: req.Payload} + return &Artifact{UUID: existing, IdempotencyKey: key, PublicationStatus: StatusRequested}, st, nil + } + uuid := "art-" + sha[:16] + _, err = tx.Exec(ctx, ` + INSERT INTO artifact_payload_staging (staging_uuid, project_id, source_event_id, payload_sha256, encrypted_payload, classification) + VALUES ($1,$2,$3,$4,$5,$6) + ON CONFLICT (source_event_id, payload_sha256) DO NOTHING + `, "stg-"+uuid, req.ProjectID, req.SourceEventID, sha, req.Payload, ClassificationOK) + if err != nil { + return nil, nil, err + } + _, err = tx.Exec(ctx, ` + INSERT INTO final_artifacts ( + artifact_uuid, idempotency_key, artifact_type, logical_resource_id, environment, project_id, + project_external_key, source_event_id, source_event_kind, source_snapshot_id, source_message_id, + source_run_id, payload_sha256, terminal_state, policy_version, publication_status, retention_class + ) VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11,$12,$13,$14,$15,$16,$17) + ON CONFLICT (idempotency_key) DO NOTHING + `, uuid, key, req.ArtifactType, req.LogicalResourceID, req.Environment, req.ProjectID, + req.ProjectExternalKey, req.SourceEventID, req.SourceEventKind, nullInt(req.SourceSnapshotID), + nullInt(req.SourceMessageID), req.SourceRunID, sha, req.TerminalState, nz(req.PolicyVersion, PolicyVersion), + StatusRequested, nz(req.RetentionClass, "governance")) + if err != nil { + return nil, nil, err + } + return &Artifact{UUID: uuid, IdempotencyKey: key, ArtifactType: req.ArtifactType, LogicalResourceID: req.LogicalResourceID, + Environment: req.Environment, ProjectExternalKey: req.ProjectExternalKey, SourceEventID: req.SourceEventID, + PayloadSHA256: sha, PublicationStatus: StatusRequested, TerminalState: req.TerminalState}, &Staging{Payload: req.Payload, PayloadSHA256: sha, SourceEventID: req.SourceEventID}, nil +} + +func nullInt(v int64) any { + if v == 0 { + return nil + } + return v +} + +func (p *PGRegistry) LoadRequested(ctx context.Context) ([]Artifact, error) { + if p == nil || p.pool == nil { + return nil, fmt.Errorf("artifactvault: no database") + } + rows, err := p.pool.Query(ctx, ` + SELECT artifact_uuid, idempotency_key, artifact_type, logical_resource_id, environment, + COALESCE(project_id,0), project_external_key, source_event_id, source_event_kind, + COALESCE(source_snapshot_id,0), COALESCE(source_message_id,0), source_run_id, + payload_sha256, COALESCE(manifest_sha256,''), COALESCE(vault_storage_identity,''), + terminal_state, policy_version, publication_status, retention_class, requested_at + FROM final_artifacts + WHERE publication_status IN ('requested','retryable_failed') + ORDER BY requested_at ASC`) + if err != nil { + return nil, err + } + defer rows.Close() + out := make([]Artifact, 0) + for rows.Next() { + var a Artifact + if err := rows.Scan(&a.UUID, &a.IdempotencyKey, &a.ArtifactType, &a.LogicalResourceID, &a.Environment, + &a.ProjectID, &a.ProjectExternalKey, &a.SourceEventID, &a.SourceEventKind, + &a.SourceSnapshotID, &a.SourceMessageID, &a.SourceRunID, &a.PayloadSHA256, &a.ManifestSHA256, + &a.VaultStorageIdentity, &a.TerminalState, &a.PolicyVersion, &a.PublicationStatus, &a.RetentionClass, &a.RequestedAt); err != nil { + return nil, err + } + out = append(out, a) + } + return out, rows.Err() +} + +func (p *PGRegistry) LoadStaging(ctx context.Context, sourceEventID string) (*Staging, error) { + if p == nil || p.pool == nil { + return nil, fmt.Errorf("artifactvault: no database") + } + var st Staging + err := p.pool.QueryRow(ctx, ` + SELECT staging_uuid, COALESCE(project_id,0), source_event_id, payload_sha256, encrypted_payload, classification, created_at + FROM artifact_payload_staging WHERE source_event_id=$1 ORDER BY created_at DESC LIMIT 1`, sourceEventID). + Scan(&st.UUID, &st.ProjectID, &st.SourceEventID, &st.PayloadSHA256, &st.Payload, &st.Classification, &st.CreatedAt) + if err != nil { + return nil, err + } + return &st, nil +} + +func (p *PGRegistry) MarkPublishing(ctx context.Context, uuid string) error { + if p == nil || p.pool == nil { + return nil + } + _, err := p.pool.Exec(ctx, `UPDATE final_artifacts SET publication_status=$2 WHERE artifact_uuid=$1`, uuid, StatusPublishing) + if err != nil { + return err + } + _, err = p.pool.Exec(ctx, ` + INSERT INTO artifact_publication_attempts (artifact_uuid, idempotency_key, attempt_no) + SELECT artifact_uuid, idempotency_key, COALESCE((SELECT MAX(attempt_no) FROM artifact_publication_attempts a WHERE a.artifact_uuid=$1),0)+1 + FROM final_artifacts WHERE artifact_uuid=$1`, uuid) + return err +} + +func (p *PGRegistry) MarkVerified(ctx context.Context, uuid, manifestSHA string) error { + if p == nil || p.pool == nil { + return nil + } + _, err := p.pool.Exec(ctx, `UPDATE final_artifacts SET publication_status=$2, manifest_sha256=$3, verified_at=now() WHERE artifact_uuid=$1`, uuid, StatusVerified, manifestSHA) + return err +} + +func (p *PGRegistry) MarkFailed(ctx context.Context, uuid, code string) error { + if p == nil || p.pool == nil { + return nil + } + _, err := p.pool.Exec(ctx, `UPDATE final_artifacts SET publication_status=$2 WHERE artifact_uuid=$1`, uuid, StatusRetryableFailed) + if err != nil { + return err + } + _, _ = p.pool.Exec(ctx, ` + UPDATE artifact_publication_attempts SET error_code=$2 + WHERE artifact_uuid=$1 AND attempt_no=(SELECT MAX(attempt_no) FROM artifact_publication_attempts WHERE artifact_uuid=$1)`, uuid, code) + return nil +} + +func (p *PGRegistry) GetByLogical(ctx context.Context, env, logical string) (*Artifact, error) { + if p == nil || p.pool == nil { + return nil, fmt.Errorf("artifactvault: no database") + } + var a Artifact + err := p.pool.QueryRow(ctx, ` + SELECT artifact_uuid, idempotency_key, artifact_type, logical_resource_id, environment, + COALESCE(project_id,0), project_external_key, source_event_id, source_event_kind, + COALESCE(source_snapshot_id,0), COALESCE(source_message_id,0), source_run_id, + payload_sha256, COALESCE(manifest_sha256,''), COALESCE(vault_storage_identity,''), + terminal_state, policy_version, publication_status + FROM final_artifacts + WHERE environment=$1 AND logical_resource_id=$2 AND publication_status='verified' + ORDER BY verified_at DESC NULLS LAST LIMIT 1`, env, logical). + Scan(&a.UUID, &a.IdempotencyKey, &a.ArtifactType, &a.LogicalResourceID, &a.Environment, + &a.ProjectID, &a.ProjectExternalKey, &a.SourceEventID, &a.SourceEventKind, + &a.SourceSnapshotID, &a.SourceMessageID, &a.SourceRunID, &a.PayloadSHA256, &a.ManifestSHA256, + &a.VaultStorageIdentity, &a.TerminalState, &a.PolicyVersion, &a.PublicationStatus) + if err != nil { + return nil, err + } + return &a, nil +} + +func (p *PGRegistry) Tombstone(ctx context.Context, uuid, reason, authority string) error { + if p == nil || p.pool == nil { + return nil + } + _, err := p.pool.Exec(ctx, `UPDATE final_artifacts SET publication_status=$2, tombstoned_at=now() WHERE artifact_uuid=$1`, uuid, StatusTombstoned) + if err != nil { + return err + } + _, err = p.pool.Exec(ctx, `INSERT INTO artifact_deletion_ledger (artifact_uuid, reason, authority, effective_at) VALUES ($1,$2,$3,now())`, uuid, reason, authority) + return err +} + +func (p *PGRegistry) IsDenied(ctx context.Context, uuid string) bool { + if p == nil || p.pool == nil { + return false + } + var n int + _ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_deletion_ledger WHERE artifact_uuid=$1`, uuid).Scan(&n) + return n > 0 +} + +func (p *PGRegistry) RecordReference(ctx context.Context, uuid, projectKey, sha, keyRef string) error { + if p == nil || p.pool == nil { + return nil + } + _, err := p.pool.Exec(ctx, ` + INSERT INTO artifact_blob_references (artifact_uuid, project_scope, payload_sha256, envelope_key_ref, reference_status) + VALUES ($1,$2,$3,$4,'active') + ON CONFLICT (artifact_uuid, payload_sha256) DO NOTHING`, uuid, projectKey, sha, keyRef) + return err +} + +func (p *PGRegistry) ActiveRefs(ctx context.Context, projectKey, sha string) int { + if p == nil || p.pool == nil { + return 0 + } + var n int + _ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_blob_references WHERE project_scope=$1 AND payload_sha256=$2 AND reference_status='active'`, projectKey, sha).Scan(&n) + return n +} + +func (p *PGRegistry) Quarantine(ctx context.Context, reason, logical string) error { + if p == nil || p.pool == nil { + return nil + } + _, err := p.pool.Exec(ctx, `INSERT INTO artifact_quarantine (logical_resource_id, reason) VALUES ($1,$2)`, logical, reason) + return err +} + +func (p *PGRegistry) QuarantineCount(ctx context.Context) int { + if p == nil || p.pool == nil { + return 0 + } + var n int + _ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_quarantine`).Scan(&n) + return n +} + +func (p *PGRegistry) Coverage(ctx context.Context) Coverage { + var c Coverage + if p == nil || p.pool == nil { + return c + } + _ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM final_artifacts`).Scan(&c.Required) + _ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM final_artifacts WHERE publication_status='verified'`).Scan(&c.Verified) + _ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_quarantine`).Scan(&c.Quarantine) + _ = p.pool.QueryRow(ctx, `SELECT COUNT(*) FROM artifact_backfill_exclusions WHERE expires_at > now()`).Scan(&c.Excluded) + return c +} + +// RequestWithTx writes staging + unpublished outbox inside an existing domain transaction. +func RequestWithTx(ctx context.Context, tx pgx.Tx, facts Facts, req PublishRequest) (*Artifact, *Staging, error) { + return requestTX(ctx, tx, facts, req) +} diff --git a/internal/artifactvault/tombstone.schema.json b/internal/artifactvault/tombstone.schema.json new file mode 100644 index 00000000..316d9a48 --- /dev/null +++ b/internal/artifactvault/tombstone.schema.json @@ -0,0 +1,16 @@ +{ + "$schema": "https://json-schema.org/draft/2020-12/schema", + "$id": "https://agent-system.local/schemas/fav-tombstone-v1.json", + "title": "FAV Tombstone", + "type": "object", + "additionalProperties": false, + "required": ["artifact_uuid", "event_uuid", "reason", "effective_at"], + "properties": { + "artifact_uuid": { "type": "string" }, + "event_uuid": { "type": "string" }, + "reason": { "type": "string" }, + "authority": { "type": "string" }, + "effective_at": { "type": "string" }, + "audit_ref": { "type": "string" } + } +} diff --git a/internal/artifactvault/types.go b/internal/artifactvault/types.go new file mode 100644 index 00000000..1ee6decd --- /dev/null +++ b/internal/artifactvault/types.go @@ -0,0 +1,208 @@ +package artifactvault + +import ( + "crypto/sha256" + "encoding/hex" + "fmt" + "strings" + "time" +) + +const ( + SchemaMajorMinor = "1.0" + PublisherVersion = "fav-publisher/1.0" + PolicyVersion = "fav-eligibility/1.0" + ClassificationOK = "ok" + ClassificationSecret = "secret" + ClassificationRestricted = "restricted" + ClassificationFailed = "classification_failed" + + TypeAgentCommit = "agent_commit" + TypeSealedVersion = "sealed_version" + TypeApprovedDocument = "approved_document" + TypeFreeformPublish = "freeform_publish" + TypeAssistantReply = "assistant_reply" + + StatusRequested = "requested" + StatusPublishing = "publishing" + StatusVerified = "verified" + StatusRetryableFailed = "retryable_failed" + StatusTombstoned = "tombstoned" + StatusPurged = "purged" + StatusQuarantine = "quarantine" + + RefActive = "active" + RefTombstoned = "tombstoned" + RefPurged = "purged" +) + +// Facts are the authoritative database facts used for eligibility. HTTP handlers must not invent them. +type Facts struct { + ArtifactType string + CandidateStatus string + OutputSnapshotID int64 + ChangeKind string + ChangeToSnapshotID int64 + VersionStatus string + SealedAt *time.Time + HeadSnapshotID int64 + DocState string + ApprovedAt *time.Time + ReviewSnapshotID int64 + HasImmutableSnapshot bool + ExplicitPublish bool + ChatTurnStatus string + CompletedAt *time.Time + AssistantMessageID int64 + ExportPolicy string + ExplicitPin bool + PinAfterCompleted bool + Payload []byte + Classification string +} + +type PublishRequest struct { + ArtifactType string + Environment string + ProjectID int64 + ProjectExternalKey string + LogicalResourceID string + SourceEventKind string + SourceEventID string + SourceSnapshotID int64 + SourceMessageID int64 + SourceRunID string + Payload []byte + TerminalState string + PolicyVersion string + Classification string + Sensitivity string + RetentionClass string +} + +type Artifact struct { + UUID string + IdempotencyKey string + ArtifactType string + LogicalResourceID string + Environment string + ProjectID int64 + ProjectExternalKey string + SourceEventID string + SourceEventKind string + SourceSnapshotID int64 + SourceMessageID int64 + SourceRunID string + PayloadSHA256 string + ManifestSHA256 string + VaultStorageIdentity string + TerminalState string + PolicyVersion string + PublicationStatus string + RetentionClass string + RequestedAt time.Time + VerifiedAt *time.Time + TombstonedAt *time.Time +} + +type Staging struct { + UUID string + ProjectID int64 + SourceEventID string + PayloadSHA256 string + Payload []byte + Classification string + CreatedAt time.Time +} + +type Manifest struct { + ArtifactSchemaVersion string `json:"artifact_schema_version"` + ArtifactUUID string `json:"artifact_uuid"` + ArtifactType string `json:"artifact_type"` + LogicalResourceID string `json:"logical_resource_id"` + Environment string `json:"environment"` + ProjectID int64 `json:"project_id,omitempty"` + ProjectExternalKey string `json:"project_external_key"` + SourceEventID string `json:"source_event_id"` + SourceEventKind string `json:"source_event_kind,omitempty"` + SourceSnapshotID *int64 `json:"source_snapshot_id"` + SourceMessageID *int64 `json:"source_message_id"` + SourceRunID string `json:"source_run_id,omitempty"` + PayloadMembers []PayloadMember `json:"payload_members"` + TerminalEvidence TerminalEvidence `json:"terminal_evidence"` + Lineage map[string]any `json:"lineage,omitempty"` + RetentionClass string `json:"retention_class,omitempty"` + Sensitivity string `json:"sensitivity,omitempty"` + EncryptionKeyRef string `json:"encryption_key_ref,omitempty"` + VaultStorageIdentity string `json:"vault_storage_identity"` + PublisherVersion string `json:"publisher_version,omitempty"` + PublishedAt string `json:"published_at,omitempty"` + ManifestSHA256 string `json:"manifest_sha256"` +} + +type PayloadMember struct { + Name string `json:"name"` + MediaType string `json:"media_type"` + ByteSize int `json:"byte_size"` + SHA256 string `json:"sha256"` + Required bool `json:"required"` +} + +type TerminalEvidence struct { + State string `json:"state"` + DecidedAt string `json:"decided_at"` + ActorRef string `json:"actor_ref,omitempty"` + PolicyVersion string `json:"policy_version"` +} + +type CommitMarker struct { + ArtifactUUID string `json:"artifact_uuid"` + ManifestSHA256 string `json:"manifest_sha256"` + MemberSHAs []string `json:"member_shas"` + VaultStorageIdentity string `json:"vault_storage_identity"` + CommittedAt string `json:"committed_at"` + PublisherVersion string `json:"publisher_version"` +} + +type Tombstone struct { + ArtifactUUID string `json:"artifact_uuid"` + EventUUID string `json:"event_uuid"` + Reason string `json:"reason"` + Authority string `json:"authority,omitempty"` + EffectiveAt string `json:"effective_at"` + AuditRef string `json:"audit_ref,omitempty"` +} + +func SHA256Hex(b []byte) string { + sum := sha256.Sum256(b) + return hex.EncodeToString(sum[:]) +} + +func IdempotencyKey(req PublishRequest, payloadSHA string) string { + return strings.Join([]string{ + req.ArtifactType, + req.Environment, + req.ProjectExternalKey, + req.LogicalResourceID, + req.SourceEventKind, + req.SourceEventID, + payloadSHA, + nz(req.PolicyVersion, PolicyVersion), + }, "|") +} + +func nz(v, def string) string { + if strings.TrimSpace(v) == "" { + return def + } + return v +} + +func ValidateEnv(env string) error { + switch env { + case "dev", "test", "main": + return nil + default: + return fmt.Errorf("artifactvault: invalid environment %q", env) + } +} diff --git a/scripts/artifact-vault-preflight.sh b/scripts/artifact-vault-preflight.sh new file mode 100755 index 00000000..3926ec6d --- /dev/null +++ b/scripts/artifact-vault-preflight.sh @@ -0,0 +1,9 @@ +#!/usr/bin/env bash +# Fail-closed FAV coverage / destroy preflight. Does not start env-service. +set -euo pipefail +DSN="${APP_DB_DSN:-${TEST_DATABASE_URL:-}}" +if [[ -z "$DSN" ]]; then + echo "APP_DB_DSN or TEST_DATABASE_URL required" >&2 + exit 2 +fi +exec go run ./cmd/artifact-vault gc-preflight --dsn "$DSN" --backup-ok="${BACKUP_OK:-false}" --corpus-ok="${CORPUS_OK:-false}"