fix(sync): enforce V2 fair-use policy parity (#10020)

## Summary

- enforce the existing fresh-upload daily audio ceiling before Sync V2
can create staged or durable work
- apply the stored subscription tier to fresh Sync V2 soft-cap
evaluation, matching V1
- preserve durable processing when the optional subscription read fails
by falling back to the default tier with shared fallback telemetry
- keep the durable coordinator within the checked Pyright complexity
boundary by isolating the policy-only plan lookup

Closes #10002

## Product invariants

none

## Failure class

Failure-Class: none

This is a narrowly scoped V2 parity repair; no registered semantic
failure class applies.

## Verification

- `cd backend && .venv/bin/python -m pytest -q
tests/unit/test_sync_v2.py tests/unit/test_sync_cloud_tasks.py
tests/unit/test_fair_use_plan_aware.py` (261 passed)
- `cd backend && BACKEND_PYTEST_WORKERS=4 bash test.sh`
- `npm run test:sync-cloud-tasks-stack:emulator`
- `backend/.venv/bin/python backend/scripts/scan_async_blockers.py
--dirs backend/routers backend/utils`
- `cd backend && bash scripts/typecheck.sh` (0 errors)
- `.github/scripts/check_product_file_line_count_ratchet.py` against
`origin/main`


<!-- This is an auto-generated description by cubic. -->
<a
href="https://cubic.dev/pr/BasedHardware/omi/pull/10020?utm_source=github"
target="_blank" rel="noopener noreferrer"
data-no-image-dialog="true"><picture><source
media="(prefers-color-scheme: dark)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"><source
media="(prefers-color-scheme: light)"
srcset="https://www.cubic.dev/buttons/review-in-cubic-light.svg"><img
alt="Review in cubic"
src="https://www.cubic.dev/buttons/review-in-cubic-dark.svg"></picture></a>
<!-- End of auto-generated description by cubic. -->

Author: Git-on-my-level

.github/scripts/product_file_line_count_ratchet_baseline/backend-routers.json

@@ -4,13 +4,13 @@
     "backend/routers/chat.py": 1577,
     "backend/routers/developer.py": 2195,
     "backend/routers/mcp_sse.py": 1858,
-    "backend/routers/sync.py": 2024,
+    "backend/routers/sync.py": 2035,
     "backend/routers/users.py": 2046
   },
   "raise_justifications": {
     "backend/routers/developer.py": "GET /v1/dev/user/memories serves the authoritative legacy memories collection for legacy-cohort accounts with no rollout state (#9892), keeping the narrow deny/fallback decision in the route that owns the read contract.",
     "backend/routers/chat.py": "PTT stereo rejection keeps the serving STT provider boundary explicit in the established chat admission owner.",
-    "backend/routers/sync.py": "Fresh Sync admission and final Cloud Tasks ledger recovery retain their coupled task, lock, ledger, and staged-blob lifecycle in the existing ingestion owner rather than a new module.",
+    "backend/routers/sync.py": "Fresh Sync's daily ceiling remains beside the existing fresh-admission gates, before staged audio or durable worker work is created.",
     "backend/routers/users.py": "GET /v1/users/subscription serializes the new mobile plus/max plans as `unlimited` for clients whose plan enum predates them, so day-one buyers read as paid instead of Free (mirrors the existing operator remap); real limits/grandfather are computed from the true plan first.",
     "backend/routers/mcp_sse.py": "MCP SSE memory reads route their legacy-fallback decision through the shared mcp_legacy_read_authorized helper (#9892); one import line."
   },

.github/scripts/product_file_line_count_ratchet_baseline/backend-utils.json

@@ -5,11 +5,11 @@
     "backend/utils/memory_ingestion/pipeline.py": 1711,
     "backend/utils/other/storage.py": 1584,
     "backend/utils/retrieval/tools/calendar_tools.py": 1513,
-    "backend/utils/sync/pipeline.py": 2299
+    "backend/utils/sync/pipeline.py": 2324
   },
   "raise_justifications": {
     "backend/utils/other/storage.py": "delete_app_logo now requires an exact app-logo bucket prefix (startswith) so a foreign URL embedding the prefix later cannot delete an unrelated object on this deletion path (David review on #9873).",
-    "backend/utils/sync/pipeline.py": "Durable-ledger convergence remains beside the Sync pipeline's run-lock ownership and terminal-result transitions instead of splitting one recovery invariant across modules."
+    "backend/utils/sync/pipeline.py": "Plan-aware soft-cap evaluation and its default-tier fallback remain in a small helper beside Sync's idempotent fresh-speech metering rather than overgrowing the durable worker coordinator."
   },
   "threshold": 1500
 }

