Add counts to sources page (#6819)

app/proprietary/build.gradle

@@ -80,6 +80,10 @@ dependencies {
     implementation "software.amazon.awssdk:s3:${awsSdkVersion}"
     implementation "software.amazon.awssdk:url-connection-client:${awsSdkVersion}"
 
+    // @DataJpaTest slice (Boot 4 ships test slices as separate starters, like webmvc-test at the
+    // root) so policy.source repositories can be exercised against embedded H2.
+    testImplementation 'org.springframework.boot:spring-boot-starter-data-jpa-test'
+
     // Testcontainers: real MinIO/LocalStack (S3) and Valkey for integration tests in CI without
     // manually-started instances. Tests skip cleanly when Docker is unavailable.
     testImplementation "org.testcontainers:testcontainers:${testcontainersMinioVersion}"

app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyRunner.java

@@ -21,6 +21,7 @@ import stirling.software.proprietary.policy.model.PolicyRun;
 import stirling.software.proprietary.policy.model.PolicyRunStatus;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
 import stirling.software.proprietary.policy.source.Source;
+import stirling.software.proprietary.policy.source.SourceDocCounter;
 import stirling.software.proprietary.policy.source.SourceStore;
 
 /**
@@ -37,6 +38,7 @@ public class PolicyRunner {
     private final PolicyEngine policyEngine;
     private final List<InputSource> inputSources;
     private final SourceStore sourceStore;
+    private final SourceDocCounter docCounter;
 
     /**
      * Trigger entry point. Pulls every referenced source; each yielded unit becomes its own run so
@@ -65,7 +67,7 @@ public class PolicyRunner {
                         policy.id());
                 continue;
             }
-            runIds.addAll(pullAndRun(policy, source.toInputSpec()));
+            runIds.addAll(pullAndRun(policy, sourceId, source.toInputSpec()));
         }
         return runIds;
     }
@@ -82,7 +84,11 @@ public class PolicyRunner {
         return policyEngine.submit(definition, inputs, listener);
     }
 
-    private List<String> pullAndRun(Policy policy, InputSpec spec) {
+    /**
+     * Resolves the source and starts a run per unit; records how many documents the source fed and
+     * returns the ids of the runs started.
+     */
+    private List<String> pullAndRun(Policy policy, String sourceId, InputSpec spec) {
         InputSource source = sourceFor(spec);
         if (source == null) {
             log.warn(
@@ -103,9 +109,12 @@ public class PolicyRunner {
             return List.of();
         }
         List<String> runIds = new ArrayList<>();
+        long docsFed = 0;
         for (ResolvedInput unit : work) {
             runIds.add(startRun(policy, unit.inputs(), unit.onComplete()));
+            docsFed += unit.inputs().primary().size();
         }
+        docCounter.record(sourceId, docsFed);
         return runIds;
     }
 

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/DocStats.java

@@ -0,0 +1,15 @@
+package stirling.software.proprietary.policy.source;
+
+/**
+ * Per-source document throughput for the overview row: how many documents a source has fed into
+ * runs in total and over the trailing 24-hour and 30-day windows. Counts documents <em>fed</em>
+ * (picked up by a run), so a snapshot-mode source that re-reads the same files each run counts them
+ * per run.
+ */
+public record DocStats(long total, long last24h, long last30d) {
+
+    /** Number of trailing daily buckets in a source's daily series. */
+    public static final int DAYS = 30;
+
+    public static final DocStats ZERO = new DocStats(0, 0, 0);
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/InProcessSourceDocCounter.java

@@ -0,0 +1,64 @@
+package stirling.software.proprietary.policy.source;
+
+import java.time.Instant;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.function.Supplier;
+
+/**
+ * In-memory {@link SourceDocCounter} for tests and any future no-database mode. Holds hourly
+ * buckets per source; the clock is injectable so window boundaries can be exercised
+ * deterministically. {@link JpaSourceDocCounter} is the runtime bean.
+ */
+public class InProcessSourceDocCounter implements SourceDocCounter {
+
+    private final Supplier<Instant> clock;
+    private final Map<String, Map<Long, Long>> bucketsBySource = new ConcurrentHashMap<>();
+
+    public InProcessSourceDocCounter() {
+        this(Instant::now);
+    }
+
+    public InProcessSourceDocCounter(Supplier<Instant> clock) {
+        this.clock = clock;
+    }
+
+    @Override
+    public void record(String sourceId, long docs) {
+        if (docs <= 0) {
+            return;
+        }
+        bucketsBySource
+                .computeIfAbsent(sourceId, key -> new ConcurrentHashMap<>())
+                .merge(currentHour(), docs, Long::sum);
+    }
+
+    @Override
+    public Map<String, DocStats> statsFor(Collection<String> sourceIds) {
+        long now = currentHour();
+        Map<String, DocStats> stats = new HashMap<>();
+        for (String id : sourceIds) {
+            Map<Long, Long> buckets = bucketsBySource.getOrDefault(id, Map.of());
+            long total = buckets.values().stream().mapToLong(Long::longValue).sum();
+            long last24h =
+                    SourceDocWindows.sumSince(buckets, now - (SourceDocWindows.HOURS_IN_24H - 1));
+            long last30d = SourceDocWindows.sumSince(buckets, SourceDocWindows.firstDayHour(now));
+            stats.put(id, new DocStats(total, last24h, last30d));
+        }
+        return stats;
+    }
+
+    @Override
+    public List<Long> dailySeriesFor(String sourceId) {
+        long now = currentHour();
+        Map<Long, Long> buckets = bucketsBySource.getOrDefault(sourceId, Map.of());
+        return SourceDocWindows.series(SourceDocWindows.byDay(buckets), now / 24);
+    }
+
+    private long currentHour() {
+        return clock.get().getEpochSecond() / 3600;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/JpaSourceDocCounter.java

@@ -0,0 +1,154 @@
+package stirling.software.proprietary.policy.source;
+
+import java.time.Instant;
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.concurrent.TimeUnit;
+import java.util.function.IntSupplier;
+import java.util.function.Supplier;
+
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.autoconfigure.condition.ConditionalOnBooleanProperty;
+import org.springframework.dao.DataIntegrityViolationException;
+import org.springframework.scheduling.annotation.Scheduled;
+import org.springframework.stereotype.Service;
+
+/**
+ * Durable {@link SourceDocCounter}; the runtime bean. {@code record} keeps two things in step: an
+ * hourly bucket ({@link SourceDocCountEntity}) that feeds the rolling 24h / 30d / daily-series
+ * windows, and a denormalized lifetime total ({@link SourceDocTotalEntity}) read directly for the
+ * all-time figure. The lifetime counter means the overview never scans a source's whole bucket
+ * history, and lets {@link #pruneOldBuckets()} retire buckets past the 30-day window so the hourly
+ * table stays bounded (~one row per source per active hour, for at most 30 days).
+ */
+@Service
+@ConditionalOnBooleanProperty(name = "policies.enabled")
+public class JpaSourceDocCounter implements SourceDocCounter {
+
+    private final SourceDocCountRepository countRepository;
+    private final SourceDocTotalRepository totalRepository;
+    private final Supplier<Instant> clock;
+
+    @Autowired
+    public JpaSourceDocCounter(
+            SourceDocCountRepository countRepository, SourceDocTotalRepository totalRepository) {
+        this(countRepository, totalRepository, Instant::now);
+    }
+
+    // Clock seam so tests can pin "now"; the runtime bean uses the wall clock above.
+    JpaSourceDocCounter(
+            SourceDocCountRepository countRepository,
+            SourceDocTotalRepository totalRepository,
+            Supplier<Instant> clock) {
+        this.countRepository = countRepository;
+        this.totalRepository = totalRepository;
+        this.clock = clock;
+    }
+
+    @Override
+    public void record(String sourceId, long docs) {
+        if (docs <= 0) {
+            return;
+        }
+        long bucketHour = currentHour();
+        upsert(
+                () -> totalRepository.increment(sourceId, docs),
+                () -> totalRepository.saveAndFlush(new SourceDocTotalEntity(sourceId, docs)));
+        upsert(
+                () -> countRepository.increment(sourceId, bucketHour, docs),
+                () ->
+                        countRepository.saveAndFlush(
+                                new SourceDocCountEntity(sourceId, bucketHour, docs)));
+    }
+
+    /**
+     * Add {@code docs} to a per-source running total: increment the existing row, else insert a new
+     * one. The insert is flushed now (in its own transaction, since {@code record} is not
+     * {@code @Transactional}) so a concurrent run's winning insert surfaces as a constraint
+     * violation we retry as an increment, rather than as a silent {@code merge} overwrite or a
+     * later doomed commit.
+     */
+    private static void upsert(IntSupplier increment, Runnable insert) {
+        if (increment.getAsInt() > 0) {
+            return;
+        }
+        try {
+            insert.run();
+        } catch (DataIntegrityViolationException concurrentInsert) {
+            increment.getAsInt();
+        }
+    }
+
+    @Override
+    public Map<String, DocStats> statsFor(Collection<String> sourceIds) {
+        if (sourceIds.isEmpty()) {
+            return Map.of();
+        }
+        long now = currentHour();
+        Map<String, Long> totals = sums(totalRepository.totalsFor(sourceIds));
+        Map<String, Long> last24h =
+                sums(
+                        countRepository.sumBySourceSince(
+                                sourceIds, now - (SourceDocWindows.HOURS_IN_24H - 1)));
+        Map<String, Long> last30d =
+                sums(
+                        countRepository.sumBySourceSince(
+                                sourceIds, SourceDocWindows.firstDayHour(now)));
+
+        Map<String, DocStats> stats = new HashMap<>();
+        for (String id : sourceIds) {
+            stats.put(
+                    id,
+                    new DocStats(
+                            totals.getOrDefault(id, 0L),
+                            last24h.getOrDefault(id, 0L),
+                            last30d.getOrDefault(id, 0L)));
+        }
+        return stats;
+    }
+
+    @Override
+    public List<Long> dailySeriesFor(String sourceId) {
+        long now = currentHour();
+        Collection<String> ids = List.of(sourceId);
+        Map<Long, Long> dailyCounts =
+                dailyBySource(
+                                countRepository.dailyCountsSince(
+                                        ids, SourceDocWindows.firstDayHour(now)))
+                        .getOrDefault(sourceId, Map.of());
+        return SourceDocWindows.series(dailyCounts, now / 24);
+    }
+
+    /**
+     * Retire hourly buckets older than the 30-day window; the lifetime total is held separately, so
+     * nothing reported is lost. Daily is ample - the read queries already ignore older buckets, so
+     * this is purely storage hygiene.
+     */
+    @Scheduled(fixedDelay = 1, timeUnit = TimeUnit.DAYS)
+    public void pruneOldBuckets() {
+        countRepository.deleteOlderThan(SourceDocWindows.firstDayHour(currentHour()));
+    }
+
+    private long currentHour() {
+        return clock.get().getEpochSecond() / 3600;
+    }
+
+    private static Map<String, Long> sums(List<SourceDocSum> rows) {
+        Map<String, Long> map = new HashMap<>();
+        for (SourceDocSum row : rows) {
+            map.put(row.sourceId(), row.count() == null ? 0L : row.count());
+        }
+        return map;
+    }
+
+    private static Map<String, Map<Long, Long>> dailyBySource(List<SourceDayDocSum> rows) {
+        Map<String, Map<Long, Long>> bySource = new HashMap<>();
+        for (SourceDayDocSum row : rows) {
+            bySource.computeIfAbsent(row.sourceId(), key -> new HashMap<>())
+                    .merge(row.day(), row.docs() == null ? 0L : row.docs(), Long::sum);
+        }
+        return bySource;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceController.java

@@ -74,6 +74,20 @@ public class SourceController {
                 .orElseGet(() -> ResponseEntity.notFound().build());
     }
 
+    @GetMapping("/{sourceId}/document-counts")
+    @Operation(
+            summary = "Daily document counts for a source",
+            description =
+                    "The trailing 30-day per-day document series (oldest first) for the source's"
+                            + " sparkline.")
+    public ResponseEntity<List<Long>> documentCounts(@PathVariable String sourceId) {
+        return sourceStore
+                .get(sourceId)
+                .filter(sourceAccessGuard::canAccess)
+                .map(source -> ResponseEntity.ok(overviewService.dailySeries(source.id())))
+                .orElseGet(() -> ResponseEntity.notFound().build());
+    }
+
     @PostMapping(consumes = MediaType.APPLICATION_JSON_VALUE)
     @Operation(
             summary = "Create or update a source",

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDayDocSum.java

@@ -0,0 +1,7 @@
+package stirling.software.proprietary.policy.source;
+
+/**
+ * A {@code (sourceId, epoch-day, summed document count)} row from the daily-aggregate query. The
+ * day is {@code floor(bucketHour / 24)}, i.e. hours-since-epoch collapsed to days-since-epoch.
+ */
+public record SourceDayDocSum(String sourceId, Long day, Long docs) {}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountEntity.java

@@ -0,0 +1,62 @@
+package stirling.software.proprietary.policy.source;
+
+import java.io.Serializable;
+
+import org.springframework.data.domain.Persistable;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.IdClass;
+import jakarta.persistence.Table;
+import jakarta.persistence.Transient;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+/**
+ * One hour's document tally for a source: {@code bucketHour} is the hours-since-epoch the documents
+ * were fed, {@code docCount} the running total for that hour. Rolling-window totals are summed from
+ * these buckets. {@code sourceId} is a plain value, not a foreign key, matching the rest of the
+ * subsystem so it stays decoupled from the security entities.
+ */
+@Entity
+@Table(name = "policy_source_doc_counts")
+@IdClass(SourceDocCountId.class)
+@NoArgsConstructor
+@Getter
+@Setter
+public class SourceDocCountEntity implements Serializable, Persistable<SourceDocCountId> {
+
+    private static final long serialVersionUID = 1L;
+
+    @Id
+    @Column(name = "source_id")
+    private String sourceId;
+
+    @Id
+    @Column(name = "bucket_hour")
+    private long bucketHour;
+
+    @Column(name = "doc_count")
+    private long docCount;
+
+    public SourceDocCountEntity(String sourceId, long bucketHour, long docCount) {
+        this.sourceId = sourceId;
+        this.bucketHour = bucketHour;
+        this.docCount = docCount;
+    }
+
+    @Override
+    @Transient
+    public SourceDocCountId getId() {
+        return new SourceDocCountId(sourceId, bucketHour);
+    }
+
+    @Override
+    @Transient
+    public boolean isNew() {
+        return true;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountId.java

@@ -0,0 +1,36 @@
+package stirling.software.proprietary.policy.source;
+
+import java.io.Serializable;
+import java.util.Objects;
+
+/** Composite key for {@link SourceDocCountEntity}: one row per source per hour bucket. */
+public class SourceDocCountId implements Serializable {
+
+    private static final long serialVersionUID = 1L;
+
+    private String sourceId;
+    private long bucketHour;
+
+    public SourceDocCountId() {}
+
+    public SourceDocCountId(String sourceId, long bucketHour) {
+        this.sourceId = sourceId;
+        this.bucketHour = bucketHour;
+    }
+
+    @Override
+    public boolean equals(Object o) {
+        if (this == o) {
+            return true;
+        }
+        if (!(o instanceof SourceDocCountId other)) {
+            return false;
+        }
+        return bucketHour == other.bucketHour && Objects.equals(sourceId, other.sourceId);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hash(sourceId, bucketHour);
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCountRepository.java

@@ -0,0 +1,69 @@
+package stirling.software.proprietary.policy.source;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
+
+@Repository
+public interface SourceDocCountRepository
+        extends JpaRepository<SourceDocCountEntity, SourceDocCountId> {
+
+    /**
+     * Add to an existing bucket; returns the number of rows updated (0 when the bucket is new).
+     * Transactional per call so {@code JpaSourceDocCounter.record} can run it (and the retry after
+     * a concurrent insert) without an enclosing transaction.
+     */
+    @Modifying
+    @Transactional
+    @Query(
+            "update SourceDocCountEntity e set e.docCount = e.docCount + :docs"
+                    + " where e.sourceId = :sourceId and e.bucketHour = :bucketHour")
+    int increment(
+            @Param("sourceId") String sourceId,
+            @Param("bucketHour") long bucketHour,
+            @Param("docs") long docs);
+
+    /**
+     * Delete hourly buckets older than {@code floor} (hours-since-epoch). Nothing reads buckets
+     * before the 30-day window ({@code SourceDocWindows.firstDayHour}); the lifetime total lives in
+     * {@code policy_source_doc_totals}, so retiring old buckets keeps the table bounded without
+     * losing any reported figure.
+     */
+    @Modifying
+    @Transactional
+    @Query("delete from SourceDocCountEntity e where e.bucketHour < :floor")
+    int deleteOlderThan(@Param("floor") long floor);
+
+    /**
+     * Document total per source restricted to buckets at or after {@code since} (the 24h window).
+     */
+    @Query(
+            "select new stirling.software.proprietary.policy.source.SourceDocSum("
+                    + "e.sourceId, sum(e.docCount))"
+                    + " from SourceDocCountEntity e"
+                    + " where e.sourceId in :ids and e.bucketHour >= :since"
+                    + " group by e.sourceId")
+    List<SourceDocSum> sumBySourceSince(
+            @Param("ids") Collection<String> ids, @Param("since") long since);
+
+    /**
+     * Per-source, per-day document totals for buckets at or after {@code since}, summed in the
+     * database so the overview reads ~one row per source per active day instead of per active hour.
+     * The day is {@code cast(floor(bucketHour / 24.0) as long)}: {@code 24.0} forces decimal
+     * division and the cast pins the result to a whole day on every dialect.
+     */
+    @Query(
+            "select new stirling.software.proprietary.policy.source.SourceDayDocSum("
+                    + "e.sourceId, cast(floor(e.bucketHour / 24.0) as long), sum(e.docCount))"
+                    + " from SourceDocCountEntity e"
+                    + " where e.sourceId in :ids and e.bucketHour >= :since"
+                    + " group by cast(floor(e.bucketHour / 24.0) as long), e.sourceId")
+    List<SourceDayDocSum> dailyCountsSince(
+            @Param("ids") Collection<String> ids, @Param("since") long since);
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocCounter.java

@@ -0,0 +1,28 @@
+package stirling.software.proprietary.policy.source;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Records and reports how many documents each source feeds into runs. Counting is bucketed by hour
+ * so the overview can report rolling totals ({@link DocStats}) cheaply; {@link JpaSourceDocCounter}
+ * is the runtime bean and {@link InProcessSourceDocCounter} backs tests.
+ */
+public interface SourceDocCounter {
+
+    /** Record that {@code docs} documents were fed from {@code sourceId} at the current time. */
+    void record(String sourceId, long docs);
+
+    /**
+     * Document totals for each given source; a source with no recorded docs maps to {@link
+     * DocStats#ZERO}.
+     */
+    Map<String, DocStats> statsFor(Collection<String> sourceIds);
+
+    /**
+     * The trailing {@link DocStats#DAYS}-day daily document series for one source, oldest first,
+     * for the detail-panel sparkline.
+     */
+    List<Long> dailySeriesFor(String sourceId);
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocSum.java

@@ -0,0 +1,6 @@
+package stirling.software.proprietary.policy.source;
+
+/**
+ * A {@code (sourceId, summed document count)} row, populated by the doc-count aggregate queries.
+ */
+public record SourceDocSum(String sourceId, Long count) {}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocTotalEntity.java

@@ -0,0 +1,59 @@
+package stirling.software.proprietary.policy.source;
+
+import java.io.Serializable;
+
+import org.springframework.data.domain.Persistable;
+
+import jakarta.persistence.Column;
+import jakarta.persistence.Entity;
+import jakarta.persistence.Id;
+import jakarta.persistence.Table;
+import jakarta.persistence.Transient;
+
+import lombok.Getter;
+import lombok.NoArgsConstructor;
+import lombok.Setter;
+
+/**
+ * A source's lifetime document total, denormalized so the overview reads the all-time count in one
+ * row instead of scanning the source's whole hourly-bucket history, and so {@link
+ * SourceDocCountEntity} buckets can be pruned to the rolling 30-day window without losing it.
+ *
+ * <p>Like {@link SourceDocCountEntity}, implements {@link Persistable} reporting {@code isNew() ==
+ * true} so a new source's first {@code save} {@code persist}s (a raw INSERT) and a concurrent
+ * insert surfaces as a constraint violation the counter retries as an increment, rather than {@code
+ * merge} silently overwriting it.
+ */
+@Entity
+@Table(name = "policy_source_doc_totals")
+@NoArgsConstructor
+@Getter
+@Setter
+public class SourceDocTotalEntity implements Serializable, Persistable<String> {
+
+    private static final long serialVersionUID = 1L;
+
+    @Id
+    @Column(name = "source_id")
+    private String sourceId;
+
+    @Column(name = "doc_total")
+    private long docTotal;
+
+    public SourceDocTotalEntity(String sourceId, long docTotal) {
+        this.sourceId = sourceId;
+        this.docTotal = docTotal;
+    }
+
+    @Override
+    @Transient
+    public String getId() {
+        return sourceId;
+    }
+
+    @Override
+    @Transient
+    public boolean isNew() {
+        return true;
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocTotalRepository.java

@@ -0,0 +1,34 @@
+package stirling.software.proprietary.policy.source;
+
+import java.util.Collection;
+import java.util.List;
+
+import org.springframework.data.jpa.repository.JpaRepository;
+import org.springframework.data.jpa.repository.Modifying;
+import org.springframework.data.jpa.repository.Query;
+import org.springframework.data.repository.query.Param;
+import org.springframework.stereotype.Repository;
+import org.springframework.transaction.annotation.Transactional;
+
+@Repository
+public interface SourceDocTotalRepository extends JpaRepository<SourceDocTotalEntity, String> {
+
+    /**
+     * Add to a source's lifetime total; returns the number of rows updated (0 when the source has
+     * no total row yet). Transactional per call so {@code JpaSourceDocCounter.record} can run it
+     * (and the retry after a concurrent insert) without an enclosing transaction.
+     */
+    @Modifying
+    @Transactional
+    @Query(
+            "update SourceDocTotalEntity e set e.docTotal = e.docTotal + :docs"
+                    + " where e.sourceId = :sourceId")
+    int increment(@Param("sourceId") String sourceId, @Param("docs") long docs);
+
+    /** Lifetime totals for the given sources, as {@code (sourceId, total)} rows. */
+    @Query(
+            "select new stirling.software.proprietary.policy.source.SourceDocSum("
+                    + "e.sourceId, e.docTotal)"
+                    + " from SourceDocTotalEntity e where e.sourceId in :ids")
+    List<SourceDocSum> totalsFor(@Param("ids") Collection<String> ids);
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceDocWindows.java

@@ -0,0 +1,61 @@
+package stirling.software.proprietary.policy.source;
+
+import java.util.Arrays;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Shared rolling-window math over a source's document buckets, so the JPA and in-memory counters
+ * agree on the window boundaries and the daily series. Both define "last 30 days" as the buckets at
+ * or after {@link #firstDayHour}, and build the series from per-day counts with {@link #series}
+ * (the JPA counter aggregates by day in SQL; the in-memory counter groups its hourly buckets with
+ * {@link #byDay}).
+ */
+final class SourceDocWindows {
+
+    static final long HOURS_IN_24H = 24;
+
+    private SourceDocWindows() {}
+
+    /**
+     * The hours-since-epoch at the start of the oldest day in the 30-day window, given the current
+     * hour bucket. Both the "last 30 days" total and the daily series are measured from here, so
+     * the KPI and the sparkline always cover the same buckets.
+     */
+    static long firstDayHour(long nowHour) {
+        return ((nowHour / 24) - (DocStats.DAYS - 1)) * 24;
+    }
+
+    /**
+     * Build the {@link DocStats#DAYS}-day daily series (oldest first) from per-day document counts
+     * (keyed by epoch-day, i.e. hours-since-epoch / 24). {@code currentDay} is today's epoch-day;
+     * the series runs back {@code DAYS} days from it.
+     */
+    static List<Long> series(Map<Long, Long> dailyCounts, long currentDay) {
+        long firstDay = currentDay - (DocStats.DAYS - 1);
+        long[] daily = new long[DocStats.DAYS];
+        for (Map.Entry<Long, Long> day : dailyCounts.entrySet()) {
+            int dayIndex = (int) (day.getKey() - firstDay);
+            if (dayIndex >= 0 && dayIndex < DocStats.DAYS) {
+                daily[dayIndex] += day.getValue();
+            }
+        }
+        return Arrays.stream(daily).boxed().toList();
+    }
+
+    /** Collapse hourly buckets (keyed by hours-since-epoch) into per-day counts (keyed by day). */
+    static Map<Long, Long> byDay(Map<Long, Long> hourlyCounts) {
+        Map<Long, Long> dailyCounts = new HashMap<>();
+        hourlyCounts.forEach((hour, count) -> dailyCounts.merge(hour / 24, count, Long::sum));
+        return dailyCounts;
+    }
+
+    /** Sum hourly buckets at or after {@code since} (hours-since-epoch). */
+    static long sumSince(Map<Long, Long> hourlyCounts, long since) {
+        return hourlyCounts.entrySet().stream()
+                .filter(entry -> entry.getKey() >= since)
+                .mapToLong(Map.Entry::getValue)
+                .sum();
+    }
+}

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceOverviewService.java

@@ -30,12 +30,15 @@ public class SourceOverviewService {
     private final PolicyStore policyStore;
     private final SourceAccessGuard sourceAccessGuard;
     private final PolicyAccessGuard policyAccessGuard;
+    private final SourceDocCounter docCounter;
 
     public SourcesResponse overview() {
         List<Source> sources = sourceAccessGuard.visibleFrom(sourceStore);
         List<Policy> policies = policyAccessGuard.visibleFrom(policyStore);
 
         Map<String, List<Policy>> referencesBySource = referencesBySource(policies);
+        Map<String, DocStats> docStats =
+                docCounter.statsFor(sources.stream().map(Source::id).toList());
 
         List<SourceView> views =
                 sources.stream()
@@ -44,7 +47,8 @@ public class SourceOverviewService {
                                         toView(
                                                 source,
                                                 referencesBySource.getOrDefault(
-                                                        source.id(), List.of())))
+                                                        source.id(), List.of()),
+                                                docStats.getOrDefault(source.id(), DocStats.ZERO)))
                         .sorted(
                                 Comparator.comparingInt(SourceView::referenceCount)
                                         .reversed()
@@ -54,6 +58,14 @@ public class SourceOverviewService {
         return new SourcesResponse(buildKpis(views), views);
     }
 
+    /**
+     * The 30-day daily document series for one source (oldest first), for the expanded row's
+     * sparkline.
+     */
+    public List<Long> dailySeries(String sourceId) {
+        return docCounter.dailySeriesFor(sourceId);
+    }
+
     /** Policies referencing each source id, across the caller's visible policies. */
     private static Map<String, List<Policy>> referencesBySource(List<Policy> policies) {
         Map<String, List<Policy>> bySource = new HashMap<>();
@@ -65,7 +77,8 @@ public class SourceOverviewService {
         return bySource;
     }
 
-    private static SourceView toView(Source source, List<Policy> referencingPolicies) {
+    private static SourceView toView(
+            Source source, List<Policy> referencingPolicies, DocStats docs) {
         List<SourceView.PolicyRef> refs =
                 referencingPolicies.stream()
                         .map(policy -> new SourceView.PolicyRef(policy.id(), policy.name()))
@@ -78,7 +91,9 @@ public class SourceOverviewService {
                 refs.size(),
                 refs,
                 configRows(source),
-                null);
+                docs.total(),
+                docs.last24h(),
+                docs.last30d());
     }
 
     /** A disabled (paused) source reads as "disabled"; an unreferenced one reads as "unused". */

app/proprietary/src/main/java/stirling/software/proprietary/policy/source/SourceView.java

@@ -4,8 +4,8 @@ import java.util.List;
 
 /**
  * One row in the Sources overview: a persisted input connection shown exactly once, with how many
- * policies reference it (and which). {@code docsTotal} is {@code null} - per-source document volume
- * is not tracked yet; the field is reserved so a later doc-accounting pass is additive.
+ * policies reference it (and which) and how many documents it has fed into runs ({@code docsTotal}
+ * lifetime plus the trailing 24-hour and 30-day windows).
  */
 public record SourceView(
         String id,
@@ -15,7 +15,9 @@ public record SourceView(
         int referenceCount,
         List<PolicyRef> referencingPolicies,
         List<DetailRow> config,
-        Long docsTotal) {
+        long docsTotal,
+        long docs24h,
+        long docs30d) {
 
     /** A policy that references this source. */
     public record PolicyRef(String id, String name) {}

app/proprietary/src/test/java/stirling/software/proprietary/policy/engine/PolicyRunnerTest.java

@@ -33,6 +33,7 @@ import stirling.software.proprietary.policy.model.PolicyInputs;
 import stirling.software.proprietary.policy.model.PolicyRun;
 import stirling.software.proprietary.policy.model.PolicyRunStatus;
 import stirling.software.proprietary.policy.progress.PolicyProgressListener;
+import stirling.software.proprietary.policy.source.InProcessSourceDocCounter;
 import stirling.software.proprietary.policy.source.InProcessSourceStore;
 import stirling.software.proprietary.policy.source.Source;
 import stirling.software.proprietary.policy.source.SourceStore;
@@ -53,7 +54,12 @@ class PolicyRunnerTest {
 
     @BeforeEach
     void setUp() {
-        runner = new PolicyRunner(policyEngine, List.of(folderSource), sourceStore);
+        runner =
+                new PolicyRunner(
+                        policyEngine,
+                        List.of(folderSource),
+                        sourceStore,
+                        new InProcessSourceDocCounter());
     }
 
     @Test

app/proprietary/src/test/java/stirling/software/proprietary/policy/source/InProcessSourceDocCounterTest.java

@@ -0,0 +1,62 @@
+package stirling.software.proprietary.policy.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.Instant;
+import java.util.Collections;
+import java.util.List;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.junit.jupiter.api.Test;
+
+/**
+ * Tests the rolling-window aggregation: documents recorded at different times roll into the total,
+ * the 24-hour / 30-day windows, and the daily series correctly as the clock advances.
+ */
+class InProcessSourceDocCounterTest {
+
+    private static InProcessSourceDocCounter seededCounter(AtomicReference<Instant> clock) {
+        InProcessSourceDocCounter counter = new InProcessSourceDocCounter(clock::get);
+        // 40 days ago: outside the 30-day window.
+        clock.set(Instant.parse("2026-05-21T12:00:00Z"));
+        counter.record("s", 100);
+        // 10 days ago: inside 30 days, outside 24 hours.
+        clock.set(Instant.parse("2026-06-20T12:00:00Z"));
+        counter.record("s", 50);
+        // 2 hours ago: inside both windows.
+        clock.set(Instant.parse("2026-06-30T10:00:00Z"));
+        counter.record("s", 7);
+        // Query as of "now".
+        clock.set(Instant.parse("2026-06-30T12:00:00Z"));
+        return counter;
+    }
+
+    @Test
+    void rollsCountsIntoTotalAnd24hAnd30dWindows() {
+        AtomicReference<Instant> clock = new AtomicReference<>();
+        DocStats stats = seededCounter(clock).statsFor(List.of("s")).get("s");
+
+        assertEquals(157, stats.total());
+        assertEquals(7, stats.last24h());
+        assertEquals(57, stats.last30d());
+    }
+
+    @Test
+    void buildsTheDailySeriesOldestFirst() {
+        AtomicReference<Instant> clock = new AtomicReference<>();
+        // Today (index 29) and 10 days ago (index 19) only; 40 days ago is outside the window.
+        List<Long> series = seededCounter(clock).dailySeriesFor("s");
+
+        assertEquals(30, series.size());
+        assertEquals(7L, series.get(29));
+        assertEquals(50L, series.get(19));
+        assertEquals(57L, series.stream().mapToLong(Long::longValue).sum());
+    }
+
+    @Test
+    void aSourceWithNoRecordedDocsIsZero() {
+        InProcessSourceDocCounter counter = new InProcessSourceDocCounter();
+        assertEquals(DocStats.ZERO, counter.statsFor(List.of("unknown")).get("unknown"));
+        assertEquals(Collections.nCopies(DocStats.DAYS, 0L), counter.dailySeriesFor("unknown"));
+    }
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/source/JpaSourceDocCounterDbTest.java

@@ -0,0 +1,93 @@
+package stirling.software.proprietary.policy.source;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+import java.time.Instant;
+import java.util.List;
+
+import org.junit.jupiter.api.Test;
+import org.springframework.beans.factory.annotation.Autowired;
+import org.springframework.boot.SpringBootConfiguration;
+import org.springframework.boot.autoconfigure.AutoConfigurationPackage;
+import org.springframework.boot.data.jpa.test.autoconfigure.DataJpaTest;
+
+/**
+ * Exercises {@link JpaSourceDocCounter} against a real (H2) database so the daily-aggregate query
+ * ({@code cast(floor(bucketHour / 24.0) as long)} grouping) and the record upsert are actually run,
+ * not just asserted against a mock. The unit-level window math lives in {@link
+ * InProcessSourceDocCounterTest}.
+ */
+@DataJpaTest
+class JpaSourceDocCounterDbTest {
+
+    // Pinned "now" so seeding and statsFor agree regardless of when the test runs (no hour-tick
+    // flake at the boundary of a wall-clock hour).
+    private static final Instant NOW = Instant.parse("2026-06-30T12:00:00Z");
+    private static final long NOW_HOUR = NOW.getEpochSecond() / 3600;
+
+    @Autowired private SourceDocCountRepository repository;
+    @Autowired private SourceDocTotalRepository totalRepository;
+
+    private JpaSourceDocCounter counter() {
+        return new JpaSourceDocCounter(repository, totalRepository, () -> NOW);
+    }
+
+    @Test
+    void recordIncrementsBothTheHourlyBucketAndTheLifetimeTotal() {
+        JpaSourceDocCounter counter = counter();
+        counter.record("s", 5);
+        counter.record("s", 3);
+
+        DocStats stats = counter.statsFor(List.of("s")).get("s");
+        assertEquals(8, stats.total()); // from the denormalized lifetime row
+        assertEquals(8, stats.last24h());
+        assertEquals(8, stats.last30d());
+        assertEquals(8L, counter.dailySeriesFor("s").get(DocStats.DAYS - 1));
+    }
+
+    @Test
+    void statsBucketDocsByDayAndWindowFromTheDatabase() {
+        // Seed buckets directly at controlled hours: today, 10 days ago, 40 days ago, plus the
+        // lifetime row (record() would write it, but here we seed history directly).
+        repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR, 7));
+        repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR - 24L * 10, 50));
+        repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR - 24L * 40, 100));
+        totalRepository.saveAndFlush(new SourceDocTotalEntity("s", 157));
+
+        DocStats stats = counter().statsFor(List.of("s")).get("s");
+        assertEquals(157, stats.total()); // lifetime, including the out-of-window 40-days-ago docs
+        assertEquals(7, stats.last24h());
+        assertEquals(57, stats.last30d());
+
+        // The daily series (fetched separately, per source) covers the same 30-day window.
+        List<Long> series = counter().dailySeriesFor("s");
+        assertEquals(DocStats.DAYS, series.size());
+        assertEquals(7L, series.get(DocStats.DAYS - 1)); // today
+        assertEquals(50L, series.get(DocStats.DAYS - 11)); // 10 days ago
+        assertEquals(57L, series.stream().mapToLong(Long::longValue).sum());
+    }
+
+    @Test
+    void pruneRetiresOutOfWindowBucketsButKeepsTheLifetimeTotal() {
+        repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR, 7)); // today
+        repository.saveAndFlush(new SourceDocCountEntity("s", NOW_HOUR - 24L * 40, 100)); // 40d ago
+        totalRepository.saveAndFlush(new SourceDocTotalEntity("s", 107));
+
+        counter().pruneOldBuckets();
+
+        assertEquals(1, repository.count()); // only the in-window bucket remains
+        DocStats stats = counter().statsFor(List.of("s")).get("s");
+        assertEquals(107, stats.total()); // lifetime survives pruning
+        assertEquals(7, stats.last24h());
+        assertEquals(7, stats.last30d());
+    }
+
+    @Test
+    void aSourceWithNoRecordedDocsIsZero() {
+        assertEquals(DocStats.ZERO, counter().statsFor(List.of("unknown")).get("unknown"));
+    }
+
+    @SpringBootConfiguration
+    @AutoConfigurationPackage
+    static class TestApp {}
+}

app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceControllerTest.java

@@ -50,7 +50,12 @@ class SourceControllerTest {
         SourceAccessGuard sourceGuard = new SourceAccessGuard(userService, properties, authority);
         PolicyAccessGuard policyGuard = new PolicyAccessGuard(userService, properties, authority);
         SourceOverviewService overviewService =
-                new SourceOverviewService(sourceStore, policyStore, sourceGuard, policyGuard);
+                new SourceOverviewService(
+                        sourceStore,
+                        policyStore,
+                        sourceGuard,
+                        policyGuard,
+                        new InProcessSourceDocCounter());
         triggerManager = mock(PolicyTriggerManager.class);
         // A permissive input source so config validation passes and save can be exercised.
         InputSource folderInput = mock(InputSource.class);

app/proprietary/src/test/java/stirling/software/proprietary/policy/source/SourceOverviewServiceTest.java

@@ -1,7 +1,6 @@
 package stirling.software.proprietary.policy.source;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.when;
@@ -31,6 +30,7 @@ class SourceOverviewServiceTest {
 
     private final SourceStore sourceStore = new InProcessSourceStore();
     private final PolicyStore policyStore = new InProcessPolicyStore();
+    private final SourceDocCounter docCounter = new InProcessSourceDocCounter();
     private SourceOverviewService service;
 
     @BeforeEach
@@ -41,7 +41,9 @@ class SourceOverviewServiceTest {
         PolicyManagementAuthority authority = mock(PolicyManagementAuthority.class);
         SourceAccessGuard sourceGuard = new SourceAccessGuard(userService, properties, authority);
         PolicyAccessGuard policyGuard = new PolicyAccessGuard(userService, properties, authority);
-        service = new SourceOverviewService(sourceStore, policyStore, sourceGuard, policyGuard);
+        service =
+                new SourceOverviewService(
+                        sourceStore, policyStore, sourceGuard, policyGuard, docCounter);
     }
 
     @Test
@@ -111,7 +113,8 @@ class SourceOverviewServiceTest {
         SourceAccessGuard sourceGuard = new SourceAccessGuard(userService, properties, authority);
         PolicyAccessGuard policyGuard = new PolicyAccessGuard(userService, properties, authority);
         SourceOverviewService scoped =
-                new SourceOverviewService(sourceStore, policyStore, sourceGuard, policyGuard);
+                new SourceOverviewService(
+                        sourceStore, policyStore, sourceGuard, policyGuard, docCounter);
 
         Source ours = teamSource("Ours", "/ours", 1L);
         teamSource("Theirs", "/theirs", 2L);
@@ -128,9 +131,19 @@ class SourceOverviewServiceTest {
     }
 
     @Test
-    void documentVolumeIsNotTrackedYet() {
+    void documentCountsReflectRecordedDocs() {
         Source a = source("A", "/a");
-        assertNull(find(service.overview(), a.id()).docsTotal());
+        Source b = source("B", "/b");
+        docCounter.record(a.id(), 5);
+        docCounter.record(a.id(), 3);
+
+        SourceView av = find(service.overview(), a.id());
+        assertEquals(8, av.docsTotal());
+        assertEquals(8, av.docs24h());
+        assertEquals(8, av.docs30d());
+
+        // A source with no recorded documents reads as zero, not null.
+        assertEquals(0, find(service.overview(), b.id()).docsTotal());
     }
 
     private Source source(String name, String directory) {

app/saas/src/main/resources/db/migration/saas/V23__policy_source_doc_counts.sql

@@ -0,0 +1,26 @@
+-- Per-source document throughput, in two tables:
+--
+--   policy_source_doc_counts  one row per source per hour bucket (hours-since-epoch), holding how
+--                             many documents that source fed into runs in that hour. Feeds the
+--                             rolling last-24h / last-30d windows and the 30-day daily series, and
+--                             is pruned to that window so it stays bounded.
+--   policy_source_doc_totals  a denormalized lifetime total per source, incremented alongside the
+--                             hourly bucket, so the overview reads the all-time figure in one row
+--                             instead of scanning a source's whole bucket history - and so the
+--                             hourly buckets can be pruned without losing it.
+--
+-- Gated by policies.enabled like the rest of the subsystem; Hibernate ddl-auto would also create
+-- these, but the migration keeps the schema explicit for the Flyway-managed deployments.
+
+CREATE TABLE IF NOT EXISTS policy_source_doc_counts (
+    source_id   VARCHAR(255) NOT NULL,
+    bucket_hour BIGINT       NOT NULL,
+    doc_count   BIGINT       NOT NULL DEFAULT 0,
+    PRIMARY KEY (source_id, bucket_hour)
+);
+
+CREATE TABLE IF NOT EXISTS policy_source_doc_totals (
+    source_id VARCHAR(255) NOT NULL,
+    doc_total BIGINT       NOT NULL DEFAULT 0,
+    PRIMARY KEY (source_id)
+);

frontend/portal/public/locales/en-US/translation.toml

@@ -330,7 +330,11 @@ subtitle = "{{type}} · {{status}}"
 closeAriaLabel = "Close detail"
 usedBy = "Used by"
 notReferenced = "Not referenced by any policy, so it's safe to delete."
-docsUntracked = "Per-source document volume isn't tracked yet."
+documents = "Documents"
+docsTotal = "Total seen"
+docs24h = "Last 24h"
+docs30d = "Last 30 days"
+docsTrend = "Documents over the last 30 days"
 edit = "Edit"
 pause = "Pause"
 resume = "Resume"

frontend/portal/src/api/sources.ts

@@ -28,8 +28,10 @@ export interface SourceView {
   referenceCount: number;
   referencingPolicies: SourcePolicyRef[];
   config: SourceDetailRow[];
-  /** Per-source document volume: not tracked yet (always null for now). */
-  docsTotal: number | null;
+  /** Documents this source has fed into runs: lifetime, plus trailing 24h / 30d windows. */
+  docsTotal: number;
+  docs24h: number;
+  docs30d: number;
 }
 
 export interface SourceKpi {
@@ -69,6 +71,17 @@ export async function fetchSource(id: string): Promise<Source> {
   );
 }
 
+/**
+ * GET /api/v1/sources/{id}/document-counts: the trailing 30-day daily document
+ * series (oldest first) for the source's sparkline. Fetched only for the expanded
+ * row, so the overview list stays lightweight.
+ */
+export async function fetchSourceDocCounts(id: string): Promise<number[]> {
+  return apiClient.local.json<number[]>(
+    `/api/v1/sources/${encodeURIComponent(id)}/document-counts`,
+  );
+}
+
 /** POST /api/v1/sources: create (blank id) or update (matched id) a source. */
 export async function createSource(source: Source): Promise<Source> {
   return apiClient.local.json<Source>("/api/v1/sources", {

frontend/portal/src/components/sources/SourceDetailCard.stories.tsx

@@ -1,6 +1,9 @@
 import type { Meta, StoryObj } from "@storybook/react-vite";
 import type { SourceView } from "@portal/api/sources";
 import { SourceDetailCard } from "@portal/components/sources/SourceDetailCard";
+import { sampleDailySeries } from "@portal/mocks/sampleDailySeries";
+
+const SAMPLE_SERIES = sampleDailySeries(330);
 
 const IN_USE: SourceView = {
   id: "src-claims",
@@ -16,7 +19,9 @@ const IN_USE: SourceView = {
     { label: "Directory", value: "/data/claims-intake" },
     { label: "Mode", value: "consume" },
   ],
-  docsTotal: null,
+  docsTotal: 45230,
+  docs24h: 312,
+  docs30d: 9870,
 };
 
 const ORPHANED: SourceView = {
@@ -27,7 +32,9 @@ const ORPHANED: SourceView = {
   referenceCount: 0,
   referencingPolicies: [],
   config: [{ label: "Directory", value: "/data/archive" }],
-  docsTotal: null,
+  docsTotal: 45230,
+  docs24h: 312,
+  docs30d: 9870,
 };
 
 const meta: Meta<typeof SourceDetailCard> = {
@@ -35,6 +42,7 @@ const meta: Meta<typeof SourceDetailCard> = {
   component: SourceDetailCard,
   parameters: { layout: "padded" },
   args: {
+    docSeries: SAMPLE_SERIES,
     onClose: () => {},
     onEdit: () => {},
     onTogglePause: () => {},

frontend/portal/src/components/sources/SourceDetailCard.tsx

@@ -7,6 +7,7 @@ import "@portal/views/Sources.css";
 
 interface SourceDetailCardProps {
   source: SourceView;
+  docSeries: number[];
   onClose: () => void;
   onEdit: (source: SourceView) => void;
   onTogglePause: (source: SourceView) => void;
@@ -18,6 +19,7 @@ interface SourceDetailCardProps {
 /** Expanded detail for the selected source row, with edit/pause/delete actions. */
 export function SourceDetailCard({
   source,
+  docSeries,
   onClose,
   onEdit,
   onTogglePause,
@@ -55,7 +57,7 @@ export function SourceDetailCard({
         </button>
       </header>
 
-      <SourceDetailPanel source={source} />
+      <SourceDetailPanel source={source} docSeries={docSeries} />
 
       <div className="portal-sources__detail-actions">
         <Button

frontend/portal/src/components/sources/SourceDetailPanel.stories.tsx

@@ -1,6 +1,9 @@
 import type { Meta, StoryObj } from "@storybook/react-vite";
 import type { SourceView } from "@portal/api/sources";
 import { SourceDetailPanel } from "@portal/components/sources/SourceDetailPanel";
+import { sampleDailySeries } from "@portal/mocks/sampleDailySeries";
+
+const SAMPLE_SERIES = sampleDailySeries(330);
 
 const IN_USE: SourceView = {
   id: "src-claims",
@@ -16,7 +19,9 @@ const IN_USE: SourceView = {
     { label: "Directory", value: "/data/claims-intake" },
     { label: "Mode", value: "consume" },
   ],
-  docsTotal: null,
+  docsTotal: 45230,
+  docs24h: 312,
+  docs30d: 9870,
 };
 
 const ORPHANED: SourceView = {
@@ -27,7 +32,9 @@ const ORPHANED: SourceView = {
   referenceCount: 0,
   referencingPolicies: [],
   config: [{ label: "Directory", value: "/data/archive" }],
-  docsTotal: null,
+  docsTotal: 1180,
+  docs24h: 0,
+  docs30d: 0,
 };
 
 const meta: Meta<typeof SourceDetailPanel> = {
@@ -45,6 +52,8 @@ const meta: Meta<typeof SourceDetailPanel> = {
 export default meta;
 type Story = StoryObj<typeof SourceDetailPanel>;
 
-export const InUse: Story = { args: { source: IN_USE } };
+export const InUse: Story = {
+  args: { source: IN_USE, docSeries: SAMPLE_SERIES },
+};
 /** A source no policy references is called out as safe to delete. */
-export const Orphaned: Story = { args: { source: ORPHANED } };
+export const Orphaned: Story = { args: { source: ORPHANED, docSeries: [] } };

frontend/portal/src/components/sources/SourceDetailPanel.tsx

@@ -1,13 +1,24 @@
 import { useTranslation } from "react-i18next";
 import { Chip, StatTile } from "@shared/components";
 import type { SourceView } from "@portal/api/sources";
+import { Sparkline } from "@portal/components/sources/Sparkline";
 import "@portal/views/Sources.css";
 
+interface SourceDetailPanelProps {
+  source: SourceView;
+  /** The 30-day daily series for the sparkline, fetched per source when expanded. */
+  docSeries: number[];
+}
+
 /**
- * Expanded detail for a source row: its config (key/value) plus which policies
- * reference it. A 0-reference source is called out as safe to delete.
+ * Expanded detail for a source row: its config (key/value), the documents it has
+ * fed into runs, and which policies reference it (a 0-reference source is called
+ * out as safe to delete).
  */
-export function SourceDetailPanel({ source }: { source: SourceView }) {
+export function SourceDetailPanel({
+  source,
+  docSeries,
+}: SourceDetailPanelProps) {
   const { t } = useTranslation();
   return (
     <div className="portal-sources__detail">
@@ -38,9 +49,31 @@ export function SourceDetailPanel({ source }: { source: SourceView }) {
         )}
       </div>
 
-      <p className="portal-sources__muted">
-        {t("sources.detail.docsUntracked")}
-      </p>
+      <div className="portal-sources__detail-section">
+        <span className="portal-sources__detail-heading">
+          {t("sources.detail.documents")}
+        </span>
+        <div className="portal-sources__stat-grid">
+          <StatTile
+            label={t("sources.detail.docsTotal")}
+            value={source.docsTotal.toLocaleString()}
+          />
+          <StatTile
+            label={t("sources.detail.docs24h")}
+            value={source.docs24h.toLocaleString()}
+          />
+          <StatTile
+            label={t("sources.detail.docs30d")}
+            value={source.docs30d.toLocaleString()}
+          />
+        </div>
+        {docSeries.length > 0 && (
+          <Sparkline
+            data={docSeries}
+            ariaLabel={t("sources.detail.docsTrend")}
+          />
+        )}
+      </div>
     </div>
   );
 }

frontend/portal/src/components/sources/SourcesTable.stories.tsx

@@ -17,7 +17,9 @@ const SOURCES: SourceView[] = [
       { label: "Directory", value: "/data/claims-intake" },
       { label: "Mode", value: "consume" },
     ],
-    docsTotal: null,
+    docsTotal: 45230,
+    docs24h: 312,
+    docs30d: 9870,
   },
   {
     id: "src-archive",
@@ -27,7 +29,9 @@ const SOURCES: SourceView[] = [
     referenceCount: 0,
     referencingPolicies: [],
     config: [{ label: "Directory", value: "/data/archive" }],
-    docsTotal: null,
+    docsTotal: 1180,
+    docs24h: 0,
+    docs30d: 0,
   },
   {
     id: "src-legacy",
@@ -37,7 +41,9 @@ const SOURCES: SourceView[] = [
     referenceCount: 0,
     referencingPolicies: [],
     config: [{ label: "Directory", value: "/mnt/legacy" }],
-    docsTotal: null,
+    docsTotal: 48600,
+    docs24h: 0,
+    docs30d: 0,
   },
 ];
 

frontend/portal/src/components/sources/Sparkline.test.tsx

@@ -0,0 +1,48 @@
+import { describe, expect, it } from "vitest";
+import { render } from "@testing-library/react";
+import { Sparkline } from "@portal/components/sources/Sparkline";
+
+function pointsOf(container: HTMLElement): string[] {
+  const poly = container.querySelector("polyline");
+  return (poly?.getAttribute("points") ?? "")
+    .trim()
+    .split(/\s+/)
+    .filter(Boolean);
+}
+
+describe("Sparkline", () => {
+  it("draws one point per value", () => {
+    const { container } = render(<Sparkline data={[1, 5, 2, 8, 3]} />);
+    expect(pointsOf(container)).toHaveLength(5);
+  });
+
+  it("renders nothing for an empty series", () => {
+    const { container } = render(<Sparkline data={[]} />);
+    expect(container.querySelector("svg")).toBeNull();
+  });
+
+  it("renders a flat finite line for an all-zero series (no divide-by-zero)", () => {
+    const { container } = render(<Sparkline data={[0, 0, 0, 0]} />);
+    const ys = pointsOf(container).map((p) => Number(p.split(",")[1]));
+    expect(ys).toHaveLength(4);
+    expect(ys.every(Number.isFinite)).toBe(true);
+    // All equal: a zero series is a single horizontal line, not NaN-laden.
+    expect(new Set(ys).size).toBe(1);
+  });
+
+  it("puts the peak value at the top of the band", () => {
+    const { container } = render(<Sparkline data={[0, 10]} height={36} />);
+    const ys = pointsOf(container).map((p) => Number(p.split(",")[1]));
+    // y grows downward in SVG, so the larger value (10) sits at the smaller y.
+    expect(ys[1]).toBeLessThan(ys[0]);
+  });
+
+  it("exposes its aria-label", () => {
+    const { container } = render(
+      <Sparkline data={[1, 2]} ariaLabel="Docs trend" />,
+    );
+    expect(container.querySelector("svg")?.getAttribute("aria-label")).toBe(
+      "Docs trend",
+    );
+  });
+});

frontend/portal/src/components/sources/Sparkline.tsx

@@ -0,0 +1,53 @@
+import "@portal/views/Sources.css";
+
+interface SparklineProps {
+  /** Series values, oldest first. */
+  data: number[];
+  /** Drawing height in px; the width fills the container. */
+  height?: number;
+  ariaLabel?: string;
+}
+
+/**
+ * A tiny dependency-free trend line: normalises {@code data} to its own peak and draws a single
+ * polyline. The viewBox is fixed while the rendered width fills the container (the stroke stays
+ * crisp via non-scaling-stroke), so it adapts to any column without distorting the line weight.
+ */
+export function Sparkline({ data, height = 36, ariaLabel }: SparklineProps) {
+  if (data.length === 0) {
+    return null;
+  }
+  const width = 240;
+  const pad = 3;
+  const max = Math.max(...data, 1);
+  const stepX = data.length > 1 ? width / (data.length - 1) : 0;
+  const points = data
+    .map((value, i) => {
+      const x = i * stepX;
+      const y = pad + (1 - value / max) * (height - 2 * pad);
+      return `${x.toFixed(1)},${y.toFixed(1)}`;
+    })
+    .join(" ");
+
+  return (
+    <svg
+      className="portal-sparkline"
+      width="100%"
+      height={height}
+      viewBox={`0 0 ${width} ${height}`}
+      preserveAspectRatio="none"
+      role="img"
+      aria-label={ariaLabel}
+    >
+      <polyline
+        points={points}
+        fill="none"
+        stroke="currentColor"
+        strokeWidth="1.5"
+        strokeLinejoin="round"
+        strokeLinecap="round"
+        vectorEffect="non-scaling-stroke"
+      />
+    </svg>
+  );
+}

frontend/portal/src/mocks/handlers/sources.ts

@@ -7,6 +7,7 @@ import type {
   SourceView,
   SourcesResponse,
 } from "@portal/api/sources";
+import { sampleDailySeries } from "@portal/mocks/sampleDailySeries";
 
 /**
  * Stateful mock for the Sources surface so the portal works fully offline with
@@ -66,6 +67,27 @@ const references: Record<string, SourcePolicyRef[]> = {
   "src-contracts": [{ id: "pol_contract", name: "Contract Review" }],
 };
 
+/** Per-source document throughput, mirroring the backend's docsTotal / 24h / 30d. */
+const docCounts: Record<
+  string,
+  { total: number; last24h: number; last30d: number }
+> = {
+  "src-claims": { total: 45230, last24h: 312, last30d: 9870 },
+  "src-contracts": { total: 12840, last24h: 96, last30d: 2310 },
+  "src-archive": { total: 1180, last24h: 0, last30d: 0 },
+  "src-legacy": { total: 48600, last24h: 0, last30d: 0 },
+};
+
+function docsFor(id: string): {
+  total: number;
+  last24h: number;
+  last30d: number;
+  daily: number[];
+} {
+  const counts = docCounts[id] ?? { total: 0, last24h: 0, last30d: 0 };
+  return { ...counts, daily: sampleDailySeries(counts.last30d / 30) };
+}
+
 let store: StoredSource[] = seedSources();
 
 let idCounter = 0;
@@ -97,6 +119,7 @@ function toSourceView(
   source: StoredSource,
   refs: SourcePolicyRef[],
 ): SourceView {
+  const docs = docsFor(source.id);
   return {
     id: source.id,
     name: source.name,
@@ -105,7 +128,9 @@ function toSourceView(
     referenceCount: refs.length,
     referencingPolicies: refs,
     config: configRows(source.options),
-    docsTotal: null,
+    docsTotal: docs.total,
+    docs24h: docs.last24h,
+    docs30d: docs.last30d,
   };
 }
 
@@ -142,6 +167,13 @@ export const sourcesHandlers = [
     return HttpResponse.json(source);
   }),
 
+  http.get("/api/v1/sources/:id/document-counts", async ({ params }) => {
+    await delay(120);
+    const source = store.find((s) => s.id === params.id);
+    if (!source) return new HttpResponse(null, { status: 404 });
+    return HttpResponse.json(docsFor(source.id).daily);
+  }),
+
   http.post("/api/v1/sources", async ({ request }) => {
     await delay(120);
     const incoming = (await request.json()) as Source;

frontend/portal/src/mocks/sampleDailySeries.ts

@@ -0,0 +1,10 @@
+/**
+ * A gentle, deterministic 30-point daily series averaging ~`avg`/day, shaped by a sine wave so mock
+ * and Storybook sparklines have something to draw. Shared so the mock handler and the source stories
+ * stay in sync instead of each carrying their own copy of the formula.
+ */
+export function sampleDailySeries(avg: number): number[] {
+  return Array.from({ length: 30 }, (_, i) =>
+    Math.round(avg * (0.5 + Math.abs(Math.sin((i + 1) / 3)))),
+  );
+}

frontend/portal/src/views/Pipelines.test.tsx

@@ -93,7 +93,9 @@ const SOURCES: SourcesResponse = {
       referenceCount: 1,
       referencingPolicies: [],
       config: [],
-      docsTotal: null,
+      docsTotal: 0,
+      docs24h: 0,
+      docs30d: 0,
     },
   ],
 };

frontend/portal/src/views/Sources.css

@@ -238,6 +238,13 @@
   font-weight: 600;
 }
 
+.portal-sparkline {
+  display: block;
+  width: 100%;
+  margin-top: 0.5rem;
+  color: var(--color-blue);
+}
+
 .portal-sources__chips {
   display: flex;
   flex-wrap: wrap;

frontend/portal/src/views/Sources.test.tsx

@@ -16,11 +16,13 @@ vi.mock("react-i18next", () => ({
 
 const fetchSources = vi.fn();
 const fetchSource = vi.fn();
+const fetchSourceDocCounts = vi.fn();
 const createSource = vi.fn();
 const deleteSource = vi.fn();
 vi.mock("@portal/api/sources", () => ({
   fetchSources: () => fetchSources(),
   fetchSource: (id: string) => fetchSource(id),
+  fetchSourceDocCounts: (id: string) => fetchSourceDocCounts(id),
   createSource: (source: unknown) => createSource(source),
   deleteSource: (id: string) => deleteSource(id),
 }));
@@ -43,7 +45,9 @@ const RESPONSE: SourcesResponse = {
         { id: "pol-2", name: "Classification" },
       ],
       config: [{ label: "Directory", value: "/data/incoming" }],
-      docsTotal: null,
+      docsTotal: 1240,
+      docs24h: 18,
+      docs30d: 540,
     },
     {
       id: "src-orphan",
@@ -53,7 +57,9 @@ const RESPONSE: SourcesResponse = {
       referenceCount: 0,
       referencingPolicies: [],
       config: [{ label: "Directory", value: "/tmp/scratch" }],
-      docsTotal: null,
+      docsTotal: 1240,
+      docs24h: 18,
+      docs30d: 540,
     },
   ],
 };
@@ -70,6 +76,8 @@ describe("Sources view", () => {
   beforeEach(() => {
     fetchSources.mockReset();
     fetchSource.mockReset();
+    fetchSourceDocCounts.mockReset();
+    fetchSourceDocCounts.mockResolvedValue([]);
     createSource.mockReset();
     deleteSource.mockReset();
   });

frontend/portal/src/views/Sources.tsx

@@ -14,6 +14,7 @@ import {
   createSource,
   deleteSource,
   fetchSource,
+  fetchSourceDocCounts,
   fetchSources,
   type Source,
   type SourcesResponse,
@@ -49,6 +50,21 @@ export function Sources() {
   const sources = data?.sources ?? [];
   const expanded = sources.find((s) => s.id === expandedId) ?? null;
 
+  // The 30-day sparkline series lives off the list endpoint; fetch it for the one
+  // expanded row only (empty while collapsed, so no request fires).
+  const docSeriesState = useAsync<{ id: string; series: number[] }>(
+    () =>
+      expandedId
+        ? fetchSourceDocCounts(expandedId).then((series) => ({
+            id: expandedId,
+            series,
+          }))
+        : Promise.resolve({ id: "", series: [] }),
+    [expandedId],
+  );
+  const docSeries =
+    docSeriesState.data?.id === expandedId ? docSeriesState.data.series : [];
+
   function openCreate() {
     setEditingSource(null);
     setWizardOpen(true);
@@ -166,6 +182,7 @@ export function Sources() {
       {expanded && (
         <SourceDetailCard
           source={expanded}
+          docSeries={docSeries}
           onClose={() => setExpandedId(null)}
           onEdit={openEdit}
           onTogglePause={togglePause}