From 5c3c5407f2809815503961479b629d7a21bb20e9 Mon Sep 17 00:00:00 2001 From: lunnt <2276214182@qq.com> Date: Wed, 29 Jul 2026 12:57:46 +0800 Subject: [PATCH 1/7] feat(acquisition): add managed Doubao capture route --- Dockerfile | 4 +- agent/Dockerfile | 4 +- backend/acquisition/capabilities.py | 124 ++++++- backend/acquisition/registry.py | 48 ++- backend/acquisition/runner.py | 252 ++++++++++--- backend/agent_server.py | 42 ++- backend/api/v1/geo_acquisition.py | 14 +- backend/channels/opencli_channel.py | 111 +++++- backend/schemas/acquisition.py | 1 + scripts/install-agent.sh | 5 +- scripts/install-managed-opencli.ps1 | 15 +- scripts/verify_managed_opencli_runtime.py | 76 ++-- tests/unit/channels/test_opencli_channel.py | 59 +++ tests/unit/test_acquisition_capabilities.py | 268 +++++++++----- tests/unit/test_acquisition_runner.py | 340 +++++++++++++++++- .../test_agent_image_runtime_packaging.py | 6 +- tests/unit/test_agent_server.py | 35 ++ tests/unit/test_geo_acquisition_api.py | 57 ++- tests/unit/test_managed_opencli_verifier.py | 45 ++- 19 files changed, 1289 insertions(+), 217 deletions(-) diff --git a/Dockerfile b/Dockerfile index ff0045b..687a17e 100644 --- a/Dockerfile +++ b/Dockerfile @@ -45,12 +45,14 @@ RUN npm install -g @jackwener/opencli@${OPENCLI_VERSION} \ && rm -rf /root/.npm ARG OHMYOPENCLI_REPO=https://github.com/2233admin/OhMyOpenCLI.git -ARG OHMYOPENCLI_COMMIT=73cc60c83586ef2c95469b3b70d6cfc80fa5bc53 +ARG OHMYOPENCLI_COMMIT=bfe1c25b4b12661058dd6e9980c562a09f230cc7 ARG OFFICIAL_SITE_CAPABILITY_COMMIT=73cc60c83586ef2c95469b3b70d6cfc80fa5bc53 +ARG DOUBAO_CAPABILITY_COMMIT=bfe1c25b4b12661058dd6e9980c562a09f230cc7 RUN git clone ${OHMYOPENCLI_REPO} /opt/ohmyopencli \ && cd /opt/ohmyopencli \ && git checkout --detach ${OHMYOPENCLI_COMMIT} \ && git merge-base --is-ancestor ${OFFICIAL_SITE_CAPABILITY_COMMIT} HEAD \ + && git merge-base --is-ancestor ${DOUBAO_CAPABILITY_COMMIT} HEAD \ && npm ci \ && test "$(git rev-parse HEAD)" = "${OHMYOPENCLI_COMMIT}" diff --git a/agent/Dockerfile b/agent/Dockerfile index 6a9ca27..1781584 100644 --- a/agent/Dockerfile +++ b/agent/Dockerfile @@ -36,12 +36,14 @@ RUN npm install -g @jackwener/opencli@${OPENCLI_VERSION} \ # identities separate: the latter is the behavior change, while the former is # the exact checkout certified by this image. ARG OHMYOPENCLI_REPO=https://github.com/2233admin/OhMyOpenCLI.git -ARG OHMYOPENCLI_COMMIT=73cc60c83586ef2c95469b3b70d6cfc80fa5bc53 +ARG OHMYOPENCLI_COMMIT=bfe1c25b4b12661058dd6e9980c562a09f230cc7 ARG OFFICIAL_SITE_CAPABILITY_COMMIT=73cc60c83586ef2c95469b3b70d6cfc80fa5bc53 +ARG DOUBAO_CAPABILITY_COMMIT=bfe1c25b4b12661058dd6e9980c562a09f230cc7 RUN git clone ${OHMYOPENCLI_REPO} /opt/ohmyopencli \ && cd /opt/ohmyopencli \ && git checkout --detach ${OHMYOPENCLI_COMMIT} \ && git merge-base --is-ancestor ${OFFICIAL_SITE_CAPABILITY_COMMIT} HEAD \ + && git merge-base --is-ancestor ${DOUBAO_CAPABILITY_COMMIT} HEAD \ && npm ci \ && test "$(git rev-parse HEAD)" = "${OHMYOPENCLI_COMMIT}" diff --git a/backend/acquisition/capabilities.py b/backend/acquisition/capabilities.py index 45d4114..46b904c 100644 --- a/backend/acquisition/capabilities.py +++ b/backend/acquisition/capabilities.py @@ -1,8 +1,11 @@ """Runtime-probed capability catalog for managed GEO acquisition.""" import asyncio +import json import os import re +from typing import Any +from urllib.parse import urlparse from backend.acquisition.registry import ( OHMYOPENCLI_COMMIT, @@ -101,15 +104,105 @@ async def _registration_is_available( return patch_rc != 0 and registration.route_probe_error in patch_output -def _anonymous_profile_available() -> bool: +def _profile_unavailable_reason(profile_kind: str) -> str: + return ( + "no_clean_profile" + if profile_kind == "anonymous" + else f"no_{profile_kind}_profile" + ) + + +def _profile_endpoints(profile_kind: str) -> tuple[object | None, list[str]]: 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 + return None, [] + return pool, [ + endpoint + for endpoint in pool.endpoints + if pool.get_profile_kind(endpoint) == profile_kind + ] + + +def _browser_environment(pool: Any, endpoint: str) -> dict[str, str]: + env = os.environ.copy() + if pool.get_mode(endpoint) == "bridge": + env.pop("OPENCLI_CDP_ENDPOINT", None) + env["OPENCLI_DAEMON_HOST"] = urlparse(endpoint).hostname or "agent-1" + env["OPENCLI_DAEMON_PORT"] = "19825" + else: + env.pop("OPENCLI_DAEMON_HOST", None) + env.pop("OPENCLI_DAEMON_PORT", None) + env["OPENCLI_CDP_ENDPOINT"] = endpoint + return env + + +def _json_payload(output: str) -> dict | None: + start = next((index for index, char in enumerate(output) if char in "[{"), None) + if start is None: + return None + try: + parsed, _ = json.JSONDecoder().raw_decode(output[start:]) + except json.JSONDecodeError: + return None + if isinstance(parsed, list): + parsed = parsed[0] if parsed else None + return parsed if isinstance(parsed, dict) else None + + +async def _session_is_ready( + registration: CapabilityRegistration, + pool: Any, + endpoint: str, +) -> bool: + if not registration.session_probe_args: + return True + from backend.config import get_settings + + if get_settings().collection_mode == "agent": + from backend.channels.opencli_channel import ( + _collect_via_agent, + _collect_via_ws_agent, + ) + + site, command = registration.session_probe_args[:2] + mode = pool.get_mode(endpoint) + get_protocol = getattr(pool, "get_agent_protocol", None) + get_agent_url = getattr(pool, "get_agent_url", None) + protocol = get_protocol(endpoint) if get_protocol else "http" + agent_url = (get_agent_url(endpoint) if get_agent_url else None) or endpoint + if protocol == "ws": + result = await _collect_via_ws_agent( + agent_url, site, command, {}, [], "json", mode, None + ) + else: + result = await _collect_via_agent( + agent_url, site, command, {}, [], "json", mode, None + ) + payload = result.items[0] if result.success and result.items else None + rc = 0 if payload is not None else 1 + else: + opencli_bin = resolve_opencli_bin() + rc, output = await _command( + opencli_bin, + *registration.session_probe_args, + env=_browser_environment(pool, endpoint), + ) + payload = _json_payload(output) + return bool( + rc == 0 + and payload + and payload.get("unattendedReady") is True + and payload.get("loginDetected") is False + and payload.get("promptInputDetected") is True + and payload.get("sendButtonDetected") is True + and ( + registration.session_expected_host is None + or urlparse(str(payload.get("url", ""))).hostname + == registration.session_expected_host + ) ) @@ -118,19 +211,38 @@ 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 + pool, endpoints = _profile_endpoints(registration.required_profile_kind) + ready = bool(endpoints) + unavailable_reason = ( + None + if ready + else _profile_unavailable_reason(registration.required_profile_kind) + ) + if ready and registration.session_probe_args: + ready = any( + [ + await _session_is_ready(registration, pool, endpoint) + for endpoint in endpoints + ] + ) + if not ready: + unavailable_reason = ( + registration.session_unavailable_reason + or "browser_session_not_ready" + ) descriptors.append( CapabilityDescriptor( capability_id=registration.capability_id, capability_version=registration.capability_version, output_schema_version=registration.output_schema_version, + target=registration.target, ready=ready, runtime=registration.runtime_identity(), - unavailable_reason=None if ready else "no_clean_profile", + unavailable_reason=None if ready else unavailable_reason, ) ) return descriptors diff --git a/backend/acquisition/registry.py b/backend/acquisition/registry.py index ecaa5ba..34a0fe3 100644 --- a/backend/acquisition/registry.py +++ b/backend/acquisition/registry.py @@ -2,8 +2,9 @@ from dataclasses import dataclass -OHMYOPENCLI_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" +OHMYOPENCLI_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" OFFICIAL_SITE_CAPABILITY_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" +DOUBAO_CAPABILITY_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" OPENCLI_VERSION = "1.8.5" @@ -19,13 +20,19 @@ class CapabilityRegistration: route_probe_args: tuple[str, ...] route_probe_error: str required_profile_kind: str = "anonymous" + url_input_field: str | None = "url" + target: str | None = None + session_probe_args: tuple[str, ...] = () + session_unavailable_reason: str | None = None + session_expected_host: str | None = None @property - def identity(self) -> tuple[str, str, str]: + def identity(self) -> tuple[str, str, str, str | None]: return ( self.capability_id, self.capability_version, self.output_schema_version, + self.target, ) def runtime_identity(self) -> dict[str, str]: @@ -59,6 +66,40 @@ 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=DOUBAO_CAPABILITY_COMMIT, + invocation={ + "site": "doubao", + "command": "capture", + "format": "json", + }, + probe_args=("doubao", "capture", "--help"), + help_marker="doubao capture", + route_probe_args=( + "doubao", + "capture", + "runtime-route-probe", + "-f", + "json", + ), + route_probe_error="CDP not reachable at http://127.0.0.1:9", + required_profile_kind="authenticated", + url_input_field=None, + target="doubao", + session_probe_args=( + "doubao", + "session-probe", + "--strict", + "true", + "-f", + "json", + ), + session_unavailable_reason="doubao_session_not_ready", + session_expected_host="www.doubao.com", + ), ) @@ -71,8 +112,9 @@ def get_capability_registration( capability_id: str, capability_version: str, output_schema_version: str, + target: str | None = None, ) -> CapabilityRegistration | None: - identity = (capability_id, capability_version, output_schema_version) + identity = (capability_id, capability_version, output_schema_version, target) return next( (registration for registration in _REGISTRATIONS if registration.identity == identity), None, diff --git a/backend/acquisition/runner.py b/backend/acquisition/runner.py index 96a65db..e1670e1 100644 --- a/backend/acquisition/runner.py +++ b/backend/acquisition/runner.py @@ -23,6 +23,65 @@ _HEARTBEAT_INTERVAL_SECONDS = 5 logger = logging.getLogger(__name__) + +def _capability_failure_code(message: str, source_code: str | None = None) -> str: + typed_codes = { + "DOUBAO_CAPTURE_LOGIN_WALL": "login_required", + "DOUBAO_CAPTURE_CAPTCHA": "captcha_required", + "DOUBAO_CAPTURE_TIMEOUT": "capture_timeout", + "DOUBAO_CAPTURE_REFUSAL": "model_refusal", + "DOUBAO_CAPTURE_EMPTY_ANSWER": "empty_answer", + "DOUBAO_CAPTURE_PAGE_DRIFT": "page_contract_drift", + } + if source_code and source_code.upper() in typed_codes: + return typed_codes[source_code.upper()] + normalized = message.lower() + if "cdp not reachable" in normalized: + return "browser_route_unavailable" + if any( + marker in normalized + for marker in ( + "login-required", + "doubao_capture_login_wall", + "logged-in browser session", + ) + ): + return "login_required" + if any( + marker in normalized + for marker in ( + "captcha", + "doubao_capture_captcha", + "verification challenge", + ) + ): + return "captcha_required" + if "timed out" in normalized or "doubao_capture_timeout" in normalized: + return "capture_timeout" + if any( + marker in normalized + for marker in ( + "refusal", + "refused", + "doubao_capture_refusal", + ) + ): + return "model_refusal" + if "empty answer" in normalized or "doubao_capture_empty_answer" in normalized: + return "empty_answer" + if any( + marker in normalized + for marker in ( + "composer-unavailable", + "could not submit", + "page drift", + "doubao_capture_page_drift", + ) + ): + return "page_contract_drift" + return "capability_execution_failed" + + async def _managed_browser_pool( session_factory: async_sessionmaker[AsyncSession], ): @@ -282,7 +341,10 @@ async def run_acquisition_execution( required_artifacts = list(execution.required_artifacts) registration = get_capability_registration( - capability_id, capability_version, output_schema_version + capability_id, + capability_version, + output_schema_version, + input_payload.get("target"), ) if registration is None: await _fail_execution( @@ -301,16 +363,33 @@ 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 registration.url_input_field is not None: + raw_url = input_payload.get(registration.url_input_field) + if not isinstance(raw_url, str) or not raw_url.strip(): + await _fail_execution( + execution_id, + { + "code": "invalid_capability_input", + "message": ( + f"{registration.url_input_field} must be a non-empty URL string" + ), + }, + session_factory, + lease_owner, + ) + return + try: + input_payload[registration.url_input_field] = await avalidate_public_url( + raw_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,41 +406,96 @@ async def run_acquisition_execution( lease_lost_task = None try: pool = await _managed_browser_pool(session_factory) - endpoint = pool.select_anonymous_endpoint() - if channel is None: - from backend.channels.opencli_channel import OpenCLIChannel + candidates = [ + candidate + for candidate in pool.endpoints + if pool.get_profile_kind(candidate) == registration.required_profile_kind + ] + if not candidates: + code = ( + "no_clean_profile" + if registration.required_profile_kind == "anonymous" + else f"no_{registration.required_profile_kind}_profile" + ) + await _fail_execution( + execution_id, + {"code": code, "message": code}, + session_factory, + lease_owner, + ) + return + endpoint = next( + ( + candidate + for candidate in candidates + if pool.available_for(candidate) + ), + candidates[0], + ) + async with pool.acquire( + endpoint=endpoint, + required_profile_kind=registration.required_profile_kind, + ) as leased_endpoint: + if registration.session_probe_args: + from backend.acquisition.capabilities import _session_is_ready + + if not await _session_is_ready(registration, pool, leased_endpoint): + await _fail_execution( + execution_id, + { + "code": "session_not_qualified", + "message": ( + f"{registration.target or capability_id} session " + "failed the execution-time readiness probe" + ), + }, + session_factory, + lease_owner, + ) + return + if channel is None: + from backend.channels.opencli_channel import OpenCLIChannel - channel = OpenCLIChannel() + channel = OpenCLIChannel() - parameters = { - **input_payload, - "chrome_endpoint": endpoint, - "required_profile_kind": "anonymous", - } - from backend.config import get_settings + capability_input = { + key: value for key, value in input_payload.items() if key != "target" + } + parameters = { + **capability_input, + "chrome_endpoint": leased_endpoint, + "required_profile_kind": registration.required_profile_kind, + "_endpoint_preacquired": True, + } + from backend.config import get_settings - if get_settings().collection_mode == "agent": - parameters["execution_id"] = execution_id - if required_artifacts: - parameters["trace"] = "on" - collection_task = asyncio.create_task( - channel.collect(registration.invocation, parameters) - ) - lease_lost_task = asyncio.create_task(lease_lost.wait()) - done, _ = await asyncio.wait( - {collection_task, lease_lost_task}, - return_when=asyncio.FIRST_COMPLETED, + if get_settings().collection_mode == "agent": + parameters["execution_id"] = execution_id + if required_artifacts: + parameters["trace"] = "on" + collection_task = asyncio.create_task( + channel.collect(registration.invocation, parameters) + ) + lease_lost_task = asyncio.create_task(lease_lost.wait()) + done, _ = await asyncio.wait( + {collection_task, lease_lost_task}, + return_when=asyncio.FIRST_COMPLETED, + ) + if lease_lost_task in done: + collection_task.cancel() + with suppress(asyncio.CancelledError): + await collection_task + return + result = await collection_task + except NoCleanProfileError: + code = ( + "no_clean_profile" + if registration.required_profile_kind == "anonymous" + else f"no_{registration.required_profile_kind}_profile" ) - if lease_lost_task in done: - collection_task.cancel() - with suppress(asyncio.CancelledError): - await collection_task - return - result = await collection_task - except NoCleanProfileError as exc: await _fail_execution( execution_id, - {"code": exc.code, "message": str(exc)}, + {"code": code, "message": code}, session_factory, lease_owner, ) @@ -425,13 +559,11 @@ async def run_acquisition_execution( execution.status = AcquisitionExecutionStatus.FAILED message = result.error or "Capability returned no payload" execution.failure = { - "code": ( - "browser_route_unavailable" - if "CDP not reachable" in message - else "capability_execution_failed" - ), + "code": _capability_failure_code(message, result.error_type), "message": message, } + if result.error_type: + execution.failure["source_code"] = result.error_type else: payload = result.items[0] from backend.config import get_settings @@ -453,6 +585,14 @@ async def run_acquisition_execution( payload.get("capabilityId") == capability_id and payload.get("capabilityVersion") == capability_version and payload.get("outputSchemaVersion") == output_schema_version + and ( + registration.target is None + or payload.get("target") == registration.target + ) + and ( + "prompt" not in input_payload + or payload.get("prompt") == input_payload["prompt"] + ) ) if not identity_matches: execution.status = AcquisitionExecutionStatus.FAILED @@ -465,9 +605,15 @@ async def run_acquisition_execution( await db.commit() return - returned_artifact_kinds = { - "trace" for value in [result.metadata.get("trace_artifact")] if value - } + trace_artifact = result.metadata.get("trace_artifact") + trace_sha256 = result.metadata.get("trace_sha256") + valid_trace = bool( + trace_artifact + and isinstance(trace_sha256, str) + and len(trace_sha256) == 64 + and all(char in "0123456789abcdef" for char in trace_sha256.lower()) + ) + returned_artifact_kinds = {"trace"} if valid_trace else set() missing_artifacts = [ kind for kind in required_artifacts if kind not in returned_artifact_kinds ] @@ -499,17 +645,21 @@ async def run_acquisition_execution( ), "browser": { "endpoint": endpoint, - "profile_kind": "anonymous", + "profile_kind": registration.required_profile_kind, }, "channel_metadata": result.metadata, }, } artifacts = payload.get("artifacts", []) artifact_refs = artifacts if isinstance(artifacts, list) else [] - if result.metadata.get("trace_artifact"): + if trace_artifact and trace_sha256: artifact_refs = [ *artifact_refs, - {"kind": "trace", "ref": result.metadata["trace_artifact"]}, + { + "kind": "trace", + "ref": trace_artifact, + "sha256": trace_sha256, + }, ] execution.artifact_refs = artifact_refs execution.failure = None diff --git a/backend/agent_server.py b/backend/agent_server.py index ccb2fd3..ae0c443 100644 --- a/backend/agent_server.py +++ b/backend/agent_server.py @@ -45,6 +45,7 @@ import asyncio import csv +import hashlib import io import json import logging @@ -55,6 +56,7 @@ import socket import subprocess from contextlib import asynccontextmanager +from pathlib import Path from typing import Any from urllib.parse import urlparse @@ -87,6 +89,27 @@ def _resolve_bin(mode: str) -> str: # noqa: ARG001 if resolved: return resolved return shutil.which(configured) or configured + + +def _artifact_sha256(artifact_ref: str) -> str | None: + root = Path(artifact_ref) + if not root.exists(): + return None + files = [root] if root.is_file() else sorted( + (path for path in root.rglob("*") if path.is_file()), + key=lambda path: path.as_posix(), + ) + digest = hashlib.sha256() + base = root.parent if root.is_file() else root + for path in files: + digest.update(path.relative_to(base).as_posix().encode()) + digest.update(b"\0") + with path.open("rb") as handle: + while chunk := handle.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + _DEFAULT_CDP = os.environ.get("OPENCLI_CDP_ENDPOINT", "http://localhost:19222") _BROWSER_PROFILE_KIND = os.environ.get( "OPENCLI_BROWSER_PROFILE_KIND", "authenticated" @@ -165,7 +188,11 @@ 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 +205,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") @@ -699,9 +727,15 @@ 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) + trace_artifact = trace_match.group(1).strip() + metadata["trace_artifact"] = trace_artifact + trace_sha256 = await asyncio.to_thread(_artifact_sha256, trace_artifact) + if trace_sha256: + metadata["trace_sha256"] = trace_sha256 return { "success": True, "items": items, diff --git a/backend/api/v1/geo_acquisition.py b/backend/api/v1/geo_acquisition.py index c8a2cd9..ab4e44b 100644 --- a/backend/api/v1/geo_acquisition.py +++ b/backend/api/v1/geo_acquisition.py @@ -51,7 +51,19 @@ async def _validate_capability(body: AcquisitionSubmission) -> CapabilityDescrip "message": body.output_schema_version, }, ) - capability = matching_schema[0] + requested_target = body.input.get("target") + matching_target = [ + c for c in matching_schema if c.target == requested_target + ] + if not matching_target: + raise HTTPException( + status_code=422, + detail={ + "code": "target_not_registered", + "message": str(requested_target or ""), + }, + ) + capability = matching_target[0] if not capability.ready: raise HTTPException( status_code=409, diff --git a/backend/channels/opencli_channel.py b/backend/channels/opencli_channel.py index c86c9f8..a4f293c 100644 --- a/backend/channels/opencli_channel.py +++ b/backend/channels/opencli_channel.py @@ -2,6 +2,7 @@ import asyncio import csv +import hashlib import io import json import logging @@ -9,6 +10,8 @@ import re import subprocess import time +from contextlib import asynccontextmanager +from pathlib import Path from typing import Any from urllib.parse import urlparse @@ -73,11 +76,39 @@ def _split_routing_parameters( cli_parameters = { key: value for key, value in parameters.items() - if key not in {"chrome_endpoint", "required_profile_kind", "execution_id"} + if key + not in { + "chrome_endpoint", + "required_profile_kind", + "execution_id", + "_endpoint_preacquired", + } } return (chrome_endpoint, required_profile_kind), cli_parameters +@asynccontextmanager +async def _browser_endpoint_lease( + pool: Any, + endpoint: str | None, + required_profile_kind: str | None, + *, + preacquired: bool, +): + """Reuse a runner-owned endpoint lease or acquire one for legacy callers.""" + if preacquired: + if not endpoint: + raise ValueError("A pre-acquired browser lease requires chrome_endpoint") + yield endpoint + return + + acquire_kwargs: dict[str, Any] = {"endpoint": endpoint} + if required_profile_kind: + acquire_kwargs["required_profile_kind"] = required_profile_kind + async with pool.acquire(**acquire_kwargs) as leased_endpoint: + yield leased_endpoint + + async def _site_bound_agent_endpoint(pool: Any, site: str, session: AsyncSession) -> str | None: if not site: return None @@ -539,12 +570,59 @@ async def _run_opencli(cmd: list[str], env: dict) -> tuple[int, str, str]: from backend.config import get_settings timeout = get_settings().opencli_timeout stdout, stderr = await asyncio.wait_for(proc.communicate(), timeout=timeout) - return proc.returncode, stdout.decode(), stderr.decode().strip() + return int(proc.returncode or 0), stdout.decode(), stderr.decode().strip() except (TimeoutError, asyncio.CancelledError): await _kill_subprocess(proc) raise +def _extract_opencli_error(stderr_text: str) -> tuple[str | None, str | None]: + """Read OpenCLI's structured error envelope without depending on its prose.""" + try: + envelope = yaml.safe_load(stderr_text) + except yaml.YAMLError: + envelope = None + if isinstance(envelope, dict) and isinstance(envelope.get("error"), dict): + error = envelope["error"] + code = str(error.get("code") or "").strip() or None + message = str(error.get("message") or "").strip() or None + return code, message + + # OpenCLI may append a human update notice after the YAML envelope. Retain + # the typed code even when that makes the complete stderr invalid YAML. + code_match = re.search( + r"(?m)^\s+code:\s*['\"]?([A-Za-z0-9_-]+)['\"]?\s*$", + stderr_text, + ) + message_match = re.search( + r"(?m)^\s+message:\s*(.+?)\s*$", + stderr_text, + ) + code = code_match.group(1) if code_match else None + message = message_match.group(1).strip(" '\"") if message_match else None + return code, message + + +def _artifact_sha256(artifact_ref: str) -> str | None: + """Hash a trace file/directory so its persisted reference is auditable.""" + root = Path(artifact_ref) + if not root.exists(): + return None + files = [root] if root.is_file() else sorted( + (path for path in root.rglob("*") if path.is_file()), + key=lambda path: path.as_posix(), + ) + digest = hashlib.sha256() + base = root.parent if root.is_file() else root + for path in files: + digest.update(path.relative_to(base).as_posix().encode()) + digest.update(b"\0") + with path.open("rb") as handle: + while chunk := handle.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() + + async def _collect_with_opencli_subprocess( cmd: list[str], env: dict, @@ -578,7 +656,12 @@ async def _collect_with_opencli_subprocess( if returncode != 0: logger.error("opencli exit=%d | stderr=%s", returncode, stderr_text[:500]) - return ChannelResult.fail(f"opencli exited with code {returncode}: {stderr_text}") + source_code, source_message = _extract_opencli_error(stderr_text) + message = source_message or stderr_text or "OpenCLI command failed" + return ChannelResult.fail( + f"opencli exited with code {returncode}: {message}", + error_type=source_code, + ) raw = stdout_text logger.debug("opencli stdout | %d chars | preview=%s", len(raw), raw[:200]) @@ -609,7 +692,11 @@ async def _collect_with_opencli_subprocess( metadata["chrome_mode"] = chrome_mode trace_match = re.search(r"OpenCLI trace artifact:\s*([^\r\n]+)", stderr_text) if trace_match: - metadata["trace_artifact"] = trace_match.group(1).strip() + trace_artifact = trace_match.group(1).strip() + metadata["trace_artifact"] = trace_artifact + trace_sha256 = await asyncio.to_thread(_artifact_sha256, trace_artifact) + if trace_sha256: + metadata["trace_sha256"] = trace_sha256 return ChannelResult.ok(items, **metadata) @@ -645,6 +732,7 @@ async def collect( output_format = config.get("format", "json") execution_id = parameters.get("execution_id") or None + endpoint_preacquired = parameters.get("_endpoint_preacquired") is True (chrome_endpoint, required_profile_kind), cli_params = ( _split_routing_parameters(parameters) ) @@ -720,10 +808,12 @@ async def collect( "No registered agent nodes available. Please add an agent node first." ) - acquire_kwargs: dict[str, Any] = {"endpoint": _acquire_endpoint} - if required_profile_kind: - acquire_kwargs["required_profile_kind"] = required_profile_kind - async with pool.acquire(**acquire_kwargs) as cdp_endpoint: + async with _browser_endpoint_lease( + pool, + _acquire_endpoint, + required_profile_kind, + preacquired=endpoint_preacquired, + ) as cdp_endpoint: mode = pool.get_mode(cdp_endpoint) # Agent mode: dispatch to remote edge node if settings.collection_mode == "agent": @@ -732,7 +822,10 @@ async def collect( if isinstance(pool, LocalBrowserPool) else "http" ) - agent_url = pool.get_agent_url(cdp_endpoint) or cdp_endpoint + get_agent_url = getattr(pool, "get_agent_url", None) + agent_url = ( + get_agent_url(cdp_endpoint) if get_agent_url else None + ) or cdp_endpoint if not protocol: return ChannelResult.fail( f"Endpoint {cdp_endpoint} has no registered agent. " diff --git a/backend/schemas/acquisition.py b/backend/schemas/acquisition.py index d87a0ad..00a76e1 100644 --- a/backend/schemas/acquisition.py +++ b/backend/schemas/acquisition.py @@ -31,6 +31,7 @@ class CapabilityDescriptor(BaseModel): capability_id: str capability_version: str output_schema_version: str + target: str | None = None ready: bool runtime: dict[str, str] = Field(default_factory=dict) unavailable_reason: str | None = None diff --git a/scripts/install-agent.sh b/scripts/install-agent.sh index 08c7091..8154644 100755 --- a/scripts/install-agent.sh +++ b/scripts/install-agent.sh @@ -333,8 +333,9 @@ install_python() { # Install the exact project-owned managed-acquisition capability package. OHMYOPENCLI_ROOT="$AGENT_DIR/ohmyopencli" - OHMYOPENCLI_COMMIT="73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" + OHMYOPENCLI_COMMIT="bfe1c25b4b12661058dd6e9980c562a09f230cc7" OFFICIAL_SITE_CAPABILITY_COMMIT="73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" + DOUBAO_CAPABILITY_COMMIT="bfe1c25b4b12661058dd6e9980c562a09f230cc7" command -v git >/dev/null 2>&1 || die "git is required to install OhMyOpenCLI" [[ -e "$OHMYOPENCLI_ROOT" ]] && die \ "Managed OhMyOpenCLI target already exists; archive it explicitly before reinstalling: $OHMYOPENCLI_ROOT" @@ -342,6 +343,8 @@ install_python() { git -C "$OHMYOPENCLI_ROOT" checkout --detach "$OHMYOPENCLI_COMMIT" git -C "$OHMYOPENCLI_ROOT" merge-base --is-ancestor \ "$OFFICIAL_SITE_CAPABILITY_COMMIT" HEAD + git -C "$OHMYOPENCLI_ROOT" merge-base --is-ancestor \ + "$DOUBAO_CAPABILITY_COMMIT" HEAD (cd "$OHMYOPENCLI_ROOT" && npm ci && npm run bootstrap) # ── Find Chrome binary ──────────────────────────────────────────────────── diff --git a/scripts/install-managed-opencli.ps1 b/scripts/install-managed-opencli.ps1 index 0616605..767238a 100644 --- a/scripts/install-managed-opencli.ps1 +++ b/scripts/install-managed-opencli.ps1 @@ -8,8 +8,11 @@ param( $ErrorActionPreference = "Stop" $OpenCliVersion = "1.8.5" -$OhMyOpenCliCommit = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" -$CapabilitySourceCommit = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" +$OhMyOpenCliCommit = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" +$CapabilitySourceCommits = @( + "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53", + "bfe1c25b4b12661058dd6e9980c562a09f230cc7" +) $requestHeaders = @{} if ($ApiAuthToken) { $requestHeaders = @{ Authorization = "Bearer $ApiAuthToken" } @@ -35,9 +38,11 @@ if (Test-Path $OhMyOpenCliRoot) { } git clone $OhMyOpenCliRepo $OhMyOpenCliRoot git -C $OhMyOpenCliRoot checkout --detach $OhMyOpenCliCommit -git -C $OhMyOpenCliRoot merge-base --is-ancestor $CapabilitySourceCommit HEAD -if ($LASTEXITCODE -ne 0) { - throw "official-site capability source commit is absent from the pinned checkout" +foreach ($CapabilitySourceCommit in $CapabilitySourceCommits) { + git -C $OhMyOpenCliRoot merge-base --is-ancestor $CapabilitySourceCommit HEAD + if ($LASTEXITCODE -ne 0) { + throw "managed capability source commit $CapabilitySourceCommit is absent from the pinned checkout" + } } Push-Location $OhMyOpenCliRoot try { diff --git a/scripts/verify_managed_opencli_runtime.py b/scripts/verify_managed_opencli_runtime.py index 53dedd5..427c8bd 100644 --- a/scripts/verify_managed_opencli_runtime.py +++ b/scripts/verify_managed_opencli_runtime.py @@ -12,8 +12,9 @@ from pathlib import Path from typing import Any -OHMYOPENCLI_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" -CAPABILITY_SOURCE_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" +OHMYOPENCLI_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" +OFFICIAL_SITE_CAPABILITY_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" +DOUBAO_CAPABILITY_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" OPENCLI_VERSION = "1.8.5" @@ -63,17 +64,21 @@ def checked(args: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: commit = checked(["git", "-C", root, "rev-parse", "HEAD"]).stdout.strip() if commit != OHMYOPENCLI_COMMIT: raise VerificationError(f"unexpected OhMyOpenCLI commit: {commit}") - checked( - [ - "git", - "-C", - root, - "merge-base", - "--is-ancestor", - CAPABILITY_SOURCE_COMMIT, - "HEAD", - ] - ) + for source_commit in ( + OFFICIAL_SITE_CAPABILITY_COMMIT, + DOUBAO_CAPABILITY_COMMIT, + ): + checked( + [ + "git", + "-C", + root, + "merge-base", + "--is-ancestor", + source_commit, + "HEAD", + ] + ) dirty = checked( ["git", "-C", root, "status", "--porcelain", "--untracked-files=no"] ).stdout.strip() @@ -84,10 +89,11 @@ def checked(args: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: if OPENCLI_VERSION not in re.findall(r"\d+\.\d+\.\d+", version_output): raise VerificationError(f"unexpected OpenCLI version: {version_output.strip()}") checked([opencli_bin, "official-site", "observe", "--help"]) + checked([opencli_bin, "doubao", "capture", "--help"]) dead_env = os.environ.copy() dead_env["OPENCLI_CDP_ENDPOINT"] = "http://127.0.0.1:9" - dead = run( + dead_route_commands = ( [ opencli_bin, "official-site", @@ -97,15 +103,32 @@ def checked(args: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: "-f", "json", ], - capture_output=True, - text=True, - timeout=30, - check=False, - env=dead_env, + [ + opencli_bin, + "doubao", + "capture", + "managed-runtime-route-probe", + "-f", + "json", + ], ) - dead_output = (dead.stdout or "") + (dead.stderr or "") - if dead.returncode == 0 or "CDP not reachable at http://127.0.0.1:9" not in dead_output: - raise VerificationError("explicit dead CDP route did not fail closed") + for dead_command in dead_route_commands: + dead = run( + dead_command, + capture_output=True, + text=True, + timeout=30, + check=False, + env=dead_env, + ) + dead_output = (dead.stdout or "") + (dead.stderr or "") + if ( + dead.returncode == 0 + or "CDP not reachable at http://127.0.0.1:9" not in dead_output + ): + raise VerificationError( + f"explicit dead CDP route did not fail closed: {' '.join(dead_command)}" + ) report: dict[str, Any] = { "platform": platform.system(), @@ -113,9 +136,16 @@ def checked(args: list[str], **kwargs: Any) -> subprocess.CompletedProcess[str]: "trace_ready": False, "runtime": { "ohmyopencli_repo_commit": commit, - "capability_source_commit": CAPABILITY_SOURCE_COMMIT, + "capability_source_commits": { + "official-site.observe": OFFICIAL_SITE_CAPABILITY_COMMIT, + "chat-ai.capture:doubao": DOUBAO_CAPABILITY_COMMIT, + }, "opencli_version": OPENCLI_VERSION, }, + "capability_contracts": { + "official-site.observe": True, + "chat-ai.capture:doubao": True, + }, } if not cdp_endpoint: return report diff --git a/tests/unit/channels/test_opencli_channel.py b/tests/unit/channels/test_opencli_channel.py index 4b6e4b9..92a77c2 100644 --- a/tests/unit/channels/test_opencli_channel.py +++ b/tests/unit/channels/test_opencli_channel.py @@ -14,7 +14,10 @@ ) from backend.channels.opencli_channel import ( OpenCLIChannel, + _artifact_sha256, + _browser_endpoint_lease, _collect_via_agent, + _extract_opencli_error, _get_named_options, _kill_subprocess, _parse_csv, @@ -31,6 +34,46 @@ def _sessionmaker(db_engine): return async_sessionmaker(db_engine, class_=AsyncSession, expire_on_commit=False) +def test_extract_opencli_error_reads_typed_yaml_envelope(): + code, message = _extract_opencli_error( + "ok: false\n" + "error:\n" + " code: DOUBAO_CAPTURE_LOGIN_WALL\n" + " message: Doubao capture requires a logged-in browser session\n" + ) + + assert code == "DOUBAO_CAPTURE_LOGIN_WALL" + assert message == "Doubao capture requires a logged-in browser session" + + +def test_extract_opencli_error_keeps_code_when_update_notice_follows_yaml(): + code, message = _extract_opencli_error( + "ok: false\n" + "error:\n" + " code: DOUBAO_CAPTURE_CAPTCHA\n" + " message: Doubao capture was blocked by a verification challenge\n\n" + "Update available: v1.8.5 -> v1.8.6\n" + ) + + assert code == "DOUBAO_CAPTURE_CAPTCHA" + assert message == "Doubao capture was blocked by a verification challenge" + + +def test_trace_artifact_hash_covers_relative_paths_and_content(tmp_path): + (tmp_path / "events.json").write_text('{"event":"answer"}', encoding="utf-8") + nested = tmp_path / "screens" + nested.mkdir() + (nested / "final.txt").write_text("doubao-final", encoding="utf-8") + + first = _artifact_sha256(str(tmp_path)) + (nested / "final.txt").write_text("changed", encoding="utf-8") + second = _artifact_sha256(str(tmp_path)) + + assert first is not None and len(first) == 64 + assert second is not None and len(second) == 64 + assert first != second + + @pytest.mark.asyncio async def test_kill_subprocess_terminates_windows_process_tree(monkeypatch): process = MagicMock(pid=4321, returncode=None) @@ -277,6 +320,7 @@ def test_managed_profile_requirement_is_not_forwarded_as_a_cli_argument(): { "chrome_endpoint": "http://clean:9222", "required_profile_kind": "anonymous", + "_endpoint_preacquired": True, "url": "https://example.com", } ) @@ -285,6 +329,21 @@ def test_managed_profile_requirement_is_not_forwarded_as_a_cli_argument(): assert cli == {"url": "https://example.com"} +@pytest.mark.asyncio +async def test_preacquired_browser_endpoint_is_reused_without_nested_pool_acquire(): + pool = MagicMock() + + async with _browser_endpoint_lease( + pool, + "http://leased:9222", + "authenticated", + preacquired=True, + ) as endpoint: + assert endpoint == "http://leased:9222" + + pool.acquire.assert_not_called() + + # ── Pure parser function tests ───────────────────────────────────────────────── def test_parse_json_list(): diff --git a/tests/unit/test_acquisition_capabilities.py b/tests/unit/test_acquisition_capabilities.py index f484991..7710fd4 100644 --- a/tests/unit/test_acquisition_capabilities.py +++ b/tests/unit/test_acquisition_capabilities.py @@ -1,10 +1,50 @@ import asyncio +import json +from types import SimpleNamespace from unittest.mock import AsyncMock import pytest -from backend.acquisition.registry import OFFICIAL_SITE_CAPABILITY_COMMIT +from backend.acquisition.registry import ( + DOUBAO_CAPABILITY_COMMIT, + OFFICIAL_SITE_CAPABILITY_COMMIT, +) from backend.browser_pool import init_pool +from backend.channels.base import ChannelResult + + +def _runtime_command(capabilities, *, dirty="", doubao_ready=True): + async def run(*args, env=None): + if args[0] == "git" and "rev-parse" in args: + return 0, f"{capabilities.OHMYOPENCLI_COMMIT}\n" + if args[0] == "git" and "merge-base" in args: + return 0, "" + if args[0] == "git" and "status" in args: + return 0, dirty + if args[-1] == "--version": + return 0, "opencli 1.8.5\n" + if args[-1] == "--help": + marker = ( + "official-site observe help" + if "official-site" in args + else "doubao capture help" + ) + return 0, marker + if env and env.get("OPENCLI_CDP_ENDPOINT") == "http://127.0.0.1:9": + return 1, "CDP not reachable at http://127.0.0.1:9" + if "session-probe" in args: + return 0, json.dumps( + { + "unattendedReady": doubao_ready, + "loginDetected": not doubao_ready, + "promptInputDetected": doubao_ready, + "sendButtonDetected": doubao_ready, + "url": "https://www.doubao.com/chat/", + } + ) + raise AssertionError(f"unexpected command: {args!r}") + + return AsyncMock(side_effect=run) @pytest.mark.asyncio @@ -51,43 +91,123 @@ async def test_catalog_does_not_publish_unpinned_runtime(monkeypatch): @pytest.mark.asyncio -async def test_catalog_reports_runtime_identity_and_clean_profile_readiness(monkeypatch): +async def test_catalog_reports_profile_and_doubao_session_readiness(monkeypatch): from backend.acquisition import capabilities - command = AsyncMock( - side_effect=[ - (0, f"{capabilities.OHMYOPENCLI_COMMIT}\n"), - (0, ""), - (0, ""), - (0, "1.8.5\n"), - (0, "official-site observe help"), - (1, "CDP not reachable at http://127.0.0.1:9"), - ] - ) + command = _runtime_command(capabilities, doubao_ready=False) monkeypatch.setattr(capabilities, "_command", command) - pool = init_pool(["http://default-profile:9222"], use_redis=False) - - [descriptor] = await capabilities.probe_capabilities() - assert descriptor.ready is False - assert descriptor.unavailable_reason == "no_clean_profile" - assert descriptor.runtime == { - "ohmyopencli_repo_commit": capabilities.OHMYOPENCLI_COMMIT, - "capability_source_commit": OFFICIAL_SITE_CAPABILITY_COMMIT, - "opencli_version": "1.8.5", + endpoint = "http://default-profile:9222" + pool = init_pool([endpoint], use_redis=False) + + descriptors = { + item.capability_id: item for item in await capabilities.probe_capabilities() + } + official = descriptors["official-site.observe"] + doubao = descriptors["chat-ai.capture"] + assert official.ready is False + assert official.unavailable_reason == "no_clean_profile" + assert official.runtime["capability_source_commit"] == OFFICIAL_SITE_CAPABILITY_COMMIT + assert doubao.ready is False + assert doubao.unavailable_reason == "doubao_session_not_ready" + assert doubao.runtime["capability_source_commit"] == DOUBAO_CAPABILITY_COMMIT + + pool.set_profile_kind(endpoint, "anonymous") + descriptors = { + item.capability_id: item for item in await capabilities.probe_capabilities() + } + assert descriptors["official-site.observe"].ready is True + assert descriptors["chat-ai.capture"].unavailable_reason == ( + "no_authenticated_profile" + ) + + +@pytest.mark.asyncio +async def test_catalog_publishes_doubao_only_after_authenticated_session_probe( + monkeypatch, +): + from backend.acquisition import capabilities + + monkeypatch.setattr( + capabilities, + "_command", + _runtime_command(capabilities, doubao_ready=True), + ) + endpoint = "http://doubao-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_profile_kind(endpoint, "authenticated") + + descriptors = { + item.capability_id: item for item in await capabilities.probe_capabilities() } - pool.set_profile_kind("http://default-profile:9222", "anonymous") - command.side_effect = [ - (0, f"{capabilities.OHMYOPENCLI_COMMIT}\n"), - (0, ""), - (0, ""), - (0, "1.8.5\n"), - (0, "official-site observe help"), - (1, "CDP not reachable at http://127.0.0.1:9"), - ] - [ready] = await capabilities.probe_capabilities() - assert ready.ready is True - assert ready.unavailable_reason is None + assert descriptors["chat-ai.capture"].ready is True + assert descriptors["chat-ai.capture"].target == "doubao" + assert descriptors["chat-ai.capture"].unavailable_reason is None + assert descriptors["official-site.observe"].unavailable_reason == "no_clean_profile" + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("protocol", "collector_name"), + [("http", "_collect_via_agent"), ("ws", "_collect_via_ws_agent")], +) +async def test_agent_mode_probes_the_selected_remote_browser_route( + monkeypatch, + protocol, + collector_name, +): + from backend.acquisition import capabilities + from backend.channels import opencli_channel + + endpoint = "agent-node-1" + agent_url = ( + "http://agent-node-1:19823" + if protocol == "http" + else "ws://agent-node-1/session" + ) + pool = init_pool([endpoint], use_redis=False) + pool.set_mode(endpoint, "cdp") + pool.set_profile_kind(endpoint, "authenticated") + pool.set_agent_url(endpoint, agent_url) + pool.set_agent_protocol(endpoint, protocol) + collector = AsyncMock( + return_value=ChannelResult.ok( + [ + { + "unattendedReady": True, + "loginDetected": False, + "promptInputDetected": True, + "sendButtonDetected": True, + "url": "https://www.doubao.com/chat/", + } + ] + ) + ) + monkeypatch.setattr(opencli_channel, collector_name, collector) + monkeypatch.setattr( + "backend.config.get_settings", + lambda: SimpleNamespace(collection_mode="agent"), + ) + local_command = AsyncMock() + monkeypatch.setattr(capabilities, "_command", local_command) + registration = next( + item + for item in capabilities.list_capability_registrations() + if item.target == "doubao" + ) + + assert await capabilities._session_is_ready(registration, pool, endpoint) is True + collector.assert_awaited_once_with( + agent_url, + "doubao", + "session-probe", + {}, + [], + "json", + "cdp", + None, + ) + local_command.assert_not_awaited() @pytest.mark.asyncio @@ -96,48 +216,38 @@ async def test_runtime_probe_uses_the_configured_opencli_binary(monkeypatch): configured_bin = r"C:\managed\opencli.cmd" monkeypatch.setenv("OPENCLI_BIN", configured_bin) - command = AsyncMock( - side_effect=[ - (0, f"{capabilities.OHMYOPENCLI_COMMIT}\n"), - (0, ""), - (0, ""), - (0, "1.8.5\n"), - (0, "official-site observe help"), - (1, "CDP not reachable at http://127.0.0.1:9"), - ] - ) + command = _runtime_command(capabilities) monkeypatch.setattr(capabilities, "_command", command) assert await capabilities._runtime_is_installed() is True - assert command.await_args_list[3].args == (configured_bin, "--version") + version_call = next( + call for call in command.await_args_list if call.args[-1] == "--version" + ) + assert version_call.args == (configured_bin, "--version") registration = capabilities.list_capability_registrations()[0] assert await capabilities._registration_is_available(registration) is True - assert command.await_args_list[4].args == ( - configured_bin, - "official-site", - "observe", - "--help", + help_call = next( + call + for call in command.await_args_list + if call.args[-3:] == ("official-site", "observe", "--help") ) + assert help_call.args[0] == configured_bin @pytest.mark.asyncio async def test_runtime_probe_rejects_tracked_checkout_changes(monkeypatch): from backend.acquisition import capabilities - command = AsyncMock( - side_effect=[ - (0, f"{capabilities.OHMYOPENCLI_COMMIT}\n"), - (0, ""), - (0, " M adapters/official-site/observe.js\n"), - ] + command = _runtime_command( + capabilities, + dirty=" M adapters/official-site/observe.js\n", ) monkeypatch.setattr(capabilities, "_command", command) assert await capabilities._runtime_is_installed() is False - assert command.await_count == 3 - assert command.await_args_list[2].args[-2:] == ( - "--porcelain", - "--untracked-files=no", + assert any( + call.args[-2:] == ("--porcelain", "--untracked-files=no") + for call in command.await_args_list ) @@ -156,14 +266,17 @@ async def test_catalog_stays_ready_while_anonymous_inventory_is_busy(monkeypatch pool.set_profile_kind(endpoint, "anonymous") async with pool.acquire(): - [descriptor] = await capabilities.probe_capabilities() + descriptors = await capabilities.probe_capabilities() - assert descriptor.ready is True - assert descriptor.unavailable_reason is None + official = next( + item for item in descriptors if item.capability_id == "official-site.observe" + ) + assert official.ready is True + assert official.unavailable_reason is None @pytest.mark.asyncio -async def test_catalog_omits_a_capability_when_its_real_command_is_not_registered( +async def test_catalog_omits_capabilities_whose_real_commands_are_not_registered( monkeypatch, ): from backend.acquisition import capabilities @@ -174,14 +287,14 @@ async def test_catalog_omits_a_capability_when_its_real_command_is_not_registere monkeypatch.setattr( capabilities, "_command", - AsyncMock(return_value=(1, "unknown command: official-site observe")), + AsyncMock(return_value=(1, "unknown command")), ) assert await capabilities.probe_capabilities() == [] @pytest.mark.asyncio -async def test_catalog_rejects_opencli_root_help_for_an_unknown_site(monkeypatch): +async def test_catalog_rejects_opencli_root_help_for_unknown_sites(monkeypatch): from backend.acquisition import capabilities monkeypatch.setattr( @@ -191,27 +304,4 @@ async def test_catalog_rejects_opencli_root_help_for_an_unknown_site(monkeypatch monkeypatch.setattr(capabilities, "_command", command) assert await capabilities.probe_capabilities() == [] - command.assert_awaited_once() - - -@pytest.mark.asyncio -async def test_catalog_does_not_invent_chat_ai_capture(monkeypatch): - from backend.acquisition import capabilities - - monkeypatch.setattr( - capabilities, "_runtime_is_installed", AsyncMock(return_value=True) - ) - monkeypatch.setattr( - capabilities, - "_command", - AsyncMock( - side_effect=[ - (0, "official-site observe help"), - (1, "CDP not reachable at http://127.0.0.1:9"), - ] - ), - ) - - descriptors = await capabilities.probe_capabilities() - - assert [item.capability_id for item in descriptors] == ["official-site.observe"] + assert command.await_count == 2 diff --git a/tests/unit/test_acquisition_runner.py b/tests/unit/test_acquisition_runner.py index 6c425b5..4e7a58d 100644 --- a/tests/unit/test_acquisition_runner.py +++ b/tests/unit/test_acquisition_runner.py @@ -5,6 +5,10 @@ import pytest from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from backend.acquisition.registry import ( + OFFICIAL_SITE_CAPABILITY_COMMIT, + OHMYOPENCLI_COMMIT, +) from backend.browser_pool import init_pool from backend.channels.base import ChannelResult from backend.models.acquisition import AcquisitionExecutionStatus @@ -41,6 +45,24 @@ def _submission() -> AcquisitionSubmission: ) +@pytest.mark.parametrize( + ("message", "code"), + [ + ("DOUBAO_CAPTURE_LOGIN_WALL", "login_required"), + ("DOUBAO_CAPTURE_CAPTCHA", "captcha_required"), + ("DOUBAO_CAPTURE_TIMEOUT", "capture_timeout"), + ("DOUBAO_CAPTURE_REFUSAL", "model_refusal"), + ("DOUBAO_CAPTURE_EMPTY_ANSWER", "empty_answer"), + ("DOUBAO_CAPTURE_PAGE_DRIFT", "page_contract_drift"), + ], +) +def test_doubao_typed_failures_are_preserved(message, code): + from backend.acquisition.runner import _capability_failure_code + + assert _capability_failure_code(message) == code + assert _capability_failure_code("generic adapter failure", message) == code + + @pytest.mark.asyncio async def test_official_site_execution_rejects_private_target_before_opencli( db_engine, monkeypatch @@ -142,6 +164,7 @@ async def collect(*_args, **_kwargs): } ], trace_artifact="artifact://trace/1", + trace_sha256="f" * 64, ) channel = AsyncMock() @@ -190,6 +213,7 @@ async def collect(*_args, **_kwargs): } ], trace_artifact="artifact://trace/1", + trace_sha256="f" * 64, ) task = asyncio.create_task( @@ -396,6 +420,44 @@ async def test_required_trace_must_be_returned_by_the_channel(db_engine): assert execution.trace_ref is None +@pytest.mark.asyncio +async def test_required_trace_must_include_a_content_hash(db_engine): + from backend.acquisition.runner import run_acquisition_execution + + sessions = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + async with sessions() as db: + outcome = await acquisition_service.submit_execution(db, _submission()) + await acquisition_service.queue_execution(db, outcome.execution) + execution_id = outcome.execution.id + + endpoint = "http://unhashed-trace-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_profile_kind(endpoint, "anonymous") + channel = AsyncMock() + channel.collect.return_value = ChannelResult.ok( + [ + { + "capabilityId": "official-site.observe", + "capabilityVersion": "1.0.0", + "outputSchemaVersion": "1", + } + ], + trace_artifact="artifact://trace/unhashed", + ) + + await run_acquisition_execution( + execution_id, session_factory=sessions, channel=channel + ) + + async with sessions() as db: + execution = await acquisition_service.get_execution(db, execution_id) + assert execution is not None + assert execution.status == AcquisitionExecutionStatus.FAILED + assert execution.failure["code"] == "required_artifact_missing" + + @pytest.mark.asyncio async def test_unknown_capability_invocation_fails_closed(db_engine): from backend.acquisition.runner import run_acquisition_execution @@ -439,13 +501,271 @@ def test_dispatch_registry_contains_only_real_versioned_capabilities(): registrations = list_capability_registrations() assert [registration.identity for registration in registrations] == [ - ("official-site.observe", "1.0.0", "1") + ("official-site.observe", "1.0.0", "1", None), + ("chat-ai.capture", "1.0.0", "1", "doubao"), ] assert registrations[0].invocation == { "site": "official-site", "command": "observe", "format": "json", } + assert registrations[1].invocation == { + "site": "doubao", + "command": "capture", + "format": "json", + } + assert registrations[1].required_profile_kind == "authenticated" + + +@pytest.mark.asyncio +async def test_doubao_execution_uses_authenticated_profile_and_frozen_prompt( + db_engine, + monkeypatch, +): + from backend.acquisition import capabilities + from backend.acquisition.runner import run_acquisition_execution + + sessions = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + submission_data = _submission().model_dump() + submission_data.update( + { + "request_id": "doubao-request-1", + "idempotency_key": "doubao-attempt-1", + "capability": {"id": "chat-ai.capture", "version": "1.0.0"}, + "input": { + "target": "doubao", + "prompt": "黑白调电竞椅值得买吗？", + }, + } + ) + submission = AcquisitionSubmission.model_validate(submission_data) + async with sessions() as db: + outcome = await acquisition_service.submit_execution(db, submission) + await acquisition_service.queue_execution(db, outcome.execution) + execution_id = outcome.execution.id + + endpoint = "http://doubao-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_mode(endpoint, "cdp") + pool.set_profile_kind(endpoint, "authenticated") + payload = { + "capabilityId": "chat-ai.capture", + "capabilityVersion": "1.0.0", + "outputSchemaVersion": "1", + "target": "doubao", + "prompt": "黑白调电竞椅值得买吗？", + "completionState": "complete", + "answer": {"text": "真实回答", "sha256": "a" * 64}, + "citations": [], + "displayedUrl": "https://www.doubao.com/chat/1", + "finalUrl": "https://www.doubao.com/chat/1", + "pageState": "answer", + "artifacts": [], + } + channel = AsyncMock() + session_probe = AsyncMock(return_value=True) + monkeypatch.setattr(capabilities, "_session_is_ready", session_probe) + + async def collect_while_leased(*_args, **_kwargs): + assert pool.available_for(endpoint) is False + return ChannelResult.ok( + [payload], + trace_artifact="artifact://trace/doubao-1", + trace_sha256="f" * 64, + ) + + channel.collect.side_effect = collect_while_leased + + await run_acquisition_execution( + execution_id, session_factory=sessions, channel=channel + ) + + channel.collect.assert_awaited_once_with( + {"site": "doubao", "command": "capture", "format": "json"}, + { + "prompt": "黑白调电竞椅值得买吗？", + "chrome_endpoint": endpoint, + "required_profile_kind": "authenticated", + "_endpoint_preacquired": True, + "trace": "on", + }, + ) + assert session_probe.await_args.args[2] == endpoint + async with sessions() as db: + execution = await acquisition_service.get_execution(db, execution_id) + assert execution is not None + assert execution.status == AcquisitionExecutionStatus.SUCCEEDED + assert execution.result_payload["payload"] == payload + assert execution.result_payload["operational"]["browser"] == { + "endpoint": endpoint, + "profile_kind": "authenticated", + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("field", "value"), + [ + ("target", "chatgpt"), + ("prompt", "a different prompt"), + ], +) +async def test_doubao_rejects_target_or_prompt_drift( + db_engine, + monkeypatch, + field, + value, +): + from backend.acquisition import capabilities + from backend.acquisition.runner import run_acquisition_execution + + sessions = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + submission_data = _submission().model_dump() + submission_data.update( + { + "request_id": f"doubao-drift-{field}", + "idempotency_key": f"doubao-drift-{field}", + "capability": {"id": "chat-ai.capture", "version": "1.0.0"}, + "input": { + "target": "doubao", + "prompt": "黑白调电竞椅值得买吗？", + }, + } + ) + async with sessions() as db: + outcome = await acquisition_service.submit_execution( + db, AcquisitionSubmission.model_validate(submission_data) + ) + await acquisition_service.queue_execution(db, outcome.execution) + execution_id = outcome.execution.id + + endpoint = "http://doubao-drift-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_profile_kind(endpoint, "authenticated") + monkeypatch.setattr( + capabilities, + "_session_is_ready", + AsyncMock(return_value=True), + ) + payload = { + "capabilityId": "chat-ai.capture", + "capabilityVersion": "1.0.0", + "outputSchemaVersion": "1", + "target": "doubao", + "prompt": "黑白调电竞椅值得买吗？", + } + payload[field] = value + channel = AsyncMock() + channel.collect.return_value = ChannelResult.ok([payload]) + + await run_acquisition_execution( + execution_id, session_factory=sessions, channel=channel + ) + + async with sessions() as db: + execution = await acquisition_service.get_execution(db, execution_id) + assert execution is not None + assert execution.status == AcquisitionExecutionStatus.FAILED + assert execution.failure["code"] == "invalid_capability_envelope" + + +@pytest.mark.asyncio +async def test_doubao_execution_fails_closed_without_authenticated_profile(db_engine): + from backend.acquisition.runner import run_acquisition_execution + + sessions = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + submission_data = _submission().model_dump() + submission_data.update( + { + "request_id": "doubao-request-2", + "idempotency_key": "doubao-attempt-2", + "capability": {"id": "chat-ai.capture", "version": "1.0.0"}, + "input": { + "target": "doubao", + "prompt": "黑白调电竞椅值得买吗？", + }, + } + ) + async with sessions() as db: + outcome = await acquisition_service.submit_execution( + db, AcquisitionSubmission.model_validate(submission_data) + ) + await acquisition_service.queue_execution(db, outcome.execution) + execution_id = outcome.execution.id + + endpoint = "http://anonymous-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_profile_kind(endpoint, "anonymous") + channel = AsyncMock() + + await run_acquisition_execution( + execution_id, session_factory=sessions, channel=channel + ) + + channel.collect.assert_not_awaited() + async with sessions() as db: + execution = await acquisition_service.get_execution(db, execution_id) + assert execution is not None + assert execution.failure == { + "code": "no_authenticated_profile", + "message": "no_authenticated_profile", + } + + +@pytest.mark.asyncio +async def test_doubao_rechecks_the_selected_session_before_prompt_submission( + db_engine, + monkeypatch, +): + from backend.acquisition import capabilities + from backend.acquisition.runner import run_acquisition_execution + + sessions = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + submission_data = _submission().model_dump() + submission_data.update( + { + "request_id": "doubao-request-3", + "idempotency_key": "doubao-attempt-3", + "capability": {"id": "chat-ai.capture", "version": "1.0.0"}, + "input": {"target": "doubao", "prompt": "黑白调电竞椅值得买吗？"}, + } + ) + async with sessions() as db: + outcome = await acquisition_service.submit_execution( + db, AcquisitionSubmission.model_validate(submission_data) + ) + await acquisition_service.queue_execution(db, outcome.execution) + execution_id = outcome.execution.id + + endpoint = "http://expired-doubao-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_profile_kind(endpoint, "authenticated") + session_probe = AsyncMock(return_value=False) + monkeypatch.setattr(capabilities, "_session_is_ready", session_probe) + channel = AsyncMock() + + await run_acquisition_execution( + execution_id, session_factory=sessions, channel=channel + ) + + session_probe.assert_awaited_once() + assert session_probe.await_args.args[2] == endpoint + channel.collect.assert_not_awaited() + async with sessions() as db: + execution = await acquisition_service.get_execution(db, execution_id) + assert execution is not None + assert execution.failure == { + "code": "session_not_qualified", + "message": "doubao session failed the execution-time readiness probe", + } @pytest.mark.asyncio @@ -540,6 +860,7 @@ async def test_redis_worker_registers_a_dynamic_anonymous_profile_before_dispatc } ], trace_artifact="artifact://trace/1", + trace_sha256="f" * 64, ) monkeypatch.setattr(pool, "_client", lambda: redis_cm) @@ -588,6 +909,7 @@ async def test_official_site_execution_preserves_payload_in_versioned_envelope( command="observe", chrome_mode="cdp", trace_artifact="artifact://trace/1", + trace_sha256="f" * 64, ) await run_acquisition_execution( @@ -602,6 +924,7 @@ async def test_official_site_execution_preserves_payload_in_versioned_envelope( "url": "https://example.com", "chrome_endpoint": "http://clean-profile:9222", "required_profile_kind": "anonymous", + "_endpoint_preacquired": True, "trace": "on", }, ) @@ -616,12 +939,8 @@ async def test_official_site_execution_preserves_payload_in_versioned_envelope( "payload": payload, "operational": { "runtime": { - "ohmyopencli_repo_commit": ( - "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" - ), - "capability_source_commit": ( - "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" - ), + "ohmyopencli_repo_commit": OHMYOPENCLI_COMMIT, + "capability_source_commit": OFFICIAL_SITE_CAPABILITY_COMMIT, "opencli_version": "1.8.5", }, "browser": { @@ -633,12 +952,17 @@ async def test_official_site_execution_preserves_payload_in_versioned_envelope( "command": "observe", "chrome_mode": "cdp", "trace_artifact": "artifact://trace/1", + "trace_sha256": "f" * 64, }, }, } assert execution.artifact_refs == [ *payload["artifacts"], - {"kind": "trace", "ref": "artifact://trace/1"}, + { + "kind": "trace", + "ref": "artifact://trace/1", + "sha256": "f" * 64, + }, ] assert execution.trace_ref == "artifact://trace/1" diff --git a/tests/unit/test_agent_image_runtime_packaging.py b/tests/unit/test_agent_image_runtime_packaging.py index 000bb76..a6a7963 100644 --- a/tests/unit/test_agent_image_runtime_packaging.py +++ b/tests/unit/test_agent_image_runtime_packaging.py @@ -17,7 +17,11 @@ def test_agent_image_pins_managed_acquisition_runtime(): assert "ARG OPENCLI_VERSION=1.8.5" in dockerfile assert ( "ARG OHMYOPENCLI_COMMIT=" - "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" + "bfe1c25b4b12661058dd6e9980c562a09f230cc7" + ) in dockerfile + assert ( + "ARG DOUBAO_CAPABILITY_COMMIT=" + "bfe1c25b4b12661058dd6e9980c562a09f230cc7" ) in dockerfile assert "git checkout --detach ${OHMYOPENCLI_COMMIT}" in dockerfile assert "npm ci" in dockerfile diff --git a/tests/unit/test_agent_server.py b/tests/unit/test_agent_server.py index 3629f4c..a8a489a 100644 --- a/tests/unit/test_agent_server.py +++ b/tests/unit/test_agent_server.py @@ -42,6 +42,41 @@ def test_resolve_bin_treats_empty_opencli_bin_as_default(monkeypatch): assert agent_server._resolve_bin("cdp") == "opencli" +@pytest.mark.asyncio +async def test_runtime_lineage_uses_the_dispatched_adapter_source(monkeypatch): + calls = [] + outputs = iter( + [ + b"runtime-commit\n", + b"doubao-source-commit\n", + b"opencli 1.8.5\n", + ] + ) + + class Process: + returncode = 0 + + async def communicate(self): + return next(outputs), b"" + + async def create(*args, **kwargs): + calls.append((args, kwargs)) + return Process() + + monkeypatch.setattr(agent_server.asyncio, "create_subprocess_exec", create) + + lineage = await agent_server._runtime_lineage( + "opencli", "doubao", "capture" + ) + + assert lineage == { + "ohmyopencli_repo_commit": "runtime-commit", + "capability_source_commit": "doubao-source-commit", + "opencli_version": "1.8.5", + } + assert calls[1][0][-2:] == ("--", "adapters/doubao/capture.js") + + # ── _auth_headers ──────────────────────────────────────────────────────────── diff --git a/tests/unit/test_geo_acquisition_api.py b/tests/unit/test_geo_acquisition_api.py index c9f9932..e6aca2c 100644 --- a/tests/unit/test_geo_acquisition_api.py +++ b/tests/unit/test_geo_acquisition_api.py @@ -8,6 +8,10 @@ from httpx import ASGITransport, AsyncClient from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker +from backend.acquisition.registry import ( + OFFICIAL_SITE_CAPABILITY_COMMIT, + OHMYOPENCLI_COMMIT, +) from backend.database import get_db from backend.main import create_app from backend.schemas.acquisition import CapabilityDescriptor @@ -25,12 +29,8 @@ async def probed_capabilities(): output_schema_version="1", ready=True, runtime={ - "ohmyopencli_repo_commit": ( - "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" - ), - "capability_source_commit": ( - "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" - ), + "ohmyopencli_repo_commit": OHMYOPENCLI_COMMIT, + "capability_source_commit": OFFICIAL_SITE_CAPABILITY_COMMIT, "opencli_version": "1.8.5", }, ) @@ -84,14 +84,11 @@ async def test_geo_can_discover_submit_observe_and_cancel_an_execution( "capability_id": "official-site.observe", "capability_version": "1.0.0", "output_schema_version": "1", + "target": None, "ready": True, "runtime": { - "ohmyopencli_repo_commit": ( - "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" - ), - "capability_source_commit": ( - "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" - ), + "ohmyopencli_repo_commit": OHMYOPENCLI_COMMIT, + "capability_source_commit": OFFICIAL_SITE_CAPABILITY_COMMIT, "opencli_version": "1.8.5", }, "unavailable_reason": None, @@ -270,6 +267,42 @@ async def test_unknown_or_mismatched_versions_are_rejected( assert response.json()["detail"]["code"] == code +@pytest.mark.asyncio +async def test_chat_ai_target_must_match_a_registered_adapter( + client, + monkeypatch, + acquisition_executor, +): + async def doubao_capability(): + return [ + CapabilityDescriptor( + capability_id="chat-ai.capture", + capability_version="1.0.0", + output_schema_version="1", + target="doubao", + ready=True, + ) + ] + + monkeypatch.setattr( + "backend.api.v1.geo_acquisition.probe_capabilities", + doubao_capability, + ) + request = _request( + capability={"id": "chat-ai.capture", "version": "1.0.0"}, + input={"target": "chatgpt", "prompt": "prompt"}, + ) + + response = await client.post(f"{BASE}/executions", json=request) + + assert response.status_code == 422 + assert response.json()["detail"] == { + "code": "target_not_registered", + "message": "chatgpt", + } + acquisition_executor.dispatch_acquisition.assert_not_awaited() + + @pytest.mark.asyncio async def test_execution_is_observable_from_a_new_app_after_restart( db_engine, acquisition_executor diff --git a/tests/unit/test_managed_opencli_verifier.py b/tests/unit/test_managed_opencli_verifier.py index 9dd5c2f..fec3cfc 100644 --- a/tests/unit/test_managed_opencli_verifier.py +++ b/tests/unit/test_managed_opencli_verifier.py @@ -4,7 +4,7 @@ from scripts.verify_managed_opencli_runtime import VerificationError, verify_runtime -PINNED_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" +PINNED_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" def _completed(args, returncode=0, stdout="", stderr=""): @@ -18,8 +18,15 @@ def test_contract_verifier_accepts_the_pinned_real_command(tmp_path): _completed([], stdout=f"{PINNED_COMMIT}\n"), _completed([]), _completed([]), + _completed([]), _completed([], stdout="opencli 1.8.5\n"), _completed([], stdout="Usage: opencli official-site observe"), + _completed([], stdout="Usage: opencli doubao capture"), + _completed( + [], + returncode=1, + stderr="CDP not reachable at http://127.0.0.1:9", + ), _completed( [], returncode=1, @@ -40,12 +47,32 @@ def run(args, **kwargs): assert report["contract_ready"] is True assert report["trace_ready"] is False - assert calls[4][0] == [ + assert calls[5][0] == [ "managed-opencli", "official-site", "observe", "--help", ] + assert calls[6][0] == ["managed-opencli", "doubao", "capture", "--help"] + assert report["capability_contracts"]["chat-ai.capture:doubao"] is True + + +def test_contract_verifier_rejects_a_missing_doubao_capture_command(tmp_path): + def run(args, **kwargs): + if args[-3:] == ["doubao", "capture", "--help"]: + return _completed(args, returncode=1, stderr="unknown command") + if args[-2:] == ["rev-parse", "HEAD"]: + return _completed(args, stdout=f"{PINNED_COMMIT}\n") + if args[-1:] == ["--version"]: + return _completed(args, stdout="opencli 1.8.5\n") + return _completed(args) + + with pytest.raises(VerificationError, match="doubao capture --help"): + verify_runtime( + ohmyopencli_root=tmp_path, + opencli_bin="managed-opencli", + run=run, + ) def test_live_verifier_requires_a_real_versioned_payload_and_trace(tmp_path): @@ -56,8 +83,15 @@ def test_live_verifier_requires_a_real_versioned_payload_and_trace(tmp_path): _completed([], stdout=f"{PINNED_COMMIT}\n"), _completed([]), _completed([]), + _completed([]), _completed([], stdout="opencli 1.8.5\n"), _completed([], stdout="Usage: opencli official-site observe"), + _completed([], stdout="Usage: opencli doubao capture"), + _completed( + [], + returncode=1, + stderr="CDP not reachable at http://127.0.0.1:9", + ), _completed( [], returncode=1, @@ -94,8 +128,15 @@ def test_live_verifier_rejects_success_without_a_trace(tmp_path): _completed([], stdout=f"{PINNED_COMMIT}\n"), _completed([]), _completed([]), + _completed([]), _completed([], stdout="opencli 1.8.5\n"), _completed([], stdout="Usage: opencli official-site observe"), + _completed([], stdout="Usage: opencli doubao capture"), + _completed( + [], + returncode=1, + stderr="CDP not reachable at http://127.0.0.1:9", + ), _completed( [], returncode=1, From 7dcdb4c73e1e8b111ca9f27e95b0ebaefd255e54 Mon Sep 17 00:00:00 2001 From: lunnt <2276214182@qq.com> Date: Wed, 29 Jul 2026 13:08:58 +0800 Subject: [PATCH 2/7] refactor(acquisition): isolate OpenCLI support helpers --- backend/channels/opencli_channel.py | 148 +---------- backend/channels/opencli_support.py | 131 ++++++++++ tests/unit/test_acquisition_runner.py | 260 +------------------ tests/unit/test_doubao_acquisition_runner.py | 232 +++++++++++++++++ 4 files changed, 378 insertions(+), 393 deletions(-) create mode 100644 backend/channels/opencli_support.py create mode 100644 tests/unit/test_doubao_acquisition_runner.py diff --git a/backend/channels/opencli_channel.py b/backend/channels/opencli_channel.py index a4f293c..9853ab6 100644 --- a/backend/channels/opencli_channel.py +++ b/backend/channels/opencli_channel.py @@ -1,23 +1,19 @@ """OpenCLI channel: invokes opencli CLI tool and parses its output.""" import asyncio -import csv -import hashlib -import io -import json import logging import os import re import subprocess import time -from contextlib import asynccontextmanager -from pathlib import Path from typing import Any from urllib.parse import urlparse import yaml from sqlalchemy.ext.asyncio import AsyncSession +import backend.channels.opencli_support as _opencli_support +import backend.opencli_runtime as _opencli_runtime from backend.channels.base import ( AbstractChannel, Capabilities, @@ -26,7 +22,17 @@ FetchResult, ) from backend.channels.registry import register_channel -from backend.opencli_runtime import configured_opencli_bin, resolve_opencli_bin + +_artifact_sha256 = _opencli_support.artifact_sha256 +_browser_endpoint_lease = _opencli_support.browser_endpoint_lease +_extract_opencli_error = _opencli_support.extract_opencli_error +_parse_csv = _opencli_support.parse_csv +_parse_json = _opencli_support.parse_json +_parse_markdown = _opencli_support.parse_markdown +_parse_table = _opencli_support.parse_table +_parse_yaml = _opencli_support.parse_yaml +configured_opencli_bin = _opencli_runtime.configured_opencli_bin +resolve_opencli_bin = _opencli_runtime.resolve_opencli_bin logger = logging.getLogger(__name__) @@ -87,28 +93,6 @@ def _split_routing_parameters( return (chrome_endpoint, required_profile_kind), cli_parameters -@asynccontextmanager -async def _browser_endpoint_lease( - pool: Any, - endpoint: str | None, - required_profile_kind: str | None, - *, - preacquired: bool, -): - """Reuse a runner-owned endpoint lease or acquire one for legacy callers.""" - if preacquired: - if not endpoint: - raise ValueError("A pre-acquired browser lease requires chrome_endpoint") - yield endpoint - return - - acquire_kwargs: dict[str, Any] = {"endpoint": endpoint} - if required_profile_kind: - acquire_kwargs["required_profile_kind"] = required_profile_kind - async with pool.acquire(**acquire_kwargs) as leased_endpoint: - yield leased_endpoint - - async def _site_bound_agent_endpoint(pool: Any, site: str, session: AsyncSession) -> str | None: if not site: return None @@ -283,65 +267,6 @@ def _resolve_bin(mode: str) -> str: # noqa: ARG001 — mode unused, kept for ca return resolve_opencli_bin() -def _parse_json(raw: str) -> list[dict]: - json_start = next((i for i, ch in enumerate(raw) if ch in ("{", "[")), None) - if json_start is None: - raise ValueError(f"No JSON found in output: {raw[:200]!r}") - data = json.loads(raw[json_start:]) - return data if isinstance(data, list) else [data] - - -def _parse_yaml(raw: str) -> list[dict]: - data = yaml.safe_load(raw) - if isinstance(data, list): - return data - if isinstance(data, dict): - return [data] - return [{"content": str(data)}] - - -def _parse_csv(raw: str) -> list[dict]: - reader = csv.DictReader(io.StringIO(raw.strip())) - return [row for row in reader] - - -def _parse_table(raw: str) -> list[dict]: - """Parse cli-table3 Unicode box-drawing table into list of dicts.""" - lines = raw.splitlines() - data_lines = [line for line in lines if line.strip().startswith("│")] - if not data_lines: - return [{"content": raw}] - - def split_row(line: str) -> list[str]: - return [cell.strip() for cell in line.strip().strip("│").split("│")] - - headers = split_row(data_lines[0]) - rows = [] - for line in data_lines[1:]: - cells = split_row(line) - if len(cells) == len(headers): - rows.append(dict(zip(headers, cells))) - return rows if rows else [{"content": raw}] - - -def _parse_markdown(raw: str) -> list[dict]: - """Parse markdown table into list of dicts.""" - lines = [line.strip() for line in raw.splitlines() if line.strip().startswith("|")] - if len(lines) < 2: - return [{"content": raw}] - - def split_row(line: str) -> list[str]: - return [cell.strip() for cell in line.strip().strip("|").split("|")] - - headers = split_row(lines[0]) - rows = [] - for line in lines[2:]: - cells = split_row(line) - if len(cells) == len(headers): - rows.append(dict(zip(headers, cells))) - return rows if rows else [{"content": raw}] - - _PARSERS = { "json": _parse_json, "yaml": _parse_yaml, @@ -576,53 +501,6 @@ async def _run_opencli(cmd: list[str], env: dict) -> tuple[int, str, str]: raise -def _extract_opencli_error(stderr_text: str) -> tuple[str | None, str | None]: - """Read OpenCLI's structured error envelope without depending on its prose.""" - try: - envelope = yaml.safe_load(stderr_text) - except yaml.YAMLError: - envelope = None - if isinstance(envelope, dict) and isinstance(envelope.get("error"), dict): - error = envelope["error"] - code = str(error.get("code") or "").strip() or None - message = str(error.get("message") or "").strip() or None - return code, message - - # OpenCLI may append a human update notice after the YAML envelope. Retain - # the typed code even when that makes the complete stderr invalid YAML. - code_match = re.search( - r"(?m)^\s+code:\s*['\"]?([A-Za-z0-9_-]+)['\"]?\s*$", - stderr_text, - ) - message_match = re.search( - r"(?m)^\s+message:\s*(.+?)\s*$", - stderr_text, - ) - code = code_match.group(1) if code_match else None - message = message_match.group(1).strip(" '\"") if message_match else None - return code, message - - -def _artifact_sha256(artifact_ref: str) -> str | None: - """Hash a trace file/directory so its persisted reference is auditable.""" - root = Path(artifact_ref) - if not root.exists(): - return None - files = [root] if root.is_file() else sorted( - (path for path in root.rglob("*") if path.is_file()), - key=lambda path: path.as_posix(), - ) - digest = hashlib.sha256() - base = root.parent if root.is_file() else root - for path in files: - digest.update(path.relative_to(base).as_posix().encode()) - digest.update(b"\0") - with path.open("rb") as handle: - while chunk := handle.read(1024 * 1024): - digest.update(chunk) - return digest.hexdigest() - - async def _collect_with_opencli_subprocess( cmd: list[str], env: dict, diff --git a/backend/channels/opencli_support.py b/backend/channels/opencli_support.py new file mode 100644 index 0000000..50af9fb --- /dev/null +++ b/backend/channels/opencli_support.py @@ -0,0 +1,131 @@ +"""Small execution helpers shared by the managed OpenCLI channel.""" + +import csv +import hashlib +import io +import json +import re +from contextlib import asynccontextmanager +from pathlib import Path +from typing import Any + +import yaml + + +def parse_json(raw: str) -> list[dict]: + json_start = next((i for i, ch in enumerate(raw) if ch in ("{", "[")), None) + if json_start is None: + raise ValueError(f"No JSON found in output: {raw[:200]!r}") + data = json.loads(raw[json_start:]) + return data if isinstance(data, list) else [data] + + +def parse_yaml(raw: str) -> list[dict]: + data = yaml.safe_load(raw) + if isinstance(data, list): + return data + if isinstance(data, dict): + return [data] + return [{"content": str(data)}] + + +def parse_csv(raw: str) -> list[dict]: + return list(csv.DictReader(io.StringIO(raw.strip()))) + + +def parse_table(raw: str) -> list[dict]: + """Parse a cli-table3 Unicode box-drawing table.""" + lines = [line for line in raw.splitlines() if line.strip().startswith("│")] + if not lines: + return [{"content": raw}] + split_row = lambda line: [ # noqa: E731 + cell.strip() for cell in line.strip().strip("│").split("│") + ] + headers = split_row(lines[0]) + rows = [ + dict(zip(headers, cells)) + for line in lines[1:] + if len(cells := split_row(line)) == len(headers) + ] + return rows or [{"content": raw}] + + +def parse_markdown(raw: str) -> list[dict]: + """Parse a markdown table.""" + lines = [line.strip() for line in raw.splitlines() if line.strip().startswith("|")] + if len(lines) < 2: + return [{"content": raw}] + split_row = lambda line: [ # noqa: E731 + cell.strip() for cell in line.strip().strip("|").split("|") + ] + headers = split_row(lines[0]) + rows = [ + dict(zip(headers, cells)) + for line in lines[2:] + if len(cells := split_row(line)) == len(headers) + ] + return rows or [{"content": raw}] + + +@asynccontextmanager +async def browser_endpoint_lease( + pool: Any, + endpoint: str | None, + required_profile_kind: str | None, + *, + preacquired: bool, +): + """Reuse a runner-owned endpoint lease or acquire one for legacy callers.""" + if preacquired: + if not endpoint: + raise ValueError("A pre-acquired browser lease requires chrome_endpoint") + yield endpoint + return + + acquire_kwargs: dict[str, Any] = {"endpoint": endpoint} + if required_profile_kind: + acquire_kwargs["required_profile_kind"] = required_profile_kind + async with pool.acquire(**acquire_kwargs) as leased_endpoint: + yield leased_endpoint + + +def extract_opencli_error(stderr_text: str) -> tuple[str | None, str | None]: + """Read OpenCLI's structured error envelope without depending on its prose.""" + try: + envelope = yaml.safe_load(stderr_text) + except yaml.YAMLError: + envelope = None + if isinstance(envelope, dict) and isinstance(envelope.get("error"), dict): + error = envelope["error"] + code = str(error.get("code") or "").strip() or None + message = str(error.get("message") or "").strip() or None + return code, message + + code_match = re.search( + r"(?m)^\s+code:\s*['\"]?([A-Za-z0-9_-]+)['\"]?\s*$", + stderr_text, + ) + message_match = re.search(r"(?m)^\s+message:\s*(.+?)\s*$", stderr_text) + code = code_match.group(1) if code_match else None + message = message_match.group(1).strip(" '\"") if message_match else None + return code, message + + +def artifact_sha256(artifact_ref: str) -> str | None: + """Hash a trace file/directory so its persisted reference is auditable.""" + root = Path(artifact_ref) + if not root.exists(): + return None + files = [root] if root.is_file() else sorted( + (path for path in root.rglob("*") if path.is_file()), + key=lambda path: path.as_posix(), + ) + digest = hashlib.sha256() + base = root.parent if root.is_file() else root + for path in files: + digest.update(path.relative_to(base).as_posix().encode()) + digest.update(b"\0") + with path.open("rb") as handle: + while chunk := handle.read(1024 * 1024): + digest.update(chunk) + return digest.hexdigest() diff --git a/tests/unit/test_acquisition_runner.py b/tests/unit/test_acquisition_runner.py index 4e7a58d..612765c 100644 --- a/tests/unit/test_acquisition_runner.py +++ b/tests/unit/test_acquisition_runner.py @@ -22,9 +22,7 @@ def _public_url_guard(monkeypatch): async def validate(url, **_kwargs): return url - monkeypatch.setattr( - "backend.security.url_guard.avalidate_public_url", validate - ) + monkeypatch.setattr("backend.security.url_guard.avalidate_public_url", validate) def _submission() -> AcquisitionSubmission: @@ -32,10 +30,7 @@ def _submission() -> AcquisitionSubmission: { "request_id": "request-1", "idempotency_key": "attempt-1", - "capability": { - "id": "official-site.observe", - "version": "1.0.0", - }, + "capability": {"id": "official-site.observe", "version": "1.0.0"}, "output_schema_version": "1", "input": {"url": "https://example.com"}, "environment": {"locale": "zh-CN", "region": "CN"}, @@ -517,257 +512,6 @@ def test_dispatch_registry_contains_only_real_versioned_capabilities(): assert registrations[1].required_profile_kind == "authenticated" -@pytest.mark.asyncio -async def test_doubao_execution_uses_authenticated_profile_and_frozen_prompt( - db_engine, - monkeypatch, -): - from backend.acquisition import capabilities - from backend.acquisition.runner import run_acquisition_execution - - sessions = async_sessionmaker( - db_engine, class_=AsyncSession, expire_on_commit=False - ) - submission_data = _submission().model_dump() - submission_data.update( - { - "request_id": "doubao-request-1", - "idempotency_key": "doubao-attempt-1", - "capability": {"id": "chat-ai.capture", "version": "1.0.0"}, - "input": { - "target": "doubao", - "prompt": "黑白调电竞椅值得买吗？", - }, - } - ) - submission = AcquisitionSubmission.model_validate(submission_data) - async with sessions() as db: - outcome = await acquisition_service.submit_execution(db, submission) - await acquisition_service.queue_execution(db, outcome.execution) - execution_id = outcome.execution.id - - endpoint = "http://doubao-profile:9222" - pool = init_pool([endpoint], use_redis=False) - pool.set_mode(endpoint, "cdp") - pool.set_profile_kind(endpoint, "authenticated") - payload = { - "capabilityId": "chat-ai.capture", - "capabilityVersion": "1.0.0", - "outputSchemaVersion": "1", - "target": "doubao", - "prompt": "黑白调电竞椅值得买吗？", - "completionState": "complete", - "answer": {"text": "真实回答", "sha256": "a" * 64}, - "citations": [], - "displayedUrl": "https://www.doubao.com/chat/1", - "finalUrl": "https://www.doubao.com/chat/1", - "pageState": "answer", - "artifacts": [], - } - channel = AsyncMock() - session_probe = AsyncMock(return_value=True) - monkeypatch.setattr(capabilities, "_session_is_ready", session_probe) - - async def collect_while_leased(*_args, **_kwargs): - assert pool.available_for(endpoint) is False - return ChannelResult.ok( - [payload], - trace_artifact="artifact://trace/doubao-1", - trace_sha256="f" * 64, - ) - - channel.collect.side_effect = collect_while_leased - - await run_acquisition_execution( - execution_id, session_factory=sessions, channel=channel - ) - - channel.collect.assert_awaited_once_with( - {"site": "doubao", "command": "capture", "format": "json"}, - { - "prompt": "黑白调电竞椅值得买吗？", - "chrome_endpoint": endpoint, - "required_profile_kind": "authenticated", - "_endpoint_preacquired": True, - "trace": "on", - }, - ) - assert session_probe.await_args.args[2] == endpoint - async with sessions() as db: - execution = await acquisition_service.get_execution(db, execution_id) - assert execution is not None - assert execution.status == AcquisitionExecutionStatus.SUCCEEDED - assert execution.result_payload["payload"] == payload - assert execution.result_payload["operational"]["browser"] == { - "endpoint": endpoint, - "profile_kind": "authenticated", - } - - -@pytest.mark.asyncio -@pytest.mark.parametrize( - ("field", "value"), - [ - ("target", "chatgpt"), - ("prompt", "a different prompt"), - ], -) -async def test_doubao_rejects_target_or_prompt_drift( - db_engine, - monkeypatch, - field, - value, -): - from backend.acquisition import capabilities - from backend.acquisition.runner import run_acquisition_execution - - sessions = async_sessionmaker( - db_engine, class_=AsyncSession, expire_on_commit=False - ) - submission_data = _submission().model_dump() - submission_data.update( - { - "request_id": f"doubao-drift-{field}", - "idempotency_key": f"doubao-drift-{field}", - "capability": {"id": "chat-ai.capture", "version": "1.0.0"}, - "input": { - "target": "doubao", - "prompt": "黑白调电竞椅值得买吗？", - }, - } - ) - async with sessions() as db: - outcome = await acquisition_service.submit_execution( - db, AcquisitionSubmission.model_validate(submission_data) - ) - await acquisition_service.queue_execution(db, outcome.execution) - execution_id = outcome.execution.id - - endpoint = "http://doubao-drift-profile:9222" - pool = init_pool([endpoint], use_redis=False) - pool.set_profile_kind(endpoint, "authenticated") - monkeypatch.setattr( - capabilities, - "_session_is_ready", - AsyncMock(return_value=True), - ) - payload = { - "capabilityId": "chat-ai.capture", - "capabilityVersion": "1.0.0", - "outputSchemaVersion": "1", - "target": "doubao", - "prompt": "黑白调电竞椅值得买吗？", - } - payload[field] = value - channel = AsyncMock() - channel.collect.return_value = ChannelResult.ok([payload]) - - await run_acquisition_execution( - execution_id, session_factory=sessions, channel=channel - ) - - async with sessions() as db: - execution = await acquisition_service.get_execution(db, execution_id) - assert execution is not None - assert execution.status == AcquisitionExecutionStatus.FAILED - assert execution.failure["code"] == "invalid_capability_envelope" - - -@pytest.mark.asyncio -async def test_doubao_execution_fails_closed_without_authenticated_profile(db_engine): - from backend.acquisition.runner import run_acquisition_execution - - sessions = async_sessionmaker( - db_engine, class_=AsyncSession, expire_on_commit=False - ) - submission_data = _submission().model_dump() - submission_data.update( - { - "request_id": "doubao-request-2", - "idempotency_key": "doubao-attempt-2", - "capability": {"id": "chat-ai.capture", "version": "1.0.0"}, - "input": { - "target": "doubao", - "prompt": "黑白调电竞椅值得买吗？", - }, - } - ) - async with sessions() as db: - outcome = await acquisition_service.submit_execution( - db, AcquisitionSubmission.model_validate(submission_data) - ) - await acquisition_service.queue_execution(db, outcome.execution) - execution_id = outcome.execution.id - - endpoint = "http://anonymous-profile:9222" - pool = init_pool([endpoint], use_redis=False) - pool.set_profile_kind(endpoint, "anonymous") - channel = AsyncMock() - - await run_acquisition_execution( - execution_id, session_factory=sessions, channel=channel - ) - - channel.collect.assert_not_awaited() - async with sessions() as db: - execution = await acquisition_service.get_execution(db, execution_id) - assert execution is not None - assert execution.failure == { - "code": "no_authenticated_profile", - "message": "no_authenticated_profile", - } - - -@pytest.mark.asyncio -async def test_doubao_rechecks_the_selected_session_before_prompt_submission( - db_engine, - monkeypatch, -): - from backend.acquisition import capabilities - from backend.acquisition.runner import run_acquisition_execution - - sessions = async_sessionmaker( - db_engine, class_=AsyncSession, expire_on_commit=False - ) - submission_data = _submission().model_dump() - submission_data.update( - { - "request_id": "doubao-request-3", - "idempotency_key": "doubao-attempt-3", - "capability": {"id": "chat-ai.capture", "version": "1.0.0"}, - "input": {"target": "doubao", "prompt": "黑白调电竞椅值得买吗？"}, - } - ) - async with sessions() as db: - outcome = await acquisition_service.submit_execution( - db, AcquisitionSubmission.model_validate(submission_data) - ) - await acquisition_service.queue_execution(db, outcome.execution) - execution_id = outcome.execution.id - - endpoint = "http://expired-doubao-profile:9222" - pool = init_pool([endpoint], use_redis=False) - pool.set_profile_kind(endpoint, "authenticated") - session_probe = AsyncMock(return_value=False) - monkeypatch.setattr(capabilities, "_session_is_ready", session_probe) - channel = AsyncMock() - - await run_acquisition_execution( - execution_id, session_factory=sessions, channel=channel - ) - - session_probe.assert_awaited_once() - assert session_probe.await_args.args[2] == endpoint - channel.collect.assert_not_awaited() - async with sessions() as db: - execution = await acquisition_service.get_execution(db, execution_id) - assert execution is not None - assert execution.failure == { - "code": "session_not_qualified", - "message": "doubao session failed the execution-time readiness probe", - } - - @pytest.mark.asyncio async def test_worker_process_hydrates_anonymous_profile_before_dispatch( db_engine, diff --git a/tests/unit/test_doubao_acquisition_runner.py b/tests/unit/test_doubao_acquisition_runner.py new file mode 100644 index 0000000..f25204a --- /dev/null +++ b/tests/unit/test_doubao_acquisition_runner.py @@ -0,0 +1,232 @@ +from unittest.mock import AsyncMock + +import pytest +import sqlalchemy.ext.asyncio as _sqlalchemy_asyncio + +from backend.browser_pool import init_pool +from backend.channels.base import ChannelResult +from backend.schemas.acquisition import AcquisitionSubmission +from backend.services import acquisition_service + +AsyncSession = _sqlalchemy_asyncio.AsyncSession +async_sessionmaker = _sqlalchemy_asyncio.async_sessionmaker + + +def _doubao_submission(request_id: str, idempotency_key: str) -> AcquisitionSubmission: + return AcquisitionSubmission.model_validate( + { + "request_id": request_id, + "idempotency_key": idempotency_key, + "capability": {"id": "chat-ai.capture", "version": "1.0.0"}, + "output_schema_version": "1", + "input": { + "target": "doubao", + "prompt": "黑白调电竞椅值得买吗？", + }, + "environment": {"locale": "zh-CN", "region": "CN"}, + "required_artifacts": ["trace"], + "geo_refs": {"attempt_id": idempotency_key}, + } + ) + + +@pytest.mark.asyncio +async def test_doubao_execution_uses_authenticated_profile_and_frozen_prompt( + db_engine, + monkeypatch, +): + from backend.acquisition import capabilities + from backend.acquisition.runner import run_acquisition_execution + + sessions = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + submission = _doubao_submission("doubao-request-1", "doubao-attempt-1") + async with sessions() as db: + outcome = await acquisition_service.submit_execution(db, submission) + await acquisition_service.queue_execution(db, outcome.execution) + execution_id = outcome.execution.id + + endpoint = "http://doubao-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_mode(endpoint, "cdp") + pool.set_profile_kind(endpoint, "authenticated") + payload = { + "capabilityId": "chat-ai.capture", + "capabilityVersion": "1.0.0", + "outputSchemaVersion": "1", + "target": "doubao", + "prompt": "黑白调电竞椅值得买吗？", + "completionState": "complete", + "answer": {"text": "真实回答", "sha256": "a" * 64}, + "citations": [], + "displayedUrl": "https://www.doubao.com/chat/1", + "finalUrl": "https://www.doubao.com/chat/1", + "pageState": "answer", + "artifacts": [], + } + channel = AsyncMock() + session_probe = AsyncMock(return_value=True) + monkeypatch.setattr(capabilities, "_session_is_ready", session_probe) + + async def collect_while_leased(*_args, **_kwargs): + assert pool.available_for(endpoint) is False + return ChannelResult.ok( + [payload], + trace_artifact="artifact://trace/doubao-1", + trace_sha256="f" * 64, + ) + + channel.collect.side_effect = collect_while_leased + + await run_acquisition_execution( + execution_id, session_factory=sessions, channel=channel + ) + + channel.collect.assert_awaited_once_with( + {"site": "doubao", "command": "capture", "format": "json"}, + { + "prompt": "黑白调电竞椅值得买吗？", + "chrome_endpoint": endpoint, + "required_profile_kind": "authenticated", + "_endpoint_preacquired": True, + "trace": "on", + }, + ) + assert session_probe.await_args.args[2] == endpoint + async with sessions() as db: + execution = await acquisition_service.get_execution(db, execution_id) + assert execution is not None + assert execution.status.value == "succeeded" + assert execution.result_payload["payload"] == payload + assert execution.result_payload["operational"]["browser"] == { + "endpoint": endpoint, + "profile_kind": "authenticated", + } + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("field", "value"), [("target", "chatgpt"), ("prompt", "a different prompt")] +) +async def test_doubao_rejects_target_or_prompt_drift( + db_engine, + monkeypatch, + field, + value, +): + from backend.acquisition import capabilities + from backend.acquisition.runner import run_acquisition_execution + + sessions = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + submission = _doubao_submission( + f"doubao-drift-{field}", + f"doubao-drift-{field}", + ) + async with sessions() as db: + outcome = await acquisition_service.submit_execution(db, submission) + await acquisition_service.queue_execution(db, outcome.execution) + execution_id = outcome.execution.id + + endpoint = "http://doubao-drift-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_profile_kind(endpoint, "authenticated") + monkeypatch.setattr( + capabilities, + "_session_is_ready", + AsyncMock(return_value=True), + ) + payload = { + "capabilityId": "chat-ai.capture", + "capabilityVersion": "1.0.0", + "outputSchemaVersion": "1", + "target": "doubao", + "prompt": "黑白调电竞椅值得买吗？", + } + payload[field] = value + channel = AsyncMock() + channel.collect.return_value = ChannelResult.ok([payload]) + + await run_acquisition_execution( + execution_id, session_factory=sessions, channel=channel + ) + + async with sessions() as db: + execution = await acquisition_service.get_execution(db, execution_id) + assert execution is not None + assert execution.status.value == "failed" + assert execution.failure["code"] == "invalid_capability_envelope" + + +@pytest.mark.asyncio +async def test_doubao_execution_fails_closed_without_authenticated_profile(db_engine): + from backend.acquisition.runner import run_acquisition_execution + + sessions = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + submission = _doubao_submission("doubao-request-2", "doubao-attempt-2") + async with sessions() as db: + outcome = await acquisition_service.submit_execution(db, submission) + await acquisition_service.queue_execution(db, outcome.execution) + execution_id = outcome.execution.id + + endpoint = "http://anonymous-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_profile_kind(endpoint, "anonymous") + channel = AsyncMock() + + await run_acquisition_execution( + execution_id, session_factory=sessions, channel=channel + ) + + channel.collect.assert_not_awaited() + async with sessions() as db: + execution = await acquisition_service.get_execution(db, execution_id) + assert execution is not None + assert execution.failure == { + "code": "no_authenticated_profile", + "message": "no_authenticated_profile", + } + + +@pytest.mark.asyncio +async def test_doubao_rechecks_the_selected_session_before_prompt_submission( + db_engine, + monkeypatch, +): + from backend.acquisition import capabilities + from backend.acquisition.runner import run_acquisition_execution + + sessions = async_sessionmaker( + db_engine, class_=AsyncSession, expire_on_commit=False + ) + submission = _doubao_submission("doubao-request-3", "doubao-attempt-3") + async with sessions() as db: + outcome = await acquisition_service.submit_execution(db, submission) + await acquisition_service.queue_execution(db, outcome.execution) + execution_id = outcome.execution.id + + endpoint = "http://expired-doubao-profile:9222" + pool = init_pool([endpoint], use_redis=False) + pool.set_profile_kind(endpoint, "authenticated") + session_probe = AsyncMock(return_value=False) + monkeypatch.setattr(capabilities, "_session_is_ready", session_probe) + channel = AsyncMock() + + await run_acquisition_execution( + execution_id, session_factory=sessions, channel=channel + ) + + session_probe.assert_awaited_once() + assert session_probe.await_args.args[2] == endpoint + channel.collect.assert_not_awaited() + async with sessions() as db: + execution = await acquisition_service.get_execution(db, execution_id) + assert execution is not None + assert execution.failure == { + "code": "session_not_qualified", + "message": "doubao session failed the execution-time readiness probe", + } From aa774da1994f9fb66dc281bbf0061d4f4ceee260 Mon Sep 17 00:00:00 2001 From: lunnt <2276214182@qq.com> Date: Thu, 30 Jul 2026 00:17:51 +0800 Subject: [PATCH 3/7] fix(acquisition): align managed Doubao runtime --- Dockerfile | 4 +- agent/Dockerfile | 4 +- backend/acquisition/capabilities.py | 26 +++++++++--- backend/acquisition/registry.py | 4 +- backend/agent_server.py | 12 +++++- backend/channels/opencli_channel.py | 2 + scripts/install-agent.sh | 4 +- scripts/install-managed-opencli.ps1 | 4 +- scripts/verify_managed_opencli_runtime.py | 4 +- tests/unit/test_acquisition_capabilities.py | 6 ++- .../test_agent_image_runtime_packaging.py | 4 +- tests/unit/test_agent_server.py | 41 +++++++++++++++++++ tests/unit/test_managed_opencli_verifier.py | 2 +- 13 files changed, 94 insertions(+), 23 deletions(-) diff --git a/Dockerfile b/Dockerfile index 687a17e..de5bdab 100644 --- a/Dockerfile +++ b/Dockerfile @@ -45,9 +45,9 @@ RUN npm install -g @jackwener/opencli@${OPENCLI_VERSION} \ && rm -rf /root/.npm ARG OHMYOPENCLI_REPO=https://github.com/2233admin/OhMyOpenCLI.git -ARG OHMYOPENCLI_COMMIT=bfe1c25b4b12661058dd6e9980c562a09f230cc7 +ARG OHMYOPENCLI_COMMIT=b0fdd513f64899b068103ddd7ff0de957d778b5c ARG OFFICIAL_SITE_CAPABILITY_COMMIT=73cc60c83586ef2c95469b3b70d6cfc80fa5bc53 -ARG DOUBAO_CAPABILITY_COMMIT=bfe1c25b4b12661058dd6e9980c562a09f230cc7 +ARG DOUBAO_CAPABILITY_COMMIT=b0fdd513f64899b068103ddd7ff0de957d778b5c RUN git clone ${OHMYOPENCLI_REPO} /opt/ohmyopencli \ && cd /opt/ohmyopencli \ && git checkout --detach ${OHMYOPENCLI_COMMIT} \ diff --git a/agent/Dockerfile b/agent/Dockerfile index 1781584..33fea48 100644 --- a/agent/Dockerfile +++ b/agent/Dockerfile @@ -36,9 +36,9 @@ RUN npm install -g @jackwener/opencli@${OPENCLI_VERSION} \ # identities separate: the latter is the behavior change, while the former is # the exact checkout certified by this image. ARG OHMYOPENCLI_REPO=https://github.com/2233admin/OhMyOpenCLI.git -ARG OHMYOPENCLI_COMMIT=bfe1c25b4b12661058dd6e9980c562a09f230cc7 +ARG OHMYOPENCLI_COMMIT=b0fdd513f64899b068103ddd7ff0de957d778b5c ARG OFFICIAL_SITE_CAPABILITY_COMMIT=73cc60c83586ef2c95469b3b70d6cfc80fa5bc53 -ARG DOUBAO_CAPABILITY_COMMIT=bfe1c25b4b12661058dd6e9980c562a09f230cc7 +ARG DOUBAO_CAPABILITY_COMMIT=b0fdd513f64899b068103ddd7ff0de957d778b5c RUN git clone ${OHMYOPENCLI_REPO} /opt/ohmyopencli \ && cd /opt/ohmyopencli \ && git checkout --detach ${OHMYOPENCLI_COMMIT} \ diff --git a/backend/acquisition/capabilities.py b/backend/acquisition/capabilities.py index 46b904c..1f25b55 100644 --- a/backend/acquisition/capabilities.py +++ b/backend/acquisition/capabilities.py @@ -19,6 +19,18 @@ COMMAND_TIMEOUT_SECONDS = 15.0 +def _opencli_environment(*, cdp_endpoint: str | None = None) -> dict[str, str]: + """Build one unambiguous OpenCLI browser-routing environment.""" + env = os.environ.copy() + env.pop("OPENCLI_DAEMON_HOST", None) + env.pop("OPENCLI_DAEMON_PORT", None) + if cdp_endpoint is None: + env.pop("OPENCLI_CDP_ENDPOINT", None) + else: + env["OPENCLI_CDP_ENDPOINT"] = cdp_endpoint + return env + + async def _command(*args: str, env: dict[str, str] | None = None) -> tuple[int, str]: try: process = await asyncio.create_subprocess_exec( @@ -76,7 +88,11 @@ async def _runtime_is_installed() -> bool: return False opencli_bin = resolve_opencli_bin() - version_rc, version_output = await _command(opencli_bin, "--version") + version_rc, version_output = await _command( + opencli_bin, + "--version", + env=_opencli_environment(), + ) versions = re.findall(r"\d+\.\d+\.\d+", version_output) if version_rc != 0 or OPENCLI_VERSION not in versions: return False @@ -89,13 +105,14 @@ async def _registration_is_available( ) -> bool: opencli_bin = resolve_opencli_bin() command_rc, command_output = await _command( - opencli_bin, *registration.probe_args + opencli_bin, + *registration.probe_args, + env=_opencli_environment(), ) if command_rc != 0 or registration.help_marker not in command_output: return False - patch_env = os.environ.copy() - patch_env["OPENCLI_CDP_ENDPOINT"] = "http://127.0.0.1:9" + patch_env = _opencli_environment(cdp_endpoint="http://127.0.0.1:9") patch_rc, patch_output = await _command( opencli_bin, *registration.route_probe_args, @@ -197,7 +214,6 @@ async def _session_is_ready( and payload.get("unattendedReady") is True and payload.get("loginDetected") is False and payload.get("promptInputDetected") is True - and payload.get("sendButtonDetected") is True and ( registration.session_expected_host is None or urlparse(str(payload.get("url", ""))).hostname diff --git a/backend/acquisition/registry.py b/backend/acquisition/registry.py index 34a0fe3..d8f32da 100644 --- a/backend/acquisition/registry.py +++ b/backend/acquisition/registry.py @@ -2,9 +2,9 @@ from dataclasses import dataclass -OHMYOPENCLI_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" +OHMYOPENCLI_COMMIT = "b0fdd513f64899b068103ddd7ff0de957d778b5c" OFFICIAL_SITE_CAPABILITY_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" -DOUBAO_CAPABILITY_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" +DOUBAO_CAPABILITY_COMMIT = "b0fdd513f64899b068103ddd7ff0de957d778b5c" OPENCLI_VERSION = "1.8.5" diff --git a/backend/agent_server.py b/backend/agent_server.py index ae0c443..a9a25c7 100644 --- a/backend/agent_server.py +++ b/backend/agent_server.py @@ -194,10 +194,18 @@ async def _runtime_lineage( command: str, ) -> dict[str, str]: """Measure the binaries/source used by this node; never echo declarations.""" + lineage_env = os.environ.copy() + lineage_env.pop("OPENCLI_DAEMON_HOST", None) + lineage_env.pop("OPENCLI_DAEMON_PORT", None) + async def output(*argv: str, cwd: str | None = None) -> str: try: proc = await asyncio.create_subprocess_exec( - *argv, stdout=asyncio.subprocess.PIPE, stderr=asyncio.subprocess.PIPE, cwd=cwd + *argv, + stdout=asyncio.subprocess.PIPE, + stderr=asyncio.subprocess.PIPE, + cwd=cwd, + env=lineage_env, ) stdout, _ = await proc.communicate() return stdout.decode(errors="replace").strip() if proc.returncode == 0 else "" @@ -671,6 +679,8 @@ async def collect(req: CollectRequest) -> dict: # running in Docker without bundled Chrome. if _AGENT_DEPLOY_TYPE == "docker" and not _AGENT_HAS_CHROME: cdp_ep = re.sub(r"(localhost|127\.0\.0\.1)", "host.docker.internal", cdp_ep) + env.pop("OPENCLI_DAEMON_HOST", None) + env.pop("OPENCLI_DAEMON_PORT", None) env["OPENCLI_CDP_ENDPOINT"] = cdp_ep logger.info("cdp | cmd=%s cdp=%s", " ".join(cmd), cdp_ep) diff --git a/backend/channels/opencli_channel.py b/backend/channels/opencli_channel.py index 9853ab6..d7b5d03 100644 --- a/backend/channels/opencli_channel.py +++ b/backend/channels/opencli_channel.py @@ -753,6 +753,8 @@ async def collect( bridge_err, ) else: + env.pop("OPENCLI_DAEMON_HOST", None) + env.pop("OPENCLI_DAEMON_PORT", None) env["OPENCLI_CDP_ENDPOINT"] = cdp_endpoint logger.info("opencli cdp | cmd=%s cdp=%s", " ".join(cmd), cdp_endpoint) diff --git a/scripts/install-agent.sh b/scripts/install-agent.sh index 8154644..e108380 100755 --- a/scripts/install-agent.sh +++ b/scripts/install-agent.sh @@ -333,9 +333,9 @@ install_python() { # Install the exact project-owned managed-acquisition capability package. OHMYOPENCLI_ROOT="$AGENT_DIR/ohmyopencli" - OHMYOPENCLI_COMMIT="bfe1c25b4b12661058dd6e9980c562a09f230cc7" + OHMYOPENCLI_COMMIT="b0fdd513f64899b068103ddd7ff0de957d778b5c" OFFICIAL_SITE_CAPABILITY_COMMIT="73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" - DOUBAO_CAPABILITY_COMMIT="bfe1c25b4b12661058dd6e9980c562a09f230cc7" + DOUBAO_CAPABILITY_COMMIT="b0fdd513f64899b068103ddd7ff0de957d778b5c" command -v git >/dev/null 2>&1 || die "git is required to install OhMyOpenCLI" [[ -e "$OHMYOPENCLI_ROOT" ]] && die \ "Managed OhMyOpenCLI target already exists; archive it explicitly before reinstalling: $OHMYOPENCLI_ROOT" diff --git a/scripts/install-managed-opencli.ps1 b/scripts/install-managed-opencli.ps1 index 767238a..0f77c93 100644 --- a/scripts/install-managed-opencli.ps1 +++ b/scripts/install-managed-opencli.ps1 @@ -8,10 +8,10 @@ param( $ErrorActionPreference = "Stop" $OpenCliVersion = "1.8.5" -$OhMyOpenCliCommit = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" +$OhMyOpenCliCommit = "b0fdd513f64899b068103ddd7ff0de957d778b5c" $CapabilitySourceCommits = @( "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53", - "bfe1c25b4b12661058dd6e9980c562a09f230cc7" + "b0fdd513f64899b068103ddd7ff0de957d778b5c" ) $requestHeaders = @{} if ($ApiAuthToken) { diff --git a/scripts/verify_managed_opencli_runtime.py b/scripts/verify_managed_opencli_runtime.py index 427c8bd..ba06926 100644 --- a/scripts/verify_managed_opencli_runtime.py +++ b/scripts/verify_managed_opencli_runtime.py @@ -12,9 +12,9 @@ from pathlib import Path from typing import Any -OHMYOPENCLI_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" +OHMYOPENCLI_COMMIT = "b0fdd513f64899b068103ddd7ff0de957d778b5c" OFFICIAL_SITE_CAPABILITY_COMMIT = "73cc60c83586ef2c95469b3b70d6cfc80fa5bc53" -DOUBAO_CAPABILITY_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" +DOUBAO_CAPABILITY_COMMIT = "b0fdd513f64899b068103ddd7ff0de957d778b5c" OPENCLI_VERSION = "1.8.5" diff --git a/tests/unit/test_acquisition_capabilities.py b/tests/unit/test_acquisition_capabilities.py index 7710fd4..21363ff 100644 --- a/tests/unit/test_acquisition_capabilities.py +++ b/tests/unit/test_acquisition_capabilities.py @@ -151,7 +151,7 @@ async def test_catalog_publishes_doubao_only_after_authenticated_session_probe( ("protocol", "collector_name"), [("http", "_collect_via_agent"), ("ws", "_collect_via_ws_agent")], ) -async def test_agent_mode_probes_the_selected_remote_browser_route( +async def test_agent_mode_accepts_ready_session_before_send_button_is_rendered( monkeypatch, protocol, collector_name, @@ -177,7 +177,7 @@ async def test_agent_mode_probes_the_selected_remote_browser_route( "unattendedReady": True, "loginDetected": False, "promptInputDetected": True, - "sendButtonDetected": True, + "sendButtonDetected": False, "url": "https://www.doubao.com/chat/", } ] @@ -216,6 +216,7 @@ async def test_runtime_probe_uses_the_configured_opencli_binary(monkeypatch): configured_bin = r"C:\managed\opencli.cmd" monkeypatch.setenv("OPENCLI_BIN", configured_bin) + monkeypatch.setenv("OPENCLI_DAEMON_PORT", "19825") command = _runtime_command(capabilities) monkeypatch.setattr(capabilities, "_command", command) @@ -224,6 +225,7 @@ async def test_runtime_probe_uses_the_configured_opencli_binary(monkeypatch): call for call in command.await_args_list if call.args[-1] == "--version" ) assert version_call.args == (configured_bin, "--version") + assert "OPENCLI_DAEMON_PORT" not in version_call.kwargs["env"] registration = capabilities.list_capability_registrations()[0] assert await capabilities._registration_is_available(registration) is True help_call = next( diff --git a/tests/unit/test_agent_image_runtime_packaging.py b/tests/unit/test_agent_image_runtime_packaging.py index a6a7963..212ce05 100644 --- a/tests/unit/test_agent_image_runtime_packaging.py +++ b/tests/unit/test_agent_image_runtime_packaging.py @@ -17,11 +17,11 @@ def test_agent_image_pins_managed_acquisition_runtime(): assert "ARG OPENCLI_VERSION=1.8.5" in dockerfile assert ( "ARG OHMYOPENCLI_COMMIT=" - "bfe1c25b4b12661058dd6e9980c562a09f230cc7" + "b0fdd513f64899b068103ddd7ff0de957d778b5c" ) in dockerfile assert ( "ARG DOUBAO_CAPABILITY_COMMIT=" - "bfe1c25b4b12661058dd6e9980c562a09f230cc7" + "b0fdd513f64899b068103ddd7ff0de957d778b5c" ) in dockerfile assert "git checkout --detach ${OHMYOPENCLI_COMMIT}" in dockerfile assert "npm ci" in dockerfile diff --git a/tests/unit/test_agent_server.py b/tests/unit/test_agent_server.py index a8a489a..1cb7998 100644 --- a/tests/unit/test_agent_server.py +++ b/tests/unit/test_agent_server.py @@ -7,6 +7,7 @@ """ import json +from unittest.mock import AsyncMock import pytest from fastapi import HTTPException @@ -64,6 +65,8 @@ async def create(*args, **kwargs): return Process() monkeypatch.setattr(agent_server.asyncio, "create_subprocess_exec", create) + monkeypatch.setenv("OPENCLI_DAEMON_HOST", "legacy-bridge") + monkeypatch.setenv("OPENCLI_DAEMON_PORT", "19825") lineage = await agent_server._runtime_lineage( "opencli", "doubao", "capture" @@ -75,6 +78,44 @@ async def create(*args, **kwargs): "opencli_version": "1.8.5", } assert calls[1][0][-2:] == ("--", "adapters/doubao/capture.js") + assert all("OPENCLI_DAEMON_HOST" not in call[1]["env"] for call in calls) + assert all("OPENCLI_DAEMON_PORT" not in call[1]["env"] for call in calls) + + +@pytest.mark.asyncio +async def test_cdp_collect_removes_inherited_bridge_environment(monkeypatch): + captured_env = {} + + class Process: + returncode = 0 + + async def communicate(self): + return b"[]", b"" + + async def create(*_args, **kwargs): + captured_env.update(kwargs["env"]) + return Process() + + monkeypatch.setenv("OPENCLI_DAEMON_HOST", "legacy-bridge") + monkeypatch.setenv("OPENCLI_DAEMON_PORT", "19825") + monkeypatch.setattr(agent_server.asyncio, "create_subprocess_exec", create) + monkeypatch.setattr(agent_server, "_snapshot_tab_ids", AsyncMock(return_value=set())) + monkeypatch.setattr(agent_server, "_cleanup_cdp_tabs", AsyncMock()) + monkeypatch.setattr(agent_server, "_runtime_lineage", AsyncMock(return_value={})) + + result = await agent_server.collect( + agent_server.CollectRequest( + site="doubao", + command="session-probe", + mode="cdp", + cdp_endpoint="http://chrome:9222", + ) + ) + + assert result["success"] is True + assert captured_env["OPENCLI_CDP_ENDPOINT"] == "http://chrome:9222" + assert "OPENCLI_DAEMON_HOST" not in captured_env + assert "OPENCLI_DAEMON_PORT" not in captured_env # ── _auth_headers ──────────────────────────────────────────────────────────── diff --git a/tests/unit/test_managed_opencli_verifier.py b/tests/unit/test_managed_opencli_verifier.py index fec3cfc..1e78b6d 100644 --- a/tests/unit/test_managed_opencli_verifier.py +++ b/tests/unit/test_managed_opencli_verifier.py @@ -4,7 +4,7 @@ from scripts.verify_managed_opencli_runtime import VerificationError, verify_runtime -PINNED_COMMIT = "bfe1c25b4b12661058dd6e9980c562a09f230cc7" +PINNED_COMMIT = "b0fdd513f64899b068103ddd7ff0de957d778b5c" def _completed(args, returncode=0, stdout="", stderr=""): From 79a0fa3634bbb62595105265166b2cadae39e0d8 Mon Sep 17 00:00:00 2001 From: lunnt <2276214182@qq.com> Date: Thu, 30 Jul 2026 00:23:36 +0800 Subject: [PATCH 4/7] fix(ci): give frontend jobs unique identifiers --- .github/workflows/ci.yml | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 1aaea72..6246938 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -10,7 +10,7 @@ on: workflow_dispatch: jobs: - frontend: + frontend-nextjs: runs-on: ubuntu-latest name: Frontend (Next.js) defaults: From b001ce7d781e9a04ddff35bacf76982e86b07b0c Mon Sep 17 00:00:00 2001 From: lunnt <2276214182@qq.com> Date: Thu, 30 Jul 2026 00:27:40 +0800 Subject: [PATCH 5/7] fix(ci): align frontend runtime dependencies --- .github/workflows/ci.yml | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 6246938..d7df522 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -28,7 +28,7 @@ jobs: - name: Setup Node uses: actions/setup-node@v4 with: - node-version: "20" + node-version: "22" cache: pnpm cache-dependency-path: frontend/pnpm-lock.yaml @@ -100,7 +100,7 @@ jobs: python-version: "3.13" - name: Install workflow contract dependency - run: python -m pip install "pydantic>=2.10.0" + run: python -m pip install "pydantic>=2.10.0" "httpx>=0.28.0" - name: Install dependencies run: | From 8c0dd229f8a90229170c6098f7733c89e5c2fb2e Mon Sep 17 00:00:00 2001 From: lunnt <2276214182@qq.com> Date: Thu, 30 Jul 2026 00:30:06 +0800 Subject: [PATCH 6/7] fix(ci): install workflow compiler dependencies --- .github/workflows/ci.yml | 6 +++++- 1 file changed, 5 insertions(+), 1 deletion(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index d7df522..2172ae1 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -100,7 +100,11 @@ jobs: python-version: "3.13" - name: Install workflow contract dependency - run: python -m pip install "pydantic>=2.10.0" "httpx>=0.28.0" + run: >- + python -m pip install + "pydantic>=2.10.0" + "httpx>=0.28.0" + "sqlalchemy>=2.0.0" - name: Install dependencies run: | From fe6a36fa00a3c0874817ebe755c7f4505855b5e2 Mon Sep 17 00:00:00 2001 From: lunnt <2276214182@qq.com> Date: Thu, 30 Jul 2026 00:32:28 +0800 Subject: [PATCH 7/7] fix(ci): install declared compiler runtime --- .github/workflows/ci.yml | 8 ++------ 1 file changed, 2 insertions(+), 6 deletions(-) diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 2172ae1..2e640e3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -99,12 +99,8 @@ jobs: with: python-version: "3.13" - - name: Install workflow contract dependency - run: >- - python -m pip install - "pydantic>=2.10.0" - "httpx>=0.28.0" - "sqlalchemy>=2.0.0" + - name: Install workflow compiler dependencies + run: python -m pip install -e .. - name: Install dependencies run: |