From ce6abe6e2384724d04f6818c02124df111e27126 Mon Sep 17 00:00:00 2001 From: ConnorYoh <40631091+ConnorYoh@users.noreply.github.com> Date: Fri, 10 Jul 2026 14:38:22 +0100 Subject: [PATCH] PAYG: size-scaled units + per-input-file PDF count + run-id grouping (#6957) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 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. --- .../common/service/AutomationRunContext.java | 56 ++++++ .../common/service/InternalApiClient.java | 8 + .../api/pipeline/PipelineProcessor.java | 13 ++ .../policy/engine/PolicyEngine.java | 124 +++++++----- .../service/AiWorkflowService.java | 58 +++--- .../ai/controller/AiCreateController.java | 3 +- .../saas/payg/api/PaygWalletController.java | 61 ++++-- .../saas/payg/api/WalletSnapshotResponse.java | 16 +- .../saas/payg/charge/ChargeContext.java | 21 +- .../saas/payg/charge/JobChargeService.java | 42 +++- .../payg/filter/PaygChargeInterceptor.java | 33 +++- .../instance/InstanceUsageIngestService.java | 3 +- .../software/saas/payg/job/JobContext.java | 18 +- .../software/saas/payg/job/JobService.java | 54 ++++- .../software/saas/payg/job/ProcessingJob.java | 21 ++ .../lineage/DefaultHashLineageDetector.java | 15 ++ .../payg/lineage/HashLineageDetector.java | 11 ++ .../saas/payg/lineage/JobLineageStore.java | 13 ++ .../saas/payg/lineage/JpaJobLineageStore.java | 18 ++ .../repository/JobArtifactHashRepository.java | 23 +++ .../repository/WalletLedgerRepository.java | 42 +++- .../saas/payg/wallet/WalletLedgerEntry.java | 20 ++ .../saas/V38__payg_run_grouping_doc_count.sql | 39 ++++ .../payg/api/PaygWalletControllerTest.java | 24 ++- .../payg/api/WalletSnapshotResponseTest.java | 12 +- .../filter/PaygChargeInterceptorTest.java | 46 +++++ .../saas/payg/job/JobServiceTest.java | 38 +++- .../public/locales/en-US/translation.toml | 11 +- frontend/editor/src/cloud/hooks/useWallet.ts | 10 +- .../billing/PdfsProcessedCard.stories.tsx | 6 +- .../components/billing/PdfsProcessedCard.tsx | 187 +++++++++++------- .../billing/SpendThisMonthCard.stories.tsx | 5 + .../components/billing/SpendThisMonthCard.tsx | 12 ++ .../src/portal/components/billing/billing.css | 7 + .../components/billing/walletFixtures.ts | 8 + .../editor/src/proprietary/billing/types.ts | 9 + .../shared/FreeLimitReachedModal.test.tsx | 4 + .../editor/src/saas/hooks/walletDevPreview.ts | 6 + 38 files changed, 897 insertions(+), 200 deletions(-) create mode 100644 app/common/src/main/java/stirling/software/common/service/AutomationRunContext.java create mode 100644 app/saas/src/main/resources/db/migration/saas/V38__payg_run_grouping_doc_count.sql diff --git a/app/common/src/main/java/stirling/software/common/service/AutomationRunContext.java b/app/common/src/main/java/stirling/software/common/service/AutomationRunContext.java new file mode 100644 index 0000000000..9ef4cb0511 --- /dev/null +++ b/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). + * + *

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 separate runs that happen to touch identical bytes stay + * distinct charges (the old content+time-window grouping merged them). + * + *

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. + * + *

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 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(); + } +} diff --git a/app/common/src/main/java/stirling/software/common/service/InternalApiClient.java b/app/common/src/main/java/stirling/software/common/service/InternalApiClient.java index 8df2d5e410..39e27457b3 100644 --- a/app/common/src/main/java/stirling/software/common/service/InternalApiClient.java +++ b/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 diff --git a/app/core/src/main/java/stirling/software/SPDF/controller/api/pipeline/PipelineProcessor.java b/app/core/src/main/java/stirling/software/SPDF/controller/api/pipeline/PipelineProcessor.java index c57a43f27f..c4f1c76f5e 100644 --- a/app/core/src/main/java/stirling/software/SPDF/controller/api/pipeline/PipelineProcessor.java +++ b/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 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 outputFiles, PipelineConfig config) throws Exception { PipelineResult result = new PipelineResult(); ByteArrayOutputStream logStream = new ByteArrayOutputStream(); diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java b/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java index 98a6f90096..cef0f29e6f 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/policy/engine/PolicyEngine.java +++ b/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 completion) { String runId = run.getRunId(); - try { - run.markRunning(); - PolicyExecutionResult result = - stepExecutor.execute(run.getDefinition(), inputs, listener); - OutputSpec output = run.getDefinition().output(); - List 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 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); } } diff --git a/app/proprietary/src/main/java/stirling/software/proprietary/service/AiWorkflowService.java b/app/proprietary/src/main/java/stirling/software/proprietary/service/AiWorkflowService.java index 0e9460eca7..74e68d4602 100644 --- a/app/proprietary/src/main/java/stirling/software/proprietary/service/AiWorkflowService.java +++ b/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 filesById = new LinkedHashMap<>(); - List 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 filesById = new LinkedHashMap<>(); + List 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( diff --git a/app/saas/src/main/java/stirling/software/saas/ai/controller/AiCreateController.java b/app/saas/src/main/java/stirling/software/saas/ai/controller/AiCreateController.java index 371060c032..28bbfbdea0 100644 --- a/app/saas/src/main/java/stirling/software/saas/ai/controller/AiCreateController.java +++ b/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( diff --git a/app/saas/src/main/java/stirling/software/saas/payg/api/PaygWalletController.java b/app/saas/src/main/java/stirling/software/saas/payg/api/PaygWalletController.java index 90616bea55..18da81ae2b 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/api/PaygWalletController.java +++ b/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 byCategory = new HashMap<>(); + Map units = new HashMap<>(); + Map 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 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 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); } } diff --git a/app/saas/src/main/java/stirling/software/saas/payg/api/WalletSnapshotResponse.java b/app/saas/src/main/java/stirling/software/saas/payg/api/WalletSnapshotResponse.java index 0b29347c85..8147bd030c 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/api/WalletSnapshotResponse.java +++ b/app/saas/src/main/java/stirling/software/saas/payg/api/WalletSnapshotResponse.java @@ -75,7 +75,21 @@ public record WalletSnapshotResponse( int spendUnitsThisPeriod, CategoryBreakdown categoryBreakdown, List members, - List recent) { + List 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) {} diff --git a/app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeContext.java b/app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeContext.java index f5914f8536..55553318dd 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/charge/ChargeContext.java +++ b/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}. + * + *

