diff --git a/.github/workflows/api-acceptance.yml b/.github/workflows/api-acceptance.yml
index d60aca75f..f548466a7 100644
--- a/.github/workflows/api-acceptance.yml
+++ b/.github/workflows/api-acceptance.yml
@@ -62,6 +62,7 @@ jobs:
           OAC_TEST_OFFICIAL_SDK_PYTHON: python
         run: |
           python services/core/tests/official_schema_test.py
+          python -m unittest discover -s services/core/tests -p qualify_public_native_test.py
           python services/core/tests/official_client.py
           go test ./services/core/tests/integration -run '^(TestFunctionStateOfficialClientReadsAndLiveEvents|TestSavedReferenceRetryOfficialClient|TestAgentUpdateOfficialClient|TestAgentDeletionOfficialClient|TestSessionAgentFilterOfficialClient|TestSessionDeletionOfficialClient|TestEnvironmentInitialFailureOfficialClient|TestSelfHostedInitialCreationOfficialClient|TestSelfHostedCancellationOfficialClient)$' -count=1
       - uses: ./.github/actions/e2b-provider
diff --git a/contracts/agents-api/environment-files.md b/contracts/agents-api/environment-files.md
index 783da51fe..2c6edde49 100644
--- a/contracts/agents-api/environment-files.md
+++ b/contracts/agents-api/environment-files.md
@@ -30,7 +30,6 @@ The response is `{"object": "page", "data": [...], "next": …, "has_more": …}
 - A `path` that does not exist, names a regular file, or passes through a symbolic link returns an empty page. Links are never followed.
 - Each page reads the directory again; there is no snapshot. If the directory's regular files (their names or sizes) or the request's parameters changed since the token was issued, the token is rejected. Unchanged names and sizes do not prove unchanged contents.
 - A directory with more than 1,024 entries of any kind returns 503 and no partial page. Permission errors, a missing workspace root and transport failures also return 503.
-- When the daemon has no local workspace binding, the Claude Code adapter answers the read instead: a missing path returns 404, and a regular file or symbolic link returns 503.
 
 Query errors, all with type and code `invalid_request_error` and a null `param` unless noted:
 
diff --git a/contracts/agents-api/harness-onboarding.md b/contracts/agents-api/harness-onboarding.md
index d84f74c16..95d8be692 100644
--- a/contracts/agents-api/harness-onboarding.md
+++ b/contracts/agents-api/harness-onboarding.md
@@ -208,6 +208,34 @@ Keep provider keys in private operator files, never in commits or logs. Existing
 | MiniMax native history binding | `mcode/session_test.go` |
 | Explicit refusals without native effects or fabricated results | `mcode/unsupported_test.go` |
 
+## Qualify the public path
+
+`services/core/tests/qualify_public_native.py` runs the pinned official SDK and raw HTTP assertions against an already installed, isolated Core deployment with its real agent host. It creates and deletes its own Sessions and uses the selected real model; it does not provision a deployment or replace the executor. Set `OPENAI_BASE_URL` to the deployment's `/v1` endpoint and `OPENAI_API_KEY` to its Project key. Supply a second Project's key in a private file. [Installation](../../docs/getting-started/install.md) owns deployment setup; [Projects and keys](./admin-api.md#projects-and-keys) owns credential issuance.
+
+The private settings JSON has exactly `agent`, `model_provider` and `environment`. `agent` contains `model` and an explicit `x_agents_core.harness`, with optional `harness_config` inside that extension. `model_provider` is the complete [provider bundle](./model-execution.md#session-override); `environment` is the public Session Environment input. Keep settings and foreign-key files absolute and mode 0600, outside the repository. Evidence must be a new absolute path under `~/.oac`; it contains public responses and check names, never the settings. The runner refuses to write evidence containing any of its three supplied credentials.
+
+```bash
+python services/core/tests/qualify_public_native.py \
+  --settings "$HOME/.oac/qualification/codex-none.json" \
+  --foreign-key-file "$HOME/.oac/qualification/foreign-project.key" \
+  --suite none \
+  --evidence "$HOME/.oac/qualification/codex-none-result.json"
+```
+
+| Suite | Operations | Placement |
+| --- | --- | --- |
+| `none` | Creation retry, foreign history rejection, two native text Turns, history recall, SSE ordering and SDK/raw Item and Turn parity | `none` |
+| `pending-actions` | Query/reconnect pending calls, success/error results, cancellation, exact target rejection, retries and durable Items | Any declared placement with function tools |
+| `functions` | SDK handlers, success/error, native file and Artifact bytes, continuation, pending cancellation and tenant isolation | Workspace |
+| `images` | Initial and active images, image results, retry/atomic rejection, native files/Artifacts, isolation and continuation | Workspace with declared image and function support |
+| `structured` | Saved and inline schema, function-assisted native files, exact JSON/SSE, cancellation and text override | Workspace with declared structured output and function support |
+
+For `self_hosted`, choose a custom absolute `workspace_directory`. When the runner prints each new Session ID, connect a separate isolated machine or container using that Session's public installation command; the runner waits up to five minutes. Multiple Sessions must not share a workspace. Use the existing Environment setup, package and capability assertions separately to qualify preparation semantics. The separate `official_environment_files_native.py` check accepts two already connected self-hosted Sessions and checks Files.list sorting, pagination and isolation through the Environment owner.
+
+Warm continuation is the default and records cold recovery as `unverified`. To qualify cold continuation, additionally pass `--compose-directory` with the owned installation's absolute directory and `--compose-project` with its exact project name. The runner restarts only that project's `agent-host`, confirms its container start time changed, and then runs the unchanged history assertions. This does not qualify a Core restart or a sandbox checkpoint restore. The `pending-actions` suite reconnects the public client, not the agent-host process, and rejects those restart options. Select cold recovery only where the [declaration and coverage ledger](./index.md#known-gaps) support it; unsupported recovery remains a gap, never a successful skipped check. An API rejection fails the selected suite.
+
+Run `python -m unittest discover -s services/core/tests -p qualify_public_native_test.py` with the pinned SDK to check credential handling and the owned restart boundary without a model. Existing deterministic Core integration tests remain the authority for schema validation, atomic admission, durable receipts and rejection semantics. Real-model results qualify only the selected suite, protocol, Harness and placement. Provider lifecycle, native identity, credential isolation, deferred discovery and unselected suites need separate evidence; view-only results do not qualify the public path.
+
 ## Native installer participation
 
 An adapter supplies `agent.Installation` from `installation.go` in its own package: its registered kind and activation environment. The agent host uses this declaration to activate the packaged Harness. Adapters own native layout; validate the packaged content and execution on the Linux agent host. Missing or incompatible native content fails; it never installs itself during a Turn. Self-hosted installers carry no Harness or Node.js.
diff --git a/contracts/agents-api/zh/environment-files.md b/contracts/agents-api/zh/environment-files.md
index 1977f2b63..d2326b64d 100644
--- a/contracts/agents-api/zh/environment-files.md
+++ b/contracts/agents-api/zh/environment-files.md
@@ -1,7 +1,7 @@
 ---
 title: "Environment 文件与 Artifact"
 source: contracts/agents-api/environment-files.md
-source_hash: 1b58aa02aaccddb9675ef41ebfe2506a6fba0bb12139efb67e0da378d879aee7
+source_hash: 11d3609d81455a3f0c203f341802e3fec1a9f318ab0e9977d2de0f2a84b8b63c
 ---
 
 Session 工作区保存由 agent 及其工具修改的实时文件。`/agents/environments/{environment_id}/files` 列出一个工作区目录，并在其中创建文件。Turn 完成时，Core 将工作区 `outputs/` 目录中的文件复制为不可变 Artifact，通过 `/agents/sessions/{session_id}/artifacts` 读取。Artifact 的生命周期长于 Environment；工作区文件则不是。
@@ -32,7 +32,6 @@ Session 工作区保存由 agent 及其工具修改的实时文件。`/agents/en
 - `path` 不存在、指向普通文件或经过符号链接时返回空页。不跟随链接。
 - 每页重新读取目录，不提供快照。token 签发后目录普通文件（名称或大小）或请求参数改变时，token 被拒绝。名称和大小未变不证明内容未变。
 - 目录中任何类型条目合计超过 1,024 个时返回 503，不返回部分页。权限错误、工作区根缺失和传输失败也返回 503。
