fix: canonical port key, remove()/restore() drift, escape-aware templates

Port keys no longer drift on zero-padding (host:080 vs host:80 collapsed
one canonical key), so a get() -> burn() pairing on a padded port no
longer raises a false "not in pool". Constructor/from_file now dedupe by
canonical key like add(). remove() decrements the rotation cursor when
the removed slot precedes it, fixing a skip/double-serve. restore() on
an unknown key now warns instead of silently no-op, matching remove()'s
contract. A genuine timed burn() under cooldown>0 is now tracked
separately from routine cooldown resting, so it logs WARNING instead of
being buried as DEBUG. Source validation uses is-not-None instead of
truthiness, so a falsy-but-provided source (e.g. template="") is
accepted. Bare "{}" template normalization is now escape-aware, so an
intentional "{{}}" literal survives instead of being corrupted.

Stored-shape change: burn()/cooldown deadlines now write int(time.time())
instead of float; comparisons are unaffected.

Docstrings compressed (module + next()); no behavior change.

Signed-off-by: disqualifier <dev@disqualifier.me>

.gitignore

@@ -1,5 +1,5 @@
 # claude
-CLAUDE.md
+.claude/
 
 # python
 __pycache__/

README.md

@@ -1,33 +1,43 @@
 # aioproxies
 
-Proxy parsing, formatting, and source management. Renders proxies for
+Proxy parsing, formatting, health, and pool management. Renders proxies for
 aiohttp/aioweb, camoufox, and socks5; manages session templates (with
-caller-supplied fields like country/ttl), rotating lists, or a static proxy.
-**Credentials are always injected — never hardcoded.**
+caller-supplied fields like country/ttl), rotating lists, or a static proxy; and
+(for rotating lists) tracks burn/timeout, usage, reuse cooldown, and live pool
+edits. **Credentials are always injected — never hardcoded.**
 
 ## Install
 
 ```
-aioproxies @ git+ssh://git@git.rethinkstudios.io/rethink-public/aioproxies.git@v0.1.0
+aioproxies @ git+ssh://git@git.rethinkstudios.io/rethink-public/aioproxies.git@v0.3.0
 # network helpers (current_ip / reset) need the extra:
-aioproxies[net] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aioproxies.git@v0.1.0
+aioproxies[net] @ git+ssh://git@git.rethinkstudios.io/rethink-public/aioproxies.git@v0.3.0
 ```
 
 The core has no dependencies. The `net` extra adds `aiohttp` for `current_ip` /
 `reset`.
 
+Drop the `@v0.3.0` suffix from the line above to install the latest unpinned.
+
 ## Formatting
 
 ```python
 from aioproxies import parse
 
-p = parse("1.2.3.4:8080:user:pass")        # or "host:port"
+p = parse("1.2.3.4:8080:user:pass")        # or "host:port" for IP-authenticated proxies
 p.aiohttp()    # {"http": "...", "https": "..."}  -> aioweb ExtendedSession(proxies=)
 p.camoufox()   # {"server": "...", "username": ..., "password": ...}
 p.socks5()     # {"server": "socks5://...", ...}
 p.url()        # "http://user:pass@host:port"
+p.key()        # "1.2.3.4:8080:user:pass"  (canonical identity; "host:port" if auth-less)
 ```
 
+Auth-less (IP-authenticated) proxies are first-class: `"host:port"` parses and
+every render shape omits the credentials. The 4-part form splits on the first three
+colons, so a password may itself contain colons (`host:port:user:pa:ss:word`).
+`url()` / `aiohttp()` percent-encode the credentials, so reserved characters
+(`/ # ? @`) in a user or password still produce a valid URL.
+
 ## Sources
 
 Construct with exactly one source:
