Merge branch 'simp/r3-13-C' into simp/r3-13

gateway/platforms/api_server_room_dispatch.py

@@ -15,52 +15,41 @@ from gateway.platforms.api_server_room_grants import _json_error
 
 
 async def _ensure_hosted_member_session(self, dispatch: Any) -> str:
-    """Create or verify the target's canonical hidden group session.
-
-    Reusing the ``Group: <room_id>`` namespace is intentional: a room that
-    moves from Desktop-assisted to hosted execution keeps one transcript.
-    A conflicting title with a different session id fails closed instead
-    of merging unrelated conversations.
-    """
+    """Create or verify the target's canonical hidden group session. The ``Group: <room_id>``
+    namespace is reused on purpose (Desktop-assisted -> hosted keeps one transcript); a
+    conflicting title under another session id fails closed rather than merging."""
     db = await self._ensure_session_db_async()
     if db is None:
         raise RuntimeError("session database unavailable")
     title = f"Group: {dispatch.room_id}"
     seed = (
         f"{dispatch.home_install_id}\0{dispatch.room_id}\0"
-        f"{dispatch.member_id}\0{dispatch.target_profile}"
-    )
+        f"{dispatch.member_id}\0{dispatch.target_profile}")
     session_id = f"room_{hashlib.sha256(seed.encode()).hexdigest()[:32]}"
 
     def atomic(conn):
-        row = conn.execute(
-            "SELECT id, title, source FROM sessions WHERE id=?", (session_id,)
-        ).fetchone()
+        row = conn.execute("SELECT id, title, source FROM sessions WHERE id=?", (session_id,)).fetchone()
         if row is not None:
             if row["title"] != title or row["source"] != "bot_room":
                 raise RuntimeError("room session identity conflicts with existing data")
             return session_id
         clean_title = db.sanitize_title(title)
         conflict = conn.execute(
-            "SELECT id FROM sessions WHERE title=? AND id!=?", (clean_title, session_id)
-        ).fetchone()
+            "SELECT id FROM sessions WHERE title=? AND id!=?", (clean_title, session_id)).fetchone()
         if conflict:
             raise RuntimeError(
                 "Another group already uses this room title on the target gateway. "
-                "Rename or migrate that group before retrying."
-            )
+                "Rename or migrate that group before retrying.")
         conn.execute(
-            "INSERT INTO sessions(id, source, title, hidden, started_at) "
-            "VALUES(?, 'bot_room', ?, 1, ?)",
-            (session_id, clean_title, time.time()),
-        )
+            "INSERT INTO sessions(id, source, title, hidden, started_at) VALUES(?, 'bot_room', ?, 1, ?)",
+            (session_id, clean_title, time.time()))
         return session_id
 
     return await asyncio.to_thread(db._execute_write, atomic)
 
 
 def _room_dispatch_error(exc: Exception, *, _openai_error) -> "web.Response":
-    message = str(exc)
+    message, code = str(exc), "invalid_room_dispatch"
     lowered = message.lower()
     if "execution policy" in lowered or "remote room execution requires" in lowered:
         message = "Room execution policy changed; reauthorization is required."
@@ -68,49 +57,29 @@ def _room_dispatch_error(exc: Exception, *, _openai_error) -> "web.Response":
     elif "capability catalog changed" in lowered:
         message = "Room capability catalog changed; reauthorization is required."
         code = "room_capability_catalog_changed"
-    else:
-        code = "invalid_room_dispatch"
     return _json_error(_openai_error, message, code=code, status=403)
 
 
 async def _normalize_room_dispatch(
-    self,
-    request: "web.Request",
-    body: Any,
-    *,
-    _api_server,
-) -> tuple[Any, "web.Response | None"]:
+    self, request: "web.Request", body: Any, *, _api_server) -> tuple[Any, "web.Response | None"]:
     """Validate and normalize a scoped RoomLink dispatch request."""
-    _api_request_profile = _api_server._api_request_profile
-    _openai_error = _api_server._openai_error
-
-    room_token = self._room_grant_token(request)
+    _openai_error, room_token = _api_server._openai_error, self._room_grant_token(request)
     if not room_token:
         return body, None
-
     if not isinstance(body, dict) or set(body) - {"input", "hosted_room_dispatch"}:
         return body, _json_error(
             _openai_error, "Room dispatch accepts only input and hosted_room_dispatch.",
-            code="invalid_room_dispatch", status=400,
-        )
+            code="invalid_room_dispatch", status=400)
     try:
         from gateway import hosted_rooms
-        from gateway.hosted_room_peer import (
-            GatewayRoomCatalog,
-            HostedMemberDispatch,
-            verify_room_grant,
-        )
+        from gateway.hosted_room_peer import GatewayRoomCatalog, HostedMemberDispatch, verify_room_grant
         from gateway.hosted_room_execution_policy import RoomExecutionPolicy
         from gateway.platforms.api_server_room_grants import _local_room_catalog
-
         dispatch = HostedMemberDispatch.from_mapping(body.get("hosted_room_dispatch"))
         verify_room_grant(self._room_grant_secret(), room_token, dispatch, permission="dispatch")
-        active_profile = _api_request_profile.get() or "default"
+        active_profile = _api_server._api_request_profile.get() or "default"
         local_install = hosted_rooms.local_authority_gateway_id()
-        if (
-            dispatch.target_profile != active_profile
-            or dispatch.target_install_id != local_install
-        ):
+        if dispatch.target_profile != active_profile or dispatch.target_install_id != local_install:
             raise ValueError("room dispatch target does not match this profile")
         _, catalog_map = _local_room_catalog(self, active_profile, local_install)
         catalog = GatewayRoomCatalog.from_mapping(catalog_map)

gateway/platforms/api_server_room_grants.py

@@ -2,7 +2,7 @@
 
 import time
 import uuid
-from typing import Any
+from typing import Any, Optional
 
 try:
     from aiohttp import web
@@ -15,58 +15,58 @@ class RoomGrantReauthorizationRequired(ValueError):
 
 
 def _json_error(_openai_error, message: str, *, status: int, **error_kwargs) -> "web.Response":
-    """``web.json_response(_openai_error(message, **kw), status=...)`` with the injected envelope builder."""
+    """JSON error response built with the injected ``_openai_error`` envelope builder."""
     return web.json_response(_openai_error(message, **error_kwargs), status=status)
 
 
-def _require_unchanged_execution_policy(
-    claims: dict[str, Any],
-    execution_policy: dict[str, Any],
-) -> None:
+def _require_unchanged_execution_policy(claims: dict[str, Any], execution_policy: dict[str, Any]) -> None:
     """Keep renewal from silently granting a changed execution policy."""
-    if str(execution_policy.get("policy_digest") or "") != str(
-        claims.get("execution_policy_digest") or ""
-    ):
+    if str(execution_policy.get("policy_digest") or "") != str(claims.get("execution_policy_digest") or ""):
         raise RoomGrantReauthorizationRequired("room execution policy changed")
 
 
-def _room_grant_error_response(exc: Exception, *, _openai_error) -> "web.Response":
-    reauth = isinstance(exc, RoomGrantReauthorizationRequired)
-    return _json_error(
-        _openai_error,
-        "Room authorization needs to be renewed." if reauth else "Room authorization is invalid or expired.",
-        err_type="gateway_auth_error",
-        code="room_reauthorization_required" if reauth else "invalid_room_grant",
-        status=403 if reauth else 401,
-    )
+def _room_grant_error_response(exc: Optional[Exception] = None, *, _openai_error) -> "web.Response":
+    """401 invalid grant, or 403 reauthorization-required for a revoked/superseded grant."""
+    if isinstance(exc, RoomGrantReauthorizationRequired):
+        message, code, status = "Room authorization needs to be renewed.", "room_reauthorization_required", 403
+    else:
+        message, code, status = "Room authorization is invalid or expired.", "invalid_room_grant", 401
+    return _json_error(_openai_error, message, err_type="gateway_auth_error", code=code, status=status)
 
 
 def _hard_expiry(claims: dict[str, Any]) -> float:
     return float(claims.get("status_expires_at", claims["expires_at"]))
 
 
-def _require_local_target(claims: dict[str, Any], profile: str, installation_id: str) -> None:
-    if claims["target_profile"] != profile or claims["target_install_id"] != installation_id:
+_ROOM_IDENTITY_FIELDS = ("room_id", "home_install_id", "authority_gateway_id", "authority_epoch", "member_id")
+
+
+def _room_identity(source: dict[str, Any], *, coerce: bool = False) -> dict[str, Any]:
+    """Room-authority kwargs for ``issue_room_grant``; *coerce* applies ``str``/``int`` to raw body values."""
+    text = str if coerce else (lambda v: v)
+    return {k: int(source[k]) if k == "authority_epoch" else text(source[k]) for k in _ROOM_IDENTITY_FIELDS}
+
+
+def _local_target(claims: dict[str, Any] | None, _api_request_profile) -> tuple[str, str]:
+    """Return ``(profile, installation_id)`` for this gateway; *claims* must target it."""
+    from gateway import hosted_rooms
+    profile = _api_request_profile.get() or "default"
+    installation_id = hosted_rooms.local_authority_gateway_id()
+    if claims is not None and (claims["target_profile"], claims["target_install_id"]) != (profile, installation_id):
         raise ValueError("room grant target does not match this profile")
+    return profile, installation_id
 
 
 def _local_room_catalog(self, profile: str, installation_id: str) -> tuple[dict, dict]:
     """Return ``(execution_policy, catalog)`` for this gateway's *profile*."""
     from gateway.hosted_room_peer import PROTOCOL_VERSION, catalog_mapping
     from gateway.hosted_room_execution_policy import execution_policy_mapping
-
     with self._profile_scope(profile):
         execution_policy = execution_policy_mapping(target_profile=profile)
     catalog = catalog_mapping(
-        installation_id=installation_id,
-        protocol_versions=(PROTOCOL_VERSION,),
-        link_modes=("direct",),
-        persistent_process=True,
-        text=True,
-        attachments=False,
-        target_profile=profile,
-        execution_policy=execution_policy,
-    )
+        installation_id=installation_id, protocol_versions=(PROTOCOL_VERSION,), link_modes=("direct",),
+        persistent_process=True, text=True, attachments=False, target_profile=profile,
+        execution_policy=execution_policy)
     return execution_policy, catalog
 
 
@@ -75,33 +75,31 @@ def _http_routes(self) -> list[tuple[str, str, Any]]:
         ("POST", "/v1/room-members/invitations", self._handle_room_member_invitation),
         ("GET", "/v1/room-members/capabilities", self._handle_room_member_capabilities),
         ("POST", "/v1/room-members/grants/refresh", self._handle_room_member_grant_refresh),
-        ("POST", "/v1/room-members/grants/revoke", self._handle_room_member_grant_revoke),
-    ]
+        ("POST", "/v1/room-members/grants/revoke", self._handle_room_member_grant_revoke)]
 
 
 def _room_grant_token(request: "web.Request") -> str:
-    authorization = str(request.headers.get("Authorization") or "")
-    scheme, separator, token = authorization.partition(" ")
-    if not separator or scheme.lower() != "hermesroom":
-        return ""
-    return token.strip()
+    scheme, separator, token = str(request.headers.get("Authorization") or "").partition(" ")
+    return token.strip() if separator and scheme.lower() == "hermesroom" else ""
 
 
 def _room_grant_secret(self) -> bytes:
     from gateway.hosted_room_peer import gateway_room_grant_secret
-
     return gateway_room_grant_secret()
 
 
-def _room_grant_claims(self, request: "web.Request", *, permission: str) -> dict[str, Any]:
+def _decode_request_grant(self, request: "web.Request", *, permission: str) -> dict[str, Any]:
+    """Signature/scope/horizon check only (no revocation lookup)."""
     from gateway.hosted_room_peer import decode_room_grant
-
     token = self._room_grant_token(request)
     if not token:
         raise ValueError("room grant is missing")
-    claims = decode_room_grant(self._room_grant_secret(), token, permission=permission)
-    from gateway import hosted_rooms
+    return decode_room_grant(self._room_grant_secret(), token, permission=permission)
 
+
+def _room_grant_claims(self, request: "web.Request", *, permission: str) -> dict[str, Any]:
+    claims = _decode_request_grant(self, request, permission=permission)
+    from gateway import hosted_rooms
     db_path = hosted_rooms.default_db_path()
     if hosted_rooms.room_grant_is_revoked(db_path, claims=claims):
         raise RoomGrantReauthorizationRequired("room grant is revoked")
@@ -111,12 +109,7 @@ def _room_grant_claims(self, request: "web.Request", *, permission: str) -> dict
 
 
 async def _handle_room_member_invitation(
-    self,
-    request: "web.Request",
-    *,
-    _openai_error,
-    _api_request_profile,
-) -> "web.Response":
+    self, request: "web.Request", *, _openai_error, _api_request_profile) -> "web.Response":
     """Mint a short-lived room/profile grant for a trusted home gateway."""
     auth_err = self._check_auth(request)
     if auth_err:
@@ -124,101 +117,57 @@ async def _handle_room_member_invitation(
     body, error = await self._read_json_body(request)
     if error:
         return error
-    required = {"room_id", "home_install_id", "authority_gateway_id", "authority_epoch", "member_id"}
+    required = set(_ROOM_IDENTITY_FIELDS)
     allowed = required | {"grant_id", "ttl_seconds", "status_ttl_seconds"}
     if set(body) - allowed or not required <= set(body):
         return _json_error(
             _openai_error, "Invitation is missing required room authority fields.",
-            code="invalid_room_invitation", status=400,
-        )
+            code="invalid_room_invitation", status=400)
     try:
         from gateway import hosted_rooms
         from gateway.hosted_room_peer import decode_room_grant, issue_room_grant
-
-        profile = _api_request_profile.get() or "default"
-        target_install_id = hosted_rooms.local_authority_gateway_id()
+        profile, target_install_id = _local_target(None, _api_request_profile)
         ttl = float(body.get("ttl_seconds", 3600))
         if not 60 <= ttl <= 24 * 60 * 60:
             raise ValueError("ttl_seconds must be between 60 and 86400")
         status_ttl = float(body.get("status_ttl_seconds", ttl))
         if not ttl <= status_ttl <= 30 * 24 * 60 * 60:
-            raise ValueError(
-                "status_ttl_seconds must be at least ttl_seconds and no more than 2592000"
-            )
+            raise ValueError("status_ttl_seconds must be at least ttl_seconds and no more than 2592000")
         execution_policy, catalog = _local_room_catalog(self, profile, target_install_id)
         token = issue_room_grant(
             self._room_grant_secret(),
             grant_id=str(body.get("grant_id") or f"grant-{uuid.uuid4().hex}"),
-            room_id=str(body["room_id"]),
-            home_install_id=str(body["home_install_id"]),
-            authority_gateway_id=str(body["authority_gateway_id"]),
-            authority_epoch=int(body["authority_epoch"]),
-            member_id=str(body["member_id"]),
-            target_install_id=target_install_id,
-            target_profile=profile,
-            execution_policy_digest=execution_policy["policy_digest"],
-            issued_at=time.time(),
-            ttl_seconds=ttl,
-            status_ttl_seconds=status_ttl,
-        )
+            **_room_identity(body, coerce=True),
+            target_install_id=target_install_id, target_profile=profile,
+            execution_policy_digest=execution_policy["policy_digest"], issued_at=time.time(),
+            ttl_seconds=ttl, status_ttl_seconds=status_ttl)
         claims = decode_room_grant(self._room_grant_secret(), token, permission="status")
         hosted_rooms.reserve_peer_room(
-            hosted_rooms.default_db_path(), claims=claims, expires_at=_hard_expiry(claims)
-        )
+            hosted_rooms.default_db_path(), claims=claims, expires_at=_hard_expiry(claims))
     except Exception as exc:
         return _json_error(_openai_error, str(exc), code="invalid_room_invitation", status=400)
-    return web.json_response(
-        {
-            "object": "hermes.room_member.invitation",
-            "grant": token,
-            "target_profile": profile,
-            "catalog": catalog,
-            "expires_at": float(claims["expires_at"]),
-            "status_expires_at": float(claims["status_expires_at"]),
-        },
-        status=201,
-    )
+    return web.json_response({
+        "object": "hermes.room_member.invitation", "grant": token, "target_profile": profile,
+        "catalog": catalog, "expires_at": float(claims["expires_at"]),
+        "status_expires_at": float(claims["status_expires_at"])}, status=201)
 
 
 async def _handle_room_member_capabilities(
-    self,
-    request: "web.Request",
-    *,
-    _openai_error,
-    _api_request_profile,
-) -> "web.Response":
+    self, request: "web.Request", *, _openai_error, _api_request_profile) -> "web.Response":
     """Verify a scoped grant and return this target's live room catalog."""
     try:
-        from gateway import hosted_rooms
-
         claims = self._room_grant_claims(request, permission="status")
-        profile = _api_request_profile.get() or "default"
-        installation_id = hosted_rooms.local_authority_gateway_id()
-        _require_local_target(claims, profile, installation_id)
+        profile, installation_id = _local_target(claims, _api_request_profile)
         _, catalog = _local_room_catalog(self, profile, installation_id)
     except Exception as exc:
         return _room_grant_error_response(exc, _openai_error=_openai_error)
-    return web.json_response(
-        {
-            "object": "hermes.room_member.capabilities",
-            "room_id": claims["room_id"],
-            "home_install_id": claims["home_install_id"],
-            "authority_gateway_id": claims["authority_gateway_id"],
-            "authority_epoch": claims["authority_epoch"],
-            "member_id": claims["member_id"],
-            "target_profile": profile,
-            "catalog": catalog,
-        }
-    )
+    return web.json_response({
+        "object": "hermes.room_member.capabilities", **{k: claims[k] for k in _ROOM_IDENTITY_FIELDS},
+        "target_profile": profile, "catalog": catalog})
 
 
 async def _handle_room_member_grant_refresh(
-    self,
-    request: "web.Request",
-    *,
-    _openai_error,
-    _api_request_profile,
-) -> "web.Response":
+    self, request: "web.Request", *, _openai_error, _api_request_profile) -> "web.Response":
     """Refresh dispatch access without a Desktop or broad gateway key."""
     body, error = await self._read_json_body(request)
     if error:
@@ -226,19 +175,13 @@ async def _handle_room_member_grant_refresh(
     if set(body) - {"ttl_seconds"}:
         return _json_error(
             _openai_error, "Grant refresh accepts only ttl_seconds.",
-            code="invalid_room_grant_refresh", status=400,
-        )
+            code="invalid_room_grant_refresh", status=400)
     try:
-        from gateway import hosted_rooms
         from gateway.hosted_room_peer import MAX_DISPATCH_GRANT_TTL_SECONDS, issue_room_grant
         from gateway.hosted_room_execution_policy import execution_policy_mapping
-
-        # A status-only bearer may observe a run but must never mint new
-        # dispatch authority: renewal requires a still-live dispatch permission.
+        # A status-only bearer must never mint dispatch authority: renewal needs live "dispatch".
         claims = self._room_grant_claims(request, permission="dispatch")
-        profile = _api_request_profile.get() or "default"
-        installation_id = hosted_rooms.local_authority_gateway_id()
-        _require_local_target(claims, profile, installation_id)
+        profile, installation_id = _local_target(claims, _api_request_profile)
         now = time.time()
         hard_expiry = _hard_expiry(claims)
         remaining = hard_expiry - now
@@ -250,41 +193,20 @@ async def _handle_room_member_grant_refresh(
             execution_policy = execution_policy_mapping(target_profile=profile)
         _require_unchanged_execution_policy(claims, execution_policy)
         token = issue_room_grant(
-            self._room_grant_secret(),
-            grant_id=f"grant-refresh-{uuid.uuid4().hex}",
-            room_id=claims["room_id"],
-            home_install_id=claims["home_install_id"],
-            authority_gateway_id=claims["authority_gateway_id"],
-            authority_epoch=int(claims["authority_epoch"]),
-            member_id=claims["member_id"],
-            target_install_id=installation_id,
-            target_profile=profile,
+            self._room_grant_secret(), grant_id=f"grant-refresh-{uuid.uuid4().hex}",
+            **_room_identity(claims), target_install_id=installation_id, target_profile=profile,
             execution_policy_digest=execution_policy["policy_digest"],
-            permissions=claims["permissions"],
-            issued_at=now,
-            ttl_seconds=dispatch_ttl,
-            status_expires_at=hard_expiry,
-        )
+            permissions=claims["permissions"], issued_at=now, ttl_seconds=dispatch_ttl,
+            status_expires_at=hard_expiry)
     except Exception as exc:
         return _room_grant_error_response(exc, _openai_error=_openai_error)
-    return web.json_response(
-        {
-            "object": "hermes.room_member.grant",
-            "grant": token,
-            "expires_at": now + dispatch_ttl,
-            "status_expires_at": hard_expiry,
-            "execution_policy": execution_policy,
-        }
-    )
+    return web.json_response({
+        "object": "hermes.room_member.grant", "grant": token, "expires_at": now + dispatch_ttl,
+        "status_expires_at": hard_expiry, "execution_policy": execution_policy})
 
 
 async def _handle_room_member_grant_revoke(
-    self,
-    request: "web.Request",
-    *,
-    _openai_error,
-    _api_request_profile,
-) -> "web.Response":
+    self, request: "web.Request", *, _openai_error, _api_request_profile) -> "web.Response":
     """Revoke exactly the scoped grant authenticating this request."""
     body, error = await self._read_json_body(request)
     if error:
@@ -292,28 +214,15 @@ async def _handle_room_member_grant_revoke(
     if body:
         return _json_error(
             _openai_error, "Grant revoke accepts no fields.",
-            code="invalid_room_grant_revoke", status=400,
-        )
+            code="invalid_room_grant_revoke", status=400)
     try:
         from gateway import hosted_rooms
-        from gateway.hosted_room_peer import decode_room_grant
-
-        token = self._room_grant_token(request)
-        if not token:
-            raise ValueError("room grant is missing")
-        # Revoke is idempotent: a response-lost retry may authenticate with the
-        # grant just denylisted, so verify signature/scope/horizon directly
-        # (not via _room_grant_claims) and upsert the same grant id.
-        claims = decode_room_grant(self._room_grant_secret(), token, permission="status")
-        profile = _api_request_profile.get() or "default"
-        installation_id = hosted_rooms.local_authority_gateway_id()
-        _require_local_target(claims, profile, installation_id)
+        # Idempotent: a response-lost retry authenticates with the grant just denylisted, so
+        # verify signature/scope/horizon directly (not _room_grant_claims) and upsert the id.
+        claims = _decode_request_grant(self, request, permission="status")
+        _local_target(claims, _api_request_profile)
         hosted_rooms.revoke_room_grant_scope(
-            hosted_rooms.default_db_path(), claims=claims, expires_at=_hard_expiry(claims)
-        )
+            hosted_rooms.default_db_path(), claims=claims, expires_at=_hard_expiry(claims))
     except Exception:
-        return _json_error(
-            _openai_error, "Room authorization is invalid or expired.",
-            err_type="gateway_auth_error", code="invalid_room_grant", status=401,
-        )
+        return _room_grant_error_response(_openai_error=_openai_error)
     return web.json_response({"object": "hermes.room_member.grant.revocation", "revoked": True})

gateway/platforms/api_server_run_idempotency.py

@@ -6,6 +6,7 @@ import logging
 import sqlite3
 import threading
 import time
+from contextlib import contextmanager
 from pathlib import Path
 from typing import Any, Dict
 
@@ -17,23 +18,19 @@ TERMINAL_STATUSES = frozenset({"completed", "failed", "cancelled", "interrupted"
 
 _SELECT_BY_KEY = (
     "SELECT fingerprint, run_id, status_json, owner_pid, owner_started, updated_at "
-    "FROM run_idempotency WHERE scope=? AND idempotency_key=?"
-)
+    "FROM run_idempotency WHERE scope=? AND idempotency_key=?")
 _EXTEND_RETENTION_BY_KEY = (
     "UPDATE run_idempotency SET retention_until=MAX(retention_until, ?) "
-    "WHERE scope=? AND idempotency_key=? AND fingerprint=?"
-)
+    "WHERE scope=? AND idempotency_key=? AND fingerprint=?")
 _EXTEND_RETENTION_BY_RUN = (
     "UPDATE run_idempotency SET retention_until=MAX(retention_until, ?) "
-    "WHERE scope=? AND run_id=?"
-)
+    "WHERE scope=? AND run_id=?")
 # Columns added after the first schema shipped; applied when missing.
 _MIGRATIONS = {
     "owner_pid": "INTEGER NOT NULL DEFAULT 0",
     "owner_started": "INTEGER NOT NULL DEFAULT 0",
     "retention_until": "REAL NOT NULL DEFAULT 0",
-    "acknowledged_at": "REAL",
-}
+    "acknowledged_at": "REAL"}
 
 
 def _encode_status(status: Dict[str, Any]) -> str:
@@ -42,22 +39,19 @@ def _encode_status(status: Dict[str, Any]) -> str:
 
 def _record(run_id, status_json, owner_pid, owner_started, updated_at) -> dict[str, Any]:
     return {
-        "run_id": run_id,
-        "status": json.loads(status_json),
-        "owner_pid": int(owner_pid or 0),
-        "owner_started": int(owner_started or 0),
-        "updated_at": float(updated_at or 0),
-    }
+        "run_id": run_id, "status": json.loads(status_json), "owner_pid": int(owner_pid or 0),
+        "owner_started": int(owner_started or 0), "updated_at": float(updated_at or 0)}
+
+
+def _outcome(row, fingerprint):
+    """Classify a stored ``(scope, key)`` row against the caller's fingerprint."""
+    return ("reused" if hmac.compare_digest(row[0], fingerprint) else "conflict"), _record(*row[1:])
 
 
 class RunIdempotencyStore:
-    """Durable, tenant-scoped reservations for ``POST /v1/runs``.
-
-    A unique ``(scope, key)`` row is inserted inside ``BEGIN IMMEDIATE`` so
-    separate gateway workers/processes cannot both admit the same request.
-    Only request fingerprints and public run status are stored; request bodies
-    and credentials are deliberately excluded.
-    """
+    """Durable, tenant-scoped reservations for ``POST /v1/runs``: a unique ``(scope, key)`` row
+    inserted inside ``BEGIN IMMEDIATE`` so separate workers cannot both admit one request. Only
+    fingerprints and public run status are stored — never request bodies or credentials."""
 
     RETENTION_SECONDS = 24 * 60 * 60
     ACKNOWLEDGED_RETENTION_SECONDS = 24 * 60 * 60
@@ -66,12 +60,10 @@ class RunIdempotencyStore:
     def durable(self) -> bool:
         """Whether reservations survive this process."""
         return self._db_path is not None
-
     def __init__(self, db_path: str = None):
         if db_path is None:
             try:
                 from hermes_cli.config import get_hermes_home
-
                 db_path = str(get_hermes_home() / "runs_idempotency.db")
             except Exception:
                 db_path = ":memory:"
@@ -81,13 +73,10 @@ class RunIdempotencyStore:
         except Exception as exc:
             logger.warning(
                 "Run idempotency storage is unavailable; falling back to "
-                "process memory, so replay will not survive a restart: %s",
-                exc,
-            )
+                "process memory, so replay will not survive a restart: %s", exc)
             self._conn = sqlite3.connect(":memory:", check_same_thread=False)
             self._db_path = None
         from hermes_state import apply_wal_with_fallback
-
         apply_wal_with_fallback(self._conn, db_label="runs_idempotency.db")
         self._conn.execute(
             """CREATE TABLE IF NOT EXISTS run_idempotency (
@@ -110,16 +99,13 @@ class RunIdempotencyStore:
             if column not in columns:
                 self._conn.execute(f"ALTER TABLE run_idempotency ADD COLUMN {column} {ddl}")
         self._conn.execute(
-            "CREATE UNIQUE INDEX IF NOT EXISTS run_idempotency_run_id ON run_idempotency(run_id)"
-        )
+            "CREATE UNIQUE INDEX IF NOT EXISTS run_idempotency_run_id ON run_idempotency(run_id)")
         self._conn.commit()
         self._lock = threading.Lock()
         self._tighten_permissions()
 
     def _tighten_permissions(self) -> None:
-        if not self._db_path:
-            return
-        for suffix in ("", "-wal", "-shm"):
+        for suffix in ("", "-wal", "-shm") if self._db_path else ():
             candidate = Path(self._db_path + suffix)
             try:
                 if candidate.exists():
@@ -127,86 +113,56 @@ class RunIdempotencyStore:
             except OSError:
                 logger.debug("Failed to restrict run idempotency store permissions", exc_info=True)
 
-    def reserve(
-        self,
-        scope: str,
-        key: str,
-        fingerprint: str,
-        run_id: str,
-        status: Dict[str, Any],
-        *,
-        owner_pid: int = 0,
-        owner_started: int = 0,
-        retention_until: float = 0,
-    ):
+    @contextmanager
+    def _immediate_txn(self):
+        """Hold the lock inside ``BEGIN IMMEDIATE``; the body commits, errors roll back."""
+        with self._lock:
+            self._conn.execute("BEGIN IMMEDIATE")
+            try:
+                yield
+            except Exception:
+                self._conn.rollback()
+                raise
+
+    def reserve(self, scope: str, key: str, fingerprint: str, run_id: str, status: Dict[str, Any], *,
+                owner_pid: int = 0, owner_started: int = 0, retention_until: float = 0):
         """Atomically reserve a key; return ``(outcome, stored_record)``."""
         now = time.time()
         retention_until = max(0.0, float(retention_until or 0))
         encoded = _encode_status(status)
-        with self._lock:
-            self._conn.execute("BEGIN IMMEDIATE")
-            try:
-                self._prune_stale_terminal_locked(now)
-                row = self._conn.execute(_SELECT_BY_KEY, (scope, key)).fetchone()
-                if row is not None:
-                    if retention_until:
-                        self._conn.execute(
-                            _EXTEND_RETENTION_BY_KEY, (retention_until, scope, key, fingerprint)
-                        )
-                    self._conn.commit()
-                    outcome = "reused" if hmac.compare_digest(row[0], fingerprint) else "conflict"
-                    return outcome, _record(*row[1:])
-                self._conn.execute(
-                    "INSERT INTO run_idempotency("
-                    "scope,idempotency_key,fingerprint,run_id,status_json,"
-                    "owner_pid,owner_started,retention_until,created_at,updated_at"
-                    ") VALUES(?,?,?,?,?,?,?,?,?,?)",
-                    (
-                        scope, key, fingerprint, run_id, encoded,
-                        int(owner_pid or 0), int(owner_started or 0), retention_until, now, now,
-                    ),
-                )
+        with self._immediate_txn():
+            self._prune_stale_terminal_locked(now)
+            row = self._conn.execute(_SELECT_BY_KEY, (scope, key)).fetchone()
+            if row is not None:
+                if retention_until:
+                    self._conn.execute(_EXTEND_RETENTION_BY_KEY, (retention_until, scope, key, fingerprint))
                 self._conn.commit()
-                return "created", {
-                    "run_id": run_id,
-                    "status": status,
-                    "owner_pid": int(owner_pid or 0),
-                    "owner_started": int(owner_started or 0),
-                    "updated_at": now,
-                }
-            except Exception:
-                self._conn.rollback()
-                raise
+                return _outcome(row, fingerprint)
+            self._conn.execute(
+                "INSERT INTO run_idempotency("
+                "scope,idempotency_key,fingerprint,run_id,status_json,"
+                "owner_pid,owner_started,retention_until,created_at,updated_at"
+                ") VALUES(?,?,?,?,?,?,?,?,?,?)",
+                (scope, key, fingerprint, run_id, encoded, int(owner_pid or 0), int(owner_started or 0),
+                 retention_until, now, now))
+            self._conn.commit()
+            return "created", _record(run_id, encoded, owner_pid, owner_started, now) | {"status": status}
 
     def lookup(self, scope: str, key: str, fingerprint: str, *, retention_until: float = 0):
         """Return ``missing``, ``reused`` or ``conflict`` without reserving."""
         now = time.time()
         retention_until = max(0.0, float(retention_until or 0))
-        with self._lock:
-            self._conn.execute("BEGIN IMMEDIATE")
-            try:
-                if retention_until:
-                    self._conn.execute(
-                        _EXTEND_RETENTION_BY_KEY, (retention_until, scope, key, fingerprint)
-                    )
-                self._prune_stale_terminal_locked(now)
-                row = self._conn.execute(_SELECT_BY_KEY, (scope, key)).fetchone()
-                self._conn.commit()
-            except Exception:
-                self._conn.rollback()
-                raise
-        if row is None:
-            return "missing", None
-        outcome = "reused" if hmac.compare_digest(row[0], fingerprint) else "conflict"
-        return outcome, _record(*row[1:])
+        with self._immediate_txn():
+            if retention_until:
+                self._conn.execute(_EXTEND_RETENTION_BY_KEY, (retention_until, scope, key, fingerprint))
+            self._prune_stale_terminal_locked(now)
+            row = self._conn.execute(_SELECT_BY_KEY, (scope, key)).fetchone()
+            self._conn.commit()
+        return ("missing", None) if row is None else _outcome(row, fingerprint)
 
     def _prune_stale_terminal_locked(self, now: float) -> None:
-        """Prune replay records only after their stored run is terminal.
-
-        The caller owns ``self._lock`` and an active transaction. Age alone
-        can never release an in-flight idempotency reservation: a long or
-        disconnected room turn may legitimately outlive the retention window.
-        """
+        """Prune aged replay records only once their stored run is terminal (caller holds the
+        lock + transaction): a long or disconnected room turn may outlive the retention window."""
         stale = self._conn.execute(
             """SELECT scope, idempotency_key, status_json
                  FROM run_idempotency
@@ -222,9 +178,7 @@ class RunIdempotencyStore:
                 terminal = False
             if terminal:
                 self._conn.execute(
-                    "DELETE FROM run_idempotency WHERE scope=? AND idempotency_key=?",
-                    (stale_scope, stale_key),
-                )
+                    "DELETE FROM run_idempotency WHERE scope=? AND idempotency_key=?", (stale_scope, stale_key))
 
     def status_for_run(self, scope: str, run_id: str, *, retention_until: float = 0) -> dict[str, Any] | None:
         """Load one durable run status inside its authenticated scope."""
@@ -236,16 +190,10 @@ class RunIdempotencyStore:
             row = self._conn.execute(
                 "SELECT status_json, owner_pid, owner_started, updated_at "
                 "FROM run_idempotency WHERE scope=? AND run_id=?",
-                (scope, run_id),
-            ).fetchone()
+                (scope, run_id)).fetchone()
         if row is None:
             return None
-        return {
-            "status": json.loads(row[0]),
-            "owner_pid": int(row[1] or 0),
-            "owner_started": int(row[2] or 0),
-            "updated_at": float(row[3] or 0),
-        }
+        return {k: v for k, v in _record(None, *row).items() if k != "run_id"}
 
     def extend_retention(self, scope: str, run_id: str, until: float) -> bool:
         """Persist the latest verified recovery horizon for an active grant."""
@@ -253,28 +201,21 @@ class RunIdempotencyStore:
         if not checked_until:
             return False
         with self._lock:
-            changed = self._conn.execute(
-                _EXTEND_RETENTION_BY_RUN, (checked_until, scope, run_id)
-            ).rowcount
+            changed = self._conn.execute(_EXTEND_RETENTION_BY_RUN, (checked_until, scope, run_id)).rowcount
             self._conn.commit()
         return changed == 1
 
     def owns_run(self, scope: str, run_id: str) -> bool:
         with self._lock:
-            return (
-                self._conn.execute(
-                    "SELECT 1 FROM run_idempotency WHERE scope=? AND run_id=?",
-                    (scope, run_id),
-                ).fetchone()
-                is not None
-            )
+            row = self._conn.execute(
+                "SELECT 1 FROM run_idempotency WHERE scope=? AND run_id=?", (scope, run_id)).fetchone()
+        return row is not None
 
     def update_status(self, run_id: str, status: Dict[str, Any]) -> None:
         with self._lock:
             self._conn.execute(
                 "UPDATE run_idempotency SET status_json=?, updated_at=? WHERE run_id=?",
-                (_encode_status(status), time.time(), run_id),
-            )
+                (_encode_status(status), time.time(), run_id))
             self._conn.commit()
 
     def close(self) -> None:

gateway/platforms/qqbot/__init__.py

@@ -1,37 +1,15 @@
-"""QQBot platform package.
+"""QQBot platform package. Re-exports adapter symbols so existing import paths
+(``from gateway.platforms.qqbot import QQAdapter, check_qq_requirements``) keep working.
+Sub-modules: constants, utils, crypto (AES-256-GCM), onboard (QR), chunked_upload, keyboards."""
 
-Re-exports the adapter symbols from ``adapter.py`` (the original ``qqbot.py``)
-so all existing import paths remain unchanged, e.g.
-``from gateway.platforms.qqbot import QQAdapter, check_qq_requirements``.
-
-Sub-modules: ``constants``, ``utils`` (User-Agent, config helpers), ``crypto``
-(AES-256-GCM), ``onboard`` (QR scan-to-configure), ``chunked_upload``, ``keyboards``.
-"""
-
-from .adapter import (  # noqa: F401
-    QQAdapter,
-    QQCloseError,
-    check_qq_requirements,
-    _coerce_list,
-    _ssrf_redirect_guard,
-)
+from .adapter import QQAdapter, QQCloseError, check_qq_requirements, _coerce_list, _ssrf_redirect_guard  # noqa: F401
 from .onboard import BindStatus, build_connect_url, qr_register  # noqa: F401
 from .crypto import decrypt_secret, generate_bind_key  # noqa: F401
 from .utils import build_user_agent, get_api_headers, coerce_list  # noqa: F401
-from .chunked_upload import (  # noqa: F401
-    ChunkedUploader,
-    UploadDailyLimitExceededError,
-    UploadFileTooLargeError,
-)
+from .chunked_upload import ChunkedUploader, UploadDailyLimitExceededError, UploadFileTooLargeError  # noqa: F401
 from .keyboards import (  # noqa: F401
-    ApprovalRequest,
-    InlineKeyboard,
-    InteractionEvent,
-    build_approval_keyboard,
-    build_approval_text,
-    build_update_prompt_keyboard,
-    parse_approval_button_data,
-    parse_interaction_event,
+    ApprovalRequest, InlineKeyboard, InteractionEvent, build_approval_keyboard, build_approval_text,
+    build_update_prompt_keyboard, parse_approval_button_data, parse_interaction_event,
     parse_update_prompt_button_data,
 )
 

gateway/platforms/qqbot/chunked_upload.py

@@ -1,21 +1,10 @@
-"""QQ Bot chunked upload flow.
-
-The QQ v2 API caps inline base64 uploads (``file_data`` / ``url``) at ~10 MB.
-Files between 10 MB and ~100 MB use the three-step chunked flow::
-
-    1. POST /v2/{users|groups}/{id}/upload_prepare
-       → upload_id, block_size, and pre-signed COS part URLs.
-    2. Per part: PUT bytes to the COS URL, then POST .../upload_part_finish.
-    3. POST /v2/{users|groups}/{id}/files with {"upload_id": ...}
-       → ``file_info`` token used in a RichMedia message.
-
-Error codes (QQ Bot v2 spec): ``40093001`` — ``upload_part_finish`` retryable
-until the server's ``retry_timeout`` (or a local cap) elapses; ``40093002`` —
-daily upload quota exceeded, surfaced as :class:`UploadDailyLimitExceededError`.
-Other API/I/O failures raise ``RuntimeError``.
-
-Ported from WideLee's qqbot-agent-sdk v1.2.2 (``media_loader.py::ChunkedUploader``).
-Authorship preserved via Co-authored-by.
+"""QQ Bot chunked upload flow (files 10 MB..~100 MB; inline uploads cap at ~10 MB):
+1. POST .../upload_prepare → upload_id, block_size, pre-signed COS part URLs;
+2. per part: PUT bytes to COS, then POST .../upload_part_finish;
+3. POST .../files with {"upload_id"} → ``file_info`` for a RichMedia message.
+biz_code 40093001 = part_finish retryable until ``retry_timeout``; 40093002 = daily
+quota (UploadDailyLimitExceededError); other API/I/O failures raise RuntimeError.
+Ported from WideLee's qqbot-agent-sdk v1.2.2 (authorship via Co-authored-by).
 """
 
 from __future__ import annotations
@@ -31,63 +20,56 @@ from gateway.platforms.qqbot.constants import FILE_UPLOAD_TIMEOUT
 
 logger = logging.getLogger(__name__)
 
-
 _BIZ_CODE_DAILY_LIMIT = 40093002     # upload_prepare: daily cumulative limit
 _BIZ_CODE_PART_RETRYABLE = 40093001  # upload_part_finish: transient
-
 _DEFAULT_CONCURRENT_PARTS = 1
 _MAX_CONCURRENT_PARTS = 10
-
 _PART_UPLOAD_TIMEOUT = 300.0        # 5 minutes per COS PUT
 _PART_UPLOAD_MAX_RETRIES = 2
 _PART_FINISH_RETRY_INTERVAL = 1.0
 _PART_FINISH_DEFAULT_TIMEOUT = 120.0
 _PART_FINISH_MAX_TIMEOUT = 600.0
-
 _COMPLETE_UPLOAD_MAX_RETRIES = 2
 _COMPLETE_UPLOAD_BASE_DELAY = 2.0
+_MD5_10M_SIZE = 10_002_432  # first N bytes used for the ``md5_10m`` hash (per QQ API spec)
 
-# First 10,002,432 bytes used for the ``md5_10m`` hash (per QQ API spec).
-_MD5_10M_SIZE = 10_002_432
+# ── Exceptions ──
+
+class _UploadError(Exception):
+    def __init__(self, file_name: str, file_size: int, message: str) -> None:
+        self.file_name = file_name
+        self.file_size = file_size
+        super().__init__(message)
+
+    @property
+    def file_size_human(self) -> str:
+        return format_size(self.file_size)
 
 
-# ── Exceptions ───────────────────────────────────────────────────────
-
-class UploadDailyLimitExceededError(Exception):
+class UploadDailyLimitExceededError(_UploadError):
     """Raised when ``upload_prepare`` returns biz_code 40093002 (daily quota hit)."""
 
     def __init__(self, file_name: str, file_size: int, message: str = "") -> None:
-        self.file_name = file_name
-        self.file_size = file_size
-        super().__init__(message or f"Daily upload limit exceeded for {file_name!r}")
-
-    @property
-    def file_size_human(self) -> str:
-        return format_size(self.file_size)
+        super().__init__(file_name, file_size, message or f"Daily upload limit exceeded for {file_name!r}")
 
 
-class UploadFileTooLargeError(Exception):
+class UploadFileTooLargeError(_UploadError):
     """Raised when a file exceeds the platform per-file size limit."""
 
     def __init__(self, file_name: str, file_size: int, limit_bytes: int = 0, message: str = "") -> None:
-        self.file_name = file_name
-        self.file_size = file_size
         self.limit_bytes = limit_bytes
         limit_str = f" ({format_size(limit_bytes)})" if limit_bytes else ""
         super().__init__(
-            message or f"File {file_name!r} ({format_size(file_size)}) exceeds platform limit{limit_str}"
+            file_name, file_size,
+            message or f"File {file_name!r} ({format_size(file_size)}) exceeds platform limit{limit_str}",
         )
 
-    @property
-    def file_size_human(self) -> str:
-        return format_size(self.file_size)
-
     @property
     def limit_human(self) -> str:
         return format_size(self.limit_bytes) if self.limit_bytes else "unknown"
 
 
-# ── Prepare-response shape ───────────────────────────────────────────
+# ── Prepare-response shape ──
 
 @dataclass
 class _PreparePart:
@@ -119,18 +101,15 @@ def _parse_prepare_response(raw: Dict[str, Any]) -> _PrepareResult:
         _PreparePart(
             index=int(p.get("part_index") or p.get("index") or 0),
             presigned_url=str(p.get("presigned_url") or p.get("url") or ""),
-            block_size=int(p.get("block_size", 0)),
-        )
+            block_size=int(p.get("block_size", 0)))
         for p in raw_parts
-        if isinstance(p, dict)
-    ]
+        if isinstance(p, dict)]
     return _PrepareResult(
         upload_id=upload_id,
         block_size=block_size,
         parts=parts,
         concurrency=int(src.get("concurrency", _DEFAULT_CONCURRENT_PARTS)) or _DEFAULT_CONCURRENT_PARTS,
-        retry_timeout=float(src.get("retry_timeout", 0.0) or 0.0),
-    )
+        retry_timeout=float(src.get("retry_timeout", 0.0) or 0.0))
 
 
 def _api_path(chat_type: str, target_id: str, endpoint: str) -> str:
@@ -138,165 +117,112 @@ def _api_path(chat_type: str, target_id: str, endpoint: str) -> str:
     return f"{base}/{target_id}/{endpoint}"
 
 
-# ── Chunked upload driver ────────────────────────────────────────────
+# ── Chunked upload driver ──
+
+@dataclass
+class _Job:
+    """Per-upload state shared by the part workers."""
+    chat_type: str
+    target_id: str
+    file_path: str
+    file_size: int
+    upload_id: str = ""
+    block_size: int = 0
+    total_parts: int = 0
+    retry_timeout: float = 0.0
+    completed: int = 0
+
+    def path(self, endpoint: str) -> str:
+        return _api_path(self.chat_type, self.target_id, endpoint)
+
 
 class ChunkedUploader:
-    """Run the prepare → PUT parts → complete sequence.
-
-    :param api_request: Bound ``_api_request(method, path, body=..., timeout=...)``
-        coroutine from the adapter (passed in rather than imported to avoid
-        circular imports). Must raise ``RuntimeError`` with the biz_code in
-        the message on API errors.
-    :param http_put: Coroutine ``(url, data, headers) -> httpx-like response``
-        for COS part uploads.
-    :param log_tag: Log prefix.
-    """
+    """Run the prepare → PUT parts → complete sequence. ``api_request`` is the adapter's
+    bound ``_api_request(method, path, body=..., timeout=...)`` (injected to avoid a
+    circular import; must raise RuntimeError with the biz_code in the message);
+    ``http_put`` is ``(url, data, headers) -> httpx-like response`` for COS PUTs."""
 
     def __init__(
-        self,
-        api_request: Callable[..., Awaitable[Dict[str, Any]]],
-        http_put: Callable[..., Awaitable[Any]],
-        log_tag: str = "QQBot",
-    ) -> None:
+        self, api_request: Callable[..., Awaitable[Dict[str, Any]]], http_put: Callable[..., Awaitable[Any]],
+        log_tag: str = "QQBot") -> None:
         self._api_request = api_request
         self._http_put = http_put
         self._log_tag = log_tag
 
-    async def upload(
-        self,
-        chat_type: str,
-        target_id: str,
-        file_path: str,
-        file_type: int,
-        file_name: str,
-    ) -> Dict[str, Any]:
-        """Run the full chunked upload and return the ``complete_upload`` response.
+    async def _post(self, job: _Job, endpoint: str, body: Dict[str, Any]) -> Dict[str, Any]:
+        return await self._api_request("POST", job.path(endpoint), body=body, timeout=FILE_UPLOAD_TIMEOUT)
 
-        :param chat_type: ``'c2c'`` or ``'group'``.
-        :param file_type: ``MEDIA_TYPE_*`` constant.
-        :returns: Raw ``complete_upload`` response dict (contains ``file_info``).
-        :raises UploadDailyLimitExceededError: On biz_code 40093002.
-        :raises UploadFileTooLargeError: When the file exceeds the platform limit.
-        :raises RuntimeError: On other API or I/O failures.
-        """
+    async def upload(self, chat_type: str, target_id: str, file_path: str, file_type: int, file_name: str) -> Dict[str, Any]:
+        """Run the full chunked upload (``chat_type`` 'c2c'|'group', ``file_type`` MEDIA_TYPE_*)
+        and return the raw ``complete_upload`` response (contains ``file_info``).
+        Raises UploadDailyLimitExceededError (40093002), UploadFileTooLargeError, RuntimeError."""
         if chat_type not in {"c2c", "group"}:
             raise ValueError(f"ChunkedUploader: unsupported chat_type {chat_type!r}")
 
-        file_size = Path(file_path).stat().st_size
+        job = _Job(chat_type, target_id, file_path, Path(file_path).stat().st_size)
         logger.info(
             "[%s] Chunked upload start: file=%s size=%s type=%d",
-            self._log_tag, file_name, format_size(file_size), file_type,
-        )
+            self._log_tag, file_name, format_size(job.file_size), file_type)
 
         # Hashing is blocking I/O → executor.
-        hashes = await asyncio.get_running_loop().run_in_executor(
-            None, _compute_file_hashes, file_path, file_size
-        )
-        prepare = await self._prepare(chat_type, target_id, file_type, file_name, file_size, hashes)
+        hashes = await asyncio.get_running_loop().run_in_executor(None, _compute_file_hashes, file_path, job.file_size)
+        prepare = await self._prepare(job, file_type, file_name, hashes)
         max_concurrent = min(prepare.concurrency, _MAX_CONCURRENT_PARTS)
-        retry_timeout = min(
+        job.upload_id, job.block_size, job.total_parts = prepare.upload_id, prepare.block_size, len(prepare.parts)
+        job.retry_timeout = min(
             prepare.retry_timeout if prepare.retry_timeout > 0 else _PART_FINISH_DEFAULT_TIMEOUT,
-            _PART_FINISH_MAX_TIMEOUT,
-        )
+            _PART_FINISH_MAX_TIMEOUT)
         logger.info(
             "[%s] Prepared: upload_id=%s block_size=%s parts=%d concurrency=%d",
-            self._log_tag, prepare.upload_id, format_size(prepare.block_size),
-            len(prepare.parts), max_concurrent,
-        )
+            self._log_tag, job.upload_id, format_size(job.block_size), job.total_parts, max_concurrent)
 
-        total_parts = len(prepare.parts)
-        completed = [0]  # shared counter for progress logging
         sem = asyncio.Semaphore(max(max_concurrent, 1))
 
         async def _run(part: _PreparePart) -> None:
             async with sem:
-                await self._upload_one_part(
-                    chat_type, target_id, file_path, file_size, prepare.upload_id,
-                    prepare.block_size, part, retry_timeout, total_parts, completed,
-                )
+                await self._upload_one_part(job, part)
 
         await asyncio.gather(*(_run(p) for p in prepare.parts))
-        logger.info("[%s] All %d parts uploaded, completing…", self._log_tag, total_parts)
-        return await self._complete(chat_type, target_id, prepare.upload_id)
+        logger.info("[%s] All %d parts uploaded, completing…", self._log_tag, job.total_parts)
+        return await self._complete(job)
 
-    async def _prepare(
-        self,
-        chat_type: str,
-        target_id: str,
-        file_type: int,
-        file_name: str,
-        file_size: int,
-        hashes: Dict[str, str],
-    ) -> _PrepareResult:
+    async def _prepare(self, job: _Job, file_type: int, file_name: str, hashes: Dict[str, str]) -> _PrepareResult:
         body = {
-            "file_type": file_type,
-            "file_name": file_name,
-            "file_size": file_size,
-            "md5": hashes["md5"],
-            "sha1": hashes["sha1"],
-            "md5_10m": hashes["md5_10m"],
-        }
+            "file_type": file_type, "file_name": file_name, "file_size": job.file_size,
+            "md5": hashes["md5"], "sha1": hashes["sha1"], "md5_10m": hashes["md5_10m"]}
         try:
-            raw = await self._api_request(
-                "POST", _api_path(chat_type, target_id, "upload_prepare"),
-                body=body, timeout=FILE_UPLOAD_TIMEOUT,
-            )
+            raw = await self._post(job, "upload_prepare", body)
         except RuntimeError as exc:
             err_msg = str(exc)
             if f"{_BIZ_CODE_DAILY_LIMIT}" in err_msg:
-                raise UploadDailyLimitExceededError(file_name, file_size, err_msg) from exc
+                raise UploadDailyLimitExceededError(file_name, job.file_size, err_msg) from exc
             raise
         return _parse_prepare_response(raw)
 
-    async def _upload_one_part(
-        self,
-        chat_type: str,
-        target_id: str,
-        file_path: str,
-        file_size: int,
-        upload_id: str,
-        rsp_block_size: int,
-        part: _PreparePart,
-        retry_timeout: float,
-        total_parts: int,
-        completed: List[int],
-    ) -> None:
+    async def _upload_one_part(self, job: _Job, part: _PreparePart) -> None:
         """PUT one part to COS, then call ``upload_part_finish``."""
-        part_index = part.index
+        part_index, total_parts = part.index, job.total_parts
         # Per-part block_size wins; fall back to the response-level value.
-        actual_block_size = part.block_size if part.block_size > 0 else rsp_block_size
-        offset = (part_index - 1) * rsp_block_size
-        length = min(actual_block_size, file_size - offset)
+        actual_block_size = part.block_size if part.block_size > 0 else job.block_size
+        offset = (part_index - 1) * job.block_size
+        length = min(actual_block_size, job.file_size - offset)
 
-        data = await asyncio.get_running_loop().run_in_executor(
-            None, _read_file_chunk, file_path, offset, length
-        )
+        data = await asyncio.get_running_loop().run_in_executor(None, _read_file_chunk, job.file_path, offset, length)
         md5_hex = hashlib.md5(data).hexdigest()
         logger.debug(
             "[%s] Part %d/%d: uploading %s (offset=%d md5=%s)",
-            self._log_tag, part_index, total_parts, format_size(length), offset, md5_hex,
-        )
+            self._log_tag, part_index, total_parts, format_size(length), offset, md5_hex)
 
         await self._put_to_presigned_url(part.presigned_url, data, part_index, total_parts)
-        await self._part_finish_with_retry(
-            chat_type, target_id, upload_id, part_index, length, md5_hex, retry_timeout,
-        )
+        await self._part_finish_with_retry(job, part_index, length, md5_hex)
 
-        completed[0] += 1
+        job.completed += 1
         logger.debug(
-            "[%s] Part %d/%d done (%d/%d total)",
-            self._log_tag, part_index, total_parts, completed[0], total_parts,
-        )
+            "[%s] Part %d/%d done (%d/%d total)", self._log_tag, part_index, total_parts, job.completed, total_parts)
 
     async def _with_retries(
-        self,
-        attempt_fn: Callable[[], Awaitable[Any]],
-        *,
-        max_retries: int,
-        base_delay: float,
-        label: str,
-        failure_label: str,
-    ) -> Any:
+        self, attempt_fn: Callable[[], Awaitable[Any]], *, max_retries: int, base_delay: float, label: str,
+        failure_label: str) -> Any:
         """Run *attempt_fn* up to ``max_retries + 1`` times with exponential backoff."""
         last_exc: Exception | None = None
         for attempt in range(max_retries + 1):
@@ -308,8 +234,7 @@ class ChunkedUploader:
                     delay = base_delay * (2 ** attempt)
                     logger.warning(
                         "[%s] %s attempt %d failed, retry in %.1fs: %s",
-                        self._log_tag, label, attempt + 1, delay, exc,
-                    )
+                        self._log_tag, label, attempt + 1, delay, exc)
                     await asyncio.sleep(delay)
         raise RuntimeError(f"{failure_label} failed after {max_retries + 1} attempts: {last_exc}")
 
@@ -319,77 +244,55 @@ class ChunkedUploader:
         async def _attempt() -> None:
             resp = await asyncio.wait_for(
                 self._http_put(url, data=data, headers={"Content-Length": str(len(data))}),
-                timeout=_PART_UPLOAD_TIMEOUT,
-            )
+                timeout=_PART_UPLOAD_TIMEOUT)
             status = getattr(resp, "status_code", 0)
             if 200 <= status < 300:
                 logger.debug("[%s] PUT part %d/%d: %d OK", self._log_tag, part_index, total_parts, status)
                 return
-            body_preview = ""
             try:
                 body_preview = getattr(resp, "text", "")[:200]
             except Exception:  # pragma: no cover — defensive
-                pass
+                body_preview = ""
             raise RuntimeError(f"COS PUT returned {status}: {body_preview}")
 
         await self._with_retries(
             _attempt, max_retries=_PART_UPLOAD_MAX_RETRIES, base_delay=1.0,
-            label=f"PUT part {part_index}/{total_parts}",
-            failure_label=f"Part {part_index}/{total_parts} upload",
-        )
+            label=f"PUT part {part_index}/{total_parts}", failure_label=f"Part {part_index}/{total_parts} upload")
 
-    async def _part_finish_with_retry(
-        self,
-        chat_type: str,
-        target_id: str,
-        upload_id: str,
-        part_index: int,
-        block_size: int,
-        md5: str,
-        retry_timeout: float,
-    ) -> None:
-        """Call ``upload_part_finish``, retrying on biz_code 40093001 until *retry_timeout*."""
-        path = _api_path(chat_type, target_id, "upload_part_finish")
-        body = {"upload_id": upload_id, "part_index": part_index, "block_size": block_size, "md5": md5}
+    async def _part_finish_with_retry(self, job: _Job, part_index: int, block_size: int, md5: str) -> None:
+        """Call ``upload_part_finish``, retrying on biz_code 40093001 until ``job.retry_timeout``."""
+        body = {"upload_id": job.upload_id, "part_index": part_index, "block_size": block_size, "md5": md5}
         loop = asyncio.get_running_loop()
         start = loop.time()
         attempt = 0
         while True:
             try:
-                await self._api_request("POST", path, body=body, timeout=FILE_UPLOAD_TIMEOUT)
+                await self._post(job, "upload_part_finish", body)
                 return
             except RuntimeError as exc:
                 if f"{_BIZ_CODE_PART_RETRYABLE}" not in str(exc):
                     raise
                 elapsed = loop.time() - start
-                if elapsed >= retry_timeout:
+                if elapsed >= job.retry_timeout:
                     raise RuntimeError(
                         f"upload_part_finish persistent retry timed out "
-                        f"after {retry_timeout:.0f}s ({attempt} retries): {exc}"
-                    ) from exc
+                        f"after {job.retry_timeout:.0f}s ({attempt} retries): {exc}") from exc
                 attempt += 1
                 logger.debug(
                     "[%s] part_finish retryable error, attempt %d, elapsed=%.1fs: %s",
-                    self._log_tag, attempt, elapsed, exc,
-                )
+                    self._log_tag, attempt, elapsed, exc)
                 await asyncio.sleep(_PART_FINISH_RETRY_INTERVAL)
 
-    async def _complete(self, chat_type: str, target_id: str, upload_id: str) -> Dict[str, Any]:
-        """Call ``complete_upload`` with retry.
-
-        Reuses the ``/files`` endpoint (same as the simple URL-based upload) but
-        signals the chunked-completion path by sending only ``upload_id``.
-        """
-        path = _api_path(chat_type, target_id, "files")
-        body = {"upload_id": upload_id}
+    async def _complete(self, job: _Job) -> Dict[str, Any]:
+        """Call ``complete_upload`` with retry — the ``/files`` endpoint (same as the simple
+        URL upload) selects the chunked-completion path when only ``upload_id`` is sent."""
         return await self._with_retries(
-            lambda: self._api_request("POST", path, body=body, timeout=FILE_UPLOAD_TIMEOUT),
+            lambda: self._post(job, "files", {"upload_id": job.upload_id}),
             max_retries=_COMPLETE_UPLOAD_MAX_RETRIES, base_delay=_COMPLETE_UPLOAD_BASE_DELAY,
-            label="complete_upload", failure_label="complete_upload",
-        )
+            label="complete_upload", failure_label="complete_upload")
 
 
-# ── Helpers (module-level for testability) ───────────────────────────
+# ── Helpers (module-level for testability) ──
 
 def format_size(size_bytes: int) -> str:
     """Return a human-readable file size string (e.g. ``'12.3 MB'``)."""
@@ -409,8 +312,7 @@ def _read_file_chunk(file_path: str, offset: int, length: int) -> bytes:
         if len(data) != length:
             raise IOError(
                 f"Short read from {file_path}: expected {length} bytes at "
-                f"offset {offset}, got {len(data)} (file may be truncated)"
-            )
+                f"offset {offset}, got {len(data)} (file may be truncated)")
         return data
 
 
@@ -435,5 +337,4 @@ def _compute_file_hashes(file_path: str, file_size: int) -> Dict[str, str]:
         "md5": full_md5,
         "sha1": sha1.hexdigest(),
         # For small files the "10m" hash is just the full md5.
-        "md5_10m": md5_10m.hexdigest() if need_10m else full_md5,
-    }
+        "md5_10m": md5_10m.hexdigest() if need_10m else full_md5}

gateway/platforms/qqbot/constants.py

@@ -4,48 +4,37 @@ from __future__ import annotations
 
 import os
 
-# Bump on functional changes to the adapter package.
-QQBOT_VERSION = "1.1.0"
-
-# ── API endpoints ──
+QQBOT_VERSION = "1.1.0"  # bump on functional changes to the adapter package
 # Portal domain is overridable (QQ_PORTAL_HOST) for corporate proxies / test environments.
 PORTAL_HOST = os.getenv("QQ_PORTAL_HOST", "q.qq.com")
 
 API_BASE = "https://api.sgroup.qq.com"
 TOKEN_URL = "https://bots.qq.com/app/getAppAccessToken"
 GATEWAY_URL_PATH = "/gateway"
-
-# QR-code onboard endpoints (on the portal host)
 ONBOARD_CREATE_PATH = "/lite/create_bind_task"
 ONBOARD_POLL_PATH = "/lite/poll_bind_result"
 QR_URL_TEMPLATE = "https://q.qq.com/qqbot/openclaw/connect.html?task_id={task_id}&_wv=2&source=hermes"
 
-# ── Timeouts & retry ──
 DEFAULT_API_TIMEOUT = 30.0
 FILE_UPLOAD_TIMEOUT = 120.0
 CONNECT_TIMEOUT_SECONDS = 20.0
-
 RECONNECT_BACKOFF = [2, 5, 10, 30, 60]
 MAX_RECONNECT_ATTEMPTS = 100
 RATE_LIMIT_DELAY = 60  # seconds
 QUICK_DISCONNECT_THRESHOLD = 5.0  # seconds
 MAX_QUICK_DISCONNECT_COUNT = 3
-
 ONBOARD_POLL_INTERVAL = 2.0  # seconds between poll_bind_result calls
 ONBOARD_API_TIMEOUT = 10.0
 
-# ── Message limits ──
 MAX_MESSAGE_LENGTH = 4000
 DEDUP_WINDOW_SECONDS = 300
 DEDUP_MAX_SIZE = 1000
 
-# ── QQ Bot message types ──
+# QQ Bot message types / file media types
 MSG_TYPE_TEXT = 0
 MSG_TYPE_MARKDOWN = 2
 MSG_TYPE_MEDIA = 7
 MSG_TYPE_INPUT_NOTIFY = 6
-
-# ── QQ Bot file media types ──
 MEDIA_TYPE_IMAGE = 1
 MEDIA_TYPE_VIDEO = 2
 MEDIA_TYPE_VOICE = 3

gateway/platforms/qqbot/crypto.py

@@ -7,12 +7,8 @@ import os
 
 
 def generate_bind_key() -> str:
-    """Generate a random 256-bit AES key as base64.
-
-    Passed to ``create_bind_task`` so the server encrypts the bot's
-    *client_secret*; only this CLI holds the key, so the secret never travels
-    in plaintext.
-    """
+    """Random 256-bit AES key (base64) passed to ``create_bind_task`` so the server
+    encrypts the bot's *client_secret*; only this CLI holds the key."""
     return base64.b64encode(os.urandom(32)).decode()
 
 
@@ -20,6 +16,5 @@ def decrypt_secret(encrypted_base64: str, key_base64: str) -> str:
     """Decrypt ``bot_encrypt_secret`` (base64 of ``IV(12) ‖ ciphertext ‖ tag(16)``) to a UTF-8 string."""
     from cryptography.hazmat.primitives.ciphers.aead import AESGCM
 
-    raw = base64.b64decode(encrypted_base64)
-    # AESGCM expects ciphertext + tag concatenated.
+    raw = base64.b64decode(encrypted_base64)  # AESGCM expects ciphertext + tag concatenated
     return AESGCM(base64.b64decode(key_base64)).decrypt(raw[:12], raw[12:], None).decode("utf-8")

gateway/platforms/qqbot/keyboards.py

@@ -1,17 +1,10 @@
 """QQ Bot inline keyboards + approval / update-prompt helpers.
 
-QQ Bot v2 attaches inline keyboards to outbound messages. A button click
-dispatches an ``INTERACTION_CREATE`` gateway event carrying the button's
-``data`` payload; the bot must ACK promptly via ``PUT /interactions/{id}`` or
-the user sees an error indicator on the button.
-
-``button_data`` formats::
-
-    approve:<session_key>:<decision>      # decision = allow-once|allow-always|deny
-    update_prompt:<answer>                # answer = y|n
-
-Ported from WideLee's qqbot-agent-sdk v1.2.2 (``approval.py`` + ``dto.py``
-keyboard types). Authorship preserved via Co-authored-by.
+A button click dispatches an ``INTERACTION_CREATE`` event carrying the button's
+``data``; the bot must ACK promptly via ``PUT /interactions/{id}`` or the user
+sees an error indicator. ``button_data`` formats: ``approve:<session_key>:<decision>``
+(decision = allow-once|allow-always|deny) and ``update_prompt:<answer>`` (y|n).
+Ported from WideLee's qqbot-agent-sdk v1.2.2 (authorship via Co-authored-by).
 """
 
 from __future__ import annotations
@@ -28,16 +21,13 @@ UPDATE_PROMPT_PREFIX = "update_prompt:"
 _APPROVAL_DATA_RE = re.compile(r"^approve:(.+):(allow-once|allow-always|deny)$")
 _UPDATE_PROMPT_RE = re.compile(r"^update_prompt:(y|n)$")
 
-
-# ── Keyboard dataclasses ─────────────────────────────────────────────
+# ── Keyboard dataclasses ──
 
 def _to_dict(value: Any) -> Any:
     """Serialize a dataclass tree in field-declaration order (the wire shape)."""
     if is_dataclass(value):
         return {f.name: _to_dict(getattr(value, f.name)) for f in fields(value)}
-    if isinstance(value, list):
-        return [_to_dict(v) for v in value]
-    return value
+    return [_to_dict(v) for v in value] if isinstance(value, list) else value
 
 
 class _Serializable:
@@ -53,9 +43,8 @@ class KeyboardButtonPermission(_Serializable):
 
 @dataclass
 class KeyboardButtonAction(_Serializable):
-    """Click behaviour: ``type`` 1 = Callback (INTERACTION_CREATE with ``data``), 2 = Link.
-    ``click_limit=1`` = single-use.
-    """
+    """Click behaviour: ``type`` 1 = Callback (INTERACTION_CREATE with ``data``), 2 = Link;
+    ``click_limit=1`` = single-use."""
     type: int
     data: str
     permission: KeyboardButtonPermission = field(default_factory=KeyboardButtonPermission)
@@ -95,12 +84,12 @@ class InlineKeyboard(_Serializable):
     content: KeyboardContent = field(default_factory=KeyboardContent)
 
 
-# ── INTERACTION_CREATE parsing ───────────────────────────────────────
+# ── INTERACTION_CREATE parsing ──
 
 def parse_approval_button_data(button_data: str) -> Optional[tuple[str, str]]:
     """Parse approval ``button_data`` into ``(session_key, decision)`` or ``None``."""
     m = _APPROVAL_DATA_RE.match(button_data or "")
-    return (m.group(1), m.group(2)) if m else None
+    return m.groups() if m else None
 
 
 def parse_update_prompt_button_data(button_data: str) -> Optional[str]:
@@ -109,54 +98,44 @@ def parse_update_prompt_button_data(button_data: str) -> Optional[str]:
     return m.group(1) if m else None
 
 
-# ── Keyboard builders ────────────────────────────────────────────────
+# ── Keyboard builders ──
 
-def _make_callback_button(
-    btn_id: str, label: str, visited_label: str, data: str, style: int, group_id: str,
-) -> KeyboardButton:
-    return KeyboardButton(
-        id=btn_id,
-        render_data=KeyboardButtonRenderData(label=label, visited_label=visited_label, style=style),
-        action=KeyboardButtonAction(type=1, data=data),
-        group_id=group_id,
-    )
-
-
-def _single_row_keyboard(buttons: List[KeyboardButton]) -> InlineKeyboard:
-    return InlineKeyboard(content=KeyboardContent(rows=[KeyboardRow(buttons=buttons)]))
+def _single_row_keyboard(group_id: str, *buttons: tuple) -> InlineKeyboard:
+    """One row of callback buttons from ``(id, label, visited_label, data, style)`` tuples."""
+    row = KeyboardRow(buttons=[
+        KeyboardButton(
+            id=btn_id, group_id=group_id, action=KeyboardButtonAction(type=1, data=data),
+            render_data=KeyboardButtonRenderData(label=label, visited_label=visited, style=style))
+        for btn_id, label, visited, data, style in buttons])
+    return InlineKeyboard(content=KeyboardContent(rows=[row]))
 
 
 def build_approval_keyboard(session_key: str, *, allow_permanent: bool = True) -> InlineKeyboard:
     """Build ``[✅ 允许一次] [⭐ 始终允许] [❌ 拒绝]`` (one group, so a click greys the rest).
-
-    The ⭐ button is hidden when persistent scope is unavailable. *session_key*
-    is embedded in ``button_data`` so the decision routes to the right approval.
-    """
+    ⭐ is hidden when persistent scope is unavailable; *session_key* is embedded in
+    ``button_data`` so the decision routes to the right approval."""
     prefix = f"{APPROVAL_BUTTON_PREFIX}{session_key}"
-    buttons = [_make_callback_button("allow", "✅ 允许一次", "已允许", f"{prefix}:allow-once", 1, "approval")]
+    buttons = [("allow", "✅ 允许一次", "已允许", f"{prefix}:allow-once", 1)]
     if allow_permanent:
-        buttons.append(_make_callback_button("always", "⭐ 始终允许", "已始终允许", f"{prefix}:allow-always", 1, "approval"))
-    buttons.append(_make_callback_button("deny", "❌ 拒绝", "已拒绝", f"{prefix}:deny", 0, "approval"))
-    return _single_row_keyboard(buttons)
+        buttons.append(("always", "⭐ 始终允许", "已始终允许", f"{prefix}:allow-always", 1))
+    buttons.append(("deny", "❌ 拒绝", "已拒绝", f"{prefix}:deny", 0))
+    return _single_row_keyboard("approval", *buttons)
 
 
 def build_update_prompt_keyboard() -> InlineKeyboard:
     """Build a Yes/No keyboard for update confirmation prompts."""
-    return _single_row_keyboard([
-        _make_callback_button("yes", "✓ 确认", "已确认", f"{UPDATE_PROMPT_PREFIX}y", 1, "update_prompt"),
-        _make_callback_button("no", "✗ 取消", "已取消", f"{UPDATE_PROMPT_PREFIX}n", 0, "update_prompt"),
-    ])
+    return _single_row_keyboard(
+        "update_prompt",
+        ("yes", "✓ 确认", "已确认", f"{UPDATE_PROMPT_PREFIX}y", 1),
+        ("no", "✗ 取消", "已取消", f"{UPDATE_PROMPT_PREFIX}n", 0))
 
 
-# ── ApprovalRequest + text builder ───────────────────────────────────
+# ── ApprovalRequest + text builder ──
 
 @dataclass
 class ApprovalRequest:
-    """Approval-request display data.
-
-    ``command_preview`` / ``cwd`` are set for exec approvals, ``tool_name`` for
-    plugin approvals; ``severity`` is ``'critical' | 'info' | ''``.
-    """
+    """Approval-request display data. ``command_preview`` / ``cwd`` are set for exec
+    approvals, ``tool_name`` for plugin approvals; ``severity`` is ``'critical' | 'info' | ''``."""
     session_key: str
     title: str
     description: str = ""
@@ -193,14 +172,11 @@ def build_approval_text(req: ApprovalRequest) -> str:
     return "\n".join(lines)
 
 
-# ── INTERACTION_CREATE event shape ───────────────────────────────────
+# ── INTERACTION_CREATE event shape ──
 
 @dataclass
 class InteractionEvent:
-    """Parsed ``INTERACTION_CREATE`` payload.
-
-    See https://bot.q.qq.com/wiki/develop/api-v2/dev-prepare/interface-framework/event-emit.html
-    """
+    """Parsed ``INTERACTION_CREATE`` payload (api-v2 event-emit docs)."""
     id: str = ""            # required for the ``PUT /interactions/{id}`` ACK
     type: int = 0           # event type code (11 = message button)
     chat_type: int = 0      # 0 = guild, 1 = group, 2 = c2c
@@ -229,16 +205,11 @@ def parse_interaction_event(raw: Dict[str, Any]) -> InteractionEvent:
     resolved = data_raw.get("resolved") or {}
     scene_code = int(raw.get("chat_type", 0) or 0)
     return InteractionEvent(
-        id=str(raw.get("id", "")),
-        type=int(data_raw.get("type", 0) or 0),
-        chat_type=scene_code,
-        scene=_SCENE_NAMES.get(scene_code, ""),
-        group_openid=str(raw.get("group_openid", "")),
+        id=str(raw.get("id", "")), type=int(data_raw.get("type", 0) or 0), chat_type=scene_code,
+        scene=_SCENE_NAMES.get(scene_code, ""), group_openid=str(raw.get("group_openid", "")),
         group_member_openid=str(raw.get("group_member_openid", "")),
-        user_openid=str(raw.get("user_openid", "")),
-        channel_id=str(raw.get("channel_id", "")),
-        guild_id=str(raw.get("guild_id", "")),
-        button_data=str(resolved.get("button_data", "")),
+        user_openid=str(raw.get("user_openid", "")), channel_id=str(raw.get("channel_id", "")),
+        guild_id=str(raw.get("guild_id", "")), button_data=str(resolved.get("button_data", "")),
         button_id=str(resolved.get("button_id", "")),
         resolver_user_id=str(resolved.get("user_id", "")),
     )

gateway/platforms/qqbot/onboard.py

@@ -1,11 +1,6 @@
-"""QQBot scan-to-configure (QR code onboard) module.
-
-Mirrors the Feishu onboarding pattern: synchronous HTTP + one public entry-point
-``qr_register()`` (create task → display QR → poll → decrypt credentials) against
-the ``q.qq.com`` ``create_bind_task`` / ``poll_bind_result`` APIs.
-
-Reference: https://bot.q.qq.com/wiki/develop/api-v2/
-"""
+"""QQBot scan-to-configure (QR code onboard). Mirrors the Feishu pattern: synchronous
+HTTP + one entry-point ``qr_register()`` (create task → display QR → poll → decrypt
+credentials) against ``q.qq.com`` ``create_bind_task`` / ``poll_bind_result``."""
 
 from __future__ import annotations
 
@@ -16,12 +11,7 @@ from typing import Optional, Tuple
 from urllib.parse import quote
 
 from .constants import (
-    ONBOARD_API_TIMEOUT,
-    ONBOARD_CREATE_PATH,
-    ONBOARD_POLL_INTERVAL,
-    ONBOARD_POLL_PATH,
-    PORTAL_HOST,
-    QR_URL_TEMPLATE,
+    ONBOARD_API_TIMEOUT, ONBOARD_CREATE_PATH, ONBOARD_POLL_INTERVAL, ONBOARD_POLL_PATH, PORTAL_HOST, QR_URL_TEMPLATE,
 )
 from .crypto import decrypt_secret, generate_bind_key
 from .utils import get_api_headers
@@ -31,11 +21,7 @@ logger = logging.getLogger(__name__)
 
 class BindStatus(IntEnum):
     """Status codes returned by ``_poll_bind_result``."""
-
-    NONE = 0
-    PENDING = 1
-    COMPLETED = 2
-    EXPIRED = 3
+    NONE, PENDING, COMPLETED, EXPIRED = 0, 1, 2, 3
 
 
 try:
@@ -84,14 +70,8 @@ def _create_bind_task(timeout: float = ONBOARD_API_TIMEOUT) -> Tuple[str, str]:
 
 def _poll_bind_result(task_id: str, timeout: float = ONBOARD_API_TIMEOUT) -> Tuple[BindStatus, str, str, str]:
     """Poll *task_id*; returns ``(status, bot_appid, bot_encrypt_secret, user_openid)``."""
-    data = _portal_post(ONBOARD_POLL_PATH, {"task_id": task_id}, timeout, "poll_bind_result failed")
-    d = data.get("data", {})
-    return (
-        BindStatus(d.get("status", 0)),
-        str(d.get("bot_appid", "")),
-        d.get("bot_encrypt_secret", ""),
-        d.get("user_openid", ""),
-    )
+    d = _portal_post(ONBOARD_POLL_PATH, {"task_id": task_id}, timeout, "poll_bind_result failed").get("data", {})
+    return BindStatus(d.get("status", 0)), str(d.get("bot_appid", "")), d.get("bot_encrypt_secret", ""), d.get("user_openid", "")
 
 
 def build_connect_url(task_id: str) -> str:
@@ -103,13 +83,8 @@ _MAX_REFRESHES = 3
 
 
 def qr_register(timeout_seconds: int = 600) -> Optional[dict]:
-    """Run the QQBot scan-to-configure QR registration flow.
-
-    Unexpected errors propagate to the caller.
-
-    :returns: ``{"app_id", "client_secret", "user_openid"}`` on success, or
-        ``None`` on failure / expiry / cancellation.
-    """
+    """Run the QR registration flow; returns ``{"app_id", "client_secret", "user_openid"}``
+    or None on failure / expiry / cancellation. Unexpected errors propagate."""
     deadline = time.monotonic() + timeout_seconds
 
     for refresh_count in range(_MAX_REFRESHES + 1):

gateway/platforms/qqbot/utils.py

@@ -21,31 +21,19 @@ def _get_hermes_version() -> str:
 def build_user_agent() -> str:
     """``QQBotAdapter/<qqbot_version> (Python/<py_version>; <os>; Hermes/<hermes_version>)``."""
     v = sys.version_info
-    return (
-        f"QQBotAdapter/{QQBOT_VERSION} (Python/{v.major}.{v.minor}.{v.micro}; "
-        f"{platform.system().lower()}; Hermes/{_get_hermes_version()})"
-    )
+    return (f"QQBotAdapter/{QQBOT_VERSION} (Python/{v.major}.{v.minor}.{v.micro}; "
+            f"{platform.system().lower()}; Hermes/{_get_hermes_version()})")
 
 
 def get_api_headers() -> Dict[str, str]:
-    """Standard QQBot API headers.
-
-    ``q.qq.com`` requires ``Accept: application/json`` — without it the server
-    returns a JavaScript anti-bot challenge page.
-    """
-    return {
-        "Content-Type": "application/json",
-        "Accept": "application/json",
-        "User-Agent": build_user_agent(),
-    }
+    """Standard QQBot API headers. ``q.qq.com`` requires ``Accept: application/json``
+    — without it the server returns a JavaScript anti-bot challenge page."""
+    return {"Content-Type": "application/json", "Accept": "application/json", "User-Agent": build_user_agent()}
 
 
 def coerce_list(value: Any) -> List[str]:
     """Coerce a comma-separated string / list / tuple / set / scalar into a trimmed string list."""
     if value is None:
         return []
-    if isinstance(value, str):
-        return [item.strip() for item in value.split(",") if item.strip()]
-    if isinstance(value, (list, tuple, set)):
-        return [str(item).strip() for item in value if str(item).strip()]
-    return [str(value).strip()] if str(value).strip() else []
+    items = value.split(",") if isinstance(value, str) else value if isinstance(value, (list, tuple, set)) else [value]
+    return [s for s in (str(item).strip() for item in items) if s]