glpi-neural-brain/internal/sourceagent/store.go

package sourceagent

import (
	"context"
	"crypto/rand"
	"crypto/sha256"
	"database/sql"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"hash/fnv"
	"math/bits"
	"net/url"
	"path/filepath"
	"sort"
	"strings"
	"sync"
	"time"
	"unicode"

	_ "modernc.org/sqlite"
)

type Store struct {
	db *sql.DB
	mu sync.Mutex

	computeMu    sync.Mutex
	computeJobs  map[string]*computeJobState
	computeOrder []string

	articleQualityMu    sync.Mutex
	articleQualityJobs  map[string]*articleQualityJobState
	articleQualityOrder []string
}

func OpenStore(dataDir string) (*Store, error) {
	path := filepath.Join(dataDir, "source-agents.db")
	db, err := sql.Open("sqlite", "file:"+filepath.ToSlash(path)+"?_pragma=busy_timeout(5000)&_pragma=foreign_keys(1)&_pragma=journal_mode(WAL)")
	if err != nil {
		return nil, err
	}
	db.SetMaxOpenConns(4)
	store := &Store{db: db, computeJobs: map[string]*computeJobState{}, articleQualityJobs: map[string]*articleQualityJobState{}}
	if err := store.init(context.Background()); err != nil {
		_ = db.Close()
		return nil, err
	}
	return store, nil
}

func (s *Store) Close() error {
	if s == nil || s.db == nil {
		return nil
	}
	return s.db.Close()
}

func (s *Store) init(ctx context.Context) error {
	statements := []string{
		`CREATE TABLE IF NOT EXISTS source_agents (
			id TEXT PRIMARY KEY, name TEXT NOT NULL, token_hash TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1,
			created_at_ns INTEGER NOT NULL, updated_at_ns INTEGER NOT NULL, last_seen_ns INTEGER NOT NULL DEFAULT 0,
			last_error TEXT NOT NULL DEFAULT '', version TEXT NOT NULL DEFAULT '', capabilities_json TEXT NOT NULL DEFAULT '[]'
		) WITHOUT ROWID`,
		`CREATE TABLE IF NOT EXISTS source_tasks (
			id TEXT PRIMARY KEY, agent_id TEXT NOT NULL REFERENCES source_agents(id) ON DELETE CASCADE,
			name TEXT NOT NULL, type TEXT NOT NULL, url TEXT NOT NULL, enabled INTEGER NOT NULL DEFAULT 1,
			poll_interval TEXT NOT NULL DEFAULT '4h', categories_json TEXT NOT NULL DEFAULT '[]', max_items INTEGER NOT NULL DEFAULT 20,
			config_json TEXT NOT NULL DEFAULT '{}', created_at_ns INTEGER NOT NULL, updated_at_ns INTEGER NOT NULL
		) WITHOUT ROWID`,
		`CREATE INDEX IF NOT EXISTS idx_source_tasks_agent ON source_tasks(agent_id, enabled, updated_at_ns)`,
		`CREATE TABLE IF NOT EXISTS source_meta (
			key TEXT PRIMARY KEY, value TEXT NOT NULL
		) WITHOUT ROWID`,
		`CREATE TABLE IF NOT EXISTS source_inbox (
			id TEXT PRIMARY KEY, agent_id TEXT NOT NULL, task_id TEXT NOT NULL, external_id TEXT NOT NULL DEFAULT '',
			url TEXT NOT NULL, canonical_url TEXT NOT NULL, title TEXT NOT NULL, published_at_ns INTEGER NOT NULL DEFAULT 0,
			discovered_at_ns INTEGER NOT NULL DEFAULT 0, received_at_ns INTEGER NOT NULL, updated_at_ns INTEGER NOT NULL,
			language TEXT NOT NULL DEFAULT '', content_type TEXT NOT NULL DEFAULT '', text_content TEXT NOT NULL,
			content_sha256 TEXT NOT NULL, source_name TEXT NOT NULL DEFAULT '', source_base_url TEXT NOT NULL DEFAULT '',
			categories_json TEXT NOT NULL DEFAULT '[]', metadata_json TEXT NOT NULL DEFAULT '{}', signature INTEGER NOT NULL DEFAULT 0,
			status TEXT NOT NULL DEFAULT 'received', relevance REAL NOT NULL DEFAULT 0, matched_node_id TEXT NOT NULL DEFAULT '',
			UNIQUE(agent_id, task_id, canonical_url, content_sha256)
		) WITHOUT ROWID`,
		`CREATE INDEX IF NOT EXISTS idx_source_inbox_status_received ON source_inbox(status, received_at_ns)`,
		`CREATE INDEX IF NOT EXISTS idx_source_inbox_candidate_updated ON source_inbox(status, updated_at_ns DESC)`,
	}
	for _, statement := range statements {
		if _, err := s.db.ExecContext(ctx, statement); err != nil {
			return err
		}
	}
	if err := s.ensureAgentColumn(ctx, "capabilities_json", `TEXT NOT NULL DEFAULT '[]'`); err != nil {
		return err
	}
	for _, column := range []struct {
		name string
		def  string
	}{
		{"proactive_state", `TEXT NOT NULL DEFAULT ''`},
		{"proactive_attempts", `INTEGER NOT NULL DEFAULT 0`},
		{"proactive_next_at_ns", `INTEGER NOT NULL DEFAULT 0`},
		{"materialized_node_id", `TEXT NOT NULL DEFAULT ''`},
	} {
		if err := s.ensureInboxColumn(ctx, column.name, column.def); err != nil {
			return err
		}
	}
	if _, err := s.db.ExecContext(ctx, `CREATE INDEX IF NOT EXISTS idx_source_inbox_proactive ON source_inbox(proactive_state, proactive_next_at_ns, updated_at_ns)`); err != nil {
		return err
	}
	if err := s.initController(ctx); err != nil {
		return err
	}
	// SQLite is a single-Brain store. If this process has just opened the DB, no
	// worker from the previous process can still own a claim. Recover all
	// in-flight claims immediately instead of leaving a 10/20 minute dead zone
	// after a crash or restart.
	now := time.Now().UTC().UnixNano()
	if _, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status='received',updated_at_ns=? WHERE status='processing'`, now); err != nil {
		return fmt.Errorf("recover source inbox classification claims: %w", err)
	}
	if _, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET proactive_state='queued',proactive_next_at_ns=0,updated_at_ns=? WHERE proactive_state='processing'`, now); err != nil {
		return fmt.Errorf("recover proactive security claims: %w", err)
	}
	return nil
}

