Reset source inactivity history on new admissions

docs/configuration.md

@@ -91,7 +91,11 @@ toward this guard; failed, skipped, truncated, or candidate-cap-affected passes
 are excluded. Sources at fallback priority 10 or better, sources with a positive
 observed probe outcome, and sources whose learned probe yield clears the
 configured minimum are protected. The one-in-ten exploration allowance remains
-in effect for cooled feeds.
+in effect for cooled feeds. Any actual new candidate insertion clears that
+source's zero-admission streak immediately, including an insertion made before
+a later partial, truncated, or capped pass ends. This immediate reset does not
+count as a completed observation or update the last full-refresh admission
+fields; duplicate or tombstone-blocked records do not reset the streak.
 
 The setting is off by default and has no environment override. Quality hashes
 expire after 30 days, and only the active configured catalog is attributed.

docs/redis-schema.md

@@ -50,13 +50,20 @@ The same hash may also hold bounded measurement counters:
 
 - `admitted` accumulates candidate IDs newly inserted into the tracked candidate
   set by atomic upsert, not rows staged by a feed. It can include real insertions
-  from a refresh that later fails.
-- `admission_refreshes`, `last_admitted`, `zero_admission_streak`, and
-  `last_admission_ms` describe completed whole-refresh observations. They are
+  from a refresh that later fails. Any such positive insertion immediately
+  resets `zero_admission_streak`, even if the containing pass later fails or is
+  truncated or capped.
+- `admission_refreshes`, `last_admitted`, and `last_admission_ms` describe
+  completed whole-refresh observations. They are
   updated only when a source participated in a successful, uncapped and
   untruncated refresh pass. Failed, skipped, truncated, or candidate-cap-affected
   source passes are excluded. `zero_admission_streak` counts consecutive
-  qualifying observations with zero new candidate IDs.
+  qualifying observations with zero new candidate IDs since the latest
+  positive admission. A partial positive
+  insertion resets that streak without incrementing `admission_refreshes` or
+  changing `last_admitted`/`last_admission_ms`; those fields still describe only
+  completed full-refresh observations. Duplicate and tombstone-blocked upserts
+  do not reset the streak.
 - `http_attempts`, `network_bytes`, `fetch_duration_ms`, and
   `fetch_cost_observations` accumulate source-fetch measurements. Attempts
   count HTTP client calls; bytes count consumed HTTP response-body reader bytes,

internal/monitor/source_efficiency_integration_test.go

@@ -73,6 +73,48 @@ func TestSourceAdmissionCooldownUsesStoredFullRefreshObservations(t *testing.T)
 	}
 }
 
