diff --git a/src/sentry/conf/server.py b/src/sentry/conf/server.py
index 277c34037c6247..b47fa7939b7ec5 100644
--- a/src/sentry/conf/server.py
+++ b/src/sentry/conf/server.py
@@ -875,7 +875,7 @@ def SOCIAL_AUTH_DEFAULT_USERNAME() -> str:
     "sentry.deletions.tasks.scheduled",
     "sentry.deletions.tasks.seer",
     "sentry.demo_mode.tasks",
-    "sentry.dynamic_sampling.per_org.tasks.scheduler",
+    "sentry.dynamic_sampling.per_org.scheduler",
     "sentry.dynamic_sampling.tasks.boost_low_volume_projects",
     "sentry.dynamic_sampling.tasks.boost_low_volume_transactions",
     "sentry.dynamic_sampling.tasks.recalibrate_orgs",
diff --git a/src/sentry/dynamic_sampling/per_org/calculations.py b/src/sentry/dynamic_sampling/per_org/calculations.py
new file mode 100644
index 00000000000000..f1924c87c716b6
--- /dev/null
+++ b/src/sentry/dynamic_sampling/per_org/calculations.py
@@ -0,0 +1,97 @@
+from __future__ import annotations
+
+import logging
+from typing import cast
+
+from sentry.dynamic_sampling.models.common import RebalancedItem
+from sentry.dynamic_sampling.models.projects_rebalancing import (
+    ProjectsRebalancingInput,
+    ProjectsRebalancingModel,
+)
+from sentry.dynamic_sampling.per_org.configuration import BaseDynamicSamplingConfiguration
+from sentry.dynamic_sampling.per_org.queries import ProjectVolume
+from sentry.dynamic_sampling.rules.utils import get_redis_client_for_ds
+from sentry.dynamic_sampling.tasks.common import sample_rate_to_float
+from sentry.dynamic_sampling.tasks.helpers.boost_low_volume_projects import (
+    generate_boost_low_volume_projects_cache_key,
+)
+
+PROJECT_BALANCING_COMPARISON_RELATIVE_TOLERANCE = 0.05
+logger = logging.getLogger(__name__)
+
+
+def run_project_balancing(
+    config: BaseDynamicSamplingConfiguration, project_volumes: list[ProjectVolume]
+) -> list[RebalancedItem]:
+    sample_rate = cast(float, config.get_sample_rate())
+    counts_by_project = {project.id: 0 for project in config.projects}
+    for project_volume in project_volumes:
+        if project_volume.project_id in counts_by_project:
+            counts_by_project[project_volume.project_id] = project_volume.total
+    return ProjectsRebalancingModel().run(
+        ProjectsRebalancingInput(
+            classes=[
+                RebalancedItem(id=project_id, count=count)
+                for project_id, count in counts_by_project.items()
+            ],
+            sample_rate=sample_rate,
+        )
+    )
+
+
+def get_cached_rebalanced_project_sample_rates(org_id: int) -> dict[int, float | None]:
+    redis_client = get_redis_client_for_ds()
+    cache_key = generate_boost_low_volume_projects_cache_key(org_id=org_id)
+    return {
+        int(project_id): sample_rate_to_float(sample_rate)
+        for project_id, sample_rate in redis_client.hgetall(cache_key).items()
+    }
+
+
+def is_within_relative_tolerance(
+    cached_sample_rate: float | None,
+    calculated_sample_rate: float,
+    relative_tolerance: float = PROJECT_BALANCING_COMPARISON_RELATIVE_TOLERANCE,
+) -> bool:
+    relative_deviation = get_relative_deviation(cached_sample_rate, calculated_sample_rate)
+    if relative_deviation is None:
+        return False
+    return relative_deviation <= relative_tolerance + 1e-12
+
+
+def get_relative_deviation(
+    cached_sample_rate: float | None, calculated_sample_rate: float
+) -> float | None:
+    if cached_sample_rate is None:
+        return None
+    if calculated_sample_rate == 0:
+        return 0.0 if abs(cached_sample_rate) <= 1e-12 else None
+    return abs(cached_sample_rate - calculated_sample_rate) / abs(calculated_sample_rate)
+
+
+def compare_rebalanced_projects_with_cache(
+    config: BaseDynamicSamplingConfiguration,
+    rebalanced_projects: list[RebalancedItem],
+    cached_sample_rates: dict[int, float | None],
+) -> None:
+    calculated_sample_rates = {
+        int(project.id): project.new_sample_rate for project in rebalanced_projects
+    }
+
+    for project_id, eap_sample_rate in sorted(calculated_sample_rates.items()):
+        generic_metrics_sample_rate = cached_sample_rates.get(project_id)
+        logger.info(
+            "dynamic_sampling.per_org.project_balancing_comparison",
+            extra={
+                "org_id": config.organization.id,
+                "project_id": project_id,
+                "generic_metrics_sample_rate": generic_metrics_sample_rate,
+                "eap_sample_rate": eap_sample_rate,
+                "relative_deviation": get_relative_deviation(
+                    generic_metrics_sample_rate, eap_sample_rate
+                ),
+                "is_equal": is_within_relative_tolerance(
+                    generic_metrics_sample_rate, eap_sample_rate
+                ),
+            },
+        )
diff --git a/src/sentry/dynamic_sampling/per_org/tasks/configuration.py b/src/sentry/dynamic_sampling/per_org/configuration.py
similarity index 78%
rename from src/sentry/dynamic_sampling/per_org/tasks/configuration.py
rename to src/sentry/dynamic_sampling/per_org/configuration.py
index a7635681759279..69f93948a102bd 100644
--- a/src/sentry/dynamic_sampling/per_org/tasks/configuration.py
+++ b/src/sentry/dynamic_sampling/per_org/configuration.py
@@ -8,8 +8,8 @@
 
 from sentry import options, quotas
 from sentry.constants import SAMPLING_MODE_DEFAULT, TARGET_SAMPLE_RATE_DEFAULT, ObjectStatus