-- daemon 无本地工作区绑定时，改由 Claude Code 适配器回答读取：路径缺失返回 404，普通文件或符号链接返回 503。
 
 查询错误的 type 和 code 均为 `invalid_request_error`，除另有说明外 `param` 为 null：
 
diff --git a/contracts/agents-api/zh/harness-onboarding.md b/contracts/agents-api/zh/harness-onboarding.md
index 0348ce0b3..7f6af512d 100644
--- a/contracts/agents-api/zh/harness-onboarding.md
+++ b/contracts/agents-api/zh/harness-onboarding.md
@@ -1,7 +1,7 @@
 ---
 title: "添加 Harness"
 source: contracts/agents-api/harness-onboarding.md
-source_hash: 2436c12691cc2f2753f39df73a6f480f081c3ddef1936e40c7da12c09c36a35e
+source_hash: 8bf3becb666da1ba0e1f6470bb468eba8d2352e63c2bddda78be6511414ed98a
 ---
 
 **Harness** 是一种运行模型和工具循环的原生代理引擎（Codex、Claude Code、MiniMax Code）。**Harness 适配器**将 Runtime 的 Executor 和 Turn 契约转换到该引擎的 SDK 或协议。本文档定义 Runtime–Harness 协议：适配器接口及其生命周期义务、注册、支持声明和验收。
@@ -210,6 +210,34 @@ Environment 验收使用 `services/core/tests/official_environment_{templates,se
 | MiniMax 原生历史绑定 | `mcode/session_test.go` |
 | 无原生副作用或伪造结果的明确拒绝 | `mcode/unsupported_test.go` |
 
+## 验证公共调用路径 {#qualify-the-public-path}
+
+`services/core/tests/qualify_public_native.py` 使用锁定的官方 SDK 和原始 HTTP 断言，访问已经安装、隔离且运行真实 agent host 的 Core 部署。它创建并删除自己的 Session，使用所选真实模型；不负责部署，也不替代执行器。将 `OPENAI_BASE_URL` 设置为部署的 `/v1` 端点，将 `OPENAI_API_KEY` 设置为其 Project 密钥。另一个 Project 的密钥通过私有文件提供。[安装](../../../docs/zh/getting-started/install.md) 负责部署设置；[Projects and keys](./admin-api.md#projects-and-keys) 负责凭据签发。
+
+私有设置 JSON 恰好包含 `agent`、`model_provider` 和 `environment`。`agent` 包含 `model` 和显式的 `x_agents_core.harness`，可在该扩展内提供 `harness_config`。`model_provider` 是完整的 [Provider 配置包](./model-execution.md#session-override)；`environment` 是公共 Session Environment 输入。设置文件和外部 Project 密钥文件必须使用绝对路径、权限 0600，并保存在仓库之外。证据必须使用 `~/.oac` 下新的绝对路径；其中保存公共响应和检查名称，不保存设置。运行器拒绝写入含有三个已提供凭据中任意一个的证据。
+
+```bash
+python services/core/tests/qualify_public_native.py \
+  --settings "$HOME/.oac/qualification/codex-none.json" \
+  --foreign-key-file "$HOME/.oac/qualification/foreign-project.key" \
+  --suite none \
+  --evidence "$HOME/.oac/qualification/codex-none-result.json"
+```
+
+| 套件 | 操作 | 放置方式 |
+| --- | --- | --- |
+| `none` | 创建重试、外部历史拒绝、两个原生文本 Turn、历史回忆、SSE 顺序，以及 SDK/原始 Item 和 Turn 一致性 | `none` |
+| `pending-actions` | 查询和重连待处理调用、成功/错误结果、取消、精确目标拒绝、重试和持久化 Item | 声明支持函数工具的任意放置方式 |
+| `functions` | SDK handler、成功/错误、原生文件和 Artifact 字节、继续执行、待处理调用取消和租户隔离 | 工作区 |
+| `images` | 初始和活动图像、图像结果、重试/原子拒绝、原生文件/Artifact、隔离和继续执行 | 声明支持图像和函数的工作区 |
+| `structured` | 保存和内联 schema、函数辅助原生文件、精确 JSON/SSE、取消和文本覆盖 | 声明支持结构化输出和函数的工作区 |
+
+对于 `self_hosted`，选择自定义绝对 `workspace_directory`。运行器打印每个新 Session ID 后，使用该 Session 的公共安装命令连接独立的隔离机器或容器；运行器最多等待五分钟。多个 Session 不得共享工作区。使用现有 Environment setup、包和能力断言单独验证准备语义。独立的 `official_environment_files_native.py` 检查接受两个已连接的 self-hosted Session，通过 Environment owner 验证 Files.list 排序、分页和隔离。
+
+默认验证热继续执行，并将冷恢复记录为 `unverified`。验证冷继续执行时，额外传入指向所拥有安装的绝对目录的 `--compose-directory`，以及指定其精确项目名称的 `--compose-project`。运行器仅重启该项目的 `agent-host`，确认容器启动时间已改变，然后执行相同的历史断言。这不能证明 Core 重启或 sandbox 检查点恢复。`pending-actions` 套件重连的是公共客户端，而非 agent-host 进程，因此拒绝这些重启选项。仅在[声明和覆盖台账](./index.md#known-gaps) 支持时选择冷恢复；不支持的恢复仍是缺口，不能把跳过的检查记为成功。API 拒绝会使所选套件失败。
+
+使用锁定的 SDK 运行 `python -m unittest discover -s services/core/tests -p qualify_public_native_test.py`，可在不调用模型的情况下检查凭据处理和所拥有的重启边界。现有确定性 Core 集成测试仍负责 schema 验证、原子准入、持久化回执和拒绝语义。真实模型结果仅证明所选套件、协议、Harness 和放置方式。Provider 生命周期、原生身份、凭据隔离、延迟发现和未选择的套件需要独立证据；仅通过 view 测试不能证明公共调用路径。
+
 ## 原生安装器参与 {#native-installer-participation}
 
 适配器从自身包中的 `installation.go` 提供 `agent.Installation`：已注册 kind 和激活环境。agent host 使用该声明激活打包的 Harness。适配器负责原生布局；必须在 Linux agent host 上验证打包内容和执行。原生内容缺失或不兼容时必须失败；绝不会在 Turn 期间自行安装。自托管安装器不携带 Harness 或 Node.js。
diff --git a/services/core/tests/official_environment_files_native.py b/services/core/tests/official_environment_files_native.py
index dae7a1b19..3f02db620 100644
--- a/services/core/tests/official_environment_files_native.py
+++ b/services/core/tests/official_environment_files_native.py
@@ -1,8 +1,4 @@
-"""Opt-in live check; stdin supplies engine, base and two tenants with session_id and token_file/token_env.
-
-Optional directory_reader is "local" (default, a local workspace binding) or, for
-claude_sdk only, "claude_sdk_adapter" for a daemon without that binding.
-"""
+"""Opt-in live check; stdin supplies engine, base and two tenants with session_id and token_file/token_env."""
 
 import importlib.metadata
 import json
@@ -52,11 +48,13 @@ def generate_files(client, session_id, label):
     workspace = PurePosixPath(session.environment.workspace_directory)
     assert workspace.is_absolute(), "Absolute workspace required"
     assert client.beta.agents.environments.retrieve(environment_id).status == "connected", "Connect the Environment first"
-    directory = str(workspace / ("files-list-" + label + "-" + uuid.uuid4().hex))
+    name = "files-list-" + label + "-" + uuid.uuid4().hex
+    directory = str(workspace / name)
+    public_directory = "/workspace/" + name
     contents = {"A.txt": "A\n", "a-b.txt": "three\n", "a.txt": "fourteen-bytes\n", "z.txt": "last\n"}
-    expected = {directory + "/" + name: len(content.encode()) for name, content in contents.items()}
+    expected = {public_directory + "/" + name: len(content.encode()) for name, content in contents.items()}
     sibling = directory + "-sibling"
-    sibling_expected = {sibling + "/one.txt": 3, sibling + "/two.txt": 3}
+    sibling_expected = {public_directory + "-sibling/one.txt": 3, public_directory + "-sibling/two.txt": 3}
     command = "mkdir -- " + shlex.quote(directory) + " " + shlex.quote(sibling)
     for name, content in contents.items():
         command += " && printf %s " + shlex.quote(content) + " > " + shlex.quote(directory + "/" + name)
@@ -74,17 +72,16 @@ def generate_files(client, session_id, label):
             assert turns[0].status not in ("failed", "cancelled"), "File generation Turn failed"
             if turns[0].status == "completed" and sessions.retrieve(session_id).status == "idle":
                 return {"session_id": session_id, "environment_id": environment_id, "turn_id": turns[0].id,
-                        "directory": directory, "expected": expected,
-                        "sibling_directory": sibling, "sibling_expected": sibling_expected}
+                        "directory": public_directory, "expected": expected,
+                        "sibling_directory": public_directory + "-sibling", "sibling_expected": sibling_expected}
         time.sleep(0.2)
     raise AssertionError("File generation Turn did not complete")
 
 
 def main():
     settings = json.load(sys.stdin)
-    assert settings["engine"] in ("codex", "claude_sdk"), "Select one qualified native engine"
-    reader = settings.get("directory_reader", "local")
-    assert reader == "local" or (reader == "claude_sdk_adapter" and settings["engine"] == "claude_sdk"), "Unsupported directory reader"
+    assert settings["engine"] in ("codex", "claude_sdk", "mcode"), "Select one qualified native engine"
+    assert "directory_reader" not in settings, "Files are served by the Environment owner"
     assert len(settings["tenants"]) == 2, "Two independent tenant Sessions are required"
     pin = json.loads((Path(__file__).resolve().parents[3] / "contracts/agents-api/upstream.json").read_text())
     distribution = importlib.metadata.distribution("openai")
@@ -107,12 +104,12 @@ def main():
                 client, http, fixture["environment_id"], fixture["sibling_directory"], fixture["sibling_expected"])
             fixture["wire_rows"] = verify_file_list_rows(client, http, fixture["environment_id"], {
                 "directory": fixture["directory"], "missing": fixture["directory"] + "-missing",
-                "file": next(iter(fixture["expected"]))}, empty_pages=reader == "local")
+                "file": next(iter(fixture["expected"]))}, empty_pages=True)
             verify_file_tenant_isolation(client, clients[1 - index], http, fixture["environment_id"],
                                         fixture["directory"], page, fixture["expected"] | fixture["sibling_expected"])
             assert [turn.to_dict() for turn in client.beta.agents.sessions.turns.list(fixture["session_id"])] == before, "Files.list changed Turns"
             fixture["cross_tenant_denied"] = True
-    proof = {"engine": settings["engine"], "directory_reader": reader, "sdk_version": distribution.version, "sdk_commit": pin["commit"],
+    proof = {"engine": settings["engine"], "sdk_version": distribution.version, "sdk_commit": pin["commit"],
              "scope": "Public input-generated flat files, SDK/raw listing, sorting, pagination and two-tenant isolation",
              "fixtures": generated, "unverified": UNVERIFIED}
     serialized = json.dumps(proof, indent=2)
diff --git a/services/core/tests/official_hosted_functions_native.py b/services/core/tests/official_hosted_functions_native.py
index a573d89ee..c4df8aa7c 100644
--- a/services/core/tests/official_hosted_functions_native.py
+++ b/services/core/tests/official_hosted_functions_native.py
@@ -1,4 +1,4 @@
-"""Common real-harness acceptance for hosted functions and workspace execution."""
+"""Common real-Harness acceptance for public functions and workspace execution."""
 
 import importlib.metadata
 import json
@@ -9,22 +9,23 @@
 from session_cleanup import delete_session
 
 
-def verify_hosted_functions(client, foreign, http, model, restart, evidence):
-    """restart(session_id, environment_id) restarts the operator-owned deployment."""
+def verify_hosted_functions(client, foreign, http, agent_options, session_options, ready, restart, record):
+    """Run against the selected workspace; restart, when supplied, restarts its agent host."""
     pin = json.loads((Path(__file__).resolve().parents[3] / "contracts/agents-api/upstream.json").read_text())
     distribution = importlib.metadata.distribution("openai")
     source = json.loads(distribution.read_text("direct_url.json") or "{}")
     assert distribution.version == pin["sdk_version"] and source["vcs_info"]["commit_id"] == pin["commit"]
     sessions = client.beta.agents.sessions
+    workspace = session_options["environment"].get("workspace_directory", "/workspace").rstrip("/")
     endpoint = str(client.base_url).rstrip("/") + "/agents/sessions"
     headers = {"Authorization": "Bearer " + client.api_key, "OpenAI-Beta": "agents=v1"}
     marker = "function-memory-" + uuid.uuid4().hex
     private_error = "private-handler-" + uuid.uuid4().hex
     calls, observed, checks = [], [], []
-    session = sessions.create(agent={"model": model, "instructions": "Follow the requested tool calls exactly. Never repeat a failed call.", "tools": [{
+    session = sessions.create(agent={**agent_options, "instructions": "Follow the requested tool calls exactly. Never repeat a failed call.", "tools": [{
         "type": "function", "name": "lookup", "description": "Retrieve the requested test value.",
         "parameters": {"type": "object", "properties": {"key": {"type": "string"}}, "required": ["key"], "additionalProperties": False},
-    }]}, environment={"type": "openai_hosted"})
+    }]}, **session_options)
 
     def until(predicate, timeout=120):
         deadline = time.monotonic() + timeout
@@ -58,12 +59,14 @@ def invoke(text, key, handler):
         return terminals[0]["turn"]["id"], result[0]["item"]
 
     try:
+        ready(session)
+
         def success(arguments):
             assert arguments == {"key": "success"}
             calls.append(arguments)
             return marker
 
-        turn, result = invoke("Call lookup once with key success. Remember its returned string in conversation. Then use the native shell to create /workspace/outputs/function.txt containing exactly that string, with no newline. Do not call any other function.", "function-success", success)
+        turn, result = invoke(f"Call lookup once with key success. Remember its returned string in conversation. Then use the native shell to create {workspace}/outputs/function.txt containing exactly that string, with no newline. Do not call any other function.", "function-success", success)
         assert result["output"] == marker and "error" in result and result["error"] is None
         artifacts = list(sessions.artifacts.list(session.id, limit=100))
         output = [a for a in artifacts if a.turn_id == turn and a.path == "/workspace/outputs/function.txt"]
@@ -75,7 +78,8 @@ def success(arguments):
         assert any(i["type"] == "function_call_output" and i.get("output") == marker for i in items())
         checks.append("same_turn_function_native_file_and_public_artifact")
 
-        restart(session.id, session.environment.id)
+        if restart is not None:
+            restart()
 
         def failure(arguments):
             assert arguments == {"key": "failure"}
@@ -87,7 +91,7 @@ def failure(arguments):
         messages = [i for i in items() if i["type"] == "message" and i["role"] == "assistant"]
         assert marker in "\n".join(p.get("text", "") for p in messages[-1]["content"])
         assert len(calls) == 2
-        checks.append("cold_history_continuation_and_public_handler_error")
+        checks.append(("cold" if restart is not None else "warm") + "_history_continuation_and_public_handler_error")
 
         sessions.events.create(session.id, events=[{"type": "agent.session.input.message", "input": [{"role": "user", "content": [{"type": "input_text", "text": "Call lookup once with key pending and wait for the result."}]}]}], idempotency_key="function-pending")
         until(lambda: sessions.retrieve(session.id).required_actions)
@@ -105,7 +109,7 @@ def failure(arguments):
         checks.append("pending_function_tenant_isolation_and_cancel_retry")
         proof = {"checks": checks, "session": session.id, "calls": calls, "events": observed, "items": items()}
         assert private_error not in json.dumps(proof)
-        Path(evidence).write_text(json.dumps(proof, indent=2))
+        record(proof)
         return checks
     finally:
         delete_session(sessions, session.id)
diff --git a/services/core/tests/official_hosted_structured_native.py b/services/core/tests/official_hosted_structured_native.py
index e4d23fb19..9ec7bcf1f 100644
--- a/services/core/tests/official_hosted_structured_native.py
+++ b/services/core/tests/official_hosted_structured_native.py
@@ -1,4 +1,4 @@
-"""Structured final answers after real native Docker workspace/tool execution."""
+"""Structured final answers after real native workspace/tool execution."""
 import importlib.metadata
 import json
 from pathlib import Path
@@ -8,16 +8,17 @@
 from session_cleanup import delete_session
 
 
-def verify_hosted_structured(client, foreign, http, model, kind, restart, evidence):
+def verify_hosted_structured(client, foreign, http, agent_options, session_options, ready, restart, record):
     pin = json.loads((Path(__file__).resolve().parents[3] / "contracts/agents-api/upstream.json").read_text())
     dist = importlib.metadata.distribution("openai")
     assert dist.version == pin["sdk_version"]
     assert json.loads(dist.read_text("direct_url.json"))["vcs_info"]["commit_id"] == pin["commit"]
     sessions = client.beta.agents.sessions
+    workspace = session_options["environment"].get("workspace_directory", "/workspace").rstrip("/")
     root = str(client.base_url).rstrip("/") + "/agents"
     headers = {"Authorization": "Bearer " + client.api_key, "OpenAI-Beta": "agents=v1"}
     other_headers = {**headers, "Authorization": "Bearer " + foreign.api_key}
-    proof = {"engine": kind, "checks": [], "runs": [], "calls": []}
+    proof = {"engine": agent_options["x_agents_core"]["harness"], "checks": [], "runs": [], "calls": []}
     owned = []
     saved = None
     schema = {"type": "object", "properties": {name: {"type": "string"} for name in ("memory", "path", "marker")},
@@ -27,7 +28,7 @@ def verify_hosted_structured(client, foreign, http, model, kind, restart, eviden
             "parameters": {"type": "object", "properties": {}, "additionalProperties": False}}
 
     def save():
-        Path(evidence).write_text(json.dumps(proof, indent=2))
+        record(proof)
 
     def check(name):
         proof["checks"].append(name)
@@ -94,11 +95,13 @@ def final(sid, eid, turn, expected, events=None):
             assert events[added]["item"]["status"] == "in_progress" and events[added]["item"]["content"] == []
             deltas = [e["delta"] for e in events if e["type"] == "agent.session.turn.output_text.delta" and e["item_id"] == answer["id"]]
             assert deltas and "".join(deltas) == raw, deltas
-        artifacts = [a for a in sessions.artifacts.list(sid, limit=100) if a.path == expected["path"] and a.turn_id == turn.id]
+        assert expected["path"].startswith(workspace + "/")
+        public_path = "/workspace" + expected["path"][len(workspace):]
+        artifacts = [a for a in sessions.artifacts.list(sid, limit=100) if a.path == public_path and a.turn_id == turn.id]
         assert len(artifacts) == 1
         with sessions.artifacts.with_streaming_response.content(artifacts[0].id, session_id=sid) as response:
             assert response.read() == expected["memory"].encode()
-        assert any(f.path == expected["path"] for f in client.beta.agents.environments.files.list(eid, path="/workspace/outputs"))
+        assert any(f.path == public_path for f in client.beta.agents.environments.files.list(eid, path="/workspace/outputs"))
         assert http.get(root + "/sessions/" + sid + "/artifacts/" + artifacts[0].id + "/content", headers=other_headers).status_code == 404
         proof.setdefault("answers", []).append(answer)
 
@@ -131,12 +134,14 @@ def run(sid, text, handler=None):
         return events
 
     try:
-        saved = client.beta.agents.create(model=model, text={"format": output_format}, tools=[tool])
+        saved = client.beta.agents.create(**{k: v for k, v in agent_options.items() if k != "x_agents_core"},
+            extra_body={"x_agents_core": agent_options["x_agents_core"]}, text={"format": output_format}, tools=[tool])
         first_memory = uuid.uuid4().hex
-        path = "/workspace/outputs/initial.txt"
-        session = sessions.create(agent_id=saved.id, environment={"type": "openai_hosted"}, input=prompt(path, "initial"))
+        path = f"{workspace}/outputs/initial.txt"
+        session = sessions.create(agent_id=saved.id, **session_options, input=prompt(path, "initial"))
         sid, eid = session.id, session.environment.id
         owned.append(sid)
+        ready(session)
         proof.update(session=sid, environment=eid, agent=saved.id, format=output_format)
         assert session.agent.text.format.to_dict() == output_format
         def pending():
@@ -150,7 +155,7 @@ def pending():
         check("saved_schema_initial_function_native_file_and_final_json")
 
         memory = uuid.uuid4().hex
-        path = "/workspace/outputs/active.txt"
+        path = f"{workspace}/outputs/active.txt"
         def active(action):
             incoming = [message("Use marker 'active' for the final result, replacing the earlier marker. Keep the requested file path and memory task.")]
             sessions.events.create(sid, events=incoming, idempotency_key="active-marker")
@@ -168,27 +173,29 @@ def active(action):
         assert http.get(root + "/environments/" + eid + "/files", headers=other_headers).status_code == 404
         assert http.post(root + "/sessions", headers=other_headers, json={"agent_id": saved.id, "environment": {"type": "openai_hosted"}}).status_code == 404
         before = items(sid)
-        restart(sid, eid)
+        if restart is not None:
+            restart()
         assert items(sid) == before and sessions.retrieve(sid).agent.text.format.to_dict() == output_format
-        path = "/workspace/outputs/resumed.txt"
+        path = f"{workspace}/outputs/resumed.txt"
         events = run(sid, prompt(path, "resumed", recall=True))
         third = terminal(sid, 3)
         final(sid, eid, third, {"memory": memory, "path": path, "marker": "resumed"}, events)
         assert len([i for i in items(sid) if i["type"] == "function_call"]) == 2
-        check("cold_core_runtime_continuation_without_replay_and_tenant_isolation")
+        check(("cold_agent_host" if restart is not None else "warm") + "_continuation_without_replay_and_tenant_isolation")
 
         def cancel(action):
             for _ in range(2):
                 sessions.events.create(sid, events=[{"type": "agent.session.input.cancel"}], idempotency_key="cancel-pending")
-        run(sid, prompt("/workspace/outputs/cancelled.txt", "cancelled"), cancel)
+        run(sid, prompt(f"{workspace}/outputs/cancelled.txt", "cancelled"), cancel)
         cancelled = terminal(sid, 4, "cancelled")
         assert not any(i.get("turn_id") == cancelled.id and i["type"] == "message" and i.get("phase") == "final_answer" for i in items(sid))
         assert not sessions.retrieve(sid).required_actions
         check("pending_cancellation_has_no_structured_final")
 
-        plain = sessions.create(agent_id=saved.id, agent={"text": {"format": {"type": "text"}}}, environment={"type": "openai_hosted"})
+        plain = sessions.create(agent_id=saved.id, agent={"text": {"format": {"type": "text"}}}, **session_options)
         owned.append(plain.id)
-        run(plain.id, "Do not call remember. Use native tools to write exactly PLAIN_OK to /workspace/plain.txt with no newline, then reply only PLAIN_OK.")
+        ready(plain)
+        run(plain.id, f"Do not call remember. Use native tools to write exactly PLAIN_OK to {workspace}/plain.txt with no newline, then reply only PLAIN_OK.")
         assert any(i["type"] == "message" and i.get("role") == "assistant" and "PLAIN_OK" in i["content"][0].get("text", "") for i in items(plain.id))
         foreign_result = proof["calls"][0]
         assert http.post(root + "/sessions/" + plain.id + "/events", headers=headers, json={"events": [foreign_result]}).status_code in {400, 404, 409}
@@ -196,10 +203,11 @@ def cancel(action):
         assert http.get(root + "/sessions/" + plain.id + "/artifacts/" + artifact.id + "/content", headers=headers).status_code == 404
         check("text_override_and_same_tenant_session_isolation")
 
-        inline = sessions.create(agent={"model": model, "text": {"format": output_format}}, environment={"type": "openai_hosted"})
+        inline = sessions.create(agent={**agent_options, "text": {"format": output_format}}, **session_options)
         owned.append(inline.id)
+        ready(inline)
         inline_memory = uuid.uuid4().hex
-        path = "/workspace/outputs/inline.txt"
+        path = f"{workspace}/outputs/inline.txt"
         events = run(inline.id, "Use native tools to create the parent directory and write exactly " + inline_memory + " with no newline to " + path + ". Return that memory, path, and marker 'inline' using the requested output format.")
         final(inline.id, inline.environment.id, terminal(inline.id, 1), {"memory": inline_memory, "path": path, "marker": "inline"}, events)
         assert inline.agent.text.format.to_dict() == output_format
diff --git a/services/core/tests/official_pending_actions_native.py b/services/core/tests/official_pending_actions_native.py
index 017801329..ffcc9af3c 100644
--- a/services/core/tests/official_pending_actions_native.py
+++ b/services/core/tests/official_pending_actions_native.py
@@ -1,4 +1,4 @@
-"""Real hosted function-action recovery through the pinned SDK and public HTTP."""
+"""Real function-action recovery through the pinned SDK and public HTTP."""
 
 import importlib.metadata
 import json
@@ -8,7 +8,7 @@
 from session_cleanup import delete_session
 
 
-def verify_pending_actions(client, foreign, http, model, evidence):
+def verify_pending_actions(client, foreign, http, agent_options, session_options, ready, record):
     """The private runner supplies an isolated Core/native provider deployment."""
     pin = json.loads((Path(__file__).resolve().parents[3] / "contracts/agents-api/upstream.json").read_text())
     distribution = importlib.metadata.distribution("openai")
@@ -21,7 +21,7 @@ def verify_pending_actions(client, foreign, http, model, evidence):
     proof = {"sdk": pin, "checks": [], "rounds": [], "passed": False}
     created = []
     marker = "pending-value-" + uuid.uuid4().hex
-    agent = {"model": model, "instructions": "Call lookup exactly once when asked. Never retry a failed tool call. Use no other tools.", "tools": [{
+    agent = {**agent_options, "instructions": "Call lookup exactly once when asked. Never retry a failed tool call. Use no other tools.", "tools": [{
         "type": "function", "name": "lookup", "description": "Retrieve the requested test value.",
         "parameters": {"type": "object", "properties": {"key": {"type": "string"}}, "required": ["key"], "additionalProperties": False},
     }]}
@@ -57,10 +57,12 @@ def submit(sid, event, key, expected=202, auth=None, error=None):
         return {"request": key, "status": response.status_code}
 
     try:
-        session = sessions.create(agent=agent, environment={"type": "openai_hosted"})
+        session = sessions.create(agent=agent, **session_options)
         created.append(session.id)
-        other = sessions.create(agent=agent, environment={"type": "openai_hosted"})
+        ready(session)
+        other = sessions.create(agent=agent, **session_options)
         created.append(other.id)
+        ready(other)
         proof.update(session=session.id, other_session=other.id)
         for index, mode in enumerate(("success", "error", "cancel")):
             current = {"mode": mode, "before_disconnect": [], "after_reconnect": [], "refusals": []}
@@ -172,6 +174,6 @@ def pending_unchanged():
         proof["passed"] = True
         return proof["checks"]
     finally:
-        Path(evidence).write_text(json.dumps(proof, indent=2))
+        record(proof)
         for sid in reversed(created):
             delete_session(sessions, sid)
diff --git a/services/core/tests/official_workspace_images_native.py b/services/core/tests/official_workspace_images_native.py
index fa9b4bfe4..00328cae4 100644
--- a/services/core/tests/official_workspace_images_native.py
+++ b/services/core/tests/official_workspace_images_native.py
@@ -1,4 +1,4 @@
-"""Common hosted image acceptance through fixed SDK, HTTP and real native tools."""
+"""Image acceptance through the pinned SDK, HTTP and real native workspace tools."""
 
 import base64
 import importlib.metadata
@@ -11,22 +11,23 @@
 from session_cleanup import delete_session
 
 
-def verify_workspace_images(client, foreign, http, model, kind, restart, evidence):
-    """restart(session_id, environment_id) cold-restarts the owned Core/Runtime."""
+def verify_workspace_images(client, foreign, http, agent_options, session_options, ready, restart, record):
+    """Run against the selected workspace; restart, when supplied, restarts its agent host."""
     pin = json.loads((Path(__file__).resolve().parents[3] / "contracts/agents-api/upstream.json").read_text())
     distribution = importlib.metadata.distribution("openai")
     assert distribution.version == pin["sdk_version"]
     assert json.loads(distribution.read_text("direct_url.json"))["vcs_info"]["commit_id"] == pin["commit"]
     sessions = client.beta.agents.sessions
+    workspace = session_options["environment"].get("workspace_directory", "/workspace").rstrip("/")
     root = str(client.base_url).rstrip("/") + "/agents"
     headers = {"Authorization": "Bearer " + client.api_key, "OpenAI-Beta": "agents=v1"}
     foreign_headers = {**headers, "Authorization": "Bearer " + foreign.api_key}
-    proof = {"engine": kind, "checks": [], "runs": [], "calls": []}
+    proof = {"engine": agent_options["x_agents_core"]["harness"], "checks": [], "runs": [], "calls": []}
     expected_messages = []
     sid = None
 
     def save():
-        Path(evidence).write_text(json.dumps(proof, indent=2))
+        record(proof)
 
     def check(name):
         proof["checks"].append(name)
@@ -80,17 +81,19 @@ def post(events, key=None, foreign_request=False):
         return http.post(root + "/sessions/" + sid + "/events", headers=hdr, json={"events": events})
 
     def verify_file(path, expected, turn):
+        assert path.startswith(workspace + "/")
+        public_path = "/workspace" + path[len(workspace):]
         answers = [i for i in items() if i["type"] == "message" and i.get("role") == "assistant"]
         answer = " ".join(p["text"] for p in answers[-1]["content"] if p["type"] == "output_text").lower()
         positions = [answer.find(color) for color in expected]
         assert all(p >= 0 for p in positions) and positions == sorted(positions), answer
         artifacts = list(sessions.artifacts.list(sid, limit=100))
-        matches = [a for a in artifacts if a.path == path and a.turn_id == turn]
+        matches = [a for a in artifacts if a.path == public_path and a.turn_id == turn]
         assert len(matches) == 1, [a.to_dict() for a in artifacts]
         artifact = matches[0]
         with sessions.artifacts.with_streaming_response.content(artifact.id, session_id=sid) as response:
             assert response.read() == ",".join(expected).encode()
-        assert any(f.path == path for f in client.beta.agents.environments.files.list(eid, path="/workspace/outputs"))
+        assert any(f.path == public_path for f in client.beta.agents.environments.files.list(eid, path="/workspace/outputs"))
         for suffix in ("", "/content"):
             assert http.get(root + "/sessions/" + sid + "/artifacts/" + artifact.id + suffix, headers=foreign_headers).status_code == 404
 
@@ -134,7 +137,7 @@ def submit(action, output, validate=False):
         key = "result-" + action.call_id
         before = items()
         if validate:
-            if kind == "claude_sdk":
+            if agent_options["x_agents_core"]["harness"] == "claude_sdk":
                 invalid = [{**result, "success": False}, {**result, "output": [{"type": "input_image", "image_url": "https://example.test/image.png"}]}]
                 for bad in invalid:
                     assert post([bad], key).status_code == 400
@@ -150,12 +153,13 @@ def submit(action, output, validate=False):
 
     colors = ["red", "green", "blue", "yellow"]
     secrets.SystemRandom().shuffle(colors)
-    initial = image_messages(picture(colors), "/workspace/outputs/initial.txt")
+    initial = image_messages(picture(colors), f"{workspace}/outputs/initial.txt")
     try:
-        session = sessions.create(agent={"model": model, "tools": [{"type": "function", "name": "get_visual",
+        session = sessions.create(agent={**agent_options, "tools": [{"type": "function", "name": "get_visual",
             "description": "Wait for a visual supplied by the caller.", "parameters": {"type": "object", "properties": {}, "additionalProperties": False}}]},
-            environment={"type": "openai_hosted"}, input=initial)
+            **session_options, input=initial)
         sid, eid = session.id, session.environment.id
+        ready(session)
         proof.update(session=sid, environment=eid, initial_colors=colors)
         expected_messages.extend(initial)
         save()
@@ -168,20 +172,20 @@ def initial_done():
             return turns[0]
 
         first = until(initial_done)
-        verify_file("/workspace/outputs/initial.txt", colors, first.id)
+        verify_file(f"{workspace}/outputs/initial.txt", colors, first.id)
         history()
         check("initial_png_native_workspace_files_artifact")
 
         jpeg = "data:image/jpeg;base64," + base64.b64encode((Path(__file__).parent / "testdata/function-bands.jpg").read_bytes()).decode()
         idle = messages([{"type": "input_image", "image_url": jpeg}]) + messages([text(
-            "Write this image's four lowercase band colors from left to right, comma-separated with no newline, to /workspace/outputs/idle.txt using native tools. Reply with those colors. Do not call get_visual.")])
+            f"Write this image's four lowercase band colors from left to right, comma-separated with no newline, to {workspace}/outputs/idle.txt using native tools. Reply with those colors. Do not call get_visual.")])
         turn = run(idle)
-        verify_file("/workspace/outputs/idle.txt", ["yellow", "blue", "red", "green"], turn.id)
+        verify_file(f"{workspace}/outputs/idle.txt", ["yellow", "blue", "red", "green"], turn.id)
         check("prepared_image_only_jpeg_and_message_boundary")
 
         active_colors = colors[1:] + colors[:1]
         def active(action):
-            incoming = image_messages(picture(active_colors), "/workspace/outputs/active.txt")
+            incoming = image_messages(picture(active_colors), f"{workspace}/outputs/active.txt")
             batch = [event(incoming)]
             sessions.events.create(sid, events=batch, idempotency_key="active-image")
             expected_messages.extend(incoming)
@@ -189,14 +193,14 @@ def active(action):
             assert post([event(messages([text("conflict")]))], "active-image").status_code == 409
             submit(action, "The user supplied an image. Follow its instructions, then stop.")
         turn = run(messages([text("Call get_visual once and wait. Then follow the incoming image instructions.")]), active)
-        verify_file("/workspace/outputs/active.txt", active_colors, turn.id)
+        verify_file(f"{workspace}/outputs/active.txt", active_colors, turn.id)
         check("active_image_input_retry_and_native_tools")
 
         result_colors = colors[2:] + colors[:2]
         output = [text("Inspect this visual."), {"type": "input_image", "image_url": picture(result_colors, 15)}, text("Remember these band colors.")]
-        turn = run(messages([text("Call get_visual exactly once. Read its returned image and use native tools to write its four lowercase band colors in left-to-right order, comma-separated with no newline, to /workspace/outputs/result.txt. Reply with the colors. Do not call get_visual again.")]),
+        turn = run(messages([text(f"Call get_visual exactly once. Read its returned image and use native tools to write its four lowercase band colors in left-to-right order, comma-separated with no newline, to {workspace}/outputs/result.txt. Reply with the colors. Do not call get_visual again.")]),
                    lambda action: submit(action, output, validate=True))
-        verify_file("/workspace/outputs/result.txt", result_colors, turn.id)
+        verify_file(f"{workspace}/outputs/result.txt", result_colors, turn.id)
         check("large_function_image_result_receipt_retry_isolation_and_artifact")
 
         before = history()
@@ -209,8 +213,9 @@ def active(action):
         assert http.get(root + "/environments/" + eid + "/files", headers=foreign_headers).status_code == 404
         check("unsupported_message_batch_is_atomic_and_foreign_resources_hidden")
 
-        other = sessions.create(agent={"model": model}, environment={"type": "openai_hosted"})
+        other = sessions.create(agent=agent_options, **session_options)
         try:
+            ready(other)
             source_artifact = list(sessions.artifacts.list(sid, limit=100))[0]
             for suffix in ("", "/content"):
                 assert http.get(root + "/sessions/" + other.id + "/artifacts/" + source_artifact.id + suffix, headers=headers).status_code == 404
@@ -227,12 +232,13 @@ def active(action):
         finally:
             delete_session(sessions, other.id)
 
-        restart(sid, eid)
+        if restart is not None:
+            restart()
         assert history() == before
-        turn = run(messages([text("Recall the most recent image returned by get_visual from conversation history, without reading any file or calling get_visual. Write its four lowercase band colors in order, comma-separated with no newline, to /workspace/outputs/resumed.txt using native tools, then reply with them.")]))
-        verify_file("/workspace/outputs/resumed.txt", result_colors, turn.id)
+        turn = run(messages([text(f"Recall the most recent image returned by get_visual from conversation history, without reading any file or calling get_visual. Write its four lowercase band colors in order, comma-separated with no newline, to {workspace}/outputs/resumed.txt using native tools, then reply with them.")]))
+        verify_file(f"{workspace}/outputs/resumed.txt", result_colors, turn.id)
         assert len(sessions.turns.list(sid, limit=100).data) == 5
-        check("cold_core_runtime_history_continuation_without_replay")
+        check(("cold_agent_host" if restart is not None else "warm") + "_history_continuation_without_replay")
 
         pending = []
         def cancel(action):
diff --git a/services/core/tests/qualify_public_native.py b/services/core/tests/qualify_public_native.py
new file mode 100644
index 000000000..ccfe45c26
--- /dev/null
+++ b/services/core/tests/qualify_public_native.py
@@ -0,0 +1,190 @@
+"""Real public API acceptance against an already installed, isolated Core deployment."""
+
+import argparse
+import importlib.metadata
+import json
+import os
+from pathlib import Path
+import re
+import stat
+import subprocess
+import time
+import traceback
+import uuid
+
+import httpx2
+from openai import OpenAI
+
+from official_hosted_functions_native import verify_hosted_functions
+from official_hosted_structured_native import verify_hosted_structured
+from official_pending_actions_native import verify_pending_actions
+from official_workspace_images_native import verify_workspace_images
+from session_cleanup import delete_session
+
+
+def private_file(path):
+    path = Path(path)
+    assert path.is_absolute() and stat.S_IMODE(path.stat().st_mode) & 0o077 == 0, "Private absolute file required"
+    return path.read_text().strip()
+
+
+def verify_none(client, foreign, http, agent_options, session_options, ready, restart, record):
+    sessions = client.beta.agents.sessions
+    marker = "memory-" + uuid.uuid4().hex
+    key = "create-" + uuid.uuid4().hex
+    session = sessions.create(agent=agent_options, **session_options, idempotency_key=key)
+    proof = {"checks": [], "session": session.id, "runs": []}
+    root = str(client.base_url).rstrip("/") + "/agents/sessions/" + session.id
+    headers = {"Authorization": "Bearer " + client.api_key, "OpenAI-Beta": "agents=v1"}
+    try:
+        assert sessions.create(agent=agent_options, **session_options, idempotency_key=key).id == session.id
+        ready(session)
+        for suffix in ("", "/items", "/turns"):
+            assert http.get(root + suffix, headers={**headers, "Authorization": "Bearer " + foreign.api_key}).status_code == 404
+        proof["checks"].append("create_retry_and_foreign_history_rejection")
+        for index, prompt in enumerate(("Remember " + marker + ". Reply with exactly that string. Do not use tools.",
+                                        "Recall the string from our previous turn. Reply with exactly that string. Do not use tools.")):
+            if index and restart is not None:
+                restart()
+            with sessions.stream(session.id, input=prompt, idempotency_key=key + str(index), timeout=240) as stream:
+                events = [event.to_dict() for event in stream]
+            proof["runs"].append(events)
+            record(proof)
+            types = [event["type"] for event in events]
+            terminals = [event for event in events if event["type"] in {
+                "agent.session.turn.completed", "agent.session.turn.failed", "agent.session.turn.cancelled"}]
+            assert len(terminals) == 1 and terminals[0]["type"] == "agent.session.turn.completed"
+            assert types[-1] == "agent.session.idle"
+            assert types.index("agent.session.turn.created") < types.index("agent.session.turn.completed") < len(types) - 1
+            assert len({event["event_id"] for event in events}) == len(events)
+            for name in ("items", "turns"):
+                response = http.get(root + "/" + name, headers=headers, params={"limit": 100, "order": "asc"})
+                assert response.status_code == 200 and not response.json()["has_more"]
+                expected = getattr(sessions, name).list(session.id, limit=100, order="asc").to_dict()
+                assert response.json() == expected
+            items = list(sessions.items.list(session.id, limit=100, order="asc"))
+            answers = [item.to_dict() for item in items if item.type == "message" and item.role == "assistant"]
+            assert marker in "".join(part.get("text", "") for part in answers[-1]["content"])
+            assert len(sessions.turns.list(session.id).data) == index + 1
+            assert sessions.retrieve(session.id).required_actions == []
+        proof["checks"].append(("cold_agent_host" if restart is not None else "warm") + "_text_history_sse_and_sdk_raw_parity")
+        return proof["checks"]
+    finally:
+        record(proof)
+        delete_session(sessions, session.id)
+
+
+def main():
+    parser = argparse.ArgumentParser(description=__doc__)
+    parser.add_argument("--settings", required=True, help="Private JSON with agent, model_provider and environment")
+    parser.add_argument("--foreign-key-file", required=True, help="Private key of a different Project")
+    parser.add_argument("--suite", required=True, choices=("none", "functions", "pending-actions", "images", "structured"))
+    parser.add_argument("--evidence", required=True, type=Path, help="New evidence file under ~/.oac")
+    parser.add_argument("--compose-directory", type=Path, help="Owned installation to restart for cold recovery")
+    parser.add_argument("--compose-project", help="Exact owned Compose project; required with --compose-directory")
+    args = parser.parse_args()
+    assert bool(args.compose_directory) == bool(args.compose_project), "Supply both Compose selectors or neither"
+    assert args.suite != "pending-actions" or args.compose_directory is None, "Pending actions tests client reconnect, not process restart"
+    evidence = args.evidence
+    assert evidence.is_absolute() and not evidence.exists(), "Use a new absolute evidence path"
+    assert evidence.parent.resolve().is_relative_to((Path.home() / ".oac").resolve()), "Evidence belongs under ~/.oac"
+    evidence.parent.mkdir(parents=True, exist_ok=True, mode=0o700)
+    settings = json.loads(private_file(args.settings))
+    assert set(settings) == {"agent", "model_provider", "environment"}, "Expected agent, model_provider and environment only"
+    agent = settings["agent"]
+    assert set(agent) <= {"model", "x_agents_core"} and agent.get("model"), "Set the model and explicit Harness configuration"
+    assert agent.get("x_agents_core", {}).get("harness"), "Explicit Harness required"
+    environment = settings["environment"]
+    placement = environment.get("type")
+    assert placement in {"none", "self_hosted", "openai_hosted"}, "Explicit Environment required"
+    if args.suite == "none":
+        assert placement == "none", "The none suite requires environment:none"
+    elif args.suite != "pending-actions":
+        assert placement != "none", "This suite verifies native workspace tools and Artifacts"
+    if placement == "self_hosted":
+        assert Path(environment["workspace_directory"]).is_absolute(), "Absolute sandbox workspace required"
+    provider = settings["model_provider"]
+    assert provider.get("api_key"), "Explicit provider key required"
+    base, key = os.environ["OPENAI_BASE_URL"], os.environ["OPENAI_API_KEY"]
+    assert base.rstrip("/").endswith("/v1") and key, "Set OPENAI_BASE_URL ending in /v1 and OPENAI_API_KEY"
+    foreign_key = private_file(args.foreign_key_file)
+    assert foreign_key and foreign_key != key, "Distinct Project credentials required"
+    pin = json.loads((Path(__file__).resolve().parents[3] / "contracts/agents-api/upstream.json").read_text())
+    dist = importlib.metadata.distribution("openai")
+    assert dist.version == pin["sdk_version"] and json.loads(dist.read_text("direct_url.json") or "{}").get("vcs_info", {}).get("commit_id") == pin["commit"], "Install the pinned official SDK"
+    secrets = (key, foreign_key, provider["api_key"])
+    report = {"suite": args.suite, "harness": agent["x_agents_core"]["harness"], "placement": placement,
+              "model_protocol": provider["protocol"], "sdk_commit": pin["commit"], "status": "running",
+              "cold_recovery": "requested" if args.compose_directory else "unverified",
+              "unverified": ["Provider lifecycle, native identity, credential isolation and unselected suites need separate qualification."]}
+
+    def record(proof):
+        report["proof"] = proof
+        serialized = json.dumps(report, indent=2)
+        assert all(secret not in serialized for secret in secrets), "Credential in public acceptance evidence"
+        # Create privately even if the caller has a permissive umask.
+        descriptor = os.open(evidence, os.O_WRONLY | os.O_CREAT | os.O_TRUNC, 0o600)
+        with os.fdopen(descriptor, "w") as output:
+            output.write(serialized + "\n")
+
+    restart = None
+    if args.compose_directory:
+        directory = args.compose_directory.resolve(strict=True)
+        assert args.compose_directory.is_absolute() and (directory / "compose.yaml").is_file()
+        assert re.fullmatch(r"[a-z0-9][a-z0-9_-]*", args.compose_project), "Exact Compose project required"
+
+        def restart():
+            command = ["docker", "compose", "--project-name", args.compose_project,
+                       "--project-directory", str(directory), "--env-file", str(directory / ".env"),
+                       "-f", str(directory / "compose.yaml")]
+            ids = subprocess.run(command + ["ps", "-q", "agent-host"], check=True, capture_output=True, text=True, timeout=30).stdout.split()
+            assert len(ids) == 1, "Exactly one running agent host in the owned project is required"
+            inspect = ["docker", "inspect", "--format", "{{.State.StartedAt}}", ids[0]]
+            before = subprocess.run(inspect, check=True, capture_output=True, text=True, timeout=30).stdout.strip()
+            subprocess.run(command + ["restart", "agent-host"], check=True, capture_output=True, timeout=120)
+            after = subprocess.run(inspect, check=True, capture_output=True, text=True, timeout=30).stdout.strip()
+            assert before and after and before != after, "Agent-host process did not restart"
+            report["restart"] = {"container": ids[0], "before": before, "after": after}
+
+    with httpx2.Client(trust_env=False, timeout=30) as http:
+        client = OpenAI(base_url=base, api_key=key, max_retries=0, _strict_response_validation=True, http_client=http)
+        foreign = OpenAI(base_url=base, api_key=foreign_key, max_retries=0, _strict_response_validation=True, http_client=http)
+
+        def ready(session):
+            if placement == "none":
+                return
+            print("Session " + session.id + ": waiting for its Environment; install self-hosted Sessions through their public installation command.", flush=True)
+            deadline = time.monotonic() + 300
+            while time.monotonic() < deadline:
+                current = client.beta.agents.environments.retrieve(session.environment.id)
+                assert current.status != "failed", "Environment preparation failed"
+                if current.status == "connected":
+                    return
+                time.sleep(0.5)
+            raise AssertionError("Environment did not connect; use a separate sandbox for each Session")
+
+        session_options = {"environment": environment, "extra_body": {"x_agents_core": {"model_provider": provider}}}
+        suites = {"none": verify_none, "functions": verify_hosted_functions, "pending-actions": verify_pending_actions,
+                  "images": verify_workspace_images, "structured": verify_hosted_structured}
+        try:
+            kwargs = {"ready": ready, "record": record}
+            if args.suite != "pending-actions":
+                kwargs["restart"] = restart
+            checks = suites[args.suite](client, foreign, http, agent, session_options, **kwargs)
+            report.update(status="passed", checks=checks)
+            if restart is not None:
+                report["cold_recovery"] = "passed"
+        except Exception:
+            report["status"] = "failed"
+            raise
+        finally:
+            record(report.get("proof", {}))
+    print("Passed " + args.suite + "; cold recovery: " + report["cold_recovery"], flush=True)
+
+
+if __name__ == "__main__":
+    try:
+        main()
+    except Exception as error:
+        locations = " -> ".join(f"{Path(frame.filename).name}:{frame.lineno}" for frame in traceback.extract_tb(error.__traceback__))
+        raise SystemExit(f"Public acceptance failed ({type(error).__name__} at {locations}); response bodies withheld.") from None
diff --git a/services/core/tests/qualify_public_native_test.py b/services/core/tests/qualify_public_native_test.py
new file mode 100644
index 000000000..d0b734b94
--- /dev/null
+++ b/services/core/tests/qualify_public_native_test.py
@@ -0,0 +1,134 @@
+"""Check qualification's secret handling and owned restart boundary without a model."""
+
+import contextlib
+import io
+import json
+import os
+from pathlib import Path
+import subprocess
+import tempfile
+import unittest
+from types import SimpleNamespace
+from unittest.mock import MagicMock, patch
+
+import httpx2
+from openai import BadRequestError, OpenAI
+
+from official_environment_files_native import generate_files
+
+import qualify_public_native as qualification
+
+
+class QualificationTests(unittest.TestCase):
+    def setUp(self):
+        root = Path.home() / ".oac/tests"
+        root.mkdir(parents=True, exist_ok=True)
+        self.temp = tempfile.TemporaryDirectory(prefix="public-qualification-", dir=root)
+        self.addCleanup(self.temp.cleanup)
+        self.root = Path(self.temp.name)
+        self.settings = {"agent": {"model": "fixture", "x_agents_core": {"harness": "codex"}},
+                         "model_provider": {"protocol": "responses", "base_url": "https://model.example/v1", "api_key": "provider-secret"},
+                         "environment": {"type": "none"}}
+        for name, value in (("settings.json", json.dumps(self.settings)), ("foreign.key", "foreign-secret")):
+            path = self.root / name
+            path.write_text(value)
+            path.chmod(0o600)
+        self.evidence = self.root / "evidence.json"
+        self.argv = ["qualify_public_native.py", "--settings", str(self.root / "settings.json"),
+                     "--foreign-key-file", str(self.root / "foreign.key"), "--suite", "none", "--evidence", str(self.evidence)]
+        self.enterContext(patch.dict(os.environ, {"OPENAI_BASE_URL": "https://core.example/v1", "OPENAI_API_KEY": "project-secret"}))
+        self.enterContext(patch.object(qualification, "OpenAI"))
+        self.enterContext(contextlib.redirect_stdout(io.StringIO()))
+
+    def run_suite(self, suite):
+        with patch("sys.argv", self.argv), patch.object(qualification, "verify_none", suite):
+            qualification.main()
+
+    def test_explicit_provider_reaches_suite_and_warm_run_does_not_claim_recovery(self):
+        def suite(client, foreign, http, agent, options, ready, restart, record):
+            self.assertEqual(agent, self.settings["agent"])
+            self.assertEqual(options["extra_body"]["x_agents_core"]["model_provider"], self.settings["model_provider"])
+            self.assertIsNone(restart)
+            record({"checks": ["warm"]})
+            return ["warm"]
+        self.run_suite(suite)
+        proof = json.loads(self.evidence.read_text())
+        self.assertEqual((proof["status"], proof["cold_recovery"]), ("passed", "unverified"))
+        self.assertEqual(self.evidence.stat().st_mode & 0o777, 0o600)
+        self.assertNotIn("provider-secret", self.evidence.read_text())
+
+    def test_leaked_credential_is_never_written_as_evidence(self):
+        def suite(*args, record, **kwargs):
+            record({"leak": "provider-secret"})
+        with self.assertRaisesRegex(AssertionError, "Credential"):
+            self.run_suite(suite)
+        self.assertFalse(self.evidence.exists())
+
+    def test_custom_workspace_uses_physical_tools_and_logical_public_file_paths(self):
+        client = MagicMock()
+        session = SimpleNamespace(status="idle", required_actions=[], environment=SimpleNamespace(
+            type="self_hosted", id="environment", workspace_directory="/custom/work"))
+        client.beta.agents.sessions.retrieve.return_value = session
+        client.beta.agents.environments.retrieve.return_value = SimpleNamespace(status="connected")
+        client.beta.agents.sessions.turns.list.side_effect = [[], [SimpleNamespace(id="turn", status="completed")]]
+        fixture = generate_files(client, "session", "custom")
+        prompt = client.beta.agents.sessions.events.create.call_args.kwargs["events"][0]["input"][0]["content"][0]["text"]
+        self.assertIn("/custom/work/files-list-custom-", prompt)
+        self.assertNotIn("/workspace/", prompt)
+        self.assertTrue(fixture["directory"].startswith("/workspace/files-list-custom-"))
+        self.assertTrue(fixture["sibling_directory"].startswith("/workspace/files-list-custom-"))
+        self.assertTrue(all(path.startswith(fixture["directory"] + "/") for path in fixture["expected"]))
+        self.assertTrue(all(path.startswith(fixture["sibling_directory"] + "/") for path in fixture["sibling_expected"]))
+        self.assertNotIn("/custom/work", json.dumps(fixture))
+
+    def test_pinned_sdk_serializes_explicit_session_selection(self):
+        bodies = []
+
+        def reject(request):
+            bodies.append(json.loads(request.content))
+            return httpx2.Response(400, json={"error": {"type": "invalid_request_error", "code": "fixture", "message": "stop before execution"}})
+
+        with httpx2.Client(transport=httpx2.MockTransport(reject)) as http:
+            client = OpenAI(base_url="https://core.example/v1", api_key="project-secret", http_client=http, max_retries=0)
+            options = {"environment": {"type": "self_hosted", "workspace_directory": "/custom/work"},
+                       "extra_body": {"x_agents_core": {"model_provider": self.settings["model_provider"]}}}
+            for verify in (qualification.verify_hosted_functions, qualification.verify_workspace_images, qualification.verify_pending_actions):
+                kwargs = {"ready": lambda session: None, "record": lambda proof: None}
+                if verify is not qualification.verify_pending_actions:
+                    kwargs["restart"] = None
+                with self.assertRaises(BadRequestError):
+                    verify(client, client, http, self.settings["agent"], options, **kwargs)
+                self.assertEqual(bodies[-1]["environment"], options["environment"])
+                self.assertEqual(bodies[-1]["agent"]["x_agents_core"], self.settings["agent"]["x_agents_core"])
+                self.assertEqual(bodies[-1]["x_agents_core"]["model_provider"], self.settings["model_provider"])
+            with self.assertRaises(BadRequestError):
+                qualification.verify_hosted_structured(client, client, http, self.settings["agent"], options,
+                    ready=lambda session: None, restart=None, record=lambda proof: None)
+            self.assertEqual(bodies[-1]["x_agents_core"], self.settings["agent"]["x_agents_core"])
+
+    def test_restart_requires_a_running_host_and_observed_new_process(self):
+        (self.root / "compose.yaml").write_text("services: {}\n")
+        self.argv += ["--compose-directory", str(self.root), "--compose-project", "owned-qualification"]
+
+        def suite(*args, restart, record, **kwargs):
+            restart()
+            record({"checks": ["cold"]})
+            return ["cold"]
+
+        results = [subprocess.CompletedProcess([], 0, value) for value in ("container-id\n", "before\n", "", "after\n")]
+        with patch.object(qualification.subprocess, "run", side_effect=results) as run:
+            self.run_suite(suite)
+            command = run.call_args_list[2].args[0]
+            self.assertEqual(command[-2:], ["restart", "agent-host"])
+            self.assertEqual(command[command.index("--project-name") + 1], "owned-qualification")
+            self.assertFalse(any(call.kwargs.get("shell") for call in run.call_args_list))
+        self.assertEqual(json.loads(self.evidence.read_text())["cold_recovery"], "passed")
+        self.evidence.unlink()
+        with patch.object(qualification.subprocess, "run", return_value=subprocess.CompletedProcess([], 0, "")):
+            with self.assertRaisesRegex(AssertionError, "Exactly one"):
+                self.run_suite(suite)
+        self.assertEqual(json.loads(self.evidence.read_text())["status"], "failed")
+
+
+if __name__ == "__main__":
+    unittest.main()
diff --git a/services/core/tests/session_cleanup.py b/services/core/tests/session_cleanup.py
index 3560948f3..3a76036c8 100644
--- a/services/core/tests/session_cleanup.py
+++ b/services/core/tests/session_cleanup.py
@@ -36,4 +36,4 @@ def delete_session(sessions, session_id, timeout=60):
     except Exception as exc:
         if original is None:
             raise
-        print(f"Session {session_id} cleanup failed after an earlier error: {exc!r}", file=sys.stderr)
+        print(f"Session {session_id} cleanup failed after an earlier error: {type(exc).__name__}", file=sys.stderr)