{@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); + } } diff --git a/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java b/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java index ea3bb9cdab..b71caddbee 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/charge/JobChargeService.java +++ b/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 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); } diff --git a/app/saas/src/main/java/stirling/software/saas/payg/filter/PaygChargeInterceptor.java b/app/saas/src/main/java/stirling/software/saas/payg/filter/PaygChargeInterceptor.java index eda04a4789..b758c89787 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/filter/PaygChargeInterceptor.java +++ b/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 = diff --git a/app/saas/src/main/java/stirling/software/saas/payg/instance/InstanceUsageIngestService.java b/app/saas/src/main/java/stirling/software/saas/payg/instance/InstanceUsageIngestService.java index 291a46880c..c672540870 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/instance/InstanceUsageIngestService.java +++ b/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) { diff --git a/app/saas/src/main/java/stirling/software/saas/payg/job/JobContext.java b/app/saas/src/main/java/stirling/software/saas/payg/job/JobContext.java index 7840effa26..a025bc5c57 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/job/JobContext.java +++ b/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); + } } diff --git a/app/saas/src/main/java/stirling/software/saas/payg/job/JobService.java b/app/saas/src/main/java/stirling/software/saas/payg/job/JobService.java index 5addb5b76d..ab4739be6b 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/job/JobService.java +++ b/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 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 findBestMatch( - Long userId, List inputs, Map> signaturesByInput) { + Long userId, + String runId, + List inputs, + Map> signaturesByInput) { List 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> signaturesByInput) { + List 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> signaturesByInput) { for (Set signatures : signaturesByInput.values()) { detector.record(jobId, signatures, ArtifactKind.INPUT); diff --git a/app/saas/src/main/java/stirling/software/saas/payg/job/ProcessingJob.java b/app/saas/src/main/java/stirling/software/saas/payg/job/ProcessingJob.java index d80bad16d4..08b73db6ac 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/job/ProcessingJob.java +++ b/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. + * + *

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; diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetector.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetector.java index 082054a09f..c9b939b975 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/lineage/DefaultHashLineageDetector.java +++ b/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 detect( + Long userId, String runId, Set 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"); diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/HashLineageDetector.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/HashLineageDetector.java index 66559eaf40..2973ad8cf0 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/lineage/HashLineageDetector.java +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/HashLineageDetector.java @@ -50,6 +50,17 @@ public interface HashLineageDetector { */ Optional detect(Long userId, Set 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 detect( + Long userId, String runId, Set 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. diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/JobLineageStore.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/JobLineageStore.java index 7ab46d9bd0..c970da34f8 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/lineage/JobLineageStore.java +++ b/app/saas/src/main/java/stirling/software/saas/payg/lineage/JobLineageStore.java @@ -34,6 +34,19 @@ public interface JobLineageStore { Optional findOpenJobForSignatures( Long userId, Set 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 findOpenJobForSignatures( + Long userId, Set 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); } diff --git a/app/saas/src/main/java/stirling/software/saas/payg/lineage/JpaJobLineageStore.java b/app/saas/src/main/java/stirling/software/saas/payg/lineage/JpaJobLineageStore.java index 4da4cd0b52..09df6ed12f 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/lineage/JpaJobLineageStore.java +++ b/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 findOpenJobForSignatures( + Long userId, Set 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 storageKeys = candidates.stream().map(LineageSignature::asStorageKey).toList(); + LocalDateTime since = LocalDateTime.now().minus(workflowWindow); + List 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) { diff --git a/app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java b/app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java index 0c3af74dbe..7b02dbaebc 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java +++ b/app/saas/src/main/java/stirling/software/saas/payg/repository/JobArtifactHashRepository.java @@ -43,6 +43,29 @@ public interface JobArtifactHashRepository @Param("signatures") Collection 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 findOpenJobsForSignaturesInRun( + @Param("userId") Long userId, + @Param("openStatus") JobStatus openStatus, + @Param("since") LocalDateTime since, + @Param("signatures") Collection 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") diff --git a/app/saas/src/main/java/stirling/software/saas/payg/repository/WalletLedgerRepository.java b/app/saas/src/main/java/stirling/software/saas/payg/repository/WalletLedgerRepository.java index 095f6cb924..f307d54241 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/repository/WalletLedgerRepository.java +++ b/app/saas/src/main/java/stirling/software/saas/payg/repository/WalletLedgerRepository.java @@ -18,14 +18,13 @@ public interface WalletLedgerRepository extends JpaRepository 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= :periodStart" + " AND e.occurredAt < :periodEnd" + " GROUP BY e.billingCategory") - List sumPeriodAmountByCategory( + List 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 > input files). + * DEBIT + non-null category only. + * + *

Returns a single-element {@code List} (aggregate-only query → always one row). Declared as + * {@code List} rather than {@code Object[]}: Spring Data treats an {@code Object[]} + * return as a collection 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 periodUsageAnalytics( @Param("teamId") Long teamId, @Param("entryType") LedgerEntryType entryType, @Param("periodStart") LocalDateTime periodStart, diff --git a/app/saas/src/main/java/stirling/software/saas/payg/wallet/WalletLedgerEntry.java b/app/saas/src/main/java/stirling/software/saas/payg/wallet/WalletLedgerEntry.java index 4971e296f3..fab6647b2c 100644 --- a/app/saas/src/main/java/stirling/software/saas/payg/wallet/WalletLedgerEntry.java +++ b/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. + * + *

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; diff --git a/app/saas/src/main/resources/db/migration/saas/V38__payg_run_grouping_doc_count.sql b/app/saas/src/main/resources/db/migration/saas/V38__payg_run_grouping_doc_count.sql new file mode 100644 index 0000000000..bc1a4e0f16 --- /dev/null +++ b/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; diff --git a/app/saas/src/test/java/stirling/software/saas/payg/api/PaygWalletControllerTest.java b/app/saas/src/test/java/stirling/software/saas/payg/api/PaygWalletControllerTest.java index 0e02c0ce50..1625116af0 100644 --- a/app/saas/src/test/java/stirling/software/saas/payg/api/PaygWalletControllerTest.java +++ b/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.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 + when(ledgerRepo.periodUsageAnalytics(eq(99L), eq(LedgerEntryType.DEBIT), any(), any())) + .thenReturn(List.of(new Object[] {142L, 120L, 30L})); when(ledgerRepo.findTop20ByTeamIdOrderByIdDesc(99L)).thenReturn(List.of()); ResponseEntity 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()); diff --git a/app/saas/src/test/java/stirling/software/saas/payg/api/WalletSnapshotResponseTest.java b/app/saas/src/test/java/stirling/software/saas/payg/api/WalletSnapshotResponseTest.java index 65d55ccb86..e8547de268 100644 --- a/app/saas/src/test/java/stirling/software/saas/payg/api/WalletSnapshotResponseTest.java +++ b/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(); diff --git a/app/saas/src/test/java/stirling/software/saas/payg/filter/PaygChargeInterceptorTest.java b/app/saas/src/test/java/stirling/software/saas/payg/filter/PaygChargeInterceptorTest.java index 128c336c32..be402dd061 100644 --- a/app/saas/src/test/java/stirling/software/saas/payg/filter/PaygChargeInterceptorTest.java +++ b/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 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 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 diff --git a/app/saas/src/test/java/stirling/software/saas/payg/job/JobServiceTest.java b/app/saas/src/test/java/stirling/software/saas/payg/job/JobServiceTest.java index ca7033405a..880a52da54 100644 --- a/app/saas/src/test/java/stirling/software/saas/payg/job/JobServiceTest.java +++ b/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( diff --git a/frontend/editor/public/locales/en-US/translation.toml b/frontend/editor/public/locales/en-US/translation.toml index 29792b0162..2244b6acdc 100644 --- a/frontend/editor/public/locales/en-US/translation.toml +++ b/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." diff --git a/frontend/editor/src/cloud/hooks/useWallet.ts b/frontend/editor/src/cloud/hooks/useWallet.ts index 07e862fb8c..0641535946 100644 --- a/frontend/editor/src/cloud/hooks/useWallet.ts +++ b/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; } diff --git a/frontend/editor/src/portal/components/billing/PdfsProcessedCard.stories.tsx b/frontend/editor/src/portal/components/billing/PdfsProcessedCard.stories.tsx index 6f4c46641c..d7da10398b 100644 --- a/frontend/editor/src/portal/components/billing/PdfsProcessedCard.stories.tsx +++ b/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, }, }, }; diff --git a/frontend/editor/src/portal/components/billing/PdfsProcessedCard.tsx b/frontend/editor/src/portal/components/billing/PdfsProcessedCard.tsx index 00d9806e6a..794ac39ae5 100644 --- a/frontend/editor/src/portal/components/billing/PdfsProcessedCard.tsx +++ b/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. * - *

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. + *

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 ( @@ -78,66 +94,99 @@ export function PdfsProcessedCard({ )}

- - {headline.toLocaleString()} - + {docs.toLocaleString()} - {t("portal.billing.pdfsProcessed.unit", "metered PDFs")} + {t("portal.billing.pdfsProcessed.unit", "PDFs")}
- {total > 0 ? ( + {hasActivity ? ( <> -
- {SEGMENTS.map((s) => - b[s.key] > 0 ? ( - - ) : null, - )} -
-
- {SEGMENTS.map((s) => ( -
- - - {t(s.labelKey, s.labelDefault)} - - - {t( - "portal.billing.pdfsProcessed.legendValue", - "{{formatted}} PDFs", - { - count: b[s.key], - formatted: b[s.key].toLocaleString(), - }, - )} - - - {t(s.descKey, s.descDefault)} - +

+ {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(), + }, + )} +

+ {totalDocs > 0 ? ( + <> +
+ {SEGMENTS.map((s) => + perDocs[s.key] > 0 ? ( + + ) : null, + )}
- ))} -
+
+ {SEGMENTS.map((s) => ( +
+ + + {t(s.labelKey, s.labelDefault)} + + + {t( + "portal.billing.pdfsProcessed.legendValue", + "{{formatted}} PDFs", + { + count: perDocs[s.key], + formatted: perDocs[s.key].toLocaleString(), + }, + )} + + + {t(s.descKey, s.descDefault)} + +
+ ))} +
+ + ) : null} + {sizeMultiplierPdfs > 0 ? ( +

+ {t( + "portal.billing.pdfsProcessed.sizeMultiplier", + "{{formatted}} PDFs used a size multiplier", + { formatted: sizeMultiplierPdfs.toLocaleString() }, + )} +

+ ) : null} ) : (

{t( "portal.billing.pdfsProcessed.emptyPeriod", - "No metered processing yet this period.", + "No processing yet this period.", )}

)} diff --git a/frontend/editor/src/portal/components/billing/SpendThisMonthCard.stories.tsx b/frontend/editor/src/portal/components/billing/SpendThisMonthCard.stories.tsx index a47b3df65b..af1ed1faa0 100644 --- a/frontend/editor/src/portal/components/billing/SpendThisMonthCard.stories.tsx +++ b/frontend/editor/src/portal/components/billing/SpendThisMonthCard.stories.tsx @@ -13,3 +13,8 @@ type Story = StoryObj; /** 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 } }, +}; diff --git a/frontend/editor/src/portal/components/billing/SpendThisMonthCard.tsx b/frontend/editor/src/portal/components/billing/SpendThisMonthCard.tsx index d4a969ae0c..10aebbe32f 100644 --- a/frontend/editor/src/portal/components/billing/SpendThisMonthCard.tsx +++ b/frontend/editor/src/portal/components/billing/SpendThisMonthCard.tsx @@ -46,6 +46,18 @@ export function SpendThisMonthCard({ wallet }: { wallet: Wallet }) { }, )}

+ {wallet.freeRemaining > 0 ? ( +

+ {t( + "portal.billing.spendThisMonth.freeRemaining", + "{{formatted}} free PDFs remaining", + { + count: wallet.freeRemaining, + formatted: wallet.freeRemaining.toLocaleString(), + }, + )} +

+ ) : null}
diff --git a/frontend/editor/src/portal/components/billing/billing.css b/frontend/editor/src/portal/components/billing/billing.css index 63ea82d748..86e9a3ff94 100644 --- a/frontend/editor/src/portal/components/billing/billing.css +++ b/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; diff --git a/frontend/editor/src/portal/components/billing/walletFixtures.ts b/frontend/editor/src/portal/components/billing/walletFixtures.ts index c751f1269a..cd754af3fc 100644 --- a/frontend/editor/src/portal/components/billing/walletFixtures.ts +++ b/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", diff --git a/frontend/editor/src/proprietary/billing/types.ts b/frontend/editor/src/proprietary/billing/types.ts index 83d0c66518..df0caa5e39 100644 --- a/frontend/editor/src/proprietary/billing/types.ts +++ b/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[]; diff --git a/frontend/editor/src/saas/components/shared/FreeLimitReachedModal.test.tsx b/frontend/editor/src/saas/components/shared/FreeLimitReachedModal.test.tsx index 7eac7b4d83..d70d807b25 100644 --- a/frontend/editor/src/saas/components/shared/FreeLimitReachedModal.test.tsx +++ b/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: [], }; diff --git a/frontend/editor/src/saas/hooks/walletDevPreview.ts b/frontend/editor/src/saas/hooks/walletDevPreview.ts index e62ec333d5..1529058f15 100644 --- a/frontend/editor/src/saas/hooks/walletDevPreview.ts +++ b/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