Sync builder workspaces from managed repo

agent_builder/builder.py

@@ -117,6 +117,13 @@ Your tools:
                                     elsewhere; use force=True only after
                                     the user explicitly accepts replacing
                                     the current repo source.
+  - sync_agent_workspace_from_repo(name)
+                                  — replace agents/<name>/ in MinIO with
+                                    the current managed repo source and
+                                    record its repo base marker. Use this
+                                    before editing an existing deployed
+                                    agent, or when deploy reports
+                                    workspace_untracked/workspace_drift.
   - cp_refresh_agent(name)        — force the control plane to re-fetch
                                     the agent's live card after an
                                     out-of-band redeploy
@@ -149,9 +156,10 @@ Discipline:
   - When deploying, return the URL the platform gave back to the user so
     they can curl it / share it.
   - If cp_deploy_tarball returns workspace_drift or workspace_untracked,
-    stop and report that the builder workspace is stale. Do not use
+    use sync_agent_workspace_from_repo when the user wants the builder to
-    force=True unless the user explicitly says to overwrite the managed
+    continue from the current repo. Do not use force=True unless the user
-    repo with the current MinIO files.
+    explicitly says to overwrite the managed repo with the current MinIO
+    files.
   - After cp_deploy_tarball returns, compare ``live_skills[].input_schema``
     with the skill signatures you wrote. If an argument is missing from the
     live schema, treat the deploy as failed, fix the source/schema version,

agent_builder/tools.py

@@ -443,6 +443,63 @@ def build_tools(ctx: ToolContext) -> list[Any]:
             "card": body.get("card") or {},
         })
 
