Stage B: decouple large file bodies from the plan JSON

src/foundation/models/file.py

@@ -81,6 +81,19 @@ class FileWriteRequest(StrictModel):
     overwrite: bool = False
 
 
+class FileWriteBriefRequest(StrictModel):
+    """Deferred write: the body is generated from a brief instead of inlined.
+
+    Keeps large file content out of the schema-constrained plan JSON. The
+    orchestrator materializes ``content_brief`` into ``content`` via a separate
+    text-generation call before the write executes.
+    """
+
+    path: str = Field(min_length=1)
+    content_brief: str = Field(min_length=1)
+    overwrite: bool = False
+
+
 class FileEditRequest(StrictModel):
     """Rewrite an existing file with conflict detection."""
 

src/foundation/services/orchestrator.py

@@ -33,6 +33,8 @@
     PolicyEvaluationRecord,
     ProviderMessage,
     ProviderMessageRole,
+    ProviderPrompt,
+    ProviderResponseFormat,
     SessionKind,
     SessionStatus,
     UserRequest,
@@ -624,6 +626,80 @@ def orchestrate(self, request: UserRequest) -> OrchestrationResult:
     # Bounded replan loop
     # ------------------------------------------------------------------
 
+    def _materialize_deferred_writes(
+        self,
+        actions: list[PlannedAction],
+        *,
+        request: UserRequest,
+        observation_messages: list[ProviderMessage],
+        request_id: str,
+        session_id: str | None,
+    ) -> None:
+        """Generate file bodies for writes that deferred content via content_brief.
+
+        On success the brief is replaced with literal ``content`` for the
+        executor. On a provider failure the brief is left in place so the
+        write fails as a normal action (the executor rejects the unknown
+        ``content_brief`` field) rather than killing the turn.
+        """
+        for action in actions:
+            if action.kind is not ActionKind.TOOL_CALL or action.tool_call is None:
+                continue
+            tool_call = action.tool_call
+            if tool_call.capability_id != "foundation.file.write":
+                continue
+            brief = tool_call.arguments.get("content_brief")
+            if not brief or tool_call.arguments.get("content"):
+                continue
+            path = str(tool_call.arguments.get("path", "<file>"))
+            try:
+                body = self._generate_file_body(
+                    path=path,
+                    brief=str(brief),
+                    request=request,
+                    observation_messages=observation_messages,
+                )
+            except ProviderError as exc:
+                logger.warning(
+                    "deferred_write_generation_failed action=%s path=%s error=%s",
+                    action.id,
+                    path,
+                    exc,
+                )
+                continue
+            tool_call.arguments.pop("content_brief", None)
+            tool_call.arguments["content"] = body
+
+    def _generate_file_body(
+        self,
+        *,
+        path: str,
+        brief: str,
+        request: UserRequest,
+        observation_messages: list[ProviderMessage],
+    ) -> str:
+        messages: list[ProviderMessage] = [
+            ProviderMessage(
+                role=ProviderMessageRole.DEVELOPER,
+                content=(
+                    f"You are generating the complete contents of the file `{path}`. "
+                    "Output ONLY the raw file body — no markdown code fences, no "
+                    "commentary, no surrounding quotes."
+                ),
+            )
+        ]
+        messages.extend(observation_messages)
+        messages.append(
+            ProviderMessage(
+                role=ProviderMessageRole.USER,
+                content=f"Original request: {request.message}\n\nFile to write: {brief}",
+            )
+        )
+        response = self._provider.complete(
+            ProviderPrompt(messages=messages, response_format=ProviderResponseFormat.TEXT)
+        )
+        return response.content
+
     def _run_replan_loop(
         self,
         *,
@@ -773,6 +849,18 @@ def _run_replan_loop(
             budget = _MAX_TOTAL_ACTIONS - total_actions_executed
             actions_to_execute = plan.actions[:budget]
 
+            # 4b. Materialize deferred file bodies (content_brief -> content) via a
+            # separate text-generation call, keeping large content out of the
+            # schema-constrained plan JSON. A generation failure leaves the brief
+            # in place so the write degrades to a normal failed action.
+            self._materialize_deferred_writes(
+                actions_to_execute,
+                request=request,
+                observation_messages=observation_messages,
+                request_id=request_id,
+                session_id=session_id,
+            )
+
             # 5. Policy evaluate + execute
             execution_results, decisions, evaluations, last_step_id = (
                 self._execute_iteration_actions(

src/foundation/services/planner.py

@@ -24,6 +24,7 @@
     FileEditRequest,
     FileReadChunkRequest,
     FileReadRequest,
+    FileWriteBriefRequest,
     FileWriteRequest,
 )
 from foundation.models.git import (
@@ -235,6 +236,11 @@ def _base_plan_messages(
                         "capability_id": "foundation.search",
                         "version": "1.0.0 | null",
                         "arguments": "tool-specific JSON object",
+                        "_file_write_note": (
+                            "for foundation.file.write use {path, content} for tiny "
+                            "files, or {path, content_brief} (NOT content) for "
+                            "anything longer — the body is generated separately"
+                        ),
                     },
                 }
             ],
@@ -264,6 +270,12 @@ def _base_plan_messages(
             "Prefer typed file capabilities (foundation.file.read, "
             "foundation.file.write, foundation.file.edit, foundation.file.apply_diff) "
             "for reading and editing files. "
+            "For foundation.file.write, do NOT inline a large file body in the "
+            "`content` argument — long content bloats this JSON plan and can be "
+            "truncated. For anything beyond a few short lines, omit `content` and "
+            "instead provide `content_brief`: a concise description of what the file "
+            "should contain. The body is generated separately. Use literal `content` "
+            "only for very short files. "
             "Prefer typed git capabilities (foundation.git.*) for repository "
             "inspection and staging. "
             "The git.commit capability requires approval and never stages implicitly. "
@@ -457,10 +469,22 @@ def _validated_tool_request(
 
             ShellAction.model_validate(arguments)
             return
+        if endpoint == "builtin.file.write":
+            has_content = bool(arguments.get("content"))
+            has_brief = bool(arguments.get("content_brief"))
+            if has_content and has_brief:
+                raise PlanningError(
+                    "foundation.file.write must provide either content or "
+                    "content_brief, not both."
+                )
+            if has_brief:
+                FileWriteBriefRequest.model_validate(arguments)
+            else:
+                FileWriteRequest.model_validate(arguments)
+            return
         _FILE_VALIDATORS: dict[str, type[BaseModel]] = {
             "builtin.file.read": FileReadRequest,
             "builtin.file.read_chunk": FileReadChunkRequest,
-            "builtin.file.write": FileWriteRequest,
             "builtin.file.edit": FileEditRequest,
             "builtin.file.apply_diff": FileApplyDiffRequest,
         }

tests/test_orchestrator.py

@@ -17,6 +17,7 @@
     PlanningStep,
     ProviderPrompt,
     ProviderResponse,
+    ProviderResponseFormat,
     ProviderResponseMetadata,
     SessionStatus,
     TraceQuery,
@@ -95,6 +96,18 @@ def _provider_response(payload: dict[str, Any]) -> ProviderResponse:
     )
 
 
+def _text_response(body: str) -> ProviderResponse:
+    return ProviderResponse(
+        content=body,
+        structured_output=None,
+        metadata=ProviderResponseMetadata(
+            provider="stub",
+            model="stub-model",
+            latency_seconds=0.01,
+        ),
+    )
+
+
 def _orchestrator(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,
@@ -315,6 +328,99 @@ def complete(self, prompt: ProviderPrompt) -> ProviderResponse:
     assert result.summary is not None
 
 
+def test_orchestrator_materializes_content_brief_via_text_call(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    body = "# Report\n\n" + ("This is a long generated body. " * 50)
+    provider = StubProvider(
+        [
+            _provider_response(
+                {
+                    "assistant_message": "Writing the report.",
+                    "actions": [
+                        {
+                            "id": "write_report",
+                            "kind": "tool_call",
+                            "summary": "Write the report",
+                            "tool_call": {
+                                "capability_id": "foundation.file.write",
+                                "arguments": {
+                                    "path": "report.md",
+                                    "content_brief": "a markdown report about the project",
+                                },
+                            },
+                        }
+                    ],
+                }
+            ),
+            _text_response(body),
+        ]
+    )
+    orchestrator, _, workspace_root = _orchestrator(tmp_path, monkeypatch, provider)
+
+    result = orchestrator.orchestrate(UserRequest(message="write a report"))
+
+    # The body was generated by a separate TEXT call, then written verbatim.
+    assert (workspace_root / "report.md").read_text(encoding="utf-8") == body
+    materialization_call = provider.calls[1]
+    assert materialization_call.response_format is ProviderResponseFormat.TEXT
+    # The plan call itself never carried the large body inline.
+    plan_call_text = "\n".join(m.content for m in provider.calls[0].messages)
+    assert body not in plan_call_text
+    assert any(r.status is ExecutionStatus.EXECUTED for r in result.execution_results)
+
+
+def test_orchestrator_deferred_write_failure_degrades_to_failed_action(
+    tmp_path: Path,
+    monkeypatch: pytest.MonkeyPatch,
+) -> None:
+    from foundation.services.provider import ProviderError, ProviderErrorCode
+
+    plan = _provider_response(
+        {
+            "assistant_message": "Writing the report.",
+            "actions": [
+                {
+                    "id": "write_report",
+                    "kind": "tool_call",
+                    "summary": "Write the report",
+                    "tool_call": {
+                        "capability_id": "foundation.file.write",
+                        "arguments": {
+                            "path": "report.md",
+                            "content_brief": "a markdown report",
+                        },
+                    },
+                }
+            ],
+        }
+    )
+
+    class _PlanThenFailBody:
+        def __init__(self) -> None:
+            self.calls: list[ProviderPrompt] = []
+
+        def complete(self, prompt: ProviderPrompt) -> ProviderResponse:
+            self.calls.append(prompt)
+            if len(self.calls) == 1:
+                return plan
+            if len(self.calls) == 2:
+                raise ProviderError(
+                    "body truncated", code=ProviderErrorCode.TRUNCATED
+                )
+            return _provider_response({"assistant_message": "Done.", "actions": []})
+
+    provider = _PlanThenFailBody()
+    orchestrator, _, workspace_root = _orchestrator(tmp_path, monkeypatch, provider)
+
+    result = orchestrator.orchestrate(UserRequest(message="write a report"))
+
+    # Body generation failed -> the write degrades to a failed action, no file written.
+    assert not (workspace_root / "report.md").exists()
+    assert any(r.status is ExecutionStatus.FAILED for r in result.execution_results)
+
+
 def test_orchestrator_retries_shell_cat_plan_without_executing_it(
     tmp_path: Path,
     monkeypatch: pytest.MonkeyPatch,