require a2a pack write prefix support

agent.py

@@ -81,6 +81,7 @@ class AgentBuilder(A2AAgent[BuilderConfig, NoAuth]):
         grant_mode="read_write_overlay",
         grant_allow_patterns=("agents/{name}/**",),
         grant_outputs_prefix="agents/{name}/",
+        grant_write_prefixes=("agents/{name}/",),
         grant_ttl_seconds=1800,
         grant_run_timeout_seconds=1740,
         grant_approval_timeout_seconds=180,
@@ -277,4 +278,4 @@ def _find_url(value: Any) -> str | None:
         m = _URL_RE.search(value)
         return m.group(0) if m else None
     return None
-# rebuild against a2a-pack 0.1.17 for nullable LLM extra_body handling
+# rebuild against a2a-pack 0.1.29 for negotiated write prefixes

agent_builder/builder.py

@@ -218,7 +218,8 @@ Your tools:
                                     for ``@skill`` semantics,
                                     ``context.py`` for ``RunContext``
                                     (workspace, sandbox, emit_progress,
-                                    ask, collect, request_scope),
+                                    ask, collect, request_scope,
+                                    ensure_read/ensure_write),
                                     ``runtime.py`` for ``AgentRuntime``
                                     fields (apt_packages, pricing,
                                     egress, etc.).
@@ -350,6 +351,7 @@ def _build_project_backend(ctx: BuilderContext, *, settings: Settings) -> Any |
             allow_patterns=(f"{prefix}/**", "outputs/**"),
             deny_patterns=(),
             outputs_prefix=None,
+            write_prefixes=(f"{prefix}/", "outputs/"),
         )
     )
     return _with_builder_skills(WorkspaceBackend(workspace))

agent_builder/skills/deepagent-agent-design/SKILL.md

@@ -17,7 +17,8 @@ Use this shape for non-trivial generated agents:
    DeepAgent with `create_deep_agent`.
 3. Pass `backend=ctx.workspace_backend()` so DeepAgents file tools write to the
    caller's durable workspace instead of LangGraph state.
-4. Pass `skills=[RUNTIME_SKILLS_ROOT]` when the generated project includes
+4. Seed project skills into the current grant's write prefix and pass
+   `skills=skill_sources or None` when the generated project includes
    `skills/<skill-name>/SKILL.md`.
 5. For custom subagents, include a `skills` field on each subagent definition.
    The general-purpose subagent inherits main skills, but custom subagents do
