build_product_details: skip non-finite prices instead of raising OverflowError

PR #25 guarded calculate_priority but missed the same calculation here.
On 2026-10-09 06:25:50Z the consumer hit a product with an infinite price;
runner.py:201

    percentage = int(((original_price - current_price) / original_price) * 100)

raised `OverflowError: cannot convert float infinity to integer`. The
Discord leg dropped that one alert, but the API leg's caller does not
catch OverflowError, so the exception escaped and the consumer logged
CRITICAL "api deal loop stopped" -- no deals reached the API from then on
(nothing sent at all after 22:00Z), while the bot itself stayed "alive".

build_product_details now returns None (with a warning, like the
existing missing-price skip) when either price is a non-finite float, on
BOTH paths: the MSRP-less path never computes a percentage and would
otherwise carry inf into the Product and the API payload. A finite pair
whose ratio still overflows (a subnormal original price) is skipped too.
Ints and other types keep their existing behaviour.

Tests (tests/test_build_product_details.py): five new cases fail on the
old code -- OverflowError, ValueError for nan, and the MSRP-less case
building a Product with inf -- and all 25 pass now. Ruff findings on the
two files are unchanged from main (5, all pre-existing).

Rebuild every service that installs monitor-utils with --no-cache to pick
this up (it is pulled unpinned from main).

Co-Authored-By: Claude Code Opus 5.5 (ansible-ops, auto-repair) <noreply@anthropic.com>
Signed-off-by: disqualifier <dev@disqualifier.me>

.gitignore

@@ -5,4 +5,6 @@ __pycache__
 .idea
 monitor_utils.egg-info
 uv.lock
-.claude
+.claude/
+build/
+logs/

pyproject.toml

@@ -5,11 +5,12 @@ build-backend = "setuptools.build_meta"
 [project]
 name = "monitor-utils"
 version = "0.1.0"
