RC-1

internal/sourceagent/agent.go

@@ -49,6 +49,24 @@ type Runner struct {
 	mu         sync.RWMutex
 	remote     RemoteConfig
 	wake       chan struct{}
+	diagMu     sync.RWMutex
+	diag       runnerDiagnostics
+}
+
+type runnerDiagnostics struct {
+	StartedAt             time.Time
+	ConfigSource          string
+	CacheLoadedAt         time.Time
+	LastConfigAttemptAt   time.Time
+	LastConfigSuccessAt   time.Time
+	LastConfigError       string
+	LastHeartbeatAt       time.Time
+	LastHeartbeatSuccess  time.Time
+	LastHeartbeatError    string
+	LastConnectionErrorAt time.Time
+	LastConnectionError   string
+	LastTaskRunAt         time.Time
+	LastTaskError         string
 }
 
 type bootstrapConfig struct {
@@ -109,13 +127,15 @@ func NewRunner(cfg RunnerConfig) (*Runner, error) {
 	if err != nil {
 		return nil, err
 	}
-	return &Runner{
+	r := &Runner{
 		cfg:        cfg,
 		http:       &http.Client{Timeout: cfg.HTTPTimeout},
 		sourceHTTP: research.NewSafeHTTPClient(cfg.AllowPrivate, cfg.HTTPTimeout),
 		state:      state,
 		wake:       make(chan struct{}, 1),
-	}, nil
+	}
+	r.diag.StartedAt = time.Now().UTC()
+	return r, nil
 }
 func (r *Runner) Close() error {
 	if r == nil || r.state == nil {
@@ -133,7 +153,56 @@ func (r *Runner) Status() map[string]any {
 	r.mu.RLock()
 	remote := r.remote
 	r.mu.RUnlock()
-	return map[string]any{"ok": true, "mode": "agent", "agent_id": r.cfg.AgentID, "brain_url": r.cfg.BrainURL, "configured_tasks": len(remote.Tasks), "config_issued_at": remote.IssuedAt, "version": r.cfg.Version}
+	r.diagMu.RLock()
+	d := r.diag
+	r.diagMu.RUnlock()
+	loopbackURL := false
+	if u, err := url.Parse(r.cfg.BrainURL); err == nil {
+		host := strings.ToLower(u.Hostname())
+		loopbackURL = host == "localhost" || host == "127.0.0.1" || host == "::1"
+	}
+	lastSuccess := d.LastConfigSuccessAt
+	if d.LastHeartbeatSuccess.After(lastSuccess) {
+		lastSuccess = d.LastHeartbeatSuccess
+	}
+	connected := !lastSuccess.IsZero() && (d.LastConnectionErrorAt.IsZero() || !d.LastConnectionErrorAt.After(lastSuccess))
+	state := "never_connected"
+	if connected {
+		state = "connected"
+		if d.LastConfigError != "" || d.LastHeartbeatError != "" {
+			state = "degraded"
+		}
+	} else if !lastSuccess.IsZero() {
+		state = "disconnected"
+	} else if d.ConfigSource == "cache" {
+		state = "cached_config_only"
+	}
+	warning := ""
+	// Loopback is perfectly valid for a native same-host test. Only surface the
+	// Docker-specific warning when the Agent is not actually connected.
+	if loopbackURL && !connected {
+		warning = "Brain URL points to loopback. In separate Docker containers, localhost/127.0.0.1 means the Agent container, not the Brain container. Use a shared service name or host.docker.internal:<published-port>."
+	}
+	hint := ""
+	errText := strings.ToLower(d.LastConnectionError)
+	switch {
+	case strings.Contains(errText, "http 401") || strings.Contains(errText, "unauthorized"):
+		hint = "The Brain rejected this token. The Agent may no longer be registered in that Brain, may be disabled, or the token was rotated. Create/verify the Agent in the Brain UI and update BRAIN_AGENT_TOKEN."
+	case strings.Contains(errText, "http 404"):
+		hint = "The configured URL does not expose the Brain Agent API. Verify BRAIN_AGENT_BRAIN_URL and the published Brain port."
+	case warning != "":
+		hint = "For separate Docker containers, do not use 127.0.0.1/localhost as the Brain URL."
+	}
+	return map[string]any{
+		"ok": true, "mode": "agent", "agent_id": r.cfg.AgentID, "brain_url": r.cfg.BrainURL,
+		"brain_connected": connected, "connection_state": state, "brain_url_warning": warning, "connection_hint": hint,
+		"configured_tasks": len(remote.Tasks), "config_source": d.ConfigSource, "cache_loaded_at": d.CacheLoadedAt,
+		"config_issued_at": remote.IssuedAt, "version": r.cfg.Version, "started_at": d.StartedAt,
+		"last_config_attempt_at": d.LastConfigAttemptAt, "last_config_success_at": d.LastConfigSuccessAt, "last_config_error": d.LastConfigError,
+		"last_heartbeat_at": d.LastHeartbeatAt, "last_heartbeat_success_at": d.LastHeartbeatSuccess, "last_heartbeat_error": d.LastHeartbeatError,
+		"last_connection_error_at": d.LastConnectionErrorAt, "last_connection_error": d.LastConnectionError,
+		"last_task_run_at": d.LastTaskRunAt, "last_task_error": d.LastTaskError,
+	}
 }
 
 func (r *Runner) loop(ctx context.Context) {
@@ -141,14 +210,32 @@ func (r *Runner) loop(ctx context.Context) {
 	defer refresh.Stop()
 	run := time.NewTicker(30 * time.Second)
 	defer run.Stop()
-	_ = r.refreshConfig(ctx)
+	heartbeat := time.NewTicker(time.Minute)
+	defer heartbeat.Stop()
+
+	// Heartbeat independently of config loading. This lets the Brain see the
+	// Agent even when its task configuration is temporarily broken.
+	if err := r.sendHeartbeat(ctx, Heartbeat{AgentID: r.cfg.AgentID, Version: r.cfg.Version, Status: "starting"}); err != nil {
+		slog.Warn("source agent initial heartbeat failed", "brain_url", r.cfg.BrainURL, "error", err)
+	}
+	if err := r.refreshConfig(ctx); err != nil {
+		slog.Warn("source agent config refresh failed", "brain_url", r.cfg.BrainURL, "error", err)
+	} else {
+		slog.Info("source agent connected to brain", "brain_url", r.cfg.BrainURL, "tasks", len(r.currentTasks()))
+	}
 	r.runDue(ctx)
 	for {
 		select {
 		case <-ctx.Done():
 			return
 		case <-refresh.C:
-			_ = r.refreshConfig(ctx)
+			if err := r.refreshConfig(ctx); err != nil {
+				slog.Warn("source agent config refresh failed", "brain_url", r.cfg.BrainURL, "error", err)
+			}
+		case <-heartbeat.C:
+			if err := r.sendHeartbeat(ctx, Heartbeat{AgentID: r.cfg.AgentID, Version: r.cfg.Version, Status: "online", Metadata: map[string]any{"configured_tasks": len(r.currentTasks())}}); err != nil {
+				slog.Warn("source agent heartbeat failed", "brain_url", r.cfg.BrainURL, "error", err)
+			}
 		case <-run.C:
 			r.runDue(ctx)
 		case <-r.wake:
@@ -157,35 +244,72 @@ func (r *Runner) loop(ctx context.Context) {
 	}
 }
 
+func (r *Runner) currentTasks() []Task {
+	r.mu.RLock()
+	defer r.mu.RUnlock()
+	return append([]Task(nil), r.remote.Tasks...)
+}
+
 func (r *Runner) refreshConfig(ctx context.Context) error {
+	r.diagMu.Lock()
+	r.diag.LastConfigAttemptAt = time.Now().UTC()
+	r.diagMu.Unlock()
 	req, err := http.NewRequestWithContext(ctx, http.MethodGet, r.cfg.BrainURL+"/api/v1/agent/config", nil)
 	if err != nil {
+		r.recordConfigResult(err)
 		return err
 	}
 	r.auth(req)
 	resp, err := r.http.Do(req)
 	if err != nil {
+		r.recordConfigResult(err)
 		return err
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode/100 != 2 {
-		return fmt.Errorf("brain config returned HTTP %d", resp.StatusCode)
+		b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		err := fmt.Errorf("brain config HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
+		r.recordConfigResult(err)
+		return err
 	}
 	var cfg RemoteConfig
 	if err := json.NewDecoder(io.LimitReader(resp.Body, 4<<20)).Decode(&cfg); err != nil {
+		r.recordConfigResult(err)
 		return err
 	}
 	if cfg.Agent.ID != "" && cfg.Agent.ID != r.cfg.AgentID {
-		return fmt.Errorf("brain returned config for unexpected agent %q", cfg.Agent.ID)
+		err := fmt.Errorf("brain returned config for unexpected agent %q", cfg.Agent.ID)
+		r.recordConfigResult(err)
+		return err
 	}
 	r.mu.Lock()
 	r.remote = cfg
 	r.mu.Unlock()
 	r.saveCachedConfig(cfg)
-	_ = r.sendHeartbeat(ctx, Heartbeat{AgentID: r.cfg.AgentID, Version: r.cfg.Version, Status: "online", Metadata: map[string]any{"configured_tasks": len(cfg.Tasks)}})
+	r.recordConfigResult(nil)
+	if err := r.sendHeartbeat(ctx, Heartbeat{AgentID: r.cfg.AgentID, Version: r.cfg.Version, Status: "online", Metadata: map[string]any{"configured_tasks": len(cfg.Tasks)}}); err != nil {
+		slog.Warn("source agent heartbeat after config refresh failed", "error", err)
+	}
 	return nil
 }
 
+func (r *Runner) recordConfigResult(err error) {
+	r.diagMu.Lock()
+	defer r.diagMu.Unlock()
+	now := time.Now().UTC()
+	if err != nil {
+		r.diag.LastConfigError = err.Error()
+		r.diag.LastConnectionErrorAt = now
+		r.diag.LastConnectionError = err.Error()
+		return
+	}
+	r.diag.LastConfigSuccessAt = now
+	r.diag.LastConfigError = ""
+	r.diag.ConfigSource = "brain"
+	r.diag.LastConnectionErrorAt = time.Time{}
+	r.diag.LastConnectionError = ""
+}
+
 func (r *Runner) runDue(ctx context.Context) {
 	r.mu.RLock()
 	tasks := append([]Task(nil), r.remote.Tasks...)
@@ -217,6 +341,9 @@ func (r *Runner) runDue(ctx context.Context) {
 
 func (r *Runner) runTask(ctx context.Context, task Task) {
 	started := time.Now().UTC()
+	r.diagMu.Lock()
+	r.diag.LastTaskRunAt = started
+	r.diagMu.Unlock()
 	docs, err := r.pollTask(ctx, task)
 	if err == nil && len(docs) > 0 {
 		for start := 0; start < len(docs); start += r.cfg.BatchSize {
@@ -234,6 +361,13 @@ func (r *Runner) runTask(ctx context.Context, task Task) {
 		}
 	}
 	_ = r.state.FinishTask(ctx, task.ID, started, err)
+	r.diagMu.Lock()
+	if err != nil {
+		r.diag.LastTaskError = err.Error()
+	} else {
+		r.diag.LastTaskError = ""
+	}
+	r.diagMu.Unlock()
 	h := Heartbeat{AgentID: r.cfg.AgentID, Version: r.cfg.Version, Status: "ok", LastRunAt: started, TasksChecked: 1, Documents: len(docs)}
 	if err != nil {
 		h.Status = "error"
@@ -516,23 +650,48 @@ func (r *Runner) sendBatch(ctx context.Context, taskID string, docs []Document)
 	return nil
 }
 func (r *Runner) sendHeartbeat(ctx context.Context, h Heartbeat) error {
+	r.diagMu.Lock()
+	r.diag.LastHeartbeatAt = time.Now().UTC()
+	r.diagMu.Unlock()
 	data, _ := json.Marshal(h)
 	req, err := http.NewRequestWithContext(ctx, http.MethodPost, r.cfg.BrainURL+"/api/v1/agent/heartbeat", bytes.NewReader(data))
 	if err != nil {
+		r.recordHeartbeatResult(err)
 		return err
 	}
 	req.Header.Set("Content-Type", "application/json")
 	r.auth(req)
 	resp, err := r.http.Do(req)
 	if err != nil {
+		r.recordHeartbeatResult(err)
 		return err
 	}
 	defer resp.Body.Close()
 	if resp.StatusCode/100 != 2 {
-		return fmt.Errorf("heartbeat HTTP %d", resp.StatusCode)
+		b, _ := io.ReadAll(io.LimitReader(resp.Body, 4096))
+		err := fmt.Errorf("heartbeat HTTP %d: %s", resp.StatusCode, strings.TrimSpace(string(b)))
+		r.recordHeartbeatResult(err)
+		return err
 	}
+	r.recordHeartbeatResult(nil)
 	return nil
 }
+
+func (r *Runner) recordHeartbeatResult(err error) {
+	r.diagMu.Lock()
+	defer r.diagMu.Unlock()
+	now := time.Now().UTC()
+	if err != nil {
+		r.diag.LastHeartbeatError = err.Error()
+		r.diag.LastConnectionErrorAt = now
+		r.diag.LastConnectionError = err.Error()
+		return
+	}
+	r.diag.LastHeartbeatSuccess = now
+	r.diag.LastHeartbeatError = ""
+	r.diag.LastConnectionErrorAt = time.Time{}
+	r.diag.LastConnectionError = ""
+}
 func (r *Runner) auth(req *http.Request) {
 	req.Header.Set("Authorization", "Bearer "+r.cfg.Token)
 	req.Header.Set("X-Brain-Agent-ID", r.cfg.AgentID)
@@ -597,6 +756,12 @@ func (r *Runner) loadCachedConfig() {
 	r.mu.Lock()
 	r.remote = cfg
 	r.mu.Unlock()
+	r.diagMu.Lock()
+	if r.diag.ConfigSource == "" {
+		r.diag.ConfigSource = "cache"
+	}
+	r.diag.CacheLoadedAt = time.Now().UTC()
+	r.diagMu.Unlock()
 }
 
 type localState struct {

internal/sourceagent/sourceagent_test.go

@@ -89,3 +89,90 @@ func TestSitemapIndexCollectsChildURLs(t *testing.T) {
 		t.Fatalf("unexpected sitemap items: %+v", items)
 	}
 }
+
+func TestRunnerStatusDoesNotTreatCachedConfigAsLiveConnection(t *testing.T) {
+	r := &Runner{
+		cfg:    RunnerConfig{BrainURL: "http://127.0.0.1:8091", AgentID: "agent-test", Version: "test"},
+		remote: RemoteConfig{IssuedAt: time.Now().UTC(), Tasks: []Task{{ID: "cached-task"}}},
+		diag:   runnerDiagnostics{ConfigSource: "cache", CacheLoadedAt: time.Now().UTC()},
+	}
+	st := r.Status()
+	if st["brain_connected"] != false {
+		t.Fatalf("cached config must not be reported as live connection: %+v", st)
+	}
+	if st["connection_state"] != "cached_config_only" {
+		t.Fatalf("unexpected state: %+v", st)
+	}
+	if st["brain_url_warning"] == "" {
+		t.Fatalf("expected loopback warning: %+v", st)
+	}
+}
+
+func TestRunnerStatusReportsFreshBrainConnection(t *testing.T) {
+	now := time.Now().UTC()
+	r := &Runner{
+		cfg:    RunnerConfig{BrainURL: "http://brain:8090", AgentID: "agent-test", Version: "test"},
+		remote: RemoteConfig{IssuedAt: now},
+		diag:   runnerDiagnostics{ConfigSource: "brain", LastConfigAttemptAt: now.Add(-time.Second), LastConfigSuccessAt: now},
+	}
+	st := r.Status()
+	if st["brain_connected"] != true || st["connection_state"] != "connected" {
+		t.Fatalf("expected live connection: %+v", st)
+	}
+}
+
+func TestRunnerStatusAcceptsConnectedLoopbackForNativeTest(t *testing.T) {
+	now := time.Now().UTC()
+	r := &Runner{
+		cfg:    RunnerConfig{BrainURL: "http://127.0.0.1:8091", AgentID: "agent-test", Version: "test"},
+		remote: RemoteConfig{IssuedAt: now},
+		diag:   runnerDiagnostics{ConfigSource: "brain", LastConfigSuccessAt: now, LastHeartbeatSuccess: now},
+	}
+	st := r.Status()
+	if st["brain_connected"] != true {
+		t.Fatalf("expected connected loopback to be accepted: %+v", st)
+	}
+	if st["brain_url_warning"] != "" {
+		t.Fatalf("connected native loopback must not be warned as broken: %+v", st)
+	}
+}
+
+func TestEnsureInboxClassifierVersionRequeuesArchivedOnce(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-a", "Agent A")
+	if err != nil {
+		t.Fatal(err)
+	}
+	task, err := store.UpsertTask(ctx, Task{ID: "task-a", AgentID: agent.ID, Name: "Security", 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/a", Title: "Security update", Text: "Dies ist ein ausreichend langer Sicherheitsartikel mit mehreren Details, damit das Dokument in die Source Inbox aufgenommen und klassifiziert werden kann."}})
+	if err != nil {
+		t.Fatal(err)
+	}
+	claimed, err := store.ClaimInbox(ctx, 1)
+	if err != nil || len(claimed) != 1 {
+		t.Fatalf("claim failed: %#v %v", claimed, err)
+	}
+	if err := store.CompleteClassification(ctx, claimed[0].ID, "archived", .4, "kb-1", map[string]any{}); err != nil {
+		t.Fatal(err)
+	}
+	n, err := store.EnsureInboxClassifierVersion(ctx, 2)
+	if err != nil || n != 1 {
+		t.Fatalf("expected one requeued archived item, n=%d err=%v", n, err)
+	}
+	items, err := store.ListInbox(ctx, "received", 10)
+	if err != nil || len(items) != 1 {
+		t.Fatalf("expected requeued received item: %#v %v", items, err)
+	}
+	n, err = store.EnsureInboxClassifierVersion(ctx, 2)
+	if err != nil || n != 0 {
+		t.Fatalf("classifier migration must be one-shot, n=%d err=%v", n, err)
+	}
+}

internal/sourceagent/store.go

@@ -63,6 +63,9 @@ func (s *Store) init(ctx context.Context) error {
 			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,
@@ -85,6 +88,43 @@ func (s *Store) init(ctx context.Context) error {
 	return nil
 }
 
+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 {
@@ -181,7 +221,7 @@ func (s *Store) ListAgents(ctx context.Context) ([]Agent, error) {
 		return nil, err
 	}
 	defer rows.Close()
-	var out []Agent
+	out := make([]Agent, 0)
 	for rows.Next() {
 		var a Agent
 		var en int
@@ -198,6 +238,18 @@ func (s *Store) ListAgents(ctx context.Context) ([]Agent, error) {
 	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")
@@ -267,6 +319,13 @@ func (s *Store) UpsertTask(ctx context.Context, t Task) (Task, error) {
 	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()
@@ -291,7 +350,7 @@ func (s *Store) ListTasks(ctx context.Context, agentID string) ([]Task, error) {
 		return nil, err
 	}
 	defer rows.Close()
-	var out []Task
+	out := make([]Task, 0)
 	for rows.Next() {
 		var t Task
 		var en int
@@ -518,7 +577,7 @@ func (s *Store) ListInbox(ctx context.Context, status string, limit int) ([]Inbo
 		return nil, err
 	}
 	defer rows.Close()
-	var out []InboxDocument
+	out := make([]InboxDocument, 0)
 	for rows.Next() {
 		d, err := scanInbox(rows)
 		if err != nil {