func (s *Store) ensureAgentColumn(ctx context.Context, name, definition string) error {
	rows, err := s.db.QueryContext(ctx, `PRAGMA table_info(source_agents)`)
	if err != nil {
		return err
	}
	found := false
	for rows.Next() {
		var cid, notnull, pk int
		var colName, colType string
		var defaultValue any
		if err := rows.Scan(&cid, &colName, &colType, &notnull, &defaultValue, &pk); err != nil {
			rows.Close()
			return err
		}
		if colName == name {
			found = true
		}
	}
	if err := rows.Close(); err != nil {
		return err
	}
	if found {
		return nil
	}
	_, err = s.db.ExecContext(ctx, `ALTER TABLE source_agents ADD COLUMN `+name+` `+definition)
	return err
}

func (s *Store) ensureInboxColumn(ctx context.Context, name, definition string) error {
	rows, err := s.db.QueryContext(ctx, `PRAGMA table_info(source_inbox)`)
	if err != nil {
		return err
	}
	found := false
	for rows.Next() {
		var cid, notnull, pk int
		var colName, colType string
		var defaultValue any
		if err := rows.Scan(&cid, &colName, &colType, &notnull, &defaultValue, &pk); err != nil {
			rows.Close()
			return err
		}
		if colName == name {
			found = true
		}
	}
	if err := rows.Close(); err != nil {
		return err
	}
	if found {
		return nil
	}
	_, err = s.db.ExecContext(ctx, `ALTER TABLE source_inbox ADD COLUMN `+name+` `+definition)
	return err
}

func (s *Store) EnsureInboxClassifierVersion(ctx context.Context, version int) (int, error) {
	if version < 1 {
		version = 1
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return 0, err
	}
	defer tx.Rollback()
	current := 0
	var raw string
	err = tx.QueryRowContext(ctx, `SELECT value FROM source_meta WHERE key='inbox_classifier_version'`).Scan(&raw)
	if err != nil && !errors.Is(err, sql.ErrNoRows) {
		return 0, err
	}
	if err == nil {
		_, _ = fmt.Sscanf(raw, "%d", &current)
	}
	if current >= version {
		return 0, tx.Commit()
	}
	res, err := tx.ExecContext(ctx, `UPDATE source_inbox SET status='received', updated_at_ns=? WHERE status='archived'`, time.Now().UTC().UnixNano())
	if err != nil {
		return 0, err
	}
	requeued64, _ := res.RowsAffected()
	if _, err := tx.ExecContext(ctx, `INSERT INTO source_meta(key,value) VALUES('inbox_classifier_version',?) ON CONFLICT(key) DO UPDATE SET value=excluded.value`, fmt.Sprint(version)); err != nil {
		return 0, err
	}
	if err := tx.Commit(); err != nil {
		return 0, err
	}
	return int(requeued64), nil
}

func GenerateToken() (string, string, error) {
	buf := make([]byte, 32)
	if _, err := rand.Read(buf); err != nil {
		return "", "", err
	}
	token := "brain_agent_" + hex.EncodeToString(buf)
	return token, hashToken(token), nil
}

func hashToken(token string) string {
	sum := sha256.Sum256([]byte(strings.TrimSpace(token)))
	return hex.EncodeToString(sum[:])
}

func randomID(prefix string) string {
	buf := make([]byte, 12)
	_, _ = rand.Read(buf)
	return prefix + "-" + hex.EncodeToString(buf)
}

func normalizeID(value, prefix string) string {
	value = strings.ToLower(strings.TrimSpace(value))
	var b strings.Builder
	for _, r := range value {
		if unicode.IsLetter(r) || unicode.IsDigit(r) || r == '-' || r == '_' {
			b.WriteRune(r)
		}
	}
	value = strings.Trim(b.String(), "-_")
	if value == "" {
		return randomID(prefix)
	}
	return value
}

func (s *Store) CreateAgent(ctx context.Context, id, name string) (Agent, string, error) {
	id = normalizeID(id, "agent")
	name = strings.TrimSpace(name)
	if name == "" {
		name = id
	}
	token, tokenHash, err := GenerateToken()
	if err != nil {
		return Agent{}, "", err
	}
	now := time.Now().UTC()
	_, err = s.db.ExecContext(ctx, `INSERT INTO source_agents(id,name,token_hash,enabled,created_at_ns,updated_at_ns) VALUES(?,?,?,?,?,?)`, id, name, tokenHash, 1, now.UnixNano(), now.UnixNano())
	if err != nil {
		return Agent{}, "", err
	}
	return Agent{ID: id, Name: name, Enabled: true, CreatedAt: now, UpdatedAt: now}, token, nil
}

func (s *Store) RotateToken(ctx context.Context, id string) (string, error) {
	token, tokenHash, err := GenerateToken()
	if err != nil {
		return "", err
	}
	result, err := s.db.ExecContext(ctx, `UPDATE source_agents SET token_hash=?, updated_at_ns=? WHERE id=?`, tokenHash, time.Now().UTC().UnixNano(), id)
	if err != nil {
		return "", err
	}
	n, _ := result.RowsAffected()
	if n == 0 {
		return "", sql.ErrNoRows
	}
	return token, nil
}

func (s *Store) SetAgentEnabled(ctx context.Context, id string, enabled bool) error {
	v := 0
	if enabled {
		v = 1
	}
	result, err := s.db.ExecContext(ctx, `UPDATE source_agents SET enabled=?,updated_at_ns=? WHERE id=?`, v, time.Now().UTC().UnixNano(), id)
	if err != nil {
		return err
	}
	n, _ := result.RowsAffected()
	if n == 0 {
		return sql.ErrNoRows
	}
	return nil
}

func (s *Store) DeleteAgent(ctx context.Context, id string) error {
	_, err := s.db.ExecContext(ctx, `DELETE FROM source_agents WHERE id=?`, id)
	return err
}

