RC-2

internal/sourceagent/agent.go

@@ -596,7 +596,7 @@ func (r *Runner) materializeItems(ctx context.Context, task Task, items []feedIt
 		if u, e := url.Parse(task.URL); e == nil {
 			baseURL = u.Scheme + "://" + u.Host
 		}
-		out = append(out, Document{ExternalID: final, URL: final, CanonicalURL: final, Title: title, PublishedAt: item.Published, DiscoveredAt: time.Now().UTC(), ContentType: ctype, Text: text, ContentSHA256: sha, SourceName: task.Name, SourceBaseURL: baseURL, Categories: task.Categories, Metadata: map[string]any{"agent_task_type": task.Type, "fetch_error_kind": diag.ErrorKind}})
+		out = append(out, Document{ExternalID: final, URL: final, CanonicalURL: final, Title: title, PublishedAt: item.Published, DiscoveredAt: time.Now().UTC(), ContentType: ctype, Text: text, ContentSHA256: sha, SourceName: task.Name, SourceBaseURL: baseURL, Categories: task.Categories, Metadata: map[string]any{"agent_task_type": task.Type, "fetch_error_kind": diag.ErrorKind, "security_proactive": strings.ToLower(strings.TrimSpace(task.Config["security_proactive"]))}})
 		if len(out) >= task.MaxItems {
 			break
 		}

internal/sourceagent/sourceagent_test.go

@@ -176,3 +176,49 @@ func TestEnsureInboxClassifierVersionRequeuesArchivedOnce(t *testing.T) {
 		t.Fatalf("classifier migration must be one-shot, n=%d err=%v", n, err)
 	}
 }
+
+func TestProactiveSecurityInboxLifecycle(t *testing.T) {
+	ctx := context.Background()
+	store, err := OpenStore(t.TempDir())
+	if err != nil {
+		t.Fatal(err)
+	}
+	defer store.Close()
+	agent, _, err := store.CreateAgent(ctx, "agent-security", "Security Agent")
+	if err != nil {
+		t.Fatal(err)
+	}
+	task, err := store.UpsertTask(ctx, Task{ID: "task-security", AgentID: agent.ID, Name: "Security Alerts", Type: "rss", URL: "https://example.org/feed", Enabled: true, PollInterval: "1h", MaxItems: 10})
+	if err != nil {
+		t.Fatal(err)
+	}
+	_, err = store.Ingest(ctx, agent.ID, task.ID, []Document{{URL: "https://example.org/advisory", CanonicalURL: "https://example.org/advisory", Title: "Critical security advisory", Text: "A concrete security advisory with enough source text to be classified and later materialized into a proactive security node.", Metadata: map[string]any{"security_proactive": "true"}}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := store.ClaimInbox(ctx, 1)
+	if err != nil || len(claimed) != 1 {
+		t.Fatalf("claim classification: %#v %v", claimed, err)
+	}
+	if err := store.CompleteClassification(ctx, claimed[0].ID, "candidate", .71, "kb-security", map[string]any{"priority_score": .71, "security_candidate": true}); err != nil {
+		t.Fatal(err)
+	}
+	if err := store.QueueProactiveSecurity(ctx, claimed[0].ID); err != nil {
+		t.Fatal(err)
+	}
+	securityItems, err := store.ClaimProactiveSecurity(ctx, 1)
+	if err != nil || len(securityItems) != 1 || securityItems[0].ProactiveState != "processing" {
+		t.Fatalf("claim proactive: %#v %v", securityItems, err)
+	}
+	if err := store.CompleteProactiveSecurity(ctx, claimed[0].ID, "external-node-1", map[string]any{"security_event_type": "vulnerability"}); err != nil {
+		t.Fatal(err)
+	}
+	materialized, err := store.ListInbox(ctx, "materialized", 10)
+	if err != nil || len(materialized) != 1 || materialized[0].MaterializedNodeID != "external-node-1" || materialized[0].ProactiveState != "done" {
+		t.Fatalf("materialized lifecycle mismatch: %#v %v", materialized, err)
+	}
+	results, err := store.SearchCandidates(ctx, "critical security advisory", 5, 0)
+	if err != nil || len(results) != 1 {
+		t.Fatalf("materialized document must remain searchable before SearXNG: %#v %v", results, err)
+	}
+}

internal/sourceagent/store.go

@@ -84,10 +84,55 @@ func (s *Store) init(ctx context.Context) error {
 			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
+	}
 	_, _ = s.db.ExecContext(ctx, `UPDATE source_inbox SET status='received' WHERE status='processing' AND updated_at_ns<?`, time.Now().UTC().Add(-10*time.Minute).UnixNano())
+	_, _ = s.db.ExecContext(ctx, `UPDATE source_inbox SET proactive_state='queued' WHERE proactive_state='processing' AND updated_at_ns<?`, time.Now().UTC().Add(-20*time.Minute).UnixNano())
 	return nil
 }
 
+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
@@ -515,6 +560,112 @@ func (s *Store) CompleteClassification(ctx context.Context, id, status string, r
 	return err
 }
 
+// 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 {
+	_, 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'`, time.Now().UTC().UnixNano(), id)
+	return err
+}
+
+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()
+	for _, id := range ids {
+		if _, 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); err != nil {
+			return nil, err
+		}
+	}
+	if err := tx.Commit(); err != nil {
+		return nil, err
+	}
+	return s.GetInboxByIDs(ctx, ids)
+}
+
+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
+	}
+	data, _ := json.Marshal(merged)
+	_, 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=?`, nodeID, string(data), time.Now().UTC().UnixNano(), id)
+	return err
+}
+
+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
+	data, _ := json.Marshal(merged)
+	_, err = s.db.ExecContext(ctx, `UPDATE source_inbox SET proactive_state='rejected', proactive_next_at_ns=0, metadata_json=?, updated_at_ns=? WHERE id=?`, string(data), time.Now().UTC().UnixNano(), id)
+	return err
+}
+
+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
+	data, _ := json.Marshal(meta)
+	_, err = s.db.ExecContext(ctx, `UPDATE source_inbox SET proactive_state='queued', proactive_next_at_ns=?, metadata_json=?, updated_at_ns=? WHERE id=?`, time.Now().UTC().Add(delay).UnixNano(), string(data), time.Now().UTC().UnixNano(), id)
+	return err
+}
+
 func (s *Store) MarkUsed(ctx context.Context, canonicalURLs []string) error {
 	now := time.Now().UTC().UnixNano()
 	for _, u := range uniqueStrings(canonicalURLs) {
@@ -539,7 +690,7 @@ func (s *Store) GetInboxByIDs(ctx context.Context, ids []string) ([]InboxDocumen
 	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 FROM source_inbox WHERE id=?`, id)
