From 55864d8a4f26bbcac11d03f5fceeda62b475315a Mon Sep 17 00:00:00 2001 From: moonyue-w <300878504+moonyue-w@users.noreply.github.com> Date: Mon, 21 Sep 2026 16:27:13 +0800 Subject: [PATCH 1/3] test(support): make tests self-contained; rebuild integration off examples MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit tests/support 承载断言层+框架+逐字场景+桩+finish_session×2+finish_dream；test_examples→test_scenarios_offline 改测 tests/support（safe_error 两分支保留）；新建 tests/integration；testpaths/marker/addopts。examples 未改。 Task: 1789906941 --- Makefile | 6 +- pyproject.toml | 5 +- tests/integration/__init__.py | 4 + tests/integration/conftest.py | 25 ++ tests/integration/test_forward.py | 12 + tests/integration/test_managed.py | 12 + tests/support/__init__.py | 29 ++ tests/support/assertions.py | 94 ++++++ tests/support/cleanup.py | 55 +++ tests/support/harness.py | 184 ++++++++++ tests/support/memory.py | 34 ++ tests/support/scenarios/__init__.py | 1 + tests/support/scenarios/forward.py | 368 ++++++++++++++++++++ tests/support/scenarios/managed.py | 241 +++++++++++++ tests/support/stub.py | 229 +++++++++++++ tests/test_examples.py | 506 ---------------------------- tests/test_scenarios_offline.py | 195 +++++++++++ 17 files changed, 1489 insertions(+), 511 deletions(-) create mode 100644 tests/integration/__init__.py create mode 100644 tests/integration/conftest.py create mode 100644 tests/integration/test_forward.py create mode 100644 tests/integration/test_managed.py create mode 100644 tests/support/__init__.py create mode 100644 tests/support/assertions.py create mode 100644 tests/support/cleanup.py create mode 100644 tests/support/harness.py create mode 100644 tests/support/memory.py create mode 100644 tests/support/scenarios/__init__.py create mode 100644 tests/support/scenarios/forward.py create mode 100644 tests/support/scenarios/managed.py create mode 100644 tests/support/stub.py delete mode 100644 tests/test_examples.py create mode 100644 tests/test_scenarios_offline.py diff --git a/Makefile b/Makefile index 46b93f0..2ec874a 100644 --- a/Makefile +++ b/Makefile @@ -24,10 +24,10 @@ build: $(PYTHON) -m build test-live: - QODER_RUN_LIVE=1 QODER_LIVE_ENV_FILE="$(LIVE_ENV_FILE)" $(PYTHON) -m pytest examples/forward -m live -v + QODER_RUN_LIVE=1 QODER_LIVE_ENV_FILE="$(LIVE_ENV_FILE)" $(PYTHON) -m pytest tests/integration/test_forward.py -m integration -v test-live-managed: - QODER_RUN_LIVE=1 QODER_LIVE_ENV_FILE="$(LIVE_ENV_FILE)" $(PYTHON) -m pytest examples/managed -m live -v + QODER_RUN_LIVE=1 QODER_LIVE_ENV_FILE="$(LIVE_ENV_FILE)" $(PYTHON) -m pytest tests/integration/test_managed.py -m integration -v test-live-all: - QODER_RUN_LIVE=1 QODER_LIVE_ENV_FILE="$(LIVE_ENV_FILE)" $(PYTHON) -m pytest examples -m live -v + QODER_RUN_LIVE=1 QODER_LIVE_ENV_FILE="$(LIVE_ENV_FILE)" $(PYTHON) -m pytest tests/integration -m integration -v diff --git a/pyproject.toml b/pyproject.toml index 999482c..502a927 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -46,10 +46,11 @@ namespaces = false "*" = ["py.typed"] [tool.pytest.ini_options] -testpaths = ["tests", "examples"] +testpaths = ["tests"] asyncio_mode = "auto" asyncio_default_fixture_loop_scope = "function" -markers = ["live: opt-in tests against a real Qoder account"] +markers = ["integration: account-backed integration tests"] +addopts = '-m "not integration" --strict-markers' [tool.ruff] target-version = "py310" diff --git a/tests/integration/__init__.py b/tests/integration/__init__.py new file mode 100644 index 0000000..8e1ad4d --- /dev/null +++ b/tests/integration/__init__.py @@ -0,0 +1,4 @@ +"""账号级集成测试（live）：默认被 -m "not integration" deselect。 + +client/凭据仅在 conftest 的 live_example fixture 内构造，模块 import 期无网络。 +""" diff --git a/tests/integration/conftest.py b/tests/integration/conftest.py new file mode 100644 index 0000000..f935c39 --- /dev/null +++ b/tests/integration/conftest.py @@ -0,0 +1,25 @@ +import os + +import pytest + +from qca import Forward, Managed +from tests.support.harness import Config, Run + + +@pytest.fixture +def live_example(request): + if os.environ.get("QODER_RUN_LIVE") != "1": + pytest.skip("Set QODER_RUN_LIVE=1 to run account-backed examples") + mode = request.param + config = Config.load( + mode, + env_file=os.environ.get("QODER_LIVE_ENV_FILE", ".env.live"), + timeout=float(os.environ.get("QODER_LIVE_TIMEOUT", "300")), + ) + context = Run(config) + client_type = Forward if mode == "forward" else Managed + with client_type(**config.client_options()) as client: + try: + yield client, context + finally: + context.cleanup() diff --git a/tests/integration/test_forward.py b/tests/integration/test_forward.py new file mode 100644 index 0000000..c5798c5 --- /dev/null +++ b/tests/integration/test_forward.py @@ -0,0 +1,12 @@ +import pytest + +from tests.support.scenarios.forward import SCENARIOS + +pytestmark = pytest.mark.integration + + +@pytest.mark.parametrize("live_example", ["forward"], indirect=True) +@pytest.mark.parametrize("scenario", SCENARIOS.values(), ids=SCENARIOS) +def test_forward_example(live_example, scenario): + client, context = live_example + scenario(client, context) diff --git a/tests/integration/test_managed.py b/tests/integration/test_managed.py new file mode 100644 index 0000000..61ac1fd --- /dev/null +++ b/tests/integration/test_managed.py @@ -0,0 +1,12 @@ +import pytest + +from tests.support.scenarios.managed import SCENARIOS + +pytestmark = pytest.mark.integration + + +@pytest.mark.parametrize("live_example", ["managed"], indirect=True) +@pytest.mark.parametrize("scenario", SCENARIOS.values(), ids=SCENARIOS) +def test_managed_example(live_example, scenario): + client, context = live_example + scenario(client, context) diff --git a/tests/support/__init__.py b/tests/support/__init__.py new file mode 100644 index 0000000..bd29dc7 --- /dev/null +++ b/tests/support/__init__.py @@ -0,0 +1,29 @@ +"""tests 自足支持层：断言层 + 运行时基元副本 + verbatim 场景 + HTTP 桩 + 清理。 + +tests 不依赖 examples、examples 亦不依赖 tests（双向断开）。场景 SCENARIOS 请从 +tests.support.scenarios.forward / tests.support.scenarios.managed 取用。 +""" + +from tests.support.assertions import TurnResult, turn, wait_reply +from tests.support.cleanup import finish_dream, finish_session_forward, finish_session_managed +from tests.support.harness import Config, Run, choose_model, marker, name, read_env, safe_error +from tests.support.memory import ProjectMemory +from tests.support.stub import ExampleService + +__all__ = [ + "Config", + "ExampleService", + "ProjectMemory", + "Run", + "TurnResult", + "choose_model", + "finish_dream", + "finish_session_forward", + "finish_session_managed", + "marker", + "name", + "read_env", + "safe_error", + "turn", + "wait_reply", +] diff --git a/tests/support/assertions.py b/tests/support/assertions.py new file mode 100644 index 0000000..c9ba9a4 --- /dev/null +++ b/tests/support/assertions.py @@ -0,0 +1,94 @@ +"""测试专用的断言层（TurnResult/wait_reply/turn）。 + +这是从 examples/common/live.py 净移出的断言层——examples 侧不再保留。 +""" + +from __future__ import annotations + +from dataclasses import dataclass +from typing import Any + +from tests.support.harness import Run, name + + +@dataclass +class TurnResult: + text: str = "" + last_id: str = "" + tool_used: bool = False + complete: bool = False + + def observe(self, event: Any) -> None: + if hasattr(event, "to_dict"): + event = event.to_dict(mode="json") + kind = event.get("type") + if event.get("id"): + self.last_id = event["id"] + if kind in ("session.error", "session.status_terminated"): + raise AssertionError(f"Execution failed: {kind}, event_id={self.last_id}") + if kind in ("agent.tool_use", "agent.mcp_tool_use"): + self.tool_used = True + elif kind == "agent.message": + # Only the latest completed assistant message can satisfy assertions. + self.text = "\n".join( + block.get("text", "") for block in event.get("content", []) if block.get("type") == "text" + ) + elif kind == "session.status_idle": + reason = event.get("stop_reason") + reason = reason.get("type") if isinstance(reason, dict) else reason + if reason not in (None, "", "end_turn", "stop_sequence"): + raise AssertionError(f"Execution stopped early: {reason}, event_id={self.last_id}") + self.complete = bool(self.text) + + def verify(self, expected: list[str], require_tool: bool = False) -> None: + if not self.complete: + raise AssertionError(f"No idle state after assistant output; last_event_id={self.last_id}") + if not all(value in self.text for value in expected): + raise AssertionError(f"Assistant output is missing expected values; last_event_id={self.last_id}") + if require_tool and not self.tool_used: + raise AssertionError(f"No actual tool execution; last_event_id={self.last_id}") + + +def wait_reply(events: Any, run: Run, session_id: str, after: str = "") -> TurnResult: + result = TurnResult(last_id=after) + while not result.complete: + run.remaining() + page = events.list( + session_id, + order="asc", + limit=100, + extra_query={"after_id": result.last_id or None, "include_tool_calls": True}, + timeout=min(run.remaining(), 30), + ) + for index, event in enumerate(page): + if index >= 2000: + raise AssertionError("Event polling exceeded 2000 events") + result.observe(event) + if result.complete: + break + if not result.complete: + run.pause() + run.output("assistant", result.text) + return result + + +def turn( + events: Any, + run: Run, + session_id: str, + prompt: str, + expected: list[str], + *, + require_tool: bool = False, +) -> TurnResult: + run.output("user", prompt) + result = events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if len(result.data) != 1 or not result.data[0].id: + raise AssertionError("Send must return exactly one user event ID") + reply = wait_reply(events, run, session_id, result.data[0].id) + reply.verify(expected, require_tool) + return reply diff --git a/tests/support/cleanup.py b/tests/support/cleanup.py new file mode 100644 index 0000000..8265b24 --- /dev/null +++ b/tests/support/cleanup.py @@ -0,0 +1,55 @@ +"""场景复用的会话与 dream 清理。 + +forward 与 managed 的 finish_session 是两份不同实现（forward: cancel+archive； +managed: user.interrupt+delete），随各自 SCENARIOS 分别迁自 examples/{forward,managed}/_cleanup.py； +finish_dream 迁自 examples/managed/dream.py。场景侧以 `as finish_session` 别名保持调用点逐字不变。 +""" + +from __future__ import annotations + +from qca import Forward, Managed +from tests.support.harness import Run + + +def finish_session_forward(client: Forward, context: Run, session_id: str) -> None: + session = client.sessions.retrieve(session_id) + if session.status not in ("idle", "terminated"): + client.sessions.cancel(session_id) + while client.sessions.retrieve(session_id).status not in ("idle", "terminated"): + context.pause() + client.sessions.archive(session_id) + + +def finish_session_managed(client: Managed, context: Run, session_id: str) -> None: + session = client.sessions.retrieve(session_id) + if session.status not in ("idle", "terminated"): + client.sessions.events.send(session_id, events=[{"type": "user.interrupt"}]) + while client.sessions.retrieve(session_id).status not in ("idle", "terminated"): + context.pause() + client.sessions.delete(session_id) + + +def finish_dream(client: Managed, context: Run, dream_id: str, input_store_id: str) -> None: + dream = client.dreams.retrieve(dream_id) + if dream.status in ("pending", "running"): + client.dreams.cancel(dream_id) + while dream.status in ("pending", "running"): + context.pause() + dream = client.dreams.retrieve(dream_id) + errors = [] + actions = [] + if dream.session_id: + actions.append(lambda: finish_session_managed(client, context, dream.session_id)) + seen = {input_store_id} + for output in dream.outputs or []: + if output.memory_store_id and output.memory_store_id not in seen: + seen.add(output.memory_store_id) + actions.append(lambda store_id=output.memory_store_id: client.memory_stores.delete(store_id)) + actions.append(lambda: client.dreams.archive(dream_id)) + for action in actions: + try: + action() + except Exception as error: + errors.append(error) + if errors: + raise RuntimeError(f"Dream cleanup had {len(errors)} failures") from errors[0] diff --git a/tests/support/harness.py b/tests/support/harness.py new file mode 100644 index 0000000..6452343 --- /dev/null +++ b/tests/support/harness.py @@ -0,0 +1,184 @@ +"""测试专用的运行时基元（Config/Run/choose_model/safe_error/read_env/name/marker）。 + +这是 examples/common/live.py 运行时基元的测试侧副本；examples 与 tests 各存一份、互不 import。 +""" + +from __future__ import annotations + +import json +import os +import re +import time +from collections.abc import Callable +from dataclasses import dataclass, field +from pathlib import Path +from typing import Any +from uuid import uuid4 + +import httpx +from pydantic import ValidationError + +from qca import APIError, APIResponseValidationError, APIStatusError + + +def name(kind: str) -> str: + return f"sdk-python-{kind}-{uuid4().hex[:12]}" + + +def marker() -> str: + return uuid4().hex + + +def read_env(path: str | Path) -> dict[str, str]: + """Read assignments as data. Never evaluate shell expressions.""" + path = Path(path) + if not path.exists(): + return {} + values = {} + for number, line in enumerate(path.read_text().splitlines(), 1): + line = line.strip() + if not line or line.startswith("#"): + continue + line = line.removeprefix("export ") + key, sep, value = line.partition("=") + if not sep or not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", key.strip()): + raise ValueError(f"Invalid env assignment at line {number}") + value = value.strip() + if value.startswith(('"', "'")): + end = value.find(value[0], 1) + if end < 0: + raise ValueError(f"Unterminated env value at line {number}") + value = value[1:end] + else: + value = value.split(" #", 1)[0].strip() + values[key.strip()] = value + return values + + +@dataclass +class Config: + mode: str + pat: str = field(repr=False) + base_url: str = "" + model: str = "" + timeout: float = 300 + poll_interval: float = 1 + + @classmethod + def load(cls, mode: str, *, env_file: str = ".env.live", region: str | None = None, timeout: float = 300) -> Config: + values = {**read_env(env_file), **os.environ} + prefix = f"QODER_{mode.upper()}_" + token = values.get(prefix + "PAT") or values.get("QODER_PAT", "") + if not token: + raise ValueError(f"Configure {prefix}PAT or QODER_PAT") + suffix = "forward" if mode == "forward" else "cloud" + base = values.get(prefix + "BASE_URL") or f"https://api.qoder.com.cn/api/v1/{suffix}" + url = httpx.URL(base) + if ( + url.scheme != "https" + or url.host not in ("api.qoder.com", "api.qoder.com.cn") + or url.userinfo + or url.query + or url.fragment + ): + raise ValueError("Examples require an HTTPS api.qoder.com.cn or api.qoder.com URL") + if url.path.rstrip("/") != f"/api/v1/{suffix}": + raise ValueError(f"Expected the {mode} API root /api/v1/{suffix}") + if region: + url = url.copy_with(host="api.qoder.com.cn" if region == "cn" else "api.qoder.com") + if timeout <= 0: + raise ValueError("timeout must be positive") + return cls(mode, token, str(url), values.get(prefix + "MODEL", ""), timeout) + + def client_options(self) -> dict[str, Any]: + return dict(pat=self.pat, base_url=self.base_url, max_retries=0, timeout=30) + + +def safe_error(error: BaseException, token: str = "") -> str: + if isinstance(error, APIResponseValidationError): + request_id = error.response.headers.get("x-request-id") or error.response.headers.get("request-id") + text = ( + f"APIResponseValidationError: HTTP {error.status_code} {error.request.method} " + f"{error.request.url.path} request_id={request_id}" + ) + cause = error.__cause__ + if isinstance(cause, ValidationError): + # Report the failing field and rule without exposing response values. + for issue in cause.errors(include_url=False, include_input=False, include_context=False)[:5]: + path = ".".join(str(part) for part in issue["loc"]) or "response" + text += f"; {path}: {issue['type']} ({issue['msg']})" + elif isinstance(error, APIStatusError): + text = ( + f"HTTP {error.status_code} {error.request.method} {error.request.url.path} " + f"code={error.code} type={error.type} request_id={error.request_id}" + ) + body = error.body if isinstance(error.body, dict) else {} + details = body.get("error", body) + if isinstance(details, dict) and isinstance(details.get("message"), str): + text += f" message={details['message']}" + elif isinstance(error, (APIError, httpx.HTTPError)): + text = type(error).__name__ + else: + text = str(error) + if token: + text = text.replace(token, "[REDACTED]") + return re.sub(r"https?://[^\s\"'<>]+", "[URL REDACTED]", text)[:1600] + + +class Run: + def __init__(self, config: Config, *, verbose: bool = False) -> None: + self.config = config + self.verbose = verbose + self.outputs: list[dict[str, Any]] = [] + self.deadline = time.monotonic() + config.timeout + self.cleanups: list[tuple[str, str, Callable[[], Any]]] = [] + + def output(self, label: str, value: Any) -> None: + self.outputs.append({"label": label, "value": value}) + if self.verbose: + text = value if isinstance(value, str) else json.dumps(value, ensure_ascii=False, indent=2) + separator = "\n" if "\n" in text else " " + print(f"{label}:{separator}{text}", flush=True) + + def track(self, kind: str, resource_id: str, cleanup: Callable[[], Any]) -> str: + if not resource_id: + raise AssertionError(f"Created {kind} has no ID") + self.cleanups.append((kind, resource_id, cleanup)) + self.output(f"{kind}_id", resource_id) + return resource_id + + def remaining(self) -> float: + remaining = self.deadline - time.monotonic() + if remaining <= 0: + raise TimeoutError("Scenario deadline exceeded") + return remaining + + def pause(self) -> None: + time.sleep(min(self.remaining(), self.config.poll_interval)) + + def cleanup(self) -> None: + had_resources = bool(self.cleanups) + failures = [] + for kind, resource_id, action in reversed(self.cleanups): + # Each resource gets its own cleanup budget even if the scenario timed out. + self.deadline = time.monotonic() + 60 + try: + action() + except Exception as error: + failures.append(f"{kind} {resource_id}: {safe_error(error, self.config.pat)}") + self.cleanups.clear() + if failures: + raise RuntimeError("Cleanup failed:\n" + "\n".join(failures)) + if had_resources: + self.output("cleanup", "completed") + + +def choose_model(models: Any, requested: str) -> str: + enabled = [model.id for model in models.data if model.is_enabled and model.id] + if requested: + if requested not in enabled: + raise AssertionError("Configured model is not enabled for this account") + return requested + if not enabled: + raise AssertionError("Account has no enabled models") + return "ultimate" if "ultimate" in enabled else sorted(enabled)[0] diff --git a/tests/support/memory.py b/tests/support/memory.py new file mode 100644 index 0000000..85b98c1 --- /dev/null +++ b/tests/support/memory.py @@ -0,0 +1,34 @@ +"""测试专用的项目记忆夹具（ProjectMemory）。 + +从 examples/common/live.py 净移出——examples 侧不再保留。 +""" + +from __future__ import annotations + +import random +from dataclasses import dataclass, field + +from tests.support.harness import marker + + +@dataclass +class ProjectMemory: + project: str = field(default_factory=lambda: "青禾订单-" + marker()[:6]) + release_time: str = field(default_factory=lambda: f"{random.randrange(20, 24):02}:{random.randrange(60):02}") + contact: str = field(default_factory=lambda: random.choice(["林岚", "陈朔", "叶澄", "苏棠"])) + rollback_version: str = field( + default_factory=lambda: f"v2.{random.randrange(100, 1000)}.{random.randrange(100, 1000)}" + ) + path = "projects/release-conventions.md" + + def content(self) -> str: + return f"---\nname: release-conventions\ndescription: {self.project} 的项目发布约定\nmetadata:\n type: project\n---\n\n# {self.project}\n\n- 北京时间 {self.release_time} 开始发布。\n- 发布异常时联系值班负责人{self.contact}。\n- 回滚使用已验证的稳定版本 {self.rollback_version}。\n\nWhy: 在值班窗口发布，并使用验证过的版本恢复服务。\nHow to apply: 为这个项目拟定发布计划时遵循以上约定。\n" + + def index(self) -> str: + return f"- [{self.project} 发布约定]({self.path}) — 发布窗口、异常联系人与回滚约定。\n" + + def prompt(self) -> str: + return f"请根据你记得的项目约定，为「{self.project}」拟一份简短上线安排，涵盖开始时间、异常联系和回滚处理。不要执行发布；缺少信息时请明确说明。" + + def expected(self) -> list[str]: + return [self.release_time, self.contact, self.rollback_version] diff --git a/tests/support/scenarios/__init__.py b/tests/support/scenarios/__init__.py new file mode 100644 index 0000000..e3d8b0b --- /dev/null +++ b/tests/support/scenarios/__init__.py @@ -0,0 +1 @@ +"""逐字迁自 examples/{forward,managed} 的综合断言场景。""" diff --git a/tests/support/scenarios/forward.py b/tests/support/scenarios/forward.py new file mode 100644 index 0000000..9f4326a --- /dev/null +++ b/tests/support/scenarios/forward.py @@ -0,0 +1,368 @@ +"""Forward 综合断言场景（逐字迁自 examples/forward/*.py）。 + +各场景与 batch_rows 保持原样，仅将对 examples.common.live / examples.forward._cleanup +的 import 改指 tests.support.*；SCENARIOS 顺序与来源一致。 +""" + +from __future__ import annotations + +import json +from typing import Any + +import httpx + +from qca import Forward +from tests.support.assertions import TurnResult, turn, wait_reply +from tests.support.cleanup import finish_session_forward as finish_session +from tests.support.harness import Run, choose_model, marker, name +from tests.support.memory import ProjectMemory + + +def models(client: Forward, context: Run) -> None: + models = client.models.list() + context.output("models", [{"id": model.id, "is_enabled": model.is_enabled} for model in models.data]) + context.output("selected_model", choose_model(models, context.config.model)) + + +def session(client: Forward, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户") + identity_id = context.track("identity", identity.id, lambda: client.identities.delete(identity.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + template = client.templates.create( + name=name("template"), + environment_id=environment_id, + model=model, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) + + session = client.sessions.create( + identity_id=identity_id, + template_id=template_id, + ) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + expected = marker() + prompt = "请用一句话介绍你能提供什么帮助，并在末尾原样附上：" + expected + context.output("user", prompt) + sent = client.sessions.events.send( + session_id, + events=[ + { + "type": "user.message", + "content": [{"type": "text", "text": prompt}], + } + ], + extra_headers={"Idempotency-Key": name("event")}, + ) + if len(sent.data) != 1 or not sent.data[0].id: + raise AssertionError("Send must return exactly one user event ID") + + reply = TurnResult(last_id=sent.data[0].id) + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": sent.data[0].id}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + reply.observe(event) + if reply.complete: + break + context.output("assistant", reply.text) + reply.verify([expected]) + + +def resources(client: Forward, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户") + identity_id = context.track("identity", identity.id, lambda: client.identities.delete(identity.id)) + + file_value, env_value, skill_value = marker(), marker(), marker() + file = client.files.upload(file=("sdk-example.txt", file_value.encode()), purpose="session_resource") + context.track("file", file.id, lambda: client.files.delete(file.id)) + skill_name = name("skill") + skill = client.skills.create( + files=[ + ( + f"{skill_name}/SKILL.md", + f"---\nname: {skill_name}\ndescription: SDK example verification code.\n---\nEXAMPLE_SKILL_CODE: {skill_value}\n".encode(), + ) + ] + ) + context.track("skill", skill.id, lambda: client.skills.delete(skill.id)) + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + template = client.templates.create( + name=name("template"), + environment_id=environment_id, + model=model, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + skills=[{"type": "custom", "skill_id": skill.id, "version": skill.latest_version}], + environment_variables={"SDK_EXAMPLE_VALUE": "template-default"}, + ) + template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) + + client.identities.configs.upsert( + template_id, + identity_id=identity_id, + identity_config={"environment_variables": {"SDK_EXAMPLE_VALUE": {"op": "set", "value": env_value}}}, + ) + session = client.sessions.create( + identity_id=identity_id, + template_id=template_id, + resources=[{"type": "file", "file_id": file.id, "mount_path": "/data/workspace/sdk-example.txt"}], + ) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + turn( + client.sessions.events, + context, + session_id, + "请使用工具读取 /data/workspace/sdk-example.txt 和 SDK_EXAMPLE_VALUE 环境变量，返回两个值。", + [file_value, env_value], + require_tool=True, + ) + turn( + client.sessions.events, + context, + session_id, + f"请使用技能 {skill_name}，读取并返回 EXAMPLE_SKILL_CODE。", + [skill_value], + require_tool=True, + ) + + +def memory(client: Forward, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户") + + def cleanup_identity() -> None: + result = client.identities.clear(identity.id, reason="SDK example cleanup") + if result.status != "completed": + raise AssertionError("Dedicated Identity cleanup did not complete") + client.identities.delete(identity.id) + + identity_id = context.track("identity", identity.id, cleanup_identity) + + memory = ProjectMemory() + store = client.memory_stores.create(name=name("memory"), idempotency_key=name("memory-key")) + context.track("memory_store", store.id, lambda: client.memory_stores.delete(store.id)) + for path, content in ((memory.path, memory.content()), ("MEMORY.md", memory.index())): + entry = client.memory_stores.memories.create(store.id, path=path, content=content) + saved = client.memory_stores.memories.retrieve(entry.id, memory_store_id=store.id) + if saved.content != content: + raise AssertionError("Persisted memory does not match the supplied content") + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + template = client.templates.create( + name=name("template"), + environment_id=environment_id, + model=model, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) + + client.identities.memory_stores.mount(template_id, identity_id=identity_id, memory_store_id=store.id) + context.track( + "memory_mount", + store.id, + lambda: client.identities.memory_stores.detach(store.id, template_id=template_id, identity_id=identity_id), + ) + mounts = client.identities.memory_stores.list(template_id, identity_id=identity_id) + if not any(mount.memory_store_id == store.id for mount in mounts.data): + raise AssertionError("Memory Store binding was not persisted") + session = client.sessions.create( + identity_id=identity_id, + template_id=template_id, + ) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + turn(client.sessions.events, context, session_id, memory.prompt(), memory.expected()) + + +def schedule(client: Forward, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户") + identity_id = context.track("identity", identity.id, lambda: client.identities.delete(identity.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + template = client.templates.create( + name=name("template"), + environment_id=environment_id, + model=model, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) + + expected = marker() + schedule = client.schedules.create( + identity_id=identity_id, + template_id=template_id, + environment_id=environment_id, + name=name("schedule"), + initial_events=[{"type": "user.message", "content": "Reply with exactly " + expected}], + trigger_policy={"type": "manual"}, + execution={"max_attempts": 1, "max_concurrent_runs": 1}, + ) + context.track("schedule", schedule.id, lambda: client.schedules.archive(schedule.id)) + execution = client.schedules.run(schedule.id, idempotency_key=name("run")) + + def cleanup_run() -> None: + while True: + current = client.schedule_runs.retrieve(execution.id, identity_id=identity_id) + if current.session_id: + finish_session(client, context, current.session_id) + return + if current.status in ("failed", "skipped"): + return + context.pause() + + context.track("schedule_run", execution.id, cleanup_run) + while True: + current = client.schedule_runs.retrieve(execution.id, identity_id=identity_id) + if current.status == "completed": + break + if current.status in ("failed", "skipped"): + raise AssertionError(f"Schedule Run failed: {current.status}") + context.pause() + if not current.session_id: + raise AssertionError("Completed Schedule Run has no session") + context.output("session_id", current.session_id) + wait_reply(client.sessions.events, context, current.session_id).verify([expected]) + + +def batch(client: Forward, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户") + identity_id = context.track("identity", identity.id, lambda: client.identities.delete(identity.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + template = client.templates.create( + name=name("template"), + environment_id=environment_id, + model=model, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) + + expected, custom_id = marker(), name("task") + data = { + "custom_id": custom_id, + "template_id": template_id, + "identity_id": identity_id, + "body": {"input": "Reply with exactly " + expected}, + } + input_file = client.files.upload( + file=("input.jsonl", (json.dumps(data) + "\n").encode()), purpose="session_resource" + ) + context.track("input_file", input_file.id, lambda: client.files.delete(input_file.id)) + batch = client.batches.create( + input_file_id=input_file.id, + completion_window="24h", + idempotency_key=name("batch"), + extra_body={"ignore_idle_window": True}, + ) + terminal = {"completed", "failed", "cancelled", "expired"} + + def cleanup_batch() -> None: + current = client.batches.retrieve(batch.id) + if current.status not in terminal: + client.batches.cancel(batch.id) + while current.status not in terminal: + context.pause() + current = client.batches.retrieve(batch.id) + if not current.output_file_id: + if current.request_counts and current.request_counts.total == 0: + return + raise AssertionError("Batch has no output for session cleanup") + rows = batch_rows(client, batch.id) + if ( + len(rows) != 1 + or rows[0].get("custom_id") != custom_id + or rows[0].get("identity_id") != identity_id + or rows[0].get("template_id") != template_id + ): + raise AssertionError("Batch cleanup output does not match this run") + if rows[0].get("session_id"): + finish_session(client, context, rows[0]["session_id"]) + + context.track("batch", batch.id, cleanup_batch) + while batch.status not in terminal: + context.pause() + batch = client.batches.retrieve(batch.id) + if ( + batch.status != "completed" + or not batch.request_counts + or batch.request_counts.completed != 1 + or batch.request_counts.failed != 0 + or not batch.output_file_id + ): + raise AssertionError("Batch did not complete exactly one successful task") + tasks = client.batches.tasks.list(batch.id) + if len(tasks.data) != 1 or tasks.data[0].custom_id != custom_id: + raise AssertionError("Batch task did not round trip") + rows = batch_rows(client, batch.id) + if len(rows) != 1: + raise AssertionError("Expected one Batch output row") + row = rows[0] + context.output("batch_output", row) + if ( + row.get("custom_id") != custom_id + or row.get("identity_id") != identity_id + or row.get("template_id") != template_id + or row.get("status") != "completed" + or row.get("error") + or not row.get("session_id") + ): + raise AssertionError("Batch output ownership or status mismatch") + if expected not in json.dumps(row.get("response")): + raise AssertionError("Batch response does not contain expected output") + wait_reply(client.sessions.events, context, row["session_id"]).verify([expected]) + + +def batch_rows(client: Forward, batch_id: str) -> list[dict[str, Any]]: + link = client.batches.retrieve_output(batch_id) + url = httpx.URL(link.url) + if url.scheme not in ("http", "https") or not url.host or url.userinfo: + raise AssertionError("Invalid Batch output URL") + # A separate HTTP client prevents API credentials from reaching storage. + with httpx.Client(timeout=30, follow_redirects=True) as download: + with download.stream("GET", url) as response: + response.raise_for_status() + content = bytearray() + for chunk in response.iter_bytes(): + content.extend(chunk) + if len(content) > 4 * 1024 * 1024: + raise AssertionError("Batch output exceeds the example's 4 MiB limit") + return [json.loads(line) for line in content.splitlines() if line.strip()] + + +SCENARIOS = { + "models": models, + "session": session, + "resources": resources, + "memory": memory, + "schedule": schedule, + "batch": batch, +} diff --git a/tests/support/scenarios/managed.py b/tests/support/scenarios/managed.py new file mode 100644 index 0000000..ea8e77e --- /dev/null +++ b/tests/support/scenarios/managed.py @@ -0,0 +1,241 @@ +"""Managed 综合断言场景（逐字迁自 examples/managed/*.py）。 + +各场景保持原样，仅将对 examples.common.live / examples.managed._cleanup / examples.managed.dream +的 import 改指 tests.support.*；finish_dream 迁至 tests.support.cleanup；SCENARIOS 顺序与来源一致。 +""" + +from __future__ import annotations + +from qca import Managed +from tests.support.assertions import TurnResult, turn, wait_reply +from tests.support.cleanup import finish_dream +from tests.support.cleanup import finish_session_managed as finish_session +from tests.support.harness import Run, choose_model, marker, name +from tests.support.memory import ProjectMemory + + +def models(client: Managed, context: Run) -> None: + models = client.models.list() + context.output("models", [{"id": model.id, "is_enabled": model.is_enabled} for model in models.data]) + context.output("selected_model", choose_model(models, context.config.model)) + + +def session(client: Managed, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + agent = client.agents.create( + name=name("agent"), + model={"id": model}, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + agent_id = context.track("agent", agent.id, lambda: client.agents.archive(agent.id)) + + session = client.sessions.create( + environment_id=environment_id, + agent=agent_id, + ) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + expected = marker() + prompt = "请用一句话介绍你能提供什么帮助，并在末尾原样附上：" + expected + context.output("user", prompt) + sent = client.sessions.events.send( + session_id, + events=[ + { + "type": "user.message", + "content": [{"type": "text", "text": prompt}], + } + ], + extra_headers={"Idempotency-Key": name("event")}, + ) + if len(sent.data) != 1 or not sent.data[0].id: + raise AssertionError("Send must return exactly one user event ID") + + reply = TurnResult(last_id=sent.data[0].id) + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": sent.data[0].id}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + reply.observe(event) + if reply.complete: + break + context.output("assistant", reply.text) + reply.verify([expected]) + + +def resources(client: Managed, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + file_value, env_value, skill_value = marker(), marker(), marker() + file = client.files.upload(file=("sdk-example.txt", file_value.encode())) + context.track("file", file.id, lambda: client.files.delete(file.id)) + skill_name = name("skill") + skill = client.skills.create( + files=[ + ( + f"{skill_name}/SKILL.md", + f"---\nname: {skill_name}\ndescription: SDK example verification code.\n---\nEXAMPLE_SKILL_CODE: {skill_value}\n".encode(), + ) + ] + ) + context.track("skill", skill.id, lambda: client.skills.delete(skill.id)) + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + agent = client.agents.create( + name=name("agent"), + model={"id": model}, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + skills=[{"type": "custom", "skill_id": skill.id, "version": skill.latest_version}], + ) + agent_id = context.track("agent", agent.id, lambda: client.agents.archive(agent.id)) + + session = client.sessions.create( + environment_id=environment_id, + agent=agent_id, + environment_variables={"SDK_EXAMPLE_VALUE": env_value}, + resources=[{"type": "file", "file_id": file.id, "mount_path": "/data/workspace/sdk-example.txt"}], + ) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + turn( + client.sessions.events, + context, + session_id, + "请使用工具读取 /data/workspace/sdk-example.txt 和 SDK_EXAMPLE_VALUE 环境变量，返回两个值。", + [file_value, env_value], + require_tool=True, + ) + turn( + client.sessions.events, + context, + session_id, + f"请使用技能 {skill_name}，读取并返回 EXAMPLE_SKILL_CODE。", + [skill_value], + require_tool=True, + ) + + +def memory(client: Managed, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + memory = ProjectMemory() + store = client.memory_stores.create(name=name("memory")) + context.track("memory_store", store.id, lambda: client.memory_stores.delete(store.id)) + for path, content in ((memory.path, memory.content()), ("MEMORY.md", memory.index())): + entry = client.memory_stores.memories.create(store.id, path=path, content=content) + saved = client.memory_stores.memories.retrieve(entry.id, memory_store_id=store.id) + if saved.content != content: + raise AssertionError("Persisted memory does not match the supplied content") + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + agent = client.agents.create( + name=name("agent"), + model={"id": model}, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + agent_id = context.track("agent", agent.id, lambda: client.agents.archive(agent.id)) + + session = client.sessions.create( + environment_id=environment_id, + agent=agent_id, + resources=[{"type": "memory_store", "memory_store_id": store.id, "access": "read_only"}], + ) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + if not any( + resource.type == "memory_store" and resource.memory_store_id == store.id + for resource in client.sessions.resources.list(session_id) + ): + raise AssertionError("Memory Store is not mounted in the new session") + turn(client.sessions.events, context, session_id, memory.prompt(), memory.expected()) + + +def deployment(client: Managed, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + agent = client.agents.create( + name=name("agent"), + model={"id": model}, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + agent_id = context.track("agent", agent.id, lambda: client.agents.archive(agent.id)) + + expected = marker() + deployment = client.deployments.create( + name=name("deployment"), + environment_id=environment_id, + agent=agent_id, + initial_events=[ + {"type": "user.message", "content": [{"type": "text", "text": "Reply with exactly " + expected}]} + ], + ) + context.track("deployment", deployment.id, lambda: client.deployments.archive(deployment.id)) + execution = client.deployments.run(deployment.id) + if not execution.session_id: + raise AssertionError("Deployment Run returned no session") + context.track("session", execution.session_id, lambda: finish_session(client, context, execution.session_id)) + saved = client.deployment_runs.retrieve(execution.id) + if saved.session_id != execution.session_id: + raise AssertionError("Deployment Run session ID changed") + wait_reply(client.sessions.events, context, execution.session_id).verify([expected]) + + +def dream(client: Managed, context: Run) -> None: + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + store = client.memory_stores.create(name=name("dream-input")) + context.track("input_memory_store", store.id, lambda: client.memory_stores.delete(store.id)) + expected = marker() + client.memory_stores.memories.create( + store.id, + path="sdk-example/source.md", + content=f"Permanent project verification code: {expected}. Preserve this exact code during consolidation.", + ) + dream = client.dreams.create( + inputs=[{"type": "memory_store", "memory_store_id": store.id}], + model=model, + instructions="Consolidate supplied memory into sdk-example/consolidated.md. Preserve the exact project verification code. Keep the original source.", + ) + dream_id = dream.id + context.track("dream", dream_id, lambda: finish_dream(client, context, dream_id, store.id)) + while dream.status in ("pending", "running"): + context.pause() + dream = client.dreams.retrieve(dream_id) + if dream.status != "completed" or not dream.outputs: + raise AssertionError(f"Dream did not complete: {dream.status}") + for output in dream.outputs: + for memory in client.memory_stores.memories.list(output.memory_store_id): + if memory.path == "sdk-example/consolidated.md": + saved = client.memory_stores.memories.retrieve(memory.id, memory_store_id=output.memory_store_id) + context.output("output_memory_store_id", output.memory_store_id) + context.output("memory_path", saved.path) + context.output("memory_content", saved.content) + if saved.content and expected in saved.content: + return + raise AssertionError("Dream did not persist consolidated memory with the original value") + + +SCENARIOS = { + "models": models, + "session": session, + "resources": resources, + "memory": memory, + "deployment": deployment, + "dream": dream, +} diff --git a/tests/support/stub.py b/tests/support/stub.py new file mode 100644 index 0000000..e886525 --- /dev/null +++ b/tests/support/stub.py @@ -0,0 +1,229 @@ +"""离线场景回放使用的 HTTP 桩（逐字迁自 tests/test_examples.py 的 ExampleService）。 + +ExampleService 的内部契约断言（挂载路径 / Idempotency-Key / skill 版本 / +ignore_idle_window / MEMORY.md 索引 / override op 等）保持原样；patch_batch_rows 把 batch +场景的输出下载改指桩数据，确保离线回放不发真实网络请求。 +""" + +from __future__ import annotations + +import json +from email.parser import BytesParser +from email.policy import default + +import httpx + +from tests.support.scenarios import forward as forward_scenarios + + +class ExampleService: + """A stateful HTTP stub: replies are derived from uploaded resources and memory.""" + + def __init__(self, mode): + self.mode = mode + self.objects = {} + self.contents = {} + self.events = {} + self.mounts = {} + self.overrides = {} + self.outputs = {} + self.counter = 0 + self.deleted = [] + + def create(self, kind, body): + self.counter += 1 + item = {**body, "id": f"{kind}-{self.counter}"} + self.objects[item["id"]] = item + return item + + def session(self, body): + session = self.create("session", {**body, "status": "idle"}) + self.events[session["id"]] = [] + for event in body.get("initial_events", []): + self.send(session["id"], event) + return session + + def send(self, session_id, event): + session = self.objects[session_id] + user_event = self.create("event", event) + self.events[session_id].append(user_event) + content = event.get("content", []) + prompt = content if isinstance(content, str) else "".join(block.get("text", "") for block in content) + text = prompt + agent = self.objects[session.get("template_id") or session.get("agent")] + if "SDK_EXAMPLE_VALUE" in prompt: + resource = next(r for r in session["resources"] if r["type"] == "file") + assert resource["mount_path"] == "/data/workspace/sdk-example.txt" + text = self.contents[resource["file_id"]].decode() + if self.mode == "forward": + override = self.overrides[session["identity_id"], session["template_id"]] + assert override["op"] == "set" + text += " " + override["value"] + else: + text += " " + session["environment_variables"]["SDK_EXAMPLE_VALUE"] + elif "EXAMPLE_SKILL_CODE" in prompt: + skill = agent["skills"][0] + text = self.contents[skill["skill_id"]].decode() + elif "项目约定" in prompt: + store_id = ( + self.mounts[session["identity_id"], session["template_id"]] + if self.mode == "forward" + else next(r["memory_store_id"] for r in session["resources"] if r["type"] == "memory_store") + ) + text = "\n".join(item["content"] for item in self.objects.values() if item.get("store_id") == store_id) + assert "MEMORY.md" in [ + item.get("path") for item in self.objects.values() if item.get("store_id") == store_id + ] + events = [ + {"type": "agent.tool_use", "name": "Read", "evaluated_permission": "allow"}, + {"type": "agent.message", "content": [{"type": "text", "text": text}]}, + {"type": "session.status_idle", "stop_reason": {"type": "end_turn"}}, + ] + self.events[session_id].extend(self.create("event", item) for item in events) + return user_event + + def __call__(self, request): + path = request.url.path.split("/api/v1/", 1)[1].split("/", 1)[1] + parts = path.split("/") + verb = request.method + body = {} + if request.content: + if request.headers.get("content-type", "").startswith("multipart/"): + message = BytesParser(policy=default).parsebytes( + b"Content-Type: " + request.headers["content-type"].encode() + b"\r\n\r\n" + request.content + ) + body = { + part.get_param("name", header="content-disposition"): part.get_payload(decode=True) + for part in message.iter_parts() + } + else: + body = json.loads(request.content) + + def reply(data): + if isinstance(data, dict) and isinstance(data.get("agent"), str): + data = {**data, "agent": {"id": data["agent"]}} + return httpx.Response(200, json=data) + + if path == "models": + return reply({"data": [{"id": "ultimate", "is_enabled": True}]}) + if parts[0] == "sessions" and len(parts) >= 3: + session_id = parts[1] + if parts[2] == "events": + if verb == "POST": + return reply({"data": [self.send(session_id, e) for e in body["events"]]}) + events = self.events[session_id] + after = request.headers.get("Last-Event-ID") or request.url.params.get("after_id") + if after: + events = events[next(i for i, e in enumerate(events) if e["id"] == after) + 1 :] + if parts[-1] == "stream": + return httpx.Response( + 200, + text="".join(f"id: {e['id']}\nevent: {e['type']}\ndata: {json.dumps(e)}\n\n" for e in events), + headers={"content-type": "text/event-stream"}, + ) + return reply({"data": events, "has_more": False}) + if parts[2] == "resources": + return reply({"data": self.objects[session_id]["resources"], "has_more": False}) + if parts[0] == "memory_stores" and len(parts) >= 3 and parts[2] == "memories": + store_id = parts[1] + if verb == "POST": + return reply(self.create("memory", {**body, "store_id": store_id})) + if len(parts) == 4: + assert self.objects[parts[3]]["store_id"] == store_id + return reply(self.objects[parts[3]]) + return reply( + { + "data": [item for item in self.objects.values() if item.get("store_id") == store_id], + "has_more": False, + } + ) + if parts[0] == "identities" and len(parts) >= 5: + identity, template = parts[1], parts[3] + if parts[4] == "config": + self.overrides[identity, template] = body["identity_config"]["environment_variables"][ + "SDK_EXAMPLE_VALUE" + ] + return reply({}) + if parts[4] == "memory_stores": + if verb == "POST": + self.mounts[identity, template] = body["memory_store_id"] + return reply({}) + if verb == "DELETE": + del self.mounts[identity, template] + return reply({}) + return reply({"data": [{"memory_store_id": self.mounts[identity, template]}]}) + if len(parts) == 1 and verb == "POST": + kind = parts[0] + if self.mode == "forward" and kind == "memory_stores": + assert request.headers.get("Idempotency-Key"), "Idempotency-Key header is required" + if kind in ("files", "skills"): + item = self.create(kind, {"latest_version": "v1"} if kind == "skills" else {}) + self.contents[item["id"]] = body["file" if kind == "files" else "files"] + elif kind in ("agents", "templates"): + for skill in body.get("skills", []): + assert skill["version"] == self.objects[skill["skill_id"]]["latest_version"] + item = self.create(kind, body) + elif kind == "sessions": + item = self.session(body) + elif kind == "dreams": + store_id = body["inputs"][0]["memory_store_id"] + source = next(item["content"] for item in self.objects.values() if item.get("store_id") == store_id) + output = self.create("memory_stores", {}) + self.create( + "memory", {"store_id": output["id"], "path": "sdk-example/consolidated.md", "content": source} + ) + item = self.create( + "dream", + {"status": "completed", "outputs": [{"type": "memory_store", "memory_store_id": output["id"]}]}, + ) + elif kind == "batches": + assert body["ignore_idle_window"] is True + task = json.loads(self.contents[body["input_file_id"]]) + session = self.session( + { + "identity_id": task["identity_id"], + "template_id": task["template_id"], + "initial_events": [{"type": "user.message", "content": task["body"]["input"]}], + } + ) + item = self.create( + "batch", + { + "status": "completed", + "output_file_id": "output", + "request_counts": {"total": 1, "completed": 1, "failed": 0}, + }, + ) + self.outputs[item["id"]] = [ + { + "custom_id": task["custom_id"], + "identity_id": task["identity_id"], + "template_id": task["template_id"], + "session_id": session["id"], + "status": "completed", + "response": task["body"]["input"], + } + ] + else: + item = self.create(kind, body) + return reply(item) + if len(parts) == 3 and parts[-1] == "run": + item = self.objects[parts[1]] + session = self.session(item) + execution = self.create("run", {"session_id": session["id"], "status": "completed"}) + return reply(execution) + if len(parts) == 3 and parts[-1] == "tasks": + return reply({"data": [{"custom_id": row["custom_id"]} for row in self.outputs[parts[1]]]}) + if len(parts) == 3 and parts[-1] == "clear": + return reply({"status": "completed"}) + if verb == "DELETE" or (len(parts) == 3 and parts[-1] == "archive"): + self.deleted.append(parts[1]) + return reply({}) + if len(parts) == 2 and verb == "GET": + return reply(self.objects[parts[1]]) + raise AssertionError(f"Unexpected request: {verb} {path}") + + +def patch_batch_rows(monkeypatch, service): + """离线回放时替换 batch 输出下载，避免真实网络请求。""" + monkeypatch.setattr(forward_scenarios, "batch_rows", lambda client, batch_id: service.outputs[batch_id]) diff --git a/tests/test_examples.py b/tests/test_examples.py deleted file mode 100644 index 3eafa6f..0000000 --- a/tests/test_examples.py +++ /dev/null @@ -1,506 +0,0 @@ -from __future__ import annotations - -import json -import sys -from email.parser import BytesParser -from email.policy import default - -import httpx -import pytest - -from examples.common.live import Config, ProjectMemory, Run, TurnResult, choose_model, read_env, run_cli, safe_error -from examples.forward import batch -from examples.forward.__main__ import SCENARIOS as FORWARD_SCENARIOS -from examples.managed.__main__ import SCENARIOS as MANAGED_SCENARIOS -from qca import Forward, Managed - - -class ExampleService: - """A stateful HTTP stub: replies are derived from uploaded resources and memory.""" - - def __init__(self, mode): - self.mode = mode - self.objects = {} - self.contents = {} - self.events = {} - self.mounts = {} - self.overrides = {} - self.outputs = {} - self.counter = 0 - self.deleted = [] - - def create(self, kind, body): - self.counter += 1 - item = {**body, "id": f"{kind}-{self.counter}"} - self.objects[item["id"]] = item - return item - - def session(self, body): - session = self.create("session", {**body, "status": "idle"}) - self.events[session["id"]] = [] - for event in body.get("initial_events", []): - self.send(session["id"], event) - return session - - def send(self, session_id, event): - session = self.objects[session_id] - user_event = self.create("event", event) - self.events[session_id].append(user_event) - content = event.get("content", []) - prompt = content if isinstance(content, str) else "".join(block.get("text", "") for block in content) - text = prompt - agent = self.objects[session.get("template_id") or session.get("agent")] - if "SDK_EXAMPLE_VALUE" in prompt: - resource = next(r for r in session["resources"] if r["type"] == "file") - assert resource["mount_path"] == "/data/workspace/sdk-example.txt" - text = self.contents[resource["file_id"]].decode() - if self.mode == "forward": - override = self.overrides[session["identity_id"], session["template_id"]] - assert override["op"] == "set" - text += " " + override["value"] - else: - text += " " + session["environment_variables"]["SDK_EXAMPLE_VALUE"] - elif "EXAMPLE_SKILL_CODE" in prompt: - skill = agent["skills"][0] - text = self.contents[skill["skill_id"]].decode() - elif "项目约定" in prompt: - store_id = ( - self.mounts[session["identity_id"], session["template_id"]] - if self.mode == "forward" - else next(r["memory_store_id"] for r in session["resources"] if r["type"] == "memory_store") - ) - text = "\n".join(item["content"] for item in self.objects.values() if item.get("store_id") == store_id) - assert "MEMORY.md" in [ - item.get("path") for item in self.objects.values() if item.get("store_id") == store_id - ] - events = [ - {"type": "agent.tool_use", "name": "Read", "evaluated_permission": "allow"}, - {"type": "agent.message", "content": [{"type": "text", "text": text}]}, - {"type": "session.status_idle", "stop_reason": {"type": "end_turn"}}, - ] - self.events[session_id].extend(self.create("event", item) for item in events) - return user_event - - def __call__(self, request): - path = request.url.path.split("/api/v1/", 1)[1].split("/", 1)[1] - parts = path.split("/") - verb = request.method - body = {} - if request.content: - if request.headers.get("content-type", "").startswith("multipart/"): - message = BytesParser(policy=default).parsebytes( - b"Content-Type: " + request.headers["content-type"].encode() + b"\r\n\r\n" + request.content - ) - body = { - part.get_param("name", header="content-disposition"): part.get_payload(decode=True) - for part in message.iter_parts() - } - else: - body = json.loads(request.content) - - def reply(data): - if isinstance(data, dict) and isinstance(data.get("agent"), str): - data = {**data, "agent": {"id": data["agent"]}} - return httpx.Response(200, json=data) - - if path == "models": - return reply({"data": [{"id": "ultimate", "is_enabled": True}]}) - if parts[0] == "sessions" and len(parts) >= 3: - session_id = parts[1] - if parts[2] == "events": - if verb == "POST": - return reply({"data": [self.send(session_id, e) for e in body["events"]]}) - events = self.events[session_id] - after = request.headers.get("Last-Event-ID") or request.url.params.get("after_id") - if after: - events = events[next(i for i, e in enumerate(events) if e["id"] == after) + 1 :] - if parts[-1] == "stream": - return httpx.Response( - 200, - text="".join(f"id: {e['id']}\nevent: {e['type']}\ndata: {json.dumps(e)}\n\n" for e in events), - headers={"content-type": "text/event-stream"}, - ) - return reply({"data": events, "has_more": False}) - if parts[2] == "resources": - return reply({"data": self.objects[session_id]["resources"], "has_more": False}) - if parts[0] == "memory_stores" and len(parts) >= 3 and parts[2] == "memories": - store_id = parts[1] - if verb == "POST": - return reply(self.create("memory", {**body, "store_id": store_id})) - if len(parts) == 4: - assert self.objects[parts[3]]["store_id"] == store_id - return reply(self.objects[parts[3]]) - return reply( - { - "data": [item for item in self.objects.values() if item.get("store_id") == store_id], - "has_more": False, - } - ) - if parts[0] == "identities" and len(parts) >= 5: - identity, template = parts[1], parts[3] - if parts[4] == "config": - self.overrides[identity, template] = body["identity_config"]["environment_variables"][ - "SDK_EXAMPLE_VALUE" - ] - return reply({}) - if parts[4] == "memory_stores": - if verb == "POST": - self.mounts[identity, template] = body["memory_store_id"] - return reply({}) - if verb == "DELETE": - del self.mounts[identity, template] - return reply({}) - return reply({"data": [{"memory_store_id": self.mounts[identity, template]}]}) - if len(parts) == 1 and verb == "POST": - kind = parts[0] - if self.mode == "forward" and kind == "memory_stores": - assert request.headers.get("Idempotency-Key"), "Idempotency-Key header is required" - if kind in ("files", "skills"): - item = self.create(kind, {"latest_version": "v1"} if kind == "skills" else {}) - self.contents[item["id"]] = body["file" if kind == "files" else "files"] - elif kind in ("agents", "templates"): - for skill in body.get("skills", []): - assert skill["version"] == self.objects[skill["skill_id"]]["latest_version"] - item = self.create(kind, body) - elif kind == "sessions": - item = self.session(body) - elif kind == "dreams": - store_id = body["inputs"][0]["memory_store_id"] - source = next(item["content"] for item in self.objects.values() if item.get("store_id") == store_id) - output = self.create("memory_stores", {}) - self.create( - "memory", {"store_id": output["id"], "path": "sdk-example/consolidated.md", "content": source} - ) - item = self.create( - "dream", - {"status": "completed", "outputs": [{"type": "memory_store", "memory_store_id": output["id"]}]}, - ) - elif kind == "batches": - assert body["ignore_idle_window"] is True - task = json.loads(self.contents[body["input_file_id"]]) - session = self.session( - { - "identity_id": task["identity_id"], - "template_id": task["template_id"], - "initial_events": [{"type": "user.message", "content": task["body"]["input"]}], - } - ) - item = self.create( - "batch", - { - "status": "completed", - "output_file_id": "output", - "request_counts": {"total": 1, "completed": 1, "failed": 0}, - }, - ) - self.outputs[item["id"]] = [ - { - "custom_id": task["custom_id"], - "identity_id": task["identity_id"], - "template_id": task["template_id"], - "session_id": session["id"], - "status": "completed", - "response": task["body"]["input"], - } - ] - else: - item = self.create(kind, body) - return reply(item) - if len(parts) == 3 and parts[-1] == "run": - item = self.objects[parts[1]] - session = self.session(item) - execution = self.create("run", {"session_id": session["id"], "status": "completed"}) - return reply(execution) - if len(parts) == 3 and parts[-1] == "tasks": - return reply({"data": [{"custom_id": row["custom_id"]} for row in self.outputs[parts[1]]]}) - if len(parts) == 3 and parts[-1] == "clear": - return reply({"status": "completed"}) - if verb == "DELETE" or (len(parts) == 3 and parts[-1] == "archive"): - self.deleted.append(parts[1]) - return reply({}) - if len(parts) == 2 and verb == "GET": - return reply(self.objects[parts[1]]) - raise AssertionError(f"Unexpected request: {verb} {path}") - - -@pytest.mark.parametrize( - "mode,scenario", [("forward", s) for s in FORWARD_SCENARIOS] + [("managed", s) for s in MANAGED_SCENARIOS] -) -def test_go_example_scenarios_with_http_stub(mode, scenario, monkeypatch, capsys): - config = Config(mode, "test-token", f"https://api.test/api/v1/{mode}", timeout=2, poll_interval=0) - run = Run(config) - service = ExampleService(mode) - cls, scenarios = (Forward, FORWARD_SCENARIOS) if mode == "forward" else (Managed, MANAGED_SCENARIOS) - with cls(**config.client_options(), http_client=httpx.Client(transport=httpx.MockTransport(service))) as client: - if mode == "forward": - monkeypatch.setattr(batch, "batch_rows", lambda client, batch_id: service.outputs[batch_id]) - try: - scenarios[scenario](client, run) - finally: - run.cleanup() - assert not run.cleanups - assert not service.mounts - created = [key for key in service.objects if not key.startswith(("event-", "memory-", "run-", "batch-"))] - assert all(key in service.deleted for key in created) - assert capsys.readouterr().out == "" - outputs = {item["label"]: item["value"] for item in run.outputs} - assert outputs["selected_model"] == "ultimate" - if scenario == "models": - assert outputs["models"] == [{"id": "ultimate", "is_enabled": True}] - elif scenario == "dream": - assert outputs["memory_path"] == "sdk-example/consolidated.md" - assert outputs["memory_content"] - assert outputs["output_memory_store_id"] in service.objects - else: - messages = [ - event["content"][0]["text"] - for events in service.events.values() - for event in events - if event["type"] == "agent.message" - ] - assert [item["value"] for item in run.outputs if item["label"] == "assistant"] == messages - if scenario != "models": - assert outputs["cleanup"] == "completed" - - -def test_turn_assertions_require_final_assistant_output_and_successful_idle(): - result = TurnResult() - for event in [ - {"type": "user.message", "content": [{"type": "text", "text": "expected"}]}, - {"type": "agent.tool_result", "content": [{"type": "text", "text": "expected"}]}, - {"type": "event_delta", "text": "expected"}, - {"type": "session.status_idle", "stop_reason": {"type": "end_turn"}}, - ]: - result.observe(event) - with pytest.raises(AssertionError, match="No idle state"): - result.verify(["expected"]) - result.observe({"type": "agent.message", "content": [{"type": "text", "text": "expected"}]}) - result.observe({"type": "agent.message", "content": [{"type": "text", "text": "final answer"}]}) - result.observe({"type": "session.status_idle", "stop_reason": {"type": "end_turn"}}) - with pytest.raises(AssertionError, match="missing"): - result.verify(["expected"]) - with pytest.raises(AssertionError, match="No actual tool"): - result.verify(["final answer"], require_tool=True) - result.observe({"type": "agent.tool_use"}) - result.verify(["final answer"], require_tool=True) - - -@pytest.mark.parametrize("reason", ["budget_exceeded", "tool_confirmation", "max_iterations"]) -def test_turn_rejects_idle_that_needs_further_action(reason): - with pytest.raises(AssertionError, match="stopped early"): - TurnResult(text="answer").observe({"type": "session.status_idle", "stop_reason": {"type": reason}}) - - -def test_memory_facts_are_absent_from_prompt_and_index(): - memory = ProjectMemory(release_time="21:34", contact="林岚", rollback_version="v2.456.789") - for fact in memory.expected(): - assert fact in memory.content() - assert fact not in memory.prompt() - assert fact not in memory.index() - - -def test_cleanup_is_reverse_order_and_continues_after_failures(): - run = Run(Config("forward", "secret")) - calls = [] - run.track("first", "first", lambda: calls.append("first")) - - def fail(): - calls.append("second") - raise ValueError("secret https://signed.test/?token=sensitive") - - run.track("second", "second", fail) - run.track("third", "third", lambda: calls.append("third")) - with pytest.raises(RuntimeError) as caught: - run.cleanup() - assert calls == ["third", "second", "first"] - assert "secret" not in str(caught.value) - assert "sensitive" not in str(caught.value) - assert not any(item["label"] == "cleanup" for item in run.outputs) - - -def test_config_file_is_data_and_environment_wins(tmp_path, monkeypatch): - path = tmp_path / ".env" - path.write_text( - "QODER_FORWARD_PAT='file-token'\nVALUE=$(touch /tmp/never-execute)\nQODER_FORWARD_MODEL=ultimate # comment\n" - ) - assert read_env(path)["VALUE"] == "$(touch /tmp/never-execute)" - monkeypatch.setenv("QODER_FORWARD_PAT", "env-token") - monkeypatch.delenv("QODER_FORWARD_BASE_URL", raising=False) - config = Config.load("forward", env_file=str(path)) - assert config.pat == "env-token" - assert config.model == "ultimate" - assert "env-token" not in repr(config) - assert "env-token" not in safe_error(ValueError("env-token"), config.pat) - - -def test_choose_model_rejects_disabled_preference(): - from types import SimpleNamespace - - models = SimpleNamespace( - data=[SimpleNamespace(id="ultimate", is_enabled=True), SimpleNamespace(id="disabled", is_enabled=False)] - ) - assert choose_model(models, "") == "ultimate" - with pytest.raises(AssertionError): - choose_model(models, "disabled") - - -@pytest.mark.parametrize("fails", [False, True]) -def test_single_scenario_cli_defaults_to_that_scenario_and_cleans_up(fails, monkeypatch, capsys): - config = Config("forward", "test-token") - monkeypatch.setattr(Config, "load", lambda *args, **kwargs: config) - monkeypatch.setattr(sys, "argv", ["examples.forward.memory", "--output", "json"]) - http_client = httpx.Client(transport=httpx.MockTransport(lambda request: httpx.Response(200))) - client = Forward(pat=config.pat, http_client=http_client) - calls = [] - - def memory(client, context): - def cleanup(): - assert not http_client.is_closed - calls.append("cleanup") - - context.track("memory_store", "store-1", cleanup) - calls.append("memory") - if fails: - raise RuntimeError("test-token failed") - - with pytest.raises(SystemExit) as caught: - run_cli("forward", lambda **kwargs: client, {"memory": memory}) - - assert caught.value.code == (1 if fails else 0) - assert calls == ["memory", "cleanup"] - assert http_client.is_closed - output = capsys.readouterr().out - result = json.loads(output) - assert result[0]["scenario"] == "memory" - assert result[0]["passed"] is not fails - assert bool(result[0]["errors"]) is fails - assert result[0]["outputs"] == [ - {"label": "memory_store_id", "value": "store-1"}, - {"label": "cleanup", "value": "completed"}, - ] - assert "test-token" not in output - - -@pytest.mark.parametrize("mode", ["forward", "managed"]) -@pytest.mark.parametrize("scenario", ["models", "session"]) -@pytest.mark.parametrize("output_format", ["text", "json"]) -def test_cli_displays_api_results(mode, scenario, output_format, monkeypatch, capsys): - config = Config(mode, "test-token", f"https://api.test/api/v1/{mode}", timeout=2, poll_interval=0) - monkeypatch.setattr(Config, "load", lambda *args, **kwargs: config) - monkeypatch.setattr(sys, "argv", [f"examples.{mode}.{scenario}", "--output", output_format]) - service = ExampleService(mode) - cls, scenarios = (Forward, FORWARD_SCENARIOS) if mode == "forward" else (Managed, MANAGED_SCENARIOS) - - def client_type(**options): - return cls(**options, http_client=httpx.Client(transport=httpx.MockTransport(service))) - - with pytest.raises(SystemExit) as caught: - run_cli(mode, client_type, {scenario: scenarios[scenario]}) - assert caught.value.code == 0 - captured = capsys.readouterr() - assert captured.err == "" - assert "test-token" not in captured.out - if output_format == "json": - result = json.loads(captured.out) - assert len(result) == 1 - assert result[0]["scenario"] == scenario - assert result[0]["passed"] is True - values = {item["label"]: item["value"] for item in result[0]["outputs"]} - assert values["selected_model"] == "ultimate" - if scenario == "models": - assert values["models"] == [{"id": "ultimate", "is_enabled": True}] - else: - session_id = next(iter(service.events)) - assert values["session_id"] == session_id - assert values["assistant"] == service.events[session_id][-2]["content"][0]["text"] - assert values["user"] == service.events[session_id][0]["content"][0]["text"] - assert values["cleanup"] == "completed" - else: - assert captured.out.startswith(f"[{mode}.{scenario}]\n") - assert captured.out.endswith(f"{scenario}: PASS\n") - assert "selected_model: ultimate\n" in captured.out - if scenario == "models": - assert '"id": "ultimate"' in captured.out - assert '"is_enabled": true' in captured.out - else: - session_id = next(iter(service.events)) - assistant = service.events[session_id][-2]["content"][0]["text"] - prompt = service.events[session_id][0]["content"][0]["text"] - assert f"session_id: {session_id}\n" in captured.out - assert f"user: {prompt}\n" in captured.out - assert f"assistant: {assistant}\n" in captured.out - assert "cleanup: completed\n" in captured.out - - -@pytest.mark.parametrize("output_format", ["text", "json"]) -def test_cli_validation_error_identifies_field_without_response_contents(output_format, monkeypatch, capsys): - config = Config("forward", "test-token", "https://api.test/api/v1/forward") - monkeypatch.setattr(Config, "load", lambda *args, **kwargs: config) - monkeypatch.setattr(sys, "argv", ["examples.forward.models", "--output", output_format]) - - def client_type(**options): - return Forward( - **options, - http_client=httpx.Client( - transport=httpx.MockTransport( - lambda _: httpx.Response( - 200, - json={"data": [{"id": "ultimate", "is_enabled": "private-response-value"}]}, - headers={"x-request-id": "req-validation"}, - ) - ) - ), - ) - - with pytest.raises(SystemExit) as caught: - run_cli("forward", client_type, {"models": FORWARD_SCENARIOS["models"]}) - assert caught.value.code == 1 - captured = capsys.readouterr() - if output_format == "json": - result = json.loads(captured.out) - assert result[0]["passed"] is False - error = result[0]["errors"][0] - else: - assert "models: FAIL" in captured.out - error = captured.err - assert "APIResponseValidationError" in error - assert "HTTP 200 GET /api/v1/forward/models" in error - assert "request_id=req-validation" in error - assert "data.0.is_enabled" in error - assert "bool_parsing" in error - assert "private-response-value" not in captured.out + captured.err - assert "test-token" not in captured.out + captured.err - - -def test_status_error_shows_server_reason_and_redacts_token_and_signed_url(): - from qca import BadRequestError - - with Forward( - pat="test-token", - base_url="https://api.test/api/v1/forward", - http_client=httpx.Client( - transport=httpx.MockTransport( - lambda _: httpx.Response( - 400, - json={ - "error": { - "type": "invalid_request_error", - "message": "Idempotency-Key header is required; test-token https://signed.test/?secret=value", - }, - "private": "unrelated-response-value", - }, - headers={"x-request-id": "req-missing-key"}, - ) - ) - ), - ) as client: - with pytest.raises(BadRequestError) as caught: - client.memory_stores.create(name="memory", idempotency_key="key") - error = safe_error(caught.value, "test-token") - assert "HTTP 400 POST /api/v1/forward/memory_stores" in error - assert "request_id=req-missing-key" in error - assert "Idempotency-Key header is required" in error - assert "test-token" not in error - assert "signed.test" not in error - assert "secret=value" not in error - assert "unrelated-response-value" not in error diff --git a/tests/test_scenarios_offline.py b/tests/test_scenarios_offline.py new file mode 100644 index 0000000..beb2f03 --- /dev/null +++ b/tests/test_scenarios_offline.py @@ -0,0 +1,195 @@ +from __future__ import annotations + +import httpx +import pytest + +from qca import Forward, Managed +from tests.support.assertions import TurnResult +from tests.support.harness import Config, Run, choose_model, read_env, safe_error +from tests.support.memory import ProjectMemory +from tests.support.scenarios.forward import SCENARIOS as FORWARD_SCENARIOS +from tests.support.scenarios.managed import SCENARIOS as MANAGED_SCENARIOS +from tests.support.stub import ExampleService, patch_batch_rows + + +@pytest.mark.parametrize( + "mode,scenario", [("forward", s) for s in FORWARD_SCENARIOS] + [("managed", s) for s in MANAGED_SCENARIOS] +) +def test_go_example_scenarios_with_http_stub(mode, scenario, monkeypatch, capsys): + config = Config(mode, "test-token", f"https://api.test/api/v1/{mode}", timeout=2, poll_interval=0) + run = Run(config) + service = ExampleService(mode) + cls, scenarios = (Forward, FORWARD_SCENARIOS) if mode == "forward" else (Managed, MANAGED_SCENARIOS) + with cls(**config.client_options(), http_client=httpx.Client(transport=httpx.MockTransport(service))) as client: + if mode == "forward": + patch_batch_rows(monkeypatch, service) + try: + scenarios[scenario](client, run) + finally: + run.cleanup() + assert not run.cleanups + assert not service.mounts + created = [key for key in service.objects if not key.startswith(("event-", "memory-", "run-", "batch-"))] + assert all(key in service.deleted for key in created) + assert capsys.readouterr().out == "" + outputs = {item["label"]: item["value"] for item in run.outputs} + assert outputs["selected_model"] == "ultimate" + if scenario == "models": + assert outputs["models"] == [{"id": "ultimate", "is_enabled": True}] + elif scenario == "dream": + assert outputs["memory_path"] == "sdk-example/consolidated.md" + assert outputs["memory_content"] + assert outputs["output_memory_store_id"] in service.objects + else: + messages = [ + event["content"][0]["text"] + for events in service.events.values() + for event in events + if event["type"] == "agent.message" + ] + assert [item["value"] for item in run.outputs if item["label"] == "assistant"] == messages + if scenario != "models": + assert outputs["cleanup"] == "completed" + + +def test_turn_assertions_require_final_assistant_output_and_successful_idle(): + result = TurnResult() + for event in [ + {"type": "user.message", "content": [{"type": "text", "text": "expected"}]}, + {"type": "agent.tool_result", "content": [{"type": "text", "text": "expected"}]}, + {"type": "event_delta", "text": "expected"}, + {"type": "session.status_idle", "stop_reason": {"type": "end_turn"}}, + ]: + result.observe(event) + with pytest.raises(AssertionError, match="No idle state"): + result.verify(["expected"]) + result.observe({"type": "agent.message", "content": [{"type": "text", "text": "expected"}]}) + result.observe({"type": "agent.message", "content": [{"type": "text", "text": "final answer"}]}) + result.observe({"type": "session.status_idle", "stop_reason": {"type": "end_turn"}}) + with pytest.raises(AssertionError, match="missing"): + result.verify(["expected"]) + with pytest.raises(AssertionError, match="No actual tool"): + result.verify(["final answer"], require_tool=True) + result.observe({"type": "agent.tool_use"}) + result.verify(["final answer"], require_tool=True) + + +@pytest.mark.parametrize("reason", ["budget_exceeded", "tool_confirmation", "max_iterations"]) +def test_turn_rejects_idle_that_needs_further_action(reason): + with pytest.raises(AssertionError, match="stopped early"): + TurnResult(text="answer").observe({"type": "session.status_idle", "stop_reason": {"type": reason}}) + + +def test_memory_facts_are_absent_from_prompt_and_index(): + memory = ProjectMemory(release_time="21:34", contact="林岚", rollback_version="v2.456.789") + for fact in memory.expected(): + assert fact in memory.content() + assert fact not in memory.prompt() + assert fact not in memory.index() + + +def test_cleanup_is_reverse_order_and_continues_after_failures(): + run = Run(Config("forward", "secret")) + calls = [] + run.track("first", "first", lambda: calls.append("first")) + + def fail(): + calls.append("second") + raise ValueError("secret https://signed.test/?token=sensitive") + + run.track("second", "second", fail) + run.track("third", "third", lambda: calls.append("third")) + with pytest.raises(RuntimeError) as caught: + run.cleanup() + assert calls == ["third", "second", "first"] + assert "secret" not in str(caught.value) + assert "sensitive" not in str(caught.value) + assert not any(item["label"] == "cleanup" for item in run.outputs) + + +def test_config_file_is_data_and_environment_wins(tmp_path, monkeypatch): + path = tmp_path / ".env" + path.write_text( + "QODER_FORWARD_PAT='file-token'\nVALUE=$(touch /tmp/never-execute)\nQODER_FORWARD_MODEL=ultimate # comment\n" + ) + assert read_env(path)["VALUE"] == "$(touch /tmp/never-execute)" + monkeypatch.setenv("QODER_FORWARD_PAT", "env-token") + monkeypatch.delenv("QODER_FORWARD_BASE_URL", raising=False) + config = Config.load("forward", env_file=str(path)) + assert config.pat == "env-token" + assert config.model == "ultimate" + assert "env-token" not in repr(config) + assert "env-token" not in safe_error(ValueError("env-token"), config.pat) + + +def test_choose_model_rejects_disabled_preference(): + from types import SimpleNamespace + + models = SimpleNamespace( + data=[SimpleNamespace(id="ultimate", is_enabled=True), SimpleNamespace(id="disabled", is_enabled=False)] + ) + assert choose_model(models, "") == "ultimate" + with pytest.raises(AssertionError): + choose_model(models, "disabled") + + +def test_validation_error_identifies_field_without_response_contents(): + from qca import APIResponseValidationError + + with Forward( + pat="test-token", + base_url="https://api.test/api/v1/forward", + http_client=httpx.Client( + transport=httpx.MockTransport( + lambda _: httpx.Response( + 200, + json={"data": [{"id": "ultimate", "is_enabled": "private-response-value"}]}, + headers={"x-request-id": "req-validation"}, + ) + ) + ), + ) as client: + with pytest.raises(APIResponseValidationError) as caught: + client.models.list() + error = safe_error(caught.value, "test-token") + assert "APIResponseValidationError" in error + assert "HTTP 200 GET /api/v1/forward/models" in error + assert "request_id=req-validation" in error + assert "data.0.is_enabled" in error + assert "bool_parsing" in error + assert "private-response-value" not in error + assert "test-token" not in error + + +def test_status_error_shows_server_reason_and_redacts_token_and_signed_url(): + from qca import BadRequestError + + with Forward( + pat="test-token", + base_url="https://api.test/api/v1/forward", + http_client=httpx.Client( + transport=httpx.MockTransport( + lambda _: httpx.Response( + 400, + json={ + "error": { + "type": "invalid_request_error", + "message": "Idempotency-Key header is required; test-token https://signed.test/?secret=value", + }, + "private": "unrelated-response-value", + }, + headers={"x-request-id": "req-missing-key"}, + ) + ) + ), + ) as client: + with pytest.raises(BadRequestError) as caught: + client.memory_stores.create(name="memory", idempotency_key="key") + error = safe_error(caught.value, "test-token") + assert "HTTP 400 POST /api/v1/forward/memory_stores" in error + assert "request_id=req-missing-key" in error + assert "Idempotency-Key header is required" in error + assert "test-token" not in error + assert "signed.test" not in error + assert "secret=value" not in error + assert "unrelated-response-value" not in error From 6bd48b745ceeed5a9ac05def1a144ecbe65e1edc Mon Sep 17 00:00:00 2001 From: moonyue-w <300878504+moonyue-w@users.noreply.github.com> Date: Mon, 21 Sep 2026 16:43:32 +0800 Subject: [PATCH 2/3] docs(examples): slim to pure runnable samples, drop test harness MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 移除断言层（TurnResult/wait_reply/turn/ProjectMemory），保留运行时；场景改纯演示 print、无 verify；删 conftest 与两个 test_live。 Task: 1789906941 --- examples/common/live.py | 107 --------------------------------- examples/conftest.py | 25 -------- examples/forward/batch.py | 66 ++++++-------------- examples/forward/memory.py | 66 ++++++++++++++------ examples/forward/resources.py | 45 ++++++++------ examples/forward/schedule.py | 19 +++--- examples/forward/session.py | 29 ++++----- examples/forward/test_live.py | 12 ---- examples/managed/deployment.py | 19 +++--- examples/managed/dream.py | 15 ++--- examples/managed/memory.py | 59 +++++++++++++----- examples/managed/resources.py | 45 ++++++++------ examples/managed/session.py | 29 ++++----- examples/managed/test_live.py | 12 ---- 14 files changed, 211 insertions(+), 337 deletions(-) delete mode 100644 examples/conftest.py delete mode 100644 examples/forward/test_live.py delete mode 100644 examples/managed/test_live.py diff --git a/examples/common/live.py b/examples/common/live.py index bcb5dc0..c74f070 100644 --- a/examples/common/live.py +++ b/examples/common/live.py @@ -3,7 +3,6 @@ import argparse import json import os -import random import re import sys import time @@ -171,89 +170,6 @@ def cleanup(self) -> None: self.output("cleanup", "completed") -@dataclass -class TurnResult: - text: str = "" - last_id: str = "" - tool_used: bool = False - complete: bool = False - - def observe(self, event: Any) -> None: - if hasattr(event, "to_dict"): - event = event.to_dict(mode="json") - kind = event.get("type") - if event.get("id"): - self.last_id = event["id"] - if kind in ("session.error", "session.status_terminated"): - raise AssertionError(f"Execution failed: {kind}, event_id={self.last_id}") - if kind in ("agent.tool_use", "agent.mcp_tool_use"): - self.tool_used = True - elif kind == "agent.message": - # Only the latest completed assistant message can satisfy assertions. - self.text = "\n".join( - block.get("text", "") for block in event.get("content", []) if block.get("type") == "text" - ) - elif kind == "session.status_idle": - reason = event.get("stop_reason") - reason = reason.get("type") if isinstance(reason, dict) else reason - if reason not in (None, "", "end_turn", "stop_sequence"): - raise AssertionError(f"Execution stopped early: {reason}, event_id={self.last_id}") - self.complete = bool(self.text) - - def verify(self, expected: list[str], require_tool: bool = False) -> None: - if not self.complete: - raise AssertionError(f"No idle state after assistant output; last_event_id={self.last_id}") - if not all(value in self.text for value in expected): - raise AssertionError(f"Assistant output is missing expected values; last_event_id={self.last_id}") - if require_tool and not self.tool_used: - raise AssertionError(f"No actual tool execution; last_event_id={self.last_id}") - - -def wait_reply(events: Any, run: Run, session_id: str, after: str = "") -> TurnResult: - result = TurnResult(last_id=after) - while not result.complete: - run.remaining() - page = events.list( - session_id, - order="asc", - limit=100, - extra_query={"after_id": result.last_id or None, "include_tool_calls": True}, - timeout=min(run.remaining(), 30), - ) - for index, event in enumerate(page): - if index >= 2000: - raise AssertionError("Event polling exceeded 2000 events") - result.observe(event) - if result.complete: - break - if not result.complete: - run.pause() - run.output("assistant", result.text) - return result - - -def turn( - events: Any, - run: Run, - session_id: str, - prompt: str, - expected: list[str], - *, - require_tool: bool = False, -) -> TurnResult: - run.output("user", prompt) - result = events.send( - session_id, - events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], - extra_headers={"Idempotency-Key": name("event")}, - ) - if len(result.data) != 1 or not result.data[0].id: - raise AssertionError("Send must return exactly one user event ID") - reply = wait_reply(events, run, session_id, result.data[0].id) - reply.verify(expected, require_tool) - return reply - - def choose_model(models: Any, requested: str) -> str: enabled = [model.id for model in models.data if model.is_enabled and model.id] if requested: @@ -265,29 +181,6 @@ def choose_model(models: Any, requested: str) -> str: return "ultimate" if "ultimate" in enabled else sorted(enabled)[0] -@dataclass -class ProjectMemory: - project: str = field(default_factory=lambda: "青禾订单-" + marker()[:6]) - release_time: str = field(default_factory=lambda: f"{random.randrange(20, 24):02}:{random.randrange(60):02}") - contact: str = field(default_factory=lambda: random.choice(["林岚", "陈朔", "叶澄", "苏棠"])) - rollback_version: str = field( - default_factory=lambda: f"v2.{random.randrange(100, 1000)}.{random.randrange(100, 1000)}" - ) - path = "projects/release-conventions.md" - - def content(self) -> str: - return f"---\nname: release-conventions\ndescription: {self.project} 的项目发布约定\nmetadata:\n type: project\n---\n\n# {self.project}\n\n- 北京时间 {self.release_time} 开始发布。\n- 发布异常时联系值班负责人{self.contact}。\n- 回滚使用已验证的稳定版本 {self.rollback_version}。\n\nWhy: 在值班窗口发布，并使用验证过的版本恢复服务。\nHow to apply: 为这个项目拟定发布计划时遵循以上约定。\n" - - def index(self) -> str: - return f"- [{self.project} 发布约定]({self.path}) — 发布窗口、异常联系人与回滚约定。\n" - - def prompt(self) -> str: - return f"请根据你记得的项目约定，为「{self.project}」拟一份简短上线安排，涵盖开始时间、异常联系和回滚处理。不要执行发布；缺少信息时请明确说明。" - - def expected(self) -> list[str]: - return [self.release_time, self.contact, self.rollback_version] - - def run_cli(mode: str, client_type: Any, scenarios: dict[str, Callable[[Any, Run], None]]) -> None: parser = argparse.ArgumentParser(description=f"Qoder {mode} SDK examples") parser.add_argument("--env", default=".env.live") diff --git a/examples/conftest.py b/examples/conftest.py deleted file mode 100644 index 09b69c6..0000000 --- a/examples/conftest.py +++ /dev/null @@ -1,25 +0,0 @@ -import os - -import pytest - -from examples.common.live import Config, Run -from qca import Forward, Managed - - -@pytest.fixture -def live_example(request): - if os.environ.get("QODER_RUN_LIVE") != "1": - pytest.skip("Set QODER_RUN_LIVE=1 to run account-backed examples") - mode = request.param - config = Config.load( - mode, - env_file=os.environ.get("QODER_LIVE_ENV_FILE", ".env.live"), - timeout=float(os.environ.get("QODER_LIVE_TIMEOUT", "300")), - ) - context = Run(config) - client_type = Forward if mode == "forward" else Managed - with client_type(**config.client_options()) as client: - try: - yield client, context - finally: - context.cleanup() diff --git a/examples/forward/batch.py b/examples/forward/batch.py index 35e19ca..6a918ac 100644 --- a/examples/forward/batch.py +++ b/examples/forward/batch.py @@ -1,4 +1,4 @@ -"""上传 JSONL 批量任务，等待完成并核对任务、输出与会话回复。 +"""上传 JSONL 批量任务，等待完成并打印任务状态与输出。 运行：python -m examples.forward.batch """ @@ -10,7 +10,7 @@ import httpx -from examples.common.live import Run, choose_model, marker, name, run_cli, wait_reply +from examples.common.live import Run, choose_model, name, run_cli from qca import Forward from ._cleanup import finish_session @@ -34,12 +34,12 @@ def run(client: Forward, context: Run) -> None: ) template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) - expected, custom_id = marker(), name("task") + custom_id = name("task") data = { "custom_id": custom_id, "template_id": template_id, "identity_id": identity_id, - "body": {"input": "Reply with exactly " + expected}, + "body": {"input": "请用一句话打个招呼。"}, } input_file = client.files.upload( file=("input.jsonl", (json.dumps(data) + "\n").encode()), purpose="session_resource" @@ -60,60 +60,32 @@ def cleanup_batch() -> None: while current.status not in terminal: context.pause() current = client.batches.retrieve(batch.id) - if not current.output_file_id: - if current.request_counts and current.request_counts.total == 0: - return - raise AssertionError("Batch has no output for session cleanup") - rows = batch_rows(client, batch.id) - if ( - len(rows) != 1 - or rows[0].get("custom_id") != custom_id - or rows[0].get("identity_id") != identity_id - or rows[0].get("template_id") != template_id - ): - raise AssertionError("Batch cleanup output does not match this run") - if rows[0].get("session_id"): - finish_session(client, context, rows[0]["session_id"]) + if current.output_file_id: + for row in batch_rows(client, batch.id): + if row.get("session_id"): + finish_session(client, context, row["session_id"]) context.track("batch", batch.id, cleanup_batch) while batch.status not in terminal: context.pause() batch = client.batches.retrieve(batch.id) - if ( - batch.status != "completed" - or not batch.request_counts - or batch.request_counts.completed != 1 - or batch.request_counts.failed != 0 - or not batch.output_file_id - ): - raise AssertionError("Batch did not complete exactly one successful task") + context.output("batch_status", batch.status) + if batch.request_counts: + context.output( + "request_counts", + {"completed": batch.request_counts.completed, "failed": batch.request_counts.failed}, + ) tasks = client.batches.tasks.list(batch.id) - if len(tasks.data) != 1 or tasks.data[0].custom_id != custom_id: - raise AssertionError("Batch task did not round trip") - rows = batch_rows(client, batch.id) - if len(rows) != 1: - raise AssertionError("Expected one Batch output row") - row = rows[0] - context.output("batch_output", row) - if ( - row.get("custom_id") != custom_id - or row.get("identity_id") != identity_id - or row.get("template_id") != template_id - or row.get("status") != "completed" - or row.get("error") - or not row.get("session_id") - ): - raise AssertionError("Batch output ownership or status mismatch") - if expected not in json.dumps(row.get("response")): - raise AssertionError("Batch response does not contain expected output") - wait_reply(client.sessions.events, context, row["session_id"]).verify([expected]) + context.output("tasks", [task.custom_id for task in tasks.data]) + if batch.output_file_id: + context.output("batch_output", batch_rows(client, batch.id)) def batch_rows(client: Forward, batch_id: str) -> list[dict[str, Any]]: link = client.batches.retrieve_output(batch_id) url = httpx.URL(link.url) if url.scheme not in ("http", "https") or not url.host or url.userinfo: - raise AssertionError("Invalid Batch output URL") + raise RuntimeError("Invalid Batch output URL") # A separate HTTP client prevents API credentials from reaching storage. with httpx.Client(timeout=30, follow_redirects=True) as download: with download.stream("GET", url) as response: @@ -122,7 +94,7 @@ def batch_rows(client: Forward, batch_id: str) -> list[dict[str, Any]]: for chunk in response.iter_bytes(): content.extend(chunk) if len(content) > 4 * 1024 * 1024: - raise AssertionError("Batch output exceeds the example's 4 MiB limit") + raise RuntimeError("Batch output exceeds the example's 4 MiB limit") return [json.loads(line) for line in content.splitlines() if line.strip()] diff --git a/examples/forward/memory.py b/examples/forward/memory.py index b055102..ed584da 100644 --- a/examples/forward/memory.py +++ b/examples/forward/memory.py @@ -1,38 +1,47 @@ -"""写入项目记忆，在新会话中挂载并验证助手按记忆回答。 +"""写入项目记忆并挂载到新会话，发送消息并打印助手回复。 运行：python -m examples.forward.memory """ from __future__ import annotations -from examples.common.live import ProjectMemory, Run, choose_model, name, run_cli, turn +from examples.common.live import Run, choose_model, name, run_cli from qca import Forward from ._cleanup import finish_session +# 一段临时的项目记忆，仅用于演示如何写入并挂载 Memory Store。 +MEMORY_PATH = "projects/release-conventions.md" +MEMORY_DOC = ( + "---\n" + "name: release-conventions\n" + "description: 青禾订单 的项目发布约定\n" + "metadata:\n" + " type: project\n" + "---\n\n" + "# 青禾订单\n\n" + "- 北京时间 22:30 开始发布。\n" + "- 发布异常时联系值班负责人林岚。\n" + "- 回滚使用已验证的稳定版本 v2.480.15。\n\n" + "Why: 在值班窗口发布，并使用验证过的版本恢复服务。\n" + "How to apply: 为这个项目拟定发布计划时遵循以上约定。\n" +) +MEMORY_INDEX = f"- [青禾订单 发布约定]({MEMORY_PATH}) — 发布窗口、异常联系人与回滚约定。\n" +MEMORY_PROMPT = "请根据你记得的项目约定，为「青禾订单」拟一份简短上线安排，涵盖开始时间、异常联系和回滚处理。不要执行发布；缺少信息时请明确说明。" + def run(client: Forward, context: Run) -> None: environment = client.environments.create(name=name("env"), config={"type": "cloud"}) environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户") + identity_id = context.track("identity", identity.id, lambda: client.identities.delete(identity.id)) - def cleanup_identity() -> None: - result = client.identities.clear(identity.id, reason="SDK example cleanup") - if result.status != "completed": - raise AssertionError("Dedicated Identity cleanup did not complete") - client.identities.delete(identity.id) - - identity_id = context.track("identity", identity.id, cleanup_identity) - - memory = ProjectMemory() store = client.memory_stores.create(name=name("memory"), idempotency_key=name("memory-key")) context.track("memory_store", store.id, lambda: client.memory_stores.delete(store.id)) - for path, content in ((memory.path, memory.content()), ("MEMORY.md", memory.index())): - entry = client.memory_stores.memories.create(store.id, path=path, content=content) - saved = client.memory_stores.memories.retrieve(entry.id, memory_store_id=store.id) - if saved.content != content: - raise AssertionError("Persisted memory does not match the supplied content") + for path, content in ((MEMORY_PATH, MEMORY_DOC), ("MEMORY.md", MEMORY_INDEX)): + client.memory_stores.memories.create(store.id, path=path, content=content) + model = choose_model(client.models.list(), context.config.model) context.output("selected_model", model) template = client.templates.create( @@ -50,16 +59,33 @@ def cleanup_identity() -> None: store.id, lambda: client.identities.memory_stores.detach(store.id, template_id=template_id, identity_id=identity_id), ) - mounts = client.identities.memory_stores.list(template_id, identity_id=identity_id) - if not any(mount.memory_store_id == store.id for mount in mounts.data): - raise AssertionError("Memory Store binding was not persisted") session = client.sessions.create( identity_id=identity_id, template_id=template_id, ) session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) - turn(client.sessions.events, context, session_id, memory.prompt(), memory.expected()) + context.output("user", MEMORY_PROMPT) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": MEMORY_PROMPT}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": sent.data[0].id}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + if event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + break if __name__ == "__main__": diff --git a/examples/forward/resources.py b/examples/forward/resources.py index 1c33d2a..71b8955 100644 --- a/examples/forward/resources.py +++ b/examples/forward/resources.py @@ -1,11 +1,11 @@ -"""上传文件与 Skill，挂载到会话，并验证环境变量与工具调用。 +"""上传文件与 Skill，挂载到会话，发送消息并打印助手回复。 运行：python -m examples.forward.resources """ from __future__ import annotations -from examples.common.live import Run, choose_model, marker, name, run_cli, turn +from examples.common.live import Run, choose_model, marker, name, run_cli from qca import Forward from ._cleanup import finish_session @@ -56,22 +56,31 @@ def run(client: Forward, context: Run) -> None: ) session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) - turn( - client.sessions.events, - context, - session_id, - "请使用工具读取 /data/workspace/sdk-example.txt 和 SDK_EXAMPLE_VALUE 环境变量，返回两个值。", - [file_value, env_value], - require_tool=True, - ) - turn( - client.sessions.events, - context, - session_id, - f"请使用技能 {skill_name}，读取并返回 EXAMPLE_SKILL_CODE。", - [skill_value], - require_tool=True, - ) + def ask(prompt: str) -> None: + context.output("user", prompt) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": sent.data[0].id}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + if event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + break + + ask("请使用工具读取 /data/workspace/sdk-example.txt 和 SDK_EXAMPLE_VALUE 环境变量，返回两个值。") + ask(f"请使用技能 {skill_name}，读取并返回 EXAMPLE_SKILL_CODE。") if __name__ == "__main__": diff --git a/examples/forward/schedule.py b/examples/forward/schedule.py index a6d0c13..5b794b7 100644 --- a/examples/forward/schedule.py +++ b/examples/forward/schedule.py @@ -1,11 +1,11 @@ -"""创建手动 Schedule，触发执行并验证关联会话的回复。 +"""创建手动 Schedule，触发执行并打印关联会话的回复。 运行：python -m examples.forward.schedule """ from __future__ import annotations -from examples.common.live import Run, choose_model, marker, name, run_cli, wait_reply +from examples.common.live import Run, choose_model, name, run_cli from qca import Forward from ._cleanup import finish_session @@ -29,13 +29,12 @@ def run(client: Forward, context: Run) -> None: ) template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) - expected = marker() schedule = client.schedules.create( identity_id=identity_id, template_id=template_id, environment_id=environment_id, name=name("schedule"), - initial_events=[{"type": "user.message", "content": "Reply with exactly " + expected}], + initial_events=[{"type": "user.message", "content": "请用一句话打个招呼。"}], trigger_policy={"type": "manual"}, execution={"max_attempts": 1, "max_concurrent_runs": 1}, ) @@ -55,15 +54,17 @@ def cleanup_run() -> None: context.track("schedule_run", execution.id, cleanup_run) while True: current = client.schedule_runs.retrieve(execution.id, identity_id=identity_id) - if current.status == "completed": + context.output("run_status", current.status) + if current.status in ("completed", "failed", "skipped"): break - if current.status in ("failed", "skipped"): - raise AssertionError(f"Schedule Run failed: {current.status}") context.pause() + if not current.session_id: - raise AssertionError("Completed Schedule Run has no session") + return context.output("session_id", current.session_id) - wait_reply(client.sessions.events, context, current.session_id).verify([expected]) + for event in client.sessions.events.list(current.session_id, order="asc"): + if event.type == "agent.message": + context.output("assistant", event.to_json()) if __name__ == "__main__": diff --git a/examples/forward/session.py b/examples/forward/session.py index 862bcbb..90ac3a4 100644 --- a/examples/forward/session.py +++ b/examples/forward/session.py @@ -1,11 +1,11 @@ -"""创建会话、发送用户消息，通过 SSE 读取最终回复。 +"""创建会话、发送用户消息，通过 SSE 读取并打印助手回复。 运行：python -m examples.forward.session """ from __future__ import annotations -from examples.common.live import Run, TurnResult, choose_model, marker, name, run_cli +from examples.common.live import Run, choose_model, name, run_cli from qca import Forward from ._cleanup import finish_session @@ -35,23 +35,17 @@ def run(client: Forward, context: Run) -> None: ) session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) - expected = marker() - prompt = "请用一句话介绍你能提供什么帮助，并在末尾原样附上：" + expected + prompt = "请用一句话介绍你能提供什么帮助。" context.output("user", prompt) sent = client.sessions.events.send( session_id, - events=[ - { - "type": "user.message", - "content": [{"type": "text", "text": prompt}], - } - ], + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], extra_headers={"Idempotency-Key": name("event")}, ) - if len(sent.data) != 1 or not sent.data[0].id: - raise AssertionError("Send must return exactly one user event ID") + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") - reply = TurnResult(last_id=sent.data[0].id) + # 从刚发送的用户事件之后开始订阅，打印助手消息，遇到 idle 即停。 with client.sessions.events.stream( session_id, extra_headers={"Last-Event-ID": sent.data[0].id}, @@ -59,11 +53,12 @@ def run(client: Forward, context: Run) -> None: ) as stream: for event in stream: context.remaining() - reply.observe(event) - if reply.complete: + if event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": break - context.output("assistant", reply.text) - reply.verify([expected]) if __name__ == "__main__": diff --git a/examples/forward/test_live.py b/examples/forward/test_live.py deleted file mode 100644 index 390a0dd..0000000 --- a/examples/forward/test_live.py +++ /dev/null @@ -1,12 +0,0 @@ -import pytest - -from .__main__ import SCENARIOS - -pytestmark = pytest.mark.live - - -@pytest.mark.parametrize("live_example", ["forward"], indirect=True) -@pytest.mark.parametrize("scenario", SCENARIOS.values(), ids=SCENARIOS) -def test_forward_example(live_example, scenario): - client, context = live_example - scenario(client, context) diff --git a/examples/managed/deployment.py b/examples/managed/deployment.py index 1bfd7cc..44855cd 100644 --- a/examples/managed/deployment.py +++ b/examples/managed/deployment.py @@ -1,11 +1,11 @@ -"""创建 Deployment，手动运行并验证关联会话的回复。 +"""创建 Deployment，手动运行并打印关联会话的回复。 运行：python -m examples.managed.deployment """ from __future__ import annotations -from examples.common.live import Run, choose_model, marker, name, run_cli, wait_reply +from examples.common.live import Run, choose_model, name, run_cli from qca import Managed from ._cleanup import finish_session @@ -25,24 +25,21 @@ def run(client: Managed, context: Run) -> None: ) agent_id = context.track("agent", agent.id, lambda: client.agents.archive(agent.id)) - expected = marker() deployment = client.deployments.create( name=name("deployment"), environment_id=environment_id, agent=agent_id, - initial_events=[ - {"type": "user.message", "content": [{"type": "text", "text": "Reply with exactly " + expected}]} - ], + initial_events=[{"type": "user.message", "content": [{"type": "text", "text": "请用一句话打个招呼。"}]}], ) context.track("deployment", deployment.id, lambda: client.deployments.archive(deployment.id)) execution = client.deployments.run(deployment.id) if not execution.session_id: - raise AssertionError("Deployment Run returned no session") + return context.track("session", execution.session_id, lambda: finish_session(client, context, execution.session_id)) - saved = client.deployment_runs.retrieve(execution.id) - if saved.session_id != execution.session_id: - raise AssertionError("Deployment Run session ID changed") - wait_reply(client.sessions.events, context, execution.session_id).verify([expected]) + context.output("session_id", execution.session_id) + for event in client.sessions.events.list(execution.session_id, order="asc"): + if event.type == "agent.message": + context.output("assistant", event.to_json()) if __name__ == "__main__": diff --git a/examples/managed/dream.py b/examples/managed/dream.py index 3e22427..0f9de33 100644 --- a/examples/managed/dream.py +++ b/examples/managed/dream.py @@ -1,11 +1,11 @@ -"""整理 Memory Store 中的记忆，读取输出并核对原始事实。 +"""整理 Memory Store 中的记忆，读取并打印整理后的输出。 运行：python -m examples.managed.dream """ from __future__ import annotations -from examples.common.live import Run, choose_model, marker, name, run_cli +from examples.common.live import Run, choose_model, name, run_cli from qca import Managed from ._cleanup import finish_session @@ -16,11 +16,10 @@ def run(client: Managed, context: Run) -> None: context.output("selected_model", model) store = client.memory_stores.create(name=name("dream-input")) context.track("input_memory_store", store.id, lambda: client.memory_stores.delete(store.id)) - expected = marker() client.memory_stores.memories.create( store.id, path="sdk-example/source.md", - content=f"Permanent project verification code: {expected}. Preserve this exact code during consolidation.", + content="Permanent project verification code: QODER-DREAM-SAMPLE. Preserve this exact code during consolidation.", ) dream = client.dreams.create( inputs=[{"type": "memory_store", "memory_store_id": store.id}], @@ -32,18 +31,14 @@ def run(client: Managed, context: Run) -> None: while dream.status in ("pending", "running"): context.pause() dream = client.dreams.retrieve(dream_id) - if dream.status != "completed" or not dream.outputs: - raise AssertionError(f"Dream did not complete: {dream.status}") - for output in dream.outputs: + context.output("dream_status", dream.status) + for output in dream.outputs or []: for memory in client.memory_stores.memories.list(output.memory_store_id): if memory.path == "sdk-example/consolidated.md": saved = client.memory_stores.memories.retrieve(memory.id, memory_store_id=output.memory_store_id) context.output("output_memory_store_id", output.memory_store_id) context.output("memory_path", saved.path) context.output("memory_content", saved.content) - if saved.content and expected in saved.content: - return - raise AssertionError("Dream did not persist consolidated memory with the original value") def finish_dream(client: Managed, context: Run, dream_id: str, input_store_id: str) -> None: diff --git a/examples/managed/memory.py b/examples/managed/memory.py index 0a581fb..6791c59 100644 --- a/examples/managed/memory.py +++ b/examples/managed/memory.py @@ -1,28 +1,44 @@ -"""写入项目记忆，在新会话中挂载并验证助手按记忆回答。 +"""写入项目记忆并挂载到新会话，发送消息并打印助手回复。 运行：python -m examples.managed.memory """ from __future__ import annotations -from examples.common.live import ProjectMemory, Run, choose_model, name, run_cli, turn +from examples.common.live import Run, choose_model, name, run_cli from qca import Managed from ._cleanup import finish_session +# 一段临时的项目记忆，仅用于演示如何写入并挂载 Memory Store。 +MEMORY_PATH = "projects/release-conventions.md" +MEMORY_DOC = ( + "---\n" + "name: release-conventions\n" + "description: 青禾订单 的项目发布约定\n" + "metadata:\n" + " type: project\n" + "---\n\n" + "# 青禾订单\n\n" + "- 北京时间 22:30 开始发布。\n" + "- 发布异常时联系值班负责人林岚。\n" + "- 回滚使用已验证的稳定版本 v2.480.15。\n\n" + "Why: 在值班窗口发布，并使用验证过的版本恢复服务。\n" + "How to apply: 为这个项目拟定发布计划时遵循以上约定。\n" +) +MEMORY_INDEX = f"- [青禾订单 发布约定]({MEMORY_PATH}) — 发布窗口、异常联系人与回滚约定。\n" +MEMORY_PROMPT = "请根据你记得的项目约定，为「青禾订单」拟一份简短上线安排，涵盖开始时间、异常联系和回滚处理。不要执行发布；缺少信息时请明确说明。" + def run(client: Managed, context: Run) -> None: environment = client.environments.create(name=name("env"), config={"type": "cloud"}) environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) - memory = ProjectMemory() store = client.memory_stores.create(name=name("memory")) context.track("memory_store", store.id, lambda: client.memory_stores.delete(store.id)) - for path, content in ((memory.path, memory.content()), ("MEMORY.md", memory.index())): - entry = client.memory_stores.memories.create(store.id, path=path, content=content) - saved = client.memory_stores.memories.retrieve(entry.id, memory_store_id=store.id) - if saved.content != content: - raise AssertionError("Persisted memory does not match the supplied content") + for path, content in ((MEMORY_PATH, MEMORY_DOC), ("MEMORY.md", MEMORY_INDEX)): + client.memory_stores.memories.create(store.id, path=path, content=content) + model = choose_model(client.models.list(), context.config.model) context.output("selected_model", model) agent = client.agents.create( @@ -40,12 +56,27 @@ def run(client: Managed, context: Run) -> None: ) session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) - if not any( - resource.type == "memory_store" and resource.memory_store_id == store.id - for resource in client.sessions.resources.list(session_id) - ): - raise AssertionError("Memory Store is not mounted in the new session") - turn(client.sessions.events, context, session_id, memory.prompt(), memory.expected()) + context.output("user", MEMORY_PROMPT) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": MEMORY_PROMPT}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": sent.data[0].id}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + if event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + break if __name__ == "__main__": diff --git a/examples/managed/resources.py b/examples/managed/resources.py index fdac185..e1ed375 100644 --- a/examples/managed/resources.py +++ b/examples/managed/resources.py @@ -1,11 +1,11 @@ -"""上传文件与 Skill，挂载到会话，并验证环境变量与工具调用。 +"""上传文件与 Skill，挂载到会话，发送消息并打印助手回复。 运行：python -m examples.managed.resources """ from __future__ import annotations -from examples.common.live import Run, choose_model, marker, name, run_cli, turn +from examples.common.live import Run, choose_model, marker, name, run_cli from qca import Managed from ._cleanup import finish_session @@ -47,22 +47,31 @@ def run(client: Managed, context: Run) -> None: ) session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) - turn( - client.sessions.events, - context, - session_id, - "请使用工具读取 /data/workspace/sdk-example.txt 和 SDK_EXAMPLE_VALUE 环境变量，返回两个值。", - [file_value, env_value], - require_tool=True, - ) - turn( - client.sessions.events, - context, - session_id, - f"请使用技能 {skill_name}，读取并返回 EXAMPLE_SKILL_CODE。", - [skill_value], - require_tool=True, - ) + def ask(prompt: str) -> None: + context.output("user", prompt) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": sent.data[0].id}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + if event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + break + + ask("请使用工具读取 /data/workspace/sdk-example.txt 和 SDK_EXAMPLE_VALUE 环境变量，返回两个值。") + ask(f"请使用技能 {skill_name}，读取并返回 EXAMPLE_SKILL_CODE。") if __name__ == "__main__": diff --git a/examples/managed/session.py b/examples/managed/session.py index 2e4e637..5d47bb4 100644 --- a/examples/managed/session.py +++ b/examples/managed/session.py @@ -1,11 +1,11 @@ -"""创建会话、发送用户消息，通过 SSE 读取最终回复。 +"""创建会话、发送用户消息，通过 SSE 读取并打印助手回复。 运行：python -m examples.managed.session """ from __future__ import annotations -from examples.common.live import Run, TurnResult, choose_model, marker, name, run_cli +from examples.common.live import Run, choose_model, name, run_cli from qca import Managed from ._cleanup import finish_session @@ -31,23 +31,17 @@ def run(client: Managed, context: Run) -> None: ) session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) - expected = marker() - prompt = "请用一句话介绍你能提供什么帮助，并在末尾原样附上：" + expected + prompt = "请用一句话介绍你能提供什么帮助。" context.output("user", prompt) sent = client.sessions.events.send( session_id, - events=[ - { - "type": "user.message", - "content": [{"type": "text", "text": prompt}], - } - ], + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], extra_headers={"Idempotency-Key": name("event")}, ) - if len(sent.data) != 1 or not sent.data[0].id: - raise AssertionError("Send must return exactly one user event ID") + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") - reply = TurnResult(last_id=sent.data[0].id) + # 从刚发送的用户事件之后开始订阅，打印助手消息，遇到 idle 即停。 with client.sessions.events.stream( session_id, extra_headers={"Last-Event-ID": sent.data[0].id}, @@ -55,11 +49,12 @@ def run(client: Managed, context: Run) -> None: ) as stream: for event in stream: context.remaining() - reply.observe(event) - if reply.complete: + if event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": break - context.output("assistant", reply.text) - reply.verify([expected]) if __name__ == "__main__": diff --git a/examples/managed/test_live.py b/examples/managed/test_live.py deleted file mode 100644 index 88ec24b..0000000 --- a/examples/managed/test_live.py +++ /dev/null @@ -1,12 +0,0 @@ -import pytest - -from .__main__ import SCENARIOS - -pytestmark = pytest.mark.live - - -@pytest.mark.parametrize("live_example", ["managed"], indirect=True) -@pytest.mark.parametrize("scenario", SCENARIOS.values(), ids=SCENARIOS) -def test_managed_example(live_example, scenario): - client, context = live_example - scenario(client, context) From 9fb9b557156c752b7756295c82f94fddef5bdab4 Mon Sep 17 00:00:00 2001 From: moonyue-w <300878504+moonyue-w@users.noreply.github.com> Date: Mon, 21 Sep 2026 17:32:26 +0800 Subject: [PATCH 3/3] docs(examples): add conversation/identity_config/custom_tools/streaming_deltas to match Go parity Task: 1789906941 --- examples/forward/__main__.py | 6 ++ examples/forward/conversation.py | 70 +++++++++++++++ examples/forward/identity_config.py | 80 +++++++++++++++++ examples/forward/streaming_deltas.py | 84 +++++++++++++++++ examples/managed/__main__.py | 6 ++ examples/managed/conversation.py | 66 ++++++++++++++ examples/managed/custom_tools.py | 129 +++++++++++++++++++++++++++ examples/managed/streaming_deltas.py | 80 +++++++++++++++++ 8 files changed, 521 insertions(+) create mode 100644 examples/forward/conversation.py create mode 100644 examples/forward/identity_config.py create mode 100644 examples/forward/streaming_deltas.py create mode 100644 examples/managed/conversation.py create mode 100644 examples/managed/custom_tools.py create mode 100644 examples/managed/streaming_deltas.py diff --git a/examples/forward/__main__.py b/examples/forward/__main__.py index 682bf84..c987a4a 100644 --- a/examples/forward/__main__.py +++ b/examples/forward/__main__.py @@ -4,19 +4,25 @@ from qca import Forward from .batch import run as batch +from .conversation import run as conversation +from .identity_config import run as identity_config from .memory import run as memory from .models import run as models from .resources import run as resources from .schedule import run as schedule from .session import run as session +from .streaming_deltas import run as streaming_deltas SCENARIOS = { "models": models, "session": session, + "conversation": conversation, "resources": resources, + "identity_config": identity_config, "memory": memory, "schedule": schedule, "batch": batch, + "streaming_deltas": streaming_deltas, } diff --git a/examples/forward/conversation.py b/examples/forward/conversation.py new file mode 100644 index 0000000..8b5ac3f --- /dev/null +++ b/examples/forward/conversation.py @@ -0,0 +1,70 @@ +"""复用同一个 Session 进行多轮对话，并分页读取会话历史。 + +运行：python -m examples.forward.conversation +""" + +from __future__ import annotations + +from examples.common.live import Run, choose_model, marker, name, run_cli +from qca import Forward + +from ._cleanup import finish_session + + +def run(client: Forward, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户") + identity_id = context.track("identity", identity.id, lambda: client.identities.delete(identity.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + template = client.templates.create( + name=name("template"), + environment_id=environment_id, + model=model, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) + + session = client.sessions.create(identity_id=identity_id, template_id=template_id) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + def ask(prompt: str) -> None: + context.output("user", prompt) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": sent.data[0].id}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + if event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + break + + # 同一个 Session 支持多轮：服务端在 session_id 下保留完整历史，无需客户端携带上文。 + code = "project-" + marker() + ask(f"这次项目代号是 {code}。请在本次对话中记住它，不要使用工具或写入记忆库。现在只回复：已记住。") + ask("只根据本次会话上文，告诉我刚才约定的项目代号。只回复代号，不要使用工具。") + + # 分页读取已有的用户消息和助手回复，重建对话文字记录。 + for event in client.sessions.events.list(session_id, order="asc", limit=100): + if event.type in ("user.message", "agent.message"): + context.output(f"history.{event.type}", event.to_json()) + + +if __name__ == "__main__": + run_cli("forward", Forward, {"conversation": run}) diff --git a/examples/forward/identity_config.py b/examples/forward/identity_config.py new file mode 100644 index 0000000..897d65b --- /dev/null +++ b/examples/forward/identity_config.py @@ -0,0 +1,80 @@ +"""两个 Identity 共用一个 Template，写入各自的个性化配置，再读取生效配置并在会话中验证。 + +运行：python -m examples.forward.identity_config +""" + +from __future__ import annotations + +from examples.common.live import Run, choose_model, marker, name, run_cli +from qca import Forward + +from ._cleanup import finish_session + + +def run(client: Forward, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + shared, baseline = marker(), marker() + template = client.templates.create( + name=name("template"), + environment_id=environment_id, + model=model, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + environment_variables={"SDK_SHARED_VALUE": shared, "SDK_PERSONAL_VALUE": baseline}, + ) + template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) + + # 先创建两个 Identity 并写入个性化配置：SDK_SHARED_VALUE 继承模板默认，SDK_PERSONAL_VALUE 被各自覆盖。 + identities: list[str] = [] + personal: list[str] = [] + for index in range(2): + identity = client.identities.create(external_id=name("identity"), name=f"SDK 示例用户 {index + 1}") + identity_id = context.track("identity", identity.id, lambda ref=identity.id: client.identities.delete(ref)) + identities.append(identity_id) + value = marker() + personal.append(value) + client.identities.configs.upsert( + template_id, + identity_id=identity_id, + identity_config={"environment_variables": {"SDK_PERSONAL_VALUE": {"op": "set", "value": value}}}, + ) + effective = client.identities.configs.retrieve_effective(template_id, identity_id=identity_id) + context.output( + f"identity_{index + 1}_effective_env", + effective.session.environment_variables if effective.session else None, + ) + + # 每个 Identity 各起一个会话，读取自己实际生效的环境变量。 + for index, identity_id in enumerate(identities): + session = client.sessions.create(identity_id=identity_id, template_id=template_id) + session_id = context.track("session", session.id, lambda ref=session.id: finish_session(client, context, ref)) + prompt = "请使用工具读取 SDK_SHARED_VALUE 和 SDK_PERSONAL_VALUE 两个环境变量，只返回这两个变量的实际值。" + context.output(f"identity_{index + 1}_user", prompt) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": sent.data[0].id}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + if event.type == "agent.message": + context.output(f"identity_{index + 1}_assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + break + + +if __name__ == "__main__": + run_cli("forward", Forward, {"identity_config": run}) diff --git a/examples/forward/streaming_deltas.py b/examples/forward/streaming_deltas.py new file mode 100644 index 0000000..8fc81e9 --- /dev/null +++ b/examples/forward/streaming_deltas.py @@ -0,0 +1,84 @@ +"""订阅 agent.message 文本增量，逐步更新消息预览，再用最终完整消息替换预览。 + +运行：python -m examples.forward.streaming_deltas +""" + +from __future__ import annotations + +from examples.common.live import Run, choose_model, marker, name, run_cli +from qca import Forward + +from ._cleanup import finish_session + + +def _delta_text(event: object) -> str: + """从一条文本增量事件中取出片段文本；不是文本增量时返回空串。""" + delta = event.to_dict().get("delta") or {} + if not isinstance(delta, dict) or delta.get("type") != "content_delta": + return "" + content = delta.get("content") or {} + if not isinstance(content, dict) or content.get("type") != "text": + return "" + return content.get("text") or "" + + +def run(client: Forward, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + identity = client.identities.create(external_id=name("identity"), name="SDK 示例用户") + identity_id = context.track("identity", identity.id, lambda: client.identities.delete(identity.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + template = client.templates.create( + name=name("template"), + environment_id=environment_id, + model=model, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + template_id = context.track("template", template.id, lambda: client.templates.archive(template.id)) + + session = client.sessions.create(identity_id=identity_id, template_id=template_id) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + marker_value = marker() + prompt = "请分三句话解释为什么多轮对话要复用 Session ID，最后原样附上：" + marker_value + # 预览增量不写入历史：先订阅、开启 agent.message 增量，再发送消息。 + with client.sessions.events.stream( + session_id, + event_deltas=["agent.message"], + timeout=context.remaining(), + ) as stream: + context.output("user", prompt) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + + previews: dict[str, str] = {} + deltas = 0 + for event in stream: + context.remaining() + if event.type == "event_delta": + text = _delta_text(event) + if text and event.event_id: + deltas += 1 + # 同一条消息的多个片段按 event_id 累加，逐步刷新该消息的预览。 + previews[event.event_id] = previews.get(event.event_id, "") + text + context.output("preview", f"{event.event_id}: {previews[event.event_id]}") + elif event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + break + context.output("deltas", deltas) + + +if __name__ == "__main__": + run_cli("forward", Forward, {"streaming_deltas": run}) diff --git a/examples/managed/__main__.py b/examples/managed/__main__.py index 50cb3de..c778c16 100644 --- a/examples/managed/__main__.py +++ b/examples/managed/__main__.py @@ -3,20 +3,26 @@ from examples.common.live import run_cli from qca import Managed +from .conversation import run as conversation +from .custom_tools import run as custom_tools from .deployment import run as deployment from .dream import run as dream from .memory import run as memory from .models import run as models from .resources import run as resources from .session import run as session +from .streaming_deltas import run as streaming_deltas SCENARIOS = { "models": models, "session": session, + "conversation": conversation, "resources": resources, + "custom_tools": custom_tools, "memory": memory, "deployment": deployment, "dream": dream, + "streaming_deltas": streaming_deltas, } diff --git a/examples/managed/conversation.py b/examples/managed/conversation.py new file mode 100644 index 0000000..c7e69d6 --- /dev/null +++ b/examples/managed/conversation.py @@ -0,0 +1,66 @@ +"""复用同一个 Session 进行多轮对话，并分页读取会话历史。 + +运行：python -m examples.managed.conversation +""" + +from __future__ import annotations + +from examples.common.live import Run, choose_model, marker, name, run_cli +from qca import Managed + +from ._cleanup import finish_session + + +def run(client: Managed, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + agent = client.agents.create( + name=name("agent"), + model={"id": model}, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + agent_id = context.track("agent", agent.id, lambda: client.agents.archive(agent.id)) + + session = client.sessions.create(environment_id=environment_id, agent=agent_id) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + def ask(prompt: str) -> None: + context.output("user", prompt) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": sent.data[0].id}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + if event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + break + + # 同一个 Session 支持多轮：服务端在 session_id 下保留完整历史，无需客户端携带上文。 + code = "project-" + marker() + ask(f"这次项目代号是 {code}。请在本次对话中记住它，不要使用工具或写入记忆库。现在只回复：已记住。") + ask("只根据本次会话上文，告诉我刚才约定的项目代号。只回复代号，不要使用工具。") + + # 分页读取已有的用户消息和助手回复，重建对话文字记录。 + for event in client.sessions.events.list(session_id, order="asc", limit=100): + if event.type in ("user.message", "agent.message"): + context.output(f"history.{event.type}", event.to_json()) + + +if __name__ == "__main__": + run_cli("managed", Managed, {"conversation": run}) diff --git a/examples/managed/custom_tools.py b/examples/managed/custom_tools.py new file mode 100644 index 0000000..ab5703d --- /dev/null +++ b/examples/managed/custom_tools.py @@ -0,0 +1,129 @@ +"""让 Agent 调用自定义工具，在本地 Python 函数中执行，再把结果回传给 Agent 得到最终回答。 + +运行：python -m examples.managed.custom_tools +""" + +from __future__ import annotations + +import json + +from examples.common.live import Run, choose_model, marker, name, run_cli +from qca import Managed + +from ._cleanup import finish_session + + +def _lookup_order(call: object, data: dict[str, str]) -> tuple[str, bool]: + """业务代码边界：只有显式注册的工具会执行。返回 (回传内容, is_error)。""" + if getattr(call, "name", None) != "lookup_order": + return "unknown tool; use lookup_order", True + order_id = (getattr(call, "input", None) or {}).get("order_id") + if not isinstance(order_id, str) or not order_id: + return "order_id must be a non-empty string", True + if order_id != data["order_id"]: + return "order not found", True + return json.dumps(data, ensure_ascii=False), False + + +def _read_tool_turn(client: Managed, context: Run, session_id: str, after: str): + """读取一段执行流：返回 (待执行的工具调用列表或 None, 断点游标)。 + + 收到 requires_action 表示 Agent 暂停等待工具结果；否则本轮已给出最终回答。 + """ + calls: dict = {} + cursor = after + with client.sessions.events.stream( + session_id, + extra_headers={"Last-Event-ID": after}, + timeout=context.remaining(), + ) as stream: + for event in stream: + context.remaining() + if event.id: + cursor = event.id + if event.type == "agent.custom_tool_use": + calls[event.id] = event + context.output("tool_call", event.name) + elif event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + reason = event.stop_reason.type if event.stop_reason else None + if reason == "requires_action": + ids = event.stop_reason.event_ids or [] + return [calls[i] for i in ids if i in calls], cursor + return None, cursor + raise RuntimeError("Stream ended before a final answer or tool-result request") + + +def run(client: Managed, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + agent = client.agents.create( + name=name("agent"), + model={"id": model}, + system="查询订单时必须调用 lookup_order。拿到工具返回后，用中文回答订单状态和完整运单号，不得编造。", + tools=[ + { + "type": "custom", + "name": "lookup_order", + "description": "根据订单 ID 查询订单状态和运单号。", + "input_schema": { + "type": "object", + "properties": {"order_id": {"type": "string", "description": "待查询的订单 ID"}}, + "required": ["order_id"], + }, + } + ], + ) + agent_id = context.track("agent", agent.id, lambda: client.agents.archive(agent.id)) + + session = client.sessions.create(environment_id=environment_id, agent=agent_id) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + # 运单号只存在于本进程，不在模型提示词里；Agent 必须调用本地工具才能拿到。 + data = {"order_id": "order-" + marker(), "status": "已发货", "tracking_number": "track-" + marker()} + prompt = f"请查询订单 {data['order_id']}，告诉我订单状态和完整运单号。" + context.output("user", prompt) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + + after = sent.data[0].id + for _ in range(8): + pending, after = _read_tool_turn(client, context, session_id, after) + if pending is None: + return + results = [] + for call in pending: + text, is_error = _lookup_order(call, data) + context.output("tool_result", f"{call.name} is_error={is_error}") + results.append( + { + "type": "user.custom_tool_result", + "custom_tool_use_id": call.id, + "is_error": is_error, + "content": [{"type": "text", "text": text}], + } + ) + # 在 idle 事件之后续订，快速到来的最终回复不会被跳过。 + response = client.sessions.events.send( + session_id, + events=results, + extra_headers={"Idempotency-Key": name("tool-results")}, + ) + if not response.data: + raise RuntimeError("Server did not acknowledge tool results") + raise RuntimeError("Custom tool loop exceeded eight rounds") + + +if __name__ == "__main__": + run_cli("managed", Managed, {"custom_tools": run}) diff --git a/examples/managed/streaming_deltas.py b/examples/managed/streaming_deltas.py new file mode 100644 index 0000000..5a844e8 --- /dev/null +++ b/examples/managed/streaming_deltas.py @@ -0,0 +1,80 @@ +"""订阅 agent.message 文本增量，逐步更新消息预览，再用最终完整消息替换预览。 + +运行：python -m examples.managed.streaming_deltas +""" + +from __future__ import annotations + +from examples.common.live import Run, choose_model, marker, name, run_cli +from qca import Managed + +from ._cleanup import finish_session + + +def _delta_text(event: object) -> str: + """从一条文本增量事件中取出片段文本；不是文本增量时返回空串。""" + delta = event.to_dict().get("delta") or {} + if not isinstance(delta, dict) or delta.get("type") != "content_delta": + return "" + content = delta.get("content") or {} + if not isinstance(content, dict) or content.get("type") != "text": + return "" + return content.get("text") or "" + + +def run(client: Managed, context: Run) -> None: + environment = client.environments.create(name=name("env"), config={"type": "cloud"}) + environment_id = context.track("environment", environment.id, lambda: client.environments.archive(environment.id)) + + model = choose_model(client.models.list(), context.config.model) + context.output("selected_model", model) + agent = client.agents.create( + name=name("agent"), + model={"id": model}, + system="你是一个 SDK 示例助手。必要时调用工具，只使用可实际读取的数据回答问题。", + tools=[{"type": "agent_toolset_20260401"}], + ) + agent_id = context.track("agent", agent.id, lambda: client.agents.archive(agent.id)) + + session = client.sessions.create(environment_id=environment_id, agent=agent_id) + session_id = context.track("session", session.id, lambda: finish_session(client, context, session.id)) + + marker_value = marker() + prompt = "请分三句话解释为什么多轮对话要复用 Session ID，最后原样附上：" + marker_value + # 预览增量不写入历史：先订阅、开启 agent.message 增量，再发送消息。 + with client.sessions.events.stream( + session_id, + event_deltas=["agent.message"], + timeout=context.remaining(), + ) as stream: + context.output("user", prompt) + sent = client.sessions.events.send( + session_id, + events=[{"type": "user.message", "content": [{"type": "text", "text": prompt}]}], + extra_headers={"Idempotency-Key": name("event")}, + ) + if not sent.data or not sent.data[0].id: + raise RuntimeError("Send returned no user event") + + previews: dict[str, str] = {} + deltas = 0 + for event in stream: + context.remaining() + if event.type == "event_delta": + text = _delta_text(event) + if text and event.event_id: + deltas += 1 + # 同一条消息的多个片段按 event_id 累加，逐步刷新该消息的预览。 + previews[event.event_id] = previews.get(event.event_id, "") + text + context.output("preview", f"{event.event_id}: {previews[event.event_id]}") + elif event.type == "agent.message": + context.output("assistant", event.to_json()) + elif event.type in ("session.error", "session.status_terminated"): + raise RuntimeError(f"Session stopped: {event.type}") + elif event.type == "session.status_idle": + break + context.output("deltas", deltas) + + +if __name__ == "__main__": + run_cli("managed", Managed, {"streaming_deltas": run})