func (s *Store) ListAgents(ctx context.Context) ([]Agent, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT a.id,a.name,a.enabled,a.created_at_ns,a.updated_at_ns,a.last_seen_ns,a.last_error,a.version,a.capabilities_json,a.controller_json,(SELECT COUNT(*) FROM source_tasks t WHERE t.agent_id=a.id) FROM source_agents a ORDER BY a.name,a.id`)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]Agent, 0)
	for rows.Next() {
		var a Agent
		var en int
		var c, u, seen int64
		var capabilities, controller string
		if err := rows.Scan(&a.ID, &a.Name, &en, &c, &u, &seen, &a.LastError, &a.Version, &capabilities, &controller, &a.TaskCount); err != nil {
			return nil, err
		}
		a.Enabled = en != 0
		a.CreatedAt = nsTime(c)
		a.UpdatedAt = nsTime(u)
		a.LastSeen = nsTime(seen)
		_ = json.Unmarshal([]byte(capabilities), &a.Capabilities)
		a.Capabilities = uniqueStrings(a.Capabilities)
		_ = json.Unmarshal([]byte(controller), &a.Controller)
		out = append(out, a)
	}
	return out, rows.Err()
}

func (s *Store) TouchAgent(ctx context.Context, id string) error {
	result, err := s.db.ExecContext(ctx, `UPDATE source_agents SET last_seen_ns=? WHERE id=?`, time.Now().UTC().UnixNano(), id)
	if err != nil {
		return err
	}
	n, _ := result.RowsAffected()
	if n == 0 {
		return sql.ErrNoRows
	}
	return nil
}

func (s *Store) Authenticate(ctx context.Context, token string) (Agent, error) {
	if strings.TrimSpace(token) == "" {
		return Agent{}, errors.New("empty agent token")
	}
	var a Agent
	var en int
	var c, u, seen int64
	var capabilities, controller string
	err := s.db.QueryRowContext(ctx, `SELECT id,name,enabled,created_at_ns,updated_at_ns,last_seen_ns,last_error,version,capabilities_json,controller_json FROM source_agents WHERE token_hash=?`, hashToken(token)).Scan(&a.ID, &a.Name, &en, &c, &u, &seen, &a.LastError, &a.Version, &capabilities, &controller)
	if err != nil {
		return Agent{}, err
	}
	a.Enabled = en != 0
	if !a.Enabled {
		return Agent{}, errors.New("agent disabled")
	}
	a.CreatedAt = nsTime(c)
	a.UpdatedAt = nsTime(u)
	a.LastSeen = nsTime(seen)
	_ = json.Unmarshal([]byte(capabilities), &a.Capabilities)
	a.Capabilities = uniqueStrings(a.Capabilities)
	_ = json.Unmarshal([]byte(controller), &a.Controller)
	return a, nil
}

func validateTask(t Task) (Task, error) {
	t.ID = normalizeID(t.ID, "task")
	t.Name = strings.TrimSpace(t.Name)
	if t.Name == "" {
		t.Name = t.ID
	}
	t.Type = strings.ToLower(strings.TrimSpace(t.Type))
	switch t.Type {
	case "rss", "atom", "sitemap", "web":
	default:
		return t, fmt.Errorf("unsupported source task type %q", t.Type)
	}
	u, err := url.Parse(strings.TrimSpace(t.URL))
	if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") {
		return t, fmt.Errorf("task URL must be absolute http(s)")
	}
	if u.User != nil {
		return t, fmt.Errorf("task URL must not contain userinfo")
	}
	t.URL = u.String()
	if strings.TrimSpace(t.PollInterval) == "" {
		t.PollInterval = "4h"
	}
	d, err := time.ParseDuration(t.PollInterval)
	if err != nil || d < 5*time.Minute || d > 30*24*time.Hour {
		return t, fmt.Errorf("poll_interval must be between 5m and 720h")
	}
	if t.MaxItems <= 0 {
		t.MaxItems = 20
	}
	if t.MaxItems > 500 {
		return t, fmt.Errorf("max_items must be <=500")
	}
	if t.Config == nil {
		t.Config = map[string]string{}
	}
	return t, nil
}

func (s *Store) UpsertTask(ctx context.Context, t Task) (Task, error) {
	var err error
	t, err = validateTask(t)
	if err != nil {
		return Task{}, err
	}
	if strings.TrimSpace(t.AgentID) == "" {
		return Task{}, errors.New("agent_id required")
	}
	var agentExists int
	if err := s.db.QueryRowContext(ctx, `SELECT 1 FROM source_agents WHERE id=?`, t.AgentID).Scan(&agentExists); err != nil {
		if errors.Is(err, sql.ErrNoRows) {
			return Task{}, fmt.Errorf("unknown source agent %q; create/register the agent in the Brain first", t.AgentID)
		}
		return Task{}, err
	}
	cats, _ := json.Marshal(uniqueStrings(t.Categories))
	cfg, _ := json.Marshal(t.Config)
	now := time.Now().UTC()
	_, err = s.db.ExecContext(ctx, `INSERT INTO source_tasks(id,agent_id,name,type,url,enabled,poll_interval,categories_json,max_items,config_json,created_at_ns,updated_at_ns) VALUES(?,?,?,?,?,?,?,?,?,?,?,?) ON CONFLICT(id) DO UPDATE SET agent_id=excluded.agent_id,name=excluded.name,type=excluded.type,url=excluded.url,enabled=excluded.enabled,poll_interval=excluded.poll_interval,categories_json=excluded.categories_json,max_items=excluded.max_items,config_json=excluded.config_json,updated_at_ns=excluded.updated_at_ns`, t.ID, t.AgentID, t.Name, t.Type, t.URL, boolInt(t.Enabled), t.PollInterval, string(cats), t.MaxItems, string(cfg), now.UnixNano(), now.UnixNano())
	if err != nil {
		return Task{}, err
	}
	t.UpdatedAt = now
	if t.CreatedAt.IsZero() {
		t.CreatedAt = now
	}
	return t, nil
}

func (s *Store) DeleteTask(ctx context.Context, id string) error {
	_, err := s.db.ExecContext(ctx, `DELETE FROM source_tasks WHERE id=?`, id)
	return err
}
func (s *Store) ListTasks(ctx context.Context, agentID string) ([]Task, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT id,agent_id,name,type,url,enabled,poll_interval,categories_json,max_items,config_json,created_at_ns,updated_at_ns FROM source_tasks WHERE (?='' OR agent_id=?) ORDER BY name,id`, agentID, agentID)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]Task, 0)
	for rows.Next() {
		var t Task
		var en int
		var cats, cfg string
		var c, u int64
		if err := rows.Scan(&t.ID, &t.AgentID, &t.Name, &t.Type, &t.URL, &en, &t.PollInterval, &cats, &t.MaxItems, &cfg, &c, &u); err != nil {
			return nil, err
		}
		t.Enabled = en != 0
		_ = json.Unmarshal([]byte(cats), &t.Categories)
		_ = json.Unmarshal([]byte(cfg), &t.Config)
		t.CreatedAt = nsTime(c)
		t.UpdatedAt = nsTime(u)
		out = append(out, t)
	}
	return out, rows.Err()
}