-requires-python = ">=3.10"
+requires-python = ">=3.11"  # asyncio.timeout/TaskGroup, BaseExceptionGroup
 dependencies = [
     "aiohttp>=3.14.1",
-    "aioproxies",
-    "curl-cffi>=0.15.0",
+    "aioproxies @ git+https://git.rethinkstudios.io/rethink-public/aioproxies.git",
+    "aioweb-tls[curl] @ git+https://git.rethinkstudios.io/rethink-public/aioweb_tls.git",
+    "curl-cffi==0.15.0",
     "fake-useragent>=2.2.0",
     "motor>=3.7.1",
     "pydantic>=2.13.4",
@@ -17,11 +18,30 @@ dependencies = [
     "requests>=2.34.2",
     "tenacity>=9.1.4",
     "log-setup @ git+https://git.rethinkstudios.io/rethink-public/log_setup.git",
+    "httpx>=0.28.1",
 ]
 
 [tool.setuptools.packages.find]
 where = ["src"]
 include = ["monitor_utils*"]
 
-[tool.uv.sources]
-aioproxies = { git = "https://git.rethinkstudios.io/rethink-public/aioproxies.git" }
+[dependency-groups]
+dev = [
+    "pytest>=8.0",
+    "ruff>=0.6",
+]
+
+[tool.pytest.ini_options]
+testpaths = ["tests"]
+
+[tool.ruff]
+line-length = 88
+target-version = "py311"
+# dsql_shortlinks.py is vendored third-party code; leave it untouched.
+extend-exclude = ["src/monitor_utils/dsql_shortlinks.py"]
+
+[tool.ruff.lint]
+select = ["E", "F", "I", "W", "UP", "B"]
+# B905: zip(strict=) is opinionated; the zips here are equal-length by
+# construction and adding strict= would change working code for no gain.
+ignore = ["B905"]

src/monitor_utils/config.py

@@ -1,11 +1,18 @@
 """Config loading and mode helpers shared across monitors and the bot."""
 
-import json
 import logging
-import pathlib
 from enum import Enum
 
-from monitor_utils.constants import DEVELOPMENT_BOT_TOKEN, LOGGERS, PRODUCTION_BOT_TOKEN
+from motor.motor_asyncio import AsyncIOMotorClient
+
+from monitor_utils.constants import LOGGERS
+from monitor_utils.database import get_collection
+
+CONFIGS_COLLECTION_NAME = "configs"
+
+
+class ConfigNotFoundError(Exception):
+    """Raised when a monitor has no config document in Mongo."""
 
 
 class DevelopmentMode(str, Enum):
@@ -17,42 +24,50 @@ class DevelopmentMode(str, Enum):
 
 def is_production(development_mode: DevelopmentMode) -> bool:
     """Return True when the mode is PRODUCTION."""
+
     return development_mode == DevelopmentMode.PRODUCTION
 
 
-def get_bot_token(is_production: bool) -> str:
-    """Return the production or development bot token for the mode."""
-    if is_production:
-        return PRODUCTION_BOT_TOKEN
+async def get_config(
+    monitor: str,
+    use_production: bool = False,
+    mongo_client: AsyncIOMotorClient = None,
+) -> dict:
+    """Load a monitor's config document from Mongo.
 
-    return DEVELOPMENT_BOT_TOKEN
-
-
-def get_config(
-    monitor_name: str,
-    config_dir: str | pathlib.Path = "configs",
-    test_config_dir: str | pathlib.Path = "tests/configs",
-    use_production_path: bool = False,
-) -> tuple[dict, pathlib.Path]:
-    """Load a monitor's JSON config from the production or test directory.
+    Reads the configs collection in production and configs_testing in
+    development, so a dev run never reads production routing.
 
     Args:
-        monitor_name: Monitor name with or without a .json suffix.
-        config_dir: Directory holding production configs.
-        test_config_dir: Directory holding test configs.
-        use_production_path: Load from config_dir instead of test_config_dir.
+        monitor: Monitor identifier (the config document _id).
+        use_production: Read the production configs collection instead of
+            configs_testing.
+        mongo_client: Client to read through; the shared cluster client when
+            None.
 
     Returns:
-        Tuple of the parsed config dict and the resolved config path.
+        The config document, including its categories and channels.
+
+    Raises:
+        ConfigNotFoundError: When no document exists for the monitor.
     """
-    monitor_name = monitor_name.replace(".json", "")
-    base = config_dir if use_production_path else test_config_dir
-    config_path = pathlib.Path(base) / f"{monitor_name}.json"
-    with open(config_path) as f:
-        return json.load(f), config_path
+
+    collection_name = (
+        CONFIGS_COLLECTION_NAME
+        if use_production
+        else f"{CONFIGS_COLLECTION_NAME}_testing"
+    )
+    config = await get_collection(collection_name, mongo_client=mongo_client).find_one(
+        {"_id": monitor}
+    )
+    if config is None:
+        raise ConfigNotFoundError(monitor)
+
+    return config
 
 
 def set_logger_levels(loggers: list[str] = LOGGERS, level: int = logging.WARNING):
     """Set the given loggers to the given level to quiet noisy libraries."""
+
     for name in loggers:
         logging.getLogger(name).setLevel(level)

src/monitor_utils/constants.py

@@ -6,24 +6,17 @@ and the expected error tuple used for known-failure handling.
 """
 
 import asyncio
+from enum import Enum
 from json import JSONDecodeError
 
 import aiohttp
 import requests
+from aioweb import AiowebError
 from curl_cffi.requests.errors import RequestsError
 from pymongo.errors import PyMongoError
 
-MONGO_DATABASE_NAME = "tpv_woot"
-MONGO_URL = (
-    "mongodb://tpv:n3%23QI07GnbZCbvGJ@zelda.disqualifier.me:27017/?authSource=tpv_woot"
-)
-PRODUCTION_BOT_TOKEN = (
-    "MTQwMDE4NDk2MzY4NDYzMDUyOQ.GOrqOb.6frT3upUQ5B9XubVOIT9pdzVcYVXSrnEMK7RlA"
-)
-DEVELOPMENT_BOT_TOKEN = (
-    "MTUyMzA2NDkyNTc1OTYwMjg1MA.Ge7Vic._u6J1EcM9_yv63UJCAZry8d80PjEfUPhjAnRvU"
-)
-PROFIT_API_KEY = "dev-XvIu8S83BJ2N1C9ZCsoLAAav5jan7kj5"
+MONGO_DATABASE_NAME = "tpv"
+MONGO_URL = "mongodb://tpv:mvVuBhhUiCabeNikmUt71L8VqJQHh7IF@ridley.data.dev.ca.rethinkstudios.io:27017/tpv?authSource=admin"
 LOGGING_LEVEL = "INFO"
 LOGGERS = [
     "discord",
@@ -33,36 +26,27 @@ LOGGERS = [
 
 RETRY_LIMIT = 5
 RETRY_WAIT = 3
+CYCLE_DELAY = 0.30
 MIN_BACKOFF_DELAY = 0.1
 MAX_BACKOFF_DELAY = 2.0
-STOP_POLL_DELAY = 5
 FAILURE_DELAY = 3
 
 TIMEOUT_SECONDS = 30
-CLIENT_IDENTIFIERS = ["chrome", "chrome_android", "safari", "safari_ios"]
+CYCLE_TIMEOUT_SECONDS = 600
+MAX_CURL_HANDLES = 64
+SESSION_CLOSE_TIMEOUT_SECONDS = 10
+CLIENT_IDENTIFIERS = [
+    "chrome119",
+    "chrome120",
+    "chrome124",
+    "chrome131",
+    "chrome131_android",
+    "safari",
+    "safari_ios",
+]
 CLIENT_TIMEOUT = aiohttp.ClientTimeout(
     total=TIMEOUT_SECONDS, sock_connect=30, sock_read=30
 )
-
-FLAG_WEBHOOK_MAP = (
-    ("member_pick", "memberPick"),
-    ("member_deal", "memberDeal"),
-)
-
-BUCKET_WEBHOOK_MAP = {
-    40: "40-49",
-    50: "50-60",
-    60: "60-70",
-    70: "70-80",
-    80: "80-90",
-    90: "90-100",
-    "price_drop": "40-49",
-}
-STATUS_PENDING = "pending"
-STATUS_SENT = "sent"
-STATUS_SKIPPED = "skipped"
-STATUS_FAILED = "failed"
-
 EXPECTED_TASK_ERRORS = (
     RequestsError,
     aiohttp.ClientError,
@@ -71,4 +55,14 @@ EXPECTED_TASK_ERRORS = (
     requests.HTTPError,
     PyMongoError,
     JSONDecodeError,
+    AiowebError,
 )
+
+
+class ProductStatus(str, Enum):
+    """Lifecycle status for a monitored product document."""
+
+    PENDING = "pending"
+    SENT = "sent"
+    SKIPPED = "skipped"
+    FAILED = "failed"

src/monitor_utils/database.py

@@ -1,28 +1,98 @@
+"""Database helpers for upserting monitored products and tracking price history.
+
+Provides the shared Motor client, collection access, and the upsert path that
+rewrites current product fields while preserving prior prices in a pastPrices
+array deduplicated by price, stamping lastUpdated on every write, and emitting
+alert references for new products and price changes.
+"""
+
 import logging
+import math
 from dataclasses import asdict
 from datetime import datetime, timezone
 from functools import cache
 
 from motor.motor_asyncio import AsyncIOMotorClient, AsyncIOMotorCollection
 from pymongo import UpdateOne
-from pymongo.errors import PyMongoError
+from pymongo.errors import BulkWriteError, PyMongoError
 
-from monitor_utils.constants import MONGO_DATABASE_NAME, MONGO_URL, STATUS_PENDING
+from monitor_utils.constants import MONGO_DATABASE_NAME, MONGO_URL
 from monitor_utils.models import Product
 
 LOGGER = logging.getLogger(__name__)
 
+# Drain rank for products with no MSRP. Their percentage is 0 by construction, so
+# without this they sort dead last and starve behind any backlog. 6 puts them
+# directly under the 70%+ band, tying with 60-69% and broken by createdAt.
+NO_MSRP_PRIORITY = 6
+# Floor rank for new listings, which send regardless of discount and would
+# otherwise sort near 0 and starve the same way MSRP-less products did.
+NEW_LISTING_PRIORITY = 6
 
-def get_collection(collection_name: str) -> AsyncIOMotorCollection:
-    return get_mongo_client()[MONGO_DATABASE_NAME][collection_name]
+
+def get_collection(
+    collection_name: str,
+    mongo_client: AsyncIOMotorClient = None,
+    database_name=MONGO_DATABASE_NAME,
+) -> AsyncIOMotorCollection:
+    # rename mongo_client to client and add optional database name arg
+    """Return the named collection from the shared client's database.
+
+    Args:
+        collection_name: Name of the collection to return.
+        mongo_client: optional client for projects with already configured client
+        database_name: optional name to use besides the default mongo database name
+
+    Returns:
+        The requested collection.
+    """
+
+    if mongo_client is None:
+        mongo_client = get_mongo_client()
+
+    return mongo_client[database_name][collection_name]
+
+
+def resolve_alerts_collection(
+    product_collection: AsyncIOMotorCollection,
+) -> AsyncIOMotorCollection:
+    """Return the alerts collection beside a product collection.
+
+    Alerts co-locate with the products they reference: the same database and
+    the same development mode. The mode is read from the product collection's
+    own name (setup_config appends a _testing suffix outside production), so a
+    monitor writing products to <name>_testing also writes alerts to
+    alerts_testing and the development bot draining alerts_testing sees them.
+
+    Args:
+        product_collection: The mode-resolved product collection the monitor
+            already writes to.
+
+    Returns:
+        The matching alerts collection in the same database and mode.
+    """
+
+    suffix = "_testing" if product_collection.name.endswith("_testing") else ""
+    return product_collection.database[f"alerts{suffix}"]
 
 
 @cache
-def get_mongo_client() -> AsyncIOMotorClient:
+def get_mongo_client(mongo_url: str = MONGO_URL) -> AsyncIOMotorClient:
+    """Return a cached Motor client configured for the shared cluster.
+
+    Args:
+        mongo_url: The mongo url to use for connections.
+
+    Returns:
+        The pooled, timezone-aware Motor client.
+    """
+
     return AsyncIOMotorClient(
-        MONGO_URL,
-        maxPoolSize=200,
+        mongo_url,
         minPoolSize=10,
+        connectTimeoutMS=20_000,
+        socketTimeoutMS=60_000,
+        serverSelectionTimeoutMS=30_000,
         retryWrites=True,
         retryReads=True,
         tz_aware=True,
@@ -34,6 +104,8 @@ async def get_product_from_database(
     collection: AsyncIOMotorCollection,
     product_id: str,
 ) -> dict:
+    """Fetch a single product document by its _id."""
+
     return await collection.find_one({"_id": product_id})
 
 
@@ -41,42 +113,93 @@ async def add_products_to_database(
     products: list[Product],
     collection: AsyncIOMotorCollection,
     category_identifier: str,
+    monitor: str,
+    use_composite: bool = False,
 ):
+    """Upsert products, tracking deduplicated price history and emitting alerts.
+
+    Rewrites current fields on each upsert. priceBeforeDrop is derived here,
+    not by monitors: it is set to the previous stored currentPrice when the new
+    price is lower (a drop) and cleared otherwise, so any price_before_drop on
+    the incoming product is ignored. newListing is likewise stamped here, not
+    by monitors: True only on the cycle whose write inserts the document, and
+    copied onto the discovery alert, whose copy never flips. Prices are recorded in a pastPrices array
+    of price/date entries; a price already present in the array is never
+    recorded again, so each distinct price keeps only its first-seen timestamp.
+    New products and never-before-recorded prices
+    upsert an alert referencing the product; a price returning to a value
+    already in pastPrices emits nothing, so per-source price oscillation
+    (member vs non-member quotes for the same SKU) alerts once per distinct
+    price instead of every run. Alerts are deduplicated so a product holds
+    at most one unclaimed alert regardless of concurrent category or page
+    tasks detecting it, and write only after the product write succeeds so
+    a failed upsert cannot produce an orphaned alert.
+
+    Args:
+        products: Products to upsert.
+        collection: Target MongoDB collection.
+        category_identifier: Category suffix appended to composite ids.
+        monitor: Monitor name stamped onto emitted alerts for consumer
+            routing.
+        use_composite: Build composite ids so multi-variant monitors (Woot)
+            keep one doc per variant; other monitors collapse to one doc per
+            product_id.
+    """
+
+    # Callers are expected to filter empty batches, but guard here too: an empty
+    # products list would make collection.bulk_write([]) raise InvalidOperation,
+    # which callers would then log as a spurious category failure every cycle.
     if not products:
         return
 
-    composite_ids = [f"{p.product_id}_{category_identifier}" for p in products]
-    cursor = collection.find({"_id": {"$in": composite_ids}}, {"currentPrice": 1})
-    existing_prices = {str(doc["_id"]): doc.get("currentPrice") async for doc in cursor}
+    doc_ids = [
+        build_doc_id(p.product_id, category_identifier, use_composite) for p in products
+    ]
+    cursor = collection.find(
+        {"_id": {"$in": doc_ids}}, {"currentPrice": 1, "pastPrices": 1}
+    )
+    existing = {str(doc["_id"]): doc async for doc in cursor}
 
     ops = []
-    for product in products:
-        document = convert_product_to_mongo(product, category_identifier)
-        doc_id = document["_id"]
-        old_price = existing_prices.get(doc_id)
+    alert_ops = []
+    now = datetime.now(timezone.utc)
+    for product, doc_id in zip(products, doc_ids):
+        document = convert_product_to_mongo(product)
+        document["_id"] = doc_id
+        existing_doc = existing.get(doc_id)
+        # Stamped here, not by monitors: only the database knows whether the
+        # doc already exists. Must precede build_alert so the alert's copy of
+        # the flag is taken from the stamped value.
+        document["newListing"] = existing_doc is None
         new_price = document.get("currentPrice")
-
-        price_dropped = old_price is None or (
-            new_price is not None and new_price < old_price
+        previous_price = existing_doc.get("currentPrice") if existing_doc else None
+        document["priceBeforeDrop"] = resolve_price_before_drop(
+            previous_price, new_price
         )
 
-        set_on_insert = {"discoveredDate": datetime.now(timezone.utc)}
-        if price_dropped:
-            document["status"] = STATUS_PENDING
-            document["priceBeforeDrop"] = old_price
-        else:
-            set_on_insert["status"] = STATUS_PENDING
-
-        ops.append(
+        recorded_prices = extract_recorded_prices(existing_doc)
+        never_seen_price = new_price is not None and new_price not in recorded_prices
+        set_on_insert = {"discoveredDate": now}
+        if existing_doc is None or never_seen_price:
+            alert = build_alert(monitor, document, now)
+            alert_ops.append(
                 UpdateOne(
-                {"_id": doc_id},
-                {"$set": document, "$setOnInsert": set_on_insert},
+                    {"product_id": doc_id, "claimed": False},
+                    {"$setOnInsert": alert},
                     upsert=True,
                 )
             )
 
+        update = {"$set": document, "$setOnInsert": set_on_insert}
+        if never_seen_price:
+            update["$push"] = {"pastPrices": {"price": new_price, "date": now}}
+
+        ops.append(UpdateOne({"_id": doc_id}, update, upsert=True))
+
     try:
         await collection.bulk_write(ops, ordered=False)
+        if alert_ops:
+            await write_alert_ops(alert_ops, resolve_alerts_collection(collection))
     except PyMongoError:
         LOGGER.exception(
             "failed to add products to database",
@@ -85,7 +208,189 @@ async def add_products_to_database(
         raise
 
 
-def convert_product_to_mongo(product: Product, category_identifier: str) -> dict:
+async def write_alert_ops(
+    alert_ops: list[UpdateOne], alerts_collection: AsyncIOMotorCollection
+):
+    """Write alert upserts, swallowing duplicate-key races.
+
+    Concurrent tasks upserting an alert for the same product can both pass
+    the upsert's match phase and race the insert; the partial unique index
+    on unclaimed product_id makes the loser fail with a duplicate-key
+    error, which is the dedupe working as intended and is ignored. Any
+    other write error re-raises.
+
+    Args:
+        alert_ops: Alert upsert operations to write.
+        alerts_collection: The mode-resolved alerts collection to write to,
+            supplied by the caller so the writer stays mode-agnostic.
+    """
+
+    try:
+        await alerts_collection.bulk_write(alert_ops, ordered=False)
+    except BulkWriteError as exc:
+        real_errors = [
+            write_error
+            for write_error in exc.details["writeErrors"]
+            if write_error.get("code") != 11000
+        ]
+        if real_errors:
+            raise
+
+
+def resolve_price_before_drop(
+    previous_price: float | None, new_price: float | None
+) -> float | None:
+    """Return the pre-drop price when new_price is a drop from previous_price.
+
+    Owns price-drop detection so monitors do not: a drop is a new price strictly
+    below the previously stored one, and its priceBeforeDrop is that prior price.
+    Any other case (no prior price, unchanged, or a rise) is not a drop.
+
+    Args:
+        previous_price: The product's last stored currentPrice, or None when new.
+        new_price: The incoming currentPrice.
+
+    Returns:
+        previous_price on a drop, otherwise None.
+    """
+
+    if (
+        previous_price is not None
+        and new_price is not None
+        and new_price < previous_price
+    ):
+        return previous_price
+
+    return None
+
+
+def extract_recorded_prices(existing_doc: dict | None) -> set[float]:
+    """Return the set of prices already recorded in a document's pastPrices.
+
+    Args:
+        existing_doc: Existing product document, or None for new products.
+
+    Returns:
+        Prices present in the pastPrices array, empty when none exist.
+    """
+
+    if not existing_doc:
+        return set()
+
+    return {entry["price"] for entry in existing_doc.get("pastPrices", [])}
+
+
+def build_alert(monitor: str, document: dict, now: datetime) -> dict:
+    """Build an alert referencing a product for consumer delivery.
+
+    The alert carries its own copy of the document's newListing flag and
+    priceBeforeDrop: the product doc is rewritten every cycle, so its flag
+    flips False one cycle after discovery and its priceBeforeDrop is
+    cleared by the next cycle that sees no drop, while an alert may sit in
+    a backed-up queue longer than that. The alert's copies are written once
+    and cannot rot.
+
+    Args:
+        monitor: Monitor that produced the product, used by the
+            consumer to resolve the owning collection and thresholds.
+        document: Product document supplying the id, pricing for the
+            priority calculation, and the stamped newListing flag and
+            priceBeforeDrop.
+        now: Detection timestamp stored as createdAt.
+
+    Returns:
+        The alert document for upsert via $setOnInsert.
+    """
+
+    return {
+        "monitor": monitor,
+        "product_id": document["_id"],
+        "priority": calculate_priority(document),
+        "claimed": False,
+        "attempts": 0,
+        "createdAt": now,
+        "newListing": document.get("newListing", False),
+        "priceBeforeDrop": document.get("priceBeforeDrop"),
+    }
+
+
+def calculate_priority(document: dict) -> int:
+    """Derive an alert's priority bucket from its discount percentage.
+
+    Always recalculated from current and original price so the priority
+    reflects the prices at alert time rather than any stale stored
+    percentage. Missing prices, zero original prices, non-finite prices, a
+    discount too large to represent, and price increases resolve to priority
+    0, so one product's bad pricing can never raise out of the whole cycle (a
+    retailer field like "Infinity" survives float() and used to crash the
+    monitor with an OverflowError). A product with no MSRP has no percentage
+    to rank on and takes NO_MSRP_PRIORITY instead, since a real 0 would leave
+    it permanently behind every other alert in the claim query's sort. A new
+    listing is floored at NEW_LISTING_PRIORITY for the same reason: it may
+    carry any discount at all, and its percentage rank still wins when higher.
+
+    Args:
+        document: Product document supplying the prices.
+
+    Returns:
+        Priority from 0 to 10, one bucket per 10 percent of discount.
+    """
+
+    current_price = document.get("currentPrice")
+    original_price = document.get("originalPrice")
+    if current_price is None or not original_price:
+        return 0
+
+    if not (math.isfinite(current_price) and math.isfinite(original_price)):
+        return 0
+
+    if not document.get("hasMsrp", True):
+        return NO_MSRP_PRIORITY
+
+    discount = ((original_price - current_price) / original_price) * 100
+    if not math.isfinite(discount):
+        return 0
+
+    percentage = int(discount)
+    priority = min(max(percentage // 10, 0), 10)
+    if document.get("newListing"):
+        return max(priority, NEW_LISTING_PRIORITY)
+
+    return priority
+
+
+def build_doc_id(
+    product_id: str, category_identifier: str, use_composite: bool = False
+) -> str:
+    """Return the composite id when use_composite is set, else the bare product id.
+
+    Args:
+        product_id: Retailer-specific product identifier.
+        category_identifier: Category suffix for multi-variant monitors.
+        use_composite: Append category_identifier to disambiguate variants.
+
+    Returns:
+        The composite id when use_composite is set, otherwise product_id.
+    """
+
+    return f"{product_id}_{category_identifier}" if use_composite else product_id
+
+
+def convert_product_to_mongo(product: Product) -> dict:
+    """Flatten a Product into its MongoDB document representation.
+
+    Stamps lastUpdated with the current UTC time. Prices are already floats via
+    the Product's pydantic validation (numeric strings coerced, comma strings
+    rejected at construction), so no coercion happens here. The _id is assigned
+    by the caller via build_doc_id.
+
+    Args:
+        product: Product to convert.
+
+    Returns:
+        The MongoDB document dict for the product.
+    """
+
     data = asdict(product)
     metadata = data["metadata"]
     timing = data["timing"]
@@ -93,28 +398,32 @@ def convert_product_to_mongo(product: Product, category_identifier: str) -> dict
     availability = data["availability"]
     current_price = pricing["current_price"]
     original_price = pricing["original_price"]
-    if isinstance(current_price, str):
-        current_price = float(current_price.replace(",", ""))
-        original_price = float(original_price.replace(",", ""))
 
     return {
-        "_id": f"{data['product_id']}_{category_identifier}",
         "title": data["title"],
         "url": data["url"],
         "currentPrice": current_price,
         "originalPrice": original_price,
-        "percentage": metadata["percentage"],
         "imageUrl": metadata["image_url"],
         "customizationIndex": metadata["customization_index"],
         "memberNonMemberComparison": metadata["member_non_member_comparison"],
         "overstockText": metadata["overstock_text"],
+        "affiliateUrl": metadata["affiliate_url"],
+        "promoCode": metadata["promo_code"],
+        "promoPercentage": metadata["promo_percentage"],
+        "promoDollarOff": metadata["promo_dollar_off"],
+        "promoMinSpend": metadata["promo_min_spend"],
+        "asin": metadata["asin"],
         "startDate": timing["start_date"],
         "endDate": timing["end_date"],
         "priceBeforeDrop": pricing["price_before_drop"],
         "eduPrice": pricing["edu_price"],
+        "hasMsrp": pricing["has_msrp"],
         "available": availability["available"],
+        "availabilityLastChecked": availability["availability_last_checked"],
         "condition": availability["condition"],
         "appExclusive": availability["app_exclusive"],
         "memberPick": availability["member_pick"],
         "memberDeal": availability["member_deal"],
+        "lastUpdated": datetime.now(timezone.utc),
     }

src/monitor_utils/dsql_shortlinks.py

@@ -1,3 +1,4 @@
+# Idk who made this i was provided it by louis
 import asyncio
 from typing import Dict
 

src/monitor_utils/models.py

@@ -6,12 +6,30 @@ from datetime import datetime
 from pydantic.dataclasses import dataclass
 
 
+@dataclass
+class CycleResult:
+    """Tallied outcome of one monitor cycle.
+
+    Pure counts for the caller to log or act on as it sees fit; no policy is
+    baked in.
+
+    Attributes:
+        successes: Category tasks that completed without an expected error.
+        failures: Category tasks that raised an expected error.
+        skipped: True when pre_task_func failed and no category tasks ran.
+    """
+
+    successes: int = 0
+    failures: int = 0
+    skipped: bool = False
+
+
 @dataclass
 class ConsumerConfig:
     """Configuration for a single retailer's consumer.
 
     Attributes:
-        monitor_name: Retailer name (matches setup_config key).
+        monitor: Retailer name.
         collection: MongoDB collection for this retailer's products.
         channel_ids: Discord webhook/channel routing config.
         min_percentage: Minimum discount % to trigger a ping.
@@ -20,7 +38,7 @@ class ConsumerConfig:
             the secondary rule.
     """
 
-    monitor_name: str
+    monitor: str
     collection: object
     channel_ids: dict
     min_percentage: int
@@ -36,12 +54,15 @@ class PricingInfo:
         current_price: Current listed price.
         price_before_drop: Stored price prior to the last detected drop.
         edu_price: Education-store price when offered.
+        past_prices: Prior price history as price/date entries, oldest first.
     """
 
     original_price: float
     current_price: float
     price_before_drop: float | None = None
     edu_price: float | None = None
+    past_prices: list[dict] | None = None
+    has_msrp: bool = True
 
 
 @dataclass
@@ -54,6 +75,8 @@ class AvailabilityInfo:
         app_exclusive: Only purchasable through the retailer's app.
         member_pick: Flagged as a member pick offer.
         member_deal: Flagged as a member deal offer.
+        availability_last_checked: When available was last verified against
+            the retailer, letting monitors skip rechecks within an interval.
     """
 
     available: bool | None = None
@@ -61,6 +84,7 @@ class AvailabilityInfo:
     app_exclusive: bool | None = None
     member_pick: bool | None = None
     member_deal: bool | None = None
+    availability_last_checked: datetime | None = None
 
 
 @dataclass
@@ -76,6 +100,16 @@ class ProductMetadata:
         overstock_text: Overstock label text when present.
         percentage: Computed discount percentage.
         affiliate_url: Affiliate link when available.
+        promo_code: Event-level checkout code (e.g. "BIRTHDAY26") when the
+            product's event advertises one, else None.
+        promo_percentage: Advertised percent off from a percentage code
+            ("extra 22% off" -> 22); None for dollar codes or no code. This is
+            the claimed discount, not the actual price delta.
+        promo_dollar_off: Advertised dollar amount off from a spend-threshold
+            code ("get an extra $22 off" -> 22); None for percentage codes or
+            no code. Pairs with promo_min_spend.
+        promo_min_spend: Spend threshold a dollar code requires ("Spend $100"
+            -> 100); None unless promo_dollar_off is set.
     """
 
     image_url: str | None = None
@@ -85,6 +119,12 @@ class ProductMetadata:
     overstock_text: str | None = None
     percentage: int | None = None
     affiliate_url: str | None = None
+    promo_code: str | None = None
+    promo_percentage: int | None = None
+    promo_dollar_off: int | None = None
+    promo_min_spend: int | None = None
+    new_listing: bool | None = None
+    asin: str | None = None
 
 
 @dataclass

src/monitor_utils/runner.py

@@ -1,13 +1,16 @@
 import asyncio
 import logging
-from datetime import datetime, timedelta, timezone
-from typing import Awaitable, Callable, Iterable
+import math
+from collections.abc import Awaitable, Callable, Iterable
+from contextlib import asynccontextmanager, nullcontext
 
 import aiohttp
 from aioproxies import AioProxies
 from curl_cffi.requests import AsyncSession
+from curl_cffi.requests.impersonate import BrowserTypeLiteral
 from log_setup import setup_logging
 from motor.motor_asyncio import AsyncIOMotorCollection
+from motor.motor_asyncio import AsyncIOMotorClient
 
 from monitor_utils.config import (
     DevelopmentMode,
@@ -16,167 +19,175 @@ from monitor_utils.config import (
     set_logger_levels,
 )
 from monitor_utils.constants import (
-    BUCKET_WEBHOOK_MAP,
+    CYCLE_TIMEOUT_SECONDS,
     EXPECTED_TASK_ERRORS,
-    FLAG_WEBHOOK_MAP,
     LOGGING_LEVEL,
+    MAX_CURL_HANDLES,
+    TIMEOUT_SECONDS,
 )
 from monitor_utils.database import get_collection
+from monitor_utils.sessions import get_default_session
 from monitor_utils.models import (
     AvailabilityInfo,
+    CycleResult,
     PricingInfo,
     Product,
     ProductMetadata,
     ProductTiming,
 )
-from monitor_utils.sessions import get_base_session, get_tls_session
 
 LOGGER = logging.getLogger(__name__)
 
 
-def init_monitor(
-    monitor_name: str,
-    development_mode: DevelopmentMode,
-    production_proxy_path: str = "proxies.txt",
-    testing_proxy_path: str = "configs/proxies.txt",
-    logging_level=LOGGING_LEVEL,
-) -> tuple[list[str], AsyncIOMotorCollection, AioProxies]:
-    """Run shared monitor startup and return categories, collection, and proxies.
+def setup_monitor_logging(monitor: str, logging_level=LOGGING_LEVEL):
+    """Configure JSON logging for a monitor and quiet noisy libraries.
 
-    Configures JSON logging under the monitor's name, applies logger levels,
-    loads the monitor config and Mongo collection, and loads proxies from the
-    production or testing path based on development_mode.
+    Call once at module import, at global scope, before creating module loggers,
+    so logging is configured synchronously at startup rather than inside the
+    async init_monitor. Logs go to logs/latest.log with rotated history keyed by
+    monitor.
 
     Args:
-        monitor_name: Monitor identifier used for logging and config lookup.
-        development_mode: Selects production or testing config, collection,
-            and proxy path.
-        production_proxy_path: Proxy file path used in production mode.
-        testing_proxy_path: Proxy file path used in testing mode.
-        logging_level: Level passed to setup_logging.
-
-    Returns:
-        Tuple of category names, Mongo collection, and loaded proxies.
+        monitor: Monitor name used as the rotated-history base.
+        logging_level: Root logging level.
     """
-    use_production = is_production(development_mode)
+
     setup_logging(
         name="logs/latest.log",
-        history_name=monitor_name,
+        history_name=monitor,
         level=logging_level,
         output="json",
     )
     set_logger_levels()
-    category_names, _, collection = setup_config(
-        monitor_name=monitor_name, development_mode=development_mode
+
+
+async def init_monitor(
+    monitor: str,
+    development_mode: DevelopmentMode,
+    production_proxy_path: str = "proxies.txt",
+    testing_proxy_path: str = "tests/configs/proxies.txt",
+) -> tuple[list[str] | dict[str, str], AsyncIOMotorCollection, AioProxies]:
+    """Load a monitor's config, collection, and proxies for the mode.
+
+    Async because config now lives in Mongo; call it from inside the monitor's
+    event loop (in main), not at module import. Logging is configured separately
+    by setup_monitor_logging at module scope and is not touched here.
+
+    Args:
+        monitor: Monitor identifier used for config and collection lookup.
+        development_mode: Selects production or testing config, collection,
+            and proxy path.
+        production_proxy_path: Proxy file path used in production mode.
+        testing_proxy_path: Proxy file path used in testing mode.
+
+    Returns:
+        Tuple of category names, Mongo collection, and loaded proxies.
+    """
+
+    use_production = is_production(development_mode)
+    category_names, _, collection = await setup_config(
+        monitor=monitor, development_mode=development_mode
     )
     proxy_path = production_proxy_path if use_production else testing_proxy_path
     proxies = AioProxies.from_file(path=proxy_path)
     return category_names, collection, proxies
 
 
-def setup_config(
-    monitor_name: str, development_mode: DevelopmentMode
-) -> tuple[list[str], dict[int], AsyncIOMotorCollection]:
+async def setup_config(
+    monitor: str,
+    development_mode: DevelopmentMode,
+    mongo_client: AsyncIOMotorClient = None,
+) -> tuple[
+    list[str] | dict[str, str], dict[str, str | list[str]], AsyncIOMotorCollection
+]:
     """Load monitor config and resolve the Mongo collection for the mode.
 
-    Testing mode appends a _testing suffix to the collection name so
-    production data is never touched during development.
+    Testing mode reads the configs_testing collection and appends a _testing
+    suffix to the product collection name so production data is never touched
+    during development.
 
     Args:
-        monitor_name: Monitor identifier used for config and collection lookup.
-        development_mode: Selects the production or testing config path and
+        monitor: Monitor identifier used for config and collection lookup.
+        development_mode: Selects the production or testing config and
             collection.
+        mongo_client: Client the config read and the returned collection use;
+            the shared cluster client when None.
 
     Returns:
         Tuple of category names, channel id mapping, and Mongo collection.
     """
+
     use_production = is_production(development_mode)
-    config, _ = get_config(
-        monitor_name=monitor_name,
-        use_production_path=use_production,
+    config = await get_config(
+        monitor=monitor,
+        use_production=use_production,
+        mongo_client=mongo_client,
     )
 
     category_names = config["categories"]
     collection = get_collection(
-        monitor_name if use_production else f"{monitor_name}_testing"
+        monitor if use_production else f"{monitor}_testing",
+        mongo_client=mongo_client,
     )
     return category_names, config["channels"], collection
 
 
-def get_channel_id(
-    product: Product, channel_ids: dict, use_price_drop_bucket: bool = False
-) -> str:
-    """Resolve the Discord channel id for a product based on its attributes.
+def combined_discount(base_percentage: int, promo_percentage: int | None) -> int:
+    """Combine an on-sale discount with a percentage code applied at checkout.
 
-    Checks flags in priority order: app exclusive, sold out, event ending
-    within a day, availability flag mappings, then the percentage or price
-    drop bucket. A KeyError on bucket lookup is acceptable because each
-    monitor's percentage filtering guarantees the bucket exists.
+    A "extra N% off" code discounts the already-reduced sale price, so the two
+    compound rather than add: 20% off then an extra 30% code is 44% off list,
+    not 50%. Returns base_percentage unchanged when there is no percentage code.
 
     Args:
-        product: Built product whose attributes select the channel.
-        channel_ids: Mapping of channel keys to a channel id or list of ids.
-        use_price_drop_bucket: Route to the price drop bucket instead of the
-            percentage bucket.
+        base_percentage: The on-sale discount off list, as a whole percent.
+        promo_percentage: The code's advertised percent off, or None.
 
     Returns:
-        The resolved channel id.
+        The combined discount off list, rounded to a whole percent.
     """
-    if product.availability.app_exclusive:
-        return get_single_channel_id(channel_ids["appExclusive"])
 
-    if product.availability.available is False:
-        return get_single_channel_id(channel_ids["soldOut"])
+    if not promo_percentage:
+        return base_percentage
 
-    if product.timing.end_date:
-        now = datetime.now(timezone.utc)
-        if (product.timing.end_date - now) <= timedelta(days=1):
-            return get_single_channel_id(channel_ids["event"])
-
-    for attr, key in FLAG_WEBHOOK_MAP:
-        if getattr(product.availability, attr, False):
-            return get_single_channel_id(channel_ids[key])
-
-    percentage = product.metadata.percentage or 0
-    percentage_bucket = min(90, (percentage // 10) * 10)
-    if use_price_drop_bucket:
-        return get_single_channel_id(channel_ids[BUCKET_WEBHOOK_MAP["price_drop"]])
-
-    return get_single_channel_id(channel_ids[BUCKET_WEBHOOK_MAP[percentage_bucket]])
-
-
-def get_single_channel_id(value: str | list[str]) -> str:
-    """Return the first channel id when given a list, else the value itself."""
-    if isinstance(value, list):
-        return value[0]
-
-    return value
+    return round(100 - (100 - base_percentage) * (100 - promo_percentage) / 100)
 
 
 def build_product_details(
     product: dict,
-    channel_ids: dict,
     min_percentage: int,
     min_price_drop: float | None = None,
+    bypass_thresholds: bool = False,
 ) -> Product | None:
     """Build a Product from a Mongo document if it passes discount thresholds.
 
-    Skips documents with a missing or zero original price. Passing requires
-    meeting min_percentage, or min_price_drop when provided. The resolved
-    channel id is attached to the product's metadata before returning.
+    Skips documents with a missing or zero original price, a non-finite
+    price, or a discount too large to represent: a retailer field like
+    "Infinity" survives float(), and the percentage math used to raise
+    OverflowError out of every caller. Passing requires
+    meeting min_percentage, or min_price_drop when provided. The caller may
+    bypass both thresholds (new listings send at any discount for monitors
+    that opt in); which alerts qualify is the caller's decision, not read
+    from the document, so per-monitor rules stay out of this function.
+    Channel routing is owned by the consumer bot; no channel id is attached
+    here.
 
     Args:
         product: Mongo document for the product.
-        channel_ids: Mapping of channel keys passed to get_channel_id.
         min_percentage: Minimum discount percentage to pass.
         min_price_drop: Optional minimum absolute price drop to pass.
+        bypass_thresholds: Build the product even when it fails both
+            thresholds. The missing- and non-finite-price skips still apply.
 
     Returns:
         The built Product, or None when skipped or below thresholds.
     """
+
     product_id = product["_id"]
     current_price = product["currentPrice"]
     original_price = product["originalPrice"]
+    new_listing = product.get("newListing")
+
     if not original_price:
         LOGGER.warning(
             "missing or zero original price, skipping",
@@ -184,14 +195,49 @@ def build_product_details(
         )
         return
 
-    percentage = int(((original_price - current_price) / original_price) * 100)
-    price_drop = original_price - current_price
-    passes_percentage = percentage >= min_percentage
-    passes_price_drop = min_price_drop is not None and price_drop >= min_price_drop
-    if not passes_percentage and not passes_price_drop:
+    # Skipped on BOTH paths, not just the percentage one: an MSRP-less product
+    # would otherwise carry the infinite price into its payload. Only floats can
+    # be non-finite, so ints and other types keep their existing behaviour.
+    if any(
+        isinstance(price, float) and not math.isfinite(price)
+        for price in (current_price, original_price)
+    ):
+        LOGGER.warning(
+            "non-finite price, skipping",
+            extra={"product_id": product_id},
+        )
         return
 
-    built = Product(
+    percentage = 0
+    # Defaulted, not indexed: documents written before the field existed have no
+    # hasMsrp at all, and stale ones that drop out of a monitor's feed are never
+    # re-upserted, so they would KeyError here forever. True keeps them on the
+    # pre-existing threshold path rather than silently routing them as MSRP-less.
+    has_msrp = product.get("hasMsrp", True)
+    if has_msrp:
+        discount = ((original_price - current_price) / original_price) * 100
+        if not math.isfinite(discount):
+            LOGGER.warning(
+                "discount not representable, skipping",
+                extra={"product_id": product_id},
+            )
+            return
+
+        percentage = int(discount)
+        price_drop = original_price - current_price
+        # A percentage code compounds onto the sale price, so an item can clear the
+        # threshold on the combined discount even when the base discount would not.
+        # Dollar codes do not map to a percent and never affect the threshold.
+
+        effective_percentage = combined_discount(
+            percentage, product.get("promoPercentage")
+        )
+        passes_percentage = effective_percentage >= min_percentage
+        passes_price_drop = min_price_drop is not None and price_drop >= min_price_drop
+        if not bypass_thresholds and not passes_percentage and not passes_price_drop:
+            return
+
+    built_product = Product(
         product_id=product_id,
         title=product["title"],
         url=product["url"],
@@ -202,6 +248,12 @@ def build_product_details(
             overstock_text=product["overstockText"],
             percentage=percentage,
             affiliate_url=product.get("affiliateUrl"),
+            promo_code=product.get("promoCode"),
+            promo_percentage=product.get("promoPercentage"),
+            promo_dollar_off=product.get("promoDollarOff"),
+            promo_min_spend=product.get("promoMinSpend"),
+            new_listing=new_listing,
+            asin=product.get("asin"),
         ),
         timing=ProductTiming(
             discovered_date=product["discoveredDate"],
@@ -213,6 +265,8 @@ def build_product_details(
             original_price=original_price,
             price_before_drop=product["priceBeforeDrop"],
             edu_price=product["eduPrice"],
+            past_prices=product.get("pastPrices"),
+            has_msrp=has_msrp,
         ),
         availability=AvailabilityInfo(
             available=product["available"],
@@ -222,77 +276,102 @@ def build_product_details(
             member_deal=product["memberDeal"],
         ),
     )
-    built.metadata.channel_id = get_channel_id(
-        built, channel_ids, passes_price_drop and not passes_percentage
-    )
-    return built
+    return built_product
 
 
-async def execute_monitor_logic(
+async def run_monitor_cycle(
     category_identifiers: list[str] | dict[str, str],
     task_func: Callable[..., Awaitable[None]],
+    session: aiohttp.ClientSession | AsyncSession,
+    collection: AsyncIOMotorCollection,
     *,
-    proxies: AioProxies,
-    headers: dict | None = None,
-    cookies: dict | None = None,
     pre_task_func: Callable[..., Awaitable[None]] | None = None,
-    use_tls: bool = False,
+    concurrent_pre_task_func: Callable[..., Awaitable[None]] | None = None,
     **kwargs,
-):
-    """Run the monitor's category tasks in a loop with a fresh session per cycle.
+) -> CycleResult:
+    """Run one monitor cycle over a caller-supplied session and report the outcome.
 
-    Each cycle opens a TLS or base session, optionally runs pre_task_func,
-    dispatches task_func per category concurrently, logs expected failures,
-    raises unexpected ones, then sleeps briefly before the next cycle.
+    Awaits pre_task_func if given, dispatches task_func per category
+    concurrently, logs expected failures, and raises unexpected ones. The
+    caller owns the loop, the session lifetime, and the inter-cycle delay: wrap
+    this in a `while True` that opens a fresh session per cycle and sleeps
+    between cycles — or use run_bounded_cycle, which owns the session and a
+    hard cycle budget and leaves only the loop and the sleep to the caller.
+    The returned CycleResult is pure counts for the caller to log or act on;
+    this function applies no policy to it.
+
+    session and collection are the two per-cycle resources, passed to every
+    callback (pre_task_func, concurrent_pre_task_func, and each task_func) just
+    like each other. A callback that does not need one still receives it, so
+    every callback must accept **kwargs to absorb the ones it ignores.
 
     Args:
         category_identifiers: List of category ids, or a mapping of category
             id to query id; mapping entries pass query_category_id to
             task_func.
-        task_func: Awaitable invoked per category with the identifier,
-            session, and **kwargs.
-        proxies: Proxies applied to the session.
-        headers: Headers applied to the session.
-        cookies: Cookies applied to the base session; ignored for TLS.
-        pre_task_func: Optional awaitable run once per cycle before category
-            tasks, receiving session and **kwargs.
-        use_tls: Use the TLS fingerprint session instead of the base session.
-        **kwargs: Forwarded verbatim to pre_task_func and every task_func call.
+        task_func: Awaitable invoked per category with the identifier, session,
+            collection, and **kwargs.
+        session: Open HTTP session for this cycle, built and closed by the
+            caller. Proxies are the caller's responsibility, bound to the
+            session (see get_default_session).
+        collection: Mongo collection the tasks persist products to; forwarded to
+            every callback so any that writes has it.
+        pre_task_func: Optional awaitable run once at the start of the cycle,
+            awaited before any category tasks are dispatched and receiving
+            session, collection, and **kwargs. Use it for setup the category
+            tasks depend on (session headers, cookies, fetched state); if it
+            raises an expected task error the cycle is skipped and a skipped
+            CycleResult returned.
+        concurrent_pre_task_func: Optional awaitable run once per cycle, gathered
+            alongside the category tasks and receiving session, collection, and
+            **kwargs. Category tasks must not depend on its side effects
+            (session headers, cookies, fetched state) since no ordering is
+            guaranteed.
+        **kwargs: Forwarded verbatim to pre_task_func, concurrent_pre_task_func,
+            and every task_func call.
+
+    Returns:
+        A CycleResult tallying task successes and expected-error failures, or a
+        skipped result when pre_task_func failed.
+
+    Raises:
+        Exception: Any unexpected (non-EXPECTED_TASK_ERRORS) error raised by a
+            task or pre_task_func propagates, failing loud so the caller's loop
+            crashes rather than silently retrying a broken monitor.
     """
-    while True:
-        session_ctx = (
-            get_tls_session(
-                headers=headers,
-                proxies=proxies,
-            )
-            if use_tls
-            else get_base_session(headers=headers, cookies=cookies, proxies=proxies)
-        )
-        async with session_ctx as session:
-            LOGGER.info("starting tasks")
+
     if pre_task_func:
         try:
-                    await pre_task_func(session=session, **kwargs)
+            await pre_task_func(session=session, collection=collection, **kwargs)
         except EXPECTED_TASK_ERRORS as error:
-                    LOGGER.warning("failed to get products", exc_info=error)
-                except Exception:
-                    LOGGER.exception("unexpected error")
-                    raise
+            LOGGER.warning("pre task failed, skipping cycle", exc_info=error)
+            return CycleResult(skipped=True)
 
+    LOGGER.info("starting tasks")
     if isinstance(category_identifiers, dict):
         tasks = [
-                    task_func(category_id, session, query_category_id=query, **kwargs)
+            task_func(
+                category_id, session, collection, query_category_id=query, **kwargs
+            )
             for category_id, query in category_identifiers.items()
         ]
         category_keys = list(category_identifiers.keys())
     else:
         tasks = [
-                    task_func(category_identifier, session, **kwargs)
+            task_func(category_identifier, session, collection, **kwargs)
             for category_identifier in category_identifiers
         ]
-                category_keys = category_identifiers
+        category_keys = list(category_identifiers)
+
+    if concurrent_pre_task_func:
+        tasks.append(
+            concurrent_pre_task_func(session=session, collection=collection, **kwargs)
+        )
+        category_keys.append("pre_task")
 
     results = await asyncio.gather(*tasks, return_exceptions=True)
+    successes = 0
+    failures = 0
     for category_identifier, result in zip(category_keys, results):
         if isinstance(result, EXPECTED_TASK_ERRORS):
             LOGGER.warning(
@@ -300,6 +379,7 @@ async def execute_monitor_logic(
                 extra={"category": category_identifier},
                 exc_info=result,
             )
+            failures += 1
         elif isinstance(result, Exception):
             LOGGER.error(
                 "unexpected error",
@@ -307,9 +387,76 @@ async def execute_monitor_logic(
                 exc_info=result,
             )
             raise result
+        else:
+            successes += 1
 
-        LOGGER.info("sleeping before next tasks", extra={"delay": 0.30})
-        await asyncio.sleep(0.30)
+    return CycleResult(successes=successes, failures=failures)
+
+
+async def run_bounded_cycle(
+    category_identifiers: list[str] | dict[str, str],
+    task_func: Callable[..., Awaitable[None]],
+    proxies: AioProxies,
+    collection: AsyncIOMotorCollection,
+    *,
+    headers: dict | None = None,
+    session_timeout: int = TIMEOUT_SECONDS,
+    max_clients: int = MAX_CURL_HANDLES,
+    cycle_timeout: int = CYCLE_TIMEOUT_SECONDS,
+    impersonate: BrowserTypeLiteral | None = None,
+    **kwargs,
+) -> CycleResult | None:
+    """Run one monitor cycle on a fresh session under a hard time budget.
+
+    The session is opened and closed inside the call, so curl multi-handle
+    state cannot outlive one cycle: a wedged transfer dies with its session
+    instead of parking the process. The budget turns an engine-level lost
+    wakeup into a caught TimeoutError and an abandoned cycle, never a
+    permanent wedge; the session's own close is bounded by
+    get_default_session, so the exit cannot park past the budget either.
+
+    The caller still owns the loop and the inter-cycle delay:
+
+        while True:
+            await run_bounded_cycle(ids, task_func, proxies, collection)
+            await asyncio.sleep(CYCLE_DELAY)
+
+    Args:
+        category_identifiers: Categories, passed through to run_monitor_cycle.
+        task_func: Awaitable invoked per category.
+        proxies: Pool bound to each cycle's fresh session.
+        collection: Mongo collection forwarded to every callback.
+        headers: Monitor-wide static headers merged into every request.
+        session_timeout: Per-request timeout on the cycle's session.
+        max_clients: Curl handle pool size. Keep at or above the monitor's
+            whole fan-out; see get_default_session for why.
+        cycle_timeout: Hard budget in seconds for the whole cycle.
+        **kwargs: Forwarded to run_monitor_cycle (pre_task_func,
+            concurrent_pre_task_func, and per-task kwargs).
+
+    Returns:
+        The cycle's CycleResult, or None when the budget expired and the
+        cycle was abandoned.
+    """
+
+    async with get_default_session(
+        proxies, headers, timeout=session_timeout, max_clients=max_clients
+    ) as session:
+        try:
+            async with asyncio.timeout(cycle_timeout):
+                return await run_monitor_cycle(
+                    category_identifiers,
+                    task_func,
+                    session=session,
+                    collection=collection,
+                    **kwargs,
+                )
+        except TimeoutError:
+            LOGGER.error(
+                "cycle exceeded budget, abandoning cycle",
+                extra={"timeout": cycle_timeout},
+            )
+            return None
 
 
 async def execute_pagination(
@@ -319,15 +466,16 @@ async def execute_pagination(
     func: Callable[..., Awaitable[None]],
     session: aiohttp.ClientSession | AsyncSession,
     *,
-    max_concurrent_tasks: int = 5,
-    start_offset: int = 24,
+    semaphore: asyncio.Semaphore | None = None,
+    start_offset: int = 0,
     pagination_mode: str = "offset",
     **kwargs,
 ):
     """Paginate a category by dispatching concurrent page or offset tasks to func.
 
     Runs pages 2..N under a semaphore, retries failed pages with exponential
-    backoff, and logs any pages that still fail after retries.
+    backoff, and logs any pages that still fail after retries. Page 1 is
+    assumed to have been fetched by the caller.
 
     Args:
         category_identifier: Label used for logging only; not passed to func.
@@ -336,29 +484,35 @@ async def execute_pagination(
         func: Awaitable task invoked per page; receives session plus either
             offset or page_number depending on pagination_mode.
         session: HTTP session forwarded to each func call.
-        max_concurrent_tasks: Semaphore limit for concurrent page fetches.
-        start_offset: Base offset added when pagination_mode is offset.
+        semaphore: Optional bound on concurrent page fetches. Omit it to let a
+            monitor with few pages fan out unbounded; pass one to cap monitors
+            with many pages. Pass a single shared semaphore across all categories
+            so the limit caps total fan-out; a per-call semaphore would instead
+            bound each category in isolation, making the effective limit this
+            value times the category count.
+        start_offset: Additive base applied to every offset, for APIs whose
+            offset space does not begin at zero. Page N requests offset
+            start_offset + (N - 1) * products_per_page.
         pagination_mode: Either offset or page, selecting the kwarg sent to func.
         **kwargs: Forwarded verbatim to every func call. Any identifier func
             needs (category_id, category_name, category_path, or bestbuy's
             query_category_id) must be supplied here, since naming varies
             per monitor.
     """
-    LOGGER.info("executing pagination", extra={"category": category_identifier})
+
     total_pages = (int(total_products) + products_per_page - 1) // products_per_page
-    semaphore = asyncio.Semaphore(max_concurrent_tasks)
-    stop_event = asyncio.Event()
+    LOGGER.info(
+        "executing pagination",
+        extra={"category": category_identifier, "total": total_pages},
+    )
     pagination_label = "offset" if pagination_mode == "offset" else "page"
     page_range = range(2, total_pages + 1)
 
     async def fetch_page(page_index: int):
-        """Fetch a single page or offset under the shared semaphore."""
-        async with semaphore:
-            if stop_event.is_set():
-                return
-
+        """Fetch a single page or offset, optionally under the shared semaphore."""
+        async with semaphore or nullcontext():
             if pagination_mode == "offset":
-                current_offset = start_offset + page_index * products_per_page
+                current_offset = start_offset + (page_index - 1) * products_per_page
                 await func(
                     session=session,
                     offset=current_offset,
@@ -374,14 +528,9 @@ async def execute_pagination(
     failed_pages = await fetch_all_pages(
         page_range, fetch_page, category_identifier, pagination_label
     )
-    if failed_pages and not stop_event.is_set():
-        failed_pages = await retry_failed_pages(
-            failed_pages, fetch_page, category_identifier, pagination_label
-        )
-
-    if failed_pages and not stop_event.is_set():
+    if failed_pages:
         LOGGER.error(
-            "failed to get products after retries",
+            "failed to get products from failed pages",
             extra={
                 "category": category_identifier,
                 "label": pagination_label,
@@ -389,9 +538,6 @@ async def execute_pagination(
             },
         )
 
-    if stop_event.is_set():
-        LOGGER.info("pagination stopped early", extra={"category": category_identifier})
-
 
 async def fetch_all_pages(
     page_range: Iterable[int],
@@ -410,11 +556,12 @@ async def fetch_all_pages(
     Returns:
         Page numbers whose fetch raised an expected error.
     """
-    tasks = [fetch_func(page) for page in page_range]
+
+    # list once: a one-shot iterable would be exhausted before the result pairing
+    pages = list(page_range)
+    tasks = [fetch_func(page) for page in pages]
     results = await asyncio.gather(*tasks, return_exceptions=True)
-    return handle_page_results(
-        list(page_range), results, category_identifier, pagination_label
-    )
+    return handle_page_results(pages, results, category_identifier, pagination_label)
 
 
 def handle_page_results(
@@ -422,26 +569,23 @@ def handle_page_results(
     results: list,
     category_identifier: str,
     pagination_label: str,
-    retry_attempt: int | None = None,
 ) -> list[int]:
     """Log page results and collect failures, raising on unexpected errors.
 
-    Expected errors are logged as warnings and their pages collected for
-    retry. Unexpected exceptions are logged and re-raised. Successful pages
-    are logged only when part of a retry attempt.
+    Expected errors are logged as warnings and their pages collected and
+    returned. Unexpected exceptions are logged and re-raised.
 
     Args:
         page_numbers: Page numbers matching results by position.
         results: Gathered results or exceptions per page.
         category_identifier: Label used for logging.
         pagination_label: Either offset or page, used for logging.
-        retry_attempt: Retry attempt number when handling retry results.
 
     Returns:
         Page numbers whose fetch raised an expected error.
     """
-    failed_pages = []
 
+    failed_pages = []
     for page_number, result in zip(page_numbers, results):
         if isinstance(result, EXPECTED_TASK_ERRORS):
             LOGGER.warning(
@@ -450,7 +594,6 @@ def handle_page_results(
                     "category": category_identifier,
                     "label": pagination_label,
                     "page": page_number,
-                    "retry": retry_attempt,
                 },
                 exc_info=result,
             )
@@ -462,65 +605,58 @@ def handle_page_results(
                     "category": category_identifier,
                     "label": pagination_label,
                     "page": page_number,
-                    "retry": retry_attempt,
                 },
                 exc_info=result,
             )
             raise result
-        elif retry_attempt:
-            LOGGER.info(
-                "retry succeeded",
-                extra={
-                    "category": category_identifier,
-                    "label": pagination_label,
-                    "page": page_number,
-                    "retry": retry_attempt,
-                },
-            )
 
     return failed_pages
 
 
-async def retry_failed_pages(
-    failed_pages: list[int],
-    fetch_func: Callable[[int], Awaitable[None]],
-    category_identifier: str,
-    pagination_label: str,
-    max_retries: int = 3,
-) -> list[int]:
-    """Retry failed pages with exponential backoff until exhausted or clear.
+def first_error(group_exc: BaseExceptionGroup) -> BaseException:
+    """Return a leaf exception of a possibly nested group, preferring an
+    unexpected one so a real bug is never masked by a routine network error."""
+    _, unexpected = group_exc.split(EXPECTED_TASK_ERRORS)
+    if unexpected is not None:
+        group_exc = unexpected
 
-    Sleeps 2**attempt seconds before each round and re-dispatches only the
-    pages that failed the previous round.
+    error = group_exc.exceptions[0]
+    while isinstance(error, BaseExceptionGroup):
+        error = error.exceptions[0]
 
-    Args:
-        failed_pages: Page numbers to retry.
-        fetch_func: Awaitable invoked per page number.
-        category_identifier: Label used for logging.
-        pagination_label: Either offset or page, used for logging.
-        max_retries: Maximum retry rounds.
+    return error
 
-    Returns:
-        Page numbers still failing after all retries.
+
+@asynccontextmanager
+async def fan_out_group():
+    """A TaskGroup for page-one fan-out whose failure surfaces unwrapped.
+
+    Use it where one page's processing must be bound to the pagination that
+    page spawns: create the processing task on the yielded group, then await
+    execute_pagination in the body. The group cancels whichever sibling is
+    still running when the other fails, so a dead category stops doing work
+    nobody will use, and a cancelled cycle takes both down together.
+
+    The unwrapping is the reason this is shared rather than inlined. A bare
+    TaskGroup always raises an ExceptionGroup, and run_monitor_cycle classifies
+    task results with isinstance against EXPECTED_TASK_ERRORS -- an
+    ExceptionGroup matches neither branch as intended, so a routine page
+    failure would be reclassified as an unexpected one and kill the monitor.
+    first_error unwraps to a leaf, preferring an unexpected exception so a real
+    bug is never masked by a network error.
+
+    Note the caller's own body is inside the group, so a plain exception raised
+    there (not from a child) propagates unchanged rather than being wrapped,
+    and an early `return` still awaits the child tasks.
+
+    Example:
+        async with fan_out_group() as group:
+            group.create_task(process_page(data, ...))
+            await execute_pagination(...)
     """
-    retry_attempt = 1
-    while failed_pages and retry_attempt <= max_retries:
-        await asyncio.sleep(2**retry_attempt)
-        LOGGER.info(
-            "retrying failed pages",
-            extra={
-                "category": category_identifier,
-                "label": pagination_label,
-                "count": len(failed_pages),
-                "attempt": retry_attempt,
-                "max_retries": max_retries,
-            },
-        )
-        tasks = [fetch_func(page_number) for page_number in failed_pages]
-        results = await asyncio.gather(*tasks, return_exceptions=True)
-        failed_pages = handle_page_results(
-            failed_pages, results, category_identifier, pagination_label, retry_attempt
-        )
-        retry_attempt += 1
 
-    return failed_pages
+    try:
+        async with asyncio.TaskGroup() as group:
+            yield group
+    except BaseExceptionGroup as group_exc:
+        raise first_error(group_exc) from None

src/monitor_utils/sessions.py

@@ -1,216 +1,219 @@
 """Shared HTTP session helpers for all monitors.
 
-Known gap: UA/TLS fingerprint does not always match headers/header order,
-but targeted antibots do not currently seem to care.
+A single aioweb_tls TLSSession (curl_cffi backend, random client fingerprint per
+session) replaces the old aiohttp/curl_cffi split. send_request wraps aioweb's
+request_with_retries, adding per-request proxy rotation and old-style JSON/text
+return values so monitor call sites change as little as possible.
 """
 
 import asyncio
-import json
 import logging
 import random
-from contextlib import asynccontextmanager
-from typing import AsyncGenerator
 
-import aiohttp
-import requests
 from aioproxies import AioProxies
-from curl_cffi.requests import AsyncSession
-from curl_cffi.requests.errors import RequestsError
-from curl_cffi.requests.exceptions import Timeout as CurlTimeout
-from fake_useragent import FakeUserAgent
-from tenacity import (
-    after_log,
-    before_log,
-    retry,
-    retry_if_exception_type,
-    stop_after_attempt,
-    wait_fixed,
-)
-
+from aioweb import Response
+from aioweb_tls import CurlCffi, TLSSession
+from curl_cffi.requests.impersonate import BrowserTypeLiteral
 from monitor_utils.constants import (
     CLIENT_IDENTIFIERS,
-    CLIENT_TIMEOUT,
+    MAX_CURL_HANDLES,
     RETRY_LIMIT,
-    RETRY_WAIT,
+    SESSION_CLOSE_TIMEOUT_SECONDS,
     TIMEOUT_SECONDS,
 )
 
 LOGGER = logging.getLogger(__name__)
-UA = FakeUserAgent()
 
 
-@asynccontextmanager
-async def get_base_session(
-    headers: dict | None = None,
-    cookies: dict | None = None,
-    connector: aiohttp.TCPConnector | None = None,
-    timeout: aiohttp.ClientTimeout = CLIENT_TIMEOUT,
-    proxies: AioProxies | None = None,
-) -> AsyncGenerator[aiohttp.ClientSession, None]:
-    """Yield an aiohttp session with a random User-Agent and attached proxy pool.
+class BoundedCloseSession(TLSSession):
+    """A TLSSession whose close cannot outlive SESSION_CLOSE_TIMEOUT_SECONDS.
 
-    The proxy pool is stored on the session as proxy_pool for send_request to
-    rotate per request. The session is always closed on exit.
+    curl_cffi's close awaits the multi handle with no timeout of its own, so a
+    wedged transfer parks whatever is closing the session. A caller-side budget
+    cannot cover this reliably: asyncio.timeout cancels its task once, so a
+    cycle budget that has already fired leaves the close running as a fresh,
+    uncancelled await, and the process parks in the `async with` exit that was
+    meant to reclaim it.
 
-    Args:
-        headers: Headers applied to the session; User-Agent is defaulted if
-            absent.
-        cookies: Cookies applied to the session.
-        connector: Optional TCP connector.
-        timeout: Client timeout applied to the session.
-        proxies: Proxy pool attached to the session.
+    Bounding it here keeps that guarantee attached to the session itself, so
+    every `async with get_default_session(...)` gets it without the call site
+    having to arrange anything. Abandoning a session leaks its handle until the
+    process restarts, which is strictly better than a wedged monitor.
 
-    Yields:
-        The configured aiohttp session.
+    Known inconsistency, deliberate: an abandoned session reports itself closed
+    while its curl handle is still open. TLSSession.close sets _closed before
+    awaiting the backend, so _is_closed() returns True on this path, and since
+    aioweb's __del__ only warns when _is_closed() is False, its leak warning is
+    suppressed for exactly the sessions that leaked. The "session close exceeded
+    budget" error logged below is therefore the only signal that it happened —
+    alert on it rather than on the library's warning.
+
+    Correcting the flag would mean overriding _is_closed too, and aioweb_tls is
+    a third-party library: this class deliberately overrides one documented
+    lifecycle method and nothing private, so an upstream release has one narrow
+    way to break it. A stale flag on a session discarded the same instant is the
+    cheaper side of that trade.
     """
-    headers = headers if headers else {}
-    headers.setdefault("User-Agent", UA.random)
-    session = aiohttp.ClientSession(
-        headers=headers, cookies=cookies, connector=connector, timeout=timeout
-    )
-    session.proxy_pool = proxies
+
+    async def close(self) -> None:
         try:
-        yield session
-    finally:
-        await session.close()
+            async with asyncio.timeout(SESSION_CLOSE_TIMEOUT_SECONDS):
+                await super().close()
+        except TimeoutError:
+            LOGGER.error(
+                "session close exceeded budget, abandoning session",
+                extra={"timeout": SESSION_CLOSE_TIMEOUT_SECONDS},
+            )
 
 
-@asynccontextmanager
-async def get_tls_session(
+def get_default_session(
+    proxies: AioProxies,
     headers: dict | None = None,
     timeout: int = TIMEOUT_SECONDS,
-    proxies: AioProxies | None = None,
-) -> AsyncGenerator[AsyncSession, None]:
-    """Yield a curl_cffi session impersonating a random client fingerprint.
+    max_clients: int = MAX_CURL_HANDLES,
+    impersonate: BrowserTypeLiteral | None = None,
+) -> TLSSession:
+    """Return a TLSSession impersonating a random client fingerprint.
 
-    The proxy pool is stored on the session as proxy_pool for send_request to
-    rotate per request.
+    Fingerprint rotation lives here so every monitor rotates the same way; the
+    caller owns the session lifetime via ``async with``.
+
+    Cookies are not set here: aioweb has no constructor cookie path, and
+    monitors usually obtain them at runtime, so call session.set_cookie(name,
+    value, domain=...) where they are fetched (e.g. a pre_task_func) instead.
 
     Args:
-        headers: Headers applied to the session; a Chrome User-Agent is
-            defaulted if absent.
-        timeout: Request timeout in seconds.
-        proxies: Proxy pool attached to the session.
+        proxies: Pool attached to the session as proxy_pool, which send_request
+            rotates through per request. Required so a session can never send
+            un-proxied.
+        headers: Default headers merged into every request by aioweb. Use for
+            monitor-wide static headers; per-request headers still go through
+            send_request. The curl_cffi impersonation already supplies a
+            matching User-Agent, so one is not defaulted here.
+        timeout: Total request timeout in seconds, defaulting to TIMEOUT_SECONDS.
+            aioweb's own default is only 10s, so this restores the wider budget
+            the old sessions used and slow retailers depend on.
+        max_clients: Size of the curl handle pool, defaulting to
+            MAX_CURL_HANDLES. This is a hard concurrency ceiling: a request
+            finding the pool empty waits in pop_curl's untimed pool.get(), and
+            `timeout` does not cover that wait, only the transfer after
+            checkout. Keep it at or above the monitor's real fan-out (the sum
+            of its pagination semaphores plus any unsemaphored per-cycle
+            requests) so the wait is never reachable.
+        impersonate: A curl_cffi profile to pin instead of drawing one from
+            CLIENT_IDENTIFIERS, for a retailer whose bot protection only
+            tolerates some profiles from a given IP (B&H's Cloudflare refuses
+            every chrome profile on a flagged IP but still accepts safari).
 
-    Yields:
-        The configured curl_cffi session.
+    Returns:
+        A BoundedCloseSession (a TLSSession whose close is bounded) over a
+        curl_cffi backend with a random impersonation, with the proxy pool
+        attached as proxy_pool.
     """
-    headers = headers if headers else {}
-    headers.setdefault("User-Agent", UA.chrome)
-    async with AsyncSession(
-        impersonate=random.choice(CLIENT_IDENTIFIERS),
-        timeout=timeout,
+    impersonate = impersonate or random.choice(CLIENT_IDENTIFIERS)
+    session = BoundedCloseSession(
+        backend=CurlCffi(impersonate=impersonate),
         headers=headers,
-    ) as session:
+        timeout=timeout,
+        max_clients=max_clients,
+    )
     session.proxy_pool = proxies
-        yield session
+    return session
 
 
-def build_session_details(
-    session: aiohttp.ClientSession | AsyncSession, headers: dict = None
-):
-    """Merge headers into an existing session, defaulting a random User-Agent.
-
-    Args:
-        session: Session whose headers are updated in place.
-        headers: Headers to merge; User-Agent is defaulted if absent.
-    """
-    headers = headers or {}
-    headers.setdefault("User-Agent", UA.random)
-    session.headers.update(headers)
-
-
-@retry(
-    stop=stop_after_attempt(RETRY_LIMIT),
-    wait=wait_fixed(RETRY_WAIT),
-    retry=retry_if_exception_type(
-        (
-            aiohttp.ClientError,
-            asyncio.TimeoutError,
-            RequestsError,
-            requests.Timeout,
-            CurlTimeout,
-        )
-    ),
-    before=before_log(LOGGER, logging.DEBUG),
-    after=after_log(LOGGER, logging.DEBUG),
-    reraise=True,
-)
 async def send_request(
-    session: aiohttp.ClientSession | AsyncSession,
+    session: TLSSession,
     url: str,
     method: str,
+    *,
     params: dict | None = None,
     json_data: dict | None = None,
     data: dict | None = None,
-    reraise_json_error: bool = False,
-    development_proxy: str | None = None,
-) -> dict | str:
-    """Send a request through either session type with retries and proxy rotation.
+    headers: dict | None = None,
+) -> Response:
+    """Send a request through a TLSSession with retries and proxy rotation.
 
-    Retries transport errors and timeouts up to RETRY_LIMIT with a fixed wait.
-    When development_proxy is set it overrides the session's proxy pool and
-    disables TLS verification for local traffic inspection; otherwise the next
-    proxy is drawn from the session's proxy_pool when present. Responses are
-    parsed as JSON when possible.
+    Wraps aioweb's request_with_retries, which retries transport errors,
+    timeouts, and retryable statuses (429/5xx) up to RETRY_LIMIT attempts with
+    exponential backoff. The next proxy is drawn from the pool per call and
+    applied to both schemes. A non-2xx response or a fully failed request raises
+    AiowebError, an expected task error, so a failed fetch surfaces the same way
+    the old HTTPError path did. On success the aioweb Response is returned for
+    the caller to read via response.json(), response.text(), etc.
 
     Args:
-        session: aiohttp or curl_cffi session to send through.
+        session: The TLSSession to send through.
         url: Request URL.
         method: HTTP method.
         params: Query parameters.
-        json_data: JSON body.
-        data: Form body.
-        reraise_json_error: Raise JSONDecodeError instead of returning raw
-            text when the response is not valid JSON.
-        development_proxy: Local proxy URL that overrides the pool and
-            disables verification.
+        json_data: JSON body. A dict here is sent as JSON with the
+            application/json content-type set by aioweb.
+        data: Raw request body, passed through unchanged (e.g. a url-encoded
+            form string). The caller owns the matching content-type via headers.
+            A dict is JSON-encoded by aioweb, so use json_data for JSON bodies.
+        headers: Per-request headers merged over the session defaults.
 
     Returns:
-        Parsed JSON dict, or raw response text when parsing fails and
-        reraise_json_error is False.
+        The aioweb Response, guaranteed 2xx (non-2xx raises). Read the body with
+        response.json() (dict/list, or None if not valid JSON) or
+        response.text().
 
     Raises:
-        json.JSONDecodeError: When parsing fails and reraise_json_error is
-            True.
+        AiowebError: On a non-2xx response or after all retries fail.
     """
-    verify = True
-    if development_proxy:
-        proxy = development_proxy
-        verify = False
-    else:
-        manager = getattr(session, "proxy_pool", None)
-        proxy = manager.next().url() if manager else None
 
-    if isinstance(session, aiohttp.ClientSession):
-        async with session.request(
-            method=method,
-            url=url,
+    proxy_url = session.proxy_pool.next().url()
+    proxy = {"http": proxy_url, "https": proxy_url}
+
+    body = json_data if json_data is not None else data
+    response = await session.request_with_retries(
+        method.upper(),
+        url,
         params=params,
-            json=json_data,
-            data=data,
-            proxy=proxy,
-            ssl=verify,
-        ) as response:
-            response.raise_for_status()
-            text = await response.text()
-    else:
-        response = await session.request(
-            method=method.upper(),
-            url=url,
-            params=params,
-            json=json_data,
-            data=data,
-            proxy=proxy,
-            verify=verify,
+        data=body,
+        headers=headers,
+        proxies=proxy,
+        attempts=RETRY_LIMIT,
     )
     response.raise_for_status()
-        text = response.text
+    return response
 
-    try:
-        return json.loads(text)
-    except json.JSONDecodeError:
-        if reraise_json_error:
-            raise
 
-        return text
+def get_response_json(
+    response: Response,
+    *,
+    logger: logging.Logger = LOGGER,
+    expected: type | tuple[type, ...] = dict,
+    **context,
+) -> dict | list | None:
+    """Parse a response body as JSON, or log it and return None.
+
+    aioweb's response.json() yields None for a non-JSON body, so a single
+    isinstance check covers both the empty/invalid case and a body of the
+    wrong shape. Callers stop on None the same way they did with the inline
+    check this replaces.
+
+    Args:
+        response: A 2xx aioweb Response from send_request.
+        logger: Logger the warning is emitted on; pass the monitor's own so
+            the record stays attributed to that monitor.
+        expected: Type(s) the parsed body must be an instance of.
+        **context: Extra fields for the warning record (category, page, ...).
+
+    Returns:
+        The parsed body when it matches `expected`, else None.
+    """
+
+    data = response.json()
+    if isinstance(data, expected):
+        return data
+
+    logger.warning(
+        "unexpected response",
+        extra={
+            **context,
+            "body_type": type(data).__name__,
+            "body": str(data)[:200],
+        },
+    )
+    return None

tests/test_alert_priority.py

@@ -0,0 +1,85 @@
+"""unit test: calculate_priority ranks MSRP-less products instead of burying them
+
+run from the repo root:
+    uv run --with pytest pytest tests/test_alert_priority.py
+
+pure and offline — passes plain dicts shaped like product documents; touches no
+Mongo. The claim query sorts priority descending, so a priority of 0 means an
+alert is only drained once everything else is.
+"""
+
+from monitor_utils.database import (
+    NEW_LISTING_PRIORITY,
+    NO_MSRP_PRIORITY,
+    calculate_priority,
+)
+
+
+def document(current, original, **overrides) -> dict:
+    doc = {"_id": "p1", "currentPrice": current, "originalPrice": original}
+    doc.update(overrides)
+    return doc
+
+
+def test_no_msrp_takes_its_own_rank():
+    assert calculate_priority(document(50.0, 50.0, hasMsrp=False)) == NO_MSRP_PRIORITY
+
+
+def test_no_msrp_rank_sits_directly_under_the_seventy_band():
+    """It must outrank every sub-70% product and lose to every 70%+ one."""
+    assert calculate_priority(document(25.0, 100.0)) == 7  # 75% off
+    assert NO_MSRP_PRIORITY < 7
+    assert NO_MSRP_PRIORITY > calculate_priority(document(45.0, 100.0))  # 55% off
+
+
+def test_msrp_products_are_unaffected():
+    assert calculate_priority(document(25.0, 100.0, hasMsrp=True)) == 7
+    assert calculate_priority(document(50.0, 100.0, hasMsrp=True)) == 5
+
+
+def test_documents_without_the_field_keep_the_percentage_path():
+    """Documents written before hasMsrp existed must not all jump the queue."""
+    assert calculate_priority(document(50.0, 100.0)) == 5
+    assert calculate_priority(document(100.0, 100.0)) == 0
+
+
+def test_broken_prices_still_outrank_nothing():
+    """A missing or zero original price is a data defect, not a missing MSRP."""
+    assert calculate_priority(document(50.0, None, hasMsrp=False)) == 0
+    assert calculate_priority(document(50.0, 0, hasMsrp=False)) == 0
+    assert calculate_priority(document(None, 100.0, hasMsrp=False)) == 0
+
+
+def test_non_finite_prices_resolve_to_zero_instead_of_raising():
+    """float("Infinity") from a retailer field once crashed the whole monitor."""
+    inf, nan = float("inf"), float("nan")
+    assert calculate_priority(document(50.0, inf)) == 0
+    assert calculate_priority(document(inf, 100.0)) == 0
+    assert calculate_priority(document(-inf, 100.0)) == 0
+    assert calculate_priority(document(nan, 100.0)) == 0
+    assert calculate_priority(document(50.0, nan)) == 0
+    assert calculate_priority(document(50.0, inf, newListing=True)) == 0
+
+
+def test_a_discount_too_large_to_represent_resolves_to_zero():
+    """Finite prices can still overflow the ratio (a subnormal original)."""
+    assert calculate_priority(document(-1e308, 5e-324)) == 0
+    assert calculate_priority(document(1e308, 5e-324)) == 0
+
+
+def test_new_listing_is_floored_not_buried():
+    """A low-discount new listing sends regardless, so it must not sort last."""
+    assert (
+        calculate_priority(document(95.0, 100.0, newListing=True))
+        == NEW_LISTING_PRIORITY
+    )
+
+
+def test_new_listing_keeps_a_higher_percentage_rank():
+    """The floor is a floor: a 75%-off new listing still ranks as 75% off."""
+    assert calculate_priority(document(25.0, 100.0, newListing=True)) == 7
+
+
+def test_repeat_alerts_keep_the_percentage_path():
+    assert calculate_priority(document(95.0, 100.0, newListing=False)) == 0
+    assert calculate_priority(document(95.0, 100.0)) == 0

tests/test_build_product_details.py

@@ -0,0 +1,243 @@
+"""unit test: build_product_details thresholding, promo read, and combined_discount
+
+run from the repo root:
+    uv run pytest tests/test_build_product_details.py
+
+pure and offline — no Mongo. Builds a Product from a raw doc dict and checks the
+discount threshold (with percentage codes compounding onto the sale price), the
+promo fields read through via .get, and the combined_discount helper.
+"""
+
+from datetime import datetime, timezone
+
+from monitor_utils.runner import build_product_details, combined_discount
+
+
+def _doc(**overrides) -> dict:
+    """A well-formed product doc; override prices/promo per test."""
+    doc = {
+        "_id": "p1_tools",
+        "title": "Widget",
+        "url": "http://x/p1",
+        "currentPrice": 80.0,
+        "originalPrice": 100.0,
+        "imageUrl": "http://img/1",
+        "customizationIndex": None,
+        "memberNonMemberComparison": None,
+        "overstockText": None,
+        "discoveredDate": datetime(2026, 7, 26, tzinfo=timezone.utc),
+        "startDate": None,
+        "endDate": None,
+        "priceBeforeDrop": None,
+        "eduPrice": None,
+        "pastPrices": [
+            {"price": 80.0, "date": datetime(2026, 7, 26, tzinfo=timezone.utc)}
+        ],
+        "available": True,
+        "condition": None,
+        "appExclusive": None,
+        "memberPick": None,
+        "memberDeal": None,
+    }
+    doc.update(overrides)
+    return doc
+
+
+# ------------------------------------------------------------ combined_discount
+
+
+def test_combined_discount_no_promo_is_base():
+    assert combined_discount(24, None) == 24
+    assert combined_discount(24, 0) == 24
+
+
+def test_combined_discount_compounds_not_adds():
+    # 20% off then extra 30% code -> 44% off list (customer pays $56), not 50%
+    assert combined_discount(20, 30) == 44
+
+
+def test_combined_discount_rounds_to_nearest():
+    # 24% sale + 22% code: 1 - .76*.78 = .4072 -> 41 (rounded, not truncated 40)
+    assert combined_discount(24, 22) == 41
+
+
+# --------------------------------------------------------- build_product_details
+
+
+def test_promo_fields_read_through():
+    doc = _doc(
+        promoCode="BIRTHDAY26",
+        promoPercentage=22,
+        promoDollarOff=None,
+        promoMinSpend=None,
+    )
+    product = build_product_details(doc, min_percentage=0)
+    assert product.metadata.promo_code == "BIRTHDAY26"
+    assert product.metadata.promo_percentage == 22
+    # base percentage is the raw price delta, unchanged by the code
+    assert product.metadata.percentage == 20
+
+
+def test_asin_reads_through():
+    """asin round-trips: convert_product_to_mongo writes it, so the consumer
+    must read it back for the SellerAmp lookup"""
+    product = build_product_details(_doc(asin="B0CHX3QBCH"), min_percentage=0)
+    assert product.metadata.asin == "B0CHX3QBCH"
+
+
+def test_missing_asin_defaults_to_none():
+    assert build_product_details(_doc(), min_percentage=0).metadata.asin is None
+
+
+def test_missing_promo_keys_default_to_none():
+    """a doc from a lagging monitor lacks promo keys -> None, not KeyError"""
+    product = build_product_details(_doc(), min_percentage=0)
+    assert product.metadata.promo_code is None
+    assert product.metadata.promo_percentage is None
+
+
+def test_percentage_code_lifts_item_over_threshold():
+    # 30% base fails min 40, but +22% code -> 45% combined clears it
+    doc = _doc(currentPrice=70.0, originalPrice=100.0, promoPercentage=22)
+    assert build_product_details(doc, min_percentage=40) is not None
+
+
+def test_below_threshold_without_code_is_skipped():
+    doc = _doc(currentPrice=70.0, originalPrice=100.0)  # 30%, no code
+    assert build_product_details(doc, min_percentage=40) is None
+
+
+def test_dollar_code_does_not_affect_threshold():
+    # a dollar code has no promoPercentage, so 30% base stays below min 40
+    doc = _doc(
+        currentPrice=70.0,
+        originalPrice=100.0,
+        promoCode="STORE22",
+        promoDollarOff=22,
+        promoMinSpend=100,
+    )
+    assert build_product_details(doc, min_percentage=40) is None
+
+
+# ----------------------------------------------------------------- pastPrices
+
+
+def test_past_prices_read_through():
+    """Price history passes through unchanged."""
+
+    history = [
+        {"price": 95.0, "date": datetime(2026, 7, 20, tzinfo=timezone.utc)},
+        {"price": 90.0, "date": datetime(2026, 7, 24, tzinfo=timezone.utc)},
+    ]
+
+    product = build_product_details(_doc(pastPrices=history), min_percentage=0)
+
+    assert product.pricing.past_prices == history
+
+
+# ----------------------------------------------------------- bypass_thresholds
+
+
+def test_bypass_builds_a_below_threshold_product():
+    """A new listing sends at any discount when the caller says so."""
+
+    doc = _doc(currentPrice=95.0, originalPrice=100.0)  # 5%, way under 40
+    product = build_product_details(doc, min_percentage=40, bypass_thresholds=True)
+    assert product is not None
+    assert product.metadata.percentage == 5
+
+
+def test_bypass_defaults_off():
+    doc = _doc(currentPrice=95.0, originalPrice=100.0)
+    assert build_product_details(doc, min_percentage=40) is None
+
+
+def test_bypass_does_not_rescue_broken_prices():
+    """The missing-price skip is a data guard, not a threshold."""
+
+    doc = _doc(currentPrice=95.0, originalPrice=None)
+    assert build_product_details(doc, min_percentage=40, bypass_thresholds=True) is None
+
+
+# --------------------------------------------------------------- newListing
+
+
+def test_new_listing_flag_reads_through():
+    product = build_product_details(_doc(newListing=True), min_percentage=0)
+    assert product.metadata.new_listing is True
+
+
+def test_legacy_docs_without_the_flag_build_as_none():
+    """Documents written before newListing existed must not KeyError."""
+
+    product = build_product_details(_doc(), min_percentage=0)
+    assert product.metadata.new_listing is None
+
+
+# ------------------------------------------------------------ msrp-less guard
+
+
+def test_msrpless_doc_with_missing_original_price_skips_not_crashes():
+    """Regression: the guard lived inside the has_msrp branch, so this doc
+    shape raised ValidationError (not an expected error) and killed the
+    consumer."""
+
+    doc = _doc(hasMsrp=False, originalPrice=None)
+    assert build_product_details(doc, min_percentage=40) is None
+
+
+def test_msrpless_doc_with_zero_original_price_skips():
+    doc = _doc(hasMsrp=False, originalPrice=0)
+    assert build_product_details(doc, min_percentage=40) is None
+
+
+def test_bypass_does_not_rescue_msrpless_broken_prices():
+    doc = _doc(hasMsrp=False, originalPrice=None)
+    assert build_product_details(doc, min_percentage=40, bypass_thresholds=True) is None
+
+
+def test_msrpless_doc_with_prices_builds_unthresholded():
+    """A priced MSRP-less doc still skips the threshold gate entirely."""
+
+    doc = _doc(hasMsrp=False, currentPrice=95.0, originalPrice=100.0)  # 5% < 40
+    product = build_product_details(doc, min_percentage=40)
+    assert product is not None
+    assert product.pricing.has_msrp is False
+
+
+# ------------------------------------------------------------ non-finite prices
+
+
+def test_infinite_price_skips_instead_of_raising():
+    """Regression (2026-10-09 06:25Z): float("Infinity") from a retailer field
+    raised OverflowError here and stopped the consumer's API deal loop."""
+
+    inf = float("inf")
+    assert build_product_details(_doc(originalPrice=inf), min_percentage=40) is None
+    assert build_product_details(_doc(currentPrice=inf), min_percentage=40) is None
+    assert build_product_details(_doc(currentPrice=-inf), min_percentage=40) is None
+
+
+def test_nan_price_skips_instead_of_raising():
+    nan = float("nan")
+    assert build_product_details(_doc(currentPrice=nan), min_percentage=40) is None
+    assert build_product_details(_doc(originalPrice=nan), min_percentage=40) is None
+
+
+def test_msrpless_infinite_price_skips_rather_than_reaching_the_payload():
+    """Without the skip the MSRP-less path would carry inf into the Product."""
+
+    doc = _doc(hasMsrp=False, currentPrice=float("inf"))
+    assert build_product_details(doc, min_percentage=40) is None
+
+
+def test_bypass_does_not_rescue_a_non_finite_price():
+    doc = _doc(originalPrice=float("inf"))
+    assert build_product_details(doc, min_percentage=40, bypass_thresholds=True) is None
+
+
+def test_unrepresentable_discount_skips_instead_of_raising():
+    """Finite prices can still overflow the ratio (a subnormal original)."""
+
+    doc = _doc(currentPrice=-1e308, originalPrice=5e-324)
+    assert build_product_details(doc, min_percentage=40) is None

tests/test_convert_product_to_mongo.py

@@ -0,0 +1,103 @@
+"""unit test: convert_product_to_mongo flattens a Product to its Mongo document
+
+run from the repo root:
+    uv run pytest tests/test_convert_product_to_mongo.py
+
+pure and offline — no Mongo. Also pins the fact that prices are already floats
+by the time they reach the flattener (pydantic coerces numeric strings on the
+Product), which is why convert_product_to_mongo no longer coerces them itself.
+"""
+
+from datetime import datetime
+
+from monitor_utils.database import convert_product_to_mongo
+from monitor_utils.models import PricingInfo, Product, ProductMetadata
+
+
+def _product(metadata: ProductMetadata | None = None, **pricing) -> Product:
+    return Product(
+        product_id="p1",
+        title="Widget",
+        url="http://x/p1",
+        pricing=PricingInfo(**pricing),
+        metadata=metadata or ProductMetadata(),
+    )
+
+
+def test_numeric_string_prices_arrive_as_floats():
+    """a numeric-string price is a float in the doc (pydantic coerced it on init)"""
+
+    doc = convert_product_to_mongo(_product(original_price="200", current_price=150.0))
+    assert doc["originalPrice"] == 200.0
+    assert isinstance(doc["originalPrice"], float)
+    assert doc["currentPrice"] == 150.0
+    assert isinstance(doc["currentPrice"], float)
+
+
+def test_lastupdated_is_stamped():
+    """every document gets a UTC lastUpdated timestamp"""
+
+    doc = convert_product_to_mongo(_product(original_price=200.0, current_price=150.0))
+    assert isinstance(doc["lastUpdated"], datetime)
+
+
+def test_core_fields_map_through():
+    """title and url pass straight through to the document"""
+
+    doc = convert_product_to_mongo(_product(original_price=200.0, current_price=150.0))
+    assert doc["title"] == "Widget"
+    assert doc["url"] == "http://x/p1"
+
+
+def test_promo_and_affiliate_fields_serialize():
+    """promo specifics + affiliate url reach the document under camelCase keys"""
+
+    doc = convert_product_to_mongo(
+        _product(
+            metadata=ProductMetadata(
+                affiliate_url="http://aff/p1",
+                promo_code="BIRTHDAY26",
+                promo_percentage=22,
+                promo_dollar_off=None,
+                promo_min_spend=None,
+            ),
+            original_price=200.0,
+            current_price=150.0,
+        )
+    )
+    assert doc["affiliateUrl"] == "http://aff/p1"
+    assert doc["promoCode"] == "BIRTHDAY26"
+    assert doc["promoPercentage"] == 22
+    assert doc["promoDollarOff"] is None
+    assert doc["promoMinSpend"] is None
+
+
+def test_dollar_promo_fields_serialize():
+    """a spend-threshold code stores amount + min spend, not a percentage"""
+
+    doc = convert_product_to_mongo(
+        _product(
+            metadata=ProductMetadata(
+                promo_code="STORE22",
+                promo_dollar_off=22,
+                promo_min_spend=100,
+            ),
+            original_price=200.0,
+            current_price=150.0,
+        )
+    )
+    assert doc["promoCode"] == "STORE22"
+    assert doc["promoDollarOff"] == 22
+    assert doc["promoMinSpend"] == 100
+    assert doc["promoPercentage"] is None
+
+
+def test_promo_fields_default_to_none():
+    """a product with no promo still carries the keys as None (uniform doc shape)"""
+
+    doc = convert_product_to_mongo(_product(original_price=200.0, current_price=150.0))
+    for key in ("promoCode", "promoPercentage", "promoDollarOff", "promoMinSpend"):
+        assert doc[key] is None
+    # percentage and channelId stay OUT of the doc on purpose
+    assert "percentage" not in doc
+    assert "channelId" not in doc

tests/test_fan_out_group.py

@@ -0,0 +1,126 @@
+"""unit test: fan_out_group binds sibling tasks and surfaces errors unwrapped
+
+run from the repo root:
+    uv run pytest tests/test_fan_out_group.py
+
+offline — no network. These lock the contract monitors depend on: a bare
+TaskGroup always raises an ExceptionGroup, which run_monitor_cycle would
+misclassify (it matches neither the EXPECTED_TASK_ERRORS branch nor the
+unexpected-Exception branch as intended), so the unwrapping is load-bearing and
+not cosmetic.
+"""
+
+import asyncio
+
+import aiohttp
+import pytest
+
+from monitor_utils.runner import fan_out_group
+
+
+async def _boom(error: BaseException):
+    raise error
+
+
+def test_child_error_surfaces_unwrapped():
+    """A failing child arrives as itself, not wrapped in an ExceptionGroup."""
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(_boom(RuntimeError("child")))
+            await asyncio.sleep(0)
+
+    with pytest.raises(RuntimeError) as exc_info:
+        asyncio.run(scenario())
+
+    assert not isinstance(exc_info.value, BaseExceptionGroup)
+    assert str(exc_info.value) == "child"
+
+
+def test_body_error_surfaces_unwrapped():
+    """An error raised by the caller's own body propagates unchanged."""
+
+    async def ok():
+        await asyncio.sleep(0)
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(ok())
+            raise ValueError("body")
+
+    with pytest.raises(ValueError) as exc_info:
+        asyncio.run(scenario())
+
+    assert not isinstance(exc_info.value, BaseExceptionGroup)
+
+
+def test_unexpected_error_is_preferred_over_an_expected_one():
+    """first_error must not let a routine network error mask a real bug."""
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(_boom(aiohttp.ClientError("routine")))
+            await asyncio.sleep(0)
+            raise ValueError("real bug")
+
+    with pytest.raises(ValueError):
+        asyncio.run(scenario())
+
+
+def test_early_return_still_awaits_children():
+    """A `return` inside the body must not abandon an in-flight child task."""
+
+    finished = []
+
+    async def child():
+        await asyncio.sleep(0.01)
+        finished.append("done")
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(child())
+            return "returned"
+
+    assert asyncio.run(scenario()) == "returned"
+    assert finished == ["done"]
+
+
+def test_parked_sibling_is_cancelled_when_the_body_fails():
+    """The binding is the point: a child that never finishes must not hang.
+
+    Without the group, a create_task + `finally: await` shape would park here
+    forever. The deadline is the assertion — a hang fails this test.
+    """
+
+    async def parked():
+        await asyncio.Event().wait()
+
+    async def scenario():
+        async with asyncio.timeout(2):
+            async with fan_out_group() as group:
+                group.create_task(parked())
+                await asyncio.sleep(0)
+                raise RuntimeError("body")
+
+    with pytest.raises(RuntimeError):
+        asyncio.run(scenario())
+
+
+def test_children_run_concurrently_with_the_body():
+    """Both siblings make progress; the group is not a sequencing device."""
+
+    order = []
+
+    async def child():
+        order.append("child-start")
+        await asyncio.sleep(0.01)
+        order.append("child-end")
+
+    async def scenario():
+        async with fan_out_group() as group:
+            group.create_task(child())
+            order.append("body")
+            await asyncio.sleep(0.02)
+
+    asyncio.run(scenario())
+    assert order == ["body", "child-start", "child-end"]

tests/test_fetch_all_pages.py

@@ -0,0 +1,43 @@
+"""unit test: fetch_all_pages result accounting for any iterable input
+
+run from the repo root:
+    uv run pytest tests/test_fetch_all_pages.py
+
+pure and offline — no HTTP. Regression: a generator page_range was exhausted
+building the tasks, so the result pairing dropped every failure and swallowed
+unexpected exceptions.
+"""
+
+import asyncio
+
+import pytest
+from aioweb import AiowebError
+
+from monitor_utils.runner import fetch_all_pages
+
+
+async def _failing_fetch(page: int):
+    raise AiowebError(f"page {page} failed")
+
+
+async def _crashing_fetch(page: int):
+    raise ValueError(f"page {page} crashed")
+
+
+def test_generator_input_still_reports_failed_pages():
+    failed = asyncio.run(
+        fetch_all_pages((p for p in [2, 3]), _failing_fetch, "cat", "page")
+    )
+    assert failed == [2, 3]
+
+
+def test_generator_input_still_raises_unexpected_errors():
+    with pytest.raises(ValueError):
+        asyncio.run(fetch_all_pages((p for p in [2]), _crashing_fetch, "cat", "page"))
+
+
+def test_range_input_reports_failed_pages():
+    """The re-iterable path the in-repo caller uses keeps working."""
+
+    failed = asyncio.run(fetch_all_pages(range(2, 4), _failing_fetch, "cat", "page"))
+    assert failed == [2, 3]

tests/test_get_response_json.py

@@ -0,0 +1,80 @@
+"""unit tests: sessions.get_response_json parses a body or logs and returns None
+
+run from the repo root:
+    uv run pytest tests/test_response_json.py
+
+offline — the aioweb Response is faked at the leaf; nothing is sent.
+"""
+
+import logging
+
+import pytest
+
+from monitor_utils.sessions import get_response_json
+
+
+class FakeResponse:
+    def __init__(self, body):
+        self.body = body
+
+    def json(self):
+        return self.body
+
+
+def warnings(caplog):
+    return [r for r in caplog.records if r.levelno == logging.WARNING]
+
+
+def test_dict_body_is_returned_without_logging(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse({"a": 1})) == {"a": 1}
+    assert warnings(caplog) == []
+
+
+def test_none_body_logs_and_returns_none(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse(None), category="c1", page=2) is None
+    (record,) = warnings(caplog)
+    assert record.message == "unexpected response"
+    assert record.body_type == "NoneType"
+    assert record.body == "None"
+    assert (record.category, record.page) == ("c1", 2)
+
+
+@pytest.mark.parametrize("body", [[1, 2], "text", 42, b"bytes"])
+def test_wrong_shape_logs_type_and_truncated_body(body, caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse(body)) is None
+    (record,) = warnings(caplog)
+    assert record.body_type == type(body).__name__
+    assert record.body == str(body)[:200]
+
+
+def test_body_preview_is_capped_at_200_chars(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        get_response_json(FakeResponse("x" * 1000))
+    (record,) = warnings(caplog)
+    assert len(record.body) == 200
+
+
+def test_expected_accepts_other_types(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse([1]), expected=list) == [1]
+        assert get_response_json(FakeResponse([1]), expected=(dict, list)) == [1]
+        assert get_response_json(FakeResponse({}), expected=list) is None
+    assert len(warnings(caplog)) == 1
+
+
+def test_warning_goes_to_the_passed_logger(caplog):
+    with caplog.at_level(logging.WARNING):
+        get_response_json(
+            FakeResponse(None), logger=logging.getLogger("woot.woot_monitor")
+        )
+    (record,) = warnings(caplog)
+    assert record.name == "woot.woot_monitor"
+
+
+def test_empty_dict_counts_as_a_dict(caplog):
+    with caplog.at_level(logging.WARNING, logger="monitor_utils.sessions"):
+        assert get_response_json(FakeResponse({})) == {}
+    assert warnings(caplog) == []

tests/test_new_listing_stamp.py

@@ -0,0 +1,146 @@
+"""unit test: add_products_to_database stamps newListing and copies it onto alerts
+
+run from the repo root:
+    uv run pytest tests/test_new_listing_stamp.py
+
+offline — drives add_products_to_database against fake collections that record
+the UpdateOne ops instead of touching Mongo. The recorded ops are inspected via
+UpdateOne's private _doc/_filter, the same fields its own __eq__ compares.
+
+The stamp is the database layer's job: only it knows whether the doc already
+exists, so monitors never set the flag. The alert carries its own copy because
+the product doc is rewritten every cycle (its flag flips False one cycle after
+discovery) while a discovery alert may outlive that in a backed-up queue.
+"""
+
+import asyncio
+
+from monitor_utils.database import add_products_to_database, build_alert
+from monitor_utils.models import PricingInfo, Product
+
+from datetime import datetime, timezone
+
+NOW = datetime(2026, 8, 10, tzinfo=timezone.utc)
+
+
+class FakeCursor:
+    def __init__(self, docs):
+        self._docs = list(docs)
+
+    def __aiter__(self):
+        return self
+
+    async def __anext__(self):
+        if not self._docs:
+            raise StopAsyncIteration
+        return self._docs.pop(0)
+
+
+class FakeCollection:
+    """Records bulk_write ops; doubles as its own .database for alerts lookup."""
+
+    def __init__(self, existing_docs=(), name="products"):
+        self.name = name
+        self._existing = list(existing_docs)
+        self.bulk_writes = []
+        self.database = self
+        self.alerts = None
+
+    def __getitem__(self, name):
+        if self.alerts is None:
+            self.alerts = FakeCollection(name=name)
+        return self.alerts
+
+    def find(self, _filter, _projection):
+        return FakeCursor(self._existing)
+
+    async def bulk_write(self, ops, ordered=True):
+        self.bulk_writes.append(list(ops))
+
+
+def _product(product_id="p1", current=95.0, original=100.0) -> Product:
+    return Product(
+        product_id=product_id,
+        title="Widget",
+        url="http://x/p1",
+        pricing=PricingInfo(current_price=current, original_price=original),
+    )
+
+
+def run(collection, products):
+    asyncio.run(
+        add_products_to_database(products, collection, "tools", "bestbuy")
+    )
+
+
+def product_update(collection):
+    return collection.bulk_writes[0][0]._doc
+
+
+def alert_doc(collection):
+    return collection.alerts.bulk_writes[0][0]._doc["$setOnInsert"]
+
+
+def test_insert_stamps_new_listing_true_on_doc_and_alert():
+    collection = FakeCollection()
+    run(collection, [_product()])
+
+    assert product_update(collection)["$set"]["newListing"] is True
+    assert alert_doc(collection)["newListing"] is True
+
+
+def test_existing_doc_with_new_price_stamps_false():
+    existing = {"_id": "p1", "currentPrice": 100.0, "pastPrices": []}
+    collection = FakeCollection([existing])
+    run(collection, [_product(current=95.0)])
+
+    assert product_update(collection)["$set"]["newListing"] is False
+    # a never-seen price still alerts, but not as a new listing
+    assert alert_doc(collection)["newListing"] is False
+
+
+def test_drop_alert_carries_price_before_drop():
+    """the doc's priceBeforeDrop is cleared by the next no-drop cycle, so the
+    alert keeps its own copy from emit time — a drop claimed late still knows
+    what it dropped from"""
+
+    existing = {"_id": "p1", "currentPrice": 100.0, "pastPrices": []}
+    collection = FakeCollection([existing])
+    run(collection, [_product(current=95.0)])
+
+    assert alert_doc(collection)["priceBeforeDrop"] == 100.0
+
+
+def test_non_drop_alert_carries_none():
+    """a never-seen higher price alerts without a drop; discoveries have no
+    prior price at all"""
+
+    existing = {"_id": "p1", "currentPrice": 100.0, "pastPrices": []}
+    collection = FakeCollection([existing])
+    run(collection, [_product(current=120.0)])
+    assert alert_doc(collection)["priceBeforeDrop"] is None
+
+    discovery = FakeCollection()
+    run(discovery, [_product()])
+    assert alert_doc(discovery)["priceBeforeDrop"] is None
+
+
+def test_existing_doc_with_known_price_emits_no_alert():
+    existing = {
+        "_id": "p1",
+        "currentPrice": 95.0,
+        "pastPrices": [{"price": 95.0, "date": NOW}],
+    }
+    collection = FakeCollection([existing])
+    run(collection, [_product(current=95.0)])
+
+    assert product_update(collection)["$set"]["newListing"] is False
+    assert collection.alerts is None
+
+
+def test_build_alert_defaults_missing_fields():
+    """Direct callers passing an unstamped document must not KeyError."""
+
+    alert = build_alert("bestbuy", {"_id": "p1"}, NOW)
+    assert alert["newListing"] is False
+    assert alert["priceBeforeDrop"] is None

tests/test_pagination_semaphore.py

@@ -0,0 +1,107 @@
+"""unit test: execute_pagination honours a caller-supplied shared semaphore
+
+run from the repo root:
+    uv run pytest tests/test_pagination_semaphore.py
+
+offline — no network. A fake page func records peak concurrency so we can
+assert the fan-out limit actually holds, both for the shared-semaphore path
+(one instance capping many categories at once) and a single call bounding one
+category with its own semaphore.
+"""
+
+import asyncio
+
+from monitor_utils.runner import execute_pagination
+
+
+def _make_tracking_func(state: dict):
+    """Return a page func that tracks peak simultaneous in-flight calls."""
+
+    async def func(session, offset, **kwargs):
+        state["live"] += 1
+        state["peak"] = max(state["peak"], state["live"])
+        await asyncio.sleep(0.01)
+        state["live"] -= 1
+
+    return func
+
+
+def test_shared_semaphore_caps_total_concurrency_across_categories():
+    """One shared semaphore bounds combined fan-out over concurrent categories.
+
+    Two categories paginate at once through the same semaphore(limit); without
+    sharing, each would build its own semaphore and the peak could reach twice
+    the limit. Peak must stay at or below the shared limit.
+    """
+
+    limit = 3
+    shared = asyncio.Semaphore(limit)
+    state = {"live": 0, "peak": 0}
+    func = _make_tracking_func(state)
+
+    async def run():
+        await asyncio.gather(
+            execute_pagination(
+                "catA",
+                total_products=300,
+                products_per_page=30,
+                func=func,
+                session=None,
+                semaphore=shared,
+            ),
+            execute_pagination(
+                "catB",
+                total_products=300,
+                products_per_page=30,
+                func=func,
+                session=None,
+                semaphore=shared,
+            ),
+        )
+
+    asyncio.run(run())
+    assert state["peak"] <= limit, f"peak {state['peak']} exceeded shared limit {limit}"
+
+
+def test_single_call_respects_its_semaphore():
+    """A single call bounds its own fan-out to the supplied semaphore limit."""
+
+    limit = 2
+    state = {"live": 0, "peak": 0}
+    func = _make_tracking_func(state)
+
+    async def run():
+        await execute_pagination(
+            "cat",
+            total_products=300,
+            products_per_page=30,
+            func=func,
+            session=None,
+            semaphore=asyncio.Semaphore(limit),
+        )
+
+    asyncio.run(run())
+    assert state["peak"] <= limit, f"peak {state['peak']} over limit {limit}"
+
+
+def test_omitted_semaphore_runs_unbounded():
+    """Omitting the semaphore lets all pages fan out concurrently.
+
+    With no semaphore the only bound is the page count; pages 2..10 (nine pages)
+    should all be in flight at once, so peak concurrency reaches nine.
+    """
+
+    state = {"live": 0, "peak": 0}
+    func = _make_tracking_func(state)
+
+    async def run():
+        await execute_pagination(
+            "cat",
+            total_products=300,
+            products_per_page=30,
+            func=func,
+            session=None,
+        )
+
+    asyncio.run(run())
+    assert state["peak"] == 9, f"expected all 9 pages concurrent, got {state['peak']}"

tests/test_price_before_drop.py

@@ -0,0 +1,25 @@
+"""unit test: resolve_price_before_drop detects a drop from the prior stored price
+
+run from the repo root:
+    uv run pytest tests/test_price_before_drop.py
+
+pure and offline — no Mongo, no product construction; just the drop rule
+add_products_to_database relies on
+"""
+
+from monitor_utils.database import resolve_price_before_drop
+
+
+def test_drop_returns_previous_price():
+    """a strictly lower new price is a drop; the prior price is returned"""
+
+    assert resolve_price_before_drop(200.0, 150.0) == 200.0
+
+
+def test_non_drops_return_none():
+    """no prior price, an unchanged price, or a rise are not drops"""
+
+    assert resolve_price_before_drop(None, 150.0) is None  # new product
+    assert resolve_price_before_drop(150.0, 150.0) is None  # unchanged
+    assert resolve_price_before_drop(150.0, 200.0) is None  # price rose
+    assert resolve_price_before_drop(200.0, None) is None  # missing new price

tests/test_run_bounded_cycle.py

@@ -0,0 +1,147 @@
+"""unit test: run_bounded_cycle session lifetime, budget, and pass-through
+
+run from the repo root:
+    uv run pytest tests/test_run_bounded_cycle.py
+
+offline — get_default_session is faked at the runner's namespace so no proxies
+or network are touched; run_monitor_cycle underneath runs real. Gap: the real
+session wiring (timeout/max_clients reaching curl) is covered by the monitors'
+own pool-size tests, not here.
+"""
+
+import asyncio
+import logging
+
+import aiohttp
+import pytest
+
+from monitor_utils import runner
+from monitor_utils.models import CycleResult
+from monitor_utils.runner import run_bounded_cycle
+
+
+class FakeSession:
+    def __init__(self):
+        self.closed = False
+
+    async def __aenter__(self):
+        return self
+
+    async def __aexit__(self, *exc):
+        self.closed = True
+
+
+def stub_session(monkeypatch) -> dict:
+    """Fake get_default_session, recording each call's kwargs and session."""
+    calls = []
+
+    def fake(proxies, headers=None, timeout=None, max_clients=None):
+        session = FakeSession()
+        calls.append(
+            {
+                "headers": headers,
+                "timeout": timeout,
+                "max_clients": max_clients,
+                "session": session,
+            }
+        )
+        return session
+
+    monkeypatch.setattr(runner, "get_default_session", fake)
+    return calls
+
+
+def test_returns_cycle_result_and_closes_session(monkeypatch):
+    calls = stub_session(monkeypatch)
+
+    async def task(category, session, collection, **kwargs):
+        pass
+
+    result = asyncio.run(run_bounded_cycle(["a", "b"], task, None, None))
+
+    assert result == CycleResult(successes=2, failures=0)
+    assert calls[0]["session"].closed
+
+
+def test_session_params_reach_the_factory(monkeypatch):
+    calls = stub_session(monkeypatch)
+
+    async def task(category, session, collection, **kwargs):
+        pass
+
+    asyncio.run(
+        run_bounded_cycle(
+            ["a"],
+            task,
+            None,
+            None,
+            headers={"referer": "https://x/"},
+            session_timeout=8,
+            max_clients=81,
+        )
+    )
+
+    assert calls[0]["headers"] == {"referer": "https://x/"}
+    assert calls[0]["timeout"] == 8
+    assert calls[0]["max_clients"] == 81
+
+
+def test_expired_budget_returns_none_not_raise(monkeypatch, caplog):
+    """A cycle past its budget is abandoned and logged, and the loop lives on."""
+
+    calls = stub_session(monkeypatch)
+
+    async def parked(category, session, collection, **kwargs):
+        await asyncio.Event().wait()
+
+    with caplog.at_level(logging.ERROR, logger="monitor_utils.runner"):
+        result = asyncio.run(
+            run_bounded_cycle(["a"], parked, None, None, cycle_timeout=0.05)
+        )
+
+    assert result is None
+    assert calls[0]["session"].closed
+    assert any(
+        r.message == "cycle exceeded budget, abandoning cycle" for r in caplog.records
+    )
+
+
+def test_expected_task_error_counts_as_failure(monkeypatch):
+    stub_session(monkeypatch)
+
+    async def flaky(category, session, collection, **kwargs):
+        raise aiohttp.ClientError("routine")
+
+    result = asyncio.run(run_bounded_cycle(["a"], flaky, None, None))
+
+    assert result == CycleResult(successes=0, failures=1)
+
+
+def test_unexpected_error_propagates_and_closes_session(monkeypatch):
+    """fail loud: a real bug must escape the helper, not vanish into a log."""
+
+    calls = stub_session(monkeypatch)
+
+    async def broken(category, session, collection, **kwargs):
+        raise RuntimeError("bug")
+
+    with pytest.raises(RuntimeError):
+        asyncio.run(run_bounded_cycle(["a"], broken, None, None))
+
+    assert calls[0]["session"].closed
+
+
+def test_each_call_opens_a_fresh_session(monkeypatch):
+    calls = stub_session(monkeypatch)
+
+    async def task(category, session, collection, **kwargs):
+        pass
+
+    async def two_cycles():
+        await run_bounded_cycle(["a"], task, None, None)
+        await run_bounded_cycle(["a"], task, None, None)
+
+    asyncio.run(two_cycles())
+
+    assert len(calls) == 2
+    assert calls[0]["session"] is not calls[1]["session"]

tests/test_session_close_budget.py

@@ -0,0 +1,135 @@
+"""unit test: a session's close cannot park the caller indefinitely
+
+run from the repo root:
+    uv run pytest tests/test_session_close_budget.py
+
+offline — no network. curl_cffi's close awaits the multi handle with no timeout
+of its own, so a wedged transfer would park whatever closes the session,
+including an `async with` exit. These build a real session and wedge the
+backend's close, asserting the bound lives on the session itself so every call
+site gets it without arranging anything.
+"""
+
+import asyncio
+import logging
+
+from aioproxies import AioProxies
+
+from monitor_utils import sessions
+from monitor_utils.sessions import BoundedCloseSession, get_default_session
+
+
+def _proxies(tmp_path) -> AioProxies:
+    """A one-entry pool in the host:port:user:pass format monitors ship."""
+    proxy_file = tmp_path / "proxies.txt"
+    proxy_file.write_text("proxy.example.io:8080:user:secret\n")
+    return AioProxies.from_file(path=str(proxy_file))
+
+
+async def _wedge_close(session) -> asyncio.Event:
+    """Replace the backend's close with one that never returns.
+
+    Builds the backend session first so TLSSession.close actually reaches the
+    backend rather than short-circuiting on an unbuilt session.
+    """
+
+    session.session  # force lazy construction
+    started = asyncio.Event()
+
+    async def wedged(inner_session):
+        started.set()
+        await asyncio.sleep(3600)
+
+    session.backend.close = wedged
+    return started
+
+
+async def _really_close(session, original_close):
+    """Close for real, so the test does not leak the curl handle it allocated."""
+    session.backend.close = original_close
+    await session.close()
+
+
+def test_close_is_abandoned_when_it_exceeds_the_budget(monkeypatch, caplog, tmp_path):
+    """A close that never returns is abandoned and logged, not awaited forever."""
+
+    monkeypatch.setattr(sessions, "SESSION_CLOSE_TIMEOUT_SECONDS", 0.05)
+    proxies = _proxies(tmp_path)
+
+    async def run():
+        session = get_default_session(proxies, timeout=8)
+        original_close = session.backend.close
+        started = await _wedge_close(session)
+
+        # wait_for is the assertion: an unbounded close never returns here
+        await asyncio.wait_for(session.close(), timeout=5)
+
+        assert started.is_set()
+        await _really_close(session, original_close)
+
+    with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"):
+        asyncio.run(run())
+
+    assert any(
+        record.message == "session close exceeded budget, abandoning session"
+        for record in caplog.records
+    )
+
+
+def test_async_with_exit_is_bounded(monkeypatch, caplog, tmp_path):
+    """The bound applies through `async with`, which is how monitors close."""
+
+    monkeypatch.setattr(sessions, "SESSION_CLOSE_TIMEOUT_SECONDS", 0.05)
+    proxies = _proxies(tmp_path)
+
+    async def run():
+        session = get_default_session(proxies, timeout=8)
+        original_close = session.backend.close
+        started = await _wedge_close(session)
+
+        async with session:
+            pass
+
+        assert started.is_set()
+        await _really_close(session, original_close)
+
+    with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"):
+        asyncio.run(asyncio.wait_for(run(), timeout=5))
+
+    assert any(
+        record.message == "session close exceeded budget, abandoning session"
+        for record in caplog.records
+    )
+
+
+def test_healthy_close_completes_and_is_not_logged(caplog, tmp_path):
+    """The normal path closes fully and reports nothing."""
+
+    proxies = _proxies(tmp_path)
+
+    async def run():
+        session = get_default_session(proxies, timeout=8)
+        session.session  # force lazy construction so close has work to do
+        await session.close()
+        return session
+
+    with caplog.at_level(logging.ERROR, logger="monitor_utils.sessions"):
+        session = asyncio.run(run())
+
+    assert session._is_closed()
+    assert not caplog.records
+
+
+def test_get_default_session_returns_a_bounded_session(tmp_path):
+    """The factory hands out the bounded type, not a bare TLSSession."""
+
+    proxies = _proxies(tmp_path)
+
+    async def run():
+        session = get_default_session(proxies, timeout=8)
+        try:
+            return isinstance(session, BoundedCloseSession)
+        finally:
+            await session.close()
+
+    assert asyncio.run(run())