PAYG: size-scaled units + per-input-file PDF count + run-id grouping (#6957)

Reworks the Processor (PAYG) meter to **size-scaled units** while
keeping a true **PDF count** visible and distinct from units, and
replaces the fragile content+time lineage grouping with **explicit
per-run grouping**. Built as one PR across three slices.

> Status: **all three slices committed + verified.** `:saas` payg suite
green (418 tests, 0 failures); FE green (typecheck 0, 1260 tests, lint
0, format 0). Remaining before it takes effect in prod: run the
size-scaled default-policy SQL (below) in the Supabase SQL editor +
attach the $0.01/unit Stripe price.

## Model (what we're implementing)
- **Size scaling**: 1 unit per 50 MiB (bytes only, no page charge, no
cap). *(policy-row config, applied separately via SQL.)*
- **Charge = number of input files**: split (1→N outputs) = 1 charge;
merge (N→1) = N charges. `doc_count` = input files, fixed at open;
joined steps add 0.
- **Grouping by run id, not time**: a pipeline/policy/AI run = one
`run_id`; its tool sub-steps group into one charge (content-lineage
still maps split/merge journeys *within* the run). Two separate runs on
identical bytes = two charges. The 5-min window survives only as a
stale-job janitor.
- **10-tool split kept**: within a run's single-file lineage, an 11th
tool run opens a 2nd charge (step limit 10).
- **Count vs units surfaced**: usage page shows unique PDFs,
per-category (automation/AI/API) counts + units, and how many PDFs hit a
size multiplier with avg units/PDF.

## Slice 1 — run-id grouping (behavioural core)
- `AutomationRunContext` (common) — thread-scoped run id.
- `InternalApiClient` — stamps `X-Stirling-Run-Id`.
- Orchestrators open a run scope **on the worker thread that
dispatches** (async-safe): `PipelineProcessor.runPipelineAgainstFiles`,
`PolicyEngine.runToCompletion` (uses `run.getRunId()`),
`AiWorkflowService.orchestrate`.
- `ChargeContext` + `JobContext`: add `runId`; the charge interceptor
reads the header.
- `JobService.joinOrOpen`: `runId == null` → always open fresh
(standalone never joins); non-null → match scoped to the same `run_id`.
`JpaJobLineageStore`/`JobArtifactHashRepository`: add `run_id` filter to
the match query. Step-limit 10 unchanged.

## Slice 2 — doc_count + document_fingerprint
- V33 migration + entity fields.
- `JobService.openFresh`: set `docCount = inputs.size()`, compute
`document_fingerprint` from input signatures, and denormalise both onto
the DEBIT row in `JobChargeService.recordLedgerDebit`.

## Slice 3 — usage analytics API + FE
- `WalletLedgerRepository`: per-category `SUM(units)` +
`SUM(doc_count)`, `COUNT(DISTINCT document_fingerprint)`, and count of
rows whose units exceed their doc_count (a size multiplier fired), over
the period.
- `WalletSnapshotResponse` + `PaygWalletController`: add `categoryDocs`,
`docsProcessedThisPeriod`, `uniquePdfsThisPeriod`,
`sizeMultiplierPdfsThisPeriod`.
- FE `types.ts` + `PdfsProcessedCard` + `useWallet` + `walletFixtures` +
i18n: headline is the **PDF count**; a summary line shows "{unique}
unique · {units} meter units · {avg} avg units/PDF"; the split bar is
per-category PDF counts; a size-multiplier line shows how many PDFs
scaled. Count is separated from meter units so a 5-unit large PDF reads
as "1 PDF, 5 units".

## Config (out of PR — run in the Supabase SQL editor)
Wrap in one transaction. The partial-unique `is_default` index only
allows one default, so the old default is flipped off **before** the new
one is inserted. The new policy carries the prior default's
`free_tier_units` forward (change the literal if the launch grant should
differ).
```sql
BEGIN;

-- 1) flip default off the current policy + close its effective window
UPDATE stirling_pdf.pricing_policy
SET is_default = FALSE, effective_to = now()
WHERE is_default = TRUE;

-- 2) new default: 1 unit / 5 MiB, no page charge, no scaling cap.
--    free_tier_units carried from whatever the last policy granted (COALESCE→0).
INSERT INTO stirling_pdf.pricing_policy
  (version, effective_from, doc_pages_per_unit, doc_bytes_per_unit,
   min_charge_units, file_unit_cap, free_tier_units, is_default, notes, created_by)
VALUES
  ('v2-size-scaled-2026-07', now(),
   2147483647,        -- doc_pages_per_unit = INT_MAX → pages never drive units
   52428800,          -- doc_bytes_per_unit = 50 MiB → +1 unit per 50 MiB
   1,                 -- min_charge_units
   2147483647,        -- file_unit_cap = INT_MAX → no cap on size scaling
   COALESCE((SELECT free_tier_units FROM stirling_pdf.pricing_policy
             ORDER BY effective_from DESC LIMIT 1), 0),
   TRUE, 'Size-scaled: 1 unit/5MiB, bytes only, no cap', 'connor');

-- 3) per-source step limits: standalone ops = own charge; pipelines split at 10
INSERT INTO stirling_pdf.pricing_policy_step_limit (policy_id, job_source, step_limit)
SELECT p.policy_id, s.src, s.lim
FROM stirling_pdf.pricing_policy p
CROSS JOIN (VALUES
  ('WEB',1),('API',1),('DESKTOP_APP',1),('LINKED_INSTANCE',1),('PIPELINE',10)
) AS s(src, lim)
WHERE p.version = 'v2-size-scaled-2026-07';

-- 4) attach the $0.01/unit Stripe price (you handle the real price id)
INSERT INTO stirling_pdf.pricing_policy_stripe_price (policy_id, stripe_price_id)
SELECT policy_id, 'price_XXXXXXXX'
FROM stirling_pdf.pricing_policy WHERE version = 'v2-size-scaled-2026-07';

COMMIT;
```
Note: the `free_tier_units` subquery reads the most-recent policy
*before* the insert — run it as written (the new row doesn't exist yet
at step 2's SELECT).

## Self-hosted parity — tracked follow-up (not in this PR)
Combined-billing (`stirling.billing.account-link.enabled`) is a
**separate metering engine** (`app/proprietary/accountlink` —
`UsageMeterService`/`LocalUsageService`/`UsageSyncService`). The unit
*math* is shared (`DocumentUnitCalculator`), so size scaling matches
once the policy is pushed. But run-id grouping, `doc_count`, and
fingerprints must be mirrored there, and the usage-sync protocol
extended to report counts/fingerprints, before the self-hosted usage
page shows the same breakdown. Frozen/deferred, so this PR does SaaS;
self-hosted mirrors when it ships.

app/common/src/main/java/stirling/software/common/service/AutomationRunContext.java

@@ -0,0 +1,56 @@
+package stirling.software.common.service;
+
+/**
+ * Thread-scoped correlation id for one automation run — a single pipeline, policy, or AI-workflow
+ * execution over its input file(s).
+ *
+ * <p>Automations dispatch each tool step as a separate internal loopback POST via {@link
+ * InternalApiClient}. The orchestrator opens a run scope around its dispatch loop; {@code
+ * InternalApiClient} reads {@link #current()} and stamps it on every sub-step request as {@link
+ * #RUN_ID_HEADER}. The SaaS PAYG interceptor uses that header so all sub-steps of ONE run group
+ * into a single charge, while two <em>separate</em> runs that happen to touch identical bytes stay
+ * distinct charges (the old content+time-window grouping merged them).
+ *
+ * <p>Sub-steps dispatch synchronously on the orchestrator's own thread (loopback {@code
+ * RestTemplate}), so this ThreadLocal is visible to {@code InternalApiClient}. The id then crosses
+ * to the receiving request thread via the HTTP header — never via this ThreadLocal.
+ *
+ * <p>No-op when the id is absent (a standalone tool call): the interceptor treats a missing run id
+ * as "its own charge", which is exactly what a one-off call should be.
+ */
+public final class AutomationRunContext {
+
+    /** Header carrying the run id on internal sub-step dispatches. */
+    public static final String RUN_ID_HEADER = "X-Stirling-Run-Id";
+
+    private static final ThreadLocal<String> CURRENT = new ThreadLocal<>();
+
+    private AutomationRunContext() {}
+
+    /**
+     * Opens a run scope on the current thread. Returns an {@link AutoCloseable} that restores the
+     * previously-active id (nesting-safe) — use in try-with-resources around the dispatch loop.
+     */
+    public static Scope open(String runId) {
+        String previous = CURRENT.get();
+        CURRENT.set(runId);
+        return () -> {
+            if (previous == null) {
+                CURRENT.remove();
+            } else {
+                CURRENT.set(previous);
+            }
+        };
+    }
+
+    /** The run id active on this thread, or {@code null} when not inside a run scope. */
+    public static String current() {
+        return CURRENT.get();
+    }
+
+    /** AutoCloseable whose {@link #close()} declares no checked exception. */
+    public interface Scope extends AutoCloseable {
+        @Override
+        void close();
+    }
+}

app/common/src/main/java/stirling/software/common/service/InternalApiClient.java

@@ -111,6 +111,14 @@ public class InternalApiClient {
         // step inside a policy run must bill as AUTOMATION, not AI). Set unconditionally because
         // every caller of this dispatcher is an automation surface by design.
         headers.add(AUTOMATION_HEADER, "true");
+        // Propagate the current automation run id (set by the orchestrator around its dispatch
+        // loop) so the PAYG interceptor groups every sub-step of this one run into a single charge,
+        // and never merges two separate runs that happen to touch identical bytes. Absent → the
+        // receiving call is treated as standalone. See AutomationRunContext.
+        String runId = AutomationRunContext.current();
+        if (runId != null && !runId.isEmpty()) {
+            headers.add(AutomationRunContext.RUN_ID_HEADER, runId);
+        }
 
         // A no-file ai/tools call (e.g. create-pdf-from-html-agent) sends only string params, so
         // without this RestTemplate would use urlencoded instead of the multipart the controller

app/core/src/main/java/stirling/software/SPDF/controller/api/pipeline/PipelineProcessor.java

@@ -10,6 +10,7 @@ import java.util.List;
 import java.util.Locale;
 import java.util.Map;
 import java.util.Map.Entry;
+import java.util.UUID;
 
 import org.springframework.core.io.FileSystemResource;
 import org.springframework.core.io.Resource;
@@ -27,6 +28,7 @@ import stirling.software.SPDF.model.PipelineConfig;
 import stirling.software.SPDF.model.PipelineOperation;
 import stirling.software.SPDF.model.PipelineResult;
 import stirling.software.SPDF.service.ApiDocService;
+import stirling.software.common.service.AutomationRunContext;
 import stirling.software.common.service.InternalApiClient;
 import stirling.software.common.util.TempFileManager;
 import stirling.software.common.util.ZipExtractionUtils;
@@ -71,6 +73,17 @@ public class PipelineProcessor {
 
     PipelineResult runPipelineAgainstFiles(List<Resource> outputFiles, PipelineConfig config)
             throws Exception {
+        // One pipeline execution = one automation run. Scope a run id so every tool sub-step
+        // dispatched via InternalApiClient groups into a single charge on the SaaS billing side
+        // (see AutomationRunContext); pipeline steps run synchronously on this thread.
+        try (AutomationRunContext.Scope ignored =
+                AutomationRunContext.open(UUID.randomUUID().toString())) {
+            return runPipelineAgainstFilesInternal(outputFiles, config);
+        }
+    }
+
+    private PipelineResult runPipelineAgainstFilesInternal(
+            List<Resource> outputFiles, PipelineConfig config) throws Exception {
         PipelineResult result = new PipelineResult();
 
         ByteArrayOutputStream logStream = new ByteArrayOutputStream();

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

@@ -21,6 +21,7 @@ import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.job.ResultFile;
+import stirling.software.common.service.AutomationRunContext;
 import stirling.software.common.service.FileStorage;
 import stirling.software.common.service.InternalApiTimeoutException;
 import stirling.software.common.service.JobOwnershipService;
@@ -199,63 +200,78 @@ public class PolicyEngine {
             PolicyProgressListener listener,
             CompletableFuture<PolicyRun> completion) {
         String runId = run.getRunId();
-        try {
-            run.markRunning();
-            PolicyExecutionResult result =
-                    stepExecutor.execute(run.getDefinition(), inputs, listener);
-            OutputSpec output = run.getDefinition().output();
-            List<ResultFile> outputs =
-                    sinkFor(output)
-                            .deliver(
-                                    new OutputDelivery(runId, run.getPolicyId()),
-                                    result.files(),
-                                    output);
-            taskManager.setMultipleFileResults(runId, outputs);
-            taskManager.setComplete(runId);
-            run.complete(outputs);
-        } catch (PolicyInputRequiredException e) {
-            // Expected path: suspend rather than fail. Persist intermediates as fileIds so the run
-            // can resume after this worker thread is gone.
-            WaitState wait = suspend(e);
-            run.waitForInput(wait);
-            taskManager.addNote(runId, "Waiting for input: " + e.getMessage());
-        } catch (InternalApiTimeoutException e) {
-            String message = toolTimeoutMessage(e);
-            log.error(
-                    "Policy run {} timed out on {}: {}",
-                    runId,
-                    e.getEndpointPath(),
-                    e.getMessage());
-            run.fail(message);
-            taskManager.setError(runId, message);
-        } catch (RestClientResponseException e) {
-            // A downstream tool call returned an error status. When it's a structured entitlement
-            // response (401/402 with a JSON `error` sentinel), surface that code onto the run so
-            // the
-            // client can react — e.g. pop the usage-limit modal — instead of only seeing a generic
-            // failure. We don't interpret the code here (that would couple this module to the saas
-            // billing layer); we just pass it through for the client to map. Other statuses fall
-            // through to the generic failure below.
-            String code = DownstreamEntitlementError.extractCode(e);
-            if (code != null) {
-                log.info("Policy run {} blocked by downstream entitlement gate ({})", runId, code);
-                String message = "Usage limit reached";
-                run.failWithCode(message, code, DownstreamEntitlementError.extractSubscribed(e));
-                taskManager.setError(runId, message);
-            } else {
-                String message = "Policy run failed: " + e.getMessage();
-                log.error("Policy run {} failed (downstream HTTP error)", runId, e);
+        // One policy run = one automation run. Scope the run id on this worker thread (the async
+        // hop already happened) so every tool sub-step dispatched via InternalApiClient groups into
+        // a single charge, and two separate policy runs on the same document stay distinct charges.
+        try (AutomationRunContext.Scope runScope = AutomationRunContext.open(runId)) {
+            try {
+                run.markRunning();
+                PolicyExecutionResult result =
+                        stepExecutor.execute(run.getDefinition(), inputs, listener);
+                OutputSpec output = run.getDefinition().output();
+                List<ResultFile> outputs =
+                        sinkFor(output)
+                                .deliver(
+                                        new OutputDelivery(runId, run.getPolicyId()),
+                                        result.files(),
+                                        output);
+                taskManager.setMultipleFileResults(runId, outputs);
+                taskManager.setComplete(runId);
+                run.complete(outputs);
+            } catch (PolicyInputRequiredException e) {
+                // Expected path: suspend rather than fail. Persist intermediates as fileIds so the
+                // run
+                // can resume after this worker thread is gone.
+                WaitState wait = suspend(e);
+                run.waitForInput(wait);
+                taskManager.addNote(runId, "Waiting for input: " + e.getMessage());
+            } catch (InternalApiTimeoutException e) {
+                String message = toolTimeoutMessage(e);
+                log.error(
+                        "Policy run {} timed out on {}: {}",
+                        runId,
+                        e.getEndpointPath(),
+                        e.getMessage());
                 run.fail(message);
                 taskManager.setError(runId, message);
+            } catch (RestClientResponseException e) {
+                // A downstream tool call returned an error status. When it's a structured
+                // entitlement
+                // response (401/402 with a JSON `error` sentinel), surface that code onto the run
+                // so
+                // the
+                // client can react — e.g. pop the usage-limit modal — instead of only seeing a
+                // generic
+                // failure. We don't interpret the code here (that would couple this module to the
+                // saas
+                // billing layer); we just pass it through for the client to map. Other statuses
+                // fall
+                // through to the generic failure below.
+                String code = DownstreamEntitlementError.extractCode(e);
+                if (code != null) {
+                    log.info(
+                            "Policy run {} blocked by downstream entitlement gate ({})",
+                            runId,
+                            code);
+                    String message = "Usage limit reached";
+                    run.failWithCode(
+                            message, code, DownstreamEntitlementError.extractSubscribed(e));
+                    taskManager.setError(runId, message);
+                } else {
+                    String message = "Policy run failed: " + e.getMessage();
+                    log.error("Policy run {} failed (downstream HTTP error)", runId, e);
+                    run.fail(message);
+                    taskManager.setError(runId, message);
+                }
+            } catch (Exception e) {
+                String message = "Policy run failed: " + e.getMessage();
+                log.error("Policy run {} failed", runId, e);
+                run.fail(message);
+                taskManager.setError(runId, message);
+            } finally {
+                // Always resolve so stream/await callers unblock.
+                completion.complete(run);
             }
-        } catch (Exception e) {
-            String message = "Policy run failed: " + e.getMessage();
-            log.error("Policy run {} failed", runId, e);
-            run.fail(message);
-            taskManager.setError(runId, message);
-        } finally {
-            // Always resolve so stream/await callers unblock.
-            completion.complete(run);
         }
     }
 

app/proprietary/src/main/java/stirling/software/proprietary/service/AiWorkflowService.java

@@ -7,6 +7,7 @@ import java.util.ArrayList;
 import java.util.LinkedHashMap;
 import java.util.List;
 import java.util.Map;
+import java.util.UUID;
 import java.util.stream.Collectors;
 
 import org.apache.commons.io.FilenameUtils;
@@ -27,6 +28,7 @@ import lombok.Data;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.model.ApplicationProperties;
+import stirling.software.common.service.AutomationRunContext;
 import stirling.software.common.service.CustomPDFDocumentFactory;
 import stirling.software.common.service.FileStorage;
 import stirling.software.common.service.InternalApiTimeoutException;
@@ -156,33 +158,41 @@ public class AiWorkflowService {
             throws IOException {
         validateRequest(request);
 
-        // Key by opaque file id, not filename. Filenames aren't guaranteed unique across an
-        // upload (users can rotate the same 'scan.pdf' twice), and the engine identifies files
-        // by id in every response shape that asks Java to look a file up again.
-        Map<String, MultipartFile> filesById = new LinkedHashMap<>();
-        List<AiFile> files = new ArrayList<>();
-        for (AiWorkflowFileInput fileInput : request.getFileInputs()) {
-            MultipartFile multipartFile = fileInput.getFileInput();
-            AiFile aiFile =
-                    new AiFile(
-                            fileIdStrategy.idFor(multipartFile),
-                            multipartFile.getOriginalFilename());
-            filesById.put(aiFile.getId(), multipartFile);
-            files.add(aiFile);
-        }
+        // One AI orchestration = one automation run. Scope a run id (on whichever thread runs
+        // orchestrate — request thread for sync, stream-executor for streaming) so every tool
+        // sub-step it dispatches via PolicyExecutor → InternalApiClient groups into one charge.
+        try (AutomationRunContext.Scope ignored =
+                AutomationRunContext.open(UUID.randomUUID().toString())) {
 
-        WorkflowTurnRequest initialRequest = new WorkflowTurnRequest();
-        initialRequest.setUserMessage(request.getUserMessage().trim());
-        initialRequest.setFiles(files);
-        initialRequest.setConversationHistory(new ArrayList<>(request.getConversationHistory()));
-        initialRequest.setEnabledEndpoints(endpointResolver.getEnabledEndpointUrls());
-        listener.onProgress(AiWorkflowProgressEvent.of(AiWorkflowPhase.ANALYZING));
+            // Key by opaque file id, not filename. Filenames aren't guaranteed unique across an
+            // upload (users can rotate the same 'scan.pdf' twice), and the engine identifies files
+            // by id in every response shape that asks Java to look a file up again.
+            Map<String, MultipartFile> filesById = new LinkedHashMap<>();
+            List<AiFile> files = new ArrayList<>();
+            for (AiWorkflowFileInput fileInput : request.getFileInputs()) {
+                MultipartFile multipartFile = fileInput.getFileInput();
+                AiFile aiFile =
+                        new AiFile(
+                                fileIdStrategy.idFor(multipartFile),
+                                multipartFile.getOriginalFilename());
+                filesById.put(aiFile.getId(), multipartFile);
+                files.add(aiFile);
+            }
 
-        WorkflowState state = new WorkflowState.Pending(initialRequest);
-        while (state instanceof WorkflowState.Pending pending) {
-            state = advance(pending.request(), filesById, listener);
+            WorkflowTurnRequest initialRequest = new WorkflowTurnRequest();
+            initialRequest.setUserMessage(request.getUserMessage().trim());
+            initialRequest.setFiles(files);
+            initialRequest.setConversationHistory(
+                    new ArrayList<>(request.getConversationHistory()));
+            initialRequest.setEnabledEndpoints(endpointResolver.getEnabledEndpointUrls());
+            listener.onProgress(AiWorkflowProgressEvent.of(AiWorkflowPhase.ANALYZING));
+
+            WorkflowState state = new WorkflowState.Pending(initialRequest);
+            while (state instanceof WorkflowState.Pending pending) {
+                state = advance(pending.request(), filesById, listener);
+            }
+            return ((WorkflowState.Terminal) state).response();
         }
-        return ((WorkflowState.Terminal) state).response();
     }
 
     private WorkflowState advance(

app/saas/src/main/java/stirling/software/saas/ai/controller/AiCreateController.java

@@ -117,7 +117,8 @@ public class AiCreateController {
                             user.getTeam().getId(),
                             source,
                             ProcessType.SINGLE_TOOL,
-                            BillingCategory.AI);
+                            BillingCategory.AI,
+                            null);
             jobChargeService.chargeStandalone(ctx, 1);
         } catch (RuntimeException e) {
             log.warn(

app/saas/src/main/java/stirling/software/saas/payg/api/PaygWalletController.java

@@ -173,7 +173,9 @@ public class PaygWalletController {
         int spend = clampToInt(snap.periodSpendUnits());
         Integer limit = snap.periodCapUnits() != null ? clampToInt(snap.periodCapUnits()) : null;
 
-        CategoryBreakdown breakdown = buildBreakdown(teamId, snap.periodStart(), snap.periodEnd());
+        BreakdownPair breakdowns = buildBreakdowns(teamId, snap.periodStart(), snap.periodEnd());
+        UsageAnalytics analytics =
+                buildUsageAnalytics(teamId, snap.periodStart(), snap.periodEnd());
 
         // Estimated bill = paid (Stripe-metered) docs this period × rate — the free portion was
         // already netted out at charge time, so this is the metered total, not spend − grant.
@@ -203,30 +205,63 @@ public class PaygWalletController {
                         noCap,
                         billing.subscriptionId(),
                         spend,
-                        breakdown,
+                        breakdowns.units(),
                         members,
-                        buildActivity(teamId));
+                        buildActivity(teamId),
+                        breakdowns.docs(),
+                        analytics.docsProcessed(),
+                        analytics.uniquePdfs(),
+                        analytics.sizeMultiplierPdfs());
         return ResponseEntity.ok(body);
     }
 
-    private CategoryBreakdown buildBreakdown(
+    /** Per-category size-scaled units + input-file counts for the same window. */
+    private record BreakdownPair(CategoryBreakdown units, CategoryBreakdown docs) {}
+
+    /** Period usage analytics: total input files, unique PDFs, and size-multiplier files. */
+    private record UsageAnalytics(int docsProcessed, int uniquePdfs, int sizeMultiplierPdfs) {}
+
+    private BreakdownPair buildBreakdowns(
             Long teamId, LocalDateTime periodStart, LocalDateTime periodEnd) {
-        Map<BillingCategory, Long> byCategory = new HashMap<>();
+        Map<BillingCategory, Long> units = new HashMap<>();
+        Map<BillingCategory, Long> docs = new HashMap<>();
         for (Object[] row :
-                ledgerRepo.sumPeriodAmountByCategory(
+                ledgerRepo.sumPeriodByCategoryWithDocs(
                         teamId, LedgerEntryType.DEBIT, periodStart, periodEnd)) {
-            if (row.length >= 2
-                    && row[0] instanceof BillingCategory cat
-                    && row[1] instanceof Number n) {
-                byCategory.put(cat, n.longValue());
+            if (row.length >= 3 && row[0] instanceof BillingCategory cat) {
+                if (row[1] instanceof Number u) {
+                    units.put(cat, u.longValue());
+                }
+                if (row[2] instanceof Number d) {
+                    docs.put(cat, d.longValue());
+                }
             }
         }
+        return new BreakdownPair(categoryBreakdown(units), categoryBreakdown(docs));
+    }
+
+    private static CategoryBreakdown categoryBreakdown(Map<BillingCategory, Long> byCategory) {
         return new CategoryBreakdown(
                 clampToInt(byCategory.getOrDefault(BillingCategory.API, 0L)),
                 clampToInt(byCategory.getOrDefault(BillingCategory.AI, 0L)),
                 clampToInt(byCategory.getOrDefault(BillingCategory.AUTOMATION, 0L)));
     }
 
+    private UsageAnalytics buildUsageAnalytics(
+            Long teamId, LocalDateTime periodStart, LocalDateTime periodEnd) {
+        List<Object[]> rows =
+                ledgerRepo.periodUsageAnalytics(
+                        teamId, LedgerEntryType.DEBIT, periodStart, periodEnd);
+        Object[] row = rows.isEmpty() ? null : rows.get(0);
+        return new UsageAnalytics(analyticsInt(row, 0), analyticsInt(row, 1), analyticsInt(row, 2));
+    }
+
+    private static int analyticsInt(Object[] row, int idx) {
+        return row != null && row.length > idx && row[idx] instanceof Number n
+                ? clampToInt(n.longValue())
+                : 0;
+    }
+
     /**
      * Latest ledger entries shaped for the FE activity feed. DEBITs read as usage, REFUNDs as
      * credits-back; system entries without a category render as {@code other}.
@@ -446,6 +481,10 @@ public class PaygWalletController {
                 0,
                 new CategoryBreakdown(0, 0, 0),
                 List.of(),
-                Collections.emptyList());
+                Collections.emptyList(),
+                new CategoryBreakdown(0, 0, 0),
+                0,
+                0,
+                0);
     }
 }

app/saas/src/main/java/stirling/software/saas/payg/api/WalletSnapshotResponse.java

@@ -75,7 +75,21 @@ public record WalletSnapshotResponse(
         int spendUnitsThisPeriod,
         CategoryBreakdown categoryBreakdown,
         List<MemberRow> members,
-        List<ActivityRow> recent) {
+        List<ActivityRow> recent,
+        CategoryBreakdown categoryDocs,
+        int docsProcessedThisPeriod,
+        int uniquePdfsThisPeriod,
+        int sizeMultiplierPdfsThisPeriod) {
+
+    // The count dimension, kept distinct from units (which now scale with file size):
+    //   categoryDocs                — per-category INPUT-file counts (parallel to
+    // categoryBreakdown,
+    //                                 which stays the size-scaled unit totals)
+    //   docsProcessedThisPeriod     — total input files processed this period (Σ doc_count)
+    //   uniquePdfsThisPeriod        — distinct input documents (a file hit by N operations counts
+    //                                 once)
+    //   sizeMultiplierPdfsThisPeriod— input files on charges where the size multiplier applied
+    //                                 (units billed > input files)
 
     /** Per-category breakdown of {@code spendUnitsThisPeriod} for the in-app analytics widget. */
     public record CategoryBreakdown(int api, int ai, int automation) {}

app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeContext.java

@@ -14,13 +14,19 @@ import stirling.software.saas.payg.model.ProcessType;
  * the interceptor before this context is built. Manual UI tools never reach {@code openProcess}
  * (they short-circuit on {@link BillingCategory#BYPASSED}); any context constructed here therefore
  * carries one of {@code API}, {@code AI}, or {@code AUTOMATION}.
+ *
+ * <p>{@code runId} is the automation-run correlation id ({@code X-Stirling-Run-Id}) when this call
+ * is a sub-step of a pipeline / policy / AI-workflow run, else {@code null} (a standalone tool
+ * call). Lineage joins are scoped to a single run id: a null run id never joins (each standalone
+ * call is its own charge), and two separate runs never merge even on identical bytes.
  */
 public record ChargeContext(
         Long ownerUserId,
         Long ownerTeamId,
         JobSource source,
         ProcessType processType,
-        BillingCategory billingCategory) {
+        BillingCategory billingCategory,
+        String runId) {
 
     public ChargeContext {
         if (ownerUserId == null) {
@@ -36,4 +42,17 @@ public record ChargeContext(
             throw new IllegalArgumentException("billingCategory is required");
         }
     }
+
+    /**
+     * Convenience for callers with no automation-run context — a standalone tool call ({@code
+     * runId} = {@code null}, so it never lineage-joins and is always its own charge).
+     */
+    public ChargeContext(
+            Long ownerUserId,
+            Long ownerTeamId,
+            JobSource source,
+            ProcessType processType,
+            BillingCategory billingCategory) {
+        this(ownerUserId, ownerTeamId, source, processType, billingCategory, null);
+    }
 }

app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java

@@ -113,7 +113,8 @@ public class JobChargeService {
                         ctx.source(),
                         ctx.processType(),
                         policy.getId(),
-                        stepLimit);
+                        stepLimit,
+                        ctx.runId());
 
         List<Path> paths = inputs.stream().map(JobInput::path).toList();
         JoinOrOpenResult result = jobService.joinOrOpen(jobCtx, paths);
@@ -122,14 +123,23 @@ public class JobChargeService {
             return new ChargeOutcome(result.job().getId(), 0, ChargeOutcome.Disposition.JOINED);
         }
 
+        ProcessingJob job = result.job();
         int units = computeUnits(inputs, policy);
-        result.job().setDocUnits(units);
+        job.setDocUnits(units);
 
         int freeUsed = consumeFreeGrant(ctx, units);
-        recordShadowRow(ctx, result.job().getId(), policy.getId(), units, freeUsed);
-        recordLedgerDebit(ctx, result.job().getId(), policy.getId(), units);
+        recordShadowRow(ctx, job.getId(), policy.getId(), units, freeUsed);
+        // doc_count + fingerprint were set on the fresh job by JobService.openFresh; carry them
+        // onto the ledger DEBIT so usage analytics query one table.
+        recordLedgerDebit(
+                ctx,
+                job.getId(),
+                policy.getId(),
+                units,
+                job.getDocCount(),
+                job.getDocumentFingerprint());
 
-        return new ChargeOutcome(result.job().getId(), units, ChargeOutcome.Disposition.OPENED);
+        return new ChargeOutcome(job.getId(), units, ChargeOutcome.Disposition.OPENED);
     }
 
     /**
@@ -164,12 +174,19 @@ public class JobChargeService {
                         ctx.source(),
                         ctx.processType(),
                         policy.getId(),
-                        stepLimit);
+                        stepLimit,
+                        ctx.runId());
         ProcessingJob job = jobService.open(jobCtx, chargeUnits);
 
         int freeUsed = consumeFreeGrant(ctx, chargeUnits);
         recordShadowRow(ctx, job.getId(), policy.getId(), chargeUnits, freeUsed);
-        recordLedgerDebit(ctx, job.getId(), policy.getId(), chargeUnits);
+        recordLedgerDebit(
+                ctx,
+                job.getId(),
+                policy.getId(),
+                chargeUnits,
+                job.getDocCount(),
+                job.getDocumentFingerprint());
 
         // Close immediately — nothing will lineage-join a standalone job — so the paid portion
         // meters via the same afterCommit hook + idempotency key as a normal process completion.
@@ -216,7 +233,12 @@ public class JobChargeService {
      * Skipped for {@code BYPASSED} / uncategorised calls — manual UI work is never billed.
      */
     private void recordLedgerDebit(
-            ChargeContext ctx, java.util.UUID jobId, Long policyId, int units) {
+            ChargeContext ctx,
+            java.util.UUID jobId,
+            Long policyId,
+            int units,
+            int docCount,
+            String documentFingerprint) {
         BillingCategory category = ctx.billingCategory();
         if (category == null || category == BillingCategory.BYPASSED) {
             return;
@@ -231,6 +253,10 @@ public class JobChargeService {
         entry.setReferenceId(jobId.toString());
         entry.setPolicyId(policyId);
         entry.setBillingCategory(category);
+        // Count dimension + input fingerprint, denormalised from the job for usage analytics
+        // (PDFs processed, unique PDFs, size-multiplier average).
+        entry.setDocCount(docCount);
+        entry.setDocumentFingerprint(documentFingerprint);
         ledgerRepository.save(entry);
     }
 

app/saas/src/main/java/stirling/software/saas/payg/filter/PaygChargeInterceptor.java

@@ -33,6 +33,7 @@ import jakarta.servlet.http.HttpServletResponse;
 import lombok.extern.slf4j.Slf4j;
 
 import stirling.software.common.annotations.AutoJobPostMapping;
+import stirling.software.common.service.AutomationRunContext;
 import stirling.software.common.util.TempFile;
 import stirling.software.common.util.TempFileManager;
 import stirling.software.proprietary.security.database.repository.UserRepository;
@@ -284,13 +285,25 @@ public class PaygChargeInterceptor implements AsyncHandlerInterceptor {
         request.setAttribute(ATTR_INPUT_BYTES, totalInputBytes);
         request.setAttribute(ATTR_TOOL_ID, resolveToolId(request));
 
+        // Automation-run correlation id, honoured ONLY from an internal automation dispatch.
+        // InternalApiClient stamps X-Stirling-Automation on every loopback sub-step alongside the
+        // run id, so a genuine pipeline / policy / AI run always carries both. A raw external
+        // request that sets X-Stirling-Run-Id on its own is ignored (each such call stays its own
+        // charge): otherwise an API caller could pin a constant run id to collapse separate
+        // same-content calls into one charge, defeating "charge per API call". Null → standalone.
+        String headerRunId = request.getHeader(AutomationRunContext.RUN_ID_HEADER);
+        String runId =
+                (hasAutomationHeader(request) && headerRunId != null && !headerRunId.isBlank())
+                        ? headerRunId
+                        : null;
         ChargeContext ctx =
                 new ChargeContext(
                         currentUser.getId(),
                         currentUser.getTeam() == null ? null : currentUser.getTeam().getId(),
                         determineSource(request, auth),
                         ProcessType.SINGLE_TOOL,
-                        category);
+                        category,
+                        runId);
 
         ChargeOutcome outcome;
         try {
@@ -498,9 +511,20 @@ public class PaygChargeInterceptor implements AsyncHandlerInterceptor {
         }
     }
 
+    /**
+     * True when the request carries the internal-dispatch marker InternalApiClient stamps on every
+     * loopback sub-step ({@code X-Stirling-Automation: true}). This is the trust boundary for both
+     * the AUTOMATION billing category and for honouring {@code X-Stirling-Run-Id}: an external
+     * caller can't group charges via a run id without also declaring itself automation (which
+     * changes its own billing category).
+     */
+    private static boolean hasAutomationHeader(HttpServletRequest request) {
+        String header = request.getHeader(AUTOMATION_HEADER);
+        return header != null && "true".equalsIgnoreCase(header.trim());
+    }
+
     private static JobSource determineSource(HttpServletRequest request, Authentication auth) {
-        String automationHeader = request.getHeader(AUTOMATION_HEADER);
-        if (automationHeader != null && "true".equalsIgnoreCase(automationHeader.trim())) {
+        if (hasAutomationHeader(request)) {
             return JobSource.PIPELINE;
         }
         String desktopHeader = request.getHeader(DESKTOP_CLIENT_HEADER);
@@ -527,8 +551,7 @@ public class PaygChargeInterceptor implements AsyncHandlerInterceptor {
      */
     private static BillingCategory determineCategory(
             HandlerMethod handler, HttpServletRequest request, Authentication auth) {
-        String automationHeader = request.getHeader(AUTOMATION_HEADER);
-        if (automationHeader != null && "true".equalsIgnoreCase(automationHeader.trim())) {
+        if (hasAutomationHeader(request)) {
             return BillingCategory.AUTOMATION;
         }
         RequiresFeature ann =

app/saas/src/main/java/stirling/software/saas/payg/instance/InstanceUsageIngestService.java

@@ -116,7 +116,8 @@ public class InstanceUsageIngestService {
                             teamId,
                             JobSource.LINKED_INSTANCE,
                             ProcessType.SINGLE_TOOL,
-                            category),
+                            category,
+                            null),
                     units);
         }
         if (row == null) {

app/saas/src/main/java/stirling/software/saas/payg/job/JobContext.java

@@ -15,7 +15,8 @@ public record JobContext(
         JobSource source,
         ProcessType processType,
         Long policyId,
-        int stepLimit) {
+        int stepLimit,
+        String runId) {
 
     public JobContext {
         if (ownerUserId == null) {
@@ -34,4 +35,19 @@ public record JobContext(
             throw new IllegalArgumentException("stepLimit must be > 0");
         }
     }
+
+    /**
+     * Convenience for callers with no automation-run context — a standalone tool call ({@code
+     * runId} = {@code null}, so {@code joinOrOpen} always opens a fresh process rather than
+     * lineage-joining).
+     */
+    public JobContext(
+            Long ownerUserId,
+            Long ownerTeamId,
+            JobSource source,
+            ProcessType processType,
+            Long policyId,
+            int stepLimit) {
+        this(ownerUserId, ownerTeamId, source, processType, policyId, stepLimit, null);
+    }
 }

app/saas/src/main/java/stirling/software/saas/payg/job/JobService.java

@@ -1,12 +1,16 @@
 package stirling.software.saas.payg.job;
 
 import java.io.IOException;
+import java.nio.charset.StandardCharsets;
 import java.nio.file.Path;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
 import java.time.Duration;
 import java.time.LocalDateTime;
 import java.util.ArrayList;
 import java.util.Comparator;
 import java.util.HashMap;
+import java.util.HexFormat;
 import java.util.List;
 import java.util.Map;
 import java.util.Objects;
@@ -97,8 +101,14 @@ public class JobService {
             signaturesByInput.put(input, detector.extractSignatures(input));
         }
 
+        // Lineage joins are scoped to one automation run: a standalone call (no run id) never
+        // joins — each is its own charge — and within a run, matching still runs by content so a
+        // run's separate input files each open their own charge (a merge of N inputs = N charges),
+        // while a single file's chain of steps + its split outputs collapse into one.
         Optional<LineageMatch> bestMatch =
-                findBestMatch(ctx.ownerUserId(), inputs, signaturesByInput);
+                ctx.runId() == null
+                        ? Optional.empty()
+                        : findBestMatch(ctx.ownerUserId(), ctx.runId(), inputs, signaturesByInput);
 
         if (bestMatch.isPresent()) {
             ProcessingJob existing =
@@ -212,10 +222,13 @@ public class JobService {
     }
 
     private Optional<LineageMatch> findBestMatch(
-            Long userId, List<Path> inputs, Map<Path, Set<LineageSignature>> signaturesByInput) {
+            Long userId,
+            String runId,
+            List<Path> inputs,
+            Map<Path, Set<LineageSignature>> signaturesByInput) {
         List<LineageMatch> matches = new ArrayList<>(inputs.size());
         for (Path input : inputs) {
-            detector.detect(userId, signaturesByInput.get(input)).ifPresent(matches::add);
+            detector.detect(userId, runId, signaturesByInput.get(input)).ifPresent(matches::add);
         }
         return matches.stream().max(Comparator.comparing(LineageMatch::jobLastStepAt));
     }
@@ -252,6 +265,11 @@ public class JobService {
         fresh.setProcessType(ctx.processType());
         fresh.setSource(ctx.source());
         fresh.setPolicyId(ctx.policyId());
+        fresh.setRunId(ctx.runId());
+        // doc_count = number of input files (the count dimension). A merge (N inputs in one call)
+        // is N; a split (1 input) is 1; a standalone bookkeeping job (no inputs) is 1.
+        fresh.setDocCount(Math.max(1, signaturesByInput.size()));
+        fresh.setDocumentFingerprint(computeFingerprint(signaturesByInput));
         fresh.setStepCount(1);
         LocalDateTime now = LocalDateTime.now();
         fresh.setStartedAt(now);
@@ -262,6 +280,36 @@ public class JobService {
         return new JoinOrOpenResult(saved, JoinOrOpenResult.Disposition.OPENED);
     }
 
+    /**
+     * Stable fingerprint of this job's input set — SHA-256 over the sorted union of the inputs'
+     * lineage storage keys. {@code COUNT(DISTINCT ...)} over these gives "unique PDFs processed".
+     * {@code null} when there are no inputs (a standalone bookkeeping job, e.g. an AI Create
+     * session) — such jobs still count toward doc_count but aren't a distinct input PDF.
+     */
+    private static String computeFingerprint(Map<Path, Set<LineageSignature>> signaturesByInput) {
+        List<String> keys =
+                signaturesByInput.values().stream()
+                        .flatMap(Set::stream)
+                        .map(LineageSignature::asStorageKey)
+                        .distinct()
+                        .sorted()
+                        .toList();
+        if (keys.isEmpty()) {
+            return null;
+        }
+        try {
+            MessageDigest md = MessageDigest.getInstance("SHA-256");
+            for (String key : keys) {
+                md.update(key.getBytes(StandardCharsets.UTF_8));
+                md.update((byte) 0);
+            }
+            return HexFormat.of().formatHex(md.digest());
+        } catch (NoSuchAlgorithmException e) {
+            // SHA-256 is guaranteed on every JRE; fall back to no fingerprint if it ever isn't.
+            return null;
+        }
+    }
+
     private void recordAllInputs(UUID jobId, Map<Path, Set<LineageSignature>> signaturesByInput) {
         for (Set<LineageSignature> signatures : signaturesByInput.values()) {
             detector.record(jobId, signatures, ArtifactKind.INPUT);

app/saas/src/main/java/stirling/software/saas/payg/job/ProcessingJob.java

@@ -62,6 +62,27 @@ public class ProcessingJob implements Serializable {
     @Column(name = "doc_units", nullable = false)
     private Integer docUnits = 0;
 
+    /**
+     * Number of input files this charge represents — the count dimension, kept distinct from {@link
+     * #docUnits} (which scales with file size). A split (1 input → many outputs) stays 1; a merge
+     * (N inputs → 1 output) is N. Fixed at open; joined steps never change it.
+     *
+     * <p>The {@code columnDefinition} default keeps the ddl-auto ADD COLUMN safe on an
+     * already-populated {@code processing_job} (a bare {@code NOT NULL} add is rejected by Postgres
+     * on a non-empty table).
+     */
+    @Column(name = "doc_count", nullable = false, columnDefinition = "integer not null default 1")
+    private Integer docCount = 1;
+
+    /**
+     * Correlation id of the automation run that opened this job ({@code X-Stirling-Run-Id}), or
+     * {@code null} for a standalone tool call. Lineage joins are scoped to a single run id, so two
+     * separate runs never merge even on identical bytes; a null run id never joins (each standalone
+     * call is its own charge).
+     */
+    @Column(name = "run_id", length = 64)
+    private String runId;
+
     @Column(name = "step_count", nullable = false)
     private Integer stepCount = 0;
 

app/saas/src/main/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetector.java

@@ -73,6 +73,21 @@ public class DefaultHashLineageDetector implements HashLineageDetector {
         return store.findOpenJobForSignatures(userId, signatures, workflowWindow);
     }
 
+    @Override
+    public Optional<LineageMatch> detect(
+            Long userId, String runId, Set<LineageSignature> signatures) {
+        Objects.requireNonNull(userId, "userId");
+        Objects.requireNonNull(signatures, "signatures");
+        if (runId == null) {
+            // No run context → standalone call; never lineage-joins (each is its own charge).
+            return Optional.empty();
+        }
+        if (signatures.isEmpty()) {
+            return Optional.empty();
+        }
+        return store.findOpenJobForSignatures(userId, signatures, workflowWindow, runId);
+    }
+
     @Override
     public void record(UUID jobId, Path file, ArtifactKind kind) throws IOException {
         Objects.requireNonNull(file, "file");

app/saas/src/main/java/stirling/software/saas/payg/lineage/HashLineageDetector.java

@@ -50,6 +50,17 @@ public interface HashLineageDetector {
      */
     Optional<LineageMatch> detect(Long userId, Set<LineageSignature> signatures);
 
+    /**
+     * Run-scoped detect: as {@link #detect(Long, Set)} but only matches open jobs belonging to the
+     * automation run {@code runId}, so lineage joins never cross runs (two separate runs on
+     * identical bytes stay distinct charges). The default ignores {@code runId} (for simple test
+     * doubles); the production detector overrides it to scope the store lookup.
+     */
+    default Optional<LineageMatch> detect(
+            Long userId, String runId, Set<LineageSignature> signatures) {
+        return detect(userId, signatures);
+    }
+
     /**
      * Same as {@link #record(UUID, Path, ArtifactKind)} but operating on pre-computed signatures.
      * Empty {@code signatures} is a no-op.

app/saas/src/main/java/stirling/software/saas/payg/lineage/JobLineageStore.java

@@ -34,6 +34,19 @@ public interface JobLineageStore {
     Optional<LineageMatch> findOpenJobForSignatures(
             Long userId, Set<LineageSignature> candidates, Duration workflowWindow);
 
+    /**
+     * Run-scoped variant: as {@link #findOpenJobForSignatures(Long, Set, Duration)} but
+     * additionally constrained to open jobs whose {@code run_id} equals {@code runId}. Lineage
+     * joins are scoped to a single automation run, so a pipeline's sub-steps group into one charge
+     * while two separate runs on identical bytes stay distinct. The default delegates to the
+     * unscoped lookup (for test doubles that don't model run ids); the production store overrides
+     * it with a filtered query.
+     */
+    default Optional<LineageMatch> findOpenJobForSignatures(
+            Long userId, Set<LineageSignature> candidates, Duration workflowWindow, String runId) {
+        return findOpenJobForSignatures(userId, candidates, workflowWindow);
+    }
+
     /** Deletes records created before {@code cutoff}. Returns the number of rows removed. */
     int pruneOlderThan(Instant cutoff);
 }

app/saas/src/main/java/stirling/software/saas/payg/lineage/JpaJobLineageStore.java

@@ -78,6 +78,24 @@ public class JpaJobLineageStore implements JobLineageStore {
         return matches.isEmpty() ? Optional.empty() : Optional.of(matches.get(0));
     }
 
+    @Override
+    public Optional<LineageMatch> findOpenJobForSignatures(
+            Long userId, Set<LineageSignature> candidates, Duration workflowWindow, String runId) {
+        Objects.requireNonNull(userId, "userId");
+        Objects.requireNonNull(candidates, "candidates");
+        Objects.requireNonNull(workflowWindow, "workflowWindow");
+        Objects.requireNonNull(runId, "runId");
+        if (candidates.isEmpty()) {
+            return Optional.empty();
+        }
+        List<String> storageKeys = candidates.stream().map(LineageSignature::asStorageKey).toList();
+        LocalDateTime since = LocalDateTime.now().minus(workflowWindow);
+        List<LineageMatch> matches =
+                hashRepository.findOpenJobsForSignaturesInRun(
+                        userId, JobStatus.OPEN, since, storageKeys, runId, Limit.of(1));
+        return matches.isEmpty() ? Optional.empty() : Optional.of(matches.get(0));
+    }
+
     @Override
     @Transactional
     public int pruneOlderThan(Instant cutoff) {

app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java

@@ -43,6 +43,29 @@ public interface JobArtifactHashRepository
             @Param("signatures") Collection<String> signatures,
             Limit limit);
 
+    /**
+     * Run-scoped lineage lookup: as {@link #findOpenJobsForSignatures} but additionally requires
+     * {@code j.run_id = :runId}, so only jobs belonging to the same automation run can be joined.
+     */
+    @Query(
+            "SELECT new stirling.software.saas.payg.lineage.LineageMatch("
+                    + " h.id.jobId, h.id.kind, j.lastStepAt)"
+                    + " FROM JobArtifactHash h"
+                    + " JOIN ProcessingJob j ON j.id = h.id.jobId"
+                    + " WHERE j.ownerUserId = :userId"
+                    + " AND j.status = :openStatus"
+                    + " AND j.lastStepAt > :since"
+                    + " AND j.runId = :runId"
+                    + " AND h.id.contentHash IN :signatures"
+                    + " ORDER BY j.lastStepAt DESC")
+    List<LineageMatch> findOpenJobsForSignaturesInRun(
+            @Param("userId") Long userId,
+            @Param("openStatus") JobStatus openStatus,
+            @Param("since") LocalDateTime since,
+            @Param("signatures") Collection<String> signatures,
+            @Param("runId") String runId,
+            Limit limit);
+
     /** Prunes rows older than {@code cutoff}; run from a scheduled task. */
     @Modifying
     @Query("DELETE FROM JobArtifactHash h WHERE h.createdAt < :cutoff")

app/saas/src/main/java/stirling/software/saas/payg/repository/WalletLedgerRepository.java

@@ -18,14 +18,13 @@ public interface WalletLedgerRepository extends JpaRepository<WalletLedgerEntry,
     List<WalletLedgerEntry> findTop20ByTeamIdOrderByIdDesc(Long teamId);
 
     /**
-     * Per-category debit totals over an arbitrary window, as positive units. Replaces the
-     * calendar-month {@code wallet_category_summary} view on the wallet endpoint — subscribed
-     * teams' billing windows are anchored to the Stripe subscription period, not month starts. Rows
-     * with {@code NULL} category (system entries) are excluded; BYPASSED never reaches the ledger
-     * by construction.
+     * Per-category debit totals with BOTH the size-scaled unit sum and the input-file count ({@code
+     * doc_count}) over a window. Rows: {@code [category, units, docs]}. Lets the wallet show, per
+     * category, "X PDFs · Y meter units" rather than conflating the two.
      */
     @Query(
-            "SELECT e.billingCategory AS category, COALESCE(SUM(-e.amountUnits), 0) AS units"
+            "SELECT e.billingCategory AS category, COALESCE(SUM(-e.amountUnits), 0) AS units,"
+                    + " COALESCE(SUM(e.docCount), 0) AS docs"
                     + " FROM WalletLedgerEntry e"
                     + " WHERE e.teamId = :teamId"
                     + " AND e.entryType = :entryType"
@@ -33,7 +32,36 @@ public interface WalletLedgerRepository extends JpaRepository<WalletLedgerEntry,
                     + " AND e.occurredAt >= :periodStart"
                     + " AND e.occurredAt < :periodEnd"
                     + " GROUP BY e.billingCategory")
-    List<Object[]> sumPeriodAmountByCategory(
+    List<Object[]> sumPeriodByCategoryWithDocs(
+            @Param("teamId") Long teamId,
+            @Param("entryType") LedgerEntryType entryType,
+            @Param("periodStart") LocalDateTime periodStart,
+            @Param("periodEnd") LocalDateTime periodEnd);
+
+    /**
+     * Period usage analytics in one row: {@code [docsProcessed, uniquePdfs, sizeMultiplierPdfs]}.
+     * {@code docsProcessed} sums input-file counts; {@code uniquePdfs} counts distinct input
+     * fingerprints (a file hit by N operations counts once); {@code sizeMultiplierPdfs} sums the
+     * input files on charges where the size multiplier kicked in (units billed &gt; input files).
+     * DEBIT + non-null category only.
+     *
+     * <p>Returns a single-element {@code List} (aggregate-only query → always one row). Declared as
+     * {@code List<Object[]>} rather than {@code Object[]}: Spring Data treats an {@code Object[]}
+     * return as a <em>collection</em> and hands back {@code Object[]{ row }}, so the caller would
+     * read the columns one level too deep — take {@code get(0)}.
+     */
+    @Query(
+            "SELECT COALESCE(SUM(e.docCount), 0) AS docs,"
+                    + " COUNT(DISTINCT e.documentFingerprint) AS uniquePdfs,"
+                    + " COALESCE(SUM(CASE WHEN (-e.amountUnits) > e.docCount THEN e.docCount ELSE 0"
+                    + " END), 0) AS sizeMultiplierPdfs"
+                    + " FROM WalletLedgerEntry e"
+                    + " WHERE e.teamId = :teamId"
+                    + " AND e.entryType = :entryType"
+                    + " AND e.billingCategory IS NOT NULL"
+                    + " AND e.occurredAt >= :periodStart"
+                    + " AND e.occurredAt < :periodEnd")
+    List<Object[]> periodUsageAnalytics(
             @Param("teamId") Long teamId,
             @Param("entryType") LedgerEntryType entryType,
             @Param("periodStart") LocalDateTime periodStart,

app/saas/src/main/java/stirling/software/saas/payg/wallet/WalletLedgerEntry.java

@@ -74,6 +74,26 @@ public class WalletLedgerEntry implements Serializable {
     @Column(name = "policy_id")
     private Long policyId;
 
+    /**
+     * Number of input files this entry billed (the count dimension, distinct from size-scaled
+     * {@link #amountUnits}). Denormalised from {@code processing_job.doc_count} so usage analytics
+     * — "PDFs processed" — sum one table. Defaults to 1; system/aggregate entries may leave it 1.
+     *
+     * <p>The {@code columnDefinition} default keeps the ddl-auto ADD COLUMN safe on an
+     * already-populated {@code wallet_ledger} (a bare {@code NOT NULL} add is rejected by Postgres
+     * on a non-empty table).
+     */
+    @Column(name = "doc_count", nullable = false, columnDefinition = "integer not null default 1")
+    private Integer docCount = 1;
+
+    /**
+     * SHA-256 of this entry's input file set; {@code COUNT(DISTINCT ...)} over a period gives
+     * unique PDFs processed. {@code null} for aggregate/system entries (grants, linked-instance
+     * sync) that don't map to a single document set.
+     */
+    @Column(name = "document_fingerprint", length = 64)
+    private String documentFingerprint;
+
     @Column(name = "stripe_event_id", length = 128)
     private String stripeEventId;
 

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

@@ -0,0 +1,39 @@
+-- PAYG size-scaled billing: run-scoped grouping + per-input-file counting.
+--
+-- Two independent axes now live on a charge:
+--   * doc_units  — billing quantity, scales with file size (existing column)
+--   * doc_count  — number of INPUT files (the "unique PDFs" dimension); a split (1→many) stays 1,
+--                  a merge (N→1) is N. Fixed at open; joined steps never change it.
+-- Plus run_id, the automation-run correlation id used to group a run's tool sub-steps into one
+-- charge (replacing the old content+time-window grouping) and to keep two separate runs distinct.
+--
+-- Everything is additive; no existing rows are modified, no columns dropped.
+
+-- ── processing_job ───────────────────────────────────────────────────────────
+ALTER TABLE processing_job ADD COLUMN IF NOT EXISTS run_id VARCHAR(64);
+ALTER TABLE processing_job ADD COLUMN IF NOT EXISTS doc_count INTEGER NOT NULL DEFAULT 1;
+
+COMMENT ON COLUMN processing_job.run_id IS
+    'Automation-run correlation id (X-Stirling-Run-Id); NULL for a standalone tool call. Lineage '
+    'joins are scoped to one run_id, so separate runs never merge even on identical bytes.';
+COMMENT ON COLUMN processing_job.doc_count IS
+    'Number of input files this charge represents (the count dimension, distinct from size-scaled '
+    'doc_units). Split=1, merge=N. Fixed at open.';
+
+-- ── wallet_ledger ────────────────────────────────────────────────────────────
+-- Denormalise the count dimension + input fingerprint onto the DEBIT row so usage analytics
+-- (unique PDFs, per-category counts, size-multiplier average) query one table and survive
+-- processing_job pruning.
+ALTER TABLE wallet_ledger ADD COLUMN IF NOT EXISTS doc_count INTEGER NOT NULL DEFAULT 1;
+ALTER TABLE wallet_ledger ADD COLUMN IF NOT EXISTS document_fingerprint VARCHAR(64);
+
+COMMENT ON COLUMN wallet_ledger.doc_count IS
+    'Input-file count for this entry (mirrors processing_job.doc_count); summed for "PDFs processed".';
+COMMENT ON COLUMN wallet_ledger.document_fingerprint IS
+    'SHA-256 of the entry''s input file set; COUNT(DISTINCT ...) gives unique PDFs. NULL for '
+    'aggregate/system entries (e.g. linked-instance sync).';
+
+-- Distinct-PDF + size-multiplier queries scan by team + period.
+CREATE INDEX IF NOT EXISTS idx_wallet_ledger_team_period_fp
+    ON wallet_ledger (team_id, occurred_at, document_fingerprint)
+    WHERE document_fingerprint IS NOT NULL;

app/saas/src/test/java/stirling/software/saas/payg/api/PaygWalletControllerTest.java

@@ -127,9 +127,11 @@ class PaygWalletControllerTest {
     }
 
     private void stubEmptyLedgerReads(long teamId) {
-        when(ledgerRepo.sumPeriodAmountByCategory(
+        when(ledgerRepo.sumPeriodByCategoryWithDocs(
                         eq(teamId), eq(LedgerEntryType.DEBIT), any(), any()))
                 .thenReturn(List.of());
+        when(ledgerRepo.periodUsageAnalytics(eq(teamId), eq(LedgerEntryType.DEBIT), any(), any()))
+                .thenReturn(List.<Object[]>of(new Object[] {0L, 0L, 0L}));
         when(ledgerRepo.findTop20ByTeamIdOrderByIdDesc(teamId)).thenReturn(List.of());
     }
 
@@ -187,12 +189,17 @@ class PaygWalletControllerTest {
         when(shadowRepo.sumPaidUnits(eq(99L), any(), any())).thenReturn(312L);
         when(billingService.estimateBillMinor(any(), eq(312L))).thenReturn(Optional.of(624L));
         when(entitlementService.getSnapshot(99L)).thenReturn(snapshot(312L, 1250L));
-        when(ledgerRepo.sumPeriodAmountByCategory(eq(99L), eq(LedgerEntryType.DEBIT), any(), any()))
+        // [category, units, docs]: units scale with size, docs = input-file count.
+        when(ledgerRepo.sumPeriodByCategoryWithDocs(
+                        eq(99L), eq(LedgerEntryType.DEBIT), any(), any()))
                 .thenReturn(
                         List.of(
-                                new Object[] {BillingCategory.API, 110L},
-                                new Object[] {BillingCategory.AI, 200L},
-                                new Object[] {BillingCategory.AUTOMATION, 2L}));
+                                new Object[] {BillingCategory.API, 110L, 90L},
+                                new Object[] {BillingCategory.AI, 200L, 50L},
+                                new Object[] {BillingCategory.AUTOMATION, 2L, 2L}));
+        // [docsProcessed, uniquePdfs, sizeMultiplierPdfs] — single-row aggregate as List<Object[]>
+        when(ledgerRepo.periodUsageAnalytics(eq(99L), eq(LedgerEntryType.DEBIT), any(), any()))
+                .thenReturn(List.<Object[]>of(new Object[] {142L, 120L, 30L}));
         when(ledgerRepo.findTop20ByTeamIdOrderByIdDesc(99L)).thenReturn(List.of());
 
         ResponseEntity<WalletSnapshotResponse> resp =
@@ -214,6 +221,13 @@ class PaygWalletControllerTest {
         assertThat(body.categoryBreakdown().api()).isEqualTo(110);
         assertThat(body.categoryBreakdown().ai()).isEqualTo(200);
         assertThat(body.categoryBreakdown().automation()).isEqualTo(2);
+        // Count dimension is surfaced separately from the size-scaled units.
+        assertThat(body.categoryDocs().api()).isEqualTo(90);
+        assertThat(body.categoryDocs().ai()).isEqualTo(50);
+        assertThat(body.categoryDocs().automation()).isEqualTo(2);
+        assertThat(body.docsProcessedThisPeriod()).isEqualTo(142);
+        assertThat(body.uniquePdfsThisPeriod()).isEqualTo(120);
+        assertThat(body.sizeMultiplierPdfsThisPeriod()).isEqualTo(30);
         assertThat(body.stripeSubscriptionId()).isEqualTo("sub_test_99");
         // Member role → ledger never queried per-user.
         verify(ledgerRepo, never()).sumPeriodAmountForMember(any(), any(), any(), any(), any());

app/saas/src/test/java/stirling/software/saas/payg/api/WalletSnapshotResponseTest.java

@@ -35,7 +35,11 @@ class WalletSnapshotResponseTest {
                 /* spendUnitsThisPeriod= */ 12,
                 new CategoryBreakdown(5, 4, 3),
                 List.of(new MemberRow("u1", "Ann", "ann@example.com", 8)),
-                List.of(new ActivityRow(1L, "api", "API usage", "2026-06-02T10:00", 4)));
+                List.of(new ActivityRow(1L, "api", "API usage", "2026-06-02T10:00", 4)),
+                /* categoryDocs= */ new CategoryBreakdown(3, 2, 1),
+                /* docsProcessedThisPeriod= */ 6,
+                /* uniquePdfsThisPeriod= */ 5,
+                /* sizeMultiplierPdfsThisPeriod= */ 2);
     }
 
     @Test
@@ -107,7 +111,11 @@ class WalletSnapshotResponseTest {
                         0,
                         new CategoryBreakdown(0, 0, 0),
                         List.of(),
-                        List.of());
+                        List.of(),
+                        new CategoryBreakdown(0, 0, 0),
+                        0,
+                        0,
+                        0);
 
         assertThat(free.billableLimit()).isNull();
         assertThat(free.pricePerDocMinor()).isNull();

app/saas/src/test/java/stirling/software/saas/payg/filter/PaygChargeInterceptorTest.java

@@ -434,6 +434,52 @@ class PaygChargeInterceptorTest {
                 .isEqualTo(stirling.software.saas.payg.model.JobSource.PIPELINE);
     }
 
+    @Test
+    void preHandle_runId_honouredOnlyWithAutomationHeader() throws Exception {
+        // An internal dispatch carries BOTH X-Stirling-Automation and X-Stirling-Run-Id, so the
+        // run id flows onto the ChargeContext (sub-steps of one run group into a single charge).
+        authenticateWithApiKey(makeUser(7L, 42L));
+        UUID jobId = UUID.randomUUID();
+        when(chargeService.openProcess(any(), anyList()))
+                .thenReturn(new ChargeOutcome(jobId, 1, ChargeOutcome.Disposition.OPENED));
+        org.mockito.ArgumentCaptor<stirling.software.saas.payg.charge.ChargeContext> ctxCaptor =
+                org.mockito.ArgumentCaptor.forClass(
+                        stirling.software.saas.payg.charge.ChargeContext.class);
+
+        MockMultipartHttpServletRequest req = newMultipart();
+        req.addFile(new MockMultipartFile("file", "x.pdf", "application/pdf", "abc".getBytes()));
+        req.addHeader("X-Stirling-Automation", "true");
+        req.addHeader("X-Stirling-Run-Id", "run-abc");
+
+        interceptor.preHandle(req, new MockHttpServletResponse(), handlerMethodForFakeController());
+
+        verify(chargeService).openProcess(ctxCaptor.capture(), anyList());
+        assertThat(ctxCaptor.getValue().runId()).isEqualTo("run-abc");
+    }
+
+    @Test
+    void preHandle_runIdWithoutAutomationHeader_isIgnored() throws Exception {
+        // A raw external API call that sets X-Stirling-Run-Id on its own must NOT be able to group
+        // charges — the run id is dropped so each call stays its own charge ("charge per API
+        // call").
+        authenticateWithApiKey(makeUser(7L, 42L));
+        UUID jobId = UUID.randomUUID();
+        when(chargeService.openProcess(any(), anyList()))
+                .thenReturn(new ChargeOutcome(jobId, 1, ChargeOutcome.Disposition.OPENED));
+        org.mockito.ArgumentCaptor<stirling.software.saas.payg.charge.ChargeContext> ctxCaptor =
+                org.mockito.ArgumentCaptor.forClass(
+                        stirling.software.saas.payg.charge.ChargeContext.class);
+
+        MockMultipartHttpServletRequest req = newMultipart();
+        req.addFile(new MockMultipartFile("file", "x.pdf", "application/pdf", "abc".getBytes()));
+        req.addHeader("X-Stirling-Run-Id", "run-spoofed");
+
+        interceptor.preHandle(req, new MockHttpServletResponse(), handlerMethodForFakeController());
+
+        verify(chargeService).openProcess(ctxCaptor.capture(), anyList());
+        assertThat(ctxCaptor.getValue().runId()).isNull();
+    }
+
     // --- BillingCategory categorisation + bypass fast-path -------------------------------------
 
     @Test

app/saas/src/test/java/stirling/software/saas/payg/job/JobServiceTest.java

@@ -183,6 +183,40 @@ class JobServiceTest {
         assertThat(detector.recorded(result.job().getId(), input, ArtifactKind.INPUT)).isTrue();
     }
 
+    @Test
+    void joinOrOpen_nullRunId_neverJoins(@TempDir Path tmp) throws IOException {
+        // A standalone call (no run id) opens its own charge even when content matches an open
+        // job — run-scoped grouping only joins within the same automation run.
+        UUID existingId = UUID.randomUUID();
+        Path input = givenFile(tmp, "in.bin");
+        detector.willMatch(
+                input, new LineageMatch(existingId, ArtifactKind.INPUT, LocalDateTime.now()));
+
+        JobContext standalone =
+                new JobContext(
+                        42L, 100L, JobSource.API, ProcessType.SINGLE_TOOL, 1L, 10); // null runId
+        JoinOrOpenResult result = service.joinOrOpen(standalone, List.of(input));
+
+        assertThat(result.disposition()).isEqualTo(JoinOrOpenResult.Disposition.OPENED);
+        verify(jobRepo, never()).findById(existingId);
+    }
+
+    @Test
+    void joinOrOpen_opened_setsRunIdAndDocCountAndFingerprint(@TempDir Path tmp)
+            throws IOException {
+        // Two input files, no match → one fresh charge with doc_count = input-file count (2),
+        // the run id stamped, and a non-null fingerprint (used for unique-PDF counting).
+        Path a = givenFile(tmp, "a.bin");
+        Path b = givenFile(tmp, "b.bin");
+
+        JoinOrOpenResult result = service.joinOrOpen(ctx(42L, 100L, 10), List.of(a, b));
+
+        assertThat(result.disposition()).isEqualTo(JoinOrOpenResult.Disposition.OPENED);
+        assertThat(result.job().getDocCount()).isEqualTo(2);
+        assertThat(result.job().getRunId()).isEqualTo("test-run");
+        assertThat(result.job().getDocumentFingerprint()).isNotNull();
+    }
+
     @Test
     void joinOrOpen_emptyInputs_throws() {
         assertThatThrownBy(() -> service.joinOrOpen(ctx(42L, 100L, 10), List.of()))
@@ -297,8 +331,10 @@ class JobServiceTest {
     // --- helpers --------------------------------------------------------------------------------
 
     private static JobContext ctx(long userId, long teamId, int stepLimit) {
+        // Lineage joins are scoped to an automation run, so the join scenarios below run inside a
+        // run id. The standalone (null run id) path is covered by joinOrOpen_nullRunId_neverJoins.
         return new JobContext(
-                userId, teamId, JobSource.WEB, ProcessType.SINGLE_TOOL, 1L, stepLimit);
+                userId, teamId, JobSource.WEB, ProcessType.SINGLE_TOOL, 1L, stepLimit, "test-run");
     }
 
     private static ProcessingJob openJob(

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

@@ -6315,18 +6315,21 @@ managedTitle = "Managed in Stripe"
 update = "Update"
 
 [portal.billing.pdfsProcessed]
-emptyPeriod = "No metered processing yet this period."
+emptyPeriod = "No processing yet this period."
 eyebrow = "PDFs processed this period"
 legendValue_one = "{{formatted}} PDFs"
 legendValue_other = "{{formatted}} PDFs"
-segbarAriaLabel = "Metered PDFs split by category"
+segbarAriaLabel = "PDFs split by category"
 segmentAgentsDesc = "AI agent actions"
 segmentAgentsLabel = "Agents"
 segmentApiDesc = "Direct API requests"
 segmentApiLabel = "API"
 segmentAutomationDesc = "Automations & pipelines"
 segmentAutomationLabel = "Automation"
-unit = "metered PDFs"
+sizeMultiplier = "{{formatted}} PDFs used a size multiplier"
+summary = "{{unique}} unique · {{units}} meter units · {{avg}} avg per PDF"
+summaryNoRate = "{{unique}} unique · {{units}} meter units"
+unit = "PDFs"
 
 [portal.billing.spendLimit]
 adjustLimit = "Adjust limit"
@@ -6355,6 +6358,8 @@ label = "Projected to exceed."
 
 [portal.billing.spendThisMonth]
 eyebrow = "Spend this month"
+freeRemaining_one = "{{formatted}} free PDF remaining"
+freeRemaining_other = "{{formatted}} free PDFs remaining"
 processed_one = "{{formattedCount}} PDF processed."
 processed_other = "{{formattedCount}} PDFs processed."
 processedWithRate_one = "{{formattedCount}} PDF processed, at {{rate}} each."

frontend/editor/src/cloud/hooks/useWallet.ts

@@ -133,7 +133,10 @@ function reuseIfEqual(prev: Wallet | null, next: Wallet): Wallet {
     prev.capUsd !== next.capUsd ||
     prev.noCap !== next.noCap ||
     prev.stripeSubscriptionId !== next.stripeSubscriptionId ||
-    prev.spendUnitsThisPeriod !== next.spendUnitsThisPeriod
+    prev.spendUnitsThisPeriod !== next.spendUnitsThisPeriod ||
+    prev.docsProcessedThisPeriod !== next.docsProcessedThisPeriod ||
+    prev.uniquePdfsThisPeriod !== next.uniquePdfsThisPeriod ||
+    prev.sizeMultiplierPdfsThisPeriod !== next.sizeMultiplierPdfsThisPeriod
   ) {
     return next;
   }
@@ -143,7 +146,10 @@ function reuseIfEqual(prev: Wallet | null, next: Wallet): Wallet {
   if (
     prev.categoryBreakdown.api !== next.categoryBreakdown.api ||
     prev.categoryBreakdown.ai !== next.categoryBreakdown.ai ||
-    prev.categoryBreakdown.automation !== next.categoryBreakdown.automation
+    prev.categoryBreakdown.automation !== next.categoryBreakdown.automation ||
+    prev.categoryDocs.api !== next.categoryDocs.api ||
+    prev.categoryDocs.ai !== next.categoryDocs.ai ||
+    prev.categoryDocs.automation !== next.categoryDocs.automation
   ) {
     return next;
   }

frontend/editor/src/portal/components/billing/PdfsProcessedCard.stories.tsx

@@ -28,7 +28,7 @@ export const WithUnsynced: Story = {
   },
 };
 
-/** Nothing metered yet this period — the split hides. */
+/** Nothing processed yet this period — the split + summary hide. */
 export const Empty: Story = {
   args: {
     wallet: {
@@ -36,6 +36,10 @@ export const Empty: Story = {
       billableUsed: 0,
       spendUnitsThisPeriod: 0,
       categoryBreakdown: { api: 0, ai: 0, automation: 0 },
+      categoryDocs: { api: 0, ai: 0, automation: 0 },
+      docsProcessedThisPeriod: 0,
+      uniquePdfsThisPeriod: 0,
+      sizeMultiplierPdfsThisPeriod: 0,
     },
   },
 };

frontend/editor/src/portal/components/billing/PdfsProcessedCard.tsx

@@ -1,19 +1,21 @@
 import { useTranslation } from "react-i18next";
 import { Card } from "@app/ui";
+import { formatMinor } from "@app/billing";
 import type { Wallet, WalletCategoryBreakdown } from "@portal/api/billing";
 import type { LocalUsage } from "@portal/api/link";
 
 /**
- * "PDFs processed this period" headline + a stacked split of where the metered
- * PDFs went. The split reuses the wallet's existing {@code categoryBreakdown}
- * (API / Agents / Automation — the same buckets the entitlement service tracks;
- * the "AI" bucket surfaces as "Agents" here). Real data only: the bar hides when
- * nothing metered has run yet.
+ * "PDFs processed this period" headline (the input-file count) plus a summary line
+ * separating that count from the size-scaled meter units, and a stacked split of
+ * where the PDFs went by category (API / Agents / Automation — the "AI" bucket
+ * surfaces as "Agents" here), driven by the wallet's per-category {@code categoryDocs}
+ * counts. Real data only: everything hides when nothing has run this period.
  *
- * <p>When a linked instance has accrued usage SaaS hasn't billed yet ({@code
- * unsynced}), it's folded into the headline + split so "current usage" reflects
- * work done since the last daily sync. The synced-vs-pending split is an internal
- * detail the customer doesn't need, so it's not surfaced — just the combined total.
+ * <p>Instance-local usage a linked instance has accrued but SaaS hasn't billed yet
+ * ({@code unsynced}) is units-only, so it folds into the meter-units figure (and the
+ * avg-per-PDF that derives from it) but NOT the PDF count or the per-category split,
+ * which reflect synced processing. The synced-vs-pending distinction is an internal
+ * detail, so only the combined unit total is surfaced.
  */
 const SEGMENTS: ReadonlyArray<{
   key: keyof WalletCategoryBreakdown;
@@ -57,17 +59,31 @@ export function PdfsProcessedCard({
   unsynced?: LocalUsage | null;
 }) {
   const { t } = useTranslation();
-  // Fold instance-local unsynced usage into both the headline and the split, so
-  // the card shows synced + not-yet-billed work as a single current-usage figure.
-  const pending = unsynced?.totalUnsyncedUnits ?? 0;
-  const base = wallet.categoryBreakdown;
-  const b: WalletCategoryBreakdown = {
-    api: base.api + (unsynced?.apiUnsyncedUnits ?? 0),
-    ai: base.ai + (unsynced?.aiUnsyncedUnits ?? 0),
-    automation: base.automation + (unsynced?.automationUnsyncedUnits ?? 0),
-  };
-  const total = b.api + b.ai + b.automation;
-  const headline = wallet.billableUsed + pending;
+  // The count dimension (PDFs) is shown separately from the size-scaled meter
+  // units. Instance-local unsynced usage (combined-billing) is units-only, so it
+  // folds into the meter-units figure; the PDF count reflects synced processing.
+  const pendingUnits = unsynced?.totalUnsyncedUnits ?? 0;
+  const docs = wallet.docsProcessedThisPeriod;
+  const uniquePdfs = wallet.uniquePdfsThisPeriod;
+  const meterUnits = wallet.spendUnitsThisPeriod + pendingUnits;
+  const sizeMultiplierPdfs = wallet.sizeMultiplierPdfsThisPeriod;
+
+  // Per-category PDF counts drive the split ("this many PDFs ran automation / AI /
+  // API"); units are surfaced in the aggregate summary line, not per bucket.
+  const perDocs: WalletCategoryBreakdown = wallet.categoryDocs;
+  const totalDocs = perDocs.api + perDocs.ai + perDocs.automation;
+
+  // Average cost per PDF in minor currency units — meter units × the per-unit rate,
+  // spread over the input files processed. Shown only when the rate is known (free-tier
+  // and unknown-price snapshots omit the term rather than imply $0.00).
+  const rate = wallet.pricePerDocMinor;
+  const showAvgCost = docs > 0 && rate != null;
+  const avgCostMinor =
+    rate != null && docs > 0 ? (meterUnits / docs) * rate : 0;
+
+  // Something ran once there are either counted PDFs or metered units (instance-local
+  // unsynced usage is units-only, so it keeps the card out of the empty state).
+  const hasActivity = docs > 0 || meterUnits > 0;
 
   return (
     <Card padding="loose">
@@ -78,66 +94,99 @@ export function PdfsProcessedCard({
         )}
       </span>
       <div className="portal-billing__bignum-row">
-        <span className="portal-billing__bignum">
-          {headline.toLocaleString()}
-        </span>
+        <span className="portal-billing__bignum">{docs.toLocaleString()}</span>
         <span className="portal-billing__bignum-unit">
-          {t("portal.billing.pdfsProcessed.unit", "metered PDFs")}
+          {t("portal.billing.pdfsProcessed.unit", "PDFs")}
         </span>
       </div>
 
-      {total > 0 ? (
+      {hasActivity ? (
         <>
-          <div
-            className="portal-billing__segbar"
-            role="img"
-            aria-label={t(
-              "portal.billing.pdfsProcessed.segbarAriaLabel",
-              "Metered PDFs split by category",
-            )}
-          >
-            {SEGMENTS.map((s) =>
-              b[s.key] > 0 ? (
-                <span
-                  key={s.key}
-                  className={`portal-billing__segbar-seg portal-billing__segbar-seg--${s.cls}`}
-                  style={{ width: `${(b[s.key] / total) * 100}%` }}
-                />
-              ) : null,
-            )}
-          </div>
-          <div className="portal-billing__seglegend">
-            {SEGMENTS.map((s) => (
-              <div className="portal-billing__seglegend-row" key={s.key}>
-                <span
-                  className={`portal-billing__dot portal-billing__dot--${s.cls}`}
-                  aria-hidden
-                />
-                <span className="portal-billing__seglegend-label">
-                  {t(s.labelKey, s.labelDefault)}
-                </span>
-                <span className="portal-billing__seglegend-val">
-                  {t(
-                    "portal.billing.pdfsProcessed.legendValue",
-                    "{{formatted}} PDFs",
-                    {
-                      count: b[s.key],
-                      formatted: b[s.key].toLocaleString(),
-                    },
-                  )}
-                </span>
-                <span className="portal-billing__seglegend-desc">
-                  {t(s.descKey, s.descDefault)}
-                </span>
+          <p className="portal-billing__section-sub">
+            {showAvgCost
+              ? t(
+                  "portal.billing.pdfsProcessed.summary",
+                  "{{unique}} unique · {{units}} meter units · {{avg}} avg per PDF",
+                  {
+                    unique: uniquePdfs.toLocaleString(),
+                    units: meterUnits.toLocaleString(),
+                    avg: formatMinor(avgCostMinor, wallet.currency),
+                  },
+                )
+              : t(
+                  "portal.billing.pdfsProcessed.summaryNoRate",
+                  "{{unique}} unique · {{units}} meter units",
+                  {
+                    unique: uniquePdfs.toLocaleString(),
+                    units: meterUnits.toLocaleString(),
+                  },
+                )}
+          </p>
+          {totalDocs > 0 ? (
+            <>
+              <div
+                className="portal-billing__segbar"
+                role="img"
+                aria-label={t(
+                  "portal.billing.pdfsProcessed.segbarAriaLabel",
+                  "PDFs split by category",
+                )}
+              >
+                {SEGMENTS.map((s) =>
+                  perDocs[s.key] > 0 ? (
+                    <span
+                      key={s.key}
+                      className={`portal-billing__segbar-seg portal-billing__segbar-seg--${s.cls}`}
+                      style={{
+                        width: `${(perDocs[s.key] / totalDocs) * 100}%`,
+                      }}
+                    />
+                  ) : null,
+                )}
               </div>
-            ))}
-          </div>
+              <div className="portal-billing__seglegend">
+                {SEGMENTS.map((s) => (
+                  <div className="portal-billing__seglegend-row" key={s.key}>
+                    <span
+                      className={`portal-billing__dot portal-billing__dot--${s.cls}`}
+                      aria-hidden
+                    />
+                    <span className="portal-billing__seglegend-label">
+                      {t(s.labelKey, s.labelDefault)}
+                    </span>
+                    <span className="portal-billing__seglegend-val">
+                      {t(
+                        "portal.billing.pdfsProcessed.legendValue",
+                        "{{formatted}} PDFs",
+                        {
+                          count: perDocs[s.key],
+                          formatted: perDocs[s.key].toLocaleString(),
+                        },
+                      )}
+                    </span>
+                    <span className="portal-billing__seglegend-desc">
+                      {t(s.descKey, s.descDefault)}
+                    </span>
+                  </div>
+                ))}
+              </div>
+            </>
+          ) : null}
+          {sizeMultiplierPdfs > 0 ? (
+            <p className="portal-billing__section-sub">
+              {t(
+                "portal.billing.pdfsProcessed.sizeMultiplier",
+                "{{formatted}} PDFs used a size multiplier",
+                { formatted: sizeMultiplierPdfs.toLocaleString() },
+              )}
+            </p>
+          ) : null}
         </>
       ) : (
         <p className="portal-billing__section-sub">
           {t(
             "portal.billing.pdfsProcessed.emptyPeriod",
-            "No metered processing yet this period.",
+            "No processing yet this period.",
           )}
         </p>
       )}

frontend/editor/src/portal/components/billing/SpendThisMonthCard.stories.tsx

@@ -13,3 +13,8 @@ type Story = StoryObj<typeof SpendThisMonthCard>;
 
 /** Actual spend + the Enterprise upsell tacked onto the foot. */
 export const Default: Story = { args: { wallet: subscribedWallet } };
+
+/** Subscribed but still holding leftover lifetime free grant — shows the free-remaining note. */
+export const WithFreeRemaining: Story = {
+  args: { wallet: { ...subscribedWallet, freeRemaining: 380 } },
+};

frontend/editor/src/portal/components/billing/SpendThisMonthCard.tsx

@@ -46,6 +46,18 @@ export function SpendThisMonthCard({ wallet }: { wallet: Wallet }) {
               },
             )}
       </p>
+      {wallet.freeRemaining > 0 ? (
+        <p className="portal-billing__section-sub portal-billing__free-remaining">
+          {t(
+            "portal.billing.spendThisMonth.freeRemaining",
+            "{{formatted}} free PDFs remaining",
+            {
+              count: wallet.freeRemaining,
+              formatted: wallet.freeRemaining.toLocaleString(),
+            },
+          )}
+        </p>
+      ) : null}
 
       <div className="portal-billing__spend-foot">
         <EnterpriseUpsell bare />

frontend/editor/src/portal/components/billing/billing.css

@@ -116,6 +116,13 @@
   margin: 0 0 1rem;
 }
 
+/* Free grant still offsetting spend — reads as a positive credit note, pulled up
+   under the "PDFs processed" line. */
+.portal-billing__free-remaining {
+  color: var(--color-green-600, var(--color-green));
+  margin-top: -0.5rem;
+}
+
 .portal-billing__skeleton {
   display: flex;
   flex-direction: column;

frontend/editor/src/portal/components/billing/walletFixtures.ts

@@ -19,6 +19,10 @@ export const freeWallet: Wallet = {
   stripeSubscriptionId: null,
   spendUnitsThisPeriod: 120,
   categoryBreakdown: { api: 40, ai: 30, automation: 50 },
+  categoryDocs: { api: 30, ai: 20, automation: 40 },
+  docsProcessedThisPeriod: 90,
+  uniquePdfsThisPeriod: 84,
+  sizeMultiplierPdfsThisPeriod: 12,
   members: [],
   recent: [],
 };
@@ -42,6 +46,10 @@ export const subscribedWallet: Wallet = {
   stripeSubscriptionId: "sub_123",
   spendUnitsThisPeriod: 2250,
   categoryBreakdown: { api: 900, ai: 600, automation: 750 },
+  categoryDocs: { api: 700, ai: 450, automation: 600 },
+  docsProcessedThisPeriod: 1750,
+  uniquePdfsThisPeriod: 1600,
+  sizeMultiplierPdfsThisPeriod: 320,
   members: [
     {
       userId: "u1",

frontend/editor/src/proprietary/billing/types.ts

@@ -60,7 +60,16 @@ export interface Wallet {
   noCap: boolean;
   stripeSubscriptionId: string | null;
   spendUnitsThisPeriod: number;
+  /** Per-category size-scaled meter units (billing quantity). */
   categoryBreakdown: WalletCategoryBreakdown;
+  /** Per-category INPUT-file counts — the count dimension, distinct from the units above. */
+  categoryDocs: WalletCategoryBreakdown;
+  /** Total input files processed this period (Σ doc_count). */
+  docsProcessedThisPeriod: number;
+  /** Distinct input documents this period — a file hit by N operations counts once. */
+  uniquePdfsThisPeriod: number;
+  /** Input files on charges where the size multiplier applied (units billed > input files). */
+  sizeMultiplierPdfsThisPeriod: number;
   /** Populated for the leader view; empty for members / single-seat tenants. */
   members: WalletMember[];
   recent: WalletActivityRow[];

frontend/editor/src/saas/components/shared/FreeLimitReachedModal.test.tsx

@@ -27,6 +27,10 @@ const wallet: Wallet = {
   stripeSubscriptionId: null,
   spendUnitsThisPeriod: 0,
   categoryBreakdown: { api: 0, ai: 0, automation: 0 },
+  categoryDocs: { api: 0, ai: 0, automation: 0 },
+  docsProcessedThisPeriod: 0,
+  uniquePdfsThisPeriod: 0,
+  sizeMultiplierPdfsThisPeriod: 0,
   members: [],
   recent: [],
 };

frontend/editor/src/saas/hooks/walletDevPreview.ts

@@ -67,6 +67,12 @@ function buildDevPreviewWallet(role: WalletRole): Wallet {
     noCap: false,
     stripeSubscriptionId: subscribed ? "sub_devpreview" : null,
     spendUnitsThisPeriod: 62,
+    // Count dimension (illustrative): input files processed vs the size-scaled
+    // meter units above — a few large PDFs pushed some charges past 1 unit.
+    docsProcessedThisPeriod: 50,
+    uniquePdfsThisPeriod: 48,
+    sizeMultiplierPdfsThisPeriod: 8,
+    categoryDocs: { api: 18, ai: 14, automation: 18 },
     // Wave 1 backend (PR #6574) returns a per-category breakdown so the
     // hero panel can split AI / automation / API. Use realistic but
     // tier-distinguishable mock values so the dev preview shows a