docs(examples): slim to pure runnable samples, drop test harness

移除断言层（TurnResult/wait_reply/turn/ProjectMemory），保留运行时；场景改纯演示 print、无 verify；删 conftest 与两个 test_live。

Task: 1789906941

Author: moonyue-w

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")

examples/conftest.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()]
 
 

examples/forward/batch.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__":