+func TestPartialPositiveAdmissionClearsCooldownWithoutFullRefreshObservation(t *testing.T) {
+	ctx := context.Background()
+	now := time.Date(2026, 10, 14, 12, 0, 0, 0, time.UTC)
+	runner, _, clock := newSourceEfficiencyTestRunner(t, []SourceSpec{{URL: "file:///synthetic/partial-admission", Scheme: "http"}})
+	*clock = now
+	runner.config.SourceLearningZeroAdmissionCooldown = time.Hour
+	runner.config.SourceLearningZeroAdmissionRefreshes = 3
+	id := runner.sourceID(runner.config.SourceSpecs[0])
+	if err := runner.store.SetActiveLearningSources(ctx, []string{id}); err != nil {
+		t.Fatal(err)
+	}
+	for i := 1; i <= 3; i++ {
+		if err := runner.store.RecordSourceAdmissionRefresh(ctx, map[string]int64{id: 0}, now.Add(time.Duration(i)*time.Minute)); err != nil {
+			t.Fatal(err)
+		}
+	}
+
+	quality, err := runner.store.LoadSourceQuality(ctx, []string{id}, now.Add(3*time.Minute))
+	if err != nil || quality[id].ZeroAdmissionStreak != 3 || quality[id].AdmissionRefreshes != 3 {
+		t.Fatalf("pre-upsert quality=%+v err=%v", quality[id], err)
+	}
+	// This direct source-aware upsert models a partial/capped pass: it commits
+	// one genuine SADD but deliberately records no completed-refresh observation.
+	counts, err := runner.store.UpsertPrioritizedSourceCounts(ctx, []string{"http://partial-new.example:8080"}, now.Add(4*time.Minute), id)
+	if err != nil || counts.Added != 1 {
+		t.Fatalf("partial admission=%+v err=%v", counts, err)
+	}
+	quality, err = runner.store.LoadSourceQuality(ctx, []string{id}, now.Add(4*time.Minute))
+	if err != nil {
+		t.Fatal(err)
+	}
+	if quality[id].Admitted != 1 || quality[id].ZeroAdmissionStreak != 0 || quality[id].AdmissionRefreshes != 3 || quality[id].LastAdmitted != 0 || !quality[id].LastAdmissionAt.Equal(now.Add(3*time.Minute)) {
+		t.Fatalf("partial admission changed full-observation fields: %+v", quality[id])
+	}
+
+	config := SourceLearningConfig{Enabled: true, MinSamples: 8, ZeroAdmissionCooldown: time.Hour, ZeroAdmissionRefreshes: 3, ExplorationInterval: 10}
+	plan := BuildSourceLearningPlan([]SourceLearningCandidate{{ID: id, FallbackPriority: 20, Quality: quality[id]}}, 1, now.Add(4*time.Minute), config)
+	if len(plan) != 1 || plan[0].AdmissionCooled {
+		t.Fatalf("source with a partial positive admission remained cooled: %+v", plan)
+	}
+}
+
 func TestCappedRefreshDoesNotRecordAdmissionObservation(t *testing.T) {
 	runner, mini, _ := capRunner(t, `, "max_candidates": 1`, feedLines("cap", 12))
 	runner.config.MaxCandidates = 0

internal/store/redis.go

@@ -401,14 +401,16 @@ local last_pair = #ARGV
 local source_id = nil
 if #ARGV % 2 == 0 then source_id = ARGV[#ARGV]; last_pair = #ARGV - 1 end
 local quality_key = nil
+local quality_hash_safe = false
 local record_admission = false
 if source_id then
   quality_key = KEYS[7]
-  -- Efficiency telemetry must never prevent candidate imports. Only enable
-  -- its increment when the existing hash field is a safe integer with room
-  -- for every possible insertion in this bounded batch.
+  -- Efficiency telemetry must never prevent candidate imports. A safe hash
+  -- can have its zero-admission streak reset after a real insertion even when
+  -- its cumulative admitted counter is malformed or full.
   local current_raw = redis.pcall('HGET', quality_key, 'admitted')
   if not (type(current_raw) == 'table' and current_raw.err) then
+    quality_hash_safe = true
     if current_raw == false then current_raw = '0' end
     local current = tonumber(current_raw)
     local max_batch = math.floor((last_pair - 1) / 2)
@@ -455,8 +457,14 @@ for i = 2, last_pair, 2 do
     end
   end
 end
-if source_id and record_admission and source_admitted > 0 then
-  redis.call('HINCRBY', quality_key, 'admitted', tostring(source_admitted))
+if source_id and quality_hash_safe and source_admitted > 0 then
+  -- A partial or capped refresh may still have admitted genuinely new IDs.
+  -- Clear stale zero-yield history immediately, but leave full-refresh
+  -- observation fields to RecordSourceAdmissionRefresh.
+  redis.call('HSET', quality_key, 'zero_admission_streak', '0')
+  if record_admission then
+    redis.call('HINCRBY', quality_key, 'admitted', tostring(source_admitted))
+  end
   redis.call('PEXPIRE', quality_key, '2592000000')
 end
 return {added, skipped}

internal/store/source_efficiency_test.go

@@ -118,6 +118,69 @@ func TestSourceAdmissionCountsOnlyActualInsertionsAndTracksCompletedRefreshes(t
 	}
 }
 