-from sentry.dynamic_sampling.per_org.tasks.queries import get_eap_organization_volume
-from sentry.dynamic_sampling.per_org.tasks.telemetry import (
+from sentry.dynamic_sampling.per_org.queries import get_eap_organization_volume
+from sentry.dynamic_sampling.per_org.telemetry import (
     DynamicSamplingException,
     DynamicSamplingStatus,
 )
@@ -99,6 +99,14 @@ def get_sample_rate(self) -> TargetSampleRate:
 
 
 class AutomaticDynamicSamplingConfiguration(BaseDynamicSamplingConfiguration):
+    """
+    This configuration is used for organizations for which sample rates are computed based on the volume sent.
+    It currently applies to AM2 organizations.
+    It consists of the following components:
+    - The sample rate is either based on the reserved volume or the ingested volume, which are then used as input to the quotas API to get the sample rate.
+    - Projects are rebalanced
+    """
+
     sample_rate: TargetSampleRate
 
     def __init__(self, organization: Organization) -> None:
@@ -125,6 +133,10 @@ def get_sample_rate(self) -> TargetSampleRate:
         return self.sample_rate
 
     def _get_sliding_window_sample_rate(self) -> TargetSampleRate:
+        """
+        The sliding window sample rate uses the ingested segment volume to compute the sample rate by extrapolating
+        the volume of the current month and then using the quotas API to get the sample rate for that volume.
+        """
         if not self.projects:
             return None
 
@@ -143,6 +155,14 @@ def _get_sliding_window_sample_rate(self) -> TargetSampleRate:
 
 
 class CustomDynamicSamplingOrganizationConfiguration(BaseDynamicSamplingConfiguration):
+    """
+    This configuration is used for organizations for which sample rates are computed based on the target sample rate option.
+    It currently applies to AM3 organizations with custom dynamic sampling enabled and sampling mode set to organization.
+    It consists of the following components:
+    - The sample rate is based on the target sample rate option that can be set by the user at the organzation level. There is no volume-based sample rate computation.
+    - Projects are rebalanced
+    """
+
     sample_rate: TargetSampleRate
 
     def __init__(self, organization: Organization) -> None:
@@ -163,6 +183,14 @@ def get_sample_rate(self) -> TargetSampleRate:
 
 
 class CustomDynamicSamplingProjectConfiguration(BaseDynamicSamplingConfiguration):
+    """
+    This configuration is used for organizations for which sample rates are computed based on the target sample rate option.
+    It currently applies to AM3 organizations with custom dynamic sampling enabled and sampling mode set to project.
+    It consists of the following components:
+    - The sample rate is based on the target sample rate option that can be set by the user at the project level. There is no volume-based sample rate computation.
+    - Projects are not rebalanced
+    """
+
     project_target_sample_rates: ProjectTargetSampleRates
     should_balance_projects: bool = False
 
diff --git a/src/sentry/dynamic_sampling/per_org/tasks/gate.py b/src/sentry/dynamic_sampling/per_org/gate.py
similarity index 100%
rename from src/sentry/dynamic_sampling/per_org/tasks/gate.py
rename to src/sentry/dynamic_sampling/per_org/gate.py
diff --git a/src/sentry/dynamic_sampling/per_org/tasks/queries.py b/src/sentry/dynamic_sampling/per_org/queries.py
similarity index 100%
rename from src/sentry/dynamic_sampling/per_org/tasks/queries.py
rename to src/sentry/dynamic_sampling/per_org/queries.py
diff --git a/src/sentry/dynamic_sampling/per_org/tasks/scheduler.py b/src/sentry/dynamic_sampling/per_org/scheduler.py
similarity index 83%
rename from src/sentry/dynamic_sampling/per_org/tasks/scheduler.py
rename to src/sentry/dynamic_sampling/per_org/scheduler.py
index 6a710e6178292b..540c6c00c7f83b 100644
--- a/src/sentry/dynamic_sampling/per_org/tasks/scheduler.py
+++ b/src/sentry/dynamic_sampling/per_org/scheduler.py
@@ -8,14 +8,19 @@
 from django.db.models.functions import Mod
 from taskbroker_client.retry import Retry
 
-from sentry.dynamic_sampling.per_org.tasks.configuration import get_configuration
-from sentry.dynamic_sampling.per_org.tasks.gate import is_org_in_rollout
-from sentry.dynamic_sampling.per_org.tasks.queries import (
+from sentry.dynamic_sampling.per_org.calculations import (
+    compare_rebalanced_projects_with_cache,
+    get_cached_rebalanced_project_sample_rates,
+    run_project_balancing,
+)
+from sentry.dynamic_sampling.per_org.configuration import get_configuration
+from sentry.dynamic_sampling.per_org.gate import is_org_in_rollout
+from sentry.dynamic_sampling.per_org.queries import (
     get_eap_organization_volume,
     get_eap_project_volumes,
     get_eap_transaction_volumes,
 )
-from sentry.dynamic_sampling.per_org.tasks.telemetry import (
+from sentry.dynamic_sampling.per_org.telemetry import (
     SCHEDULER_BUCKET_ORG_STATUS_METRIC,
     DynamicSamplingStatus,
     emit_status,
@@ -116,6 +121,9 @@ def run_calculations_per_org_task(org_id: OrganizationId) -> DynamicSamplingStat
         project_volumes = get_eap_project_volumes(config)
         if not project_volumes:
             return DynamicSamplingStatus.NO_PROJECT_VOLUMES
+        rebalanced_projects = run_project_balancing(config, project_volumes)
+        cached_sample_rates = get_cached_rebalanced_project_sample_rates(config.organization.id)
+        compare_rebalanced_projects_with_cache(config, rebalanced_projects, cached_sample_rates)
 
     if not get_eap_transaction_volumes(config):
         return DynamicSamplingStatus.NO_TRANSACTION_VOLUMES
diff --git a/src/sentry/dynamic_sampling/per_org/tasks/__init__.py b/src/sentry/dynamic_sampling/per_org/tasks/__init__.py
deleted file mode 100644
index e69de29bb2d1d6..00000000000000
diff --git a/src/sentry/dynamic_sampling/per_org/tasks/telemetry.py b/src/sentry/dynamic_sampling/per_org/telemetry.py
similarity index 98%
rename from src/sentry/dynamic_sampling/per_org/tasks/telemetry.py
rename to src/sentry/dynamic_sampling/per_org/telemetry.py
index fa9d87e98d2459..74c84b51afd4de 100644
--- a/src/sentry/dynamic_sampling/per_org/tasks/telemetry.py
+++ b/src/sentry/dynamic_sampling/per_org/telemetry.py
@@ -8,7 +8,7 @@
 
 import sentry_sdk
 
-from sentry.dynamic_sampling.per_org.tasks.gate import (
+from sentry.dynamic_sampling.per_org.gate import (
     is_killswitch_engaged,
     is_rollout_enabled,
     metrics_sample_rate,
diff --git a/src/sentry/features/temporary.py b/src/sentry/features/temporary.py
index c39c63008d1929..cb94a3b9a5f780 100644
--- a/src/sentry/features/temporary.py
+++ b/src/sentry/features/temporary.py
@@ -117,6 +117,7 @@ def register_temporary_features(manager: FeatureManager) -> None:
     manager.add("organizations:gen-ai-consent", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable LLM-generated title and description for external issue details
     manager.add("organizations:external-issues-ai-generate", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
+    manager.add("organizations:inbound-filters-v2", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=True)
     # Enable increased issue_owners rate limit for auto-assignment
     manager.add("organizations:increased-issue-owners-rate-limit", OrganizationFeature, FeatureHandlerStrategy.FLAGPOLE, api_expose=False)
     # Starfish: extract metrics from the spans
diff --git a/static/app/components/core/info/index.tsx b/static/app/components/core/info/index.tsx
index 484563ac10663f..527b17bc14e49d 100644
--- a/static/app/components/core/info/index.tsx
+++ b/static/app/components/core/info/index.tsx
@@ -1,2 +1,2 @@
-export {InfoText} from './infoText';
+export {InfoText, type InfoTextProps} from './infoText';
 export {InfoTip, DisabledTip} from './infoTip';
diff --git a/static/app/components/core/info/infoText.tsx b/static/app/components/core/info/infoText.tsx
index 3a31d84b62e038..a41be66070365c 100644
--- a/static/app/components/core/info/infoText.tsx
+++ b/static/app/components/core/info/infoText.tsx
@@ -4,18 +4,17 @@ import type {DistributedOmit} from 'type-fest';
 import {Text, type TextProps} from '@sentry/scraps/text';
 import {Tooltip, type TooltipProps} from '@sentry/scraps/tooltip';
 
-type InfoTextProps<T extends 'span' | 'p' | 'label' | 'div'> = DistributedOmit<
-  TextProps<T>,
-  'title' | 'variant'
-> & {
-  title: React.ReactNode;
-  variant?: TooltipProps['underlineColor'] | 'inherit';
-} & Pick<TooltipProps, 'position'>;
+export type InfoTextProps<T extends 'span' | 'p' | 'label' | 'div' | 'time'> =
+  DistributedOmit<TextProps<T>, 'title' | 'variant'> & {
+    title: React.ReactNode;
+    variant?: TooltipProps['underlineColor'] | 'inherit';
+  } & Pick<TooltipProps, 'position' | 'maxWidth'>;
 
-export function InfoText<T extends 'span' | 'p' | 'label' | 'div' = 'span'>({
+export function InfoText<T extends 'span' | 'p' | 'label' | 'div' | 'time' = 'span'>({
   title,
   children,
   position,
+  maxWidth,
   ...textProps
 }: InfoTextProps<T>) {
   if (!title) {
@@ -25,6 +24,7 @@ export function InfoText<T extends 'span' | 'p' | 'label' | 'div' = 'span'>({
     <Tooltip
       title={title}
       position={position}
+      maxWidth={maxWidth}
       skipWrapper
       isHoverable
       showUnderline
diff --git a/static/app/components/feedback/feedbackItem/messageTitle.tsx b/static/app/components/feedback/feedbackItem/messageTitle.tsx
index 7ce8583c808db0..f3ccbe72d1556e 100644
--- a/static/app/components/feedback/feedbackItem/messageTitle.tsx
+++ b/static/app/components/feedback/feedbackItem/messageTitle.tsx
@@ -44,12 +44,13 @@ export function MessageTitle({feedbackItem, eventData}: Props) {
         ) : null}
         <StyledTimeSince
           date={feedbackItem.firstSeen}
-          tooltipProps={{
-            title: eventData ? (
+          disabledAbsoluteTooltip={!eventData}
+          tooltipBody={
+            eventData ? (
               <FeedbackTimestampsTooltip feedbackItem={feedbackItem} />
-            ) : undefined,
-            overlayStyle: {maxWidth: 300},
-          }}
+            ) : undefined
+          }
+          maxWidth={300}
         />
       </Flex>
     </Flex>
diff --git a/static/app/components/timeSince.tsx b/static/app/components/timeSince.tsx
index 12aded88dd5827..0f0d2a4abbf34b 100644
--- a/static/app/components/timeSince.tsx
+++ b/static/app/components/timeSince.tsx
@@ -2,8 +2,7 @@ import {Fragment, useEffect, useMemo, useState} from 'react';
 import isNumber from 'lodash/isNumber';
 import moment from 'moment-timezone';
 
-import type {TooltipProps} from '@sentry/scraps/tooltip';
-import {Tooltip} from '@sentry/scraps/tooltip';
+import {InfoText, type InfoTextProps} from '@sentry/scraps/info';
 
 import {t} from 'sentry/locale';
 import {getDuration} from 'sentry/utils/duration/getDuration';
@@ -19,7 +18,10 @@ type RelaxedDateType = string | number | Date;
 
 type UnitStyle = 'human' | 'regular' | 'short' | 'extraShort';
 
-interface Props extends React.TimeHTMLAttributes<HTMLTimeElement> {
+interface Props extends Omit<
+  React.TimeHTMLAttributes<HTMLTimeElement>,
+  'color' | 'title'
+> {
   /**
    * The date value, can be string, number (e.g. timestamp), or instance of Date
    *
@@ -31,10 +33,6 @@ interface Props extends React.TimeHTMLAttributes<HTMLTimeElement> {
    * that
    */
   disabledAbsoluteTooltip?: boolean;
-  /**
-   * Tooltip text to be hoverable when isTooltipHoverable is true
-   */
-  isTooltipHoverable?: boolean;
   /**
    * How often should the component live update the timestamp.
    *
@@ -43,6 +41,10 @@ interface Props extends React.TimeHTMLAttributes<HTMLTimeElement> {
    * @default minute
    */
   liveUpdateInterval?: 'minute' | 'second' | number;
+  /**
+   * Max width of the tooltip
+   */
+  maxWidth?: InfoTextProps<'time'>['maxWidth'];
   /**
    * Prefix before upcoming time (when the date is in the future)
    *
@@ -65,23 +67,10 @@ interface Props extends React.TimeHTMLAttributes<HTMLTimeElement> {
    * time is for
    */
   tooltipPrefix?: React.ReactNode;
-  /**
-   * Any other props for the <Tooltip>
-   */
-  tooltipProps?: Partial<TooltipProps>;
   /**
    * Include seconds in the tooltip
    */
   tooltipShowSeconds?: boolean;
-  /**
-   * Suffix content to add to the tooltip. Useful to indicate what the relative
-   * time is for
-   */
-  tooltipSuffix?: React.ReactNode;
-  /**
-   * Change the color of the underline
-   */
-  tooltipUnderlineColor?: 'warning' | 'danger' | 'success' | 'muted';
   /**
    * How much text should be used for the suffix:
    *
@@ -103,6 +92,10 @@ interface Props extends React.TimeHTMLAttributes<HTMLTimeElement> {
    * @default human
    */
   unitStyle?: UnitStyle;
+  /**
+   * Change the color of the underline
+   */
+  variant?: InfoTextProps<'time'>['variant'];
 }
 
 export function TimeSince({
@@ -111,10 +104,8 @@ export function TimeSince({
   tooltipShowSeconds,
   tooltipPrefix: tooltipTitle,
   tooltipBody,
-  tooltipSuffix,
-  tooltipUnderlineColor,
-  tooltipProps,
-  isTooltipHoverable = false,
+  variant = 'inherit',
+  maxWidth,
   unitStyle,
   prefix = t('in'),
   suffix = t('ago'),
@@ -159,24 +150,23 @@ export function TimeSince({
   const tooltip = moment.tz(dateObj, tz).format(format);
 
   return (
-    <Tooltip
-      disabled={disabledAbsoluteTooltip}
-      underlineColor={tooltipUnderlineColor}
-      showUnderline
-      isHoverable={isTooltipHoverable}
+    <InfoText
+      as="time"
+      dateTime={dateObj?.toISOString()}
+      variant={variant}
+      maxWidth={maxWidth}
       title={
-        <Fragment>
-          {tooltipTitle && <div>{tooltipTitle}</div>}
-          {tooltipBody ?? tooltip}
-          {tooltipSuffix && <div>{tooltipSuffix}</div>}
-        </Fragment>
+        disabledAbsoluteTooltip ? null : (
+          <Fragment>
+            {tooltipTitle && <div>{tooltipTitle}</div>}
+            {tooltipBody ?? tooltip}
+          </Fragment>
+        )
       }
-      {...tooltipProps}
+      {...props}
     >
-      <time dateTime={dateObj?.toISOString()} {...props}>
-        {relative}
-      </time>
-    </Tooltip>
+      {relative}
+    </InfoText>
   );
 }
 
diff --git a/static/app/views/explore/conversations/components/messagesPanel.tsx b/static/app/views/explore/conversations/components/messagesPanel.tsx
index 8de2a320d773ba..5827afb7dfe9e0 100644
--- a/static/app/views/explore/conversations/components/messagesPanel.tsx
+++ b/static/app/views/explore/conversations/components/messagesPanel.tsx
@@ -133,7 +133,11 @@ export function MessagesPanel({nodes, selectedNodeId, onSelectNode}: MessagesPan
               >
                 <Container padding="md">
                   <MessageText size="sm" align="left">
-                    <AIContentRenderer text={message.content} inline />
+                    <AIContentRenderer
+                      text={message.content}
+                      inline
+                      autoCollapseLimit={10}
+                    />
                   </MessageText>
                 </Container>
               </StyledClippedBox>
diff --git a/static/app/views/explore/releases/detail/commitsAndFiles/releaseCommit.tsx b/static/app/views/explore/releases/detail/commitsAndFiles/releaseCommit.tsx
index 56a69969618e78..af8b8c56ddb5a1 100644
--- a/static/app/views/explore/releases/detail/commitsAndFiles/releaseCommit.tsx
+++ b/static/app/views/explore/releases/detail/commitsAndFiles/releaseCommit.tsx
@@ -102,7 +102,7 @@ export function ReleaseCommit({commit}: ReleaseCommitProps) {
                 />
               ),
             })}
-            <TimeSince date={commit.dateCreated} tooltipUnderlineColor="muted" />
+            <TimeSince date={commit.dateCreated} variant="muted" />
           </Meta>
         </MetaWrapper>
       </CommitContent>
diff --git a/static/app/views/issueDetails/activitySection/index.tsx b/static/app/views/issueDetails/activitySection/index.tsx
index 4111701496e137..1386643232aa74 100644
--- a/static/app/views/issueDetails/activitySection/index.tsx
+++ b/static/app/views/issueDetails/activitySection/index.tsx
@@ -176,7 +176,7 @@ function TimelineItem({
           )}
         </Flex>
       }
-      timestamp={<Timestamp date={item.dateCreated} tooltipProps={{skipWrapper: true}} />}
+      timestamp={<Timestamp date={item.dateCreated} />}
       marker={useTwoColumnLayout ? getActivityMarker(item, colorConfig.icon) : undefined}
       colorConfig={useTwoColumnLayout ? colorConfig : undefined}
       icon={
diff --git a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/lowValueSpanIssueDetails.spec.tsx b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/lowValueSpanIssueDetails.spec.tsx
index 137107f27629d2..3befb3b1076cac 100644
--- a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/lowValueSpanIssueDetails.spec.tsx
+++ b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/lowValueSpanIssueDetails.spec.tsx
@@ -13,9 +13,10 @@ describe('LowValueSpanIssueDetails', () => {
         event={EventFixture({
           occurrence: {
             evidenceData: {
-              op: 'function',
-              description: 'compute_checksum',
-              count: 1234,
+              span_op: 'function',
+              span_description: 'compute_checksum',
+              span_count: 1234,
+              extrapolated_count: 60_000,
               value_score: 0.15,
               avg_duration_ms: 0.4,
               estimated_cost_usd: 12.34,
@@ -32,6 +33,7 @@ describe('LowValueSpanIssueDetails', () => {
     expect(screen.getByText('Problem')).toBeInTheDocument();
     expect(screen.getByText('Troubleshooting')).toBeInTheDocument();
     expect(screen.getAllByText('function - compute_checksum').length).toBeGreaterThan(0);
+    expect(screen.getByText('60,000')).toBeInTheDocument();
     expect(screen.getByText('$12.34')).toBeInTheDocument();
     expect(screen.queryByText('Value score')).not.toBeInTheDocument();
     expect(screen.queryByText('15%')).not.toBeInTheDocument();
@@ -39,12 +41,15 @@ describe('LowValueSpanIssueDetails', () => {
     expect(screen.queryByText('Impact')).not.toBeInTheDocument();
   });
 
-  it('only reads the backend evidence field names', () => {
+  it('only reads the new backend evidence field names', () => {
     render(
       <LowValueSpanIssueDetails
         event={EventFixture({
           occurrence: {
             evidenceData: {
+              op: 'function',
+              description: 'compute_checksum',
+              count: 1234,
               spanOp: 'function',
               spanDescription: 'compute_checksum',
               seenCount: 1234,
diff --git a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/problemSection.spec.tsx b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/problemSection.spec.tsx
index 84660cd6aae94e..58a847768ce2c5 100644
--- a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/problemSection.spec.tsx
+++ b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/problemSection.spec.tsx
@@ -6,10 +6,12 @@ import type {LowValueSpanEvidenceData} from './types';
 const evidenceData: LowValueSpanEvidenceData = {
   op: 'function',
   description: 'compute_checksum',
-  count: 1234,
+  spanCount: 1234,
+  extrapolatedCount: 60_000,
   avgDurationMs: 0.4,
   estimatedCostUsd: 12.34,
   sdkName: 'sentry.javascript.nextjs',
+  spanOrigin: 'auto',
 };
 
 describe('LowValueSpanIssues ProblemSection', () => {
@@ -17,18 +19,28 @@ describe('LowValueSpanIssues ProblemSection', () => {
     render(<ProblemSection evidenceData={evidenceData} />);
 
     expect(screen.getByText('Problem')).toBeInTheDocument();
-    expect(screen.getByText('function')).toBeInTheDocument();
     expect(screen.getByText('function - compute_checksum')).toBeInTheDocument();
-    expect(screen.getByText('1,234')).toBeInTheDocument();
+    expect(screen.getByText('Span count')).toBeInTheDocument();
+    expect(screen.getByText('60,000')).toBeInTheDocument();
+    expect(screen.getAllByLabelText('More information')).toHaveLength(2);
     expect(screen.getByText('Estimated cost')).toBeInTheDocument();
     expect(screen.getByText('$12.34')).toBeInTheDocument();
-    expect(
-      screen.getByLabelText('More information', {
-        selector: '[aria-label="More information"]',
-      })
-    ).toBeInTheDocument();
     expect(screen.getByText('<1ms')).toBeInTheDocument();
-    expect(screen.getByText('sentry.javascript.nextjs')).toBeInTheDocument();
+    expect(screen.queryByText('sentry.javascript.nextjs')).not.toBeInTheDocument();
+  });
+
+  it('falls back to the sampled span count when extrapolated count is unavailable', () => {
+    render(
+      <ProblemSection
+        evidenceData={{
+          ...evidenceData,
+          extrapolatedCount: null,
+        }}
+      />
+    );
+
+    expect(screen.getByText('1,234')).toBeInTheDocument();
+    expect(screen.getAllByLabelText('More information')).toHaveLength(1);
   });
 
   it('does not render estimated cost when unavailable', () => {
diff --git a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/problemSection.tsx b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/problemSection.tsx
index 0ebe2f70566604..1cdca46d963496 100644
--- a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/problemSection.tsx
+++ b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/problemSection.tsx
@@ -1,6 +1,5 @@
 import type {ReactNode} from 'react';
 
-import {InlineCode} from '@sentry/scraps/code';
 import {InfoTip} from '@sentry/scraps/info';
 import {Flex, Stack} from '@sentry/scraps/layout';
 import {Heading, Text} from '@sentry/scraps/text';
@@ -32,16 +31,16 @@ function DetailRow({label, value}: {label: string; value: ReactNode}) {
 export function ProblemSection({evidenceData}: ProblemSectionProps) {
   const hasEstimatedCost =
     evidenceData.estimatedCostUsd !== null && evidenceData.estimatedCostUsd > 0;
+  const spanCount = evidenceData.extrapolatedCount ?? evidenceData.spanCount;
 
   return (
     <Stack gap="lg" padding="lg">
       <Flex align="baseline" gap="sm" wrap="wrap">
         <Heading as="h3">{t('Problem')}</Heading>
-        {evidenceData.op && <InlineCode>{evidenceData.op}</InlineCode>}
       </Flex>
       <Text>
         {t(
-          'Sentry detected a span that appears frequently but adds low-value telemetry. It can make traces noisier and increase stored span volume without adding useful debugging context.'
+          'Sentry found a frequently created span that adds little value. It can make traces harder to read and increase stored span volume.'
         )}
       </Text>
       <Text>
@@ -50,7 +49,20 @@ export function ProblemSection({evidenceData}: ProblemSectionProps) {
         })}
       </Text>
       <Stack gap="xs">
-        <DetailRow label={t('Seen')} value={formatCount(evidenceData.count)} />
+        <Flex align="baseline" gap="sm" wrap="wrap">
+          <Text variant="muted">{t('Span count')}</Text>
+          <Flex align="center" gap="xs">
+            <Text>{formatCount(spanCount)}</Text>
+            {evidenceData.extrapolatedCount !== null && (
+              <InfoTip
+                size="xs"
+                title={t(
+                  'This monthly volume is extrapolated from a recent sample of this span, so it may not match the final span volume for the billing period.'
+                )}
+              />
+            )}
+          </Flex>
+        </Flex>
         {hasEstimatedCost && (
           <Flex align="baseline" gap="sm" wrap="wrap">
             <Text variant="muted">{t('Estimated cost')}</Text>
@@ -69,12 +81,6 @@ export function ProblemSection({evidenceData}: ProblemSectionProps) {
           label={t('Average duration')}
           value={formatDurationMs(evidenceData.avgDurationMs)}
         />
-        {evidenceData.sdkName && (
-          <DetailRow
-            label={t('SDK')}
-            value={<InlineCode>{evidenceData.sdkName}</InlineCode>}
-          />
-        )}
       </Stack>
     </Stack>
   );
diff --git a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/troubleshootingSection.spec.tsx b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/troubleshootingSection.spec.tsx
index 472ac6cf5a6b6c..212bbb3604623d 100644
--- a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/troubleshootingSection.spec.tsx
+++ b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/troubleshootingSection.spec.tsx
@@ -6,25 +6,43 @@ import type {LowValueSpanEvidenceData} from './types';
 const baseEvidenceData: LowValueSpanEvidenceData = {
   op: 'function',
   description: 'compute_checksum',
-  count: 1234,
+  spanCount: 1234,
+  extrapolatedCount: 60_000,
   avgDurationMs: 0.4,
   estimatedCostUsd: 12.34,
   sdkName: 'sentry.javascript.nextjs',
+  spanOrigin: 'auto',
 };
 
 describe('LowValueSpanIssues TroubleshootingSection', () => {
-  it('renders the two troubleshooting paths', () => {
+  it('renders automatic instrumentation guidance for auto spans', () => {
     render(<TroubleshootingSection evidenceData={baseEvidenceData} />);
 
     expect(screen.getByText('Troubleshooting')).toBeInTheDocument();
-    expect(screen.getByText('1. Find where the span is created')).toBeInTheDocument();
-    expect(
-      screen.getByText('2. Remove custom instrumentation when possible')
-    ).toBeInTheDocument();
-    expect(
-      screen.getByText('3. Filter automatic instrumentation exactly')
-    ).toBeInTheDocument();
-    expect(screen.getByText('function - compute_checksum')).toBeInTheDocument();
+    expect(screen.getByText('ignoreSpans')).toBeInTheDocument();
+    expect(screen.queryByText('function - compute_checksum')).not.toBeInTheDocument();
+    expect(screen.queryByText('sentry.javascript.nextjs')).not.toBeInTheDocument();
+    expect(screen.queryByText('1. Find the custom span')).not.toBeInTheDocument();
+    expect(screen.queryByText('2. Remove or replace the span')).not.toBeInTheDocument();
+  });
+
+  it('renders manual instrumentation guidance only for manual spans', () => {
+    render(
+      <TroubleshootingSection
+        evidenceData={{
+          ...baseEvidenceData,
+          spanOrigin: 'manual',
+        }}
+      />
+    );
+
+    expect(screen.getByText('1. Find the custom span')).toBeInTheDocument();
+    expect(screen.getByText('2. Remove or replace the span')).toBeInTheDocument();
+    expect(screen.getByText(/delete the custom span line/)).toBeInTheDocument();
+    expect(screen.getAllByText('function - compute_checksum').length).toBeGreaterThan(0);
+    expect(screen.queryByText('sentry.javascript.nextjs')).not.toBeInTheDocument();
+    expect(screen.queryByText('ignoreSpans')).not.toBeInTheDocument();
+    expect(screen.queryByText('before_send_transaction')).not.toBeInTheDocument();
   });
 
   it('recommends JavaScript span filtering and mentions beforeSendSpan', () => {
@@ -61,9 +79,21 @@ describe('LowValueSpanIssues TroubleshootingSection', () => {
       />
     );
 
-    expect(
-      screen.getByText(/Check your SDK tracing options and add an exact-match filter/)
-    ).toBeInTheDocument();
+    expect(screen.getByText(/Add an exact-match span filter/)).toBeInTheDocument();
     expect(screen.queryByText(/beforeSendTransaction/)).not.toBeInTheDocument();
   });
+
+  it('treats missing span origin as automatic instrumentation', () => {
+    render(
+      <TroubleshootingSection
+        evidenceData={{
+          ...baseEvidenceData,
+          spanOrigin: null,
+        }}
+      />
+    );
+
+    expect(screen.getByText('ignoreSpans')).toBeInTheDocument();
+    expect(screen.queryByText('1. Find the custom span')).not.toBeInTheDocument();
+  });
 });
diff --git a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/troubleshootingSection.tsx b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/troubleshootingSection.tsx
index 5045c21eea6b82..9c5e6da209732e 100644
--- a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/troubleshootingSection.tsx
+++ b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/troubleshootingSection.tsx
@@ -30,13 +30,10 @@ function AutomaticInstrumentationFix({
     return (
       <Stack gap="md">
         <Text>
-          {tct(
-            'For Python automatic instrumentation, use [hook] to remove matching spans from [spans].',
-            {
-              hook: <InlineCode>before_send_transaction</InlineCode>,
-              spans: <InlineCode>event["spans"]</InlineCode>,
-            }
-          )}
+          {tct('Use [hook] to remove matching spans from [spans].', {
+            hook: <InlineCode>before_send_transaction</InlineCode>,
+            spans: <InlineCode>event["spans"]</InlineCode>,
+          })}
         </Text>
         <CodeBlock language="python">
           {getPythonSpanFilterSnippet(evidenceData)}
@@ -50,7 +47,7 @@ function AutomaticInstrumentationFix({
       <Stack gap="md">
         <Text>
           {tct(
-            'For JavaScript automatic instrumentation, use [ignoreSpans] to drop this exact span before it is sent. Use [beforeSendSpan] only if you need to transform span data instead of dropping it.',
+            'Use [ignoreSpans] to drop this exact span. Use [beforeSendSpan] only when you need to change span data.',
             {
               beforeSendSpan: <InlineCode>beforeSendSpan</InlineCode>,
               ignoreSpans: <InlineCode>ignoreSpans</InlineCode>,
@@ -67,51 +64,54 @@ function AutomaticInstrumentationFix({
   return (
     <Text>
       {tct(
-        'Check your SDK tracing options and add an exact-match filter for [span]. Filter only this operation and description so useful spans with similar names keep flowing.',
+        'Add an exact-match span filter for [span]. Match both operation and description so similar useful spans still get sent.',
         {span: <SpanCode>{getSpanLabel(evidenceData)}</SpanCode>}
       )}
     </Text>
   );
 }
 
-export function TroubleshootingSection({evidenceData}: TroubleshootingSectionProps) {
+function ManualInstrumentationFix({
+  evidenceData,
+}: {
+  evidenceData: LowValueSpanEvidenceData;
+}) {
   return (
-    <Stack gap="lg" padding="lg">
-      <Heading as="h3">{t('Troubleshooting')}</Heading>
+    <Stack gap="md">
+      <Text>{t('This appears to come from custom instrumentation in your code.')}</Text>
+      <Stack gap="xs">
+        <Heading as="h4">{t('1. Find the custom span')}</Heading>
+        <Text>
+          {tct('Search your codebase for [span].', {
+            span: <SpanCode>{getSpanLabel(evidenceData)}</SpanCode>,
+          })}
+        </Text>
+      </Stack>
+      <Stack gap="xs">
+        <Heading as="h4">{t('2. Remove or replace the span')}</Heading>
+        <Text>
+          {t(
+            'If this span is not useful for debugging, delete the custom span line or replace it with a more meaningful span.'
+          )}
+        </Text>
+      </Stack>
+    </Stack>
+  );
+}
+
+function AutomaticInstrumentationTroubleshooting({
+  evidenceData,
+}: {
+  evidenceData: LowValueSpanEvidenceData;
+}) {
+  return (
+    <Stack gap="md">
       <Text>
         {t(
-          'Start by confirming whether this span is created by custom code or by SDK automatic instrumentation. Then remove it at the source or filter the exact span before it is sent.'
+          'This appears to come from SDK automatic instrumentation. Filter the exact operation and description before the span is sent.'
         )}
       </Text>
-      <Stack gap="md">
-        <Stack gap="xs">
-          <Heading as="h4">{t('1. Find where the span is created')}</Heading>
-          <Text>
-            {tct('Search for code or SDK configuration that creates [span].', {
-              span: <SpanCode>{getSpanLabel(evidenceData)}</SpanCode>,
-            })}
-          </Text>
-          {evidenceData.sdkName && (
-            <Text>
-              {tct('The latest evidence points to the [sdk] SDK.', {
-                sdk: <InlineCode>{evidenceData.sdkName}</InlineCode>,
-              })}
-            </Text>
-          )}
-        </Stack>
-        <Stack gap="xs">
-          <Heading as="h4">{t('2. Remove custom instrumentation when possible')}</Heading>
-          <Text>
-            {t(
-              'If this span is manually instrumented and does not describe work you need for debugging, delete the custom span creation line or replace it with a more meaningful span.'
-            )}
-          </Text>
-        </Stack>
-        <Stack gap="xs">
-          <Heading as="h4">{t('3. Filter automatic instrumentation exactly')}</Heading>
-          <AutomaticInstrumentationFix evidenceData={evidenceData} />
-        </Stack>
-      </Stack>
+      <AutomaticInstrumentationFix evidenceData={evidenceData} />
       <Flex align="center" gap="xs">
         <IconDocs size="xs" />
         <ExternalLink href={getSpanFilteringDocsUrl(evidenceData)}>
@@ -121,3 +121,18 @@ export function TroubleshootingSection({evidenceData}: TroubleshootingSectionPro
     </Stack>
   );
 }
+
+export function TroubleshootingSection({evidenceData}: TroubleshootingSectionProps) {
+  const isManualInstrumentation = evidenceData.spanOrigin === 'manual';
+
+  return (
+    <Stack gap="lg" padding="lg">
+      <Heading as="h3">{t('Troubleshooting')}</Heading>
+      {isManualInstrumentation ? (
+        <ManualInstrumentationFix evidenceData={evidenceData} />
+      ) : (
+        <AutomaticInstrumentationTroubleshooting evidenceData={evidenceData} />
+      )}
+    </Stack>
+  );
+}
diff --git a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/types.ts b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/types.ts
index 39e4f0a30616f5..bf777d677ed87e 100644
--- a/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/types.ts
+++ b/static/app/views/issueDetails/configurationIssues/lowValueSpanIssues/types.ts
@@ -2,11 +2,13 @@ import type {EventOccurrence} from 'sentry/types/event';
 
 export interface LowValueSpanEvidenceData {
   avgDurationMs: number | null;
-  count: number | null;
   description: string | null;
   estimatedCostUsd: number | null;
+  extrapolatedCount: number | null;
   op: string | null;
   sdkName: string | null;
+  spanCount: number | null;
+  spanOrigin: string | null;
 }
 
 function getStringValue(value: unknown): string | null {
@@ -21,11 +23,13 @@ export function getLowValueSpanEvidenceData(
   evidenceData: EventOccurrence['evidenceData'] | null | undefined
 ): LowValueSpanEvidenceData {
   return {
-    op: getStringValue(evidenceData?.op),
-    description: getStringValue(evidenceData?.description),
-    count: getNumberValue(evidenceData?.count),
+    op: getStringValue(evidenceData?.span_op),
+    description: getStringValue(evidenceData?.span_description),
+    spanCount: getNumberValue(evidenceData?.span_count),
+    extrapolatedCount: getNumberValue(evidenceData?.extrapolated_count),
     avgDurationMs: getNumberValue(evidenceData?.avg_duration_ms),
     estimatedCostUsd: getNumberValue(evidenceData?.estimated_cost_usd),
     sdkName: getStringValue(evidenceData?.sdk_name),
+    spanOrigin: getStringValue(evidenceData?.span_origin),
   };
 }
diff --git a/static/app/views/issueDetails/eventTitle.tsx b/static/app/views/issueDetails/eventTitle.tsx
index 456ba5281f3074..93213d3fda2467 100644
--- a/static/app/views/issueDetails/eventTitle.tsx
+++ b/static/app/views/issueDetails/eventTitle.tsx
@@ -98,7 +98,7 @@ export function EventTitle({event, group, ref, ...props}: EventNavigationProps)
           </EventIdWrapper>
           <StyledTimeSince
             tooltipBody={<EventCreatedTooltip event={event} />}
-            tooltipProps={{maxWidth: 300, isHoverable: true}}
+            maxWidth={300}
             date={event.dateCreated ?? event.dateReceived}
             css={grayText}
             aria-label={t('Event timestamp')}
diff --git a/static/app/views/performance/newTraceDetails/traceDrawer/details/span/eapSections/aiContentRenderer.tsx b/static/app/views/performance/newTraceDetails/traceDrawer/details/span/eapSections/aiContentRenderer.tsx
index b591d91e92760f..11837563ec16c8 100644
--- a/static/app/views/performance/newTraceDetails/traceDrawer/details/span/eapSections/aiContentRenderer.tsx
+++ b/static/app/views/performance/newTraceDetails/traceDrawer/details/span/eapSections/aiContentRenderer.tsx
@@ -15,6 +15,7 @@ import {TraceDrawerComponents} from 'sentry/views/performance/newTraceDetails/tr
 
 interface AIContentRendererProps {
   text: string;
+  autoCollapseLimit?: number;
   inline?: boolean;
   maxJsonDepth?: number;
 }
@@ -67,6 +68,7 @@ export function AIContentRenderer({
   text,
   inline = false,
   maxJsonDepth = 2,
+  autoCollapseLimit,
 }: AIContentRendererProps) {
   const detection = useMemo(() => detectAIContentType(text), [text]);
 
@@ -77,6 +79,7 @@ export function AIContentRenderer({
         <TraceDrawerComponents.MultilineJSON
           value={detection.parsedData}
           maxDefaultDepth={maxJsonDepth}
+          autoCollapseLimit={autoCollapseLimit}
         />
       );
 
@@ -86,6 +89,7 @@ export function AIContentRenderer({
           <TraceDrawerComponents.MultilineJSON
             value={detection.parsedData}
             maxDefaultDepth={maxJsonDepth}
+            autoCollapseLimit={autoCollapseLimit}
           />
           <Text size="xs" variant="muted">
             {t('Truncated')}
diff --git a/static/app/views/performance/newTraceDetails/traceDrawer/details/styles.tsx b/static/app/views/performance/newTraceDetails/traceDrawer/details/styles.tsx
index a11747d21f4395..6469e32d31874b 100644
--- a/static/app/views/performance/newTraceDetails/traceDrawer/details/styles.tsx
+++ b/static/app/views/performance/newTraceDetails/traceDrawer/details/styles.tsx
@@ -1316,8 +1316,10 @@ const MultilineTextWrapper = styled('div')`
 function MultilineJSON({
   value,
   maxDefaultDepth = 2,
+  autoCollapseLimit,
 }: {
   value: any;
+  autoCollapseLimit?: number;
   maxDefaultDepth?: number;
 }) {
   const [showRaw, setShowRaw] = useState(false);
@@ -1328,9 +1330,9 @@ function MultilineJSON({
 
   // Ensure root ('$') is always expanded, while children follow maxDefaultDepth rules
   const computedExpandedPaths = useMemo(() => {
-    const childPaths = getDefaultExpanded(maxDefaultDepth, json);
+    const childPaths = getDefaultExpanded(maxDefaultDepth, json, autoCollapseLimit);
     return Array.from(new Set(['$', ...childPaths]));
-  }, [maxDefaultDepth, json]);
+  }, [maxDefaultDepth, json, autoCollapseLimit]);
 
   return (
     <MultilineTextWrapperMonospace {...hoverProps}>
@@ -1367,6 +1369,7 @@ function MultilineJSON({
           }}
           value={json}
           maxDefaultDepth={maxDefaultDepth}
+          autoCollapseLimit={autoCollapseLimit}
           initialExpandedPaths={computedExpandedPaths}
           withAnnotatedText
         />
diff --git a/static/app/views/projectDetail/projectLatestAlerts.tsx b/static/app/views/projectDetail/projectLatestAlerts.tsx
index 68e3d4205cc82d..44cff24c2c9ea8 100644
--- a/static/app/views/projectDetail/projectLatestAlerts.tsx
+++ b/static/app/views/projectDetail/projectLatestAlerts.tsx
@@ -5,6 +5,7 @@ import pick from 'lodash/pick';
 
 import {AlertBadge} from '@sentry/scraps/badge';
 import {Link} from '@sentry/scraps/link';
+import {Text} from '@sentry/scraps/text';
 
 import {SectionHeading} from 'sentry/components/charts/styles';
 import {EmptyStateWarning} from 'sentry/components/emptyStateWarning';
@@ -41,7 +42,7 @@ function AlertRow({alert}: AlertRowProps) {
 
   const Icon = isResolved ? IconCheckmark : isWarning ? IconExclamation : IconFire;
 
-  const statusProps = {isResolved, isWarning};
+  const variant = isResolved ? 'success' : isWarning ? 'warning' : 'danger';
 
   return (
     <AlertRowLink
@@ -56,21 +57,16 @@ function AlertRow({alert}: AlertRowProps) {
       </AlertBadgeWrapper>
       <AlertDetails>
         <AlertTitle>{title}</AlertTitle>
-        <AlertDate {...statusProps}>
+        <Text variant={variant}>
           {isResolved ? t('Resolved') : t('Triggered')}{' '}
           {isResolved ? (
             dateClosed ? (
-              <TimeSince date={dateClosed} />
+              <TimeSince date={dateClosed} variant={variant} />
             ) : null
           ) : (
-            <TimeSince
-              date={dateStarted}
-              tooltipUnderlineColor={
-                isResolved ? 'success' : isWarning ? 'warning' : 'danger'
-              }
-            />
+            <TimeSince date={dateStarted} variant={variant} />
           )}
-        </AlertDate>
+        </Text>
       </AlertDetails>
     </AlertRowLink>
   );
@@ -218,11 +214,6 @@ const AlertRowLink = styled(Link)`
   }
 `;
 
-type StatusColorProps = {
-  isResolved: boolean;
-  isWarning: boolean;
-};
-
 const AlertBadgeWrapper = styled('div')<{icon: typeof IconExclamation}>`
   display: flex;
   align-items: center;
@@ -249,15 +240,6 @@ const AlertTitle = styled('div')`
   text-overflow: ellipsis;
 `;
 
-const AlertDate = styled('span')<StatusColorProps>`
-  color: ${p =>
-    p.isResolved
-      ? p.theme.tokens.content.success
-      : p.isWarning
-        ? p.theme.tokens.content.warning
-        : p.theme.tokens.content.danger};
-`;
-
 const StyledEmptyStateWarning = styled(EmptyStateWarning)`
   height: ${PLACEHOLDER_AND_EMPTY_HEIGHT};
   justify-content: center;
diff --git a/tests/sentry/dynamic_sampling/per_org/tasks/__init__.py b/tests/sentry/dynamic_sampling/per_org/tasks/__init__.py
deleted file mode 100644
index e69de29bb2d1d6..00000000000000
diff --git a/tests/sentry/dynamic_sampling/per_org/test_calculations.py b/tests/sentry/dynamic_sampling/per_org/test_calculations.py
new file mode 100644
index 00000000000000..dc50eb46e81d22
--- /dev/null
+++ b/tests/sentry/dynamic_sampling/per_org/test_calculations.py
@@ -0,0 +1,119 @@
+from __future__ import annotations
+
+from unittest.mock import Mock, patch
+
+import pytest
+
+from sentry.dynamic_sampling.models.common import RebalancedItem
+from sentry.dynamic_sampling.models.projects_rebalancing import ProjectsRebalancingInput
+from sentry.dynamic_sampling.per_org.calculations import (
+    compare_rebalanced_projects_with_cache,
+    get_cached_rebalanced_project_sample_rates,
+    is_within_relative_tolerance,
+    run_project_balancing,
+)
+from sentry.dynamic_sampling.per_org.queries import ProjectVolume
+from sentry.dynamic_sampling.rules.utils import get_redis_client_for_ds
+from sentry.dynamic_sampling.tasks.helpers.boost_low_volume_projects import (
+    generate_boost_low_volume_projects_cache_key,
+)
+from sentry.testutils.cases import TestCase
+
+
+def _project_volume(project_id: int, total: int = 100, keep: int = 25) -> ProjectVolume:
+    return ProjectVolume(project_id=project_id, total=total, keep=keep, drop=max(total - keep, 0))
+
+
+class ProjectBalancingCalculationsTest(TestCase):
+    def setUp(self) -> None:
+        super().setUp()
+        self.redis = get_redis_client_for_ds()
+
+    def test_run_project_balancing_returns_rebalanced_projects(self) -> None:
+        org = self.create_organization()
+        project_with_volume = self.create_project(organization=org)
+        project_without_volume = self.create_project(organization=org)
+        config = Mock()
+        config.organization = org
+        config.projects = [project_with_volume, project_without_volume]
+        config.get_sample_rate.return_value = 0.5
+        rebalanced_projects = [
+            RebalancedItem(id=project_with_volume.id, count=100, new_sample_rate=0.25),
+            RebalancedItem(id=project_without_volume.id, count=0, new_sample_rate=1.0),
+        ]
+
+        with patch(
+            "sentry.dynamic_sampling.per_org.calculations.ProjectsRebalancingModel.run",
+            return_value=rebalanced_projects,
+        ) as model_run:
+            result = run_project_balancing(config, [_project_volume(project_with_volume.id)])
+
+        model_run.assert_called_once()
+        model_input = model_run.call_args.args[-1]
+        assert isinstance(model_input, ProjectsRebalancingInput)
+        assert model_input.sample_rate == 0.5
+        assert model_input.classes == [
+            RebalancedItem(id=project_with_volume.id, count=100),
+            RebalancedItem(id=project_without_volume.id, count=0),
+        ]
+        assert result == rebalanced_projects
+
+    def test_compare_rebalanced_projects_with_cache_logs_per_project(self) -> None:
+        org = self.create_organization()
+        project_with_volume = self.create_project(organization=org)
+        project_without_volume = self.create_project(organization=org)
+        config = Mock()
+        config.organization = org
+        rebalanced_projects = [
+            RebalancedItem(id=project_with_volume.id, count=100, new_sample_rate=0.25),
+            RebalancedItem(id=project_without_volume.id, count=0, new_sample_rate=1.0),
+        ]
+        cached_sample_rates: dict[int, float | None] = {
+            project_with_volume.id: 0.2,
+            project_without_volume.id: 0.96,
+        }
+
+        with patch("sentry.dynamic_sampling.per_org.calculations.logger.info") as logger_info:
+            compare_rebalanced_projects_with_cache(config, rebalanced_projects, cached_sample_rates)
+
+        assert [call.args for call in logger_info.call_args_list] == [
+            ("dynamic_sampling.per_org.project_balancing_comparison",),
+            ("dynamic_sampling.per_org.project_balancing_comparison",),
+        ]
+        assert [call.kwargs["extra"] for call in logger_info.call_args_list] == [
+            {
+                "org_id": org.id,
+                "project_id": project_with_volume.id,
+                "generic_metrics_sample_rate": 0.2,
+                "eap_sample_rate": 0.25,
+                "relative_deviation": pytest.approx(0.2),
+                "is_equal": False,
+            },
+            {
+                "org_id": org.id,
+                "project_id": project_without_volume.id,
+                "generic_metrics_sample_rate": 0.96,
+                "eap_sample_rate": 1.0,
+                "relative_deviation": pytest.approx(0.04),
+                "is_equal": True,
+            },
+        ]
+
+    def test_project_balancing_relative_tolerance(self) -> None:
+        assert is_within_relative_tolerance(0.95, 1.0)
+        assert is_within_relative_tolerance(1.05, 1.0)
+        assert not is_within_relative_tolerance(0.94, 1.0)
+        assert not is_within_relative_tolerance(1.06, 1.0)
+        assert is_within_relative_tolerance(0.0, 0.0)
+        assert not is_within_relative_tolerance(0.01, 0.0)
+        assert not is_within_relative_tolerance(None, 1.0)
+
+    def test_get_cached_rebalanced_project_sample_rates(self) -> None:
+        org = self.create_organization()
+        project = self.create_project(organization=org)
+        cache_key = generate_boost_low_volume_projects_cache_key(org.id)
+        self.redis.delete(cache_key)
+        self.addCleanup(self.redis.delete, cache_key)
+        self.redis.hset(cache_key, str(project.id), "0.25")
+
+        assert get_cached_rebalanced_project_sample_rates(org.id) == {project.id: 0.25}
diff --git a/tests/sentry/dynamic_sampling/per_org/tasks/test_configuration.py b/tests/sentry/dynamic_sampling/per_org/test_configuration.py
similarity index 89%
rename from tests/sentry/dynamic_sampling/per_org/tasks/test_configuration.py
rename to tests/sentry/dynamic_sampling/per_org/test_configuration.py
index 8fda8542883540..e4761489133df9 100644
--- a/tests/sentry/dynamic_sampling/per_org/tasks/test_configuration.py
+++ b/tests/sentry/dynamic_sampling/per_org/test_configuration.py
@@ -8,14 +8,14 @@
 import pytest
 from django.core.exceptions import ObjectDoesNotExist
 
-from sentry.dynamic_sampling.per_org.tasks.configuration import (
+from sentry.dynamic_sampling.per_org.configuration import (
     AutomaticDynamicSamplingConfiguration,
     CustomDynamicSamplingOrganizationConfiguration,
     CustomDynamicSamplingProjectConfiguration,
     NoDynamicSamplingConfiguration,
     get_configuration,
 )
-from sentry.dynamic_sampling.per_org.tasks.telemetry import (
+from sentry.dynamic_sampling.per_org.telemetry import (
     DynamicSamplingException,
     DynamicSamplingStatus,
 )
@@ -55,7 +55,7 @@ def test_subscription_backed_org_uses_blended_sample_rate(self) -> None:
         org = self.create_organization()
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+            "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
             return_value=0.5,
         ) as get_blended_sample_rate:
             configuration = get_configuration(org.id)
@@ -78,15 +78,15 @@ def test_subscription_backed_org_uses_eap_sliding_window_sample_rate(self) -> No
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=0.5,
             ),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.configuration.get_eap_organization_volume",
                 return_value=sliding_window_volume,
             ) as get_volume,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.compute_sliding_window_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.compute_sliding_window_sample_rate",
                 return_value=0.25,
             ) as compute_sample_rate,
         ):
@@ -113,15 +113,15 @@ def test_subscription_backed_org_falls_back_to_blended_sample_rate_without_volum
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=0.5,
             ),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.configuration.get_eap_organization_volume",
                 return_value=None,
             ),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.compute_sliding_window_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.compute_sliding_window_sample_rate",
             ) as compute_sample_rate,
         ):
             configuration = get_configuration(org.id)
@@ -136,11 +136,11 @@ def test_subscription_backed_org_without_sample_rate_is_disabled(self) -> None:
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=None,
             ),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.configuration.get_eap_organization_volume",
             ) as get_volume,
         ):
             configuration = get_configuration(org.id)
@@ -158,7 +158,7 @@ def test_subscription_backed_org_without_subscription_bubbles_terminal_status(se
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 side_effect=ObjectDoesNotExist,
             ),
             pytest.raises(DynamicSamplingException) as exc_info,
@@ -172,7 +172,7 @@ def test_am2_ignores_project_mode_option(self) -> None:
         org.update_option("sentry:sampling_mode", DynamicSamplingMode.PROJECT)
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+            "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
             return_value=0.5,
         ):
             configuration = get_configuration(org.id)
@@ -197,7 +197,7 @@ def test_org_mode_custom_dynamic_sampling_uses_org_target_sample_rate(self) -> N
                         }
                     ),
                     patch(
-                        "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate"
+                        "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate"
                     ) as get_blended_sample_rate,
                 ):
                     configuration = get_configuration(org.id)
@@ -235,7 +235,7 @@ def test_project_mode_custom_dynamic_sampling_stores_project_sample_rates(self)
                         }
                     ),
                     patch(
-                        "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate"
+                        "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate"
                     ) as get_blended_sample_rate,
                 ):
                     configuration = get_configuration(org.id)
@@ -322,7 +322,7 @@ def test_subscription_backed_org_uses_measure_options(self) -> None:
                         }
                     ),
                     patch(
-                        "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                        "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                         return_value=1.0,
                     ),
                 ):
diff --git a/tests/sentry/dynamic_sampling/per_org/tasks/test_queries.py b/tests/sentry/dynamic_sampling/per_org/test_queries.py
similarity index 94%
rename from tests/sentry/dynamic_sampling/per_org/tasks/test_queries.py
rename to tests/sentry/dynamic_sampling/per_org/test_queries.py
index 7a0770184c6c16..a0ceda59bf29e8 100644
--- a/tests/sentry/dynamic_sampling/per_org/tasks/test_queries.py
+++ b/tests/sentry/dynamic_sampling/per_org/test_queries.py
@@ -5,11 +5,11 @@
 
 from sentry_protos.snuba.v1.trace_item_attribute_pb2 import ExtrapolationMode
 
-from sentry.dynamic_sampling.per_org.tasks.configuration import (
+from sentry.dynamic_sampling.per_org.configuration import (
     BaseDynamicSamplingConfiguration,
     get_configuration,
 )
-from sentry.dynamic_sampling.per_org.tasks.queries import (
+from sentry.dynamic_sampling.per_org.queries import (
     DynamicSamplingQueryFields,
     DynamicSamplingQueryFilters,
     ProjectVolume,
@@ -86,11 +86,11 @@ def get_config(
     ) -> BaseDynamicSamplingConfiguration:
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=1.0,
             ),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.configuration.get_eap_organization_volume",
                 return_value=None,
             ),
         ):
@@ -101,7 +101,7 @@ def test_get_eap_organization_volume_existing_org(self) -> None:
         project = self.create_project(organization=organization)
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.queries.Spans.run_table_query",
+            "sentry.dynamic_sampling.per_org.queries.Spans.run_table_query",
             return_value={"data": [{DynamicSamplingQueryFields.COUNT: 2, "count_sample()": 2}]},
         ) as run_table_query:
             org_volume = get_eap_organization_volume(
@@ -129,7 +129,7 @@ def test_get_eap_organization_volume_returns_raw_and_extrapolated_counts(self) -
         self.create_project(organization=organization)
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.queries.Spans.run_table_query",
+            "sentry.dynamic_sampling.per_org.queries.Spans.run_table_query",
             return_value={"data": [{"count()": 10, DynamicSamplingQueryFields.COUNT_SAMPLE: 1}]},
         ):
             org_volume = get_eap_organization_volume(
@@ -152,7 +152,7 @@ def test_get_eap_organization_volume_without_projects(self) -> None:
         organization = self.create_organization()
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.queries.Spans.run_table_query",
+            "sentry.dynamic_sampling.per_org.queries.Spans.run_table_query",
             return_value={"data": []},
         ) as run_table_query:
             org_volume = get_eap_organization_volume(
@@ -171,7 +171,7 @@ def test_get_eap_project_volumes_existing_org(self) -> None:
         self.create_project(organization=other_organization)
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.queries.run_eap_spans_table_query_in_chunks",
+            "sentry.dynamic_sampling.per_org.queries.run_eap_spans_table_query_in_chunks",
             return_value=[
                 {
                     "sentry.dsc.project_id": project.id,
@@ -213,7 +213,7 @@ def test_get_eap_project_volumes_without_traffic(self) -> None:
         self.create_project(organization=organization)
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.queries.run_eap_spans_table_query_in_chunks",
+            "sentry.dynamic_sampling.per_org.queries.run_eap_spans_table_query_in_chunks",
             return_value=[],
         ):
             project_volumes = get_eap_project_volumes(
@@ -227,7 +227,7 @@ def test_get_eap_project_volumes_handles_missing_aggregate_values(self) -> None:
         project = self.create_project(organization=organization)
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.queries.run_eap_spans_table_query_in_chunks",
+            "sentry.dynamic_sampling.per_org.queries.run_eap_spans_table_query_in_chunks",
             return_value=[
                 {
                     "sentry.dsc.project_id": project.id,
@@ -243,7 +243,7 @@ def test_get_eap_project_volumes_skips_rows_without_dsc_project_id(self) -> None
         project = self.create_project(organization=organization)
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.queries.run_eap_spans_table_query_in_chunks",
+            "sentry.dynamic_sampling.per_org.queries.run_eap_spans_table_query_in_chunks",
             return_value=[
                 {
                     "sentry.dsc.project_id": None,
@@ -265,7 +265,7 @@ def test_get_eap_project_volumes_without_projects(self) -> None:
         organization = self.create_organization()
 
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.queries.run_eap_spans_table_query_in_chunks",
+            "sentry.dynamic_sampling.per_org.queries.run_eap_spans_table_query_in_chunks",
             return_value=[],
         ) as run_table_query:
             project_volumes = get_eap_project_volumes(
@@ -281,7 +281,7 @@ def test_get_eap_project_volumes_without_projects(self) -> None:
 class EAPTransactionVolumesTest(TestCase, SnubaTestCase, SpanTestCase):
     def get_config(self, organization: Organization) -> BaseDynamicSamplingConfiguration:
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+            "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
             return_value=1.0,
         ):
             return get_configuration(organization.id)
diff --git a/tests/sentry/dynamic_sampling/per_org/tasks/test_scheduler.py b/tests/sentry/dynamic_sampling/per_org/test_scheduler.py
similarity index 63%
rename from tests/sentry/dynamic_sampling/per_org/tasks/test_scheduler.py
rename to tests/sentry/dynamic_sampling/per_org/test_scheduler.py
index 7386e37083ae04..83250f6e68911e 100644
--- a/tests/sentry/dynamic_sampling/per_org/tasks/test_scheduler.py
+++ b/tests/sentry/dynamic_sampling/per_org/test_scheduler.py
@@ -4,15 +4,17 @@
 
 from django.core.exceptions import ObjectDoesNotExist
 
-from sentry.dynamic_sampling.per_org.tasks.configuration import BaseDynamicSamplingConfiguration
-from sentry.dynamic_sampling.per_org.tasks.scheduler import (
+from sentry.dynamic_sampling.models.common import RebalancedItem
+from sentry.dynamic_sampling.per_org.configuration import BaseDynamicSamplingConfiguration
+from sentry.dynamic_sampling.per_org.queries import ProjectVolume
+from sentry.dynamic_sampling.per_org.scheduler import (
     BUCKET_COUNT,
     BUCKET_CURSOR_KEY,
     _next_bucket_index,
     run_calculations_per_org_task,
     schedule_per_org_calculations,
 )
-from sentry.dynamic_sampling.per_org.tasks.telemetry import DynamicSamplingStatus
+from sentry.dynamic_sampling.per_org.telemetry import DynamicSamplingStatus
 from sentry.dynamic_sampling.rules.utils import get_redis_client_for_ds
 from sentry.dynamic_sampling.tasks.common import OrganizationDataVolume
 from sentry.dynamic_sampling.types import DynamicSamplingMode
@@ -33,6 +35,10 @@ def _assert_called_once_with_config(
     return config
 
 
+def _project_volume(project_id: int, total: int = 100, keep: int = 25) -> ProjectVolume:
+    return ProjectVolume(project_id=project_id, total=total, keep=keep, drop=max(total - keep, 0))
+
+
 def _drain_dispatched_org_ids(burst) -> list[int]:
     ids = [args[0] for _task, args, _kwargs in burst.queue]
     burst.queue.clear()
@@ -135,15 +141,15 @@ def test_run_calculations_per_org_returns_no_volume_without_traffic(self) -> Non
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=1.0,
             ) as get_blended_sample_rate,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume",
                 return_value=None,
             ) as get_volume,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_project_volumes"
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_project_volumes"
             ) as get_project_volumes,
         ):
             result = run_calculations_per_org_task(org.id)
@@ -156,28 +162,42 @@ def test_run_calculations_per_org_returns_no_volume_without_traffic(self) -> Non
     @override_options({"dynamic-sampling.per_org.rollout-rate": 1.0})
     def test_run_calculations_per_org_continues_with_traffic(self) -> None:
         org = self.create_organization()
-        self.create_project(organization=org)
+        project = self.create_project(organization=org)
         org_volume = OrganizationDataVolume(org_id=org.id, total=100, indexed=25)
+        project_volumes = [_project_volume(project.id)]
+        rebalanced_projects = [RebalancedItem(id=project.id, count=100, new_sample_rate=1.0)]
+        cached_sample_rates: dict[int, float | None] = {}
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=1.0,
             ) as get_blended_sample_rate,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume",
                 return_value=org_volume,
             ) as get_volume,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_project_volumes",
-                return_value=[(1, 100, 25, 75)],
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_project_volumes",
+                return_value=project_volumes,
             ) as get_project_volumes,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_transaction_volumes",
+                "sentry.dynamic_sampling.per_org.scheduler.run_project_balancing",
+                return_value=rebalanced_projects,
+            ) as project_balancing,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.get_cached_rebalanced_project_sample_rates",
+                return_value=cached_sample_rates,
+            ) as get_cached_sample_rates,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.compare_rebalanced_projects_with_cache"
+            ) as compare_rebalanced_projects,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_transaction_volumes",
                 return_value=[
                     {
                         "org_id": org.id,
-                        "project_id": 1,
+                        "project_id": project.id,
                         "transaction_counts": [("checkout", 1.0)],
                         "total_num_transactions": 1.0,
                         "total_num_classes": 1,
@@ -190,7 +210,12 @@ def test_run_calculations_per_org_continues_with_traffic(self) -> None:
         assert result is None
         _assert_called_once_with_config(get_volume, org.id)
         get_blended_sample_rate.assert_called_once_with(organization_id=org.id)
-        _assert_called_once_with_config(get_project_volumes, org.id)
+        project_config = _assert_called_once_with_config(get_project_volumes, org.id)
+        project_balancing.assert_called_once_with(project_config, project_volumes)
+        get_cached_sample_rates.assert_called_once_with(org.id)
+        compare_rebalanced_projects.assert_called_once_with(
+            project_config, rebalanced_projects, cached_sample_rates
+        )
         _assert_called_once_with_config(get_transaction_volumes, org.id)
 
     @override_options({"dynamic-sampling.per_org.rollout-rate": 1.0})
@@ -201,20 +226,24 @@ def test_run_calculations_per_org_returns_no_volume_without_project_volumes(self
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=1.0,
             ) as get_blended_sample_rate,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume",
                 return_value=org_volume,
             ) as get_volume,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_project_volumes",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_project_volumes",
                 return_value=[],
             ) as get_project_volumes,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_transaction_volumes"
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_transaction_volumes"
             ) as get_transaction_volumes,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.run_project_balancing",
+                return_value=None,
+            ) as project_balancing,
         ):
             result = run_calculations_per_org_task(org.id)
 
@@ -222,29 +251,44 @@ def test_run_calculations_per_org_returns_no_volume_without_project_volumes(self
         get_blended_sample_rate.assert_called_once_with(organization_id=org.id)
         _assert_called_once_with_config(get_volume, org.id)
         _assert_called_once_with_config(get_project_volumes, org.id)
+        project_balancing.assert_not_called()
         get_transaction_volumes.assert_not_called()
 
     @override_options({"dynamic-sampling.per_org.rollout-rate": 1.0})
     def test_run_calculations_per_org_returns_no_volume_without_transaction_volumes(self) -> None:
         org = self.create_organization()
-        self.create_project(organization=org)
+        project = self.create_project(organization=org)
         org_volume = OrganizationDataVolume(org_id=org.id, total=100, indexed=25)
+        project_volumes = [_project_volume(project.id)]
+        rebalanced_projects = [RebalancedItem(id=project.id, count=100, new_sample_rate=1.0)]
+        cached_sample_rates: dict[int, float | None] = {}
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=1.0,
             ) as get_blended_sample_rate,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume",
                 return_value=org_volume,
             ) as get_volume,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_project_volumes",
-                return_value=[(1, 100, 25, 75)],
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_project_volumes",
+                return_value=project_volumes,
             ) as get_project_volumes,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_transaction_volumes",
+                "sentry.dynamic_sampling.per_org.scheduler.run_project_balancing",
+                return_value=rebalanced_projects,
+            ) as project_balancing,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.get_cached_rebalanced_project_sample_rates",
+                return_value=cached_sample_rates,
+            ) as get_cached_sample_rates,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.compare_rebalanced_projects_with_cache"
+            ) as compare_rebalanced_projects,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_transaction_volumes",
                 return_value=[],
             ) as get_transaction_volumes,
         ):
@@ -253,7 +297,12 @@ def test_run_calculations_per_org_returns_no_volume_without_transaction_volumes(
         assert result == DynamicSamplingStatus.NO_TRANSACTION_VOLUMES
         get_blended_sample_rate.assert_called_once_with(organization_id=org.id)
         _assert_called_once_with_config(get_volume, org.id)
-        _assert_called_once_with_config(get_project_volumes, org.id)
+        project_config = _assert_called_once_with_config(get_project_volumes, org.id)
+        project_balancing.assert_called_once_with(project_config, project_volumes)
+        get_cached_sample_rates.assert_called_once_with(org.id)
+        compare_rebalanced_projects.assert_called_once_with(
+            project_config, rebalanced_projects, cached_sample_rates
+        )
         _assert_called_once_with_config(get_transaction_volumes, org.id)
 
     @override_options({"dynamic-sampling.per_org.rollout-rate": 1.0})
@@ -267,17 +316,17 @@ def test_run_calculations_per_org_skips_project_balancing_for_project_mode(self)
         with (
             self.feature("organizations:dynamic-sampling-custom"),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate"
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate"
             ) as get_blended_sample_rate,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume",
                 return_value=org_volume,
             ) as get_volume,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_project_volumes",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_project_volumes",
             ) as get_project_volumes,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_transaction_volumes",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_transaction_volumes",
                 return_value=[
                     {
                         "org_id": org.id,
@@ -300,29 +349,43 @@ def test_run_calculations_per_org_skips_project_balancing_for_project_mode(self)
     @override_options({"dynamic-sampling.per_org.rollout-rate": 1.0})
     def test_run_calculations_per_org_queries_projects_for_am3_org_mode(self) -> None:
         org = self.create_organization()
-        self.create_project(organization=org)
+        project = self.create_project(organization=org)
         org.update_option("sentry:sampling_mode", DynamicSamplingMode.ORGANIZATION)
         org_volume = OrganizationDataVolume(org_id=org.id, total=100, indexed=25)
+        project_volumes = [_project_volume(project.id)]
+        rebalanced_projects = [RebalancedItem(id=project.id, count=100, new_sample_rate=1.0)]
+        cached_sample_rates: dict[int, float | None] = {}
 
         with (
             self.feature("organizations:dynamic-sampling-custom"),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate"
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate"
             ) as get_blended_sample_rate,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume",
                 return_value=org_volume,
             ) as get_volume,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_project_volumes",
-                return_value=[(1, 100, 25, 75)],
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_project_volumes",
+                return_value=project_volumes,
             ) as get_project_volumes,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_transaction_volumes",
+                "sentry.dynamic_sampling.per_org.scheduler.run_project_balancing",
+                return_value=rebalanced_projects,
+            ) as project_balancing,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.get_cached_rebalanced_project_sample_rates",
+                return_value=cached_sample_rates,
+            ) as get_cached_sample_rates,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.compare_rebalanced_projects_with_cache"
+            ) as compare_rebalanced_projects,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_transaction_volumes",
                 return_value=[
                     {
                         "org_id": org.id,
-                        "project_id": 1,
+                        "project_id": project.id,
                         "transaction_counts": [("checkout", 1.0)],
                         "total_num_transactions": 1.0,
                         "total_num_classes": 1,
@@ -335,7 +398,12 @@ def test_run_calculations_per_org_queries_projects_for_am3_org_mode(self) -> Non
         assert result is None
         get_blended_sample_rate.assert_not_called()
         _assert_called_once_with_config(get_volume, org.id)
-        _assert_called_once_with_config(get_project_volumes, org.id)
+        project_config = _assert_called_once_with_config(get_project_volumes, org.id)
+        project_balancing.assert_called_once_with(project_config, project_volumes)
+        get_cached_sample_rates.assert_called_once_with(org.id)
+        compare_rebalanced_projects.assert_called_once_with(
+            project_config, rebalanced_projects, cached_sample_rates
+        )
         _assert_called_once_with_config(get_transaction_volumes, org.id)
 
     @override_options({"dynamic-sampling.per_org.rollout-rate": 1.0})
@@ -347,13 +415,13 @@ def test_run_calculations_per_org_skips_project_mode_without_project_rates(self)
         with (
             self.feature("organizations:dynamic-sampling-custom"),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate"
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate"
             ) as get_blended_sample_rate,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume"
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume"
             ) as get_volume,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_project_volumes"
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_project_volumes"
             ) as get_project_volumes,
         ):
             result = run_calculations_per_org_task(org.id)
@@ -366,28 +434,42 @@ def test_run_calculations_per_org_skips_project_mode_without_project_rates(self)
     @override_options({"dynamic-sampling.per_org.rollout-rate": 1.0})
     def test_run_calculations_per_org_queries_projects_for_am2(self) -> None:
         org = self.create_organization()
-        self.create_project(organization=org)
+        project = self.create_project(organization=org)
         org_volume = OrganizationDataVolume(org_id=org.id, total=100, indexed=25)
+        project_volumes = [_project_volume(project.id)]
+        rebalanced_projects = [RebalancedItem(id=project.id, count=100, new_sample_rate=1.0)]
+        cached_sample_rates: dict[int, float | None] = {}
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=1.0,
             ) as get_blended_sample_rate,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume",
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume",
                 return_value=org_volume,
             ) as get_volume,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_project_volumes",
-                return_value=[(1, 100, 25, 75)],
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_project_volumes",
+                return_value=project_volumes,
             ) as get_project_volumes,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_transaction_volumes",
+                "sentry.dynamic_sampling.per_org.scheduler.run_project_balancing",
+                return_value=rebalanced_projects,
+            ) as project_balancing,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.get_cached_rebalanced_project_sample_rates",
+                return_value=cached_sample_rates,
+            ) as get_cached_sample_rates,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.compare_rebalanced_projects_with_cache"
+            ) as compare_rebalanced_projects,
+            patch(
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_transaction_volumes",
                 return_value=[
                     {
                         "org_id": org.id,
-                        "project_id": 1,
+                        "project_id": project.id,
                         "transaction_counts": [("checkout", 1.0)],
                         "total_num_transactions": 1.0,
                         "total_num_classes": 1,
@@ -400,7 +482,12 @@ def test_run_calculations_per_org_queries_projects_for_am2(self) -> None:
         assert result is None
         get_blended_sample_rate.assert_called_once_with(organization_id=org.id)
         _assert_called_once_with_config(get_volume, org.id)
-        _assert_called_once_with_config(get_project_volumes, org.id)
+        project_config = _assert_called_once_with_config(get_project_volumes, org.id)
+        project_balancing.assert_called_once_with(project_config, project_volumes)
+        get_cached_sample_rates.assert_called_once_with(org.id)
+        compare_rebalanced_projects.assert_called_once_with(
+            project_config, rebalanced_projects, cached_sample_rates
+        )
         _assert_called_once_with_config(get_transaction_volumes, org.id)
 
     @override_options({"dynamic-sampling.per_org.rollout-rate": 1.0})
@@ -409,11 +496,11 @@ def test_run_calculations_per_org_skips_org_without_transaction_sample_rate(self
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=None,
             ) as get_blended_sample_rate,
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume"
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume"
             ) as get_volume,
         ):
             result = run_calculations_per_org_task(org.id)
@@ -428,11 +515,11 @@ def test_run_calculations_per_org_skips_org_without_projects(self) -> None:
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 return_value=1.0,
             ),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume"
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume"
             ) as get_volume,
         ):
             result = run_calculations_per_org_task(org.id)
@@ -446,11 +533,11 @@ def test_run_calculations_per_org_skips_org_without_subscription(self) -> None:
 
         with (
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.configuration.quotas.backend.get_blended_sample_rate",
+                "sentry.dynamic_sampling.per_org.configuration.quotas.backend.get_blended_sample_rate",
                 side_effect=ObjectDoesNotExist,
             ),
             patch(
-                "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume"
+                "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume"
             ) as get_volume,
         ):
             result = run_calculations_per_org_task(org.id)
@@ -461,7 +548,7 @@ def test_run_calculations_per_org_skips_org_without_subscription(self) -> None:
     @override_options({"dynamic-sampling.per_org.rollout-rate": 1.0})
     def test_run_calculations_per_org_skips_missing_org(self) -> None:
         with patch(
-            "sentry.dynamic_sampling.per_org.tasks.scheduler.get_eap_organization_volume"
+            "sentry.dynamic_sampling.per_org.scheduler.get_eap_organization_volume"
         ) as get_volume:
             result = run_calculations_per_org_task(99999999)
 
diff --git a/tests/sentry/dynamic_sampling/per_org/tasks/test_telemetry.py b/tests/sentry/dynamic_sampling/per_org/test_telemetry.py
similarity index 74%
rename from tests/sentry/dynamic_sampling/per_org/tasks/test_telemetry.py
rename to tests/sentry/dynamic_sampling/per_org/test_telemetry.py
index 0bb66c451fa7c3..9d42fb883f9ccc 100644
--- a/tests/sentry/dynamic_sampling/per_org/tasks/test_telemetry.py
+++ b/tests/sentry/dynamic_sampling/per_org/test_telemetry.py
@@ -5,7 +5,7 @@
 
 import pytest
 
-from sentry.dynamic_sampling.per_org.tasks.telemetry import (
+from sentry.dynamic_sampling.per_org.telemetry import (
     DynamicSamplingException,
     DynamicSamplingStatus,
     track_dynamic_sampling,
@@ -41,9 +41,9 @@ def boom() -> None:
     timer, timer_tags = _capture_timer_tags()
 
     with (
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.metrics") as mock_metrics,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.emit_status") as emit,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.sentry_sdk") as sdk,
+        patch("sentry.dynamic_sampling.per_org.telemetry.metrics") as mock_metrics,
+        patch("sentry.dynamic_sampling.per_org.telemetry.emit_status") as emit,
+        patch("sentry.dynamic_sampling.per_org.telemetry.sentry_sdk") as sdk,
         pytest.raises(ValueError),
     ):
         mock_metrics.timer.side_effect = timer
@@ -65,9 +65,9 @@ def boom() -> None:
     timer, timer_tags = _capture_timer_tags()
 
     with (
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.metrics") as mock_metrics,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.emit_status") as emit,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.sentry_sdk") as sdk,
+        patch("sentry.dynamic_sampling.per_org.telemetry.metrics") as mock_metrics,
+        patch("sentry.dynamic_sampling.per_org.telemetry.emit_status") as emit,
+        patch("sentry.dynamic_sampling.per_org.telemetry.sentry_sdk") as sdk,
         pytest.raises(SnubaRPCTimeout),
     ):
         mock_metrics.timer.side_effect = timer
@@ -91,9 +91,9 @@ def boom() -> None:
     timer, timer_tags = _capture_timer_tags()
 
     with (
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.metrics") as mock_metrics,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.emit_status") as emit,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.sentry_sdk") as sdk,
+        patch("sentry.dynamic_sampling.per_org.telemetry.metrics") as mock_metrics,
+        patch("sentry.dynamic_sampling.per_org.telemetry.emit_status") as emit,
+        patch("sentry.dynamic_sampling.per_org.telemetry.sentry_sdk") as sdk,
         pytest.raises(SnubaRPCError),
     ):
         mock_metrics.timer.side_effect = timer
@@ -112,9 +112,9 @@ def add(x: int, y: int) -> int:
 
     timer, timer_tags = _capture_timer_tags()
     with (
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.metrics") as mock_metrics,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.emit_status") as emit,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.sentry_sdk") as sdk,
+        patch("sentry.dynamic_sampling.per_org.telemetry.metrics") as mock_metrics,
+        patch("sentry.dynamic_sampling.per_org.telemetry.emit_status") as emit,
+        patch("sentry.dynamic_sampling.per_org.telemetry.sentry_sdk") as sdk,
     ):
         mock_metrics.timer.side_effect = timer
         assert add(2, 3) == 5
@@ -133,9 +133,9 @@ def skipped() -> DynamicSamplingStatus:
 
     timer, timer_tags = _capture_timer_tags()
     with (
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.metrics") as mock_metrics,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.emit_status") as emit,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.sentry_sdk") as sdk,
+        patch("sentry.dynamic_sampling.per_org.telemetry.metrics") as mock_metrics,
+        patch("sentry.dynamic_sampling.per_org.telemetry.emit_status") as emit,
+        patch("sentry.dynamic_sampling.per_org.telemetry.sentry_sdk") as sdk,
     ):
         mock_metrics.timer.side_effect = timer
         assert skipped() == DynamicSamplingStatus.NOT_IN_ROLLOUT
@@ -156,9 +156,9 @@ def skipped() -> None:
 
     timer, timer_tags = _capture_timer_tags()
     with (
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.metrics") as mock_metrics,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.emit_status") as emit,
-        patch("sentry.dynamic_sampling.per_org.tasks.telemetry.sentry_sdk") as sdk,
+        patch("sentry.dynamic_sampling.per_org.telemetry.metrics") as mock_metrics,
+        patch("sentry.dynamic_sampling.per_org.telemetry.emit_status") as emit,
+        patch("sentry.dynamic_sampling.per_org.telemetry.sentry_sdk") as sdk,
     ):
         mock_metrics.timer.side_effect = timer
         assert skipped() == DynamicSamplingStatus.NO_SUBSCRIPTION