refactor(test): decouple examples from integration harness

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

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/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,
 }
 
 

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()]
 
 

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