@@ -86,6 +96,97 @@ The credentials are baked in once with `.format()`; the per-call fields and `{se
 are double-braced (`{{country}}`) so they pass through that `.format()` untouched and
 remain for `next(**fields)` / the lib to fill.
 
+## Proxy health & pool management (rotating list source)
+
+For `proxies=` / `from_file` sources, the manager tracks each proxy's health and
+usage and lets you edit the pool live. (On `template=` / `static=` these methods are
+**no-ops that log a warning** and return cleanly — generic caller code can call them
+regardless of source.)
+
+```python
+from aioproxies import AioProxies, ProxiesExhaustedError
+
+pm = AioProxies(proxies=[...], cooldown=5)   # 5s reuse spacing; cooldown defaults to 0 (off)
+
+try:
+    proxy = pm.get()                          # next usable proxy, aiohttp dict
+    resp = await session.get(url, proxies=proxy)
+    if response_looks_blocked(resp):
+        pm.burn(proxy, 600)                   # time out 10 min ... or pm.burn(proxy) for dead
+except ProxiesExhaustedError:
+    ...                                       # whole pool permanently dead — back off / refetch
+
+pm.replace(fresh_batch)                       # swap in a new provider batch
+pm.stats()                                    # monitor uses + timeout state
+```
+
+### Selection
+
+Rotation is **sequential round-robin over usable proxies**:
+
+1. proxies that are fine (never burned, or a timed burn already expired) cycle in
+   order — same as v0.1.0.
+2. if none are fine but some are merely timed, the manager **warns** and hands out
+   the one recovering soonest (still counts a use).
+3. if every proxy is permanently dead (`-1`), `next()`/`get()` raise
+   `ProxiesExhaustedError`.
+
+`next()` still returns a `Proxy`; `get()` still returns an aiohttp dict.
+
+### Burn / restore
+
+```python
+pm.burn(proxy)            # dead/permanent (-1) — only manual restore() brings it back
+pm.burn(proxy, 600)       # timed — usable again automatically after 600s (lazy, no timers)
+pm.restore(proxy)         # clear any burn/timeout, back to fine
+pm.is_burned(proxy)       # current state (expired timed burns read False)
+```
+
+`burn`/`restore`/`is_burned`/`remove` accept **any proxy shape** — a spec string, a
+`Proxy`, an aiohttp/camoufox/socks5 dict, or a url — all resolve to the same canonical
+key (`host:port:user:pass`, or `host:port` auth-less; the port is normalized so
+`host:080` and `host:80` are one slot). The password is part of the key, so two proxies
+differing only by password are distinct slots. `burn` on a proxy not in the pool raises
+`ValueError`; `restore` on a proxy not in the pool logs a warning and no-ops (matching
+`remove`'s contract, as of v0.3.0 — previously it silently did nothing with no signal).
+
+### Cooldown
+
+`AioProxies(proxies=[...], cooldown=5)` spaces reuse: each handout times the proxy out
+for `cooldown` seconds so it isn't reused if avoidable. It is **soft** — under load
+(everything cooling) it falls through to the soonest-to-recover and never raises on
+cooldown alone. Default `0` = off (exact v0.1.0 behavior).
+
+### Stats
+
+```python
+pm.stats()        # [{"proxy": "h:p:u:pw", "uses": int, "state": "active"|"timed"|"dead",
+                  #   "timeout": <ts | -1 | None>}, ...]
+pm.reset_stats()  # zero all use counters; leave timeouts untouched
+```
+
+`uses` is a pure counter (every handout, including forced ones); it never drives
+selection and survives burns — a proxy can read "used 500× and dead". The `proxy`
+field is the full canonical spec (passwords included).
+
+### Live pool edits
+
+```python
+pm.replace(new_batch)                    # swap the whole list; wipes per-proxy state
+pm.replace(new_batch, keep_state=True)   # survivors keep uses/timeout; new ones start clean
+pm.add("h:p:u:pw")                       # append (single or list); skip exact-duplicate keys
+pm.remove(proxy)                         # drop a slot entirely (any shape) — distinct from burn
+```
+
+`replace` resets the rotation index and honors the manager's `shuffle` setting on the
+incoming list. `remove` differs from `burn`: burn = unusable but still tracked; remove =
+gone from the pool — removing a slot that precedes the rotation cursor adjusts the
+cursor so `next()` doesn't skip a proxy. Like the burn family, `add`/`replace` accept
+**any proxy shape** (spec/`Proxy`/url/aiohttp dict/camoufox/socks5 dict). The
+constructor and `from_file` also dedupe by canonical key, same as `add`.
+`canonical_key(shape)` and `to_proxy(shape)` are exported if you need the key or a
+normalized `Proxy` yourself.
+
 ## Network helpers (optional)
 
 ```python
@@ -103,3 +204,60 @@ await reset("https://provider/reset-url")    # rotate upstream ip
 - A missing proxy file raises, it does not exit the process.
 - Country/ASN tables, provider accounts, and reset URLs are project config —
   inject them; do not hardcode credentials in shared code.
+- aioweb integration is the manual loop shown above (get → use → burn on block).
+  A provider-protocol auto-rotation is a possible later enhancement, not in this lib.
+
+## Changelog
+
+### v0.3.0
+
+- **Stored-shape change: int unix deadlines.** `burn(proxy, seconds)` and the
+  cooldown timeout write `int(time.time()) + n` instead of a `float`. Comparisons
+  (`is_burned`/`stats`/selection) are unaffected — this only tightens what gets
+  persisted into per-proxy state.
+- **Port key normalization:** `key()` strips leading zeros from the port, so
+  `host:080` and `host:80` are the same canonical slot. Previously a zero-padded
+  port could break a `get()` → `burn()` pairing (the burn would raise "not in
+  pool" against the proxy that was just handed out).
+- **Constructor / `from_file` dedupe by canonical key**, matching `add()`. A
+  proxy list with repeated entries (e.g. same host:port:user:pass twice) no
+  longer inflates the pool or double-weights rotation.
+- **`remove()` no longer skews rotation.** Removing a proxy that precedes the
+  rotation cursor now decrements the cursor, so the next `next()` call doesn't
+  skip or double-serve a proxy.
+- **`restore()` on an unknown proxy now logs a warning and no-ops**, matching
+  `remove()`'s contract (previously it silently did nothing, with `burn()`
+  raising for the same precondition and `remove()` warning — `restore()` was the
+  odd one out).
+- **Genuine timed burns under `cooldown>0` now log at WARNING**, not DEBUG. A
+  real `burn(proxy, seconds)` is now tracked separately from the manager's own
+  cooldown resting, so it no longer gets buried as routine cooldown noise when
+  every proxy in the pool happens to also be cooling down.
+- **Source truthiness → presence.** The constructor's exactly-one-source check
+  now uses `is not None` instead of truthiness, so `template=""` (or another
+  falsy-but-explicitly-provided source) is accepted rather than silently
+  rejected as "no source given."
+- **Escape-aware bare-`{}` template normalization.** A template's bare `{}` is
+  still filled as the session slot, but an escaped `{{}}` (str.format's own
+  convention for a literal `{}` in the output) now survives untouched instead of
+  being corrupted into `{{session}}`.
+
+### v0.2.1
+
+- **Legible missing-template-field error:** a `template=` placeholder not supplied to
+  `next(**fields)` now raises a clear `ValueError` naming the field, instead of leaking
+  a bare `KeyError` from `str.format`.
+
+### v0.2.0
+
+- **Proxy health for rotating lists:** `burn`/`restore`/`is_burned` (dead `-1` vs
+  timed), `stats`/`reset_stats`, and the new `ProxiesExhaustedError` (all-dead pool).
+- **Cooldown:** new `cooldown=` constructor arg spaces reuse; default `0` = off.
+- **Live pool edits:** `replace` (with `keep_state=`), `add`, `remove`, keyed by a
+  canonical proxy key that accepts every input shape (incl. auth-less / IP-auth).
+- **`{session}` default is now 8-char alphanumeric** (was 10-digit numeric);
+  `session_len` default is `8`. Templates that set `session_len` explicitly are
+  unaffected by the length change; the charset is now alphanumeric regardless.
+- **Backward-compatible:** a v0.1.0-style manager (no burns, `cooldown=0`) behaves
+  byte-for-byte identically — sequential round-robin, `next()`→`Proxy`,
+  `get()`→aiohttp dict, never raises.

pyproject.toml

@@ -4,8 +4,8 @@ build-backend = "hatchling.build"
 
 [project]
 name = "aioproxies"
-version = "0.1.0"
-description = "proxy parsing, formatting, and source management for aiohttp/aioweb, camoufox, and socks5"
+version = "0.3.0"
+description = "proxy parsing, formatting, health, and pool management for aiohttp/aioweb, camoufox, and socks5"
 requires-python = ">=3.10"
 dependencies = []
 

src/aioproxies/__init__.py

@@ -4,9 +4,18 @@ renders proxies for aiohttp/aioweb, camoufox, and socks5; manages session
 templates (with caller-supplied fields like country/ttl), rotating lists, or a
 static proxy. credentials are always injected, never hardcoded.
 """
-from .manager import AioProxies, ProxyManager, aioproxies
-from .proxy import Proxy, parse
+from .manager import AioProxies, ProxiesExhaustedError, ProxyManager, aioproxies
+from .proxy import Proxy, canonical_key, parse, to_proxy
 
-__all__ = ["AioProxies", "ProxyManager", "aioproxies", "Proxy", "parse"]
+__all__ = [
+    "AioProxies",
+    "ProxyManager",
+    "aioproxies",
+    "ProxiesExhaustedError",
+    "Proxy",
+    "parse",
+    "canonical_key",
+    "to_proxy",
+]
 
-__version__ = "0.1.0"
+__version__ = "0.3.0"

src/aioproxies/manager.py

@@ -1,26 +1,50 @@
 """proxy source management: session templates, rotation, static.
 
-`AioProxies` is constructed with exactly one source and hands out `Proxy`
-objects. it carries no module-level globals (rotation state is per-instance) and
-never exits the process — a missing proxy file raises, it does not `sys.exit`.
+`AioProxies` is constructed with exactly one source and hands out `Proxy` objects.
+no module-level globals (rotation state is per-instance); a missing proxy file
+raises, never `sys.exit`.
 
 sources:
-- template: a format string with named placeholders. `{session}` is filled with
-  a fresh session id on each `next()`; any other placeholder (e.g. `{country}`,
-  `{ttl}`) is filled from keyword args passed to `next(**fields)`. a bare `{}` is
-  also accepted and treated as the session slot (back-compat with simple templates).
-- proxies:  a list of specs cycled round-robin
+- template: a format string. `{session}` is filled with a fresh id on each
+  `next()`; other placeholders (`{country}`, `{ttl}`, ...) come from
+  `next(**fields)`. a bare `{}` is treated as the session slot unless escaped as
+  `{{}}` (back-compat with simple templates).
+- proxies:  a list of specs cycled round-robin (deduped by canonical key)
 - static:   one fixed proxy
+
+proxy health (rotating list source only) — burn/timeout, usage counters, reuse
+cooldown, pool edits (replace/add/remove) — keyed by each proxy's canonical key
+(`host:port:user:pass`, or `host:port` auth-less; port is normalized so `:080`
+and `:80` collapse to one key). on template/static sources these are no-ops that
+log a warning and return, so generic caller code can call them regardless of
+source.
+
+per-proxy state (keyed by canonical key): `uses` is a pure counter, incremented on
+every handout, never drives selection, survives burns. `timeout` is the
+availability state — `None`/`0` fine, `-1` dead/permanent (manual restore only), a
+future unix ts = timed out until then. this core is sync: no timers, lazy expiry
+checked only against `time.time()` at call time. cooldown and timed burns share
+this field; durations only ever exist as call arguments, converted to `now +
+seconds` and discarded — a raw duration is never stored.
 """
 import logging
 import random
+import re
 import string
+import time
 from typing import Dict, List, Optional, Union
 
-from .proxy import Proxy, parse
+from .proxy import Proxy, canonical_key, parse, to_proxy
 
 log = logging.getLogger(__name__)
 
+_DEAD = -1
+_BARE_SESSION_SLOT = re.compile(r"(?<!\{)\{\}(?!\})")
+
+
+class ProxiesExhaustedError(Exception):
+    """raised by next() when every proxy in the pool is permanently dead (-1)"""
+
 
 class AioProxies:
     """hands out proxies from a template, a rotating list, or a static value"""
@@ -31,21 +55,52 @@ class AioProxies:
         template: Optional[str] = None,
         proxies: Optional[List[Union[str, Proxy]]] = None,
         static: Optional[Union[str, Proxy]] = None,
-        session_len: int = 10,
+        session_len: int = 8,
         shuffle: bool = True,
+        cooldown: int = 0,
     ):
-        sources = [s for s in (template, proxies, static) if s]
+        if proxies is not None and len(proxies) == 0:
+            raise ValueError("proxies list is empty; provide at least one proxy")
+        sources = [s for s in (template, proxies, static) if s is not None]
         if len(sources) != 1:
             raise ValueError("provide exactly one of: template, proxies, static")
-        # normalize a bare {} session slot to the named {session} form
-        self.template = template.replace("{}", "{session}") if template else None
+        self.template = self._normalize_template(template) if template is not None else None
         self.session_len = session_len
-        self._static = parse(static) if static else None
-        self._proxies = [parse(p) for p in proxies] if proxies else []
+        self.cooldown = cooldown
+        self._shuffle = shuffle
+        self._static = parse(static) if static is not None else None
+        self._proxies = self._dedupe([parse(p) for p in proxies]) if proxies is not None else []
         if self._proxies and shuffle:
             random.shuffle(self._proxies)
+        self._state: Dict[str, Dict[str, object]] = {}
+        for proxy in self._proxies:
+            self._state.setdefault(proxy.key(), self._fresh_state())
         self._index = 0
 
+    @staticmethod
+    def _dedupe(proxies: List[Proxy]) -> List[Proxy]:
+        """drop later entries sharing a canonical key, preserving first-seen order"""
+        seen: Dict[str, Proxy] = {}
+        for proxy in proxies:
+            seen.setdefault(proxy.key(), proxy)
+        return list(seen.values())
+
+    @staticmethod
+    def _fresh_state() -> Dict[str, object]:
+        """a clean per-proxy state entry: no uses, no timeout, not burned"""
+        return {"uses": 0, "timeout": None, "burned": False}
+
+    @staticmethod
+    def _normalize_template(template: str) -> str:
+        """fill a bare `{}` session slot, escape-aware
+
+        a lone `{}` is treated as `{session}` (back-compat with simple templates).
+        an escaped literal `{{}}` — str.format's own convention for a literal `{}`
+        in the output — is left untouched so it survives to render as `{}`, not
+        `{session}`.
+        """
+        return _BARE_SESSION_SLOT.sub("{session}", template)
+
     @classmethod
     def from_file(cls, path: str, **kwargs) -> "AioProxies":
         """build a rotating manager from a newline-delimited proxy file
@@ -59,28 +114,273 @@ class AioProxies:
         return cls(proxies=lines, **kwargs)
 
     def session_id(self) -> str:
-        """generate a fresh numeric session id for template filling"""
-        return "".join(random.choices(string.digits, k=self.session_len))
+        """generate a fresh alphanumeric session id for template filling"""
+        alphabet = string.ascii_letters + string.digits
+        return "".join(random.choices(alphabet, k=self.session_len))
+
+    def _is_list_source(self) -> bool:
+        """whether this manager rotates a proxy list (vs template/static)"""
+        return self._static is None and self.template is None
+
+    def _warn_non_list(self, method: str) -> None:
+        """log that a health/pool method only applies to list sources"""
+        log.warning("%s applies only to list sources; no-op on template/static", method)
+
+    def _available(self, timeout: object, now: float) -> bool:
+        """whether a timeout value means the proxy is selectable right now"""
+        if timeout is None or timeout == 0:
+            return True
+        if timeout == _DEAD:
+            return False
+        return now >= timeout
 
     def next(self, **fields: object) -> Proxy:
         """return the next proxy from the configured source
 
-        for template sources, `{session}` is always filled with a fresh id and any
-        other named placeholder is filled from `fields` (e.g. next(country="ca",
-        ttl=30)). fields are ignored by list/static sources.
+        template sources: `{session}` is always filled with a fresh id, other
+        placeholders come from `fields` (e.g. next(country="ca", ttl=30)). fields
+        are ignored by list/static sources.
+
+        list sources: skips proxies whose timeout is active (-1 dead, or a future
+        ts not yet passed), increments `uses` on handout, applies the manager's
+        cooldown. if none are fine but some are merely timed, hands out the one
+        recovering soonest (warns); raises ProxiesExhaustedError if all are dead.
         """
         if self._static is not None:
             return self._static
         if self.template is not None:
-            return parse(self.template.format(session=self.session_id(), **fields))
-        proxy = self._proxies[self._index]
-        self._index = (self._index + 1) % len(self._proxies)
+            if "session" in fields:
+                # `session` is auto-filled with a fresh id; a caller-supplied one would
+                # collide in str.format with an opaque TypeError — reject it clearly
+                raise ValueError("'session' is filled automatically; do not pass it to next()")
+            try:
+                filled = self.template.format(session=self.session_id(), **fields)
+            except (KeyError, IndexError) as exc:
+                raise ValueError(
+                    f"template placeholder {exc} not provided; pass it to next(**fields)"
+                ) from exc
+            except ValueError as exc:
+                # a malformed template (e.g. an unmatched '{') makes str.format raise a
+                # bare ValueError; re-raise naming the cause so it isn't cryptic
+                raise ValueError(f"malformed proxy template {self.template!r}: {exc}") from exc
+            return parse(filled)
+        return self._next_from_list()
+
+    def _next_from_list(self) -> Proxy:
+        """rotation + health selection over the proxy list"""
+        if not self._proxies:
+            raise ProxiesExhaustedError("no proxies in pool")
+
+        now = time.time()
+        count = len(self._proxies)
+        fine: List[int] = []
+        for offset in range(count):
+            idx = (self._index + offset) % count
+            timeout = self._state[self._proxies[idx].key()]["timeout"]
+            if self._available(timeout, now):
+                fine.append(idx)
+
+        if fine:
+            chosen = fine[0]
+        else:
+            chosen = self._soonest_recovering()
+            if chosen is None:
+                raise ProxiesExhaustedError("all proxies are permanently dead (-1)")
+            if self._has_burned():
+                log.warning("no proxies available; handing out the one recovering soonest")
+            else:
+                log.debug("all proxies cooling down; handing out the one recovering soonest")
+
+        proxy = self._proxies[chosen]
+        self._index = (chosen + 1) % count
+        state = self._state[proxy.key()]
+        state["uses"] = int(state["uses"]) + 1
+        if self.cooldown > 0:
+            state["timeout"] = int(now) + self.cooldown
         return proxy
 
+    def _has_burned(self) -> bool:
+        """whether the empty 'fine' tier reflects a genuine burn(), not just cooldown
+
+        tracked directly via each state's `burned` flag (set by burn(), cleared by
+        restore()) so a real burn is never masked by the manager's own cooldown
+        resting on the same `timeout` field, regardless of whether cooldown is on.
+        used to pick warning (real trouble) vs debug (normal cooldown) on the
+        soonest-recovering path.
+        """
+        return any(state["burned"] for state in self._state.values())
+
+    def _soonest_recovering(self) -> Optional[int]:
+        """index of the timed (non-dead) proxy recovering soonest, or None"""
+        best_idx: Optional[int] = None
+        best_ts: Optional[float] = None
+        for idx, proxy in enumerate(self._proxies):
+            timeout = self._state[proxy.key()]["timeout"]
+            if timeout is None or timeout == 0 or timeout == _DEAD:
+                continue
+            ts = float(timeout)  # type: ignore[arg-type]
+            if best_ts is None or ts < best_ts:
+                best_ts = ts
+                best_idx = idx
+        return best_idx
+
     def get(self, **fields: object) -> Dict[str, str]:
         """convenience: next proxy as an aiohttp / aioweb proxies dict"""
         return self.next(**fields).aiohttp()
 
+    def burn(self, proxy: Union[str, Proxy, Dict[str, str]], seconds: Optional[int] = None) -> None:
+        """mark a proxy unusable: dead (-1) by default, or timed for `seconds`
+
+        accepts any supported proxy shape. raises ValueError (naming the key) if the
+        proxy is not in the pool. no-op + warning on template/static sources.
+        """
+        if not self._is_list_source():
+            self._warn_non_list("burn()")
+            return
+        key = canonical_key(proxy)
+        if key not in self._state:
+            raise ValueError(f"proxy not in pool: {key}")
+        if seconds is None:
+            self._state[key]["timeout"] = _DEAD
+        else:
+            self._state[key]["timeout"] = int(time.time()) + seconds
+        self._state[key]["burned"] = True
+
+    def restore(self, proxy: Union[str, Proxy, Dict[str, str]]) -> None:
+        """clear any burn/timeout on a proxy (back to fine). no-op if already fine
+
+        matches remove()'s contract for an unknown proxy: logs a warning and returns
+        rather than silently doing nothing.
+        """
+        if not self._is_list_source():
+            self._warn_non_list("restore()")
+            return
+        key = canonical_key(proxy)
+        if key not in self._state:
+            log.warning("restore(): proxy not in pool: %s", key)
+            return
+        self._state[key]["timeout"] = None
+        self._state[key]["burned"] = False
+
+    def is_burned(self, proxy: Union[str, Proxy, Dict[str, str]]) -> bool:
+        """whether a proxy is currently unavailable (lazy expiry of timed burns)
+
+        -1 is always burned; an expired timed burn reads False. unknown proxies and
+        non-list sources read False.
+        """
+        if not self._is_list_source():
+            return False
+        key = canonical_key(proxy)
+        if key not in self._state:
+            return False
+        timeout = self._state[key]["timeout"]
+        return not self._available(timeout, time.time())
+
+    def stats(self) -> List[Dict[str, object]]:
+        """per-proxy usage + state snapshot (list sources only, else empty)
+
+        each entry: {"proxy": <spec>, "uses": int, "state": active|timed|dead,
+        "timeout": <ts | -1 | None>}. the spec is the full canonical key.
+        """
+        if not self._is_list_source():
+            self._warn_non_list("stats()")
+            return []
+        now = time.time()
+        out: List[Dict[str, object]] = []
+        for proxy in self._proxies:
+            state = self._state[proxy.key()]
+            timeout = state["timeout"]
+            if timeout == _DEAD:
+                label = "dead"
+            elif self._available(timeout, now):
+                label = "active"
+            else:
+                label = "timed"
+            out.append({
+                "proxy": proxy.key(),
+                "uses": int(state["uses"]),
+                "state": label,
+                "timeout": timeout,
+            })
+        return out
+
+    def reset_stats(self) -> None:
+        """zero every proxy's use counter; leave timeouts untouched"""
+        if not self._is_list_source():
+            self._warn_non_list("reset_stats()")
+            return
+        for state in self._state.values():
+            state["uses"] = 0
+
+    def replace(self, proxies: List[Union[str, Proxy]], *, keep_state: bool = False) -> None:
+        """swap the entire proxy list
+
+        keep_state=False (default) wipes all per-proxy state (fresh batch).
+        keep_state=True preserves uses/timeout for proxies whose canonical key
+        survives the swap; new proxies start clean, dropped ones are forgotten.
+        resets the rotation index and honors the manager's shuffle setting.
+        """
+        if not self._is_list_source():
+            self._warn_non_list("replace()")
+            return
+        old_state = self._state
+        incoming = [to_proxy(p) for p in proxies]
+        if self._shuffle:
+            random.shuffle(incoming)
+        new_proxies: List[Proxy] = []
+        new_state: Dict[str, Dict[str, object]] = {}
+        for proxy in incoming:
+            key = proxy.key()
+            if key in new_state:
+                continue
+            new_proxies.append(proxy)
+            if keep_state and key in old_state:
+                new_state[key] = old_state[key]
+            else:
+                new_state[key] = self._fresh_state()
+        self._proxies = new_proxies
+        self._state = new_state
+        self._index = 0
+
+    def add(self, proxies: Union[str, Proxy, List[Union[str, Proxy]]]) -> None:
+        """append proxies to the pool, keeping existing state; skip duplicate keys"""
+        if not self._is_list_source():
+            self._warn_non_list("add()")
+            return
+        items = proxies if isinstance(proxies, list) else [proxies]
+        for item in items:
+            proxy = to_proxy(item)
+            key = proxy.key()
+            if key in self._state:
+                continue
+            self._proxies.append(proxy)
+            self._state[key] = self._fresh_state()
+
+    def remove(self, proxy: Union[str, Proxy, Dict[str, str]]) -> None:
+        """drop a proxy from the pool entirely (by canonical key, any shape)
+
+        distinct from burn (burn = unusable but tracked; remove = gone). decrements
+        the rotation index when the removed slot precedes it, so the next() cursor
+        still lands on the same upcoming proxy instead of skipping one. no-op +
+        warning if the proxy is not present.
+        """
+        if not self._is_list_source():
+            self._warn_non_list("remove()")
+            return
+        key = canonical_key(proxy)
+        if key not in self._state:
+            log.warning("remove(): proxy not in pool: %s", key)
+            return
+        removed_idx = next(i for i, p in enumerate(self._proxies) if p.key() == key)
+        self._proxies = [p for p in self._proxies if p.key() != key]
+        del self._state[key]
+        if not self._proxies:
+            self._index = 0
+        else:
+            if removed_idx < self._index:
+                self._index -= 1
+            self._index %= len(self._proxies)
+
 
 # name aliases — same class, call it whichever reads best at your call site
 ProxyManager = AioProxies

src/aioproxies/net.py

@@ -37,8 +37,10 @@ async def current_ip(
     try:
         async with aiohttp.ClientSession(timeout=t) as session:
             async with session.get(test_url, proxy=p.url()) as resp:
-                data = await resp.json()
-                return data.get("ip")
+                # content_type=None: an echo endpoint may return text/plain json; the
+                # default would raise ContentTypeError and silently return None
+                data = await resp.json(content_type=None)
+                return data.get("ip") if isinstance(data, dict) else None
     except Exception as exc:
         log.warning("ip check failed: %s", exc)
         return None

src/aioproxies/proxy.py

@@ -3,15 +3,31 @@
 a `Proxy` holds host/port/optional-auth and renders the shapes different clients
 want: an aiohttp/aioweb proxies dict, a camoufox proxy dict, a socks5 dict, or a
 plain url. `parse` accepts the common "host:port" and "host:port:user:pass" string
-forms (the user field may itself contain commas, e.g. session-param proxies).
+forms (the user field may itself contain commas, e.g. session-param proxies; the
+password field may itself contain colons). auth-less (IP-authenticated) proxies
+with no creds are first-class — every render shape handles them via `has_auth`.
+
+`key()` produces a stable canonical identity (`host:port:user:pass`, or
+`host:port` when auth-less) so burn/remove/stats recognize the same proxy from any
+input shape. `canonical_key()` extends that to dict/url forms.
 """
 from dataclasses import dataclass
 from typing import Dict, Optional, Union
+from urllib.parse import quote, unquote, urlsplit
 
 SCHEME_HTTP = "http"
 SCHEME_SOCKS5 = "socks5"
 
 
+def normalize_port(port: Union[str, int]) -> str:
+    """canonical port string: parsed to int, rendered without zero-padding
+
+    keeps host:080 and host:80 as one canonical key. raises ValueError on a
+    non-integer port rather than silently keying it as-is.
+    """
+    return str(int(port))
+
+
 @dataclass
 class Proxy:
     """a single proxy endpoint with optional auth"""
@@ -26,10 +42,32 @@ class Proxy:
         """whether credentials are present"""
         return bool(self.user and self.password)
 
-    def url(self, scheme: str = SCHEME_HTTP) -> str:
-        """render as a url, embedding auth when present"""
+    def key(self) -> str:
+        """stable canonical identity: host:port:user:pass, or host:port if auth-less
+
+        the host is lowercased (hostnames are case-insensitive per DNS, so
+        PROXY.example.com and proxy.example.com are the same host and collapse to one
+        key); credentials are kept verbatim. the port is normalized (leading zeros
+        stripped) so host:080 and host:80 collapse to one key. the password is
+        included in full (two proxies differing only by password are distinct
+        slots). auth-less proxies collapse to host:port with no trailing colons.
+        """
+        host = self.host.lower()
+        port = normalize_port(self.port)
         if self.has_auth:
-            return f"{scheme}://{self.user}:{self.password}@{self.host}:{self.port}"
+            return f"{host}:{port}:{self.user}:{self.password}"
+        return f"{host}:{port}"
+
+    def url(self, scheme: str = SCHEME_HTTP) -> str:
+        """render as a url, embedding auth when present
+
+        credentials are percent-encoded so reserved chars (/ # ? @ :) in a user or
+        password produce a valid url; this mirrors the unquote() on the parse side.
+        """
+        if self.has_auth:
+            user = quote(str(self.user), safe="")
+            password = quote(str(self.password), safe="")
+            return f"{scheme}://{user}:{password}@{self.host}:{self.port}"
         return f"{scheme}://{self.host}:{self.port}"
 
     def aiohttp(self) -> Dict[str, str]:
@@ -62,16 +100,79 @@ def parse(spec: Union[str, Proxy]) -> Proxy:
     """parse a proxy spec into a Proxy
 
     accepts an existing Proxy (returned as-is) or a colon-delimited string in
-    `host:port` or `host:port:user:pass` form. raises ValueError on anything else
-    rather than guessing.
+    `host:port` or `host:port:user:pass` form. the 4-part form splits on the first
+    three colons only, so a password may itself contain colons. raises ValueError
+    on anything else rather than guessing.
     """
     if isinstance(spec, Proxy):
         return spec
-    parts = spec.split(":")
-    if len(parts) == 4:
-        host, port, user, password = parts
+    if spec.count(":") >= 3:
+        host, port, user, password = spec.split(":", 3)
         return Proxy(host, port, user, password)
+    parts = spec.split(":")
     if len(parts) == 2:
         host, port = parts
         return Proxy(host, port)
     raise ValueError("expected 'host:port' or 'host:port:user:pass'")
+
+
+def canonical_key(spec: Union[str, Proxy, Dict[str, str]]) -> str:
+    """canonical identity key for any supported proxy shape
+
+    accepts a spec string, a Proxy, a url string (`http://user:pass@host:port`,
+    `socks5://...`), an aiohttp dict (`{"http": url, "https": url}`), or a
+    camoufox/socks5 dict (`{"server": ..., "username": ..., "password": ...}`).
+    all forms collapse to the same `host:port:user:pass` (or `host:port` auth-less)
+    key, so burn/remove/stats recognize one proxy regardless of how it was passed.
+    """
+    return to_proxy(spec).key()
+
+
+def to_proxy(spec: Union[str, Proxy, Dict[str, str]]) -> Proxy:
+    """normalize any supported proxy shape into a Proxy
+
+    accepts a spec string, a Proxy, a url string, an aiohttp dict, or a
+    camoufox/socks5 dict — the same shapes `canonical_key` accepts. used both for
+    keying and for adding proxies to a pool from any shape.
+    """
+    if isinstance(spec, Proxy):
+        return spec
+    if isinstance(spec, dict):
+        return _proxy_from_dict(spec)
+    if isinstance(spec, str):
+        if "://" in spec:
+            return _proxy_from_url(spec)
+        return parse(spec)
+    raise ValueError(f"unsupported proxy shape: {type(spec).__name__}")
+
+
+def _proxy_from_dict(spec: Dict[str, str]) -> Proxy:
+    """normalize an aiohttp / camoufox / socks5 dict into a Proxy"""
+    if "server" in spec:
+        proxy = _proxy_from_url(spec["server"])
+        # prefer explicit dict auth; otherwise fall back to auth embedded in the server
+        # URL (http://user:pass@host:port) so it isn't silently dropped — which would
+        # key the proxy auth-less and collide with a genuinely auth-less one
+        user = spec.get("username") or proxy.user
+        password = spec.get("password") or proxy.password
+        if user and password:
+            return Proxy(proxy.host, proxy.port, user, password)
+        return Proxy(proxy.host, proxy.port)
+    for field in ("http", "https"):
+        if field in spec:
+            return _proxy_from_url(spec[field])
+    raise ValueError("dict proxy must have 'server' or 'http'/'https'")
+
+
+def _proxy_from_url(url: str) -> Proxy:
+    """normalize a proxy url (any scheme, optional auth) into a Proxy"""
+    parts = urlsplit(url)
+    host = parts.hostname
+    port = str(parts.port) if parts.port is not None else ""
+    if host is None:
+        raise ValueError(f"could not parse proxy url: {url}")
+    if parts.username:
+        user = unquote(parts.username)
+        password = unquote(parts.password) if parts.password is not None else ""
+        return Proxy(host, port, user, password)
+    return Proxy(host, port)