func (s *Store) RemoteConfig(ctx context.Context, agent Agent) (RemoteConfig, error) {
	tasks, err := s.ListTasks(ctx, agent.ID)
	if err != nil {
		return RemoteConfig{}, err
	}
	return RemoteConfig{SchemaVersion: SchemaVersion, Agent: agent, Tasks: tasks, IssuedAt: time.Now().UTC()}, nil
}

func (s *Store) Heartbeat(ctx context.Context, agentID string, h Heartbeat) error {
	capabilities := []string{}
	controllerJSON := "{}"
	if h.Metadata != nil {
		for _, key := range []string{"compute_kinds", "capabilities"} {
			switch value := h.Metadata[key].(type) {
			case []string:
				capabilities = append(capabilities, value...)
			case []any:
				for _, item := range value {
					if text, ok := item.(string); ok {
						capabilities = append(capabilities, text)
					}
				}
			}
		}
		if controller, ok := h.Metadata["docker_controller"]; ok {
			if raw, err := json.Marshal(controller); err == nil && len(raw) <= 64<<10 {
				controllerJSON = string(raw)
			}
		}
	}
	capabilities = uniqueStrings(capabilities)
	capJSON, _ := json.Marshal(capabilities)
	now := time.Now().UTC().UnixNano()
	_, err := s.db.ExecContext(ctx, `UPDATE source_agents SET last_seen_ns=?,last_error=?,version=?,capabilities_json=?,controller_json=?,updated_at_ns=? WHERE id=?`, now, strings.TrimSpace(h.LastError), strings.TrimSpace(h.Version), string(capJSON), controllerJSON, now, agentID)
	return err
}

func (s *Store) HasOnlineComputeAgent(ctx context.Context, kind string, maxAge time.Duration) (bool, error) {
	if maxAge <= 0 {
		maxAge = 3 * time.Minute
	}
	agents, err := s.ListAgents(ctx)
	if err != nil {
		return false, err
	}
	now := time.Now().UTC()
	kind = strings.TrimSpace(kind)
	for _, agent := range agents {
		if !agent.Enabled || agent.LastSeen.IsZero() || now.Sub(agent.LastSeen) > maxAge {
			continue
		}
		for _, capability := range agent.Capabilities {
			if capability == kind {
				return true, nil
			}
		}
	}
	return false, nil
}

func normalizeDocument(d Document) (Document, error) {
	d.URL = strings.TrimSpace(d.URL)
	if d.URL == "" {
		return d, errors.New("document url required")
	}
	u, err := url.Parse(d.URL)
	if err != nil || u.Host == "" || (u.Scheme != "http" && u.Scheme != "https") {
		return d, errors.New("invalid document url")
	}
	d.CanonicalURL = strings.TrimSpace(d.CanonicalURL)
	if d.CanonicalURL == "" {
		d.CanonicalURL = d.URL
	}
	d.Title = strings.TrimSpace(d.Title)
	d.Text = strings.TrimSpace(d.Text)
	if len([]rune(d.Text)) < 80 {
		return d, errors.New("document text too short")
	}
	if d.DiscoveredAt.IsZero() {
		d.DiscoveredAt = time.Now().UTC()
	}
	if d.ExternalID == "" {
		d.ExternalID = d.CanonicalURL
	}
	d.Categories = uniqueStrings(d.Categories)
	if d.Metadata == nil {
		d.Metadata = map[string]any{}
	}
	if d.ContentSHA256 == "" {
		sum := sha256.Sum256([]byte(d.Text))
		d.ContentSHA256 = hex.EncodeToString(sum[:])
	}
	return d, nil
}

func (s *Store) Ingest(ctx context.Context, agentID, taskID string, docs []Document) (IngestResult, error) {
	var taskOwner string
	if err := s.db.QueryRowContext(ctx, `SELECT agent_id FROM source_tasks WHERE id=? AND enabled=1`, taskID).Scan(&taskOwner); err != nil {
		return IngestResult{}, fmt.Errorf("unknown or disabled source task: %w", err)
	}
	if taskOwner != agentID {
		return IngestResult{}, errors.New("source task does not belong to authenticated agent")
	}
	result := IngestResult{BatchID: randomID("batch")}
	now := time.Now().UTC()
	s.mu.Lock()
	defer s.mu.Unlock()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return result, err
	}
	defer tx.Rollback()
	for _, raw := range docs {
		d, e := normalizeDocument(raw)
		if e != nil {
			result.Rejected++
			continue
		}
		cats, _ := json.Marshal(d.Categories)
		meta, _ := json.Marshal(d.Metadata)
		idSum := sha256.Sum256([]byte(agentID + "\x00" + taskID + "\x00" + d.CanonicalURL + "\x00" + d.ContentSHA256))
		id := hex.EncodeToString(idSum[:12])
		sig := int64(simhash64(d.Title + "\n" + d.Text))
		res, e := tx.ExecContext(ctx, `INSERT OR IGNORE INTO source_inbox(id,agent_id,task_id,external_id,url,canonical_url,title,published_at_ns,discovered_at_ns,received_at_ns,updated_at_ns,language,content_type,text_content,content_sha256,source_name,source_base_url,categories_json,metadata_json,signature,status) VALUES(?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)`, id, agentID, taskID, d.ExternalID, d.URL, d.CanonicalURL, d.Title, timeNS(d.PublishedAt), timeNS(d.DiscoveredAt), now.UnixNano(), now.UnixNano(), d.Language, d.ContentType, d.Text, d.ContentSHA256, d.SourceName, d.SourceBaseURL, string(cats), string(meta), sig, "received")
		if e != nil {
			result.Rejected++
			continue
		}
		n, _ := res.RowsAffected()
		if n == 0 {
			result.Duplicates++
		} else {
			result.Accepted++
		}
	}
	if err := tx.Commit(); err != nil {
		return result, err
	}
	return result, nil
}

