weichen
3e01f18756
User authorized lifting CURRENT FAV whole-branch merge protection on 2026-09-23. Keep FAV_DEGRADED_READ/FAV_PUBLISHER_ENABLED default false. Drop duplicate goose 00034 (local_dev already has 00036). Preserve agentx workspace admin routes and pin-export wiring.
285 lines
8.2 KiB
Go
285 lines
8.2 KiB
Go
package artifactvault
|
|
|
|
import (
|
|
"context"
|
|
"fmt"
|
|
"strings"
|
|
"time"
|
|
|
|
"github.com/jackc/pgx/v5"
|
|
)
|
|
|
|
const (
|
|
KindSnapshot = "snapshot"
|
|
KindAgentCommit = "agent_commit"
|
|
KindApproved = "approved"
|
|
KindChatTurn = "chat_turn"
|
|
|
|
HazardNone = ""
|
|
HazardMissingSnapshot = "missing_snapshot"
|
|
HazardProjectionFork = "projection_fork"
|
|
HazardLegacyV0 = "legacy_v0"
|
|
HazardHighID = "high_id"
|
|
HazardOldSchema = "old_schema"
|
|
HazardWorkspacePath = "workspace_path"
|
|
HazardSensitivePayload = "sensitive_payload"
|
|
|
|
HighIDThreshold int64 = 1_000_000_000
|
|
OldestSupportedSchema = 28
|
|
BackfillBatchSize = 200
|
|
)
|
|
|
|
// Candidate is a durable ledger row eligible for dataset/FAV backfill enumeration.
|
|
type Candidate struct {
|
|
Kind string
|
|
SourcePK int64
|
|
LogicalID string
|
|
ProjectID int64
|
|
SnapshotID int64
|
|
HeadSnapshotID int64
|
|
MessageID int64
|
|
EventID string
|
|
RunID string
|
|
Payload []byte
|
|
PayloadSHA string
|
|
CompletedAt *time.Time
|
|
ApprovedAt *time.Time
|
|
SourceSchema int
|
|
SchemaVer int
|
|
ProjectionSHA string
|
|
LedgerSHA string
|
|
ExportPolicy string
|
|
WorkDir string
|
|
}
|
|
|
|
type ProcessResult struct {
|
|
Status string
|
|
Hazard string
|
|
ArtifactID string
|
|
}
|
|
|
|
const (
|
|
SQLSnapshots = `
|
|
SELECT s.id, s.doc_id, COALESCE(s.content_sha256,''), COALESCE(b.content_markdown,''), s.source_kind
|
|
FROM document_snapshots s
|
|
LEFT JOIN document_content_blobs b ON b.content_sha256 = s.content_sha256
|
|
WHERE s.id > $1
|
|
ORDER BY s.id
|
|
LIMIT $2`
|
|
|
|
SQLAgentCommits = `
|
|
SELECT e.id, e.doc_id, e.to_snapshot_id, e.idempotency_key, COALESCE(b.content_markdown,''), COALESCE(s.content_sha256,'')
|
|
FROM document_change_events e
|
|
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' AND e.id > $1
|
|
ORDER BY e.id
|
|
LIMIT $2`
|
|
|
|
SQLApproved = `
|
|
SELECT r.id, r.doc_id, r.project_id, r.approved_at, COALESCE(r.content_sha256,''),
|
|
COALESCE(v.head_snapshot_id, 0), COALESCE(hs.content_sha256,''), COALESCE(b.content_markdown,'')
|
|
FROM document_records r
|
|
LEFT JOIN document_versions v ON v.id = r.ledger_version_id
|
|
LEFT JOIN document_snapshots hs ON hs.id = v.head_snapshot_id
|
|
LEFT JOIN document_content_blobs b ON b.content_sha256 = r.content_sha256
|
|
WHERE r.current_state = 'approved' AND r.approved_at IS NOT NULL AND r.id > $1
|
|
ORDER BY r.id
|
|
LIMIT $2`
|
|
|
|
SQLChatTurns = `
|
|
SELECT t.id, COALESCE(t.project_id, 0), t.assistant_message_id, t.completed_at, t.session_id, COALESCE(m.content,'')
|
|
FROM chat_turns t
|
|
JOIN chat_messages m ON m.id = t.assistant_message_id
|
|
WHERE t.status = 'completed' AND t.completed_at IS NOT NULL
|
|
AND t.assistant_message_id IS NOT NULL AND t.id > $1
|
|
ORDER BY t.id
|
|
LIMIT $2`
|
|
)
|
|
|
|
func EnumeratorSQL(kind string) (string, error) {
|
|
var sql string
|
|
switch kind {
|
|
case KindSnapshot:
|
|
sql = SQLSnapshots
|
|
case KindAgentCommit:
|
|
sql = SQLAgentCommits
|
|
case KindApproved:
|
|
sql = SQLApproved
|
|
case KindChatTurn:
|
|
sql = SQLChatTurns
|
|
default:
|
|
return "", fmt.Errorf("artifactvault: unknown backfill kind %q", kind)
|
|
}
|
|
if err := ForbidWorkspaceScan(sql); err != nil {
|
|
return "", err
|
|
}
|
|
return sql, nil
|
|
}
|
|
|
|
func ForbidWorkspaceScan(sql string) error {
|
|
low := strings.ToLower(sql)
|
|
parts := [][]string{
|
|
{"work", "spaces/"},
|
|
{"output", ".md"},
|
|
{"work", "_dir"},
|
|
{"data/", "dev"},
|
|
{"data/", "test"},
|
|
{"data/", "main"},
|
|
}
|
|
for _, p := range parts {
|
|
needle := strings.Join(p, "")
|
|
if strings.Contains(low, needle) {
|
|
return fmt.Errorf("artifactvault: enumerator must not scan workspace or env data roots")
|
|
}
|
|
}
|
|
return nil
|
|
}
|
|
|
|
func ClassifyHazard(c Candidate) string {
|
|
if strings.TrimSpace(c.WorkDir) != "" || strings.Contains(strings.ToLower(c.LogicalID), strings.Join([]string{"work", "spaces/"}, "")) {
|
|
return HazardWorkspacePath
|
|
}
|
|
if c.SourcePK > HighIDThreshold {
|
|
return HazardHighID
|
|
}
|
|
if c.SourceSchema > 0 && c.SourceSchema < OldestSupportedSchema {
|
|
return HazardOldSchema
|
|
}
|
|
if c.Kind == KindSnapshot {
|
|
if len(c.Payload) == 0 || c.PayloadSHA == "" {
|
|
return HazardMissingSnapshot
|
|
}
|
|
if c.SchemaVer == 0 {
|
|
return HazardLegacyV0
|
|
}
|
|
return HazardNone
|
|
}
|
|
if c.Kind == KindAgentCommit || c.Kind == KindApproved {
|
|
if c.SnapshotID <= 0 || len(c.Payload) == 0 {
|
|
return HazardMissingSnapshot
|
|
}
|
|
if c.Kind == KindApproved && c.HeadSnapshotID > 0 && c.LedgerSHA != "" && c.ProjectionSHA != "" && c.LedgerSHA != c.ProjectionSHA {
|
|
return HazardProjectionFork
|
|
}
|
|
}
|
|
if c.Kind == KindChatTurn && (c.MessageID <= 0 || c.CompletedAt == nil || len(c.Payload) == 0) {
|
|
return HazardMissingSnapshot
|
|
}
|
|
if ClassifyPayload(c.Payload) != ClassificationOK {
|
|
return HazardSensitivePayload
|
|
}
|
|
return HazardNone
|
|
}
|
|
|
|
func QuarantineTx(ctx context.Context, tx pgx.Tx, reason, logical string) error {
|
|
_, err := tx.Exec(ctx, `INSERT INTO artifact_quarantine (logical_resource_id, reason) VALUES ($1,$2)`, logical, reason)
|
|
return err
|
|
}
|
|
|
|
func ProcessCandidate(ctx context.Context, tx pgx.Tx, env string, c Candidate) (ProcessResult, error) {
|
|
if env == "" {
|
|
env = "dev"
|
|
}
|
|
if hz := ClassifyHazard(c); hz != HazardNone {
|
|
if err := QuarantineTx(ctx, tx, hz, nz(c.LogicalID, fmt.Sprintf("%s:%d", c.Kind, c.SourcePK))); err != nil {
|
|
return ProcessResult{}, err
|
|
}
|
|
return ProcessResult{Status: "quarantine", Hazard: hz}, nil
|
|
}
|
|
if c.Kind == KindSnapshot {
|
|
return ProcessResult{Status: "done"}, nil
|
|
}
|
|
ing := &Ingest{Env: env}
|
|
var err error
|
|
switch c.Kind {
|
|
case KindAgentCommit:
|
|
err = ing.AgentCommitTX(ctx, tx, c.ProjectID, c.LogicalID, nz(c.EventID, fmt.Sprintf("commit:%d", c.SourcePK)), c.SnapshotID, c.Payload)
|
|
case KindApproved:
|
|
err = ing.ApprovedTX(ctx, tx, c.ProjectID, c.LogicalID, c.SnapshotID, c.ApprovedAt, c.Payload)
|
|
case KindChatTurn:
|
|
policy := nz(c.ExportPolicy, "full")
|
|
err = ing.requestTX(ctx, tx, AssistantFacts("completed", c.CompletedAt, c.MessageID, policy), PublishRequest{
|
|
ArtifactType: TypeAssistantReply, Environment: env, ProjectID: c.ProjectID,
|
|
LogicalResourceID: nz(c.LogicalID, fmt.Sprintf("turn:%d", c.SourcePK)),
|
|
SourceEventKind: "assistant_completed", SourceEventID: nz(c.EventID, fmt.Sprintf("turn:%d", c.SourcePK)),
|
|
SourceMessageID: c.MessageID, SourceRunID: c.RunID, Payload: c.Payload, TerminalState: "completed",
|
|
})
|
|
default:
|
|
return ProcessResult{}, fmt.Errorf("artifactvault: unsupported backfill kind %q", c.Kind)
|
|
}
|
|
if err != nil {
|
|
return ProcessResult{}, err
|
|
}
|
|
return ProcessResult{Status: "done"}, nil
|
|
}
|
|
|
|
type rowQuerier interface {
|
|
Query(ctx context.Context, sql string, args ...any) (pgx.Rows, error)
|
|
}
|
|
|
|
func Enumerate(ctx context.Context, q rowQuerier, kind string, lastPK int64, limit int) ([]Candidate, error) {
|
|
sql, err := EnumeratorSQL(kind)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
if limit <= 0 || limit > BackfillBatchSize {
|
|
limit = BackfillBatchSize
|
|
}
|
|
rows, err := q.Query(ctx, sql, lastPK, limit)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
defer rows.Close()
|
|
out := make([]Candidate, 0, limit)
|
|
for rows.Next() {
|
|
c, err := scanCandidate(kind, rows)
|
|
if err != nil {
|
|
return nil, err
|
|
}
|
|
out = append(out, c)
|
|
}
|
|
return out, rows.Err()
|
|
}
|
|
|
|
type scannable interface {
|
|
Scan(dest ...any) error
|
|
}
|
|
|
|
func scanCandidate(kind string, row scannable) (Candidate, error) {
|
|
var c Candidate
|
|
c.Kind = kind
|
|
switch kind {
|
|
case KindSnapshot:
|
|
var srcKind string
|
|
if err := row.Scan(&c.SourcePK, &c.LogicalID, &c.PayloadSHA, &c.Payload, &srcKind); err != nil {
|
|
return c, err
|
|
}
|
|
c.SchemaVer = 1
|
|
if len(c.Payload) == 0 {
|
|
c.SchemaVer = 0
|
|
}
|
|
_ = srcKind
|
|
case KindAgentCommit:
|
|
if err := row.Scan(&c.SourcePK, &c.LogicalID, &c.SnapshotID, &c.EventID, &c.Payload, &c.PayloadSHA); err != nil {
|
|
return c, err
|
|
}
|
|
case KindApproved:
|
|
if err := row.Scan(&c.SourcePK, &c.LogicalID, &c.ProjectID, &c.ApprovedAt, &c.LedgerSHA, &c.HeadSnapshotID, &c.ProjectionSHA, &c.Payload); err != nil {
|
|
return c, err
|
|
}
|
|
c.SnapshotID = c.HeadSnapshotID
|
|
case KindChatTurn:
|
|
var session string
|
|
if err := row.Scan(&c.SourcePK, &c.ProjectID, &c.MessageID, &c.CompletedAt, &session, &c.Payload); err != nil {
|
|
return c, err
|
|
}
|
|
c.LogicalID = session
|
|
c.EventID = fmt.Sprintf("turn:%d", c.SourcePK)
|
|
c.ExportPolicy = "full"
|
|
default:
|
|
return c, fmt.Errorf("artifactvault: scan %s", kind)
|
|
}
|
|
return c, nil
|
|
}
|