backend/routers/sync.py

@@ -967,6 +967,17 @@ async def sync_local_files_v2(
                 request_id=x_request_id if isinstance(x_request_id, str) else None,
                 cloud_trace_context=x_cloud_trace_context if isinstance(x_cloud_trace_context, str) else None,
             )
+        if await run_blocking(db_executor, is_daily_audio_ceiling_exceeded, uid):
+            logger.info('sync_v2: daily audio ceiling reached uid=%s', uid)
+            return await _fair_use_restriction_response(
+                uid=uid,
+                retry_after=_retry_after_until_next_utc_day(),
+                client_platform=client_device_context.platform,
+                device_hash=client_device_context.device_hash,
+                app_version=client_device_context.app_version,
+                request_id=x_request_id if isinstance(x_request_id, str) else None,
+                cloud_trace_context=x_cloud_trace_context if isinstance(x_cloud_trace_context, str) else None,
+            )
 
     should_lock = not await run_blocking(critical_executor, has_transcription_credits, uid)
 

backend/tests/unit/test_sync_cloud_tasks.py

@@ -1006,6 +1006,7 @@ def _submit_with_context(executor, fn, *args, **kwargs):
 
     sys.modules['utils.fair_use'].is_hard_restricted = MagicMock(return_value=False)
     sys.modules['utils.fair_use'].get_hard_restriction_status = MagicMock(return_value=(False, None))
+    sys.modules['utils.fair_use'].is_daily_audio_ceiling_exceeded = MagicMock(return_value=False)
     sys.modules['utils.fair_use'].is_dg_budget_exhausted = MagicMock(return_value=False)
     sys.modules['utils.fair_use'].get_enforcement_stage = MagicMock(return_value='off')
     sys.modules['utils.fair_use'].FAIR_USE_ENABLED = False
@@ -2018,6 +2019,89 @@ def _rmtree_raises(*args, **kwargs):
                 sys.modules[mod_name] = orig
 
 
+@pytest.mark.asyncio
+async def test_fresh_admission_daily_ceiling_prevents_staging_or_dispatch():
+    """A fresh V2 upload over the hard ceiling must leave its WAL audio untouched."""
+    from starlette.datastructures import UploadFile
+
+    module, saved_modules, mock_sync_jobs, BytesIO, _, _ = _load_sync_router_for_fast_path()
+    module.is_daily_audio_ceiling_exceeded = MagicMock(return_value=True)
+    module._fair_use_restriction_response = AsyncMock(
+        return_value=module.JSONResponse(status_code=429, content={'code': 'fair_use_restricted'})
+    )
+    module.start_background_task = MagicMock()
+    module.classify_sync_lane = MagicMock(
+        return_value=types.SimpleNamespace(
+            lane=module.SyncLane.FRESH,
+            trust=types.SimpleNamespace(value='device_bound'),
+            reason='recent_capture',
+            maximum_age_seconds=60,
+            automatic_recovery_allowed=True,
+        )
+    )
+
+    try:
+        upload = UploadFile(filename='test.opus', file=BytesIO(b'\x00' * 10))
+        response = await module.sync_local_files_v2(files=[upload], uid='test-uid')
+
+        assert response.status_code == 429
+        assert json.loads(response.body) == {'code': 'fair_use_restricted'}
+        module.is_daily_audio_ceiling_exceeded.assert_called_once_with('test-uid')
+        module._fair_use_restriction_response.assert_awaited_once()
+        restriction_kwargs = module._fair_use_restriction_response.await_args.kwargs
+        assert restriction_kwargs['uid'] == 'test-uid'
+        assert restriction_kwargs['retry_after'] > 0
+        module._retrieve_file_paths_v2.assert_not_called()
+        mock_sync_jobs.create_sync_job.assert_not_called()
+        module.claim_sync_content.assert_not_called()
+        module.enqueue_sync_job.assert_not_called()
+        module.start_background_task.assert_not_called()
+        module.has_transcription_credits.assert_not_called()
+    finally:
+        sys.modules.pop('routers.sync', None)
+        sys.modules.pop('utils.sync.pipeline', None)
+        for mod_name, orig in saved_modules.items():
+            if orig is None:
+                sys.modules.pop(mod_name, None)
+            else:
+                sys.modules[mod_name] = orig
+
+
+@pytest.mark.asyncio
+async def test_backfill_admission_does_not_consume_fresh_daily_ceiling():
+    """Historical recovery keeps its independent pacing policy, even above the live ceiling."""
+    from starlette.datastructures import UploadFile
+
+    module, saved_modules, mock_sync_jobs, BytesIO, _, _ = _load_sync_router_for_fast_path()
+    module.is_daily_audio_ceiling_exceeded = MagicMock(return_value=True)
+    module.start_background_task = MagicMock()
+    module.classify_sync_lane = MagicMock(
+        return_value=types.SimpleNamespace(
+            lane=module.SyncLane.BACKFILL,
+            trust=types.SimpleNamespace(value='legacy'),
+            reason='historical_recovery',
+            maximum_age_seconds=24 * 60 * 60,
+            automatic_recovery_allowed=True,
+        )
+    )
+
+    try:
+        upload = UploadFile(filename='test.opus', file=BytesIO(b'\x00' * 10))
+        response = await module.sync_local_files_v2(files=[upload], uid='test-uid')
+
+        assert response.status_code == 202
+        module.is_daily_audio_ceiling_exceeded.assert_not_called()
+        mock_sync_jobs.create_sync_job.assert_called_once()
+    finally:
+        sys.modules.pop('routers.sync', None)
+        sys.modules.pop('utils.sync.pipeline', None)
+        for mod_name, orig in saved_modules.items():
+            if orig is None:
+                sys.modules.pop(mod_name, None)
+            else:
+                sys.modules[mod_name] = orig
+
+
 @pytest.mark.asyncio
 async def test_sync_dispatch_cloud_tasks_success_increments_attempts(monkeypatch):
     from starlette.datastructures import UploadFile

