From d43533df808d17e820eed35c0b94cbf44f85456d Mon Sep 17 00:00:00 2001 From: robert Date: Tue, 19 May 2026 09:18:39 -0300 Subject: [PATCH] Sync builder workspaces from managed repo --- agent_builder/builder.py | 14 +++- agent_builder/tools.py | 131 +++++++++++++++++++++++++++++++----- tests/test_template_init.py | 63 +++++++++++++++-- 3 files changed, 185 insertions(+), 23 deletions(-) diff --git a/agent_builder/builder.py b/agent_builder/builder.py index 39a54aa..01a4b6f 100644 --- a/agent_builder/builder.py +++ b/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// 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 - force=True unless the user explicitly says to overwrite the managed - repo with the current MinIO files. + use sync_agent_workspace_from_repo when the user wants the builder to + continue from the current repo. Do not use force=True unless the user + 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, diff --git a/agent_builder/tools.py b/agent_builder/tools.py index b00f592..68bf3b4 100644 --- a/agent_builder/tools.py +++ b/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//`` 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 diff --git a/tests/test_template_init.py b/tests/test_template_init.py index 1554df0..b5345c3 100644 --- a/tests/test_template_init.py +++ b/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: