feat(acquisition): dispatch managed ChatGPT capture

backend/acquisition/capabilities.py

@@ -101,15 +101,15 @@ async def _registration_is_available(
     return patch_rc != 0 and registration.route_probe_error in patch_output
 
 
-def _anonymous_profile_available() -> bool:
+def _profile_available(profile_kind: str) -> bool:
     from backend.browser_pool import get_pool
 
     try:
         pool = get_pool()
     except RuntimeError:
         return False
     return any(
-        pool.get_profile_kind(endpoint) == "anonymous" for endpoint in pool.endpoints
+        pool.get_profile_kind(endpoint) == profile_kind for endpoint in pool.endpoints
     )
 
 
@@ -118,19 +118,27 @@ async def probe_capabilities() -> list[CapabilityDescriptor]:
     if not await _runtime_is_installed():
         return []
 
-    ready = _anonymous_profile_available()
     descriptors = []
     for registration in list_capability_registrations():
         if not await _registration_is_available(registration):
             continue
+        ready = _profile_available(registration.required_profile_kind)
         descriptors.append(
             CapabilityDescriptor(
                 capability_id=registration.capability_id,
                 capability_version=registration.capability_version,
                 output_schema_version=registration.output_schema_version,
                 ready=ready,
                 runtime=registration.runtime_identity(),
-                unavailable_reason=None if ready else "no_clean_profile",
+                unavailable_reason=(
+                    None
+                    if ready
+                    else (
+                        "no_clean_profile"
+                        if registration.required_profile_kind == "anonymous"
+                        else f"no_{registration.required_profile_kind}_profile"
+                    )
+                ),
             )
         )
     return descriptors

backend/acquisition/registry.py

@@ -1,9 +1,11 @@
 """Versioned managed-acquisition capabilities backed by real OpenCLI commands."""
 
 from dataclasses import dataclass
+from typing import Any
 
-OHMYOPENCLI_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53"
+OHMYOPENCLI_COMMIT = "54f729d6447238918c4098e322d80b1c64ab8a7a"
 OFFICIAL_SITE_CAPABILITY_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53"
+CHAT_AI_CAPABILITY_COMMIT = "54f729d6447238918c4098e322d80b1c64ab8a7a"
 OPENCLI_VERSION = "1.8.5"
 
 
@@ -13,7 +15,7 @@ class CapabilityRegistration:
     capability_version: str
     output_schema_version: str
     source_commit: str
-    invocation: dict[str, str]
+    invocation: dict[str, Any]
     probe_args: tuple[str, ...]
     help_marker: str
     route_probe_args: tuple[str, ...]
@@ -59,6 +61,29 @@ def runtime_identity(self) -> dict[str, str]:
         ),
         route_probe_error="CDP not reachable at http://127.0.0.1:9",
     ),
+    CapabilityRegistration(
+        capability_id="chat-ai.capture",
+        capability_version="1.0.0",
+        output_schema_version="1",
+        source_commit=CHAT_AI_CAPABILITY_COMMIT,
+        invocation={
+            "site": "chatgpt",
+            "command": "capture",
+            "format": "json",
+            "args": {"timeout": 180},
+        },
+        probe_args=("chatgpt", "capture", "--help"),
+        help_marker="chat-ai.capture@1",
+        route_probe_args=(
+            "chatgpt",
+            "capture",
+            "如何选择适合团队的知识库工具？",
+            "-f",
+            "json",
+        ),
+        route_probe_error="CDP not reachable at http://127.0.0.1:9",
+        required_profile_kind="authenticated",
+    ),
 )
 
 

backend/acquisition/runner.py

@@ -14,6 +14,7 @@
 from backend.browser_pool import (
     LocalBrowserPool,
     NoCleanProfileError,
+    NoMatchingProfileError,
     RedisBrowserPool,
 )
 from backend.models.acquisition import AcquisitionExecution, AcquisitionExecutionStatus
@@ -301,16 +302,17 @@ async def run_acquisition_execution(
 
     from backend.security.url_guard import SSRFValidationError, avalidate_public_url
 
-    try:
-        input_payload["url"] = await avalidate_public_url(input_payload.get("url"))
-    except SSRFValidationError as exc:
-        await _fail_execution(
-            execution_id,
-            {"code": "ssrf_rejected", "message": str(exc)},
-            session_factory,
-            lease_owner,
-        )
-        return
+    if "url" in input_payload:
+        try:
+            input_payload["url"] = await avalidate_public_url(input_payload["url"])
+        except SSRFValidationError as exc:
+            await _fail_execution(
+                execution_id,
+                {"code": "ssrf_rejected", "message": str(exc)},
+                session_factory,
+                lease_owner,
+            )
+            return
 
     heartbeat_stop = asyncio.Event()
     lease_lost = asyncio.Event()
@@ -327,7 +329,7 @@ async def run_acquisition_execution(
     lease_lost_task = None
     try:
         pool = await _managed_browser_pool(session_factory)
-        endpoint = pool.select_anonymous_endpoint()
+        endpoint = pool.select_endpoint(registration.required_profile_kind)
         if channel is None:
             from backend.channels.opencli_channel import OpenCLIChannel
 
@@ -336,7 +338,7 @@ async def run_acquisition_execution(
         parameters = {
             **input_payload,
             "chrome_endpoint": endpoint,
-            "required_profile_kind": "anonymous",
+            "required_profile_kind": registration.required_profile_kind,
         }
         from backend.config import get_settings
 
@@ -358,7 +360,7 @@ async def run_acquisition_execution(
                 await collection_task
             return
         result = await collection_task
-    except NoCleanProfileError as exc:
+    except (NoCleanProfileError, NoMatchingProfileError) as exc:
         await _fail_execution(
             execution_id,
             {"code": exc.code, "message": str(exc)},
@@ -499,7 +501,7 @@ async def run_acquisition_execution(
                     ),
                     "browser": {
                         "endpoint": endpoint,
-                        "profile_kind": "anonymous",
+                        "profile_kind": registration.required_profile_kind,
                     },
                     "channel_metadata": result.metadata,
                 },

backend/agent_server.py

@@ -40,7 +40,7 @@
     OPENCLI_CDP_BIN         Path to opencli 0.9 binary (default: /opt/opencli-cdp/bin/opencli)
     OPENCLI_CDP_ENDPOINT    Default Chrome CDP endpoint (default: http://localhost:19222)
     OPENCLI_DAEMON_PORT     Bridge daemon port (default: 19825)
-    OPENCLI_TIMEOUT         opencli subprocess timeout in seconds (default: 120)
+    OPENCLI_TIMEOUT         opencli subprocess timeout in seconds (default: 240)
 """
 
 import asyncio
@@ -110,7 +110,7 @@ def _resolve_bin(mode: str) -> str:  # noqa: ARG001
 #   off  — disable auto-registration entirely
 _AGENT_REGISTER = os.environ.get("AGENT_REGISTER", "http").lower()
 # opencli subprocess execution timeout in seconds
-_OPENCLI_TIMEOUT = int(os.environ.get("OPENCLI_TIMEOUT", "120"))
+_OPENCLI_TIMEOUT = int(os.environ.get("OPENCLI_TIMEOUT", "240"))
 # Outbound proxy for agent → center communication (optional)
 _HTTP_PROXY = os.environ.get("HTTP_PROXY") or os.environ.get("http_proxy") or ""
 _HTTPS_PROXY = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy") or ""
@@ -165,7 +165,7 @@ async def _kill_process_tree(proc: asyncio.subprocess.Process) -> None:
     await proc.wait()
 
 
-async def _runtime_lineage(bin_path: str) -> dict[str, str]:
+async def _runtime_lineage(bin_path: str, site: str, command: str) -> dict[str, str]:
     """Measure the binaries/source used by this node; never echo declarations."""
     async def output(*argv: str, cwd: str | None = None) -> str:
         try:
@@ -178,8 +178,9 @@ async def output(*argv: str, cwd: str | None = None) -> str:
             return ""
 
     repo_commit = await output("git", "rev-parse", "HEAD", cwd=_OHMYOPENCLI_ROOT)
+    adapter_path = f"adapters/{site}/{command}.js"
     source_commit = await output(
-        "git", "log", "-1", "--format=%H", "--", "adapters/official-site/observe.js",
+        "git", "log", "-1", "--format=%H", "--", adapter_path,
         cwd=_OHMYOPENCLI_ROOT,
     )
     version_text = await output(bin_path, "--version")
@@ -662,7 +663,7 @@ async def collect(req: CollectRequest) -> dict:
             hostname = "host.docker.internal"
         env.pop("OPENCLI_CDP_ENDPOINT", None)
         env["OPENCLI_DAEMON_HOST"] = hostname
-        env["OPENCLI_DAEMON_PORT"] = str(_DAEMON_PORT)
+        env.pop("OPENCLI_DAEMON_PORT", None)
         logger.info("bridge | cmd=%s daemon=%s:%s", " ".join(cmd), hostname, _DAEMON_PORT)
     else:
         # Same logic for CDP: remap localhost to host.docker.internal only when
@@ -695,7 +696,11 @@ async def collect(req: CollectRequest) -> dict:
             await _kill_process_tree(proc)
         if mode == "cdp":
             await _cleanup_cdp_tabs(cdp_ep, pre_tab_ids)
-        return {"success": False, "items": [], "error": "opencli timed out after 120s"}
+        return {
+            "success": False,
+            "items": [],
+            "error": f"opencli timed out after {_OPENCLI_TIMEOUT}s",
+        }
     except Exception as exc:
         logger.exception("subprocess error | %s", exc)
         if mode == "cdp":
@@ -725,7 +730,9 @@ async def collect(req: CollectRequest) -> dict:
 
     logger.info("done | site=%s cmd=%s items=%d", req.site, req.command, len(items))
     trace_match = re.search(r"OpenCLI trace artifact:\s*([^\r\n]+)", stderr_str)
-    metadata: dict[str, Any] = {"runtime": await _runtime_lineage(bin_path)}
+    metadata: dict[str, Any] = {
+        "runtime": await _runtime_lineage(bin_path, req.site, req.command)
+    }
     if trace_match:
         metadata["trace_artifact"] = trace_match.group(1)
     return {

backend/browser_pool.py

@@ -35,6 +35,14 @@ def __init__(self) -> None:
         super().__init__(self.code)
 
 
+class NoMatchingProfileError(RuntimeError):
+    """No browser profile of the capability's required kind is available."""
+
+    def __init__(self, profile_kind: str) -> None:
+        self.code = f"no_{profile_kind}_profile"
+        super().__init__(self.code)
+
+
 class LocalBrowserPool:
     """In-process pool backed by per-endpoint asyncio.Queue slots.
 
@@ -62,10 +70,11 @@ def __init__(self, endpoints: list[str]) -> None:
         self._agent_protocols: dict[str, str | None] = {ep: None for ep in endpoints}
         # node_type per endpoint: "docker" (started in container) | "shell" (native process)
         self._node_types: dict[str, str] = {ep: "docker" for ep in endpoints}
-        # Fail closed: an endpoint is potentially personalized until an agent
-        # or operator explicitly registers it as a dedicated anonymous profile.
+        # Managed acquisition routes only to explicitly classified profiles.
+        # Merely appearing in a legacy endpoint setting does not certify an
+        # account-backed session or a dedicated anonymous browser.
         self._profile_kinds: dict[str, str] = {
-            ep: "authenticated" for ep in endpoints
+            ep: "unclassified" for ep in endpoints
         }
         logger.info(
             "BrowserPool (local): %d Chrome instance(s): %s",
@@ -143,13 +152,28 @@ def anonymous_endpoints(self) -> list[str]:
             raise NoCleanProfileError()
         return candidates
 
-    def select_anonymous_endpoint(self) -> str:
-        candidates = self.anonymous_endpoints()
+    def profile_endpoints(self, profile_kind: str) -> list[str]:
+        candidates = [
+            endpoint
+            for endpoint in self.endpoints
+            if self.get_profile_kind(endpoint) == profile_kind
+        ]
+        if not candidates:
+            if profile_kind == "anonymous":
+                raise NoCleanProfileError()
+            raise NoMatchingProfileError(profile_kind)
+        return candidates
+
+    def select_endpoint(self, profile_kind: str) -> str:
+        candidates = self.profile_endpoints(profile_kind)
         return next(
             (endpoint for endpoint in candidates if self.available_for(endpoint)),
             candidates[0],
         )
 
+    def select_anonymous_endpoint(self) -> str:
+        return self.select_endpoint("anonymous")
+
     async def _acquire_any(self) -> str:
         """Wait for whichever endpoint slot becomes free first."""
         tasks: dict[asyncio.Task[str], str] = {
@@ -226,7 +250,7 @@ def set_node_type(self, endpoint: str, node_type: str) -> None:
         logger.info("BrowserPool: endpoint %s node_type set to %s", endpoint, node_type)
 
     def get_profile_kind(self, endpoint: str) -> str:
-        return self._profile_kinds.get(endpoint, "authenticated")
+        return self._profile_kinds.get(endpoint, "unclassified")
 
     def set_profile_kind(self, endpoint: str, profile_kind: str) -> None:
         if profile_kind not in {"anonymous", "authenticated"}:
@@ -249,7 +273,7 @@ def add_endpoint(self, endpoint: str) -> None:
         self._agent_urls.setdefault(endpoint, None)
         self._agent_protocols.setdefault(endpoint, None)
         self._node_types.setdefault(endpoint, "docker")
-        self._profile_kinds.setdefault(endpoint, "authenticated")
+        self._profile_kinds.setdefault(endpoint, "unclassified")
         self._total += 1
         logger.info("BrowserPool: added endpoint %s (total: %d)", endpoint, self._total)
 
@@ -311,7 +335,7 @@ def __init__(self, endpoints: list[str], redis_url: str) -> None:
         self._total = len(endpoints)
         self._modes: dict[str, str] = {ep: "bridge" for ep in endpoints}
         self._profile_kinds: dict[str, str] = {
-            ep: "authenticated" for ep in endpoints
+            ep: "unclassified" for ep in endpoints
         }
 
     def _client(self):
@@ -349,7 +373,7 @@ async def register_endpoint(self, endpoint: str) -> None:
             self._endpoints.append(endpoint)
             self._total += 1
             self._modes.setdefault(endpoint, "bridge")
-            self._profile_kinds.setdefault(endpoint, "authenticated")
+            self._profile_kinds.setdefault(endpoint, "unclassified")
 
         async with self._client() as r:
             added = await r.sadd(self._REGISTRY_KEY, endpoint)
@@ -460,7 +484,7 @@ def set_mode(self, endpoint: str, mode: str) -> None:
         self._modes[endpoint] = mode
 
     def get_profile_kind(self, endpoint: str) -> str:
-        return self._profile_kinds.get(endpoint, "authenticated")
+        return self._profile_kinds.get(endpoint, "unclassified")
 
     def set_profile_kind(self, endpoint: str, profile_kind: str) -> None:
         if profile_kind not in {"anonymous", "authenticated"}:
@@ -477,8 +501,23 @@ def anonymous_endpoints(self) -> list[str]:
             raise NoCleanProfileError()
         return candidates
 
+    def profile_endpoints(self, profile_kind: str) -> list[str]:
+        candidates = [
+            endpoint
+            for endpoint in self.endpoints
+            if self.get_profile_kind(endpoint) == profile_kind
+        ]
+        if not candidates:
+            if profile_kind == "anonymous":
+                raise NoCleanProfileError()
+            raise NoMatchingProfileError(profile_kind)
+        return candidates
+
+    def select_endpoint(self, profile_kind: str) -> str:
+        return self.profile_endpoints(profile_kind)[0]
+
     def select_anonymous_endpoint(self) -> str:
-        return self.anonymous_endpoints()[0]
+        return self.select_endpoint("anonymous")
 
 
 # ── Module-level singleton ────────────────────────────────────────────────────

backend/channels/opencli_channel.py

@@ -558,9 +558,13 @@ async def _collect_with_opencli_subprocess(
     try:
         returncode, stdout_text, stderr_text = await _run_opencli(cmd, env)
     except TimeoutError as exc:
+        from backend.config import get_settings
+
+        timeout = get_settings().opencli_timeout
         logger.error("opencli timeout | cmd=%s", " ".join(cmd))
         return ChannelResult.fail(
-            "opencli command timed out after 120s", error_type=type(exc).__name__
+            f"opencli command timed out after {timeout}s",
+            error_type=type(exc).__name__,
         )
     except FileNotFoundError as exc:
         logger.error("opencli binary not found: %s", cmd[0])
@@ -768,7 +772,9 @@ async def collect(
                 daemon_host = urlparse(cdp_endpoint).hostname or "agent-1"
                 env.pop("OPENCLI_CDP_ENDPOINT", None)
                 env["OPENCLI_DAEMON_HOST"] = daemon_host
-                env["OPENCLI_DAEMON_PORT"] = str(_DAEMON_PORT)
+                # OpenCLI 1.8.5 fixes Browser Bridge to localhost:19825 and
+                # rejects the legacy port override even when it is 19825.
+                env.pop("OPENCLI_DAEMON_PORT", None)
                 logger.info(
                     "opencli bridge | cmd=%s daemon=%s:%s",
                     " ".join(cmd),