+    @tool
+    async def sync_agent_workspace_from_repo(name: str) -> str:
+        """Replace ``agents/<name>/`` in MinIO with the current managed repo source.
+
+        Use this before editing an existing deployed agent, or after
+        ``cp_deploy_tarball`` returns ``workspace_untracked`` /
+        ``workspace_drift``. It downloads the owner's current managed repo
+        through the control plane, deletes the old MinIO files for that
+        agent, writes the repo source files, and records the repo head as the
+        deploy base marker. This intentionally overwrites the builder
+        workspace copy so it matches the repo.
+        """
+        try:
+            prefix = _agent_prefix(name)
+        except ValueError as exc:
+            return json.dumps({"error": str(exc)})
+        if not ctx.cp_jwt:
+            return json.dumps({
+                "error": "no CP JWT forwarded — agent declaration missing wants_cp_jwt=True",
+            })
+        try:
+            async with httpx.AsyncClient(timeout=60.0) as c:
+                r = await c.get(
+                    f"{settings.cp_url}/v1/agents/{name}/source",
+                    headers={"authorization": f"bearer {ctx.cp_jwt}"},
+                )
+        except httpx.HTTPError as exc:
+            return json.dumps({"error": f"cp unreachable: {exc}"})
+        if r.status_code >= 400:
+            return json.dumps({"error": f"cp {r.status_code}", "detail": r.text[:1000]})
+        head_sha = r.headers.get("x-a2a-repo-head-sha")
+        try:
+            files = _files_from_tarball(r.content)
+        except ValueError as exc:
+            return json.dumps({"error": f"invalid source tarball: {exc}"})
+        if not files:
+            return json.dumps({"error": "managed repo source export was empty"})
+        written = _replace_workspace_files(s3, bucket, prefix, files)
+        _write_builder_state(
+            s3,
+            bucket,
+            name,
+            {
+                "agent": name,
+                "repo_head_sha": head_sha,
+                "file_count": len(written),
+                "updated_at": int(time.time()),
+                "source": "repo-sync",
+            },
+        )
+        return json.dumps({
+            "ok": True,
+            "agent": name,
+            "repo_head_sha": head_sha,
+            "files": written,
+        })
+
     @tool
     def list_a2a_pack(subdir: str = "") -> str:
         """List ``.py`` files under the installed ``a2a_pack`` package.
@@ -527,6 +584,7 @@ def build_tools(ctx: ToolContext) -> list[Any]:
     return [
         init_agent_template, list_agent_files, write_agent_file, read_agent_file,
         test_agent_in_sandbox, cp_deploy_tarball, cp_refresh_agent,
+        sync_agent_workspace_from_repo,
         list_a2a_pack, read_a2a_pack,
     ]
 
@@ -588,6 +646,62 @@ def _write_builder_state(
     )
 
 
+def _delete_workspace_objects(s3: Any, bucket: str, prefix: str) -> None:
+    paginator = s3.get_paginator("list_objects_v2")
+    batch: list[dict[str, str]] = []
+    for page in paginator.paginate(Bucket=bucket, Prefix=prefix):
+        for obj in page.get("Contents") or ():
+            batch.append({"Key": obj["Key"]})
+            if len(batch) >= 1000:
+                s3.delete_objects(Bucket=bucket, Delete={"Objects": batch})
+                batch = []
+    if batch:
+        s3.delete_objects(Bucket=bucket, Delete={"Objects": batch})
+
+
+def _files_from_tarball(bundle: bytes) -> dict[str, bytes]:
+    out: dict[str, bytes] = {}
+    try:
+        with tarfile.open(fileobj=io.BytesIO(bundle), mode="r:gz") as tf:
+            for member in tf.getmembers():
+                if not member.isfile():
+                    continue
+                rel = member.name
+                while rel.startswith("./"):
+                    rel = rel[2:]
+                if not rel or rel.startswith("/") or ".." in rel.split("/"):
+                    raise ValueError(f"unsafe member path: {member.name}")
+                if rel == _BUILDER_STATE_FILE:
+                    continue
+                extracted = tf.extractfile(member)
+                if extracted is None:
+                    continue
+                out[rel] = extracted.read()
+    except tarfile.TarError as exc:
+        raise ValueError(str(exc)) from exc
+    return out
+
+
+def _replace_workspace_files(
+    s3: Any,
+    bucket: str,
+    prefix: str,
+    files: dict[str, bytes],
+) -> list[dict[str, Any]]:
+    _delete_workspace_objects(s3, bucket, prefix)
+    written: list[dict[str, Any]] = []
+    for rel, body in sorted(files.items()):
+        key = prefix + rel
+        s3.put_object(
+            Bucket=bucket,
+            Key=key,
+            Body=body,
+            ContentType="application/octet-stream",
+        )
+        written.append({"path": rel, "size": len(body)})
+    return written
+
+
 async def _latest_cp_deployment(
     cp_url: str,
     cp_jwt: str,
@@ -623,22 +737,6 @@ def _deployment_drift_error(
         return None
     latest_head = latest_deployment.get("head_sha")
     base_head = builder_state.get("repo_head_sha")
-    if isinstance(latest_head, str) and latest_head:
-        if isinstance(base_head, str) and base_head == latest_head:
-            return None
-        return {
-            "ok": False,
-            "error": "workspace_drift",
-            "agent": name,
-            "message": (
-                "This MinIO workspace is not based on the latest managed "
-                "repo deployment. Refresh it before deploying, or use "
-                "force=True only if the user wants to replace the repo."
-            ),
-            "workspace_base_head_sha": base_head,
-            "current_head_sha": latest_head,
-            "current_deployment_id": latest_deployment.get("deploy_id"),
-        }
     if not isinstance(base_head, str) or not base_head:
         return {
             "ok": False,
@@ -650,6 +748,7 @@ def _deployment_drift_error(
                 "deploying, or use force=True only if the user wants to "
                 "replace the repo."
             ),
+            "current_head_sha": latest_head,
             "current_deployment_id": latest_deployment.get("deploy_id"),
         }
     return None

tests/test_template_init.py

@@ -7,6 +7,8 @@ import unittest
 from agent_builder.tools import (
     _BUILDER_STATE_FILE,
     _deployment_drift_error,
+    _files_from_tarball,
+    _replace_workspace_files,
     _render_a2a_init_template,
     _tarball_workspace_dir,
 )
@@ -40,18 +42,17 @@ class TemplateInitTests(unittest.TestCase):
 
         self.assertIn("A2AAgent", files["agent.py"])
 
-    def test_deployment_drift_error_blocks_stale_workspace(self) -> None:
+    def test_deployment_drift_error_blocks_untracked_existing_workspace(self) -> None:
         latest = {"deploy_id": "dpl_1", "head_sha": "abc1234"}
 
         err = _deployment_drift_error("demo-agent", latest, {}, force=False)
 
         self.assertIsNotNone(err)
         assert err is not None
-        self.assertEqual(err["error"], "workspace_drift")
+        self.assertEqual(err["error"], "workspace_untracked")
         self.assertEqual(err["current_head_sha"], "abc1234")
-        self.assertIsNone(err["workspace_base_head_sha"])
 
-    def test_deployment_drift_error_allows_matching_or_forced_workspace(self) -> None:
+    def test_deployment_drift_error_allows_tracked_or_forced_workspace(self) -> None:
         latest = {"deploy_id": "dpl_1", "head_sha": "abc1234"}
 
         self.assertIsNone(
@@ -62,6 +63,14 @@ class TemplateInitTests(unittest.TestCase):
                 force=False,
             )
         )
+        self.assertIsNone(
+            _deployment_drift_error(
+                "demo-agent",
+                latest,
+                {"repo_head_sha": "repohead999"},
+                force=False,
+            )
+        )
         self.assertIsNone(
             _deployment_drift_error(
                 "demo-agent",
@@ -83,6 +92,38 @@ class TemplateInitTests(unittest.TestCase):
         with tarfile.open(fileobj=io.BytesIO(bundle), mode="r:gz") as tf:
             self.assertEqual(tf.getnames(), ["agent.py"])
 
+    def test_files_from_tarball_rejects_unsafe_paths(self) -> None:
+        bundle = _tarball({"../agent.py": "bad"})
+
+        with self.assertRaises(ValueError):
+            _files_from_tarball(bundle)
+
+    def test_replace_workspace_files_deletes_stale_objects(self) -> None:
+        prefix = "agents/demo-agent/"
+        s3 = _FakeS3({
+            prefix + "old.py": b"old",
+            prefix + _BUILDER_STATE_FILE: b"{}",
+        })
+
+        written = _replace_workspace_files(
+            s3,
+            "bucket",
+            prefix,
+            {"agent.py": b"print('ok')\n", "a2a.yaml": b"name: demo-agent\n"},
+        )
+
+        self.assertEqual(
+            sorted(s3.objects),
+            [prefix + "a2a.yaml", prefix + "agent.py"],
+        )
+        self.assertEqual(
+            written,
+            [
+                {"path": "a2a.yaml", "size": len(b"name: demo-agent\n")},
+                {"path": "agent.py", "size": len(b"print('ok')\n")},
+            ],
+        )
+
 
 def _tarball(files: dict[str, str]) -> bytes:
     buf = io.BytesIO()
@@ -106,6 +147,20 @@ class _FakeS3:
     def get_object(self, *, Bucket: str, Key: str) -> dict[str, io.BytesIO]:
         return {"Body": io.BytesIO(self.objects[Key])}
 
+    def put_object(
+        self,
+        *,
+        Bucket: str,
+        Key: str,
+        Body: bytes,
+        ContentType: str,
+    ) -> None:
+        self.objects[Key] = bytes(Body)
+
+    def delete_objects(self, *, Bucket: str, Delete: dict[str, object]) -> None:
+        for obj in Delete["Objects"]:  # type: ignore[index]
+            self.objects.pop(obj["Key"], None)
+
 
 class _FakePaginator:
     def __init__(self, objects: dict[str, bytes]) -> None: