diff --git a/apps/backend/agents/coder.py b/apps/backend/agents/coder.py index 16e7f5f75..1d9a1c82a 100644 --- a/apps/backend/agents/coder.py +++ b/apps/backend/agents/coder.py @@ -80,6 +80,7 @@ RuntimeCapabilityError, create_runtime_session, get_runtime_mode, + is_mutating_subagent_task, requirements_for_runtime_mode, resolve_direct_api_autonomous_gate, resolve_runtime_mode_with_fallback, @@ -1353,6 +1354,21 @@ def subagent_session_factory( child_session = child_provider.create_session( child_session_config ) + if is_mutating_subagent_task(task): + # A transactional_write child runs a write-confined + # generic_edit session that stages its mutations and + # exports them as a changeset for the parent to merge + # — it never commits to the shared workspace itself. + return create_runtime_session( + provider_name=child_provider.name, + agent_session=child_session, + claude_session_runner=run_agent_session, + runtime_mode="generic_edit", + project_dir=project_dir, + agent_type=child_agent_type, + write_scope_guard=tuple(task.write_scope), + changeset_export=True, + ) return create_runtime_session( provider_name=child_provider.name, agent_session=child_session, diff --git a/apps/backend/agents/runtime/adapters/__init__.py b/apps/backend/agents/runtime/adapters/__init__.py index e48c87199..9ab8d21ac 100644 --- a/apps/backend/agents/runtime/adapters/__init__.py +++ b/apps/backend/agents/runtime/adapters/__init__.py @@ -32,13 +32,17 @@ def create_runtime_session( subagent_session_factory: Callable[..., Awaitable[Any] | Any] | None = None, allow_direct_api_autonomous: bool = False, write_scope_guard: tuple[str, ...] | list[str] | None = None, + changeset_export: bool = False, ) -> Any: """Create a runtime adapter for a provider session. ``write_scope_guard`` confines a Generic Edit session (or its promoted direct-API variant) to a declared write scope; the runtime then blocks - mutations outside that scope. Used to build mutating subagent child - sessions whose write contract is enforced, not advisory. + mutations outside that scope. ``changeset_export`` additionally makes the + session stage its mutations and export them as a changeset at finish + instead of committing them to the shared workspace. Together they build + mutating subagent child sessions whose write contract is enforced, not + advisory. """ provider_name = provider_name.lower() @@ -63,6 +67,7 @@ def create_runtime_session( agent_type=agent_type, subagent_session_factory=subagent_session_factory, write_scope_guard=write_scope_guard, + changeset_export=changeset_export, ) if ( @@ -81,6 +86,7 @@ def create_runtime_session( agent_type=agent_type, subagent_session_factory=subagent_session_factory, write_scope_guard=write_scope_guard, + changeset_export=changeset_export, ) if runtime_mode == "analysis_only": diff --git a/apps/backend/agents/runtime/adapters/generic_edit.py b/apps/backend/agents/runtime/adapters/generic_edit.py index 34d88d9e2..410488db5 100644 --- a/apps/backend/agents/runtime/adapters/generic_edit.py +++ b/apps/backend/agents/runtime/adapters/generic_edit.py @@ -352,6 +352,12 @@ class JsonActionExecutionResult: COMMIT_BATCH_TOOL = "commit_batch" ABORT_BATCH_TOOL = "abort_batch" BATCH_CONTROL_TOOLS = frozenset({BEGIN_BATCH_TOOL, COMMIT_BATCH_TOOL, ABORT_BATCH_TOOL}) +# Phase 1.2: implicit staging batch for changeset-exporting (mutating child) +# sessions. The runtime owns this batch: it auto-opens on the first staged +# mutation and is finalized into an exported changeset at finish instead of +# ever being committed to the shared workspace. +CHANGESET_BATCH_ID = "__changeset__" +GENERIC_EDIT_CHANGESET_SCHEMA_VERSION = 1 MUTATING_LOCAL_ACTIONS = frozenset( { "write_file", @@ -598,6 +604,7 @@ def __init__( sandbox_policy: SandboxPolicy | None = None, sandbox_backend: SandboxBackendInfo | None = None, write_scope_guard: tuple[str, ...] | list[str] | None = None, + changeset_export: bool = False, ): self.provider_name = provider_name self.agent_session = agent_session @@ -606,6 +613,12 @@ def __init__( # actions may only target paths inside this scope and opaque shell # commands are blocked entirely (they cannot be scope-checked). self.write_scope_guard = normalize_write_scope_entries(write_scope_guard) + # Phase 1.2: changeset-exporting session. Mutations are auto-staged + # into the runtime-owned CHANGESET_BATCH_ID batch and finish exports + # the staged pre/postimages instead of committing them to the shared + # workspace; transaction batching is runtime-owned in this mode. + self.changeset_export = bool(changeset_export) + self._exported_changeset: dict[str, Any] | None = None self._subagent_session_factory = subagent_session_factory self._max_subagent_concurrency = max(1, max_subagent_concurrency) self._max_subagent_task_seconds = max_subagent_task_seconds @@ -1163,7 +1176,11 @@ def _finish_json_action_loop( results=action_results, ) finish_trace = [*trace, iteration_entry] - if has_open_transaction_batch(finish_trace): + # In a changeset-exporting session the only possible open batch is the + # runtime-owned staging batch (model batch control is rejected), and + # finish finalizes it into the exported changeset — so the open-batch + # veto only applies to model-managed batches. + if not self.changeset_export and has_open_transaction_batch(finish_trace): return self._open_batch_finish_result( trace=finish_trace, spec_dir=spec_dir, @@ -1177,6 +1194,7 @@ def _finish_json_action_loop( observation_path=observation_path, subtask_id=subtask_id, ) + changeset = self._finalize_changeset_export() trace.append(iteration_entry) artifacts = save_generic_edit_artifacts( spec_dir=spec_dir, @@ -1203,6 +1221,7 @@ def _finish_json_action_loop( return AgentRunResult( status="continue", response_text="\n".join(response_lines), + changeset=changeset, ) async def _run_native_tool_iteration( @@ -1649,7 +1668,9 @@ def _finish_native_tool_loop( results=action_results, ) finish_trace = [*trace, iteration_entry] - if has_open_transaction_batch(finish_trace): + # See the JSON finish path: the open-batch veto only applies to + # model-managed batches, never to the runtime-owned staging batch. + if not self.changeset_export and has_open_transaction_batch(finish_trace): return self._open_batch_finish_result( trace=finish_trace, spec_dir=spec_dir, @@ -1663,6 +1684,7 @@ def _finish_native_tool_loop( observation_path=observation_path, subtask_id=subtask_id, ) + changeset = self._finalize_changeset_export() trace.append(iteration_entry) artifacts = save_generic_edit_artifacts( spec_dir=spec_dir, @@ -1689,6 +1711,7 @@ def _finish_native_tool_loop( return AgentRunResult( status="continue", response_text="\n".join(response_lines), + changeset=changeset, ) def _open_batch_finish_result( @@ -1812,10 +1835,16 @@ async def _execute_action( if scope_error is not None: return scope_error + changeset_error = self._changeset_batch_control_result(action) + if changeset_error is not None: + return changeset_error + isolation_error = self._opaque_batch_mutation_result(action) if isolation_error is not None: return isolation_error + self._auto_open_changeset_batch(action) + tool = action_tool(action) try: staged_workspace = self._materialize_staged_batch_workspace(action) @@ -1986,6 +2015,70 @@ def _write_scope_violation_result( }, ) + def _changeset_batch_control_result( + self, + action: dict[str, Any], + ) -> ToolActionResult | None: + """Reject model-issued batch control in a changeset-exporting session. + + Changeset sessions stage mutations automatically into the + runtime-owned changeset batch; committing to the shared workspace is + the parent's job, so begin/commit/abort batch are not available to + the child model. + """ + if not self.changeset_export: + return None + tool = action_tool(action) + if tool not in BATCH_CONTROL_TOOLS: + return None + return ToolActionResult( + tool=tool, + ok=False, + message=( + f"{tool} is unavailable in a changeset-exporting session: " + "mutations are staged automatically and exported at finish " + "for the parent to merge transactionally. Edit files inside " + "your write scope and call finish." + ), + data={ + "changeset_batch_control_rejected": True, + "changeset_batch_id": CHANGESET_BATCH_ID, + }, + ) + + def _auto_open_changeset_batch(self, action: dict[str, Any]) -> None: + """Open the runtime-owned staging batch before the first mutation.""" + if ( + self.changeset_export + and self._active_batch_id is None + and action_tool(action) in SNAPSHOT_MUTATING_ACTIONS + ): + self._active_batch_id = CHANGESET_BATCH_ID + + def _finalize_changeset_export(self) -> dict[str, Any] | None: + """Export staged changeset mutations at finish without committing them. + + Returns the changeset payload (empty entries when the child mutated + nothing) and closes the runtime-owned staging batch so the session + ends clean. The shared workspace is left at its baseline state. + """ + if not self.changeset_export: + return None + changeset = build_generic_edit_changeset( + batch_id=CHANGESET_BATCH_ID, + mutation_snapshots=self._mutation_snapshots, + write_scope=self.write_scope_guard, + ) + for snapshot in self._mutation_snapshots: + if generic_edit_snapshot_is_active_staged( + snapshot, batch_id=CHANGESET_BATCH_ID + ): + snapshot["staged_status"] = "exported" + if self._active_batch_id == CHANGESET_BATCH_ID: + self._active_batch_id = None + self._exported_changeset = changeset + return changeset + def _opaque_batch_mutation_result( self, action: dict[str, Any], @@ -3807,6 +3900,72 @@ def active_generic_edit_staged_postimages( return postimages +def build_generic_edit_changeset( + *, + batch_id: str, + mutation_snapshots: list[dict[str, Any]], + write_scope: tuple[str, ...] | None = None, +) -> dict[str, Any]: + """Build an exportable changeset from active staged mutation snapshots. + + One entry per path: the FIRST captured preimage (the shared-workspace + baseline the parent must still observe at merge time) and the LAST staged + postimage (the content the parent applies). An entry without full + postimage content (over the preimage size cap, or non-UTF-8) is marked + non-exportable, and the changeset as a whole is only ``exportable`` when + every entry is — the parent refuses to apply partial content. + """ + first_preimage_by_path: dict[str, dict[str, Any]] = {} + last_postimage_by_path: dict[str, dict[str, Any]] = {} + snapshot_ids: list[str] = [] + for snapshot in mutation_snapshots: + if not generic_edit_snapshot_is_active_staged(snapshot, batch_id=batch_id): + continue + snapshot_id = str(snapshot.get("id") or "") + if snapshot_id: + snapshot_ids.append(snapshot_id) + for preimage in snapshot.get("preimages") or []: + if not isinstance(preimage, dict): + continue + path = str(preimage.get("path") or "") + if path and path not in first_preimage_by_path: + first_preimage_by_path[path] = preimage + for postimage in snapshot.get("postimages") or []: + if not isinstance(postimage, dict): + continue + path = str(postimage.get("path") or "") + if path: + last_postimage_by_path[path] = postimage + + entries: list[dict[str, Any]] = [] + for path in sorted(last_postimage_by_path): + postimage = last_postimage_by_path[path] + # The parent applies entries with the same machinery that + # materializes staged states, so exportability == materializability. + entry_exportable = generic_edit_file_state_is_materializable(postimage) + entries.append( + { + "path": path, + "exportable": entry_exportable, + "preimage": first_preimage_by_path.get(path), + "postimage": postimage, + } + ) + + return { + "schema_version": GENERIC_EDIT_CHANGESET_SCHEMA_VERSION, + "batch_id": batch_id, + "write_scope": list(write_scope or ()), + "entry_count": len(entries), + "exportable": all(entry["exportable"] for entry in entries), + "non_exportable_paths": [ + entry["path"] for entry in entries if not entry["exportable"] + ], + "snapshot_ids": list(dict.fromkeys(snapshot_ids)), + "entries": entries, + } + + def materialize_generic_edit_staged_workspace( *, project_dir: Path, diff --git a/apps/backend/agents/runtime/result.py b/apps/backend/agents/runtime/result.py index 3e2d04c02..ba515b6d4 100644 --- a/apps/backend/agents/runtime/result.py +++ b/apps/backend/agents/runtime/result.py @@ -13,3 +13,8 @@ class AgentRunResult: usage_metadata: dict[str, Any] | None = None decision_tracker: Any = None artifacts: dict[str, str] | None = None + # Phase 1.2 mutating subagents: a changeset-exporting session (a + # write-confined child) finishes WITHOUT committing its staged mutations + # to the shared workspace; the staged pre/postimages are returned here so + # the parent can merge them transactionally. + changeset: dict[str, Any] | None = None diff --git a/apps/backend/agents/runtime/subagents.py b/apps/backend/agents/runtime/subagents.py index da8a1db3d..e63819428 100644 --- a/apps/backend/agents/runtime/subagents.py +++ b/apps/backend/agents/runtime/subagents.py @@ -118,6 +118,10 @@ class RuntimeSubagentResult: merge_policy: str = DEFAULT_SUBAGENT_MERGE_POLICY write_scope: tuple[str, ...] = () artifact_path: str | None = None + # Staged mutations exported by a changeset session (transactional_write + # children). None for read-only children and for mutating children that + # did not reach a clean finish. + changeset: dict[str, Any] | None = None def to_dict(self) -> dict[str, Any]: """Serialize result for artifact output.""" @@ -372,7 +376,14 @@ async def _run_child_session( verbose: bool, phase: Any, ) -> AgentRunResult: - """Run one child runtime session with the configured timeout.""" + """Run one child runtime session with the configured timeout. + + Children get a per-child artifact namespace under the parent spec + dir so a runtime child (for example a confined generic_edit session) + never clobbers the parent's own runtime artifacts — or a sibling's — + with same-named trace/checkpoint files. + """ + child_spec_dir = subagent_child_spec_dir(self.spec_dir, task) return await asyncio.wait_for( run_runtime_session( runtime_session, @@ -381,10 +392,10 @@ async def _run_child_session( attempt=attempt, child_context_id=child_context_id, ), - self.spec_dir, + child_spec_dir, verbose=verbose, phase=phase, - requirements=task.requirements, + requirements=child_execution_requirements(task), subtask_id=task.subtask_id or task.id, ), timeout=self.max_task_seconds, @@ -424,6 +435,7 @@ def runtime_result_to_subagent_result( usage_metadata=result.usage_metadata, artifacts=result.artifacts, max_attempts=bounded_subagent_attempts(task.max_attempts), + changeset=getattr(result, "changeset", None), ) @@ -559,9 +571,31 @@ def mutating_child_confinement_error( f"Mutating subagent task '{task.id}' child session allows paths " f"outside the declared write scope: {', '.join(extra_paths)}." ) + if not getattr(runtime_session, "changeset_export", False): + return ( + f"Mutating subagent task '{task.id}' requires a changeset-exporting " + "child session (changeset_export=True) so its staged mutations " + "reach the parent as a changeset instead of landing on the shared " + "workspace; the session factory returned a directly-writing " + "session." + ) return None +def child_execution_requirements(task: RuntimeSubagentTask) -> RuntimeRequirements: + """Return the capability requirements for RUNNING one child session. + + ``task.requirements`` is the parent-side gate (it includes the + policy-granted ``subagents`` capability that keeps mutating children + behind the autonomy policy). The child session itself is a Generic Edit + runtime for mutating tasks — validate it against what it physically + does, not against the parent gate. + """ + if is_mutating_subagent_task(task): + return RuntimeRequirements.generic_edit() + return task.requirements + + def resolve_runtime_subagent_support( *, provider_name: str, @@ -688,6 +722,11 @@ def subagent_child_context_id( return f"child-{safe_artifact_id(task.id)}-attempt-{attempt}" +def subagent_child_spec_dir(spec_dir: Path, task: RuntimeSubagentTask) -> Path: + """Return the per-child artifact namespace for one delegated task.""" + return spec_dir / "artifacts" / "subagents" / safe_artifact_id(task.id) + + def summarize_subagent_status(results: list[RuntimeSubagentResult]) -> str: """Return aggregate status for a subagent run.""" if any(result.status == "error" for result in results): @@ -799,6 +838,21 @@ def build_subagent_merge_plan( scope_owner[scope] = result.id conflict_result_ids = list(dict.fromkeys(conflict_result_ids)) + changeset_result_ids = [ + result.id + for result in results + if result.id in mutating_result_ids and result.changeset is not None + ] + # A mutating child that finished WITHOUT exporting a changeset has + # nothing the parent can apply — the merge executor treats it as failed. + # (A real generic_edit child reports "continue" on a clean finish.) + missing_changeset_result_ids = [ + result.id + for result in results + if result.id in mutating_result_ids + and result.changeset is None + and result.status in {"complete", "continue"} + ] return { "strategy": ( "read_only" if not mutating_result_ids else "transactional_parent_merge" @@ -806,6 +860,8 @@ def build_subagent_merge_plan( "requires_parent_merge": bool(mutating_result_ids), "read_only_result_ids": read_only_result_ids, "mutating_result_ids": mutating_result_ids, + "changeset_result_ids": changeset_result_ids, + "missing_changeset_result_ids": missing_changeset_result_ids, "write_scopes": write_scopes, "conflict_result_ids": conflict_result_ids, "has_conflicts": bool(conflict_result_ids), diff --git a/tests/test_agent_runtime.py b/tests/test_agent_runtime.py index 37de58e23..1c6b38f68 100644 --- a/tests/test_agent_runtime.py +++ b/tests/test_agent_runtime.py @@ -1589,6 +1589,7 @@ async def run( response_text=self.response, usage_metadata={"input_tokens": 1, "output_tokens": 2}, artifacts=None, + changeset=getattr(self, "changeset", None), ) async def cancel(self): @@ -1761,6 +1762,8 @@ def session_factory(task: RuntimeSubagentTask): "requires_parent_merge": False, "read_only_result_ids": ["explore-api", "explore-ui"], "mutating_result_ids": [], + "changeset_result_ids": [], + "missing_changeset_result_ids": [], "write_scopes": {}, "conflict_result_ids": [], "has_conflicts": False, @@ -3699,14 +3702,17 @@ async def test_direct_api_autonomous_runs_mutating_subagents_under_bold( def session_factory(task: RuntimeSubagentTask): session = FakeSubagentRuntimeSession(f"changes from {task.id}") - session.capabilities = RuntimeCapabilities( - text_completion=True, - structured_output=True, - filesystem_read=True, - filesystem_edit=True, - subagents=True, - ) + # A mutating child runs as a confined changeset-exporting generic_edit + # session; mirror that contract on the fake. + session.capabilities = RuntimeCapabilities.generic_edit() session.write_scope_guard = tuple(task.write_scope) + session.changeset_export = True + session.changeset = { + "schema_version": 1, + "entry_count": 1, + "exportable": True, + "entries": [{"path": "src/feature_a/a.txt", "exportable": True}], + } created_sessions.append(session) return session @@ -3766,7 +3772,10 @@ def session_factory(task: RuntimeSubagentTask): assert subagent_artifact["merge_plan"]["strategy"] == "transactional_parent_merge" assert subagent_artifact["merge_plan"]["requires_parent_merge"] is True assert subagent_artifact["merge_plan"]["mutating_result_ids"] == ["edit-a"] + assert subagent_artifact["merge_plan"]["changeset_result_ids"] == ["edit-a"] + assert subagent_artifact["merge_plan"]["missing_changeset_result_ids"] == [] assert subagent_artifact["results"][0]["write_scope"] == ["src/feature_a"] + assert subagent_artifact["results"][0]["changeset"]["entry_count"] == 1 @pytest.mark.asyncio @@ -3874,6 +3883,258 @@ async def test_write_scope_guard_blocks_out_of_scope_mutations(tmp_path: Path): assert "batch_boundary_error" not in batched_command["data"] +@pytest.mark.asyncio +async def test_changeset_session_exports_without_touching_workspace(tmp_path: Path): + """A changeset session stages mutations and exports them at finish.""" + (tmp_path / "src").mkdir() + (tmp_path / "src" / "existing.txt").write_text("before\n", encoding="utf-8") + session = FakeGenericEditSession( + [ + { + "thought": "edit inside scope", + "actions": [ + { + "tool": "write_file", + "path": "src/existing.txt", + "content": "after\n", + }, + { + "tool": "write_file", + "path": "src/created.txt", + "content": "new file\n", + }, + ], + }, + { + "thought": "done", + "actions": [ + { + "tool": "finish", + "summary": "changes staged for export", + "tests": [], + "risks": [], + } + ], + }, + ] + ) + runtime_session = create_runtime_session( + provider_name="openai", + agent_session=session, + runtime_mode="generic_edit", + project_dir=tmp_path, + write_scope_guard=["src"], + changeset_export=True, + ) + assert runtime_session.changeset_export is True + + result = await run_runtime_session( + runtime_session, + "stage edits for the parent", + tmp_path, + requirements=RuntimeRequirements.generic_edit(), + subtask_id="1.1", + ) + + # The shared workspace is untouched: staged content never committed. + assert result.status == "continue" + assert (tmp_path / "src" / "existing.txt").read_text( + encoding="utf-8" + ) == "before\n" + assert not (tmp_path / "src" / "created.txt").exists() + + changeset = result.changeset + assert changeset is not None + assert changeset["schema_version"] == 1 + assert changeset["exportable"] is True + assert changeset["entry_count"] == 2 + assert changeset["write_scope"] == ["src"] + entries = {entry["path"]: entry for entry in changeset["entries"]} + assert entries["src/existing.txt"]["preimage"]["content"] == "before\n" + assert entries["src/existing.txt"]["postimage"]["content"] == "after\n" + assert entries["src/created.txt"]["preimage"]["exists"] is False + assert entries["src/created.txt"]["postimage"]["content"] == "new file\n" + + +@pytest.mark.asyncio +async def test_changeset_session_rejects_model_batch_control(tmp_path: Path): + """Batch control is runtime-owned in a changeset-exporting session.""" + session = FakeGenericEditSession( + [ + { + "thought": "try to manage my own batch", + "actions": [{"tool": "begin_batch", "batch_id": "mine"}], + }, + { + "thought": "stage and finish instead", + "actions": [ + { + "tool": "write_file", + "path": "src/a.txt", + "content": "staged\n", + } + ], + }, + { + "thought": "done", + "actions": [ + { + "tool": "finish", + "summary": "staged despite batch refusal", + "tests": [], + "risks": [], + } + ], + }, + ] + ) + runtime_session = create_runtime_session( + provider_name="openai", + agent_session=session, + runtime_mode="generic_edit", + project_dir=tmp_path, + write_scope_guard=["src"], + changeset_export=True, + ) + + result = await run_runtime_session( + runtime_session, + "exercise batch control rejection", + tmp_path, + requirements=RuntimeRequirements.generic_edit(), + subtask_id="1.1", + ) + + observation_lines = [ + json.loads(line) + for line in (tmp_path / "artifacts" / "generic_edit_observations.jsonl") + .read_text(encoding="utf-8") + .splitlines() + ] + batch_control = observation_lines[0]["result"] + assert batch_control["ok"] is False + assert batch_control["data"]["changeset_batch_control_rejected"] is True + # The refusal does not poison the session: staging + export still work. + assert result.changeset is not None + assert result.changeset["entry_count"] == 1 + assert not (tmp_path / "src" / "a.txt").exists() + + +@pytest.mark.asyncio +async def test_orchestrator_runs_changeset_children_end_to_end(tmp_path: Path): + """A real confined generic_edit child exports a changeset to the parent.""" + + def session_factory(task: RuntimeSubagentTask): + child_agent_session = FakeGenericEditSession( + [ + { + "thought": "edit my scope", + "actions": [ + { + "tool": "write_file", + "path": "src/feature_a/a.txt", + "content": "child change\n", + } + ], + }, + { + "thought": "done", + "actions": [ + { + "tool": "finish", + "summary": "feature a staged", + "tests": [], + "risks": [], + } + ], + }, + ] + ) + return create_runtime_session( + provider_name="openai", + agent_session=child_agent_session, + runtime_mode="generic_edit", + project_dir=tmp_path, + write_scope_guard=tuple(task.write_scope), + changeset_export=True, + ) + + orchestrator = RuntimeSubagentOrchestrator( + session_factory=session_factory, + spec_dir=tmp_path, + max_concurrency=2, + ) + run = await orchestrator.run( + [ + RuntimeSubagentTask( + id="edit-a", + prompt="edit feature a", + merge_policy=TRANSACTIONAL_WRITE_SUBAGENT_MERGE_POLICY, + write_scope=("src/feature_a",), + ) + ] + ) + + assert run.status == "continue" + result = run.results[0] + assert result.status == "continue" + assert result.changeset is not None + assert result.changeset["exportable"] is True + assert result.changeset["entries"][0]["path"] == "src/feature_a/a.txt" + # The shared workspace stays untouched until the parent merges. + assert not (tmp_path / "src" / "feature_a" / "a.txt").exists() + # Merge plan records the exported changeset for the parent executor. + merge_plan = run.merge_plan or {} + assert merge_plan["changeset_result_ids"] == ["edit-a"] + assert merge_plan["missing_changeset_result_ids"] == [] + # The per-child parent artifact embeds the changeset payload. + child_artifact = json.loads( + Path(str(result.artifact_path)).read_text(encoding="utf-8") + ) + assert ( + child_artifact["result"]["changeset"]["entries"][0]["postimage"]["content"] + == "child change\n" + ) + # The child runtime writes its own artifacts into a per-child namespace, + # never clobbering the parent's generic_edit artifacts. + child_runtime_artifacts = ( + tmp_path / "artifacts" / "subagents" / "edit-a" / "artifacts" + ) + assert (child_runtime_artifacts / "generic_edit_trace.json").exists() + assert not (tmp_path / "artifacts" / "generic_edit_trace.json").exists() + + +@pytest.mark.asyncio +async def test_mutating_child_requires_changeset_export(tmp_path: Path): + """A confined but directly-writing child session is refused.""" + + def session_factory(task: RuntimeSubagentTask): + session = FakeSubagentRuntimeSession(f"changes from {task.id}") + session.capabilities = RuntimeCapabilities.generic_edit() + session.write_scope_guard = tuple(task.write_scope) + # changeset_export deliberately absent. + return session + + orchestrator = RuntimeSubagentOrchestrator( + session_factory=session_factory, + spec_dir=tmp_path, + max_concurrency=1, + ) + run = await orchestrator.run( + [ + RuntimeSubagentTask( + id="edit-a", + prompt="edit feature a", + merge_policy=TRANSACTIONAL_WRITE_SUBAGENT_MERGE_POLICY, + write_scope=("src/feature_a",), + ) + ] + ) + + assert run.results[0].status == "error" + assert "changeset-exporting" in (run.results[0].error or "") + + @pytest.mark.asyncio async def test_generic_edit_runtime_bridges_auto_claude_mcp_tools( tmp_path: Path,