+	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)
 }
 
@@ -547,10 +698,11 @@ 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 string
+	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)
+	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
 	}
@@ -560,6 +712,9 @@ func scanInbox(row rowScanner) (InboxDocument, error) {
 	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
@@ -572,7 +727,7 @@ func (s *Store) ListInbox(ctx context.Context, status string, limit int) ([]Inbo
 	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 FROM source_inbox WHERE (?='' OR status=?) ORDER BY received_at_ns DESC LIMIT ?`, status, status, limit)
+	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
 	}
@@ -611,11 +766,36 @@ func (s *Store) Stats(ctx context.Context) (InboxStats, error) {
 			st.Candidate = n
 		case "archived":
 			st.Archived = n
+		case "materialized":
+			st.Materialized = n
 		case "used":
 			st.Used = n
 		}
 	}
-	return st, rows.Err()
+	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) SearchCandidates(ctx context.Context, query string, limit int, maxAge time.Duration) ([]ScoredDocument, error) {
@@ -629,7 +809,7 @@ func (s *Store) SearchCandidates(ctx context.Context, query string, limit int, m
 	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,signature FROM source_inbox WHERE status IN ('candidate','used') AND (?=0 OR COALESCE(NULLIF(published_at_ns,0),received_at_ns)>=?) ORDER BY updated_at_ns DESC LIMIT 5000`, since, since)
+	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
 	}
@@ -639,10 +819,11 @@ func (s *Store) SearchCandidates(ctx context.Context, query string, limit int, m
 	var out []ScoredDocument
 	for rows.Next() {
 		var d InboxDocument
-		var ext, urlv, canon, title, lang, ctype, text, sha, source, base, cats, meta, status, matched string
+		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, &sig); err != nil {
+		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}
@@ -651,6 +832,9 @@ func (s *Store) SearchCandidates(ctx context.Context, query string, limit int, m
 		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))

internal/sourceagent/types.go

@@ -81,25 +81,32 @@ type Heartbeat struct {
 }
 
 type InboxDocument struct {
-	ID            string         `json:"id"`
-	AgentID       string         `json:"agent_id"`
-	TaskID        string         `json:"task_id"`
-	Document      Document       `json:"document"`
-	Status        string         `json:"status"`
-	Relevance     float64        `json:"relevance,omitempty"`
-	MatchedNodeID string         `json:"matched_node_id,omitempty"`
-	ReceivedAt    time.Time      `json:"received_at"`
-	UpdatedAt     time.Time      `json:"updated_at"`
-	Metadata      map[string]any `json:"metadata,omitempty"`
+	ID                 string         `json:"id"`
+	AgentID            string         `json:"agent_id"`
+	TaskID             string         `json:"task_id"`
+	Document           Document       `json:"document"`
+	Status             string         `json:"status"`
+	Relevance          float64        `json:"relevance,omitempty"`
+	MatchedNodeID      string         `json:"matched_node_id,omitempty"`
+	ProactiveState     string         `json:"proactive_state,omitempty"`
+	ProactiveAttempts  int            `json:"proactive_attempts,omitempty"`
+	MaterializedNodeID string         `json:"materialized_node_id,omitempty"`
+	ReceivedAt         time.Time      `json:"received_at"`
+	UpdatedAt          time.Time      `json:"updated_at"`
+	Metadata           map[string]any `json:"metadata,omitempty"`
 }
 
 type InboxStats struct {
-	Total      int `json:"total"`
-	Received   int `json:"received"`
-	Processing int `json:"processing"`
-	Candidate  int `json:"candidate"`
-	Archived   int `json:"archived"`
-	Used       int `json:"used"`
+	Total              int `json:"total"`
+	Received           int `json:"received"`
+	Processing         int `json:"processing"`
+	Candidate          int `json:"candidate"`
+	Archived           int `json:"archived"`
+	Materialized       int `json:"materialized"`
+	Used               int `json:"used"`
+	SecurityQueued     int `json:"security_queued"`
+	SecurityProcessing int `json:"security_processing"`
+	SecurityRejected   int `json:"security_rejected"`
 }
 
 type ScoredDocument struct {