backend/tests/unit/test_sync_v2.py

@@ -23,6 +23,7 @@
 
 import pytest
 from fastapi.routing import APIRoute
+from models.users import PlanType
 from utils.executors import run_blocking as _production_run_blocking
 
 PIPELINE_PATH = os.path.join(os.path.dirname(__file__), '..', '..', 'utils', 'sync', 'pipeline.py')
@@ -2372,6 +2373,67 @@ def _vad_with_segments(path, segmented_paths, errors):
         finally:
             self._cleanup(stubs['saved_modules'])
 
+    @pytest.mark.asyncio
+    @pytest.mark.parametrize(
+        ('subscription', 'expected_plan', 'lookup_error'),
+        [
+            (types.SimpleNamespace(plan=PlanType.unlimited_v2), PlanType.unlimited_v2, None),
+            (types.SimpleNamespace(plan=PlanType.basic), PlanType.basic, None),
+            (None, None, None),
+            (None, None, RuntimeError('subscription store unavailable')),
+        ],
+        ids=['unlimited', 'basic', 'missing-subscription', 'subscription-read-failure'],
+    )
+    async def test_fresh_soft_caps_use_the_current_subscription_plan(self, subscription, expected_plan, lookup_error):
+        """Queued fresh work applies the persisted plan instead of the default cap tier."""
+        module, stubs = self._load_sync_module()
+        try:
+            pipeline = stubs['pipeline']
+            pipeline.decode_files_to_wav = MagicMock(return_value=['/tmp/w.wav'])
+            pipeline._cleanup_files = MagicMock()
+
+            def _vad_with_segments(path, segmented_paths, errors):
+                segmented_paths.add('/tmp/seg_1700000001.wav')
+
+            speech_totals = {'daily_ms': 5_000, 'three_day_ms': 5_000, 'weekly_ms': 5_000}
+            pipeline.retrieve_vad_segments = _vad_with_segments
+            pipeline.get_wav_duration = MagicMock(return_value=5.0)
+            pipeline.FAIR_USE_ENABLED = True
+            pipeline.FAIR_USE_RESTRICT_DAILY_DG_MS = 0
+            pipeline.record_speech_ms = MagicMock()
+            pipeline.get_rolling_speech_ms = MagicMock(return_value=speech_totals)
+            pipeline.check_soft_caps = MagicMock(return_value=[])
+            pipeline.users_db = MagicMock()
+            pipeline.users_db.get_existing_user_subscription = MagicMock(return_value=subscription)
+            if lookup_error:
+                pipeline.users_db.get_existing_user_subscription.side_effect = lookup_error
+            pipeline.users_db.get_user_transcription_preferences = MagicMock(return_value={})
+            pipeline.users_db.get_user_private_cloud_sync_enabled = MagicMock(return_value=False)
+            pipeline.users_db.get_data_protection_level = MagicMock(return_value=None)
+            pipeline.build_person_embeddings_cache = MagicMock(return_value={})
+            pipeline.process_segment = MagicMock()
+            pipeline.record_dg_usage_ms = MagicMock()
+            pipeline.record_usage = MagicMock()
+
+            await module._run_full_pipeline_background_async('j-plan', 'uid', ['/tmp/f.opus'], 'omi', False, '/tmp/job')
+
+            pipeline.users_db.get_existing_user_subscription.assert_called_once_with('uid')
+            pipeline.check_soft_caps.assert_called_once_with('uid', speech_totals=speech_totals, plan=expected_plan)
+            stubs['sync_jobs'].finalize_sync_job.assert_called_once()
+            if lookup_error:
+                pipeline.record_fallback.assert_called_once_with(
+                    component='other',
+                    from_mode='subscription_plan',
+                    to_mode='default_cap',
+                    reason='policy',
+                    outcome='degraded',
+                    log=pipeline.logger,
+                )
+            else:
+                pipeline.record_fallback.assert_not_called()
+        finally:
+            self._cleanup(stubs['saved_modules'])
+
     @pytest.mark.asyncio
     async def test_partial_segment_failure_completes(self):
         """Partial segment failure must complete (not fail) with error count."""
@@ -2789,6 +2851,7 @@ def _submit_with_context(executor, fn, *args, **kwargs):
         # Set up fair_use defaults
         sys.modules['utils.fair_use'].is_hard_restricted = MagicMock(return_value=False)
         sys.modules['utils.fair_use'].get_hard_restriction_status = MagicMock(return_value=(False, None))
+        sys.modules['utils.fair_use'].is_daily_audio_ceiling_exceeded = MagicMock(return_value=False)
         sys.modules['utils.fair_use'].is_dg_budget_exhausted = MagicMock(return_value=False)
         sys.modules['utils.fair_use'].get_enforcement_stage = MagicMock(return_value='off')
         sys.modules['utils.fair_use'].FAIR_USE_ENABLED = False

backend/utils/sync/pipeline.py

@@ -95,6 +95,7 @@
     upload_audio_chunk,
     upload_syncing_temporal_file,
 )
+from utils.observability.fallback import record_fallback
 from utils.observability.transcription import record_sync_transcription_outcome
 from utils.speaker_assignment import process_speaker_assigned_segments
 from utils.speaker_identification import detect_speaker_from_text
@@ -156,6 +157,27 @@ def _bounded_exception_type(error: BaseException) -> str:
     return name if name.replace('_', '').isalnum() and len(name) <= 64 else 'Exception'
 
 
+async def _resolve_fair_use_soft_cap_plan(uid: str):
+    """Return the stored plan, falling back to the default soft-cap tier on read failure."""
+    try:
+        fair_use_sub = await run_blocking(db_executor, users_db.get_existing_user_subscription, uid)
+        return fair_use_sub.plan if fair_use_sub else None
+    except Exception as e:
+        logger.warning(
+            'event=sync_fair_use outcome=subscription_plan_fallback exception_type=%s',
+            _bounded_exception_type(e),
+        )
+        record_fallback(
+            component='other',
+            from_mode='subscription_plan',
+            to_mode='default_cap',
+            reason='policy',
+            outcome='degraded',
+            log=logger,
+        )
+        return None
+
+
 def _bounded_sync_failure_reason(reason: str | None) -> str:
     return reason if reason in _SYNC_FAILURE_REASON_CODES else 'other'
 
@@ -1789,8 +1811,11 @@ async def _run_full_pipeline_background_async(
                         raise_on_error=bool(content_id),
                     )
                 if sync_lane == SyncLane.FRESH.value:
+                    fair_use_plan = await _resolve_fair_use_soft_cap_plan(uid)
                     speech_totals = await run_blocking(db_executor, get_rolling_speech_ms, uid)
-                    triggered_caps = await run_blocking(db_executor, check_soft_caps, uid, speech_totals=speech_totals)
+                    triggered_caps = await run_blocking(
+                        db_executor, check_soft_caps, uid, speech_totals=speech_totals, plan=fair_use_plan
+                    )
                     if triggered_caps:
                         logger.info(
                             'event=sync_fair_use outcome=soft_cap_triggered cap_count=%d',