func (s *Store) ClaimInbox(ctx context.Context, limit int) ([]InboxDocument, error) {
	if limit < 1 {
		limit = 1
	}
	if limit > 100 {
		limit = 100
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	rows, err := tx.QueryContext(ctx, `SELECT id FROM source_inbox WHERE status='received' ORDER BY received_at_ns LIMIT ?`, limit)
	if err != nil {
		return nil, err
	}
	var ids []string
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			rows.Close()
			return nil, err
		}
		ids = append(ids, id)
	}
	rows.Close()
	if len(ids) == 0 {
		return nil, tx.Commit()
	}
	now := time.Now().UTC().UnixNano()
	claimed := make([]string, 0, len(ids))
	for _, id := range ids {
		res, err := tx.ExecContext(ctx, `UPDATE source_inbox SET status='processing',updated_at_ns=? WHERE id=? AND status='received'`, now, id)
		if err != nil {
			return nil, err
		}
		rowsAffected, err := res.RowsAffected()
		if err != nil {
			return nil, err
		}
		if rowsAffected == 1 {
			claimed = append(claimed, id)
		}
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return s.GetInboxByIDs(ctx, claimed)
}

func (s *Store) CompleteClassification(ctx context.Context, id, status string, relevance float64, matchedNodeID string, meta map[string]any) error {
	if status != "candidate" && status != "archived" {
		return errors.New("invalid inbox classification status")
	}
	data, _ := json.Marshal(meta)
	res, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status=?,relevance=?,matched_node_id=?,metadata_json=?,updated_at_ns=? WHERE id=? AND status='processing'`, status, relevance, matchedNodeID, string(data), time.Now().UTC().UnixNano(), id)
	return requireOneInboxRow(res, err, id, "complete inbox classification")
}

// QueueProactiveSecurity marks a classified candidate for the bounded proactive
// security pipeline while keeping its public inbox status as candidate. This
// lets the document remain available for normal evidence lookup even while the
// security worker is pending or retrying it.
func (s *Store) QueueProactiveSecurity(ctx context.Context, id string) error {
	res, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET proactive_state=CASE WHEN proactive_state='' THEN 'queued' ELSE proactive_state END, proactive_next_at_ns=0, updated_at_ns=? WHERE id=? AND status='candidate' AND proactive_state IN ('','queued')`, time.Now().UTC().UnixNano(), id)
	return requireOneInboxRow(res, err, id, "queue proactive security")
}

func (s *Store) ClaimProactiveSecurity(ctx context.Context, limit int) ([]InboxDocument, error) {
	if limit < 1 {
		limit = 1
	}
	if limit > 20 {
		limit = 20
	}
	s.mu.Lock()
	defer s.mu.Unlock()
	tx, err := s.db.BeginTx(ctx, nil)
	if err != nil {
		return nil, err
	}
	defer tx.Rollback()
	now := time.Now().UTC().UnixNano()
	rows, err := tx.QueryContext(ctx, `SELECT id FROM source_inbox WHERE status='candidate' AND proactive_state='queued' AND (proactive_next_at_ns=0 OR proactive_next_at_ns<=?) ORDER BY relevance DESC, received_at_ns LIMIT ?`, now, limit)
	if err != nil {
		return nil, err
	}
	ids := make([]string, 0, limit)
	for rows.Next() {
		var id string
		if err := rows.Scan(&id); err != nil {
			rows.Close()
			return nil, err
		}
		ids = append(ids, id)
	}
	rows.Close()
	claimed := make([]string, 0, len(ids))
	for _, id := range ids {
		res, err := tx.ExecContext(ctx, `UPDATE source_inbox SET proactive_state='processing', proactive_attempts=proactive_attempts+1, updated_at_ns=? WHERE id=? AND proactive_state='queued'`, now, id)
		if err != nil {
			return nil, err
		}
		rowsAffected, err := res.RowsAffected()
		if err != nil {
			return nil, err
		}
		if rowsAffected == 1 {
			claimed = append(claimed, id)
		}
	}
	if err := tx.Commit(); err != nil {
		return nil, err
	}
	return s.GetInboxByIDs(ctx, claimed)
}

func (s *Store) StartProactiveSecurityRun(ctx context.Context, id string) (string, time.Time, error) {
	current, err := s.getInbox(ctx, id)
	if err != nil {
		return "", time.Time{}, err
	}
	started := time.Now().UTC()
	runID := fmt.Sprintf("security-%s-%d", id, started.UnixNano())
	meta := make(map[string]any, len(current.Metadata)+4)
	for k, v := range current.Metadata {
		meta[k] = v
	}
	meta["proactive_run_id"] = runID
	meta["proactive_started_at"] = started
	meta["proactive_completed_at"] = nil
	meta["proactive_duration_ms"] = int64(0)
	meta["proactive_outcome"] = "running"
	meta["proactive_last_error"] = ""
	data, _ := json.Marshal(meta)
	res, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET metadata_json=?,updated_at_ns=? WHERE id=? AND proactive_state='processing'`, string(data), started.UnixNano(), id)
	if err != nil {
		return "", time.Time{}, err
	}
	rows, err := res.RowsAffected()
	if err != nil {
		return "", time.Time{}, err
	}
	if rows != 1 {
		return "", time.Time{}, fmt.Errorf("security inbox %s is not in processing state", id)
	}
	return runID, started, nil
}

func proactiveLifecycleFinish(meta map[string]any, outcome, lastError string) map[string]any {
	if meta == nil {
		meta = map[string]any{}
	}
	completed := time.Now().UTC()
	started := metadataTimeValue(meta["proactive_started_at"])
	if started.IsZero() {
		started = completed
	}
	meta["proactive_completed_at"] = completed
	meta["proactive_duration_ms"] = completed.Sub(started).Milliseconds()
	meta["proactive_outcome"] = outcome
	meta["proactive_last_error"] = lastError
	return meta
}

func metadataTimeValue(value any) time.Time {
	switch v := value.(type) {
	case time.Time:
		return v.UTC()
	case string:
		if parsed, err := time.Parse(time.RFC3339Nano, strings.TrimSpace(v)); err == nil {
			return parsed.UTC()
		}
	}
	return time.Time{}
}

func metadataStringValue(value any) string {
	if value == nil {
		return ""
	}
	if s, ok := value.(string); ok {
		return strings.TrimSpace(s)
	}
	return strings.TrimSpace(fmt.Sprint(value))
}

func metadataFloatValue(value any) float64 {
	switch v := value.(type) {
	case float64:
		return v
	case float32:
		return float64(v)
	case int:
		return float64(v)
	case int64:
		return float64(v)
	case json.Number:
		f, _ := v.Float64()
		return f
	}
	return 0
}

func (s *Store) CompleteProactiveSecurity(ctx context.Context, id, nodeID string, meta map[string]any) error {
	current, err := s.getInbox(ctx, id)
	if err != nil {
		return err
	}
	merged := make(map[string]any, len(current.Metadata)+len(meta))
	for k, v := range current.Metadata {
		merged[k] = v
	}
	for k, v := range meta {
		merged[k] = v
	}
	merged = proactiveLifecycleFinish(merged, "materialized", "")
	data, _ := json.Marshal(merged)
	res, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status='materialized', proactive_state='done', proactive_next_at_ns=0, materialized_node_id=?, metadata_json=?, updated_at_ns=? WHERE id=? AND proactive_state='processing'`, nodeID, string(data), time.Now().UTC().UnixNano(), id)
	return requireOneInboxRow(res, err, id, "complete proactive security")
}