@@ -47,32 +48,46 @@ before `create_deep_agent`:
 
 ```python
 from pathlib import Path
+from typing import Any
 
-RUNTIME_SKILLS_ROOT = "/outputs/{{ agent_name }}/.deepagents/skills/"
+RUNTIME_SKILLS_DIR = "{{ agent_name }}/.deepagents/skills/"
 
 
-def _seed_runtime_skills(backend: Any) -> None:
+def _runtime_skills_root(ctx: Any) -> str:
+    workspace = getattr(ctx, "_workspace", None)
+    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
+    if not prefixes:
+        outputs_prefix = getattr(workspace, "outputs_prefix", None)
+        prefixes = (outputs_prefix or "outputs/",)
+    prefix = str(prefixes[0]).strip("/")
+    return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
+
+
+def _seed_runtime_skills(backend: Any, ctx: Any) -> list[str]:
     root = Path(__file__).parent / "skills"
     if not root.exists():
-        return
+        return []
+    runtime_skills_root = _runtime_skills_root(ctx)
     uploads: list[tuple[str, bytes]] = []
     for path in root.rglob("*"):
         if path.is_file():
             rel = path.relative_to(root).as_posix()
-            uploads.append((RUNTIME_SKILLS_ROOT + rel, path.read_bytes()))
+            uploads.append((runtime_skills_root + rel, path.read_bytes()))
     if uploads:
         backend.upload_files(uploads)
+        return [runtime_skills_root]
+    return []
 ```
 
 Then:
 
 ```python
 backend = ctx.workspace_backend()
-_seed_runtime_skills(backend)
+skill_sources = _seed_runtime_skills(backend, ctx)
 return create_deep_agent(
     model=model,
     backend=backend,
-    skills=[RUNTIME_SKILLS_ROOT],
+    skills=skill_sources or None,
     tools=[small_deterministic_tool],
     subagents=[...],
     system_prompt=SYSTEM_PROMPT,

agent_builder/skills/deepagents-implementation-patterns/SKILL.md

@@ -21,12 +21,12 @@ from langchain_core.tools import tool
 from langchain_openai import ChatOpenAI
 from deepagents import create_deep_agent
 
-RUNTIME_SKILLS_ROOT = "/outputs/research-agent/.deepagents/skills/"
+RUNTIME_SKILLS_DIR = "research-agent/.deepagents/skills/"
 
 SYSTEM_PROMPT = """\
 You are a careful research agent. Use project skills for workflow rules and
 use workspace-backed tools for durable files. Prefer /workspace/outputs/ for
-stable output paths.
+stable output paths only when that is one of the grant write prefixes.
 """
 
 
@@ -50,7 +50,7 @@ def build_graph(ctx: Any) -> Any:
         return json.dumps({"ok": not missing, "missing_indexes": missing})
 
     backend = ctx.workspace_backend()
-    skill_sources = seed_runtime_skills(backend)
+    skill_sources = seed_runtime_skills(backend, ctx)
     return create_deep_agent(
         model=model,
         backend=backend,
@@ -71,18 +71,29 @@ def build_graph(ctx: Any) -> Any:
     )
 
 
-def seed_runtime_skills(backend: Any) -> list[str]:
+def runtime_skills_root(ctx: Any) -> str:
+    workspace = getattr(ctx, "_workspace", None)
+    prefixes = tuple(getattr(workspace, "write_prefixes", ()) or ())
+    if not prefixes:
+        outputs_prefix = getattr(workspace, "outputs_prefix", None)
+        prefixes = (outputs_prefix or "outputs/",)
+    prefix = str(prefixes[0]).strip("/")
+    return f"/{prefix}/{RUNTIME_SKILLS_DIR}" if prefix else f"/{RUNTIME_SKILLS_DIR}"
+
+
+def seed_runtime_skills(backend: Any, ctx: Any) -> list[str]:
     root = Path(__file__).parent / "skills"
     if not root.exists():
         return []
+    runtime_root = runtime_skills_root(ctx)
     uploads: list[tuple[str, bytes]] = []
     for path in root.rglob("*"):
         if path.is_file():
             rel = path.relative_to(root).as_posix()
-            uploads.append((RUNTIME_SKILLS_ROOT + rel, path.read_bytes()))
+            uploads.append((runtime_root + rel, path.read_bytes()))
     if uploads:
         backend.upload_files(uploads)
-        return [RUNTIME_SKILLS_ROOT]
+        return [runtime_root]
     return []
 ```
 

agent_builder/skills/workspace-artifact-safety/SKILL.md

@@ -77,21 +77,23 @@ async def run_checked(script: str, timeout_s: int = 120) -> dict[str, str | int]
     }
 ```
 
-Commands may write to `/workspace/outputs/...` for stable paths, or to normal
+Commands may write under the grant's `write_prefixes` for stable paths, or to
-tool defaults such as `/tmp`, `/root`, or `/app`. The platform sandbox mirrors
+normal tool defaults such as `/tmp`, `/root`, or `/app`. The platform sandbox
-changed files outside `/workspace` into `outputs/rootfs-captures/...`. Do not
+mirrors changed files outside `/workspace` into `outputs/rootfs-captures/...`.
-use `asyncio.create_subprocess_exec(...)` for durable outputs. Plain subprocesses
+Do not use `asyncio.create_subprocess_exec(...)` for durable outputs. Plain
-run inside the agent container, which is not mounted to the caller workspace and
+subprocesses run inside the agent container, which is not mounted to the caller
-is not rootfs-captured.
+workspace and is not rootfs-captured.
 
 If a nonzero command still produced a usable output, preserve the output and
 include the command failure as a warning instead of discarding the file.
 
 ## Scope Expansion
 
-Only use `ctx.request_scope(...)` when the public `@skill` declares
+Only use `ctx.request_scope(...)`, `ctx.ensure_read(...)`, or
+`ctx.ensure_write(...)` when the public `@skill` declares
 `allow_scope_expansion=True`. Explain the reason in plain language and keep
-the requested read/write patterns narrow.
+the requested read/write patterns narrow. New write prefixes always require
+caller approval.
 
 ```python
 @skill(
@@ -99,10 +101,9 @@ the requested read/write patterns narrow.
     allow_scope_expansion=True,
 )
 async def analyze_folder(ctx: RunContext[NoAuth], folder: str) -> dict[str, str]:
