From 39f67df3baaa5bd1c9036a351a2b3648a02d1ae0 Mon Sep 17 00:00:00 2001 From: evanlowe <62918515+evanlowe@users.noreply.github.com> Date: Mon, 21 Sep 2026 11:55:19 +0800 Subject: [PATCH 1/3] feat(studio): store evaluation datasets and feedback in TOS --- frontend/README.md | 33 + frontend/server/evaluation/__init__.py | 19 + frontend/server/evaluation/automatic.py | 96 ++ frontend/server/evaluation/models.py | 69 ++ frontend/server/evaluation/repository.py | 355 ++++++ frontend/server/evaluation/routes.py | 463 +++++++ frontend/server/evaluation/sessions.py | 52 + .../server/evaluation_automation/__init__.py | 32 +- .../server/evaluation_automation/datasets.py | 46 +- .../evaluation_automation/model_gateway.py | 6 +- .../evaluation_automation/repository.py | 345 +----- frontend/server/runtime_readiness.py | 57 + tests/cli/conftest.py | 10 + .../cli/test_frontend_evaluation_feedback.py | 1090 +++-------------- tests/cli/test_studio_rbac.py | 119 +- tests/cli/test_studio_startup_imports.py | 29 + .../test_tos_evaluation_repository.py | 337 +++++ .../evaluation_automation/test_datasets.py | 46 +- .../test_model_gateway.py | 56 + .../evaluation_automation/test_repository.py | 191 --- .../frontend/server/test_runtime_readiness.py | 92 ++ veadk/cli/cli_frontend.py | 708 ++--------- veadk/cli/studio_start.py | 8 +- 23 files changed, 2036 insertions(+), 2223 deletions(-) create mode 100644 frontend/server/evaluation/__init__.py create mode 100644 frontend/server/evaluation/automatic.py create mode 100644 frontend/server/evaluation/models.py create mode 100644 frontend/server/evaluation/repository.py create mode 100644 frontend/server/evaluation/routes.py create mode 100644 frontend/server/evaluation/sessions.py create mode 100644 frontend/server/runtime_readiness.py create mode 100644 tests/frontend/server/evaluation/test_tos_evaluation_repository.py create mode 100644 tests/frontend/server/evaluation_automation/test_model_gateway.py create mode 100644 tests/frontend/server/test_runtime_readiness.py diff --git a/frontend/README.md b/frontend/README.md index d669f9fd7..fde76852a 100644 --- a/frontend/README.md +++ b/frontend/README.md @@ -1565,3 +1565,36 @@ Tokens 使用带展开提示的轻量文字按钮，可通过悬浮、键盘聚 `usage` 事件仍保留兼容；已有数据库只增加 `run_turns.metrics` 列。回滚到旧代码时应 使用新的短期数据库路径，因为旧版本按固定列数写入 `run_turns`。当前单实例云部署 在实例替换后丢失短期记录的约定不变。 + +### Runtime evaluation storage + +Studio stores evaluation sets and samples in the configured private TOS bucket +(`VEADK_STUDIO_TOS_BUCKET` and `VEADK_STUDIO_TOS_REGION`). Volcengine and BytePlus +use their respective server credentials. No AgentKit evaluation APIs are used. + +Objects live under `veadk-studio/v1/evaluation//`: + +- `sets/good.json` and `sets/bad.json` are the two default sets +- `sets/.json` stores a user-created set and its editable name and description +- `samples/.json` stores an individual sample, its set ID and `user` or `auto` source + +App names, projects and cloud regions are not part of the storage identity. +Use separate buckets for local and production installations. Existing AgentKit +sets are not imported. + +New Runtime deployments apply the requested instance limits at creation and +wait for the configured instances to become ready before reporting success. +This avoids starting the first conversation on a temporary startup instance. +Runtime sessions still use the Agent's configured short-term memory backend; +use database-backed memory when sessions must survive instance replacement. + +The existing evaluation page and conversation feedback controls use the same +TOS records. Runtime-scoped APIs support custom set CRUD, sample editing and +moving, source filtering, search and pagination. Updates use TOS ETag conditions; +a stale edit returns a conflict. Deletion removes +sample content and retains a small marker so automatic retries cannot recreate +it. Deleting a default set stops feedback writes to that set and prevents background recreation; renaming it keeps feedback routing intact. + +List filtering and counts currently scan only the selected Runtime's objects, +with bounded parallel reads. This is intended for the initial dataset sizes; +large collections will need a separate rebuildable query index. diff --git a/frontend/server/evaluation/__init__.py b/frontend/server/evaluation/__init__.py new file mode 100644 index 000000000..85716b1c5 --- /dev/null +++ b/frontend/server/evaluation/__init__.py @@ -0,0 +1,19 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Studio-owned evaluation storage and management""" + +from .repository import EvaluationStorage + +__all__ = ["EvaluationStorage"] diff --git a/frontend/server/evaluation/automatic.py b/frontend/server/evaluation/automatic.py new file mode 100644 index 000000000..8ec3873ec --- /dev/null +++ b/frontend/server/evaluation/automatic.py @@ -0,0 +1,96 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Adapt automatic evaluation output to the shared Runtime sample store""" + +from __future__ import annotations + +from typing import TYPE_CHECKING + +if TYPE_CHECKING: + from frontend.server.evaluation_automation.models import AutoEvaluationCase + +from .models import Sample +from .repository import TosEvaluationRepository, sample_id + + +class TosAutomaticCases: + def __init__(self, repository: TosEvaluationRepository) -> None: + self.repository = repository + + async def upsert(self, case: AutoEvaluationCase) -> AutoEvaluationCase: + identifier = sample_id( + "auto", + case.user_id, + case.session_id, + case.message_id, + case.evaluator_version, + ) + saved = await self.repository.save_automatic( + Sample( + id=identifier, + evaluationSetId=case.kind, + source="auto", + kind=case.kind, + input=case.input, + output=case.output, + referenceOutput=case.reference_output, + comment=case.comment, + userId=case.user_id, + sessionId=case.session_id, + messageId=case.message_id, + invocationId=case.invocation_id, + score=case.score, + reason=case.reason, + evaluatorVersion=case.evaluator_version, + ) + ) + return case.model_copy( + update={ + "id": identifier, + "item_key": identifier, + "evaluation_set_id": saved.evaluation_set_id if saved else case.kind, + } + ) + + async def list_cases( + self, *, agent_name: str, page_size: int = 100 + ) -> list[AutoEvaluationCase]: + from frontend.server.evaluation_automation.models import AutoEvaluationCase + + rows = await self.repository.list_samples() + return [ + AutoEvaluationCase( + id=row.id, + itemKey=row.id, + kind=row.kind, + input=row.input, + output=row.output, + referenceOutput=row.reference_output, + comment=row.comment, + agentName=agent_name, + sessionId=row.session_id, + messageId=row.message_id, + runtimeId=self.repository.runtime_id, + invocationId=row.invocation_id, + userId=row.user_id, + createdAt=row.created_at, + evaluationSetId=row.evaluation_set_id, + score=row.score, + reason=row.reason, + evaluatorVersion=row.evaluator_version, + ) + for row in rows + if row.source == "auto" and row.kind is not None and row.score is not None + ][:page_size] diff --git a/frontend/server/evaluation/models.py b/frontend/server/evaluation/models.py new file mode 100644 index 000000000..17a5ba56e --- /dev/null +++ b/frontend/server/evaluation/models.py @@ -0,0 +1,69 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Runtime-owned evaluation collections and samples""" + +from datetime import datetime, timezone +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field + + +def now() -> str: + return datetime.now(timezone.utc).isoformat() + + +class Record(BaseModel): + model_config = ConfigDict(extra="forbid", populate_by_name=True) + schema_version: Literal[1] = Field(default=1, alias="schemaVersion") + id: str + created_at: str = Field(default_factory=now, alias="createdAt") + updated_at: str = Field(default_factory=now, alias="updatedAt") + deleted_at: str | None = Field(default=None, alias="deletedAt") + revision: str = Field(default="", exclude=True) + + +class EvaluationSet(Record): + name: str = Field(min_length=1, max_length=100) + description: str = Field(default="", max_length=4000) + default_kind: Literal["good", "bad"] | None = Field( + default=None, alias="defaultKind" + ) + created_by: str = Field(default="", alias="createdBy") + + +class Sample(Record): + evaluation_set_id: str = Field(alias="evaluationSetId", min_length=1) + source: Literal["user", "auto"] + kind: Literal["good", "bad"] | None = None + input: str = Field(max_length=200_000) + output: str = Field(max_length=200_000) + reference_output: str = Field( + default="", alias="referenceOutput", max_length=200_000 + ) + comment: str = Field(default="", max_length=20_000) + user_id: str = Field(default="", alias="userId") + session_id: str = Field(default="", alias="sessionId") + message_id: str = Field(default="", alias="messageId") + invocation_id: str = Field(default="", alias="invocationId") + score: float | None = Field(default=None, ge=0, le=1) + reason: str = Field(default="", max_length=20_000) + evaluator_version: str = Field(default="", alias="evaluatorVersion") + + +def public_record(record: Record) -> dict: + return { + **record.model_dump(mode="json", by_alias=True), + "revision": record.revision, + } diff --git a/frontend/server/evaluation/repository.py b/frontend/server/evaluation/repository.py new file mode 100644 index 000000000..c109b702d --- /dev/null +++ b/frontend/server/evaluation/repository.py @@ -0,0 +1,355 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""TOS is the source of truth for Runtime evaluation data""" + +from __future__ import annotations + +import asyncio +import hashlib +import json +import re +from collections.abc import Callable +from concurrent.futures import ThreadPoolExecutor +from typing import Any +from urllib.parse import quote +from uuid import uuid4 + +from frontend.server.storage import STUDIO_STORAGE_ROOT_PREFIX, StudioStorageConfig +from frontend.server.storage.tos import create_cached_tos_client_factory + +from .models import EvaluationSet, Record, Sample, now + +MAX_OBJECT_BYTES = 1024 * 1024 + + +class EvaluationError(RuntimeError): + def __init__(self, message: str, status: int = 502) -> None: + super().__init__(message) + self.status = status + + +def sample_id( + source: str, user_id: str, session_id: str, message_id: str, version: str = "" +) -> str: + return hashlib.sha256( + json.dumps([source, user_id, session_id, message_id, version]).encode() + ).hexdigest() + + +class EvaluationStorage: + def __init__(self, provider: Any, resolve_credentials: Callable) -> None: + self.config = StudioStorageConfig.from_env(provider) + self.factory = ( + create_cached_tos_client_factory(self.config, resolve_credentials) + if self.config.configured + else None + ) + + def for_runtime(self, runtime_id: str) -> TosEvaluationRepository: + if self.factory is None: + raise EvaluationError(self.config.unavailable_reason, 503) + return TosEvaluationRepository(self.config.bucket, self.factory, runtime_id) + + +class TosEvaluationRepository: + def __init__( + self, bucket: str, client_factory: Callable[[], Any], runtime_id: str + ) -> None: + if not runtime_id or runtime_id in {".", ".."}: + raise ValueError("Runtime ID is required") + self.bucket = bucket + self.factory = client_factory + self.runtime_id = runtime_id + self.prefix = ( + f"{STUDIO_STORAGE_ROOT_PREFIX}/evaluation/{quote(runtime_id, safe='')}/" + ) + + def _key(self, folder: str, identifier: str) -> str: + if not re.fullmatch(r"[a-zA-Z0-9_-]{1,128}", identifier): + raise EvaluationError("记录 ID 无效", 400) + return f"{self.prefix}{folder}/{identifier}.json" + + def _read(self, folder: str, identifier: str) -> tuple[dict, str] | None: + try: + response = self.factory().get_object( + bucket=self.bucket, key=self._key(folder, identifier) + ) + data = response.read(MAX_OBJECT_BYTES + 1) + except Exception as error: + if getattr(error, "status_code", None) == 404: + return None + if isinstance(error, EvaluationError): + raise + raise EvaluationError( + "读取评测存储失败，请检查 TOS 配置和访问权限" + ) from error + if len(data) > MAX_OBJECT_BYTES: + raise EvaluationError("评测记录超过大小限制") + try: + content = json.loads(data) + except (ValueError, UnicodeDecodeError) as error: + raise EvaluationError("评测记录不是有效的 JSON 数据") from error + if not isinstance(content, dict) or content.get("id") != identifier: + raise EvaluationError("评测记录格式无效") + etag = str(response.etag) + if not etag: + raise EvaluationError("评测记录缺少版本信息") + return content, etag + + def _write( + self, folder: str, identifier: str, content: dict, revision: str = "" + ) -> str: + data = json.dumps(content, ensure_ascii=False).encode() + if len(data) > MAX_OBJECT_BYTES: + raise EvaluationError("单条评测数据不能超过 1 MB", 400) + options = {"if_match": revision} if revision else {"forbid_overwrite": True} + try: + response = self.factory().put_object( + bucket=self.bucket, + key=self._key(folder, identifier), + content=data, + content_type="application/json", + **options, + ) + return str(response.etag) + except Exception as error: + if getattr(error, "status_code", None) in {409, 412}: + raise EvaluationError("数据已被修改，请刷新后重试", 409) from error + if isinstance(error, EvaluationError): + raise + raise EvaluationError( + "保存评测存储失败，请检查 TOS 配置和访问权限" + ) from error + + def _save(self, folder: str, record: Record, revision: str = "") -> Any: + updated = record.model_copy(update={"updated_at": now()}) + etag = self._write( + folder, updated.id, updated.model_dump(mode="json", by_alias=True), revision + ) + return updated.model_copy(update={"revision": etag}) + + def _get_set(self, identifier: str, include_deleted: bool = False) -> EvaluationSet: + result = self._read("sets", identifier) + if result is None or (result[0].get("deletedAt") and not include_deleted): + raise EvaluationError("评测集不存在或已删除", 404) + return EvaluationSet.model_validate({**result[0], "revision": result[1]}) + + def _scan(self, folder: str) -> list[tuple[dict, str]]: + keys: list[str] = [] + token = "" + seen: set[str] = set() + while True: + try: + response = self.factory().list_objects_type2( + bucket=self.bucket, + prefix=f"{self.prefix}{folder}/", + continuation_token=token, + max_keys=1000, + ) + except Exception as error: + raise EvaluationError( + "读取评测列表失败，请检查 TOS 配置和访问权限" + ) from error + keys.extend( + item.key.rsplit("/", 1)[-1][:-5] + for item in response.contents + if item.key.endswith(".json") + ) + if not response.is_truncated: + break + token = response.next_continuation_token + if not token or token in seen: + raise EvaluationError("TOS 列表分页响应无效") + seen.add(token) + with ThreadPoolExecutor(max_workers=8) as pool: + return [ + item + for item in pool.map(lambda key: self._read(folder, key), keys) + if item is not None + ] + + async def ensure_defaults(self) -> list[EvaluationSet]: + return await asyncio.to_thread(self._ensure_defaults) + + def _ensure_defaults(self) -> list[EvaluationSet]: + return [self._ensure_default(kind) for kind in ("good", "bad")] + + def _ensure_default(self, kind: str) -> EvaluationSet: + record = EvaluationSet( + id=kind, + name="Good Case" if kind == "good" else "Bad Case", + defaultKind="good" if kind == "good" else "bad", + ) + try: + return self._save("sets", record) + except EvaluationError as error: + if error.status != 409: + raise + # A deletion marker deliberately prevents background re-creation + return self._get_set(kind) + + async def create_set( + self, name: str, description: str, actor: str + ) -> EvaluationSet: + record = EvaluationSet( + id=uuid4().hex, name=name, description=description, createdBy=actor + ) + return await asyncio.to_thread(self._save, "sets", record) + + async def get_set(self, identifier: str) -> EvaluationSet: + return await asyncio.to_thread(self._get_set, identifier) + + async def update_set( + self, identifier: str, name: str, description: str, revision: str + ) -> EvaluationSet: + record = await self.get_set(identifier) + return await asyncio.to_thread( + self._save, + "sets", + record.model_copy(update={"name": name, "description": description}), + revision, + ) + + async def list_sets(self) -> list[EvaluationSet]: + rows = await asyncio.to_thread(self._scan, "sets") + return [ + EvaluationSet.model_validate({**row, "revision": etag}) + for row, etag in rows + if not row.get("deletedAt") + ] + + async def list_samples(self) -> list[Sample]: + sets, rows = await asyncio.gather( + self.list_sets(), asyncio.to_thread(self._scan, "samples") + ) + active = {item.id for item in sets} + return sorted( + [ + Sample.model_validate({**row, "revision": etag}) + for row, etag in rows + if not row.get("deletedAt") and row.get("evaluationSetId") in active + ], + key=lambda row: (row.created_at, row.id), + reverse=True, + ) + + async def get_sample(self, identifier: str) -> Sample: + result = await asyncio.to_thread(self._read, "samples", identifier) + if result is None or result[0].get("deletedAt"): + raise EvaluationError("样本不存在或已删除", 404) + record = Sample.model_validate({**result[0], "revision": result[1]}) + await self.get_set(record.evaluation_set_id) + return record + + async def save_sample(self, record: Sample, revision: str = "") -> Sample: + collection = await self.get_set(record.evaluation_set_id) + if collection.default_kind: + record = record.model_copy(update={"kind": collection.default_kind}) + result = await asyncio.to_thread(self._save, "samples", record, revision) + try: + await self.get_set(record.evaluation_set_id) + except EvaluationError as error: + if error.status == 404: + await self.delete_sample(result.id, result.revision) + raise + return result + + async def save_feedback(self, record: Sample) -> Sample: + await asyncio.to_thread(self._ensure_default, record.evaluation_set_id) + previous = await asyncio.to_thread(self._read, "samples", record.id) + if previous and not previous[0].get("deletedAt"): + record = record.model_copy(update={"created_at": previous[0]["createdAt"]}) + return await self.save_sample(record, previous[1] if previous else "") + + async def save_automatic(self, record: Sample) -> Sample | None: + previous = await asyncio.to_thread(self._read, "samples", record.id) + if previous: + if previous[0].get("deletedAt"): + return None + return await self.get_sample(record.id) + await asyncio.to_thread(self._ensure_default, record.evaluation_set_id) + try: + return await self.save_sample(record) + except EvaluationError as error: + if error.status != 409: + raise + return await self.get_sample(record.id) + + async def delete_sample(self, identifier: str, revision: str = "") -> bool: + previous = await asyncio.to_thread(self._read, "samples", identifier) + if previous is None or previous[0].get("deletedAt"): + return False + marker = { + "schemaVersion": 1, + "id": identifier, + "evaluationSetId": previous[0]["evaluationSetId"], + "source": previous[0]["source"], + "userId": previous[0].get("userId", ""), + "sessionId": previous[0].get("sessionId", ""), + "messageId": previous[0].get("messageId", ""), + "deletedAt": now(), + } + await asyncio.to_thread( + self._write, "samples", identifier, marker, revision or previous[1] + ) + return True + + async def delete_set(self, identifier: str, revision: str) -> None: + record = await asyncio.to_thread(self._get_set, identifier, True) + if not record.deleted_at: + await asyncio.to_thread( + self._save, + "sets", + record.model_copy(update={"deleted_at": now(), "description": ""}), + revision, + ) + rows = await asyncio.to_thread(self._scan, "samples") + for row, etag in rows: + if row.get("evaluationSetId") == identifier and not row.get("deletedAt"): + await self.delete_sample(str(row["id"]), etag) + + async def feedback_states( + self, user_id: str, session_id: str, event_ids: list[str] + ) -> dict: + async def read(event_id: str): + identifier = sample_id("user", user_id, session_id, event_id) + result = await asyncio.to_thread(self._read, "samples", identifier) + if result is None or result[0].get("deletedAt"): + return event_id, {"rating": None, "syncStatus": "synced"} + row, _ = result + try: + collection = await self.get_set(row["evaluationSetId"]) + except EvaluationError as error: + if error.status != 404: + raise + return event_id, {"rating": None, "syncStatus": "synced"} + return event_id, { + "rating": row.get("kind"), + "comment": row.get("comment", ""), + "evaluationSetId": collection.id, + "evaluationSetName": collection.name, + "evaluationItemId": identifier, + "workspaceId": None, + "syncStatus": "synced", + "statePersistence": "runtime", + } + + result = {} + for start in range(0, len(event_ids), 8): + for event_id, state in await asyncio.gather( + *(read(value) for value in event_ids[start : start + 8]) + ): + result[f"veadk_feedback:{event_id}"] = state + return result diff --git a/frontend/server/evaluation/routes.py b/frontend/server/evaluation/routes.py new file mode 100644 index 000000000..32f363a76 --- /dev/null +++ b/frontend/server/evaluation/routes.py @@ -0,0 +1,463 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Evaluation APIs backed exclusively by Studio TOS storage""" + +from __future__ import annotations + +import asyncio +import time +from collections.abc import Callable +from typing import Any, Literal +from urllib.parse import quote +from uuid import uuid4 + +from fastapi import FastAPI, HTTPException, Query, Request +from fastapi.responses import JSONResponse +from pydantic import BaseModel, ConfigDict, Field + +from veadk.integrations.agentkit.evaluation.feedback import ( + extract_feedback_sample, + feedback_state_key, +) + +from .models import Sample, public_record +from .repository import EvaluationError, EvaluationStorage, sample_id + + +class Input(BaseModel): + model_config = ConfigDict( + extra="forbid", populate_by_name=True, str_strip_whitespace=True + ) + + +class SetInput(Input): + name: str = Field(min_length=1, max_length=100) + description: str = Field(default="", max_length=4000) + + +class SetUpdate(SetInput): + revision: str = Field(min_length=1) + + +class SampleInput(Input): + evaluation_set_id: str = Field(alias="evaluationSetId", min_length=1) + input: str = Field(min_length=1, max_length=200_000) + output: str = Field(min_length=1, max_length=200_000) + reference_output: str = Field( + default="", alias="referenceOutput", max_length=200_000 + ) + comment: str = Field(default="", max_length=20_000) + + +class SampleUpdate(SampleInput): + revision: str = Field(min_length=1) + + +class FeedbackInput(Input): + runtime_id: str = Field(alias="runtimeId", min_length=1) + region: str = "" + app_name: str = Field(alias="appName", min_length=1) + user_id: str = Field(alias="userId", min_length=1) + session_id: str = Field(alias="sessionId", min_length=1) + event_id: str = Field(alias="eventId", min_length=1) + rating: Literal["good", "bad"] | None + comment: str = Field(default="", max_length=20_000) + + +class DeleteInput(Input): + runtime_id: str = Field(alias="runtimeId", min_length=1) + region: str = "" + app_name: str = Field(alias="appName", min_length=1) + item_ids: list[str] = Field(alias="itemIds", min_length=1, max_length=200) + + +class FeedbackStateInput(Input): + runtime_id: str = Field(alias="runtimeId", min_length=1) + region: str = "" + user_id: str = Field(alias="userId", min_length=1) + session_id: str = Field(alias="sessionId", min_length=1) + event_ids: list[str] = Field(alias="eventIds", max_length=200) + + +def case_view(sample: Sample, runtime_id: str, set_names: dict[str, str]) -> dict: + annotated_bad = ( + sample.source == "user" + and sample.kind == "bad" + and bool(sample.comment.strip()) + ) + return { + **public_record(sample), + "itemKey": sample.id, + "runtimeId": runtime_id, + "agentName": "", + "workspaceId": "", + "evaluationSetName": set_names.get(sample.evaluation_set_id, ""), + "score": 0 if annotated_bad else sample.score, + "reason": sample.comment if annotated_bad else sample.reason, + } + + +def mount_routes( + app: FastAPI, + *, + storage: EvaluationStorage, + authorize: Callable, + principal: Callable, + runtime_request: Callable, + normalize_region: Callable, +) -> None: + async def storage_error(_request: Request, error: Exception) -> JSONResponse: + if not isinstance(error, EvaluationError): + raise error + return JSONResponse({"detail": str(error)}, status_code=error.status) + + app.add_exception_handler(EvaluationError, storage_error) + + def repository(request: Request, runtime_id: str, region: str): + authorize(request, runtime_id, normalize_region(region)) + return storage.for_runtime(runtime_id) + + def actor(request: Request) -> str: + value = principal(request) + return str(value.owner_id) if value else "local" + + @app.get("/web/evaluation/sets") + async def list_sets( + request: Request, runtimeId: str = Query(min_length=1), region: str = "" + ): + repo = repository(request, runtimeId, region) + sets, samples = await asyncio.gather(repo.list_sets(), repo.list_samples()) + return { + "items": [ + { + **public_record(item), + "itemCount": sum( + row.evaluation_set_id == item.id for row in samples + ), + } + for item in sets + ] + } + + @app.post("/web/evaluation/sets", status_code=201) + async def create_set( + data: SetInput, + request: Request, + runtimeId: str = Query(min_length=1), + region: str = "", + ): + repo = repository(request, runtimeId, region) + return public_record( + await repo.create_set(data.name, data.description, actor(request)) + ) + + @app.post("/web/evaluation/sets/defaults") + async def ensure_defaults( + request: Request, runtimeId: str = Query(min_length=1), region: str = "" + ): + return { + "items": [ + public_record(item) + for item in await repository( + request, runtimeId, region + ).ensure_defaults() + ] + } + + @app.get("/web/evaluation/sets/{identifier}") + async def get_set( + identifier: str, + request: Request, + runtimeId: str = Query(min_length=1), + region: str = "", + ): + return public_record( + await repository(request, runtimeId, region).get_set(identifier) + ) + + @app.patch("/web/evaluation/sets/{identifier}") + async def update_set( + identifier: str, + data: SetUpdate, + request: Request, + runtimeId: str = Query(min_length=1), + region: str = "", + ): + repo = repository(request, runtimeId, region) + return public_record( + await repo.update_set( + identifier, data.name, data.description, data.revision + ) + ) + + @app.delete("/web/evaluation/sets/{identifier}") + async def delete_set( + identifier: str, + request: Request, + revision: str = Query(min_length=1), + runtimeId: str = Query(min_length=1), + region: str = "", + ): + await repository(request, runtimeId, region).delete_set(identifier, revision) + return {"deleted": True} + + @app.get("/web/evaluation/samples") + async def list_samples( + request: Request, + runtimeId: str = Query(min_length=1), + region: str = "", + setId: str | None = None, + q: str = "", + source: Literal["user", "auto"] | None = None, + page: int = Query(default=1, ge=1), + pageSize: int = Query(default=50, ge=1, le=200), + ): + repo = repository(request, runtimeId, region) + if setId: + await repo.get_set(setId) + rows = [ + row + for row in await repo.list_samples() + if (not setId or row.evaluation_set_id == setId) + and (not source or row.source == source) + and ( + not q + or q.casefold() + in "\n".join((row.input, row.output, row.comment)).casefold() + ) + ] + offset = (page - 1) * pageSize + return { + "items": [public_record(row) for row in rows[offset : offset + pageSize]], + "total": len(rows), + "page": page, + "pageSize": pageSize, + } + + @app.post("/web/evaluation/samples", status_code=201) + async def create_sample( + data: SampleInput, + request: Request, + runtimeId: str = Query(min_length=1), + region: str = "", + ): + repo = repository(request, runtimeId, region) + sample = Sample( + id=uuid4().hex, source="user", userId=actor(request), **data.model_dump() + ) + return public_record(await repo.save_sample(sample)) + + @app.get("/web/evaluation/samples/{identifier}") + async def get_sample( + identifier: str, + request: Request, + runtimeId: str = Query(min_length=1), + region: str = "", + ): + return public_record( + await repository(request, runtimeId, region).get_sample(identifier) + ) + + @app.patch("/web/evaluation/samples/{identifier}") + async def update_sample( + identifier: str, + data: SampleUpdate, + request: Request, + runtimeId: str = Query(min_length=1), + region: str = "", + ): + repo = repository(request, runtimeId, region) + sample = await repo.get_sample(identifier) + return public_record( + await repo.save_sample( + sample.model_copy(update=data.model_dump(exclude={"revision"})), + data.revision, + ) + ) + + @app.delete("/web/evaluation/samples/{identifier}") + async def delete_sample( + identifier: str, + request: Request, + revision: str = Query(min_length=1), + runtimeId: str = Query(min_length=1), + region: str = "", + ): + return { + "deleted": await repository(request, runtimeId, region).delete_sample( + identifier, revision + ) + } + + async def persist_feedback_state( + request: Request, runtime: Any, data: FeedbackInput, value: dict + ) -> None: + try: + await runtime_request( + request, + runtime=runtime, + runtime_id=data.runtime_id, + region=normalize_region(data.region), + method="PATCH", + path=f"apps/{quote(data.app_name, safe='')}/users/{quote(data.user_id, safe='')}/sessions/{quote(data.session_id, safe='')}", + payload={"state_delta": {feedback_state_key(data.event_id): value}}, + ) + except HTTPException as error: + if error.status_code != 404: + raise + value["statePersistence"] = "browser" + + @app.post("/web/evaluation/feedback-state") + async def feedback_state(data: FeedbackStateInput, request: Request): + current = principal(request) + if current is None or data.user_id.casefold() not in current.identifiers: + raise HTTPException(403, "只能读取当前用户的会话反馈") + authorize(request, data.runtime_id, normalize_region(data.region), True) + return await storage.for_runtime(data.runtime_id).feedback_states( + data.user_id, data.session_id, list(dict.fromkeys(data.event_ids)) + ) + + @app.post("/web/evaluation/feedback") + async def feedback(data: FeedbackInput, request: Request): + current = principal(request) + if current is None or data.user_id.casefold() not in current.identifiers: + raise HTTPException(403, "只能提交当前用户的会话反馈") + region = normalize_region(data.region) + runtime = authorize(request, data.runtime_id, region, True) + repo = storage.for_runtime(data.runtime_id) + identifier = sample_id("user", data.user_id, data.session_id, data.event_id) + saved = None + collection = None + if data.rating is not None: + session = await runtime_request( + request, + runtime=runtime, + runtime_id=data.runtime_id, + region=region, + method="GET", + path=f"apps/{quote(data.app_name, safe='')}/users/{quote(data.user_id, safe='')}/sessions/{quote(data.session_id, safe='')}", + ) + try: + sample = extract_feedback_sample( + session, + target_event_id=data.event_id, + runtime_id=data.runtime_id, + agent_name="", + user_id=data.user_id, + ) + except ValueError as error: + raise HTTPException(400, str(error)) from error + saved = await repo.save_feedback( + Sample( + id=identifier, + evaluationSetId=data.rating, + source="user", + kind=data.rating, + input=sample.input, + output=sample.output, + referenceOutput=sample.output if data.rating == "good" else "", + comment=data.comment, + userId=data.user_id, + sessionId=data.session_id, + messageId=data.event_id, + invocationId=sample.invocation_id, + ) + ) + collection = await repo.get_set(saved.evaluation_set_id) + else: + await repo.delete_sample(identifier) + value = { + "rating": data.rating, + "comment": data.comment if saved else "", + "evaluationSetId": collection.id if collection else None, + "evaluationSetName": collection.name if collection else None, + "evaluationItemId": saved.id if saved else None, + "workspaceId": None, + "syncStatus": "synced", + "statePersistence": "runtime", + "updatedAt": time.time(), + } + await persist_feedback_state(request, runtime, data, value) + return value + + @app.get("/web/evaluation/feedback-cases") + async def feedback_cases( + request: Request, + runtimeId: str = Query(min_length=1), + appName: str = "", + region: str = "", + page_size: int = Query(default=100, ge=1, le=200), + ): + repo = repository(request, runtimeId, region) + sets, rows = await asyncio.gather(repo.list_sets(), repo.list_samples()) + names = {item.id: item.name for item in sets} + default_ids = {item.id for item in sets if item.default_kind} + rows = [row for row in rows if row.evaluation_set_id in default_ids] + return { + "runtimeId": runtimeId, + "agentName": appName, + "region": normalize_region(region), + "projectName": "", + "sets": [ + { + "kind": item.default_kind, + "evaluationSetId": item.id, + "evaluationSetName": item.name, + "workspaceId": "", + "itemCount": sum(row.evaluation_set_id == item.id for row in rows), + } + for item in sets + if item.default_kind + ], + "items": [case_view(row, runtimeId, names) for row in rows[:page_size]], + } + + @app.post("/web/evaluation/feedback-cases/delete") + async def delete_feedback(data: DeleteInput, request: Request): + repo = repository(request, data.runtime_id, data.region) + runtime = authorize(request, data.runtime_id, normalize_region(data.region)) + count = 0 + for identifier in dict.fromkeys(data.item_ids): + try: + row = await repo.get_sample(identifier) + except EvaluationError as error: + if error.status == 404: + continue + raise + count += await repo.delete_sample(identifier, row.revision) + if row.source == "user" and row.session_id and row.message_id: + value = { + "rating": None, + "evaluationSetId": None, + "evaluationItemId": None, + "syncStatus": "synced", + "statePersistence": "runtime", + "updatedAt": time.time(), + } + await persist_feedback_state( + request, + runtime, + FeedbackInput( + runtimeId=data.runtime_id, + region=data.region, + appName=data.app_name, + userId=row.user_id, + sessionId=row.session_id, + eventId=row.message_id, + rating=None, + ), + value, + ) + return {"deletedCount": count} diff --git a/frontend/server/evaluation/sessions.py b/frontend/server/evaluation/sessions.py new file mode 100644 index 000000000..f494e43f5 --- /dev/null +++ b/frontend/server/evaluation/sessions.py @@ -0,0 +1,52 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Keep existing conversation responses consistent with persisted feedback""" + +from urllib.parse import unquote + +from .repository import EvaluationStorage + + +def session_identity(path: str) -> tuple[str, str] | None: + parts = path.strip("/").split("/") + if len(parts) != 6 or parts[::2] != ["apps", "users", "sessions"]: + return None + return unquote(parts[3]), unquote(parts[5]) + + +async def enrich_session( + storage: EvaluationStorage, + runtime_id: str, + identity: tuple[str, str], + session: dict, +) -> dict: + if storage.factory is None: + return session + state = session.get("state") or {} + events = session.get("events") or [] + event_ids = { + key.removeprefix("veadk_feedback:") + for key in state + if key.startswith("veadk_feedback:") + } + event_ids.update( + event["id"] + for event in events + if event.get("id") and event.get("author") != "user" + ) + feedback = await storage.for_runtime(runtime_id).feedback_states( + *identity, sorted(event_ids) + ) + return {**session, "state": {**state, **feedback}} diff --git a/frontend/server/evaluation_automation/__init__.py b/frontend/server/evaluation_automation/__init__.py index dbff9d93c..76f1e8ecf 100644 --- a/frontend/server/evaluation_automation/__init__.py +++ b/frontend/server/evaluation_automation/__init__.py @@ -25,11 +25,12 @@ from frontend.server.storage.tos import create_tos_client_factory from veadk.utils.logger import get_logger -from .datasets import ensure_feedback_sets +from frontend.server.evaluation import EvaluationStorage +from frontend.server.evaluation.automatic import TosAutomaticCases + from .model_gateway import StructuredEvaluationModels from .models import RunSseActivity from .repository import ( - AgentKitAutoEvaluationRepository, InMemoryOptimizationRepository, TosOptimizationRepository, ) @@ -44,7 +45,7 @@ def create_service( *, - openapi_post: OpenApiPost, + evaluation_storage: EvaluationStorage | None = None, provider: StudioProvider = "volcengine", resolve_credentials: CredentialResolver | None = None, quiet_seconds: float = 300, @@ -84,26 +85,10 @@ async def runtime_get( raise TypeError("Runtime returned an invalid JSON response") return payload - async def case_repository( - activity: RunSseActivity, - ) -> AgentKitAutoEvaluationRepository: - async def post( - *, - action: str, - payload: dict[str, Any], - query: dict[str, str] | None = None, - ) -> dict[str, Any]: - return await openapi_post( - region=activity.region, - action=action, - payload=payload, - query=query, - ) - - return AgentKitAutoEvaluationRepository( - post, - project_name=activity.project_name, - ) + async def case_repository(activity: RunSseActivity) -> TosAutomaticCases: + if evaluation_storage is None: + raise RuntimeError("管理员未配置评测持久化存储") + return TosAutomaticCases(evaluation_storage.for_runtime(activity.runtime_id)) return EvaluationAutomationService( evaluator=models, @@ -120,7 +105,6 @@ async def post( "RunSseActivity", "RunSseObservation", "create_service", - "ensure_feedback_sets", "mount_routes", "observed_sse_stream", ] diff --git a/frontend/server/evaluation_automation/datasets.py b/frontend/server/evaluation_automation/datasets.py index f7b0bbd94..1c12f641e 100644 --- a/frontend/server/evaluation_automation/datasets.py +++ b/frontend/server/evaluation_automation/datasets.py @@ -12,48 +12,10 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Evaluation-set initialization for Studio deployments.""" +"""Initialize the two default Runtime evaluation collections in TOS""" -from __future__ import annotations +from frontend.server.evaluation.repository import TosEvaluationRepository -from collections.abc import Awaitable, Callable -from typing import Any -from veadk.integrations.agentkit.evaluation import ( - AgentKitEvaluationDatasetsClient, -) - -OpenApiPost = Callable[..., Awaitable[dict[str, Any]]] - - -async def ensure_feedback_sets( - *, - openapi_post: OpenApiPost, - region: str, - project_name: str, - agent_name: str, -) -> list[str]: - """Idempotently ensure the Studio Agent's good and bad feedback sets.""" - - async def post( - *, - action: str, - payload: dict[str, Any], - query: dict[str, str] | None = None, - ) -> dict[str, Any]: - return await openapi_post( - region=region, - action=action, - payload=payload, - query=query, - ) - - client = AgentKitEvaluationDatasetsClient( - post, - project_name=project_name, - ) - created = [] - for rating in ("good", "bad"): - evaluation_set = await client.ensure_feedback_set(agent_name, rating) - created.append(evaluation_set.name) - return created +async def ensure_feedback_sets(repository: TosEvaluationRepository) -> list[str]: + return [item.name for item in await repository.ensure_defaults()] diff --git a/frontend/server/evaluation_automation/model_gateway.py b/frontend/server/evaluation_automation/model_gateway.py index 894fdfef4..c79e23947 100644 --- a/frontend/server/evaluation_automation/model_gateway.py +++ b/frontend/server/evaluation_automation/model_gateway.py @@ -24,6 +24,8 @@ from pydantic import BaseModel +from veadk.utils.cloud_provider import cloud_provider_from_env + from .models import ( AutoEvaluationCase, AutoEvaluationOutput, @@ -42,7 +44,9 @@ class StructuredEvaluationModels: def __init__(self, model_name: str | None = None) -> None: self._model_name = model_name or os.getenv( "VEADK_STUDIO_EVALUATION_MODEL", - DEFAULT_AUTOMATION_MODEL, + "seed-2-0-lite-260228" + if cloud_provider_from_env() == "byteplus" + else DEFAULT_AUTOMATION_MODEL, ) async def evaluate( diff --git a/frontend/server/evaluation_automation/repository.py b/frontend/server/evaluation_automation/repository.py index b64dea047..59f1fc789 100644 --- a/frontend/server/evaluation_automation/repository.py +++ b/frontend/server/evaluation_automation/repository.py @@ -12,71 +12,22 @@ # See the License for the specific language governing permissions and # limitations under the License. -"""Persistence adapters for automatic evaluations and optimization snapshots.""" +"""Storage for automatic optimization snapshots""" from __future__ import annotations import asyncio import hashlib -import re from collections.abc import Callable -from dataclasses import dataclass -from typing import Any, Protocol +from typing import Any from urllib.parse import quote -from veadk.integrations.agentkit.evaluation import AgentKitOpenApiError +from .models import OptimizationSnapshot -from .models import AutoEvaluationCase, OptimizationSnapshot - -AUTO_FIELD_KEYS = ( - "input", - "reference_output", - "output", - "feedback_comment", - "agent_name", - "session_id", - "message_id", - "runtime_id", - "invocation_id", - "user_id", - "created_at", - "source", - "score", - "reason", - "evaluator_version", -) -_MAX_DATASET_NAME_LENGTH = 50 -_DUPLICATE_DATASET_ERROR_CODE = "601104504" -_LOOKUP_DELAYS = (0.25, 0.5, 1.0, 2.0, 4.0) _OPTIMIZATION_KEY_PREFIX = "veadk-studio/v1/evaluation-optimizations" _MAX_OPTIMIZATION_BYTES = 8 * 1024 * 1024 -class AgentKitPost(Protocol): - async def __call__( - self, - *, - action: str, - payload: dict[str, Any], - query: dict[str, str] | None = None, - ) -> dict[str, Any]: ... - - -@dataclass(frozen=True) -class _SetRef: - id: str - workspace_id: str - name: str - - -def auto_set_name(agent_name: str, kind: str) -> str: - if kind not in {"good", "bad"}: - raise ValueError(f"unsupported automatic evaluation kind: {kind}") - suffix = f"_auto_{kind}_case" - normalized = re.sub(r"\s+", "_", agent_name.strip()) or "agent" - return f"{normalized[: _MAX_DATASET_NAME_LENGTH - len(suffix)]}{suffix}" - - def auto_item_key( *, project_name: str, @@ -85,255 +36,9 @@ def auto_item_key( message_id: str, evaluator_version: str, ) -> str: - value = ( - f"auto\0{project_name}\0{runtime_id}\0{session_id}\0{message_id}\0" - f"{evaluator_version}" - ) - return hashlib.sha256(value.encode("utf-8")).hexdigest() - - -class AgentKitAutoEvaluationRepository: - """Store automatic cases in AgentKit datasets separate from user feedback.""" - - def __init__(self, post: AgentKitPost, *, project_name: str) -> None: - self._post = post - self._project_name = project_name.strip() or "default" - self._workspace_id: str | None = None - - @property - def _project_query(self) -> dict[str, str]: - return {"ProjectName": self._project_name} - - async def upsert(self, case: AutoEvaluationCase) -> AutoEvaluationCase: - evaluation_set = await self._ensure_set(case.agent_name, case.kind) - fields = case.field_values() - response = await self._post( - action="BatchCreateEvaluationSetItems", - query={ - **self._project_query, - "WorkspaceId": evaluation_set.workspace_id, - "EvaluationSetId": evaluation_set.id, - }, - payload={ - "Items": [ - { - "ItemKey": case.item_key, - "Turns": [ - { - "Id": "0", - "FieldDataList": [ - { - "Key": key, - "Name": key, - "Content": { - "ContentType": "Text", - "Text": fields.get(key, ""), - }, - } - for key in AUTO_FIELD_KEYS - ], - } - ], - } - ], - "SkipInvalidItems": False, - "AllowPartialAdd": False, - }, - ) - outputs = (response.get("Result") or {}).get("ItemOutputs") or [] - if not outputs or not outputs[0].get("ItemId"): - raise RuntimeError("AgentKit did not return the automatic evaluation item") - return case.model_copy( - update={ - "id": str(outputs[0]["ItemId"]), - "evaluation_set_id": evaluation_set.id, - "evaluation_set_name": evaluation_set.name, - "workspace_id": evaluation_set.workspace_id, - } - ) - - async def list_cases( - self, - *, - agent_name: str, - page_size: int = 100, - ) -> list[AutoEvaluationCase]: - workspace_id = await self._resolve_workspace_id() - if workspace_id is None: - return [] - result: list[AutoEvaluationCase] = [] - for kind in ("good", "bad"): - name = auto_set_name(agent_name, kind) - evaluation_set = await self._find_set(name, workspace_id=workspace_id) - if evaluation_set is None: - evaluation_set = await self._find_set( - self._fallback_name(name, workspace_id), - workspace_id=workspace_id, - ) - if evaluation_set is None: - continue - response = await self._post( - action="ListEvaluationSetItems", - query={ - **self._project_query, - "WorkspaceId": workspace_id, - "EvaluationSetId": evaluation_set.id, - }, - payload={"PageNumber": 1, "PageSize": max(1, min(page_size, 200))}, - ) - for item in _extract_items(response): - fields = _extract_fields(item) - try: - score = float(fields.get("score", "")) - except ValueError: - continue - result.append( - AutoEvaluationCase( - id=str(item.get("ItemId") or item.get("Id") or ""), - itemKey=str(item.get("ItemKey") or ""), - kind=kind, - input=fields.get("input", ""), - output=fields.get("output", ""), - referenceOutput=fields.get("reference_output", ""), - comment=fields.get("feedback_comment", ""), - agentName=fields.get("agent_name", agent_name), - sessionId=fields.get("session_id", ""), - messageId=fields.get("message_id", ""), - runtimeId=fields.get("runtime_id", ""), - invocationId=fields.get("invocation_id", ""), - userId=fields.get("user_id", ""), - createdAt=fields.get("created_at", ""), - evaluationSetId=evaluation_set.id, - evaluationSetName=evaluation_set.name, - workspaceId=evaluation_set.workspace_id, - score=score, - reason=fields.get("reason", ""), - evaluatorVersion=fields.get("evaluator_version", "v1"), - ) - ) - return result - - async def _ensure_set(self, agent_name: str, kind: str) -> _SetRef: - workspace_id = await self._resolve_workspace_id() - name = auto_set_name(agent_name, kind) - if workspace_id is None: - try: - return await self._create_set(name, workspace_id=None) - except AgentKitOpenApiError as error: - if error.code != _DUPLICATE_DATASET_ERROR_CODE: - raise - existing = await self._wait_for_set(name, workspace_id=None) - if existing is not None: - return existing - raise - - fallback = self._fallback_name(name, workspace_id) - for candidate in (name, fallback): - existing = await self._find_set(candidate, workspace_id=workspace_id) - if existing is not None: - return existing - try: - return await self._create_set(name, workspace_id=workspace_id) - except AgentKitOpenApiError as error: - if error.code != _DUPLICATE_DATASET_ERROR_CODE: - raise - existing = await self._wait_for_set(name, workspace_id=workspace_id) - if existing is not None: - return existing - return await self._create_set(fallback, workspace_id=workspace_id) - - async def _create_set(self, name: str, *, workspace_id: str | None) -> _SetRef: - query = dict(self._project_query) - if workspace_id: - query["WorkspaceId"] = workspace_id - response = await self._post( - action="CreateEvaluationSet", - query=query, - payload={ - "Name": name, - "Description": "AgentKit Studio 会话静默自动评测集", - "EvaluationSetSchema": { - "FieldSchemas": [ - { - "Key": key, - "Name": key, - "ContentType": "Text", - "TextSchema": '{"type":"string"}', - "Status": 1, - "DefaultDisplayFormat": 1, - "IsRequired": key - in {"input", "output", "session_id", "message_id"}, - "Hidden": False, - } - for key in AUTO_FIELD_KEYS - ] - }, - }, - ) - created_id = str((response.get("Result") or {}).get("EvaluationSetId") or "") - if not created_id: - raise RuntimeError("AgentKit did not return an automatic evaluation set ID") - visible = await self._wait_for_set(name, workspace_id=workspace_id) - if visible is None or visible.id != created_id: - raise RuntimeError("AgentKit automatic evaluation set is not visible") - return visible - - async def _resolve_workspace_id(self) -> str | None: - if self._workspace_id: - return self._workspace_id - response = await self._post( - action="ListEvaluationSets", - query=self._project_query, - payload={"PageNumber": 1, "PageSize": 1}, - ) - for item in (response.get("Result") or {}).get("EvaluationSets") or []: - version = item.get("EvaluationSetVersion") or {} - workspace_id = str( - item.get("WorkspaceId") or version.get("WorkspaceId") or "" - ) - if workspace_id: - self._workspace_id = workspace_id - return workspace_id - return None - - async def _find_set(self, name: str, *, workspace_id: str | None) -> _SetRef | None: - query = dict(self._project_query) - if workspace_id: - query["WorkspaceId"] = workspace_id - response = await self._post( - action="ListEvaluationSets", - query=query, - payload={"Name": name, "PageNumber": 1, "PageSize": 100}, - ) - for item in (response.get("Result") or {}).get("EvaluationSets") or []: - if str(item.get("Name") or "") != name: - continue - version = item.get("EvaluationSetVersion") or {} - resolved_workspace_id = str( - item.get("WorkspaceId") or version.get("WorkspaceId") or "" - ) - item_id = str(item.get("Id") or "") - if not item_id or not resolved_workspace_id: - raise RuntimeError("AgentKit evaluation set has invalid identifiers") - self._workspace_id = resolved_workspace_id - return _SetRef(item_id, resolved_workspace_id, name) - return None - - async def _wait_for_set( - self, name: str, *, workspace_id: str | None - ) -> _SetRef | None: - for delay in _LOOKUP_DELAYS: - await asyncio.sleep(delay) - existing = await self._find_set(name, workspace_id=workspace_id) - if existing is not None: - return existing - return None - - @staticmethod - def _fallback_name(name: str, workspace_id: str) -> str: - digest = hashlib.sha256(f"{workspace_id}\0{name}".encode()).hexdigest() - suffix = f"_{digest[:8]}" - return f"{name[: _MAX_DATASET_NAME_LENGTH - len(suffix)]}{suffix}" + return hashlib.sha256( + f"{runtime_id}\0{session_id}\0{message_id}\0{evaluator_version}".encode() + ).hexdigest() class InMemoryOptimizationRepository: @@ -412,41 +117,3 @@ def _key(runtime_id: str, app_name: str) -> str: runtime_segment = quote(runtime_id, safe="") app_segment = quote(app_name, safe="") return f"{_OPTIMIZATION_KEY_PREFIX}/{runtime_segment}/{app_segment}.json" - - -def _extract_items(response: dict[str, Any]) -> list[dict[str, Any]]: - result = response.get("Result") or {} - for key in ("Items", "EvaluationSetItems", "ItemDetails", "EvaluationItems"): - value = result.get(key) - if isinstance(value, list): - return [item for item in value if isinstance(item, dict)] - return [] - - -def _extract_fields(item: dict[str, Any]) -> dict[str, str]: - turns = item.get("Turns") - turn = turns[0] if isinstance(turns, list) and turns else {} - if not isinstance(turn, dict): - turn = {} - field_data = ( - turn.get("FieldDataList") - or turn.get("FieldData") - or item.get("FieldDataList") - or item.get("FieldData") - or [] - ) - result: dict[str, str] = {} - if not isinstance(field_data, list): - return result - for field in field_data: - if not isinstance(field, dict): - continue - key = str(field.get("Key") or field.get("Name") or "") - content = field.get("Content") - if not key: - continue - if isinstance(content, dict): - result[key] = str(content.get("Text") or content.get("Value") or "") - elif content is not None: - result[key] = str(content) - return result diff --git a/frontend/server/runtime_readiness.py b/frontend/server/runtime_readiness.py new file mode 100644 index 000000000..5a7a2e482 --- /dev/null +++ b/frontend/server/runtime_readiness.py @@ -0,0 +1,57 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Wait for newly deployed Runtime instances before accepting conversations""" + +from threading import Event +from time import monotonic +from typing import Any + +from agentkit.sdk.runtime.types import ListRuntimeInstancesRequest + + +def wait_for_runtime_instances( + client: Any, + runtime_id: str, + minimum: int, + *, + cancelled: Event, + timeout: float = 300, +) -> None: + # Control-plane Ready can precede the configured instances. A conversation + # started then may land on a temporary instance and lose its in-memory session. + if minimum <= 0: + return + deadline = monotonic() + timeout + while True: + if cancelled.is_set(): + raise RuntimeError("Deployment cancelled") + if monotonic() >= deadline: + raise RuntimeError("Runtime 实例未在时限内就绪，请在控制台检查实例状态") + response = client.list_runtime_instances( + ListRuntimeInstancesRequest(RuntimeId=runtime_id) + ) + ready = { + item.instance_name + for item in response.instance_items or [] + if item.runtime_id == runtime_id + and item.instance_status == "Ready" + and item.instance_name + # ListRuntimeInstances omits InstanceType. VeFaaS encodes reserved + # versus elastic instances in the instance name returned here. + and "-reserved-" in item.instance_name + } + if len(ready) >= minimum: + return + cancelled.wait(2) diff --git a/tests/cli/conftest.py b/tests/cli/conftest.py index 94269d1bc..fbfd4dad2 100644 --- a/tests/cli/conftest.py +++ b/tests/cli/conftest.py @@ -65,6 +65,16 @@ def _stub_studio_deploy_role_confirmation( ) +@pytest.fixture(autouse=True) +def _stub_studio_runtime_readiness(monkeypatch: pytest.MonkeyPatch) -> MagicMock: + """Keep deployment route tests independent of cloud instance polling""" + wait = MagicMock() + monkeypatch.setattr( + "frontend.server.runtime_readiness.wait_for_runtime_instances", wait + ) + return wait + + @pytest.fixture(autouse=True) def _stub_studio_runtime_role(monkeypatch: pytest.MonkeyPatch) -> MagicMock: """Keep deployment tests isolated from live Runtime IAM operations""" diff --git a/tests/cli/test_frontend_evaluation_feedback.py b/tests/cli/test_frontend_evaluation_feedback.py index f7314c12f..36c8a6543 100644 --- a/tests/cli/test_frontend_evaluation_feedback.py +++ b/tests/cli/test_frontend_evaluation_feedback.py @@ -14,9 +14,12 @@ from __future__ import annotations +import hashlib +import io import json from pathlib import Path from types import SimpleNamespace +from threading import Lock from typing import Any import pytest @@ -121,307 +124,69 @@ async def search_findskill(_catalog: object, **kwargs: Any) -> dict[str, object] assert response.json()["items"][0]["slug"] == "clawhub/pdf-reader" -@pytest.mark.parametrize( - ("agent_info_status", "expected_agent_name", "initially_empty"), - [ - (200, "客服助手", False), - (404, "agent", False), - (200, "客服助手", True), - ], - ids=["agent-info", "legacy-runtime-fallback", "first-feedback-dataset"], -) -@pytest.mark.parametrize("rating", ["good", "bad"]) -def test_message_feedback_writes_dataset_and_session_state( - monkeypatch: pytest.MonkeyPatch, - tmp_path: Path, - agent_info_status: int, - expected_agent_name: str, - initially_empty: bool, - rating: str, -) -> None: - app = _create_frontend_app(monkeypatch, tmp_path) - openapi_calls: list[dict[str, Any]] = [] - session_patches: list[dict[str, Any]] = [] - annotation_comment = "选中片段：回答\n\n批注：事实错误" - created = False - - class _FakeRuntimeClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs - - def get_runtime(self, request: Any) -> SimpleNamespace: - del request - return SimpleNamespace( - project_name="support", - tags=[], - network_configurations=[ - SimpleNamespace( - endpoint="https://runtime.example", - network_type="public", - ) - ], - authorizer_configuration=SimpleNamespace( - key_auth=SimpleNamespace(api_key="runtime-key"), - custom_jwt_authorizer=None, - ), - ) - - class _FakeAsyncClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs - - async def __aenter__(self) -> "_FakeAsyncClient": - return self - - async def __aexit__(self, *args: Any) -> None: - del args - - async def request(self, method: str, url: str, **kwargs: Any) -> _FakeResponse: - if method == "GET" and "/sessions/session-1" in url: - return _FakeResponse( - { - "id": "session-1", - "state": {}, - "events": [ - { - "id": "user-event", - "author": "user", - "content": {"parts": [{"text": "问题"}]}, - }, - { - "id": "assistant-event", - "author": "agent", - "invocationId": "invocation-1", - "content": {"parts": [{"text": "回答"}]}, - }, - ], - } - ) - if method == "GET" and "/web/agent-info/agent" in url: - return _FakeResponse( - {"name": "客服助手"} if agent_info_status == 200 else {}, - status_code=agent_info_status, - ) - if method == "PATCH" and "/sessions/session-1" in url: - json_headers = [ - (name.lower(), value) - for name, value in kwargs["headers"].items() - if name.lower() in {"accept", "content-type"} - ] - assert json_headers == [ - ("accept", "application/json"), - ("content-type", "application/json"), - ] - session_patches.append(kwargs["json"]) - return _FakeResponse({}, status_code=404) - raise AssertionError((method, url)) - - async def post(self, url: str, **kwargs: Any) -> _FakeResponse: - nonlocal created - del url - action = kwargs["params"]["Action"] - openapi_calls.append( - { - "action": action, - "params": kwargs["params"], - "body": kwargs["content"], - } - ) - if action == "ListEvaluationSets": - if initially_empty and not created: - return _FakeResponse({"Result": {"EvaluationSets": []}}) - return _FakeResponse( - { - "Result": { - "EvaluationSets": [ - { - "Id": "set-1", - "Name": f"{expected_agent_name}_{rating}_case", - "WorkspaceId": "workspace-1", - } - ] - } - } - ) - if action == "CreateEvaluationSet": - created = True - return _FakeResponse({"Result": {"EvaluationSetId": "set-1"}}) - if action == "BatchCreateEvaluationSetItems": - return _FakeResponse( - { - "Result": { - "ItemOutputs": [ - { - "ItemId": "item-1", - "ItemKey": "stable-key", - "IsNewItem": True, - } - ] - } - } - ) - raise AssertionError(action) - - monkeypatch.setattr( - "agentkit.sdk.runtime.client.AgentkitRuntimeClient", - _FakeRuntimeClient, - ) - monkeypatch.setattr("httpx.AsyncClient", _FakeAsyncClient) - - with TestClient(app) as client: - response = client.post( - "/web/evaluation/feedback", - headers={"X-VeADK-Local-User": "user-1"}, - json={ - "runtimeId": "runtime-1", - "region": "cn-beijing", - "appName": "agent", - "userId": "user-1", - "sessionId": "session-1", - "eventId": "assistant-event", - "rating": rating, - "comment": annotation_comment, - }, +class _StorageFailure(Exception): + def __init__(self, status: int): + self.status_code = status + + +class _FakeTos: + def __init__(self): + self.objects = {} + self.lock = Lock() + + def get_object(self, *, bucket, key): + with self.lock: + if key not in self.objects: + raise _StorageFailure(404) + content, etag = self.objects[key] + return SimpleNamespace(read=io.BytesIO(content).read, etag=etag) + + def put_object( + self, *, bucket, key, content, forbid_overwrite=False, if_match=None, **kwargs + ): + with self.lock: + previous = self.objects.get(key) + if (forbid_overwrite and previous) or ( + if_match and (not previous or previous[1] != if_match) + ): + raise _StorageFailure(412) + etag = hashlib.sha256(content).hexdigest() + self.objects[key] = content, etag + return SimpleNamespace(etag=etag) + + def list_objects_type2(self, *, bucket, prefix, **kwargs): + with self.lock: + keys = sorted(key for key in self.objects if key.startswith(prefix)) + return SimpleNamespace( + contents=[SimpleNamespace(key=key) for key in keys], is_truncated=False ) - assert response.status_code == 200 - assert response.json()["rating"] == rating - assert response.json()["comment"] == annotation_comment - assert response.json()["evaluationItemId"] == "item-1" - assert response.json()["evaluationSetName"] == ( - f"{expected_agent_name}_{rating}_case" - ) - assert response.json()["statePersistence"] == "browser" - expected_actions = [ - "ListEvaluationSets", - "ListEvaluationSets", - "BatchCreateEvaluationSetItems", - "ListEvaluationSets", - "ListEvaluationSets", - ] - if initially_empty: - expected_actions.insert(1, "CreateEvaluationSet") - assert [call["action"] for call in openapi_calls] == expected_actions - assert openapi_calls[0]["params"]["ProjectName"] == "support" - if initially_empty: - assert "WorkspaceId" not in openapi_calls[1]["params"] - assert "WorkspaceId" not in openapi_calls[2]["params"] - else: - assert openapi_calls[1]["params"]["WorkspaceId"] == "workspace-1" - state = session_patches[0]["state_delta"]["veadk_feedback:assistant-event"] - assert state["rating"] == rating - assert state["comment"] == annotation_comment - assert state["evaluationItemId"] == "item-1" - create_item_call = next( - call - for call in openapi_calls - if call["action"] == "BatchCreateEvaluationSetItems" - ) - create_item_payload = json.loads(create_item_call["body"]) - field_data = create_item_payload["Items"][0]["Turns"][0]["FieldDataList"] - fields = {field["Key"]: field["Content"]["Text"] for field in field_data} - assert fields["feedback_comment"] == annotation_comment - - -@pytest.mark.parametrize("rating", ["good", "bad"]) -def test_message_feedback_byteplus_is_noop( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path, rating: str -) -> None: - app = _create_frontend_app( - monkeypatch, - tmp_path, - studio=True, - provider="byteplus", - ) - - class _FakeRuntimeClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs - - def get_runtime(self, request: Any) -> SimpleNamespace: - del request - return SimpleNamespace( - project_name="support", - tags=[], - network_configurations=[ - SimpleNamespace( - endpoint="https://runtime.example", - network_type="public", - ) - ], - authorizer_configuration=SimpleNamespace( - key_auth=SimpleNamespace(api_key="runtime-key"), - custom_jwt_authorizer=None, - ), - ) - - class _UnexpectedAsyncClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs - raise AssertionError("BytePlus feedback should not call remote APIs") +@pytest.fixture(params=["volcengine", "byteplus"]) +def feedback_client(request, monkeypatch, tmp_path): + provider = request.param + region = "cn-beijing" if provider == "volcengine" else "ap-southeast-1" + backend = _FakeTos() + monkeypatch.setenv("VEADK_STUDIO_TOS_BUCKET", "evaluation-test") + monkeypatch.setenv("VEADK_STUDIO_TOS_REGION", region) monkeypatch.setattr( - "agentkit.sdk.runtime.client.AgentkitRuntimeClient", - _FakeRuntimeClient, - ) - monkeypatch.setattr("httpx.AsyncClient", _UnexpectedAsyncClient) - - with TestClient(app) as client: - response = client.post( - "/web/evaluation/feedback", - headers={"X-VeADK-Local-User": "user-1"}, - json={ - "runtimeId": "runtime-1", - "region": "ap-southeast-1", - "appName": "agent", - "userId": "user-1", - "sessionId": "session-1", - "eventId": "assistant-event", - "rating": rating, - }, - ) - - assert response.status_code == 200 - assert response.json() == { - "rating": None, - "evaluationSetId": None, - "evaluationSetName": None, - "workspaceId": None, - "evaluationItemId": None, - "syncStatus": "synced", - "statePersistence": "browser", - "updatedAt": response.json()["updatedAt"], - } - - -def test_message_feedback_uncheck_deletes_case_by_stable_item_key( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - from veadk.integrations.agentkit.evaluation.feedback import feedback_item_key - - app = _create_frontend_app(monkeypatch, tmp_path) - openapi_calls: list[dict[str, Any]] = [] - session_patches: list[dict[str, Any]] = [] - stable_key = feedback_item_key( - project_name="support", - runtime_id="runtime-1", - session_id="session-1", - message_id="assistant-event", + "frontend.server.evaluation.repository.create_cached_tos_client_factory", + lambda *args, **kwargs: lambda: backend, ) + app = _create_frontend_app(monkeypatch, tmp_path, studio=True, provider=provider) + patches = [] - class _FakeRuntimeClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs + class RuntimeClient: + def __init__(self, **kwargs): + pass - def get_runtime(self, request: Any) -> SimpleNamespace: - del request + def get_runtime(self, request): return SimpleNamespace( project_name="support", tags=[], network_configurations=[ SimpleNamespace( - endpoint="https://runtime.example", - network_type="public", + endpoint="https://runtime.example", network_type="public" ) ], authorizer_configuration=SimpleNamespace( @@ -430,18 +195,19 @@ def get_runtime(self, request: Any) -> SimpleNamespace: ), ) - class _FakeAsyncClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs + class HttpClient: + def __init__(self, **kwargs): + pass - async def __aenter__(self) -> "_FakeAsyncClient": + async def __aenter__(self): return self - async def __aexit__(self, *args: Any) -> None: - del args + async def __aexit__(self, *args): + pass - async def request(self, method: str, url: str, **kwargs: Any) -> _FakeResponse: - if method == "GET" and "/sessions/session-1" in url: + async def request(self, method, url, **kwargs): + assert "/apps/agent/users/user-1/sessions/session-1" in url + if method == "GET": return _FakeResponse( { "id": "session-1", @@ -455,625 +221,167 @@ async def request(self, method: str, url: str, **kwargs: Any) -> _FakeResponse: { "id": "assistant-event", "author": "agent", + "invocationId": "invocation-1", "content": {"parts": [{"text": "回答"}]}, }, ], } ) - if method == "GET" and "/web/agent-info/agent" in url: - return _FakeResponse({"name": "客服助手"}) - if method == "PATCH" and "/sessions/session-1" in url: - session_patches.append(kwargs["json"]) - return _FakeResponse({}) - raise AssertionError((method, url)) - - async def post(self, url: str, **kwargs: Any) -> _FakeResponse: - del url - action = kwargs["params"]["Action"] - body = kwargs["content"].decode("utf-8") - openapi_calls.append( - { - "action": action, - "params": kwargs["params"], - "body": body, - } + assert method == "PATCH" + headers = {name.lower(): value for name, value in kwargs["headers"].items()} + assert headers["content-type"] == "application/json" + patches.append(kwargs["json"]) + # Older Runtime gateways lack PATCH; the TOS record must remain usable + return _FakeResponse({}, status_code=404) + + async def post(self, url, **kwargs): + raise AssertionError( + "Evaluation storage must not call cloud evaluation APIs" ) - if action == "ListEvaluationSets": - if "客服助手_good_case" in body: - return _FakeResponse( - { - "Result": { - "EvaluationSets": [ - { - "Id": "good-set", - "Name": "客服助手_good_case", - "WorkspaceId": "workspace-1", - } - ] - } - } - ) - if "客服助手_bad_case" in body: - return _FakeResponse({"Result": {"EvaluationSets": []}}) - return _FakeResponse( - { - "Result": { - "EvaluationSets": [ - { - "Id": "probe-set", - "Name": "其他评测集", - "WorkspaceId": "workspace-1", - } - ] - } - } - ) - if action == "ListEvaluationSetItems": - return _FakeResponse( - { - "Result": { - "Items": [ - { - "ItemId": "good-item", - "ItemKey": stable_key, - "Turns": [{"FieldDataList": []}], - } - ] - } - } - ) - if action == "BatchDeleteEvaluationSetItems": - assert kwargs["params"]["EvaluationSetId"] == "good-set" - assert '"good-item"' in body - return _FakeResponse({"Result": {}}) - raise AssertionError(action) monkeypatch.setattr( - "agentkit.sdk.runtime.client.AgentkitRuntimeClient", - _FakeRuntimeClient, + "agentkit.sdk.runtime.client.AgentkitRuntimeClient", RuntimeClient ) - monkeypatch.setattr("httpx.AsyncClient", _FakeAsyncClient) - - with TestClient(app) as client: - response = client.post( - "/web/evaluation/feedback", - headers={"X-VeADK-Local-User": "user-1"}, - json={ - "runtimeId": "runtime-1", - "region": "cn-beijing", - "appName": "agent", - "userId": "user-1", - "sessionId": "session-1", - "eventId": "assistant-event", - "rating": None, - }, + monkeypatch.setattr("httpx.AsyncClient", HttpClient) + with TestClient(app, headers={"X-VeADK-Local-User": "user-1"}) as client: + yield SimpleNamespace( + client=client, backend=backend, patches=patches, region=region ) - assert response.status_code == 200 - assert response.json()["rating"] is None - assert "BatchDeleteEvaluationSetItems" in [call["action"] for call in openapi_calls] - patch = session_patches[0]["state_delta"]["veadk_feedback:assistant-event"] - assert patch["rating"] is None - assert patch["evaluationItemId"] is None - - -def test_feedback_cases_list_agentkit_dataset_items( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - annotation_comment = "选中片段：回答\n\n批注：事实错误" - - class _Automation: - async def get_optimizations(self, runtime_id: str, app_name: str) -> None: - del runtime_id, app_name - - async def close(self) -> None: - pass - - monkeypatch.setattr( - "frontend.server.evaluation_automation.create_service", - lambda **kwargs: _Automation(), - ) - app = _create_frontend_app(monkeypatch, tmp_path, studio=True) - openapi_calls: list[dict[str, Any]] = [] - - class _FakeRuntimeClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs - def get_runtime(self, request: Any) -> SimpleNamespace: - del request - return SimpleNamespace( - project_name="support", - tags=[], - network_configurations=[ - SimpleNamespace( - endpoint="https://runtime.example", - network_type="public", - ) - ], - authorizer_configuration=SimpleNamespace( - key_auth=SimpleNamespace(api_key="runtime-key"), - custom_jwt_authorizer=None, - ), - ) - - class _FakeAsyncClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs +def _feedback(region, rating: str | None = "good"): + return { + "runtimeId": "runtime-1", + "region": region, + "appName": "agent", + "userId": "user-1", + "sessionId": "session-1", + "eventId": "assistant-event", + "rating": rating, + "comment": "选中片段：回答\n\n批注：事实错误", + } - async def __aenter__(self) -> "_FakeAsyncClient": - return self - async def __aexit__(self, *args: Any) -> None: - del args +def _query(region): + return {"runtimeId": "runtime-1", "region": region, "appName": "agent"} - async def request(self, method: str, url: str, **kwargs: Any) -> _FakeResponse: - del kwargs - if method == "GET" and "/web/agent-info/agent" in url: - return _FakeResponse({}, status_code=404) - raise AssertionError((method, url)) - async def post(self, url: str, **kwargs: Any) -> _FakeResponse: - del url - action = kwargs["params"]["Action"] - body = kwargs["content"].decode("utf-8") - openapi_calls.append( - { - "action": action, - "params": kwargs["params"], - "body": body, - } - ) - if action == "ListEvaluationSets": - if "agent_auto_good_case" in body: - return _FakeResponse( - { - "Result": { - "EvaluationSets": [ - { - "Id": "auto-good-set", - "Name": "agent_auto_good_case", - "WorkspaceId": "workspace-1", - } - ] - } - } - ) - if "agent_auto_bad_case" in body: - return _FakeResponse({"Result": {"EvaluationSets": []}}) - if "agent_good_case" in body: - return _FakeResponse( - { - "Result": { - "EvaluationSets": [ - { - "Id": "good-set", - "Name": "agent_good_case", - "WorkspaceId": "workspace-1", - } - ] - } - } - ) - if "agent_bad_case" in body: - return _FakeResponse( - { - "Result": { - "EvaluationSets": [ - { - "Id": "bad-set", - "Name": "agent_bad_case", - "WorkspaceId": "workspace-1", - } - ] - } - } - ) - return _FakeResponse( - { - "Result": { - "EvaluationSets": [ - { - "Id": "probe-set", - "Name": "其他评测集", - "WorkspaceId": "workspace-1", - } - ] - } - } - ) - if action == "ListEvaluationSetItems": - set_id = kwargs["params"]["EvaluationSetId"] - if set_id == "auto-good-set": - return _FakeResponse( - { - "Result": { - "Items": [ - { - "ItemId": "auto-good-item", - "ItemKey": "auto-good-key", - "Turns": [ - { - "FieldDataList": [ - { - "Key": "input", - "Content": {"Text": "自动问题"}, - }, - { - "Key": "output", - "Content": {"Text": "自动回答"}, - }, - { - "Key": "runtime_id", - "Content": { - "Text": "runtime-1" - }, - }, - { - "Key": "score", - "Content": {"Text": "0.82"}, - }, - { - "Key": "reason", - "Content": { - "Text": "回答完整。" - }, - }, - { - "Key": "evaluator_version", - "Content": {"Text": "v1"}, - }, - ] - } - ], - } - ] - } - } - ) - rating = "good" if set_id == "good-set" else "bad" - return _FakeResponse( - { - "Result": { - "Items": [ - { - "ItemId": f"{rating}-item", - "ItemKey": f"{rating}-key", - "Turns": [ - { - "FieldDataList": [ - { - "Key": "input", - "Content": {"Text": "问题"}, - }, - { - "Key": "output", - "Content": {"Text": "回答"}, - }, - { - "Key": "user_id", - "Content": {"Text": "user-1"}, - }, - { - "Key": "feedback_comment", - "Content": { - "Text": ( - annotation_comment - if rating == "bad" - else "" - ) - }, - }, - ] - } - ], - } - ] - } - } - ) - raise AssertionError(action) +@pytest.mark.parametrize("rating", ["good", "bad"]) +def test_message_feedback_writes_tos_and_session_state(feedback_client, rating): + from frontend.server.evaluation.repository import sample_id - monkeypatch.setattr( - "agentkit.sdk.runtime.client.AgentkitRuntimeClient", - _FakeRuntimeClient, + env = feedback_client + response = env.client.post( + "/web/evaluation/feedback", json=_feedback(env.region, rating) ) - monkeypatch.setattr("httpx.AsyncClient", _FakeAsyncClient) - - with TestClient(app) as client: - response = client.get( - "/web/evaluation/feedback-cases", - headers={"X-VeADK-Local-User": "user-1"}, - params={ - "runtimeId": "runtime-1", - "region": "cn-beijing", - "appName": "agent", - }, - ) - assert response.status_code == 200 - payload = response.json() - assert payload["agentName"] == "agent" - assert {item["kind"] for item in payload["items"]} == {"good", "bad"} - assert payload["items"][0]["input"] == "问题" - manual_items = [item for item in payload["items"] if item["source"] == "user"] - automatic_items = [item for item in payload["items"] if item["source"] == "auto"] - assert len(manual_items) == 2 - manual_good = next(item for item in manual_items if item["kind"] == "good") - manual_bad = next(item for item in manual_items if item["kind"] == "bad") - assert manual_good["score"] is None - assert manual_good["reason"] == "" - assert manual_bad["score"] == 0 - assert manual_bad["reason"] == annotation_comment - assert manual_bad["comment"] == annotation_comment - assert len(automatic_items) == 1 - assert automatic_items[0]["score"] == 0.82 - assert automatic_items[0]["reason"] == "回答完整。" - assert {item["itemCount"] for item in payload["sets"]} == {1, 2} - listed_set_ids = { - call["params"].get("EvaluationSetId") - for call in openapi_calls - if call["action"] == "ListEvaluationSetItems" - } - assert listed_set_ids == {"good-set", "bad-set", "auto-good-set"} - - -def test_feedback_cases_byteplus_404_reports_unsupported( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - app = _create_frontend_app( - monkeypatch, - tmp_path, - studio=True, - provider="byteplus", + saved = response.json() + identifier = sample_id("user", "user-1", "session-1", "assistant-event") + assert saved["evaluationItemId"] == identifier + assert saved["evaluationSetId"] == rating + assert saved["statePersistence"] == "browser" + assert saved["comment"] == _feedback(env.region)["comment"] + state = env.patches[0]["state_delta"]["veadk_feedback:assistant-event"] + assert state["rating"] == rating + key = f"veadk-studio/v1/evaluation/runtime-1/samples/{identifier}.json" + content = json.loads(env.backend.objects[key][0]) + assert content["input"] == "问题" + assert content["output"] == "回答" + assert content["source"] == "user" + assert content["evaluationSetId"] == rating + assert "appName" not in content and "agentName" not in content + states = env.client.post( + "/web/evaluation/feedback-state", + json={ + "runtimeId": "runtime-1", + "region": env.region, + "userId": "user-1", + "sessionId": "session-1", + "eventIds": ["assistant-event"], + }, ) - - class _FakeRuntimeClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs - - def get_runtime(self, request: Any) -> SimpleNamespace: - del request - return SimpleNamespace( - project_name="support", - tags=[], - network_configurations=[ - SimpleNamespace( - endpoint="https://runtime.example", - network_type="public", - ) - ], - authorizer_configuration=SimpleNamespace( - key_auth=SimpleNamespace(api_key="runtime-key"), - custom_jwt_authorizer=None, - ), - ) - - class _FakeAsyncClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs - - async def __aenter__(self) -> "_FakeAsyncClient": - return self - - async def __aexit__(self, *args: Any) -> None: - del args - - async def request(self, method: str, url: str, **kwargs: Any) -> _FakeResponse: - del kwargs - if method == "GET" and "/web/agent-info/agent" in url: - return _FakeResponse({"name": "agent"}) - raise AssertionError((method, url)) - - async def post(self, url: str, **kwargs: Any) -> _FakeResponse: - del url, kwargs - return _FakeResponse( - { - "ResponseMetadata": { - "RequestId": ( - "02178602503621000000000000000000000ffffac101ef9ae9c3d" - ) - } - }, - status_code=404, - ) - - monkeypatch.setattr( - "agentkit.sdk.runtime.client.AgentkitRuntimeClient", - _FakeRuntimeClient, + assert states.status_code == 200 + assert states.json()["veadk_feedback:assistant-event"]["rating"] == rating + + +def test_message_feedback_reclassification_and_uncheck_use_one_tos_record( + feedback_client, +): + env = feedback_client + first = env.client.post( + "/web/evaluation/feedback", json=_feedback(env.region) + ).json() + changed = env.client.post( + "/web/evaluation/feedback", json=_feedback(env.region, "bad") ) - monkeypatch.setattr("httpx.AsyncClient", _FakeAsyncClient) - - with TestClient(app) as client: - response = client.get( - "/web/evaluation/feedback-cases", - headers={"X-VeADK-Local-User": "user-1"}, - params={ - "runtimeId": "runtime-1", - "region": "ap-southeast-1", - "appName": "agent", - }, - ) - - assert response.status_code == 200 - payload = response.json() - assert payload["unsupported"] is True - assert payload["unsupportedMessage"] == "BytePlus 暂不支持 AgentKit 评测集。" - assert payload["sets"] == [] - assert payload["items"] == [] - - -def test_feedback_cases_delete_removes_dataset_items_and_clears_rating( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - app = _create_frontend_app(monkeypatch, tmp_path) - openapi_calls: list[dict[str, Any]] = [] - session_patches: list[dict[str, Any]] = [] - - class _FakeRuntimeClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs - - def get_runtime(self, request: Any) -> SimpleNamespace: - del request - return SimpleNamespace( - project_name="support", - tags=[], - network_configurations=[ - SimpleNamespace( - endpoint="https://runtime.example", - network_type="public", - ) - ], - authorizer_configuration=SimpleNamespace( - key_auth=SimpleNamespace(api_key="runtime-key"), - custom_jwt_authorizer=None, - ), - ) - - class _FakeAsyncClient: - def __init__(self, **kwargs: Any) -> None: - del kwargs - - async def __aenter__(self) -> "_FakeAsyncClient": - return self - - async def __aexit__(self, *args: Any) -> None: - del args - - async def request(self, method: str, url: str, **kwargs: Any) -> _FakeResponse: - if method == "GET" and "/web/agent-info/agent" in url: - return _FakeResponse({"name": "客服助手"}) - if method == "PATCH" and "/sessions/session-1" in url: - session_patches.append(kwargs["json"]) - return _FakeResponse({}) - raise AssertionError((method, url)) - - async def post(self, url: str, **kwargs: Any) -> _FakeResponse: - del url - action = kwargs["params"]["Action"] - body = kwargs["content"].decode("utf-8") - openapi_calls.append( - { - "action": action, - "params": kwargs["params"], - "body": body, - } - ) - if action == "ListEvaluationSets": - if "客服助手_good_case" in body: - return _FakeResponse( - { - "Result": { - "EvaluationSets": [ - { - "Id": "good-set", - "Name": "客服助手_good_case", - "WorkspaceId": "workspace-1", - } - ] - } - } - ) - if "客服助手_bad_case" in body: - return _FakeResponse({"Result": {"EvaluationSets": []}}) - return _FakeResponse( - { - "Result": { - "EvaluationSets": [ - { - "Id": "probe-set", - "Name": "其他评测集", - "WorkspaceId": "workspace-1", - } - ] - } - } - ) - if action == "ListEvaluationSetItems": - return _FakeResponse( - { - "Result": { - "Items": [ - { - "ItemId": "good-item", - "ItemKey": "good-key", - "Turns": [ - { - "FieldDataList": [ - { - "Key": "session_id", - "Content": {"Text": "session-1"}, - }, - { - "Key": "message_id", - "Content": { - "Text": "assistant-event" - }, - }, - { - "Key": "user_id", - "Content": {"Text": "user-1"}, - }, - ] - } - ], - } - ] - } - } - ) - if action == "BatchDeleteEvaluationSetItems": - assert kwargs["params"]["EvaluationSetId"] == "good-set" - assert kwargs["params"]["WorkspaceId"] == "workspace-1" - assert '"good-item"' in body - return _FakeResponse({"Result": {}}) - raise AssertionError(action) - - monkeypatch.setattr( - "agentkit.sdk.runtime.client.AgentkitRuntimeClient", - _FakeRuntimeClient, + assert changed.status_code == 200 + assert changed.json()["evaluationItemId"] == first["evaluationItemId"] + rows = env.client.get( + "/web/evaluation/feedback-cases", params=_query(env.region) + ).json()["items"] + assert len(rows) == 1 and rows[0]["kind"] == "bad" + assert ( + rows[0]["score"] == 0 and rows[0]["reason"] == _feedback(env.region)["comment"] + ) + cleared = env.client.post( + "/web/evaluation/feedback", json=_feedback(env.region, None) + ) + assert cleared.status_code == 200 and cleared.json()["rating"] is None + assert ( + env.client.get( + "/web/evaluation/feedback-cases", params=_query(env.region) + ).json()["items"] + == [] + ) + assert all( + "回答" not in content.decode() for content, _ in env.backend.objects.values() ) - monkeypatch.setattr("httpx.AsyncClient", _FakeAsyncClient) - with TestClient(app) as client: - response = client.post( - "/web/evaluation/feedback-cases/delete", - headers={"X-VeADK-Local-User": "user-1"}, - json={ - "runtimeId": "runtime-1", - "region": "cn-beijing", - "appName": "agent", - "itemIds": ["good-item"], - }, - ) +def test_feedback_cases_keep_existing_page_contract_and_delete_clears_rating( + feedback_client, +): + env = feedback_client + initialized = env.client.post( + "/web/evaluation/sets/defaults", params=_query(env.region) + ) + assert initialized.status_code == 200 + saved = env.client.post( + "/web/evaluation/feedback", json=_feedback(env.region) + ).json() + response = env.client.get( + "/web/evaluation/feedback-cases", params=_query(env.region) + ) assert response.status_code == 200 - assert response.json()["deletedCount"] == 1 - assert "BatchDeleteEvaluationSetItems" in [call["action"] for call in openapi_calls] - feedback_patch = session_patches[0]["state_delta"]["veadk_feedback:assistant-event"] - assert feedback_patch["rating"] is None - assert feedback_patch["evaluationItemId"] is None - assert feedback_patch["syncStatus"] == "synced" - - -def test_message_feedback_rejects_another_user( - monkeypatch: pytest.MonkeyPatch, tmp_path: Path -) -> None: - app = _create_frontend_app(monkeypatch, tmp_path) + body = response.json() + assert {item["kind"] for item in body["sets"]} == {"good", "bad"} + assert body["agentName"] == "agent" + assert len(body["items"]) == 1 + row = body["items"][0] + assert row["source"] == "user" and row["kind"] == "good" + assert row["runtimeId"] == "runtime-1" + deleted = env.client.post( + "/web/evaluation/feedback-cases/delete", + json={**_query(env.region), "itemIds": [saved["evaluationItemId"]]}, + ) + assert deleted.status_code == 200 and deleted.json()["deletedCount"] == 1 + assert ( + env.patches[-1]["state_delta"]["veadk_feedback:assistant-event"]["rating"] + is None + ) + assert ( + env.client.get( + "/web/evaluation/feedback-cases", params=_query(env.region) + ).json()["items"] + == [] + ) - with TestClient(app) as client: - response = client.post( - "/web/evaluation/feedback", - headers={"X-VeADK-Local-User": "user-1"}, - json={ - "runtimeId": "runtime-1", - "appName": "agent", - "userId": "user-2", - "sessionId": "session-1", - "eventId": "assistant-event", - "rating": "bad", - }, - ) +def test_feedback_rejects_another_users_session(feedback_client): + env = feedback_client + response = env.client.post( + "/web/evaluation/feedback", + json={**_feedback(env.region), "userId": "another-user"}, + ) assert response.status_code == 403 + assert not env.backend.objects diff --git a/tests/cli/test_studio_rbac.py b/tests/cli/test_studio_rbac.py index 1cca8e416..741261284 100644 --- a/tests/cli/test_studio_rbac.py +++ b/tests/cli/test_studio_rbac.py @@ -1596,12 +1596,12 @@ def launch(*, config_file: str, **_kwargs: Any) -> SimpleNamespace: ), ) - async def initialize_evaluation_sets(**_kwargs: Any) -> list[str]: - raise AssertionError("BytePlus deploy should not create evaluation sets") + def initialize_evaluation_sets(_storage, runtime_id): + raise AssertionError("Default deployment should not create evaluation sets") monkeypatch.setattr("agentkit.toolkit.sdk.launch", launch) monkeypatch.setattr( - "frontend.server.evaluation_automation.datasets.ensure_feedback_sets", + "frontend.server.evaluation.repository.EvaluationStorage.for_runtime", initialize_evaluation_sets, ) app = _create_studio_app( @@ -1672,14 +1672,14 @@ def launch(*, config_file: str, **_kwargs: Any) -> SimpleNamespace: ), ) - async def initialize_evaluation_sets(**_kwargs: Any) -> list[str]: + def initialize_evaluation_sets(_storage, runtime_id): nonlocal evaluation_set_calls evaluation_set_calls += 1 - return ["unexpected"] + raise AssertionError("Default deployment should not create evaluation sets") monkeypatch.setattr("agentkit.toolkit.sdk.launch", launch) monkeypatch.setattr( - "frontend.server.evaluation_automation.datasets.ensure_feedback_sets", + "frontend.server.evaluation.repository.EvaluationStorage.for_runtime", initialize_evaluation_sets, ) app = _create_studio_app(monkeypatch, tmp_path, developers="developer") @@ -6179,7 +6179,7 @@ def test_update_deployment_reuses_owned_runtime_and_returns_new_version( captured_config: dict[str, Any] = {} captured_dockerfile = "" update_requests: list[Any] = [] - evaluation_set_calls: list[dict[str, Any]] = [] + evaluation_set_calls: list[str] = [] resolved_model_keys: list[dict[str, Any]] = [] runtime_tag_sync_calls: list[dict[str, Any]] = [] @@ -6272,14 +6272,18 @@ def resolve_model_key(**kwargs: Any) -> str: resolve_model_key, ) - async def initialize_evaluation_sets(**kwargs: Any) -> list[str]: - evaluation_set_calls.append(kwargs) - if evaluation_error: - raise RuntimeError(evaluation_error) - return ["updated-agent_good_case", "updated-agent_bad_case"] + def initialize_evaluation_sets(_storage, runtime_id): + evaluation_set_calls.append(runtime_id) + + async def ensure_defaults(): + if evaluation_error: + raise RuntimeError(evaluation_error) + return [] + + return SimpleNamespace(ensure_defaults=ensure_defaults) monkeypatch.setattr( - "frontend.server.evaluation_automation.datasets.ensure_feedback_sets", + "frontend.server.evaluation.repository.EvaluationStorage.for_runtime", initialize_evaluation_sets, ) if provider == "byteplus": @@ -6414,36 +6418,22 @@ async def _mark_validated_oauth_token(request: Request, call_next): evaluation_frames = [ frame for frame in frames if frame.get("phase") == "evaluation" ] - if provider == "byteplus": - assert evaluation_frames == [] - assert "warnings" not in frames[-1] - assert evaluation_set_calls == [] + assert evaluation_frames[0]["message"] == ("正在创建 Good Case 和 Bad Case 评测集") + if evaluation_error: + assert evaluation_frames[-1]["level"] == "warning" + assert evaluation_frames[-1]["message"] == ( + "Good Case 和 Bad Case 评测集创建失败" + ) + assert frames[-1]["warnings"] == [ + "Runtime 已部署，但评测集创建失败：evaluation workspace unavailable" + ] else: - assert evaluation_frames[0]["message"] == ( - "正在创建 Good Case 和 Bad Case 评测集" + assert evaluation_frames[-1]["level"] == "success" + assert evaluation_frames[-1]["message"] == ( + "Good Case 和 Bad Case 评测集已创建" ) - if evaluation_error: - assert evaluation_frames[-1]["level"] == "warning" - assert evaluation_frames[-1]["message"] == ( - "Good Case 和 Bad Case 评测集创建失败" - ) - assert frames[-1]["warnings"] == [ - "Runtime 已部署，但评测集创建失败：evaluation workspace unavailable" - ] - else: - assert evaluation_frames[-1]["level"] == "success" - assert evaluation_frames[-1]["message"] == ( - "Good Case 和 Bad Case 评测集已创建" - ) - assert "warnings" not in frames[-1] - assert len(evaluation_set_calls) == 1 - assert callable(evaluation_set_calls[0]["openapi_post"]) - assert evaluation_set_calls[0] | {"openapi_post": None} == { - "openapi_post": None, - "region": region, - "project_name": "default", - "agent_name": "updated-agent", - } + assert "warnings" not in frames[-1] + assert evaluation_set_calls == [runtime.runtime_id] cloud = captured_config["launch_types"]["cloud"] assert cloud["runtime_id"] == runtime.runtime_id assert cloud["runtime_name"] == runtime.name @@ -6794,28 +6784,27 @@ def get_completed_capability(client: TestClient) -> httpx.Response: "session_storage", "min_instance", "max_instance", - "expects_update", "quick_mode", ), [ - ("in-memory", 1, 1, True, False), - ("persistent", 1, 5, False, False), - ("persistent", 0, 5, True, False), - ("persistent", 2, 4, True, False), - ("persistent", 1, 5, False, True), + ("in-memory", 1, 1, False), + ("persistent", 1, 5, False), + ("persistent", 0, 5, False), + ("persistent", 2, 4, False), + ("persistent", 1, 5, True), ], ) @pytest.mark.parametrize("provider", ["volcengine", "byteplus"]) -def test_new_deployment_only_updates_non_default_instance_range( +def test_new_deployment_creates_requested_instance_range_without_republishing( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, session_storage: str, min_instance: int, max_instance: int, - expects_update: bool, quick_mode: bool, provider: str, _stub_studio_runtime_role, + _stub_studio_runtime_readiness, ) -> None: from agentkit.sdk.runtime.client import AgentkitRuntimeClient @@ -6908,6 +6897,11 @@ def launch(*, config_file: str, **_kwargs: Any) -> SimpleNamespace: assert frames[-1]["success"] is True assert frames[-1]["agentName"] == "demo-agent" assert frames[-1]["runtimeName"] == "generated-runtime-name" + _stub_studio_runtime_readiness.assert_called_once() + assert _stub_studio_runtime_readiness.call_args.args[1:] == ( + runtime_id, + min_instance, + ) created_tags = {tag.key: tag.value for tag in create_requests[-1].tags} assert created_tags["veadk:environment-id"] == "default" assert created_tags["veadk:author"] == "developer" @@ -6944,9 +6938,10 @@ def launch(*, config_file: str, **_kwargs: Any) -> SimpleNamespace: assert "ENABLE_APMPLUS" not in runtime_envs assert "OBSERVABILITY_OPENTELEMETRY_APMPLUS_API_KEY" not in runtime_envs assert not any(frame.get("phase") == "evaluation" for frame in frames) - assert bool(update_requests) is expects_update - assert all(request.apmplus_enable is True for request in update_requests) - assert any(frame.get("phase") == "update" for frame in frames) is expects_update + assert update_requests == [] + assert create_requests[0].min_instance == min_instance + assert create_requests[0].max_instance == max_instance + assert not any(frame.get("phase") == "update" for frame in frames) assert captured_config["launch_types"]["cloud"]["runtime_role_name"] == ( "shared-runtime-role" ) @@ -6961,12 +6956,6 @@ def launch(*, config_file: str, **_kwargs: Any) -> SimpleNamespace: or "快速模式 Runtime 已具备 AgentKit 资源访问权限" == frame.get("message") for frame in frames ) - if expects_update: - request = update_requests[0] - assert request.runtime_id == runtime_id - assert request.min_instance == min_instance - assert request.max_instance == max_instance - assert request.release_enable is True def test_deployment_rejects_internal_runtime_environment( @@ -7751,13 +7740,18 @@ def reject_registry_conflict( } -def test_single_instance_update_failure_fails_the_deployment_at_update_phase( +@pytest.mark.parametrize("ready", [True, False]) +def test_single_instance_deployment_does_not_start_a_second_release( monkeypatch: pytest.MonkeyPatch, tmp_path: Path, + _stub_studio_runtime_readiness, + ready: bool, ) -> None: from agentkit.sdk.runtime.client import AgentkitRuntimeClient runtime_id = "r-update-failure" + if not ready: + _stub_studio_runtime_readiness.side_effect = RuntimeError("实例未在时限内就绪") monkeypatch.setattr( AgentkitRuntimeClient, @@ -7810,9 +7804,10 @@ def launch(**_kwargs: Any) -> SimpleNamespace: ] assert response.status_code == 200 - assert frames[-1]["success"] is False - assert frames[-1]["phase"] == "update" - assert "instance update failed" in frames[-1]["error"] + assert frames[-1]["success"] is ready + if not ready: + assert "实例未在时限内就绪" in frames[-1]["error"] + assert not any(frame.get("phase") == "update" for frame in frames) def test_deployment_maps_create_runtime_duplicate_name_to_actionable_error( diff --git a/tests/cli/test_studio_startup_imports.py b/tests/cli/test_studio_startup_imports.py index 199ba758f..4553c7e77 100644 --- a/tests/cli/test_studio_startup_imports.py +++ b/tests/cli/test_studio_startup_imports.py @@ -1104,6 +1104,35 @@ def test_studio_deferred_genai_models_support_openapi_first_use() -> None: ) +def test_studio_deferred_genai_types_accept_evaluation_output_schema() -> None: + root = Path(__file__).resolve().parents[2] + script = """ +import os +import tempfile +os.environ["MODEL_AGENT_API_KEY"] = "test-only-key" +from veadk.cli.studio_start import studio_fast_api_factory +with tempfile.TemporaryDirectory() as agents_dir: + studio_fast_api_factory()(agents_dir=agents_dir, web=False) + from veadk import Agent + from frontend.server.evaluation_automation.models import AutoEvaluationOutput + agent = Agent( + name="evaluation_startup_test", + model_name="doubao-seed-2-0-lite-260428", + output_schema=AutoEvaluationOutput, + enable_responses=True, + ) + assert agent.output_schema is AutoEvaluationOutput +""" + result = subprocess.run( + [sys.executable, "-c", script], + cwd=root, + capture_output=True, + text=True, + timeout=60, + ) + assert result.returncode == 0, result.stderr + + def test_session_metadata_defers_agentkit_models_until_first_request() -> None: root = Path(__file__).resolve().parents[2] script = """ diff --git a/tests/frontend/server/evaluation/test_tos_evaluation_repository.py b/tests/frontend/server/evaluation/test_tos_evaluation_repository.py new file mode 100644 index 000000000..ab8b89f87 --- /dev/null +++ b/tests/frontend/server/evaluation/test_tos_evaluation_repository.py @@ -0,0 +1,337 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +import hashlib +import io +import json +from types import SimpleNamespace +from threading import Lock +from typing import Literal +from unittest.mock import MagicMock + +import pytest + +from frontend.server.evaluation.models import Sample +from frontend.server.evaluation.repository import ( + EvaluationError, + EvaluationStorage, + TosEvaluationRepository, + sample_id, +) + + +class StorageFailure(Exception): + def __init__(self, status): + self.status_code = status + + +class FakeTos: + def __init__(self): + self.objects = {} + self.lock = Lock() + + def put_object( + self, + *, + bucket, + key, + content, + content_type, + forbid_overwrite=False, + if_match=None, + ): + with self.lock: + old = self.objects.get(key) + if (forbid_overwrite and old) or ( + if_match and (not old or old[1] != if_match) + ): + raise StorageFailure(412) + etag = hashlib.sha256(content).hexdigest() + self.objects[key] = (content, etag) + return SimpleNamespace(etag=etag) + + def get_object(self, *, bucket, key): + if key not in self.objects: + raise StorageFailure(404) + data, etag = self.objects[key] + return SimpleNamespace(read=io.BytesIO(data).read, etag=etag) + + def list_objects_type2(self, *, bucket, prefix, continuation_token, max_keys): + keys = sorted(key for key in self.objects if key.startswith(prefix)) + start = int(continuation_token or 0) + end = start + 2 # Deliberately exercise pagination + return SimpleNamespace( + contents=[SimpleNamespace(key=key) for key in keys[start:end]], + is_truncated=end < len(keys), + next_continuation_token=str(end), + ) + + +@pytest.fixture +def storage(): + client = FakeTos() + return client, TosEvaluationRepository("bucket", lambda: client, "runtime") + + +def sample( + identifier="message", source: Literal["user", "auto"] = "user", collection="good" +): + return Sample( + id=identifier, + evaluationSetId=collection, + source=source, + kind="good", + input="问题", + output="回答", + userId="user", + sessionId="session", + messageId="message", + ) + + +@pytest.mark.asyncio +async def test_two_defaults_contain_both_sources_and_isolate_runtimes(storage): + client, repo = storage + await repo.ensure_defaults() + await repo.save_feedback(sample()) + await repo.save_automatic(sample("automatic", "auto")) + assert {row.id for row in await repo.list_sets()} == {"good", "bad"} + assert {row.source for row in await repo.list_samples()} == {"user", "auto"} + other = TosEvaluationRepository("bucket", lambda: client, "other") + assert await other.list_samples() == [] + for key, (content, _) in client.objects.items(): + assert key.startswith("veadk-studio/v1/evaluation/runtime/") + assert "appName" not in json.loads(content) + + +@pytest.mark.asyncio +async def test_custom_set_rename_move_and_stale_edit(storage): + _, repo = storage + custom = await repo.create_set("回归", "说明", "owner") + row = await repo.save_sample(sample(collection=custom.id)) + renamed = await repo.update_set(custom.id, "回归二", "更新", custom.revision) + assert renamed.id == custom.id + with pytest.raises(EvaluationError, match="数据已被修改"): + await repo.update_set(custom.id, "过期修改", "", custom.revision) + await repo.ensure_defaults() + moved = await repo.save_sample( + row.model_copy(update={"evaluation_set_id": "bad"}), row.revision + ) + assert moved.kind == "bad" + assert len(await repo.list_samples()) == 1 + with pytest.raises(EvaluationError): + await repo.delete_sample(row.id, row.revision) + assert await repo.delete_sample(row.id, moved.revision) + + +@pytest.mark.asyncio +async def test_deleted_auto_sample_and_set_do_not_reappear(storage): + client, repo = storage + await repo.ensure_defaults() + row = await repo.save_automatic(sample(source="auto")) + await repo.delete_sample(row.id, row.revision) + assert await repo.save_automatic(sample(source="auto")) is None + assert "回答" not in client.objects[repo._key("samples", row.id)][0].decode() + collection = await repo.get_set("good") + await repo.delete_set("good", collection.revision) + with pytest.raises(EvaluationError, match="评测集不存在"): + await repo.save_feedback(sample("new")) + assert {row.id for row in await repo.list_sets()} == {"bad"} + + +@pytest.mark.asyncio +async def test_feedback_reclassification_updates_one_sample(storage): + _, repo = storage + await repo.save_feedback(sample()) + await repo.save_feedback(sample(collection="bad")) + rows = await repo.list_samples() + assert len(rows) == 1 + assert rows[0].evaluation_set_id == "bad" + await repo.delete_sample(rows[0].id) + assert await repo.list_samples() == [] + await repo.save_feedback(sample()) + assert len(await repo.list_samples()) == 1 + + +def test_id_includes_user_and_source_without_app_or_project(): + assert sample_id("user", "u", "s", "m") != sample_id("auto", "u", "s", "m") + assert sample_id("user", "u", "s", "m") != sample_id("user", "other", "s", "m") + + +@pytest.mark.asyncio +async def test_feedback_state_clears_after_sample_or_set_deletion(storage): + _, repo = storage + identifier = sample_id("user", "user", "session", "message") + row = await repo.save_feedback(sample(identifier)) + states = await repo.feedback_states("user", "session", ["message"]) + assert states["veadk_feedback:message"]["rating"] == "good" + await repo.delete_sample(identifier, row.revision) + assert (await repo.feedback_states("user", "session", ["message"]))[ + "veadk_feedback:message" + ]["rating"] is None + await repo.save_feedback(sample(identifier)) + collection = await repo.get_set("good") + await repo.delete_set("good", collection.revision) + assert (await repo.feedback_states("user", "session", ["message"]))[ + "veadk_feedback:message" + ]["rating"] is None + + +def test_native_routes_support_custom_sets_and_enforce_runtime_access(storage): + from fastapi import FastAPI, HTTPException + from fastapi.testclient import TestClient + from frontend.server.evaluation.routes import mount_routes + + backend, _ = storage + app = FastAPI() + + def authorize(request, runtime_id, region, shared=False): + if runtime_id != "runtime": + raise HTTPException(404, "not found") + return SimpleNamespace() + + async def runtime_request(*args, **kwargs): + if kwargs["method"] == "PATCH": + return {} + return { + "id": "session", + "events": [ + { + "id": "question", + "author": "user", + "content": {"parts": [{"text": "问题"}]}, + }, + { + "id": "message", + "author": "agent", + "content": {"parts": [{"text": "回答"}]}, + }, + ], + } + + mount_routes( + app, + storage=MagicMock( + spec=EvaluationStorage, + for_runtime=lambda runtime_id: TosEvaluationRepository( + "bucket", lambda: backend, runtime_id + ), + ), + authorize=authorize, + principal=lambda request: SimpleNamespace( + owner_id="user", identifiers={"user"} + ), + runtime_request=runtime_request, + normalize_region=lambda region: region or "cn-beijing", + ) + client = TestClient(app) + suffix = "?runtimeId=runtime" + assert client.get("/web/evaluation/sets?runtimeId=other").status_code == 404 + created = client.post( + "/web/evaluation/sets" + suffix, json={"name": "自定义", "description": "描述"} + ) + assert created.status_code == 201 + collection = created.json() + value = {"evaluationSetId": collection["id"], "input": "问题", "output": "回答"} + response = client.post("/web/evaluation/samples" + suffix, json=value) + assert response.status_code == 201 + row = response.json() + assert row["source"] == "user" + assert "appName" not in row + edit = {**value, "output": "修改", "revision": row["revision"]} + assert ( + client.patch( + f"/web/evaluation/samples/{row['id']}" + suffix, json=edit + ).status_code + == 200 + ) + assert ( + client.patch( + f"/web/evaluation/samples/{row['id']}" + suffix, json=edit + ).status_code + == 409 + ) + assert ( + client.get("/web/evaluation/samples" + suffix + "&q=修改").json()["total"] == 1 + ) + assert client.get("/web/evaluation/samples" + suffix + "&page=0").status_code == 422 + payload = { + "runtimeId": "runtime", + "appName": "agent", + "userId": "user", + "sessionId": "session", + "eventId": "message", + "rating": "good", + } + assert client.post("/web/evaluation/feedback", json=payload).status_code == 200 + assert ( + client.post( + "/web/evaluation/feedback", json={**payload, "rating": "bad"} + ).status_code + == 200 + ) + assert ( + client.post( + "/web/evaluation/feedback", json={**payload, "userId": "other"} + ).status_code + == 403 + ) + assert len(client.get("/web/evaluation/sets" + suffix).json()["items"]) == 3 + assert ( + client.post( + "/web/evaluation/feedback", json={**payload, "rating": None} + ).status_code + == 200 + ) + assert ( + client.delete( + f"/web/evaluation/sets/{collection['id']}" + + suffix + + "&revision=" + + collection["revision"] + ).status_code + == 200 + ) + assert client.get("/web/evaluation/samples" + suffix).json()["total"] == 0 + + +@pytest.mark.asyncio +async def test_session_response_overrides_stale_feedback_without_frontend_changes( + storage, +): + from frontend.server.evaluation.sessions import enrich_session, session_identity + + _, repo = storage + service = MagicMock( + spec=EvaluationStorage, factory=True, for_runtime=lambda runtime_id: repo + ) + identifier = sample_id("user", "user", "session", "message") + row = await repo.save_feedback(sample(identifier)) + original = { + "state": {"custom": 1, "veadk_feedback:message": {"rating": "bad"}}, + "events": [{"id": "message", "author": "agent"}], + } + assert session_identity("apps/agent/users/user/sessions/session") == ( + "user", + "session", + ) + assert session_identity("apps/agent/users/user/sessions") is None + assert session_identity("apps/agent/users/user/sessions/session/events") is None + result = await enrich_session(service, "runtime", ("user", "session"), original) + assert result["state"]["custom"] == 1 + assert result["state"]["veadk_feedback:message"]["rating"] == "good" + await repo.delete_sample(row.id, row.revision) + result = await enrich_session(service, "runtime", ("user", "session"), original) + assert result["state"]["veadk_feedback:message"]["rating"] is None + assert original["state"]["veadk_feedback:message"]["rating"] == "bad" diff --git a/tests/frontend/server/evaluation_automation/test_datasets.py b/tests/frontend/server/evaluation_automation/test_datasets.py index 574795248..c6bb86141 100644 --- a/tests/frontend/server/evaluation_automation/test_datasets.py +++ b/tests/frontend/server/evaluation_automation/test_datasets.py @@ -12,49 +12,17 @@ # See the License for the specific language governing permissions and # limitations under the License. -from __future__ import annotations - -from types import SimpleNamespace -from typing import Any - import pytest -from frontend.server.evaluation_automation import datasets +from frontend.server.evaluation_automation.datasets import ensure_feedback_sets @pytest.mark.asyncio -async def test_ensure_feedback_sets_creates_good_and_bad_sets( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[tuple[str, str]] = [] - - class _FakeDatasetsClient: - def __init__(self, post: Any, *, project_name: str) -> None: - assert project_name == "default" - self.post = post - - async def ensure_feedback_set( - self, agent_name: str, rating: str - ) -> SimpleNamespace: - calls.append((agent_name, rating)) - return SimpleNamespace(name=f"{agent_name}_{rating}_case") - - async def openapi_post(**kwargs: Any) -> dict[str, Any]: - assert kwargs["region"] == "cn-beijing" - return {} - - monkeypatch.setattr( - datasets, - "AgentKitEvaluationDatasetsClient", - _FakeDatasetsClient, - ) +async def test_ensure_feedback_sets_uses_two_runtime_defaults(): + from types import SimpleNamespace - created = await datasets.ensure_feedback_sets( - openapi_post=openapi_post, - region="cn-beijing", - project_name="default", - agent_name="客服助手", - ) + class Repository: + async def ensure_defaults(self): + return [SimpleNamespace(name="Good Case"), SimpleNamespace(name="Bad Case")] - assert created == ["客服助手_good_case", "客服助手_bad_case"] - assert calls == [("客服助手", "good"), ("客服助手", "bad")] + assert await ensure_feedback_sets(Repository()) == ["Good Case", "Bad Case"] diff --git a/tests/frontend/server/evaluation_automation/test_model_gateway.py b/tests/frontend/server/evaluation_automation/test_model_gateway.py new file mode 100644 index 000000000..325c593a5 --- /dev/null +++ b/tests/frontend/server/evaluation_automation/test_model_gateway.py @@ -0,0 +1,56 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from types import SimpleNamespace +from unittest.mock import AsyncMock, Mock + +import pytest + +from frontend.server.evaluation_automation.model_gateway import ( + StructuredEvaluationModels, +) +from frontend.server.evaluation_automation.models import AutoEvaluationOutput + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + ("provider", "override", "expected"), + [ + ("volcengine", None, "doubao-seed-2-0-lite-260428"), + ("byteplus", None, "seed-2-0-lite-260228"), + ("byteplus", "custom-evaluator", "custom-evaluator"), + ], +) +async def test_evaluator_uses_provider_model_and_validates_structured_output( + monkeypatch, provider, override, expected +): + import veadk + + monkeypatch.setenv("CLOUD_PROVIDER", provider) + monkeypatch.setenv("AGENTKIT_CLOUD_PROVIDER", provider) + monkeypatch.delenv("VEADK_STUDIO_EVALUATION_MODEL", raising=False) + if override: + monkeypatch.setenv("VEADK_STUDIO_EVALUATION_MODEL", override) + agent = Mock() + run = AsyncMock(return_value='{"score": 0.9, "reason": "正确"}') + monkeypatch.setattr(veadk, "Agent", agent) + monkeypatch.setattr(veadk, "Runner", Mock(return_value=SimpleNamespace(run=run))) + result = await StructuredEvaluationModels().evaluate( + user_input="1+1", agent_output="2", agent_info={"name": "math"} + ) + assert agent.call_args.kwargs["model_name"] == expected + assert agent.call_args.kwargs["output_schema"] is AutoEvaluationOutput + assert result.score == 0.9 + assert result.reason == "正确" + assert "1+1" in run.call_args.args[0] diff --git a/tests/frontend/server/evaluation_automation/test_repository.py b/tests/frontend/server/evaluation_automation/test_repository.py index 327db9077..431e8a171 100644 --- a/tests/frontend/server/evaluation_automation/test_repository.py +++ b/tests/frontend/server/evaluation_automation/test_repository.py @@ -23,13 +23,11 @@ from frontend.server.evaluation_automation import create_service from frontend.server.evaluation_automation.models import ( - AutoEvaluationCase, OptimizationGroup, OptimizationSnapshot, OptimizationSuggestion, ) from frontend.server.evaluation_automation.repository import ( - AgentKitAutoEvaluationRepository, TosOptimizationRepository, ) @@ -148,7 +146,6 @@ async def openapi_post(**kwargs: Any) -> dict[str, Any]: raise AssertionError(kwargs) service = create_service( - openapi_post=openapi_post, provider="volcengine", resolve_credentials=resolve_credentials, ) @@ -161,191 +158,3 @@ async def openapi_post(**kwargs: Any) -> dict[str, Any]: assert restored == snapshot assert credentials_calls == 1 await service.close() - - -@pytest.mark.asyncio -async def test_agentkit_repository_creates_and_lists_an_auto_dataset() -> None: - calls: list[dict[str, Any]] = [] - created = False - - async def post( - *, action: str, payload: dict[str, Any], query: dict[str, str] | None = None - ) -> dict[str, Any]: - nonlocal created - calls.append({"action": action, "payload": payload, "query": query}) - if action == "ListEvaluationSets" and query == {"ProjectName": "support"}: - return { - "Result": { - "EvaluationSets": [ - {"Id": "other", "Name": "other", "WorkspaceId": "ws"} - ] - } - } - if action == "ListEvaluationSets": - name = payload.get("Name") - if created and name == "客服助手_auto_good_case": - return { - "Result": { - "EvaluationSets": [ - { - "Id": "auto-good", - "Name": name, - "WorkspaceId": "ws", - } - ] - } - } - return {"Result": {"EvaluationSets": []}} - if action == "CreateEvaluationSet": - created = True - return {"Result": {"EvaluationSetId": "auto-good"}} - if action == "BatchCreateEvaluationSetItems": - return { - "Result": { - "ItemOutputs": [ - {"ItemId": "item-1", "ItemKey": "key-1", "IsNewItem": True} - ] - } - } - if action == "ListEvaluationSetItems": - return { - "Result": { - "Items": [ - { - "ItemId": "item-1", - "ItemKey": "key-1", - "Turns": [ - { - "FieldDataList": [ - { - "Key": "score", - "Content": {"Text": "0.92"}, - }, - { - "Key": "reason", - "Content": {"Text": "回答完整。"}, - }, - { - "Key": "source", - "Content": {"Text": "auto"}, - }, - ] - } - ], - } - ] - } - } - raise AssertionError(action) - - repository = AgentKitAutoEvaluationRepository(post, project_name="support") - case = AutoEvaluationCase( - id="", - itemKey="key-1", - kind="good", - input="问题", - output="回答", - agentName="客服助手", - sessionId="session", - messageId="message", - runtimeId="runtime", - invocationId="invocation", - userId="user", - createdAt="2026-08-05T10:00:00+08:00", - score=0.92, - reason="回答完整。", - evaluatorVersion="v1", - ) - - stored = await repository.upsert(case) - items = await repository.list_cases(agent_name="客服助手") - - assert stored.id == "item-1" - assert items[0].score == 0.92 - create_call = next( - call for call in calls if call["action"] == "CreateEvaluationSet" - ) - field_keys = { - field["Key"] - for field in create_call["payload"]["EvaluationSetSchema"]["FieldSchemas"] - } - assert {"source", "score", "reason", "evaluator_version"} <= field_keys - - -@pytest.mark.asyncio -async def test_agentkit_repository_creates_first_dataset_without_workspace( - monkeypatch: pytest.MonkeyPatch, -) -> None: - calls: list[dict[str, Any]] = [] - created = False - - async def post( - *, action: str, payload: dict[str, Any], query: dict[str, str] | None = None - ) -> dict[str, Any]: - nonlocal created - calls.append({"action": action, "payload": payload, "query": query}) - if action == "ListEvaluationSets": - if created and payload.get("Name") == "客服助手_auto_good_case": - return { - "Result": { - "EvaluationSets": [ - { - "Id": "auto-good", - "Name": "客服助手_auto_good_case", - "WorkspaceId": "workspace-1", - } - ] - } - } - return {"Result": {"EvaluationSets": []}} - if action == "CreateEvaluationSet": - created = True - return {"Result": {"EvaluationSetId": "auto-good"}} - if action == "BatchCreateEvaluationSetItems": - return { - "Result": { - "ItemOutputs": [ - {"ItemId": "item-1", "ItemKey": "key-1", "IsNewItem": True} - ] - } - } - raise AssertionError(action) - - async def no_sleep(delay: float) -> None: - del delay - - monkeypatch.setattr( - "frontend.server.evaluation_automation.repository.asyncio.sleep", - no_sleep, - ) - repository = AgentKitAutoEvaluationRepository(post, project_name="support") - case = AutoEvaluationCase( - id="", - itemKey="key-1", - kind="good", - input="问题", - output="回答", - agentName="客服助手", - sessionId="session", - messageId="message", - runtimeId="runtime", - invocationId="invocation", - userId="user", - createdAt="2026-08-05T10:00:00+08:00", - score=0.92, - reason="回答完整。", - evaluatorVersion="v1", - ) - - stored = await repository.upsert(case) - - assert stored.id == "item-1" - assert stored.workspace_id == "workspace-1" - assert [call["action"] for call in calls] == [ - "ListEvaluationSets", - "CreateEvaluationSet", - "ListEvaluationSets", - "BatchCreateEvaluationSetItems", - ] - assert calls[1]["query"] == {"ProjectName": "support"} - assert calls[2]["query"] == {"ProjectName": "support"} diff --git a/tests/frontend/server/test_runtime_readiness.py b/tests/frontend/server/test_runtime_readiness.py new file mode 100644 index 000000000..56755ad05 --- /dev/null +++ b/tests/frontend/server/test_runtime_readiness.py @@ -0,0 +1,92 @@ +# Copyright (c) 2025 Beijing Volcano Engine Technology Co., Ltd. and/or its affiliates. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +from threading import Event +from types import SimpleNamespace +from unittest.mock import Mock + +import pytest + +from frontend.server.runtime_readiness import wait_for_runtime_instances + + +def instance(name: str, status: str = "Ready", runtime: str = "r-test"): + return SimpleNamespace( + instance_name=f"function-revision-reserved-{name}", + instance_status=status, + runtime_id=runtime, + ) + + +def test_waits_for_all_configured_instances_to_be_ready(monkeypatch): + client = Mock() + client.list_runtime_instances.side_effect = [ + SimpleNamespace(instance_items=[]), + SimpleNamespace( + instance_items=[ + SimpleNamespace( + instance_name="function-revision-elastic-one", + instance_status="Ready", + runtime_id="r-test", + ) + ] + ), + SimpleNamespace(instance_items=[instance("one", "Starting")]), + SimpleNamespace( + instance_items=[instance("one"), instance("other", runtime="r-other")] + ), + SimpleNamespace(instance_items=[instance("one"), instance("two")]), + ] + cancelled = Mock(spec=Event) + cancelled.is_set.return_value = False + cancelled.wait.return_value = False + wait_for_runtime_instances(client, "r-test", 2, cancelled=cancelled) + assert client.list_runtime_instances.call_count == 5 + assert cancelled.wait.call_count == 4 + assert client.list_runtime_instances.call_args.args[0].runtime_id == "r-test" + + +def test_zero_minimum_does_not_wait_for_scale_to_zero_runtime(): + client = Mock() + wait_for_runtime_instances(client, "r-test", 0, cancelled=Event()) + client.list_runtime_instances.assert_not_called() + + +def test_cancellation_stops_readiness_wait(): + cancelled = Event() + cancelled.set() + client = Mock() + with pytest.raises(RuntimeError, match="cancelled"): + wait_for_runtime_instances(client, "r-test", 1, cancelled=cancelled) + client.list_runtime_instances.assert_not_called() + + +def test_timeout_is_not_reported_as_success(monkeypatch): + clock = iter([0, 0, 301]) + monkeypatch.setattr( + "frontend.server.runtime_readiness.monotonic", lambda: next(clock) + ) + client = Mock() + client.list_runtime_instances.return_value = SimpleNamespace(instance_items=[]) + cancelled = Mock(spec=Event) + cancelled.is_set.return_value = False + with pytest.raises(RuntimeError, match="实例未在时限内就绪"): + wait_for_runtime_instances(client, "r-test", 1, cancelled=cancelled) + + +def test_cloud_errors_are_not_hidden(): + client = Mock() + client.list_runtime_instances.side_effect = RuntimeError("access denied") + with pytest.raises(RuntimeError, match="access denied"): + wait_for_runtime_instances(client, "r-test", 1, cancelled=Event()) diff --git a/veadk/cli/cli_frontend.py b/veadk/cli/cli_frontend.py index 1c04f2d9a..fe80291dc 100644 --- a/veadk/cli/cli_frontend.py +++ b/veadk/cli/cli_frontend.py @@ -6452,32 +6452,6 @@ def _delete_agentkit_runtime(runtime_id: str, region: str) -> None: ) client.delete_runtime(_rt.DeleteRuntimeRequest(RuntimeId=runtime_id)) - def _set_agentkit_runtime_instance_range( - runtime_id: str, - region: str, - min_instance: int, - max_instance: int, - ) -> None: - """Set a Runtime instance range and publish the configuration update.""" - from agentkit.sdk.runtime import types as _rt - from agentkit.sdk.runtime.client import AgentkitRuntimeClient - - ak, sk, token = _resolve_ve_credentials() - client = AgentkitRuntimeClient( - access_key=ak, - secret_key=sk, - session_token=token or "", - region=region, - ) - client.update_runtime( - _rt.UpdateRuntimeRequest( - RuntimeId=runtime_id, - MinInstance=min_instance, - MaxInstance=max_instance, - ReleaseEnable=True, - ) - ) - def _destroy_deploy_task_runtime(task: dict[str, Any]) -> bool: """Destroy a task's Runtime once, if creation has reached that stage.""" with _deploy_tasks_lock: @@ -6965,8 +6939,6 @@ def _draft_has_model_fallbacks(value: Any) -> bool: status_code=400, detail="createEvaluationSets must be a boolean", ) - if provider == "byteplus": - create_evaluation_sets = False min_instance = data.get("minInstance", 1) max_instance = data.get("maxInstance", 5) @@ -7079,11 +7051,6 @@ def _draft_has_model_fallbacks(value: Any) -> bool: use_managed_sidecar_release = ( sidecar_enabled and not source_preserving_requested ) - needs_instance_update = ( - not sidecar_enabled - and not runtime_id - and (min_instance != 1 or max_instance != 5) - ) region = config.get("region") or _default_cloud_region() project_name = config.get("projectName", "default") @@ -9052,6 +9019,10 @@ def _tagged_create(self, req, _orig=orig_create): for key, value in _merged_runtime_tags(req.tags).items() ] req.apmplus_enable = True + # Set limits before the first release so a later deployment + # cannot replace the instance holding the first session. + req.min_instance = min_instance + req.max_instance = max_instance created = _create_runtime_with_description_fallback( _orig, self, req ) @@ -9257,33 +9228,42 @@ def _launch_config(config: dict[str, Any]): result = _launch_config(final_config) else: result = _launch_config(sdk_agentkit_config) - if ( - result is not None - and getattr(result, "success", False) - and needs_instance_update - ): - created_runtime_id = str(task_state.get("runtime_id") or "") - if not created_runtime_id: - raise RuntimeError("Runtime 创建成功，但未返回 Runtime ID") - state["phase"] = "update" - _emit( - "info", - f"正在将 Runtime 实例数调整为 {min_instance}～{max_instance}", - 0, - ) - _set_agentkit_runtime_instance_range( - created_runtime_id, - region, - min_instance, - max_instance, - ) - _emit( - "success", - f"Runtime 实例数已调整为 {min_instance}～{max_instance}", - 100, - ) if result is not None and getattr(result, "success", False): _verify_sdk_sidecar_release(result) + if existing_runtime is None and not sidecar_enabled: + from agentkit.sdk.runtime.client import ( + AgentkitRuntimeClient, + ) + from frontend.server.runtime_readiness import ( + wait_for_runtime_instances, + ) + + deployed = getattr(result, "deploy_result", None) + metadata = getattr(deployed, "metadata", None) or {} + created_id = str( + task_state.get("runtime_id") + or metadata.get("runtime_id") + or "" + ) + if not created_id: + raise RuntimeError( + "Runtime 创建成功，但未返回 Runtime ID" + ) + access_key, secret_key, session_token = ( + _resolve_ve_credentials() + ) + _emit("info", "正在等待 Runtime 实例就绪", 95) + wait_for_runtime_instances( + AgentkitRuntimeClient( + access_key=access_key, + secret_key=secret_key, + session_token=session_token or "", + region=region, + ), + created_id, + min_instance, + cancelled=task_state["cancel_event"], + ) if existing_runtime is not None and provider != "byteplus": try: access_key, secret_key, session_token = ( @@ -9515,16 +9495,9 @@ async def _stream(): "\n\n" ) try: - from frontend.server.evaluation_automation.datasets import ( - ensure_feedback_sets, - ) - - await ensure_feedback_sets( - openapi_post=_agentkit_openapi_post, - region=region, - project_name=project_name, - agent_name=agent_name, - ) + await evaluation_storage.for_runtime( + deployed_runtime_id, + ).ensure_defaults() evaluation_complete = { "level": "success", "phase": "evaluation", @@ -10856,6 +10829,10 @@ async def _connect_runtime_route_channel(runtime_id: str, request: Request): ), } + from frontend.server.evaluation import EvaluationStorage + from frontend.server.evaluation.sessions import enrich_session, session_identity + + evaluation_storage = EvaluationStorage(provider, _resolve_ve_credentials) evaluation_automation: EvaluationAutomationService | None = None agent_usage_service: Any | None = None if studio: @@ -10867,24 +10844,10 @@ async def _connect_runtime_route_channel(runtime_id: str, request: Request): mount_routes as mount_agent_usage_routes, ) - async def _evaluation_automation_openapi_post( - *, - region: str, - action: str, - payload: dict[str, Any], - query: dict[str, str] | None = None, - ) -> dict[str, Any]: - return await _agentkit_openapi_post( - region=region, - action=action, - payload=payload, - query=query, - ) - evaluation_automation = create_evaluation_automation_service( - openapi_post=_evaluation_automation_openapi_post, provider=provider, resolve_credentials=_resolve_ve_credentials, + evaluation_storage=evaluation_storage, ) agent_usage_service = create_agent_usage_service( provider=provider, @@ -11755,6 +11718,22 @@ async def _studio_channel_body(): ), ) + identity = session_identity(path) if upstream_method == "GET" else None + if identity is not None and upstream.status_code == 200: + try: + session = json.loads(await upstream.aread()) + return JSONResponse( + await enrich_session( + evaluation_storage, runtime_id, identity, session + ), + headers=studio_runtime_context_headers( + runtime_request_context(upstream.headers) + ), + ) + finally: + await upstream.aclose() + await client.aclose() + async def _body(): try: if observation is None: @@ -13417,559 +13396,24 @@ async def _web_runtime_mcp_credentials( headers={"Cache-Control": "no-store", "Pragma": "no-cache"}, ) - @app.post("/web/evaluation/feedback") - async def _web_message_feedback( - feedback: _MessageFeedbackRequest, - request: Request, - ) -> dict[str, Any]: - """Persist one message rating in ADK state and AgentKit evaluation sets.""" - feedback.region = _coerce_cloud_region(feedback.region) - principal = _current_principal(request) - if ( - principal is None - or feedback.user_id.casefold() not in principal.identifiers - ): - raise HTTPException( - status_code=403, - detail="Feedback can only be submitted for the current user", - ) - runtime = _authorized_runtime( - request, - feedback.runtime_id, - feedback.region, - coded_access_error=True, - allow_shared=True, - ) - if provider == "byteplus": - return { - "rating": None, - "evaluationSetId": None, - "evaluationSetName": None, - "workspaceId": None, - "evaluationItemId": None, - "syncStatus": "synced", - "statePersistence": "browser", - "updatedAt": time.time(), - } - session_path = ( - f"apps/{quote(feedback.app_name, safe='')}/users/" - f"{quote(feedback.user_id, safe='')}/sessions/" - f"{quote(feedback.session_id, safe='')}" - ) - agent_info_path = f"web/agent-info/{quote(feedback.app_name, safe='')}" - - async def _feedback_agent_info() -> dict[str, Any]: - try: - return await _runtime_json_request( - request, - runtime=runtime, - runtime_id=feedback.runtime_id, - region=feedback.region, - method="GET", - path=agent_info_path, - ) - except HTTPException as error: - if error.status_code != 404: - raise - logger.info( - "Runtime %s does not expose Agent info; using app name %s " - "for evaluation feedback", - feedback.runtime_id, - feedback.app_name, - ) - return {} - - try: - session, agent_info = await asyncio.gather( - _runtime_json_request( - request, - runtime=runtime, - runtime_id=feedback.runtime_id, - region=feedback.region, - method="GET", - path=session_path, - ), - _feedback_agent_info(), - ) - from veadk.integrations.agentkit.evaluation import ( - AgentKitEvaluationDatasetsClient, - ) - from veadk.integrations.agentkit.evaluation.feedback import ( - extract_feedback_sample, - feedback_item_key, - feedback_state_key, - ) - - agent_name = str(agent_info.get("name") or feedback.app_name) - project_name = str(getattr(runtime, "project_name", "") or "default") - sample = extract_feedback_sample( - session, - target_event_id=feedback.event_id, - runtime_id=feedback.runtime_id, - agent_name=agent_name, - user_id=feedback.user_id, - ) - state_key = feedback_state_key(feedback.event_id) - session_state = session.get("state") - previous_value = ( - session_state.get(state_key) - if isinstance(session_state, dict) - else None - ) - previous: dict[str, Any] = ( - previous_value if isinstance(previous_value, dict) else {} - ) - - async def _evaluation_post( - *, - action: str, - payload: dict[str, Any], - query: dict[str, str] | None = None, - ) -> dict[str, Any]: - return await _agentkit_openapi_post( - region=feedback.region, - action=action, - payload=payload, - query=query, - ) - - evaluation = AgentKitEvaluationDatasetsClient( - _evaluation_post, - project_name=project_name, - ) - item_key = feedback_item_key( - project_name=project_name, - runtime_id=feedback.runtime_id, - session_id=feedback.session_id, - message_id=feedback.event_id, - ) - deleted_previous_item_ids: set[str] = set() - evaluation_set = None - evaluation_item = None - if feedback.rating is not None: - evaluation_set = await evaluation.ensure_feedback_set( - agent_name, - feedback.rating, - ) - evaluation_item = await evaluation.upsert_item( - evaluation_set_id=evaluation_set.id, - workspace_id=evaluation_set.workspace_id, - item_key=item_key, - fields=sample.fields( - rating=feedback.rating, - comment=feedback.comment, - ), - ) - - previous_rating = str(previous.get("rating") or "") - previous_item_id = str(previous.get("evaluationItemId") or "") - previous_set_id = str(previous.get("evaluationSetId") or "") - previous_workspace_id = str(previous.get("workspaceId") or "") - replacing_previous = previous_item_id and ( - feedback.rating is None or previous_rating != feedback.rating - ) - if replacing_previous and previous_set_id and previous_workspace_id: - await evaluation.delete_item( - evaluation_set_id=previous_set_id, - workspace_id=previous_workspace_id, - item_id=previous_item_id, - ) - deleted_previous_item_ids.add(previous_item_id) - - fallback_delete_ratings: tuple[str, ...] = () - if feedback.rating is None: - fallback_delete_ratings = ("good", "bad") - elif feedback.rating == "good": - fallback_delete_ratings = ("bad",) - elif feedback.rating == "bad": - fallback_delete_ratings = ("good",) - for stale_rating in fallback_delete_ratings: - stale_set, stale_items = await evaluation.list_feedback_items( - agent_name=agent_name, - rating=stale_rating, - page_size=200, - ) - if stale_set is None: - continue - for stale_item in stale_items: - if ( - stale_item.item_key != item_key - or stale_item.id in deleted_previous_item_ids - ): - continue - await evaluation.delete_item( - evaluation_set_id=stale_set.id, - workspace_id=stale_set.workspace_id, - item_id=stale_item.id, - ) - deleted_previous_item_ids.add(stale_item.id) - - feedback_state = { - "rating": feedback.rating, - "comment": feedback.comment if feedback.rating is not None else "", - "evaluationSetId": evaluation_set.id if evaluation_set else None, - "evaluationSetName": evaluation_set.name if evaluation_set else None, - "workspaceId": ( - evaluation_set.workspace_id if evaluation_set else None - ), - "evaluationItemId": evaluation_item.id if evaluation_item else None, - "syncStatus": "synced", - "statePersistence": "runtime", - "updatedAt": time.time(), - } - try: - await _runtime_json_request( - request, - runtime=runtime, - runtime_id=feedback.runtime_id, - region=feedback.region, - method="PATCH", - path=session_path, - payload={"state_delta": {state_key: feedback_state}}, - ) - except HTTPException as error: - if error.status_code != 404: - raise - feedback_state["statePersistence"] = "browser" - logger.warning( - "Runtime %s does not expose Session PATCH through its gateway; " - "feedback state will use the browser compatibility cache", - feedback.runtime_id, - ) - return feedback_state - except HTTPException: - raise - except ValueError as error: - raise HTTPException(status_code=400, detail=str(error)) from error - except RuntimeError as error: - raise HTTPException( - status_code=502, - detail=( - "同步反馈到 AgentKit 评测集失败：" + _safe_exception_detail(error) - ), - ) from error + from frontend.server.evaluation.routes import ( + mount_routes as mount_evaluation_routes, + ) - @app.get("/web/evaluation/feedback-cases") - async def _web_feedback_cases( - request: Request, - runtimeId: str = Query(..., min_length=1), - appName: str = Query(..., min_length=1), - region: str = Query(default="", min_length=0), - page_size: int = Query(default=100, ge=1, le=200), - ) -> dict[str, Any]: - """List AgentKit evaluation-set items created from message feedback.""" - region = _coerce_cloud_region(region) - runtime = _authorized_runtime( + mount_evaluation_routes( + app, + storage=evaluation_storage, + authorize=lambda request, runtime_id, region, shared=False: _authorized_runtime( request, - runtimeId, + runtime_id, region, coded_access_error=True, - ) - agent_info_path = f"web/agent-info/{quote(appName, safe='')}" - try: - try: - agent_info = await _runtime_json_request( - request, - runtime=runtime, - runtime_id=runtimeId, - region=region, - method="GET", - path=agent_info_path, - ) - except HTTPException as error: - if error.status_code != 404: - raise - agent_info = {"name": appName} - from frontend.server.evaluation_automation.repository import ( - AgentKitAutoEvaluationRepository, - ) - from veadk.integrations.agentkit.evaluation import ( - AgentKitEvaluationDatasetsClient, - ) - - agent_name = str(agent_info.get("name") or appName) - project_name = str(getattr(runtime, "project_name", "") or "default") - - async def _evaluation_post( - *, - action: str, - payload: dict[str, Any], - query: dict[str, str] | None = None, - ) -> dict[str, Any]: - return await _agentkit_openapi_post( - region=region, - action=action, - payload=payload, - query=query, - ) - - evaluation = AgentKitEvaluationDatasetsClient( - _evaluation_post, - project_name=project_name, - ) - response_sets: list[dict[str, Any]] = [] - response_items: list[dict[str, Any]] = [] - for rating in ("good", "bad"): - evaluation_set, items = await evaluation.list_feedback_items( - agent_name=agent_name, - rating=rating, - page_size=page_size, - ) - if evaluation_set is None: - response_sets.append( - { - "kind": rating, - "evaluationSetId": None, - "evaluationSetName": None, - "workspaceId": None, - "itemCount": 0, - } - ) - continue - response_sets.append( - { - "kind": rating, - "evaluationSetId": evaluation_set.id, - "evaluationSetName": evaluation_set.name, - "workspaceId": evaluation_set.workspace_id, - "itemCount": len(items), - } - ) - for item in items: - fields = item.fields - comment = fields.get("feedback_comment", "") - is_annotated_bad_case = rating == "bad" and bool(comment.strip()) - response_items.append( - { - "id": item.id or item.item_key, - "itemKey": item.item_key, - "kind": rating, - "input": fields.get("input", ""), - "output": fields.get("output", ""), - "referenceOutput": fields.get("reference_output", ""), - "comment": comment, - "agentName": fields.get("agent_name", agent_name), - "sessionId": fields.get("session_id", ""), - "messageId": fields.get("message_id", ""), - "runtimeId": fields.get("runtime_id", runtimeId), - "invocationId": fields.get("invocation_id", ""), - "userId": fields.get("user_id", ""), - "createdAt": fields.get("created_at", ""), - "evaluationSetId": evaluation_set.id, - "evaluationSetName": evaluation_set.name, - "workspaceId": evaluation_set.workspace_id, - "source": "user", - "score": 0 if is_annotated_bad_case else None, - "reason": comment if is_annotated_bad_case else "", - } - ) - if studio: - automatic = AgentKitAutoEvaluationRepository( - _evaluation_post, - project_name=project_name, - ) - automatic_cases = await automatic.list_cases( - agent_name=agent_name, - page_size=page_size, - ) - response_items.extend( - case.model_dump(mode="json", by_alias=True) - for case in automatic_cases - ) - for item in response_sets: - item["itemCount"] = sum( - case["kind"] == item["kind"] for case in response_items - ) - return { - "agentName": agent_name, - "runtimeId": runtimeId, - "region": region, - "projectName": project_name, - "sets": response_sets, - "items": response_items, - } - except HTTPException: - raise - except ValueError as error: - raise HTTPException(status_code=400, detail=str(error)) from error - except RuntimeError as error: - if provider == "byteplus" and "AgentKit OpenAPI returned HTTP 404" in str( - error - ): - return { - "agentName": appName, - "runtimeId": runtimeId, - "region": region, - "projectName": getattr(runtime, "project_name", "") or "default", - "sets": [], - "items": [], - "unsupported": True, - "unsupportedMessage": "BytePlus 暂不支持 AgentKit 评测集。", - } - raise HTTPException( - status_code=502, - detail="读取 AgentKit 评测集失败：" + _safe_exception_detail(error), - ) from error - - @app.post("/web/evaluation/feedback-cases/delete") - async def _web_delete_feedback_cases( - deletion: _DeleteFeedbackCasesRequest, - request: Request, - ) -> dict[str, Any]: - """Remove feedback cases and clear their thumbs state without deleting chat.""" - deletion.region = _coerce_cloud_region(deletion.region) - requested_ids = { - item_id.strip() - for item_id in deletion.item_ids - if item_id and item_id.strip() - } - if not requested_ids: - raise HTTPException(status_code=400, detail="No feedback cases selected") - runtime = _authorized_runtime( - request, - deletion.runtime_id, - deletion.region, - coded_access_error=True, - ) - agent_info_path = f"web/agent-info/{quote(deletion.app_name, safe='')}" - try: - try: - agent_info = await _runtime_json_request( - request, - runtime=runtime, - runtime_id=deletion.runtime_id, - region=deletion.region, - method="GET", - path=agent_info_path, - ) - except HTTPException as error: - if error.status_code != 404: - raise - agent_info = {"name": deletion.app_name} - from frontend.server.evaluation_automation.repository import ( - AgentKitAutoEvaluationRepository, - ) - from veadk.integrations.agentkit.evaluation import ( - AgentKitEvaluationDatasetsClient, - ) - from veadk.integrations.agentkit.evaluation.feedback import ( - feedback_state_key, - ) - - agent_name = str(agent_info.get("name") or deletion.app_name) - project_name = str(getattr(runtime, "project_name", "") or "default") - - async def _evaluation_post( - *, - action: str, - payload: dict[str, Any], - query: dict[str, str] | None = None, - ) -> dict[str, Any]: - return await _agentkit_openapi_post( - region=deletion.region, - action=action, - payload=payload, - query=query, - ) - - evaluation = AgentKitEvaluationDatasetsClient( - _evaluation_post, - project_name=project_name, - ) - matched: list[tuple[str, str, dict[str, str]]] = [] - for rating in ("good", "bad"): - evaluation_set, items = await evaluation.list_feedback_items( - agent_name=agent_name, - rating=rating, - page_size=200, - ) - if evaluation_set is None: - continue - for item in items: - if item.id not in requested_ids: - continue - matched.append( - (evaluation_set.id, evaluation_set.workspace_id, item.fields) - ) - await evaluation.delete_item( - evaluation_set_id=evaluation_set.id, - workspace_id=evaluation_set.workspace_id, - item_id=item.id, - ) - - automatic_deleted = 0 - if studio: - automatic = AgentKitAutoEvaluationRepository( - _evaluation_post, - project_name=project_name, - ) - automatic_cases = await automatic.list_cases( - agent_name=agent_name, - page_size=200, - ) - for case in automatic_cases: - if case.id not in requested_ids: - continue - await evaluation.delete_item( - evaluation_set_id=case.evaluation_set_id, - workspace_id=case.workspace_id, - item_id=case.id, - ) - automatic_deleted += 1 - - for _set_id, _workspace_id, fields in matched: - session_id = str(fields.get("session_id") or "") - message_id = str(fields.get("message_id") or "") - user_id = str(fields.get("user_id") or "") - if not session_id or not message_id or not user_id: - continue - session_path = ( - f"apps/{quote(deletion.app_name, safe='')}/users/" - f"{quote(user_id, safe='')}/sessions/" - f"{quote(session_id, safe='')}" - ) - feedback_state = { - "rating": None, - "evaluationSetId": None, - "evaluationSetName": None, - "workspaceId": None, - "evaluationItemId": None, - "syncStatus": "synced", - "statePersistence": "runtime", - "updatedAt": time.time(), - } - try: - await _runtime_json_request( - request, - runtime=runtime, - runtime_id=deletion.runtime_id, - region=deletion.region, - method="PATCH", - path=session_path, - payload={ - "state_delta": { - feedback_state_key(message_id): feedback_state, - } - }, - ) - except HTTPException as error: - if error.status_code != 404: - raise - logger.warning( - "Runtime %s does not expose Session PATCH; feedback case " - "was deleted but message state could not be cleared", - deletion.runtime_id, - ) - return {"deletedCount": len(matched) + automatic_deleted} - except HTTPException: - raise - except ValueError as error: - raise HTTPException(status_code=400, detail=str(error)) from error - except RuntimeError as error: - raise HTTPException( - status_code=502, - detail="删除 AgentKit 评测案例失败：" + _safe_exception_detail(error), - ) from error + allow_shared=shared, + ), + principal=_current_principal, + runtime_request=_runtime_json_request, + normalize_region=_coerce_cloud_region, + ) @app.get("/web/a2a-spaces") async def _web_list_a2a_spaces( diff --git a/veadk/cli/studio_start.py b/veadk/cli/studio_start.py index 5eb84bb2a..4f188c658 100644 --- a/veadk/cli/studio_start.py +++ b/veadk/cli/studio_start.py @@ -196,9 +196,13 @@ def _validate(value: Any) -> Any: model_validate = getattr(target, "model_validate", None) if callable(model_validate): return model_validate(value) - from pydantic import TypeAdapter + from pydantic import ConfigDict, TypeAdapter - return TypeAdapter(target).validate_python(value) + # GenAI aliases such as SchemaUnion include GenericAlias and + # UnionType, matching its own arbitrary-types model configuration. + return TypeAdapter( + target, config=ConfigDict(arbitrary_types_allowed=True) + ).validate_python(value) def _core_schema( cls: type[Any], From 82c8e1f33bdd70930a552925f76455a4df2e99ee Mon Sep 17 00:00:00 2001 From: evanlowe <62918515+evanlowe@users.noreply.github.com> Date: Mon, 21 Sep 2026 12:52:22 +0800 Subject: [PATCH 2/3] fix(ci): update runtime instance regression selector --- .github/workflows/harness-sidecar-release-gate.yaml | 2 +- tests/test_harness_sidecar_release_gate.py | 5 ++++- 2 files changed, 5 insertions(+), 2 deletions(-) diff --git a/.github/workflows/harness-sidecar-release-gate.yaml b/.github/workflows/harness-sidecar-release-gate.yaml index 540e33d32..4e44c9ac9 100644 --- a/.github/workflows/harness-sidecar-release-gate.yaml +++ b/.github/workflows/harness-sidecar-release-gate.yaml @@ -136,7 +136,7 @@ jobs: tests/cli/test_generated_agent_mcp.py \ tests/cli/test_generated_agent_backend_codegen_extended.py::test_generated_debug_applies_published_mcp_credential_contract_before_discovery \ tests/cli/test_studio_rbac.py::test_deployment_status_recovers_completed_update_from_fresh_instance \ - tests/cli/test_studio_rbac.py::test_new_deployment_only_updates_non_default_instance_range \ + tests/cli/test_studio_rbac.py::test_new_deployment_creates_requested_instance_range_without_republishing \ tests/cli/test_studio_rbac.py::test_sidecar_deployment_uses_agentkit_cli_structured_release \ tests/cli/test_studio_rbac.py::test_source_preserving_disabled_sidecar_contract_filters_harness_env \ tests/cli/test_studio_rbac.py::test_source_preserving_output_repository_migrates_only_tagless_runtimes \ diff --git a/tests/test_harness_sidecar_release_gate.py b/tests/test_harness_sidecar_release_gate.py index bf1f3e6dc..4e6fc40bb 100644 --- a/tests/test_harness_sidecar_release_gate.py +++ b/tests/test_harness_sidecar_release_gate.py @@ -65,7 +65,10 @@ def test_sidecar_release_gate_runs_backend_and_frontend_in_parallel() -> None: in backend_run ) assert "tests/frontend/server/test_runtime_iam.py" in backend_run - assert "test_new_deployment_only_updates_non_default_instance_range" in backend_run + assert ( + "test_new_deployment_creates_requested_instance_range_without_republishing" + in backend_run + ) assert ( "test_sidecar_update_resolves_or_explicitly_reuses_stored_mcp_credentials" in (backend_run) From 283e26c608cc9ceacaa6ddc407ae40829d8128b9 Mon Sep 17 00:00:00 2001 From: evanlowe <62918515+evanlowe@users.noreply.github.com> Date: Mon, 21 Sep 2026 13:11:14 +0800 Subject: [PATCH 3/3] test(studio): allow background recovery in credential reuse checks --- tests/cli/test_studio_rbac.py | 7 +++++++ 1 file changed, 7 insertions(+) diff --git a/tests/cli/test_studio_rbac.py b/tests/cli/test_studio_rbac.py index 741261284..f102fe01a 100644 --- a/tests/cli/test_studio_rbac.py +++ b/tests/cli/test_studio_rbac.py @@ -7380,6 +7380,13 @@ def test_sidecar_update_resolves_or_explicitly_reuses_stored_mcp_credentials( from agentkit.sdk.runtime.client import AgentkitRuntimeClient from veadk.extensions.harness import sidecar + # Credential reuse must not depend on background recovery finishing within + # two seconds on busy CI workers; separate tests cover pending latency + monkeypatch.setattr( + "veadk.cli.cli_frontend._RUNTIME_UPDATE_CAPABILITY_INITIAL_WAIT_SECONDS", + 30.0, + ) + agent_name = "stored_mcp_agent" unnamed = mode == "changed-unnamed-explicit-reuse" published_tool_name = "" if unnamed else "orders"