func (s *Store) RejectProactiveSecurity(ctx context.Context, id, reason string, meta map[string]any) error {
	current, err := s.getInbox(ctx, id)
	if err != nil {
		return err
	}
	merged := make(map[string]any, len(current.Metadata)+len(meta)+1)
	for k, v := range current.Metadata {
		merged[k] = v
	}
	for k, v := range meta {
		merged[k] = v
	}
	merged["proactive_security_reason"] = reason
	merged = proactiveLifecycleFinish(merged, "rejected", "")
	data, _ := json.Marshal(merged)
	res, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET proactive_state='rejected', proactive_next_at_ns=0, metadata_json=?, updated_at_ns=? WHERE id=? AND proactive_state='processing'`, string(data), time.Now().UTC().UnixNano(), id)
	return requireOneInboxRow(res, err, id, "reject proactive security")
}

func (s *Store) ReleaseProactiveSecurity(ctx context.Context, id, reason string) error {
	current, err := s.getInbox(ctx, id)
	if err != nil {
		return err
	}
	delay := 30 * time.Minute
	if current.ProactiveAttempts >= 2 {
		delay = 2 * time.Hour
	}
	if current.ProactiveAttempts >= 4 {
		delay = 12 * time.Hour
	}
	meta := make(map[string]any, len(current.Metadata)+1)
	for k, v := range current.Metadata {
		meta[k] = v
	}
	meta["proactive_security_error"] = reason
	meta = proactiveLifecycleFinish(meta, "failed_retry", reason)
	data, _ := json.Marshal(meta)
	res, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET proactive_state='queued', proactive_next_at_ns=?, metadata_json=?, updated_at_ns=? WHERE id=? AND proactive_state='processing'`, time.Now().UTC().Add(delay).UnixNano(), string(data), time.Now().UTC().UnixNano(), id)
	return requireOneInboxRow(res, err, id, "release proactive security")
}

func requireOneInboxRow(result sql.Result, err error, id, operation string) error {
	if err != nil {
		return err
	}
	rows, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if rows != 1 {
		return fmt.Errorf("%s for inbox %s affected %d rows", operation, id, rows)
	}
	return nil
}

// RequeueMissingMaterializedSecurity repairs the only non-transactional gap
// between source-agents.db and graph.db: the source store may have committed
// "done" just before a hard process exit while the graph changes were still
// waiting for their batched flush. Requeueing is safe because Security node/
// edge IDs are deterministic and graph upserts are idempotent.
func (s *Store) RequeueMissingMaterializedSecurity(ctx context.Context, id, reason string) error {
	current, err := s.getInbox(ctx, id)
	if err != nil {
		return err
	}
	meta := make(map[string]any, len(current.Metadata)+3)
	for k, v := range current.Metadata {
		meta[k] = v
	}
	meta["proactive_recovery_reason"] = strings.TrimSpace(reason)
	meta["proactive_outcome"] = "recovery_requeued"
	meta["proactive_last_error"] = ""
	data, _ := json.Marshal(meta)
	res, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status='candidate', proactive_state='queued', proactive_next_at_ns=0, materialized_node_id='', metadata_json=?, updated_at_ns=? WHERE id=? AND proactive_state='done' AND status IN ('materialized','used')`, string(data), time.Now().UTC().UnixNano(), id)
	return requireOneInboxRow(res, err, id, "requeue missing materialized security")
}

func (s *Store) MarkUsedByIDs(ctx context.Context, ids []string) error {
	now := time.Now().UTC().UnixNano()
	for _, id := range uniqueStrings(ids) {
		res, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status='used',updated_at_ns=? WHERE id=? AND status IN ('candidate','materialized','used')`, now, id)
		if err != nil {
			return err
		}
		rows, err := res.RowsAffected()
		if err != nil {
			return err
		}
		if rows != 1 {
			return fmt.Errorf("mark source inbox %s used affected %d rows", id, rows)
		}
	}
	return nil
}

func (s *Store) MarkUsed(ctx context.Context, canonicalURLs []string) error {
	now := time.Now().UTC().UnixNano()
	for _, u := range uniqueStrings(canonicalURLs) {
		if _, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status='used',updated_at_ns=? WHERE canonical_url=?`, now, u); err != nil {
			return err
		}
	}
	return nil
}

func (s *Store) GetInboxByIDs(ctx context.Context, ids []string) ([]InboxDocument, error) {
	if len(ids) == 0 {
		return nil, nil
	}
	out := make([]InboxDocument, 0, len(ids))
	for _, id := range ids {
		doc, err := s.getInbox(ctx, id)
		if err == nil {
			out = append(out, doc)
		}
	}
	return out, nil
}
func (s *Store) getInbox(ctx context.Context, id string) (InboxDocument, error) {
	row := s.db.QueryRowContext(ctx, `SELECT id,agent_id,task_id,external_id,url,canonical_url,title,published_at_ns,discovered_at_ns,received_at_ns,updated_at_ns,language,content_type,text_content,content_sha256,source_name,source_base_url,categories_json,metadata_json,status,relevance,matched_node_id,proactive_state,proactive_attempts,materialized_node_id FROM source_inbox WHERE id=?`, id)
	return scanInbox(row)
}

type rowScanner interface{ Scan(...any) error }

func scanInbox(row rowScanner) (InboxDocument, error) {
	var d InboxDocument
	var ext, urlv, canon, title, lang, ctype, text, sha, source, base, cats, meta, status, matched, proactiveState, materializedNodeID string
	var pub, disc, recv, upd int64
	var proactiveAttempts int
	var rel float64
	err := row.Scan(&d.ID, &d.AgentID, &d.TaskID, &ext, &urlv, &canon, &title, &pub, &disc, &recv, &upd, &lang, &ctype, &text, &sha, &source, &base, &cats, &meta, &status, &rel, &matched, &proactiveState, &proactiveAttempts, &materializedNodeID)
	if err != nil {
		return d, err
	}
	d.Document = Document{ExternalID: ext, URL: urlv, CanonicalURL: canon, Title: title, PublishedAt: nsTime(pub), DiscoveredAt: nsTime(disc), Language: lang, ContentType: ctype, Text: text, ContentSHA256: sha, SourceName: source, SourceBaseURL: base}
	_ = json.Unmarshal([]byte(cats), &d.Document.Categories)
	_ = json.Unmarshal([]byte(meta), &d.Metadata)
	d.Status = status
	d.Relevance = rel
	d.MatchedNodeID = matched
	d.ProactiveState = proactiveState
	d.ProactiveAttempts = proactiveAttempts
	d.MaterializedNodeID = materializedNodeID
	d.ReceivedAt = nsTime(recv)
	d.UpdatedAt = nsTime(upd)
	return d, nil
}

func (s *Store) ListInbox(ctx context.Context, status string, limit int) ([]InboxDocument, error) {
	if limit < 1 {
		limit = 100
	}
	if limit > 1000 {
		limit = 1000
	}
	rows, err := s.db.QueryContext(ctx, `SELECT id,agent_id,task_id,external_id,url,canonical_url,title,published_at_ns,discovered_at_ns,received_at_ns,updated_at_ns,language,content_type,text_content,content_sha256,source_name,source_base_url,categories_json,metadata_json,status,relevance,matched_node_id,proactive_state,proactive_attempts,materialized_node_id FROM source_inbox WHERE (?='' OR status=?) ORDER BY received_at_ns DESC LIMIT ?`, status, status, limit)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]InboxDocument, 0)
	for rows.Next() {
		d, err := scanInbox(rows)
		if err != nil {
			return nil, err
		}
		out = append(out, d)
	}
	return out, rows.Err()
}

func (s *Store) Stats(ctx context.Context) (InboxStats, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT status,COUNT(*) FROM source_inbox GROUP BY status`)
	if err != nil {
		return InboxStats{}, err
	}
	defer rows.Close()
	var st InboxStats
	for rows.Next() {
		var status string
		var n int
		if err := rows.Scan(&status, &n); err != nil {
			return st, err
		}
		st.Total += n
		switch status {
		case "received":
			st.Received = n
		case "processing":
			st.Processing = n
		case "candidate":
			st.Candidate = n
		case "archived":
			st.Archived = n
		case "materialized":
			st.Materialized = n
		case "used":
			st.Used = n
		}
	}
	if err := rows.Err(); err != nil {
		return st, err
	}
	proactiveRows, err := s.db.QueryContext(ctx, `SELECT proactive_state,COUNT(*) FROM source_inbox WHERE proactive_state<>'' GROUP BY proactive_state`)
	if err != nil {
		return st, err
	}
	defer proactiveRows.Close()
	for proactiveRows.Next() {
		var state string
		var n int
		if err := proactiveRows.Scan(&state, &n); err != nil {
			return st, err
		}
		switch state {
		case "queued":
			st.SecurityQueued = n
		case "processing":
			st.SecurityProcessing = n
		case "rejected":
			st.SecurityRejected = n
		}
	}
	return st, proactiveRows.Err()
}

func (s *Store) SecurityLifecycles(ctx context.Context, since time.Time) ([]SecurityLifecycle, error) {
	rows, err := s.db.QueryContext(ctx, `SELECT id,title,status,proactive_state,materialized_node_id,metadata_json FROM source_inbox WHERE proactive_state<>'' AND updated_at_ns>=? ORDER BY updated_at_ns DESC`, since.UTC().UnixNano())
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	out := make([]SecurityLifecycle, 0)
	for rows.Next() {
		var record SecurityLifecycle
		var metadataJSON string
		if err := rows.Scan(&record.InboxID, &record.Title, &record.Status, &record.ProactiveState, &record.MaterializedNodeID, &metadataJSON); err != nil {
			return nil, err
		}
		meta := map[string]any{}
		_ = json.Unmarshal([]byte(metadataJSON), &meta)
		record.RunID = metadataStringValue(meta["proactive_run_id"])
		record.StartedAt = metadataTimeValue(meta["proactive_started_at"])
		record.CompletedAt = metadataTimeValue(meta["proactive_completed_at"])
		record.DurationMS = int64(metadataFloatValue(meta["proactive_duration_ms"]))
		record.Outcome = metadataStringValue(meta["proactive_outcome"])
		record.LastError = metadataStringValue(meta["proactive_last_error"])
		record.Confidence = metadataFloatValue(meta["security_confidence"])
		record.Severity = metadataStringValue(meta["security_severity"])
		record.EventType = metadataStringValue(meta["security_event_type"])
		record.NodesCreated = uint64(metadataFloatValue(meta["run_nodes_created"]))
		record.NodesUpdated = uint64(metadataFloatValue(meta["run_nodes_updated"]))
		record.NodesDeleted = uint64(metadataFloatValue(meta["run_nodes_deleted"]))
		record.EdgesCreated = uint64(metadataFloatValue(meta["run_edges_created"]))
		record.EdgesUpdated = uint64(metadataFloatValue(meta["run_edges_updated"]))
		record.EdgesDeleted = uint64(metadataFloatValue(meta["run_edges_deleted"]))
		record.VectorsCreated = uint64(metadataFloatValue(meta["run_vectors_created"]))
		record.VectorsUpdated = uint64(metadataFloatValue(meta["run_vectors_updated"]))
		record.VectorsDeleted = uint64(metadataFloatValue(meta["run_vectors_deleted"]))
		out = append(out, record)
	}
	return out, rows.Err()
}

