From 1fa031089a45efab54959eb581e095147e8115ce Mon Sep 17 00:00:00 2001
From: SaladDay <1203511142@qq.com>
Date: Thu, 8 Oct 2026 14:06:14 +0000
Subject: [PATCH 1/2] Qualify Environment composition through Core

---
 contracts/agents-api/harness-onboarding.md    |   9 +-
 contracts/agents-api/zh/harness-onboarding.md |  11 +-
 .../tests/official_environment_composition.py | 156 ++++++++++++++----
 .../official_environment_initial_files.py     |  34 +---
 .../tests/official_environment_plugin_mcp.py  |  62 ++-----
 .../core/tests/official_environment_setup.py  | 153 ++---------------
 .../official_environment_skill_references.py  |   4 +-
 .../core/tests/official_environment_skills.py |  13 +-
 services/core/tests/qualify_public_native.py  |   7 +-
 .../core/tests/qualify_public_native_test.py  |  87 ++++++++++
 services/core/tests/session_cleanup.py        |   9 +
 11 files changed, 281 insertions(+), 264 deletions(-)

diff --git a/contracts/agents-api/harness-onboarding.md b/contracts/agents-api/harness-onboarding.md
index dad44cd80..8a22dc571 100644
--- a/contracts/agents-api/harness-onboarding.md
+++ b/contracts/agents-api/harness-onboarding.md
@@ -195,7 +195,7 @@ Before starting, record the operation set, expected results, exclusions and stop
 4. **Regression.** Existing Harnesses keep working. Run targeted tests, then `make check`; run `make openapi` after API changes and `make sqlc-generate` after query changes.
 5. **Review.** Follow the [blind review workflow](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/CONTRIBUTING.md#review).
 
-Environment acceptance uses `services/core/tests/official_environment_{templates,setup,skills,plugins,plugin_mcp,composition,initial_files,network,skill_references}.py`. For composed preparation, change the Skill default and Template, delete the sources, retry and restart; verify frozen bytes, one setup execution and MCP cancellation. `official_hosted_structured_native.py` covers hosted structured output. Record exact source revisions, native versions and commands with each acceptance result.
+Environment acceptance uses `services/core/tests/official_environment_{templates,setup,skills,plugins,plugin_mcp,composition,initial_files,network,skill_references}.py`. Run composed preparation through the [public qualifier](#qualify-the-public-path)'s `composition` suite. `official_hosted_structured_native.py` covers hosted structured output. Record exact source revisions, native versions and commands with each acceptance result.
 
 Keep provider keys in private operator files, never in commits or logs. Existing focused tests, relative to `apps/daemon/internal/agent`:
 
@@ -229,8 +229,13 @@ python services/core/tests/qualify_public_native.py \
 | `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 |
+| `composition` | Frozen source snapshots, initial bytes, ordered setup, packages, Skills, stdio MCP, continuation and cancellation | `openai_hosted` |
 
-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.
+For `composition`, settings must supply exactly `{"type":"openai_hosted"}` as `environment`; the suite creates its own Template, file and Skill sources. It checks initial binary bytes, ordered setup, npm and Python packages, uploaded, plugin and directory Skills, and three real stdio MCP tool identities, Items and results. After changing or deleting source resources, it verifies frozen preparation through warm continuation and, when selected, the existing Compose-verified agent-host restart. Cancellation must stop the MCP descendant's file effects. The suite cleans up all resources it creates.
+
+Private-owner and credential isolation remain `unverified`; positive canary evidence requires separate proof using operator-owned resources. The suite does not request `packages.system` or stdio MCP `env_vars`, which the current contracts reject.
+
+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. 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.
 
diff --git a/contracts/agents-api/zh/harness-onboarding.md b/contracts/agents-api/zh/harness-onboarding.md
index fee7ec11f..e4e271fa3 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: a571ca6ea6b43f89f1669be989ecaf52838181705ba5facbc4f66a72cccd9e57
+source_hash: 32b85617efe41a23bd25768a42fd7a57019c75ae723d07d8cc5106124709f402
 ---
 
 **Harness** 是一种运行模型和工具循环的原生代理引擎（Codex、Claude Code、MiniMax Code）。**Harness 适配器**将 Runtime 的 Executor 和 Turn 契约转换到该引擎的 SDK 或协议。本文档定义 Runtime–Harness 协议：适配器接口及其生命周期义务、注册、支持声明和验收。
@@ -197,7 +197,7 @@ Core 会识别[内置 Harness 注册项](harness-catalog.md)。向 `internal/har
 4. **回归。** 现有 Harness 必须继续正常工作。先运行定向测试，然后运行 `make check`；API 更改后运行 `make openapi`，查询更改后运行 `make sqlc-generate`。
 5. **审查。** 遵循 [blind review workflow](https://github.com/MiniMax-AI/OpenAgentCore/blob/main/CONTRIBUTING.md#review)。
 
-Environment 验收使用 `services/core/tests/official_environment_{templates,setup,skills,plugins,plugin_mcp,composition,initial_files,network,skill_references}.py`。对于组合式准备，请更改 Skill 默认值和 Template，删除源文件，重试并重启；验证冻结字节、一次 setup 执行和 MCP 取消。`official_hosted_structured_native.py` 覆盖托管结构化输出。随每项验收结果记录精确源修订版本、原生版本和命令。
+Environment 验收使用 `services/core/tests/official_environment_{templates,setup,skills,plugins,plugin_mcp,composition,initial_files,network,skill_references}.py`。通过[公共验收运行器](#qualify-the-public-path)的 `composition` 套件运行组合式准备验收。`official_hosted_structured_native.py` 覆盖托管结构化输出。随每项验收结果记录精确源修订版本、原生版本和命令。
 
 将 Provider 密钥保存在私有操作员文件中，绝不能放入提交或日志。相对于 `apps/daemon/internal/agent` 的现有定向测试如下：
 
@@ -231,8 +231,13 @@ python services/core/tests/qualify_public_native.py \
 | `functions` | SDK handler、成功/错误、原生文件和 Artifact 字节、继续执行、待处理调用取消和租户隔离 | 工作区 |
 | `images` | 初始和活动图像、图像结果、重试/原子拒绝、原生文件/Artifact、隔离和继续执行 | 声明支持图像和函数的工作区 |
 | `structured` | 保存和内联 schema、函数辅助原生文件、精确 JSON/SSE、取消和文本覆盖 | 声明支持结构化输出和函数的工作区 |
+| `composition` | 冻结的源快照、初始字节、有序 setup、包、Skills、stdio MCP、继续执行和取消 | `openai_hosted` |
 
-对于 `self_hosted`，选择自定义绝对 `workspace_directory`。运行器打印每个新 Session ID 后，使用该 Session 的公共安装命令连接独立的隔离机器或容器；运行器最多等待五分钟。多个 Session 不得共享工作区。使用现有 Environment setup、包和能力断言单独验证准备语义。独立的 `official_environment_files_native.py` 检查接受两个已连接的 self-hosted Session，通过 Environment owner 验证 Files.list 排序、分页和隔离。
+对于 `composition`，设置中的 `environment` 必须恰好为 `{"type":"openai_hosted"}`；套件创建自己的 Template、文件和 Skill 来源。它检查初始二进制字节、有序 setup、npm 和 Python 包、上传的 Skill、插件 Skill 和目录 Skill，以及三个真实 stdio MCP 工具的身份、Item 和结果。更改或删除源资源后，它通过热继续执行验证冻结的准备结果；选用重启时，还通过现有的 Compose 验证 agent-host 重启流程进行检查。取消必须使 MCP 后代进程停止产生文件副作用。套件清理其创建的全部资源。
+
+私有所有者与凭据隔离仍为 `unverified`；肯定性的 canary 证据需要使用操作员拥有的资源独立验证。套件不请求当前契约拒绝的 `packages.system` 或 stdio MCP `env_vars`。
+
+对于 `self_hosted`，选择自定义绝对 `workspace_directory`。运行器打印每个新 Session ID 后，使用该 Session 的公共安装命令连接独立的隔离机器或容器；运行器最多等待五分钟。多个 Session 不得共享工作区。独立的 `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 拒绝会使所选套件失败。
 
diff --git a/services/core/tests/official_environment_composition.py b/services/core/tests/official_environment_composition.py
index 285d02398..6267594a0 100644
--- a/services/core/tests/official_environment_composition.py
+++ b/services/core/tests/official_environment_composition.py
@@ -1,23 +1,27 @@
 """Compose existing real acceptance fixtures; no model or Runtime emulation."""
 import base64
+from contextlib import ExitStack
 import io
 import json
+import time
+import uuid
 import zipfile
 
 from official_environment_initial_files import initial_files, assert_initial_bytes_script
-from official_environment_plugin_mcp import plugin_mcp_fixture, verify_plugin_mcp_metadata
+from official_environment_plugin_mcp import plugin_mcp_fixture, verify_plugin_mcp_metadata, verify_plugin_mcp_items
 from official_environment_setup import setup_configuration, native_setup_script, verify_setup_metadata
 from official_environment_skill_references import upload_reference_skill
 from official_environment_skills import inline_skill
+from official_session_artifacts import verify_session_artifacts
+from session_cleanup import delete_session, delete_source
 
 
-def composition_fixture(client, foreign, http, agent, proxy):
+def composition_fixture(client, foreign, http, agent, model_provider, cleanup):
     """The runner owns source cleanup as well as Session/Template cleanup."""
     fixture = plugin_mcp_fixture(installed_dependencies=True)
-    environment, source_id, initial = initial_files(client, foreign, http, agent)
-    skill, expected = upload_reference_skill(client)
-    setup, marker = setup_configuration(proxy)
-    setup['packages']['system'] = ['jq']
+    environment, source_id, initial = initial_files(client, foreign, http, agent, model_provider, cleanup)
+    skill, expected = upload_reference_skill(client, cleanup)
+    setup, marker = setup_configuration()
     fixture['env'].update(setup['env'])
     fixture['packages'] = setup['packages']
     fixture['setup_commands'] = setup['setup_commands'] + fixture['setup_commands']
@@ -30,32 +34,21 @@ def composition_fixture(client, foreign, http, agent, proxy):
     fixture['source_skill_id'] = skill.id
     fixture['setup_configuration'] = setup
     proof = {'marker': fixture['marker'], 'checks': ['initial_files', 'ordered_setup',
-             'caller_env', 'system_npm_python', 'uploaded_skill', 'readonly_system_root', 'private_paths_denied']}
+             'caller_env', 'npm_python', 'uploaded_skill']}
+    fixture['composition_proof'] = proof
     script = native_setup_script(marker) + assert_initial_bytes_script(initial) + '''
-import json
-assert subprocess.check_output(['jq', '-r', '.value'], input=b'{"value":42}').strip() == b'42'
-private_paths = json.loads(Path('/workspace/plugin-isolation-paths.json').read_text())
-assert set(private_paths) == {'staging', 'native_history', 'daemon_auth'}
-for path in private_paths.values():
-    try:
-        with open(path, 'rb') as stream:
-            exposed = stream.read(1)
-    except OSError:
-        continue
-    assert not exposed, 'private Runtime data readable by uploaded Skill'
-try:
-    Path('/usr/bin/jq').open('r+b').close()
-except OSError:
-    pass
-else:
-    raise AssertionError('system package root is writable')
-Path('/workspace/outputs').mkdir(exist_ok=True)
+assert Path('plugin-mcp-setup-count').read_text() == '1'
+Path('outputs').mkdir(exist_ok=True)
+'''
+    script += 'import json\nproof = ' + repr(proof) + '''
+counter = Path('composition-run-count')
+proof['run'] = int(counter.read_text()) + 1 if counter.exists() else 1
+counter.write_text(str(proof['run']))
+Path('outputs/composition.json').write_text(json.dumps(proof, sort_keys=True) + '\\n')
 '''
-    script += 'Path("/workspace/outputs/composition.json").write_text(' + repr(
-        json.dumps(proof, sort_keys=True) + '\n') + ')\n'
     fixture['files'].append({'type': 'inline', 'path': '/workspace/verify.py',
                              'data': base64.b64encode(script.encode()).decode()})
-    fixture['outputs']['/workspace/outputs/composition.json'] = (json.dumps(proof, sort_keys=True) + '\n').encode()
+    fixture['outputs']['/workspace/outputs/composition.json'] = (json.dumps({**proof, 'run': 1}, sort_keys=True) + '\n').encode()
     fixture['outputs']['/workspace/outputs/skill-proof.txt'] = expected
     fixture['prompt'] = ('Use the installed proof-skill Skill and run its packaged script. '
                          'It must verify initialized files, environment, setup and installed dependencies. '
@@ -85,6 +78,11 @@ def verify_composition_metadata(client, http, session, fixture):
         assert {value['path']: value['size_bytes'] for value in environment['files']} == fixture['initial_sizes']
         assert len({value['id'] for value in environment['files']}) == len(fixture['files'])
         assert 'env' not in environment and 'setup_commands' not in environment
+        files = {value['path']: value for value in environment['files']}
+        inline = files['/workspace/initial-inline.bin']
+        source = files['/workspace/initial-source.bin']
+        assert inline['type'] == 'inline' and 'data' not in inline and 'file_id' not in inline
+        assert source['type'] == 'file_id' and source['file_id'] == fixture['source_file_id']
 
 
 def change_and_delete_sources(client, fixture, template_id=None):
@@ -102,3 +100,105 @@ def change_and_delete_sources(client, fixture, template_id=None):
         assert updated.skills == [] and updated.plugins == [] and updated.files == []
     client.files.delete(fixture['source_file_id'])
     client.skills.delete(fixture['source_skill_id'])
+
+
+def verify_composition(client, foreign, http, agent_options, session_options, ready, restart, record):
+    """Qualify installed composition through the public Core path, without a deployment rig."""
+    assert session_options['environment'] == {'type': 'openai_hosted'}, 'Composition requires a fresh hosted Environment'
+    sessions = client.beta.agents.sessions
+    templates = client.beta.agents.environments.templates
+    proof = {'checks': [], 'runs': [], 'unverified': ['Private-owner and credential isolation require positive operator evidence.'],
+             'unsupported': ['packages.system', 'Plugin stdio env_vars']}
+    with ExitStack() as cleanup:
+        fixture = composition_fixture(client, foreign, http, agent_options,
+            session_options['extra_body']['x_agents_core']['model_provider'], cleanup)
+        template = templates.create(**fixture['configuration'])
+        cleanup.callback(delete_source, templates, template.id)
+        session = sessions.create(agent=agent_options, **{**session_options, 'environment': {
+            'type': 'openai_hosted', 'environment_template_id': template.id}})
+        cleanup.callback(delete_session, sessions, session.id)
+        proof['session'] = session.id
+        try:
+            ready(session)
+            expected_artifacts = {}
+            stages = ['initial', 'warm'] + (['cold'] if restart is not None else [])
+            for run_number, stage in enumerate(stages, 1):
+                if stage == 'warm':
+                    change_and_delete_sources(client, fixture, template.id)
+                    templates.delete(template.id)
+                elif stage == 'cold':
+                    restart()
+                    ready(session)
+                current = sessions.retrieve(session.id)
+                verify_composition_metadata(client, http, current, fixture)
+                prompt = fixture['prompt'] + ' Run all checks again even if proof files already exist; do not reuse an earlier answer.'
+                with sessions.stream(session.id, input=prompt, idempotency_key=uuid.uuid4().hex, timeout=600) as stream:
+                    events = [event.to_dict() for event in stream]
+                proof['runs'].append({'stage': stage, 'events': events})
+                record(proof)
+                terminal = [event for event in events if event['type'] in {
+                    'agent.session.turn.completed', 'agent.session.turn.failed', 'agent.session.turn.cancelled'}]
+                assert len(terminal) == 1 and terminal[0]['type'] == 'agent.session.turn.completed'
+                assert events[-1]['type'] == 'agent.session.idle'
+                turn = terminal[0]['turn']['id']
+                run = verify_plugin_mcp_items(client, http, session.id, turn, fixture, events=events)
+                proof['runs'][-1]['mcp'] = run
+                if stage == 'initial':
+                    expected_artifacts[turn] = dict(fixture['outputs'])
+                else:
+                    expected_artifacts[turn] = {'/workspace/outputs/composition.json': (json.dumps(
+                        {**fixture['composition_proof'], 'run': run_number}, sort_keys=True) + '\n').encode()}
+                verify_session_artifacts(client, foreign, http, session.id, session.environment.id, expected_artifacts)
+                rows = client.beta.agents.environments.files.list(session.environment.id, path='/workspace/outputs')
+                assert {row.path: row.size_bytes for row in rows} == {path: len(body) for path, body in fixture['outputs'].items()}
+                assert sessions.retrieve(session.id).required_actions == []
+                proof['checks'].append(stage + '_composition_bytes_mcp_and_frozen_sources')
+                record(proof)
+
+            before = {turn.id for turn in sessions.turns.list(session.id)}
+            sessions.events.create(session.id, events=[{'type': 'agent.session.input.message', 'input': [
+                {'role': 'user', 'content': [{'type': 'input_text', 'text': fixture['hold_prompt']}]}]}],
+                                   idempotency_key=uuid.uuid4().hex)
+            deadline = time.monotonic() + 180
+            observed = None
+            while time.monotonic() < deadline:
+                turns = [turn for turn in sessions.turns.list(session.id) if turn.id not in before]
+                assert len(turns) <= 1 and not any(turn.status in {'completed', 'failed', 'cancelled'} for turn in turns)
+                rows = client.beta.agents.environments.files.list(session.environment.id, path='/workspace/plugin-mcp-hold')
+                sizes = {row.path: row.size_bytes for row in rows}
+                size = sizes.get(fixture['hold_paths']['ticks'], 0)
+                if observed is not None and size > observed and fixture['hold_paths']['invocation'] in sizes:
+                    break
+                if size:
+                    observed = size
+                time.sleep(0.5)
+            else:
+                raise AssertionError('Native MCP descendant did not produce growing effects')
+            turn_id = turns[0].id
+            key = uuid.uuid4().hex
+            for _ in range(2):
+                sessions.events.create(session.id, events=[{'type': 'agent.session.input.cancel'}], idempotency_key=key)
+            deadline = time.monotonic() + 120
+            while time.monotonic() < deadline:
+                turn = sessions.turns.retrieve(turn_id, session_id=session.id)
+                if turn.status == 'cancelled' and sessions.retrieve(session.id).status == 'idle':
+                    break
+                assert turn.status not in {'completed', 'failed'}
+                time.sleep(0.5)
+            else:
+                raise AssertionError('MCP cancellation did not settle')
+            settled = {row.path: row.size_bytes for row in client.beta.agents.environments.files.list(
+                session.environment.id, path='/workspace/plugin-mcp-hold')}
+            assert settled[fixture['hold_paths']['ticks']] >= size
+            time.sleep(3)
+            assert {row.path: row.size_bytes for row in client.beta.agents.environments.files.list(
+                session.environment.id, path='/workspace/plugin-mcp-hold')} == settled, 'MCP effects continued after cancellation'
+            proof['release'] = verify_plugin_mcp_items(client, http, session.id, turn_id, fixture,
+                tool='hold', servers=[fixture['hold_server']], expected_status='incomplete', expected_turn_status='cancelled')
+            proof['release']['stable_seconds'] = 3
+            proof['release']['ticks_before_cancel_bytes'] = size
+            proof['release']['ticks_after_cancel_bytes'] = settled[fixture['hold_paths']['ticks']]
+            proof['checks'].append('native_mcp_cancel_retry_stops_descendant_effects')
+            return proof['checks']
+        finally:
+            record(proof)
diff --git a/services/core/tests/official_environment_initial_files.py b/services/core/tests/official_environment_initial_files.py
index a5cbfca70..6b7e7b82f 100644
--- a/services/core/tests/official_environment_initial_files.py
+++ b/services/core/tests/official_environment_initial_files.py
@@ -1,13 +1,15 @@
 """Real-deployment checks for confidential initial files and frozen metadata."""
 import base64
-import json
 import secrets
 
+from session_cleanup import delete_source
 
-def initial_files(client, foreign, http, agent, template_id=None):
+
+def initial_files(client, foreign, http, agent, model_provider, cleanup):
     inline = secrets.token_bytes(40)
     source_body = bytes(range(256))
     source = client.files.create(file=('initial.bin', source_body), purpose='user_data')
+    cleanup.callback(delete_source, client.files, source.id)
     files = [{'type': 'inline', 'path': '/workspace/initial-inline.bin',
               'data': base64.b64encode(inline).decode()},
              {'type': 'file_id', 'path': '/workspace/initial-source.bin', 'file_id': source.id}]
@@ -15,34 +17,12 @@ def initial_files(client, foreign, http, agent, template_id=None):
     headers = {'Authorization': 'Bearer ' + client.api_key, 'OpenAI-Beta': 'agents=v1'}
     foreign_headers = {**headers, 'Authorization': 'Bearer ' + foreign.api_key}
     attempted = http.post(endpoint + '/agents/sessions', headers=foreign_headers,
-                          json={'agent': agent, 'environment': {'type': 'openai_hosted', 'files': [files[1]]}})
+                          json={'agent': agent, 'x_agents_core': {'model_provider': model_provider}, 'environment': {'type': 'openai_hosted', 'files': [files[1]]}})
     assert attempted.status_code == 404 and source.id not in attempted.text
-    if template_id:
-        response = client.beta.agents.environments.templates.with_raw_response.update(template_id, files=files)
-        body = response.http_response.json()
-        assert body['files'] == [{'type': 'inline', 'path': files[0]['path'], 'size_bytes': len(inline)},
-                                 {'type': 'file_id', 'path': files[1]['path'], 'file_id': source.id}]
-        assert files[0]['data'] not in json.dumps(body)
-        listing = client.beta.agents.environments.templates.list().to_dict()
-        assert files[0]['data'] not in json.dumps(listing)
-        environment = {'type': 'openai_hosted', 'environment_template_id': template_id}
-    else:
-        environment = {'type': 'openai_hosted', 'network': {'access': 'disabled'}, 'files': files}
+    environment = {'type': 'openai_hosted', 'network': {'access': 'enabled'}, 'files': files}
     return environment, source.id, {files[0]['path']: inline, files[1]['path']: source_body}
 
 
-def verify_initial_snapshot(client, http, session, expected, source_id):
-    metadata = [value.to_dict() for value in session.environment.files]
-    assert len(metadata) == 2 and len({value['id'] for value in metadata}) == 2
-    assert {value['path']: value['size_bytes'] for value in metadata} == {p: len(b) for p, b in expected.items()}
-    assert metadata[0]['type'] == 'inline' and 'data' not in metadata[0] and 'file_id' not in metadata[0]
-    assert metadata[1]['type'] == 'file_id' and metadata[1]['file_id'] == source_id
-    endpoint = str(client.base_url).rstrip('/') + '/agents/environments/' + session.environment.id
-    response = http.get(endpoint, headers={'Authorization': 'Bearer ' + client.api_key, 'OpenAI-Beta': 'agents=v1'})
-    assert response.status_code == 200 and response.json()['files'] == metadata
-    client.files.delete(source_id)
-
-
 def assert_initial_bytes_script(expected):
-    return ''.join('assert Path(' + repr(path) + ').read_bytes() == bytes.fromhex(' + repr(body.hex()) + ')\n'
+    return ''.join('assert Path(' + repr(path.removeprefix('/workspace/')) + ').read_bytes() == bytes.fromhex(' + repr(body.hex()) + ')\n'
                    for path, body in expected.items())
diff --git a/services/core/tests/official_environment_plugin_mcp.py b/services/core/tests/official_environment_plugin_mcp.py
index 5a33298b9..8368377be 100644
--- a/services/core/tests/official_environment_plugin_mcp.py
+++ b/services/core/tests/official_environment_plugin_mcp.py
@@ -1,13 +1,12 @@
 """Fixtures for real native Plugin MCP acceptance through the pinned public API.
 
-The caller owns Core, Runtime, positive private canaries, model calls, recovery and
+The caller owns Core, Runtime, model calls, recovery and
 cancellation. Reuse official_environment_files.verify_environment_files with
 {path: len(body) for path, body in fixture['outputs'].items()}, and
 official_session_artifacts.verify_session_artifacts with those same bytes.
 These helpers never emulate a model or claim native acceptance from fixture checks.
 """
 import base64
-import hashlib
 import io
 import json
 import secrets
@@ -17,9 +16,8 @@
 from official_environment_plugins import verify_plugin_resources
 
 
-_CHECKS = ['selected_user_env', 'unselected_env_absent', 'native_env_absent',
-           'package_cwd', 'private_contents_denied', 'installed_package_readonly']
-_SERVER = r'''import hashlib, json, os, subprocess, sys
+_CHECKS = ['native_env_absent', 'package_cwd']
+_SERVER = r'''import json, os, subprocess, sys
 from pathlib import Path
 
 root = Path(__file__).resolve().parent
@@ -27,41 +25,14 @@
 
 def verify(marker):
     assert marker == config['marker'], 'wrong invocation marker'
-    selected = os.environ.get('PLUGIN_MCP_SELECTED', '')
-    assert selected and hashlib.sha256(selected.encode()).hexdigest() == config['selected_sha256'], 'selected user env missing or wrong'
-    assert 'PLUGIN_MCP_UNSELECTED' not in os.environ, 'unselected user env present'
     prefixes = ('PARSAR_', 'CODEX_', 'OPENAI_', 'ANTHROPIC_', 'MINIMAX_', 'MOONSHOT_', 'KIMI_')
     assert not any(name.startswith(prefixes) or name == 'NATIVE_MCP_CANARY' for name in os.environ), 'native environment present'
     assert Path.cwd() == root / 'resources', 'relative package cwd not applied'
     assert json.loads(Path('proof.json').read_text()) == config, 'cwd resource differs'
-    private_paths = json.loads(Path('/workspace/plugin-isolation-paths.json').read_text())
-    assert set(private_paths) == {'staging', 'native_history', 'daemon_auth'}, 'positive private paths missing'
-    assert all(isinstance(path, str) and path.startswith('/') for path in private_paths.values()), 'invalid private paths'
-    for path in private_paths.values():
-        try:
-            with open(path, 'rb') as stream:
-                exposed = stream.read(1)
-        except OSError:
-            continue
-        assert not exposed, 'private Runtime content readable'
-    for path in [root / '.codex-plugin/plugin.json', root / '.mcp.json', root / 'resources/proof.json']:
-        try:
-            with path.open('r+b'):
-                pass
-        except OSError:
-            continue
-        raise AssertionError('installed package writable')
     if config.get('installed_dependencies'):
         import packaging
         assert packaging.__version__ == '26.0', 'installed Python dependency missing'
         assert subprocess.check_output(['semver', '1.2.3']).strip() == b'1.2.3', 'installed npm dependency missing'
-        assert subprocess.check_output(['jq', '-r', '.value'], input=b'{"value":42}').strip() == b'42', 'installed system dependency missing'
-        try:
-            Path('/usr/bin/jq').open('r+b').close()
-        except OSError:
-            pass
-        else:
-            raise AssertionError('installed system root writable')
     return {'marker': marker, 'server': config['server'], 'checks': config['checks']}
 
 def invoke(name, arguments):
@@ -122,7 +93,6 @@ def invoke(name, arguments):
 def _skill(name, marker, output):
     script = ("from pathlib import Path\n"
               "root = Path(__file__).resolve().parent\n"
-              "assert '/initialization/capabilities/' in str(root)\n"
               "Path('/workspace/outputs').mkdir(exist_ok=True)\n"
               "Path(" + repr(output) + ").write_bytes((root / 'proof.txt').read_bytes())\n"
               "print('INSTALLED_PLUGIN_SKILL_VERIFIED')\n")
@@ -132,13 +102,11 @@ def _skill(name, marker, output):
     return {'SKILL.md': manifest, 'check.py': script, 'proof.txt': marker + '\n'}
 
 
-def _package(server, marker, selected, skill=False, *, installed_dependencies=False):
+def _package(server, marker, skill=False, *, installed_dependencies=False):
     manifest = {'name': server, 'description': 'Native MCP isolation proof.', 'mcpServers': './.mcp.json'}
     files = {'.mcp.json': json.dumps({'mcpServers': {server: {
-        'command': 'python3', 'args': ['../server.py'], 'cwd': 'resources',
-        'env_vars': ['PLUGIN_MCP_SELECTED']}}}), 'server.py': _SERVER,
+        'command': 'python3', 'args': ['../server.py'], 'cwd': 'resources'}}}), 'server.py': _SERVER,
         'resources/proof.json': json.dumps({'server': server, 'marker': marker,
-            'selected_sha256': hashlib.sha256(selected.encode()).hexdigest(),
             'installed_dependencies': installed_dependencies,
             'checks': _CHECKS + (['installed_dependencies'] if installed_dependencies else [])})}
     if skill:
@@ -162,23 +130,21 @@ def _inline_plugin(name, files):
 def plugin_mcp_fixture(*, installed_dependencies=False):
     """Return one hosted configuration and exact expected public proof bytes.
 
-    Before a native Turn, the runner must create nonempty private canary files
-    outside tool authority and publish only their paths in plugin-isolation-paths.json.
-    Never log the returned env or source bodies. The marker itself is nonsecret.
+    Private-owner isolation needs a separate positive operator check.
+    Never log source bodies. The invocation marker itself is nonsecret.
     """
     marker = 'plugin-mcp-proof-' + secrets.token_hex(20)
-    env = {'PLUGIN_MCP_SELECTED': 'selected-' + secrets.token_hex(24),
-           'PLUGIN_MCP_UNSELECTED': 'unselected-' + secrets.token_hex(24)}
+    env = {}
     servers = ['mcp_only_proof', 'combined_proof', 'generated_proof']
-    plugins = [_inline_plugin(name, _package(name, marker, env['PLUGIN_MCP_SELECTED'], skill=index == 1,
+    plugins = [_inline_plugin(name, _package(name, marker, skill=index == 1,
                        installed_dependencies=installed_dependencies))
                for index, name in enumerate(servers[:2])]
     exact, parent = '/workspace/generated/mcp-exact', '/workspace/generated/skill-parent'
     generated = {exact + '/' + path: body for path, body in _package(
-        servers[2], marker, env['PLUGIN_MCP_SELECTED'],
+        servers[2], marker,
         installed_dependencies=installed_dependencies).items()}
     # Selecting the parent discovers a nested Skill, but never the child's MCP.
-    child = _package('unselected_child_mcp', marker, env['PLUGIN_MCP_SELECTED'])
+    child = _package('unselected_child_mcp', marker)
     child_manifest = json.loads(child['.codex-plugin/plugin.json'])
     child_manifest['skills'] = ['./skills']
     child['.codex-plugin/plugin.json'] = json.dumps(child_manifest)
@@ -189,9 +155,9 @@ def plugin_mcp_fixture(*, installed_dependencies=False):
     initial = [{'type': 'inline', 'path': '/workspace/plugin-mcp-seed.json',
                 'data': base64.b64encode(seed.encode()).decode()}]
     setup = ("import json\nfrom pathlib import Path\n"
-             "for name, body in json.loads(Path('/workspace/plugin-mcp-seed.json').read_text()).items():\n"
-             " p = Path(name)\n p.parent.mkdir(parents=True, exist_ok=True)\n p.write_text(body)\n"
-             "p = Path('/workspace/plugin-mcp-setup-count')\n"
+             "for name, body in json.loads(Path('plugin-mcp-seed.json').read_text()).items():\n"
+             " p = Path(name.removeprefix('/workspace/'))\n p.parent.mkdir(parents=True, exist_ok=True)\n p.write_text(body)\n"
+             "p = Path('plugin-mcp-setup-count')\n"
              "p.write_text(str(int(p.read_text()) + 1) if p.exists() else '1')\n")
     checks = _CHECKS + (['installed_dependencies'] if installed_dependencies else [])
     outputs = {'/workspace/outputs/' + server + '.json': (json.dumps(
diff --git a/services/core/tests/official_environment_setup.py b/services/core/tests/official_environment_setup.py
index ed3b225b0..c47cb0a38 100644
--- a/services/core/tests/official_environment_setup.py
+++ b/services/core/tests/official_environment_setup.py
@@ -2,56 +2,28 @@
 import json
 import secrets
 
-from openai import NotFoundError
 
-
-def setup_configuration(proxy=None, *, system_packages=False):
+def setup_configuration():
     marker = 'setup-private-' + secrets.token_hex(16) + "' $()"
     env = {'SETUP_VALUE': marker}
-    if proxy:
-        env.update(HTTPS_PROXY=proxy, HTTP_PROXY=proxy, https_proxy=proxy, http_proxy=proxy)
-    first = """test -f /workspace/initial-inline.bin && mkdir -p /workspace/setup-sub && python3 - <<'SCRIPT'
+    first = """test -f initial-inline.bin && mkdir -p setup-sub && python3 - <<'SCRIPT'
 import os
 from pathlib import Path
 import packaging
 assert packaging.__version__ == '26.0'
 assert os.environ['SETUP_VALUE']
-Path('/workspace/setup-sub/order').write_text('first')
+Path('setup-sub/order').write_text('first')
 SCRIPT
-semver 1.2.3 > /workspace/setup-version
+semver 1.2.3 > setup-version
 """
-    second = "test \"$(cat order)\" = first && printf second > order && printf initialized > /workspace/setup-once"
+    second = "test \"$(cat order)\" = first && printf second > order && printf initialized > ../setup-once"
     packages = {'npm': ['semver@7.7.2'], 'python': ['packaging==26.0']}
-    if system_packages:
-        packages['system'] = ['jq', 'build-essential', 'libpq-dev']
-        first = """printf '{"value":42}' | jq -e '.value == 42' &&
-printf '#include <libpq-fe.h>\\nint main(void){return PQlibVersion() > 0 ? 0 : 1;}\\n' > /workspace/system-library.c &&
-cc -I/usr/include/postgresql /workspace/system-library.c -lpq -o /workspace/system-library &&
-/workspace/system-library &&
-""" + first
     return {
         'env': env, 'packages': packages,
         'setup_commands': [{'command': first}, {'command': second, 'cwd': '/workspace/setup-sub'}],
     }, marker
 
 
-def attach_setup(client, foreign, http, environment, configuration, template_id=None, network="disabled"):
-    endpoint = str(client.base_url).rstrip('/')
-    if template_id:
-        raw = client.beta.agents.environments.templates.with_raw_response.update(
-            template_id, network={'access': network}, **configuration)
-        resource = raw.http_response.json()
-        assert resource['packages'] == {'system': [], **configuration['packages']}
-        headers = {'Authorization': 'Bearer ' + foreign.api_key, 'OpenAI-Beta': 'agents=v1'}
-        rejected = http.get(endpoint + '/agents/environments/templates/' + template_id, headers=headers)
-        assert rejected.status_code == 404
-        metadata = [resource, client.beta.agents.environments.templates.list().to_dict()]
-        assert configuration['env']['SETUP_VALUE'] not in json.dumps(metadata)
-        assert all('env' not in value and 'setup_commands' not in value for value in [resource])
-        return environment
-    return {**environment, 'network': {'access': network}, **configuration}
-
-
 def verify_setup_metadata(client, session, configuration):
     resource = client.beta.agents.environments.retrieve(session.environment.id).to_dict()
     assert session.environment.to_dict()['packages'] == {'system': [], **configuration['packages']}
@@ -61,121 +33,16 @@ def verify_setup_metadata(client, session, configuration):
         assert 'env' not in environment and 'setup_commands' not in environment
 
 
-def native_setup_script(marker, network_target=None, *, system_packages=False):
+def native_setup_script(marker):
     script = f'''from pathlib import Path
-import os, subprocess, socket
+import os, subprocess
 import packaging
 assert packaging.__version__ == '26.0'
 assert os.environ['SETUP_VALUE'] == {marker!r}
-assert Path('/workspace/setup-sub/order').read_text() == 'second'
-assert Path('/workspace/setup-once').read_text() == 'initialized'
-for cwd in ['/workspace', '/workspace/setup-sub']:
+assert Path('setup-sub/order').read_text() == 'second'
+assert Path('setup-once').read_text() == 'initialized'
+for cwd in ['.', 'setup-sub']:
     assert subprocess.check_output(['semver', '1.2.3'], cwd=cwd).strip() == b'1.2.3'
     subprocess.run(['python3', '-c', 'import packaging; assert packaging.__version__ == "26.0"'], cwd=cwd, check=True)
-'''
-    if system_packages:
-        script += '''assert Path('/workspace').samefile('/environment/workspace')
-for cwd in ['/environment/workspace', '/environment/workspace/setup-sub']:
-    assert subprocess.check_output(['jq', '-r', '.value'], input=b'{"value":42}', cwd=cwd).strip() == b'42'
-assert subprocess.check_output(['jq', '-r', '.value'], input=b'{"value":42}').strip() == b'42'
-subprocess.run(['/workspace/system-library'], check=True)
-for path in ['/usr/bin/system-package-write', '/environment/packages/system/usr/bin/system-package-write']:
-    try:
-        Path(path).write_text('changed')
-    except OSError:
-        pass
-    else:
-        raise AssertionError('installed system root is writable')
-'''
-    if network_target:
-        script += f'''try:
-    connection = socket.create_connection({network_target!r}, timeout=2)
-except OSError:
-    pass
-else:
-    connection.close()
-    raise AssertionError('runtime network policy was not applied after setup')
 '''
     return script
-
-
-def verify_system_package_configuration(client, http):
-    templates = client.beta.agents.environments.templates
-    template = templates.create(packages={'system': ['jq'], 'npm': ['semver@7.7.2']})
-    endpoint = str(client.base_url).rstrip('/') + '/agents/environments/templates/' + template.id
-    headers = {'Authorization': 'Bearer ' + client.api_key, 'OpenAI-Beta': 'agents=v1'}
-    try:
-        expected = {'system': ['jq'], 'npm': ['semver@7.7.2'], 'python': []}
-        assert http.get(endpoint, headers=headers).json()['packages'] == expected
-        assert templates.update(template.id, name='System tools').to_dict()['packages'] == expected
-        assert templates.update(template.id, packages={'system': ['libpq-dev']}).to_dict()['packages'] == {
-            'system': ['libpq-dev'], 'npm': [], 'python': []}
-        empty = {'system': [], 'npm': [], 'python': []}
-        for packages in [{'system': None}, {'system': []}, None]:
-            assert templates.update(template.id, packages=packages).to_dict()['packages'] == empty
-        for packages in [{'system': [None]}, {'system': ['']}, {'system': ['-unsafe-option']}]:
-            response = http.post(endpoint, headers=headers, json={'packages': packages})
-            assert response.status_code == 400
-            assert '-unsafe-option' not in response.text
-            assert templates.retrieve(template.id).to_dict()['packages'] == empty
-    finally:
-        templates.delete(template.id)
-
-
-def verify_setup_failure(client, http, agent, until):
-    """Actual Provider initialization must fail before any native Turn starts."""
-    sessions = client.beta.agents.sessions
-    for command in [{'command': 'echo confidential-setup-failure >&2; exit 7'},
-                    {'command': 'touch /workspace/unexpected', 'cwd': '/missing-setup-cwd'}]:
-        session = sessions.create(agent=agent, environment={
-            'type': 'openai_hosted', 'setup_commands': [command,
-                {'command': 'touch /workspace/unexpected-later-step'}]})
-        try:
-            until(lambda: client.beta.agents.environments.retrieve(session.environment.id).status == 'failed', 180)
-            assert sessions.turns.list(session.id).data == []
-            resource = sessions.retrieve(session.id).to_dict()
-            assert 'confidential-setup-failure' not in json.dumps(resource)
-            assert 'setup_commands' not in resource['environment']
-        finally:
-            sessions.delete(session.id)
-
-
-def verify_saved_agent_setup_identity(client, http):
-    sessions = client.beta.agents.sessions
-    agent = client.beta.agents.create(model='kimi-k3')
-    created = []
-    secret = 'intent-canary-' + secrets.token_hex(12)
-    variants = [
-        {'type': 'openai_hosted'},
-        {'type': 'openai_hosted', 'env': {'VALUE': secret}},
-        {'type': 'openai_hosted', 'env': {'VALUE': secret + '-changed'}},
-        {'type': 'openai_hosted', 'setup_commands': [{'command': 'true'}]},
-        {'type': 'openai_hosted', 'setup_commands': [{'command': 'false'}]},
-        {'type': 'openai_hosted', 'env': {'VALUE': secret}, 'setup_commands': [{'command': 'true'}]},
-    ]
-    try:
-        for index in [0, 1, 3]:
-            key = 'setup-intent-' + secrets.token_hex(12)
-            original = sessions.create(agent_id=agent.id, environment=variants[index],
-                                       extra_headers={'Idempotency-Key': key})
-            created.append((original, key, index))
-        client.beta.agents.delete(agent.id)
-        endpoint = str(client.base_url).rstrip('/') + '/agents/sessions'
-        for original, key, index in created:
-            assert sessions.create(agent_id=agent.id, environment=variants[index],
-                                   extra_headers={'Idempotency-Key': key}).id == original.id
-            for changed, environment in enumerate(variants):
-                if changed == index:
-                    continue
-                response = http.post(endpoint, json={'agent_id': agent.id, 'environment': environment},
-                                     headers={'Authorization': 'Bearer ' + client.api_key,
-                                              'OpenAI-Beta': 'agents=v1', 'Idempotency-Key': key})
-                assert response.status_code == 409, (index, changed, response.status_code)
-                assert secret not in response.text
-    finally:
-        for session, _, _ in created:
-            sessions.delete(session.id)
-        try:
-            client.beta.agents.delete(agent.id)
-        except NotFoundError:
-            pass
diff --git a/services/core/tests/official_environment_skill_references.py b/services/core/tests/official_environment_skill_references.py
index b07afebb2..01a0051cd 100644
--- a/services/core/tests/official_environment_skill_references.py
+++ b/services/core/tests/official_environment_skill_references.py
@@ -6,13 +6,15 @@
 
 from official_environment_skills import inline_skill
 from official_session_artifacts import verify_session_artifacts
+from session_cleanup import delete_source
 
 
-def upload_reference_skill(client):
+def upload_reference_skill(client, cleanup):
     inline, expected = inline_skill()
     with zipfile.ZipFile(io.BytesIO(base64.b64decode(inline["source"]["data"]))) as archive:
         files = [(name, archive.read(name), "application/octet-stream") for name in archive.namelist()]
     skill = client.skills.create(files=files)
+    cleanup.callback(delete_source, client.skills, skill.id)
     assert skill.default_version == "1"
     return skill, expected
 
diff --git a/services/core/tests/official_environment_skills.py b/services/core/tests/official_environment_skills.py
index 48af5137b..a60c9ae61 100644
--- a/services/core/tests/official_environment_skills.py
+++ b/services/core/tests/official_environment_skills.py
@@ -12,22 +12,15 @@ def inline_skill():
 name: proof-skill
 description: Verify the initialized workspace and publish the Skill proof.
 ---
-Run `python3 /environment/initialization/capabilities/skills/proof-skill/scripts/check.py`.
+Run the packaged `scripts/check.py` by its installed absolute path, with the workspace as the working directory.
 Stop on any failed assertion. Report INITIAL_FILES_VERIFIED and SKILL_VERIFIED.
 """
     script = f'''import os, runpy
 from pathlib import Path
 for name in ['ANTHROPIC_API_KEY', 'ANTHROPIC_AUTH_TOKEN', 'OPENAI_API_KEY', 'MINIMAX_API_KEY']:
     assert name not in os.environ, 'native credential reached a Skill helper'
-manifest = Path('/environment/initialization/capabilities/skills/proof-skill/SKILL.md')
-try:
-    manifest.write_text('tampered')
-except OSError:
-    pass
-else:
-    raise AssertionError('Skill content was writable')
-runpy.run_path('/workspace/verify.py')
-Path('/workspace/outputs/skill-proof.txt').write_text({marker!r})
+runpy.run_path('verify.py')
+Path('outputs/skill-proof.txt').write_text({marker!r})
 print('SKILL_VERIFIED')
 '''
     data = io.BytesIO()
diff --git a/services/core/tests/qualify_public_native.py b/services/core/tests/qualify_public_native.py
index ccfe45c26..c5dd67afc 100644
--- a/services/core/tests/qualify_public_native.py
+++ b/services/core/tests/qualify_public_native.py
@@ -19,6 +19,7 @@
 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 official_environment_composition import verify_composition
 from session_cleanup import delete_session
 
 
@@ -78,7 +79,7 @@ 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("--suite", required=True, choices=("none", "functions", "pending-actions", "images", "structured", "composition"))
     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")
@@ -99,6 +100,8 @@ def main():
     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 == "composition":
+        assert environment == {"type": "openai_hosted"}, "Composition supplies its own fresh hosted preparation"
     elif args.suite != "pending-actions":
         assert placement != "none", "This suite verifies native workspace tools and Artifacts"
     if placement == "self_hosted":
@@ -165,7 +168,7 @@ def ready(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}
+                  "images": verify_workspace_images, "structured": verify_hosted_structured, "composition": verify_composition}
         try:
             kwargs = {"ready": ready, "record": record}
             if args.suite != "pending-actions":
diff --git a/services/core/tests/qualify_public_native_test.py b/services/core/tests/qualify_public_native_test.py
index d0b734b94..c9aa601e8 100644
--- a/services/core/tests/qualify_public_native_test.py
+++ b/services/core/tests/qualify_public_native_test.py
@@ -1,6 +1,7 @@
 """Check qualification's secret handling and owned restart boundary without a model."""
 
 import contextlib
+import base64
 import io
 import json
 import os
@@ -8,6 +9,7 @@
 import subprocess
 import tempfile
 import unittest
+import zipfile
 from types import SimpleNamespace
 from unittest.mock import MagicMock, patch
 
@@ -15,6 +17,8 @@
 from openai import BadRequestError, OpenAI
 
 from official_environment_files_native import generate_files
+from official_environment_composition import composition_fixture
+from official_environment_initial_files import assert_initial_bytes_script
 
 import qualify_public_native as qualification
 
@@ -129,6 +133,89 @@ def suite(*args, restart, record, **kwargs):
                 self.run_suite(suite)
         self.assertEqual(json.loads(self.evidence.read_text())["status"], "failed")
 
+    def test_composition_rejects_preselected_environment_before_creating_resources(self):
+        self.settings['environment'] = {'type': 'openai_hosted', 'environment_template_id': 'foreign-template'}
+        (self.root / 'settings.json').write_text(json.dumps(self.settings))
+        self.argv[self.argv.index('none')] = 'composition'
+        with patch('sys.argv', self.argv), patch.object(qualification, 'verify_composition') as suite:
+            with self.assertRaisesRegex(AssertionError, 'fresh hosted'):
+                qualification.main()
+            suite.assert_not_called()
+
+    def test_composition_partial_failure_cleans_sources_and_keeps_provider_selection(self):
+        client, http = MagicMock(), MagicMock()
+        client.base_url = 'https://core.example/v1'
+        client.api_key = 'project-secret'
+        client.files.create.return_value = SimpleNamespace(id='source-file')
+        client.skills.create.side_effect = RuntimeError('upload failed')
+        http.post.return_value = SimpleNamespace(status_code=404, text='not found')
+        with self.assertRaisesRegex(RuntimeError, 'upload failed'), contextlib.ExitStack() as cleanup:
+            composition_fixture(client, SimpleNamespace(api_key='foreign-secret'), http,
+                self.settings['agent'], self.settings['model_provider'], cleanup)
+        client.files.delete.assert_called_once_with('source-file')
+        self.assertEqual(http.post.call_args.kwargs['json']['x_agents_core']['model_provider'], self.settings['model_provider'])
+        self.assertEqual(http.post.call_args.kwargs['headers']['Authorization'], 'Bearer foreign-secret')
+
+    def test_composition_session_failure_cleans_template_and_all_sources(self):
+        client, http = MagicMock(), MagicMock()
+        client.base_url = 'https://core.example/v1'
+        client.api_key = 'project-secret'
+        client.files.create.return_value = SimpleNamespace(id='source-file')
+        client.skills.create.return_value = SimpleNamespace(id='source-skill', name='proof-skill', description='proof', default_version='1')
+        client.beta.agents.environments.templates.create.return_value = SimpleNamespace(id='template')
+        client.beta.agents.sessions.create.side_effect = RuntimeError('session failed')
+        http.post.return_value = SimpleNamespace(status_code=404, text='not found')
+        options = {'environment': {'type': 'openai_hosted'}, 'extra_body': {'x_agents_core': {'model_provider': self.settings['model_provider']}}}
+        with self.assertRaisesRegex(RuntimeError, 'session failed'):
+            qualification.verify_composition(client, SimpleNamespace(api_key='foreign-secret'), http,
+                self.settings['agent'], options, ready=MagicMock(), restart=None, record=MagicMock())
+        client.files.delete.assert_called_once_with('source-file')
+        client.skills.delete.assert_called_once_with('source-skill')
+        client.beta.agents.environments.templates.delete.assert_called_once_with('template')
+        self.assertEqual(client.beta.agents.sessions.create.call_args.kwargs['extra_body'], options['extra_body'])
+        configuration = client.beta.agents.environments.templates.create.call_args.kwargs
+        self.assertEqual(configuration['network'], {'access': 'enabled'})
+        self.assertEqual(set(configuration['packages']), {'npm', 'python'})
+        for plugin in configuration['plugins']:
+            with zipfile.ZipFile(io.BytesIO(base64.b64decode(plugin['source']['data']))) as archive:
+                mcp = json.loads(archive.read('proof/.mcp.json'))
+                for server in mcp['mcpServers'].values():
+                    self.assertNotIn('env_vars', server)
+        # Exercise the actual prepared proof script's byte and freshness checks;
+        # dependency installation and native execution remain live-only checks.
+        workspace = self.root / 'prepared'
+        workspace.mkdir()
+        for item in configuration['files']:
+            body = base64.b64decode(item['data']) if item['type'] == 'inline' else bytes(range(256))
+            (workspace / item['path'].removeprefix('/workspace/')).write_bytes(body)
+        (workspace / 'setup-sub').mkdir()
+        (workspace / 'setup-sub/order').write_text('second')
+        (workspace / 'setup-once').write_text('initialized')
+        (workspace / 'plugin-mcp-setup-count').write_text('1')
+        script = compile((workspace / 'verify.py').read_text(), 'prepared verify.py', 'exec')
+        with contextlib.chdir(workspace), patch.dict(os.environ, configuration['env']), \
+                patch.dict('sys.modules', {'packaging': SimpleNamespace(__version__='26.0')}), \
+                patch.object(subprocess, 'check_output', return_value=b'1.2.3\n'), patch.object(subprocess, 'run'):
+            for run in (1, 2):
+                exec(script, {})
+                self.assertEqual(json.loads((workspace / 'outputs/composition.json').read_text())['run'], run)
+            (workspace / 'initial-source.bin').write_bytes(b'changed')
+            with self.assertRaises(AssertionError):
+                exec(script, {})
+            self.assertEqual((workspace / 'composition-run-count').read_text(), '2')
+
+    def test_initial_byte_assertion_uses_declared_working_directory(self):
+        workspace = self.root / 'custom-workspace'
+        workspace.mkdir()
+        expected = b'\x00\xffbinary'
+        (workspace / 'initial.bin').write_bytes(expected)
+        script = assert_initial_bytes_script({'/workspace/initial.bin': expected})
+        with contextlib.chdir(workspace):
+            exec(compile(script, '<initial byte assertion>', 'exec'), {'Path': Path})
+            (workspace / 'initial.bin').write_bytes(b'changed')
+            with self.assertRaises(AssertionError):
+                exec(compile(script, '<initial byte assertion>', 'exec'), {'Path': Path})
+
 
 if __name__ == "__main__":
     unittest.main()
diff --git a/services/core/tests/session_cleanup.py b/services/core/tests/session_cleanup.py
index 3a76036c8..f9afb240a 100644
--- a/services/core/tests/session_cleanup.py
+++ b/services/core/tests/session_cleanup.py
@@ -6,6 +6,15 @@
 from openai import APIStatusError
 
 
+def delete_source(resource, identifier):
+    """Clean an owned source that the frozen-snapshot check may already delete."""
+    try:
+        resource.delete(identifier)
+    except APIStatusError as exc:
+        if exc.status_code != 404:
+            raise
+
+
 def delete_session(sessions, session_id, timeout=60):
     """Delete an owned Session from a ``finally`` block.
 

From fcbec5ae66391d03c2ae4d13ec016b053524cf45 Mon Sep 17 00:00:00 2001
From: SaladDay <1203511142@qq.com>
Date: Thu, 8 Oct 2026 14:16:27 +0000
Subject: [PATCH 2/2] Wait for the native MCP invocation directory

---
 .../tests/official_environment_composition.py | 15 ++++--
 .../core/tests/qualify_public_native_test.py  | 50 ++++++++++++++++++-
 2 files changed, 61 insertions(+), 4 deletions(-)

diff --git a/services/core/tests/official_environment_composition.py b/services/core/tests/official_environment_composition.py
index 6267594a0..ac445cb6a 100644
--- a/services/core/tests/official_environment_composition.py
+++ b/services/core/tests/official_environment_composition.py
@@ -7,6 +7,8 @@
 import uuid
 import zipfile
 
+from openai import NotFoundError
+
 from official_environment_initial_files import initial_files, assert_initial_bytes_script
 from official_environment_plugin_mcp import plugin_mcp_fixture, verify_plugin_mcp_metadata, verify_plugin_mcp_items
 from official_environment_setup import setup_configuration, native_setup_script, verify_setup_metadata
@@ -161,13 +163,20 @@ def verify_composition(client, foreign, http, agent_options, session_options, re
                                    idempotency_key=uuid.uuid4().hex)
             deadline = time.monotonic() + 180
             observed = None
+            directory_seen = False
             while time.monotonic() < deadline:
                 turns = [turn for turn in sessions.turns.list(session.id) if turn.id not in before]
                 assert len(turns) <= 1 and not any(turn.status in {'completed', 'failed', 'cancelled'} for turn in turns)
-                rows = client.beta.agents.environments.files.list(session.environment.id, path='/workspace/plugin-mcp-hold')
-                sizes = {row.path: row.size_bytes for row in rows}
+                try:
+                    rows = client.beta.agents.environments.files.list(session.environment.id, path='/workspace/plugin-mcp-hold')
+                    sizes = {row.path: row.size_bytes for row in rows}
+                    directory_seen = True
+                except NotFoundError:
+                    if directory_seen:
+                        raise
+                    sizes = {}
                 size = sizes.get(fixture['hold_paths']['ticks'], 0)
-                if observed is not None and size > observed and fixture['hold_paths']['invocation'] in sizes:
+                if turns and observed is not None and size > observed and fixture['hold_paths']['invocation'] in sizes:
                     break
                 if size:
                     observed = size
diff --git a/services/core/tests/qualify_public_native_test.py b/services/core/tests/qualify_public_native_test.py
index c9aa601e8..5612da39c 100644
--- a/services/core/tests/qualify_public_native_test.py
+++ b/services/core/tests/qualify_public_native_test.py
@@ -14,11 +14,12 @@
 from unittest.mock import MagicMock, patch
 
 import httpx2
-from openai import BadRequestError, OpenAI
+from openai import BadRequestError, NotFoundError, OpenAI
 
 from official_environment_files_native import generate_files
 from official_environment_composition import composition_fixture
 from official_environment_initial_files import assert_initial_bytes_script
+import official_environment_composition as composition
 
 import qualify_public_native as qualification
 
@@ -216,6 +217,53 @@ def test_initial_byte_assertion_uses_declared_working_directory(self):
             with self.assertRaises(AssertionError):
                 exec(compile(script, '<initial byte assertion>', 'exec'), {'Path': Path})
 
+    def test_mcp_hold_waits_for_directory_and_turn_without_hiding_errors(self):
+        request = httpx2.Request('GET', 'https://core.example/v1/files')
+        missing = NotFoundError('missing', response=httpx2.Response(404, request=request), body=None)
+        invalid = BadRequestError('invalid', response=httpx2.Response(400, request=request), body=None)
+        output = SimpleNamespace(path='/workspace/outputs/composition.json', size_bytes=2)
+        ticks = '/workspace/plugin-mcp-hold/ticks.jsonl'
+        invocation = '/workspace/plugin-mcp-hold/invocation.json'
+        growing = [[SimpleNamespace(path=ticks, size_bytes=size), SimpleNamespace(path=invocation, size_bytes=1)]
+                   for size in (5, 10, 15, 20)]
+        fixture = {'configuration': {}, 'prompt': 'verify', 'outputs': {output.path: b'{}'},
+                   'composition_proof': {}, 'hold_prompt': 'hold', 'hold_server': 'server',
+                   'hold_paths': {'ticks': ticks, 'invocation': invocation}}
+        for label, polling, error in (
+            ('delayed directory and turn', [missing, *growing, growing[-1]], None),
+            ('directory disappeared', [missing, growing[0], missing], NotFoundError),
+            ('other error', [invalid], BadRequestError),
+        ):
+            with self.subTest(label), contextlib.ExitStack() as patches:
+                client = MagicMock()
+                sessions = client.beta.agents.sessions
+                session = SimpleNamespace(id='session', environment=SimpleNamespace(id='environment'), status='idle', required_actions=[])
+                sessions.create.return_value = sessions.retrieve.return_value = session
+                sessions.turns.list.side_effect = [[], [], [], [], [SimpleNamespace(id='hold-turn', status='in_progress')]]
+                sessions.turns.retrieve.return_value = SimpleNamespace(status='cancelled')
+                stream_events = [{'type': 'agent.session.turn.completed', 'turn': {'id': 'completed-turn'}},
+                                 {'type': 'agent.session.idle'}]
+                sessions.stream.return_value.__enter__.return_value = [
+                    SimpleNamespace(to_dict=lambda event=event: event) for event in stream_events]
+                client.beta.agents.environments.files.list.side_effect = [[output], [output], *polling]
+                patches.enter_context(patch.object(composition, 'composition_fixture', return_value=fixture))
+                for name in ('change_and_delete_sources', 'verify_composition_metadata', 'verify_session_artifacts', 'verify_plugin_mcp_items'):
+                    patches.enter_context(patch.object(composition, name, return_value={}))
+                patches.enter_context(patch.object(composition.time, 'sleep'))
+                options = {'environment': {'type': 'openai_hosted'},
+                           'extra_body': {'x_agents_core': {'model_provider': self.settings['model_provider']}}}
+                if error is not None:
+                    with self.assertRaises(error):
+                        composition.verify_composition(client, MagicMock(), MagicMock(), self.settings['agent'],
+                            options, ready=MagicMock(), restart=None, record=MagicMock())
+                    sessions.turns.retrieve.assert_not_called()
+                else:
+                    checks = composition.verify_composition(client, MagicMock(), MagicMock(), self.settings['agent'],
+                        options, ready=MagicMock(), restart=None, record=MagicMock())
+                    self.assertIn('native_mcp_cancel_retry_stops_descendant_effects', checks)
+                    sessions.turns.retrieve.assert_called_once_with('hold-turn', session_id='session')
+                    self.assertEqual(sessions.events.create.call_count, 3)
+
 
 if __name__ == "__main__":
     unittest.main()