-    await ctx.request_scope(
+    await ctx.ensure_read(
         reason=f"Need to read {folder} to analyze the requested files",
-        read=(f"{folder.rstrip('/')}/**",),
+        patterns=(f"{folder.rstrip('/')}/**",),
-        mode="read_only",
     )
     return {"status": "approved"}
 ```

agent_builder/tools.py

@@ -28,7 +28,7 @@ if TYPE_CHECKING:
     from .config import Settings
 
 
-A2A_PACK_MIN_VERSION = "0.1.28"
+A2A_PACK_MIN_VERSION = "0.1.29"
 
 
 @dataclass(frozen=True)

requirements.txt

@@ -1,4 +1,4 @@
-a2a-pack>=0.1.28
+a2a-pack>=0.1.29
 httpx>=0.27
 boto3>=1.34
 deepagents>=0.5.0

tests/test_builder_prompt.py

@@ -23,6 +23,7 @@ class BuilderPromptTests(unittest.TestCase):
             agent_source,
         )
         self.assertIn("stream=True", agent_source)
+        self.assertIn('grant_write_prefixes=("agents/{name}/",)', agent_source)
 
     def test_outer_builder_accepts_user_creds_or_platform_llm_grants(self) -> None:
         agent_source = Path(__file__).resolve().parents[1].joinpath("agent.py").read_text()
@@ -93,18 +94,20 @@ class BuilderPromptTests(unittest.TestCase):
         self.assertIn("deepagent-agent-design", SYSTEM_PROMPT)
         self.assertIn("write_agent_skill", SYSTEM_PROMPT)
         self.assertIn("workspace_access = WorkspaceAccess.dynamic", SYSTEM_PROMPT)
+        self.assertIn("ensure_read/ensure_write", SYSTEM_PROMPT)
 
         files = _builder_skill_files()
         self.assertTrue(
             {path.split("/", 1)[0] for path in files}.issuperset(expected)
         )
-        self.assertIn("skills=[RUNTIME_SKILLS_ROOT]", files["deepagent-agent-design/SKILL.md"])
+        self.assertIn("skills=skill_sources or None", files["deepagent-agent-design/SKILL.md"])
         self.assertIn("from a2a_pack import", files["a2apack-agent-authoring/SKILL.md"])
         self.assertIn("ctx.workspace_shell", files["a2apack-agent-authoring/SKILL.md"])
         self.assertIn("create_deep_agent", files["deepagents-implementation-patterns/SKILL.md"])
         self.assertIn('config={"recursion_limit": 500}', files["deepagents-implementation-patterns/SKILL.md"])
         self.assertIn("ctx.write_artifact", files["workspace-artifact-safety/SKILL.md"])
         self.assertIn("ctx.workspace_shell", files["workspace-artifact-safety/SKILL.md"])
+        self.assertIn("write_prefixes", files["workspace-artifact-safety/SKILL.md"])
         self.assertIn("outputs/rootfs-captures/...", files["workspace-artifact-safety/SKILL.md"])
         self.assertIn("live_skills[].input_schema", files["agent-quality-review/SKILL.md"])
         self.assertIn("ctx.workspace_python", files["agent-quality-review/SKILL.md"])

tests/test_template_init.py

@@ -33,7 +33,8 @@ class TemplateInitTests(unittest.TestCase):
         self.assertIn('description = "Research helper"', files["agent.py"])
         self.assertIn("LLMProvisioning.CALLER_PROVIDED", files["agent.py"])
         self.assertIn("WorkspaceAccess.dynamic", files["agent.py"])
-        self.assertIn("RUNTIME_SKILLS_ROOT", files["agent.py"])
+        self.assertIn("RUNTIME_SKILLS_DIR", files["agent.py"])
+        self.assertIn("_runtime_skills_root", files["agent.py"])
         self.assertIn("DEEPAGENTS_RECURSION_LIMIT = 500", files["agent.py"])
         self.assertIn(
             'config={"recursion_limit": DEEPAGENTS_RECURSION_LIMIT}',