func (s *Store) SearchCandidates(ctx context.Context, query string, limit int, maxAge time.Duration) ([]ScoredDocument, error) {
	if limit < 1 {
		limit = 3
	}
	if limit > 50 {
		limit = 50
	}
	since := int64(0)
	if maxAge > 0 {
		since = time.Now().UTC().Add(-maxAge).UnixNano()
	}
	rows, err := s.db.QueryContext(ctx, `SELECT id,agent_id,task_id,external_id,url,canonical_url,title,published_at_ns,discovered_at_ns,received_at_ns,updated_at_ns,language,content_type,text_content,content_sha256,source_name,source_base_url,categories_json,metadata_json,status,relevance,matched_node_id,proactive_state,proactive_attempts,materialized_node_id,signature FROM source_inbox WHERE status IN ('candidate','materialized','used') AND (?=0 OR COALESCE(NULLIF(published_at_ns,0),received_at_ns)>=?) ORDER BY updated_at_ns DESC LIMIT 5000`, since, since)
	if err != nil {
		return nil, err
	}
	defer rows.Close()
	qsig := simhash64(query)
	qterms := termSet(query)
	var out []ScoredDocument
	for rows.Next() {
		var d InboxDocument
		var ext, urlv, canon, title, lang, ctype, text, sha, source, base, cats, meta, status, matched, proactiveState, materializedNodeID string
		var pub, disc, recv, upd, sig int64
		var proactiveAttempts int
		var rel float64
		if err := rows.Scan(&d.ID, &d.AgentID, &d.TaskID, &ext, &urlv, &canon, &title, &pub, &disc, &recv, &upd, &lang, &ctype, &text, &sha, &source, &base, &cats, &meta, &status, &rel, &matched, &proactiveState, &proactiveAttempts, &materializedNodeID, &sig); err != nil {
			return nil, err
		}
		d.Document = Document{ExternalID: ext, URL: urlv, CanonicalURL: canon, Title: title, PublishedAt: nsTime(pub), DiscoveredAt: nsTime(disc), Language: lang, ContentType: ctype, Text: text, ContentSHA256: sha, SourceName: source, SourceBaseURL: base}
		_ = json.Unmarshal([]byte(cats), &d.Document.Categories)
		_ = json.Unmarshal([]byte(meta), &d.Metadata)
		d.Status = status
		d.Relevance = rel
		d.MatchedNodeID = matched
		d.ProactiveState = proactiveState
		d.ProactiveAttempts = proactiveAttempts
		d.MaterializedNodeID = materializedNodeID
		d.ReceivedAt = nsTime(recv)
		d.UpdatedAt = nsTime(upd)
		hamming := bits.OnesCount64(qsig ^ uint64(sig))
		hashScore := 1 - float64(hamming)/64.0
		lex := jaccard(qterms, termSet(title+" "+text))
		score := 0.45*hashScore + 0.45*lex + 0.10*rel
		if lex < 0.03 && hashScore < 0.58 {
			continue
		}
		out = append(out, ScoredDocument{InboxDocument: d, QueryScore: score})
	}
	sort.Slice(out, func(i, j int) bool { return out[i].QueryScore > out[j].QueryScore })
	if len(out) > limit {
		out = out[:limit]
	}
	return out, nil
}

func boolInt(v bool) int {
	if v {
		return 1
	}
	return 0
}
func timeNS(t time.Time) int64 {
	if t.IsZero() {
		return 0
	}
	return t.UTC().UnixNano()
}
func nsTime(v int64) time.Time {
	if v <= 0 {
		return time.Time{}
	}
	return time.Unix(0, v).UTC()
}
func uniqueStrings(values []string) []string {
	seen := map[string]bool{}
	out := make([]string, 0, len(values))
	for _, v := range values {
		v = strings.TrimSpace(v)
		if v == "" || seen[v] {
			continue
		}
		seen[v] = true
		out = append(out, v)
	}
	return out
}
func termSet(value string) map[string]bool {
	out := map[string]bool{}
	for _, r := range strings.FieldsFunc(strings.ToLower(value), func(r rune) bool { return !unicode.IsLetter(r) && !unicode.IsDigit(r) }) {
		if len([]rune(r)) >= 3 {
			out[r] = true
		}
	}
	return out
}
func jaccard(a, b map[string]bool) float64 {
	if len(a) == 0 || len(b) == 0 {
		return 0
	}
	inter := 0
	union := len(a)
	for k := range b {
		if a[k] {
			inter++
		} else {
			union++
		}
	}
	if union == 0 {
		return 0
	}
	return float64(inter) / float64(union)
}
func simhash64(value string) uint64 {
	weights := [64]int{}
	for token := range termSet(value) {
		h := fnv.New64a()
		_, _ = h.Write([]byte(token))
		x := h.Sum64()
		for i := 0; i < 64; i++ {
			if x&(1<<i) != 0 {
				weights[i]++
			} else {
				weights[i]--
			}
		}
	}
	var sig uint64
	for i, w := range weights {
		if w >= 0 {
			sig |= 1 << i
		}
	}
	return sig
}

func (s *Store) ReleaseInbox(ctx context.Context, id, message string) error {
	current, err := s.getInbox(ctx, id)
	if err != nil {
		return err
	}
	meta := make(map[string]any, len(current.Metadata)+1)
	for k, v := range current.Metadata {
		meta[k] = v
	}
	meta["classification_error"] = strings.TrimSpace(message)
	data, _ := json.Marshal(meta)
	res, err := s.db.ExecContext(ctx, `UPDATE source_inbox SET status='received',metadata_json=?,updated_at_ns=? WHERE id=? AND status='processing'`, string(data), time.Now().UTC().UnixNano(), id)
	return requireOneInboxRow(res, err, id, "release inbox classification")
}