+func TestPartialPositiveUpsertResetsStreakWithoutRecordingFullRefresh(t *testing.T) {
+	ctx := context.Background()
+	_, s := newTestStore(t, "source-admission-partial-positive:v1")
+	now := time.Unix(1_800_000_000, 0)
+	id := StableSourceID("https://feeds.example/partial-positive", "http")
+	if err := s.SetActiveLearningSources(ctx, []string{id}); err != nil {
+		t.Fatal(err)
+	}
+	for i := 1; i <= 3; i++ {
+		if err := s.RecordSourceAdmissionRefresh(ctx, map[string]int64{id: 0}, now.Add(time.Duration(i)*time.Minute)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	quality, err := s.LoadSourceQuality(ctx, []string{id}, now.Add(3*time.Minute))
+	if err != nil || quality[id].ZeroAdmissionStreak != 3 || quality[id].AdmissionRefreshes != 3 {
+		t.Fatalf("pre-upsert admission state=%+v err=%v", quality[id], err)
+	}
+
+	counts, err := s.UpsertPrioritizedSourceCounts(ctx, []string{"http://new-source.example:8080"}, now.Add(4*time.Minute), id)
+	if err != nil || counts.Added != 1 {
+		t.Fatalf("partial positive upsert=%+v err=%v", counts, err)
+	}
+	quality, err = s.LoadSourceQuality(ctx, []string{id}, now.Add(4*time.Minute))
+	if err != nil {
+		t.Fatal(err)
+	}
+	got := quality[id]
+	if got.Admitted != 1 || got.ZeroAdmissionStreak != 0 || got.AdmissionRefreshes != 3 || got.LastAdmitted != 0 || !got.LastAdmissionAt.Equal(now.Add(3*time.Minute)) {
+		t.Fatalf("partial upsert should reset only zero history, got %+v", got)
+	}
+}
+
+func TestDuplicateAndTombstoneOnlyUpsertsDoNotResetZeroAdmissionStreak(t *testing.T) {
+	ctx := context.Background()
+	mini, s := newTestStore(t, "source-admission-nonpositive-upsert:v1")
+	now := time.Unix(1_800_000_000, 0)
+	id := StableSourceID("https://feeds.example/nonpositive", "http")
+	if err := s.SetActiveLearningSources(ctx, []string{id}); err != nil {
+		t.Fatal(err)
+	}
+	existing := "http://already-present.example:8080"
+	if _, err := s.UpsertPrioritizedSourceCounts(ctx, []string{existing}, now, id); err != nil {
+		t.Fatal(err)
+	}
+	for i := 1; i <= 3; i++ {
+		if err := s.RecordSourceAdmissionRefresh(ctx, map[string]int64{id: 0}, now.Add(time.Duration(i)*time.Minute)); err != nil {
+			t.Fatal(err)
+		}
+	}
+	blocked := "http://tombstoned.example:8080"
+	if err := mini.Set(s.tombstoneKey(proxyID(blocked)), fmt.Sprintf("1:%d", now.Add(time.Hour).UnixMilli())); err != nil {
+		t.Fatal(err)
+	}
+	counts, err := s.UpsertPrioritizedSourceCounts(ctx, []string{existing, blocked}, now.Add(4*time.Minute), id)
+	if err != nil || counts.Added != 0 || counts.Tombstoned != 1 {
+		t.Fatalf("duplicate/tombstoned upsert=%+v err=%v", counts, err)
+	}
+	quality, err := s.LoadSourceQuality(ctx, []string{id}, now.Add(4*time.Minute))
+	if err != nil || quality[id].ZeroAdmissionStreak != 3 || quality[id].AdmissionRefreshes != 3 || quality[id].LastAdmitted != 0 || !quality[id].LastAdmissionAt.Equal(now.Add(3*time.Minute)) {
+		t.Fatalf("nonpositive upsert changed completed history: quality=%+v err=%v", quality[id], err)
+	}
+}
+
 func TestCorruptAdmissionTelemetryNeverBlocksCandidateImports(t *testing.T) {
 	ctx := context.Background()
 	mini, s := newTestStore(t, "source-admission-corrupt:v1")
@@ -131,16 +194,19 @@ func TestCorruptAdmissionTelemetryNeverBlocksCandidateImports(t *testing.T) {
 		name       string
 		seed       func(key string)
 		wantStored string
+		wantStreak string
 	}{
 		{
 			name:       "malformed counter",
-			seed:       func(key string) { mini.HSet(key, "admitted", "not-an-integer") },
+			seed:       func(key string) { mini.HSet(key, "admitted", "not-an-integer", "zero_admission_streak", "3") },
 			wantStored: "not-an-integer",
+			wantStreak: "0",
 		},
 		{
 			name:       "counter at exact cap",
-			seed:       func(key string) { mini.HSet(key, "admitted", fmt.Sprint(int64(1<<53-1))) },
+			seed:       func(key string) { mini.HSet(key, "admitted", fmt.Sprint(int64(1<<53-1)), "zero_admission_streak", "3") },
 			wantStored: fmt.Sprint(int64(1<<53 - 1)),
+			wantStreak: "0",
 		},
 		{
 			name: "wrong-type quality key",
@@ -150,6 +216,7 @@ func TestCorruptAdmissionTelemetryNeverBlocksCandidateImports(t *testing.T) {
 				}
 			},
 			wantStored: "legacy non-hash value",
+			wantStreak: "",
 		},
 	}
 	for index, tc := range tests {
@@ -182,6 +249,11 @@ func TestCorruptAdmissionTelemetryNeverBlocksCandidateImports(t *testing.T) {
 			if err != nil || stored != tc.wantStored {
 				t.Fatalf("corrupt admission counter changed: got=%q err=%v want=%q", stored, err, tc.wantStored)
 			}
+			if tc.name != "wrong-type quality key" {
+				if got := mini.HGet(idKey, "zero_admission_streak"); got != tc.wantStreak {
+					t.Fatalf("zero admission streak=%q, want reset to %q", got, tc.wantStreak)
+				}
+			}
 		})
 	}
 }