From a3466c7a9a160867379a6e44f8cbea9bcb389011 Mon Sep 17 00:00:00 2001 From: robert Date: Tue, 19 May 2026 09:10:20 -0300 Subject: [PATCH] Track builder workspace deployment base --- agent_builder/builder.py | 12 ++- agent_builder/tools.py | 172 ++++++++++++++++++++++++++++++++++-- tests/test_template_init.py | 74 ++++++++++++++++ 3 files changed, 251 insertions(+), 7 deletions(-) diff --git a/agent_builder/builder.py b/agent_builder/builder.py index 949a2d8..39a54aa 100644 --- a/agent_builder/builder.py +++ b/agent_builder/builder.py @@ -110,9 +110,13 @@ Your tools: - test_agent_in_sandbox(name) — pip install + ``a2a card`` round-trip; check exit_code == 0 and the card JSON looks right - - cp_deploy_tarball(name, version="0.1.0", public=True) + - cp_deploy_tarball(name, version="0.1.0", public=True, force=False) — ship it to the platform; returns the - public URL + public URL. It refuses stale MinIO + workspaces if the managed repo changed + elsewhere; use force=True only after + the user explicitly accepts replacing + the current repo source. - cp_refresh_agent(name) — force the control plane to re-fetch the agent's live card after an out-of-band redeploy @@ -144,6 +148,10 @@ Discipline: scaffold. - 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. - 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 b406925..b00f592 100644 --- a/agent_builder/tools.py +++ b/agent_builder/tools.py @@ -92,6 +92,7 @@ def _ensure_bucket(s3: Any, bucket: str) -> None: _AGENT_NAME_RE = re.compile(r"^[a-z][a-z0-9-]{1,62}$") +_BUILDER_STATE_FILE = ".a2a-builder-state.json" def _validate_name(name: str) -> None: @@ -106,6 +107,10 @@ def _agent_prefix(name: str) -> str: return f"agents/{name}/" +def _builder_state_key(name: str) -> str: + return _agent_prefix(name) + _BUILDER_STATE_FILE + + def build_tools(ctx: ToolContext) -> list[Any]: bucket = ctx.bucket settings = ctx.settings @@ -156,8 +161,11 @@ def build_tools(ctx: ToolContext) -> list[Any]: paginator = s3.get_paginator("list_objects_v2") for page in paginator.paginate(Bucket=bucket, Prefix=prefix): for obj in page.get("Contents") or (): + rel = obj["Key"][len(prefix):] + if rel == _BUILDER_STATE_FILE: + continue out.append({ - "path": obj["Key"][len(prefix):], + "path": rel, "size": int(obj.get("Size") or 0), }) return json.dumps({"agent": name, "files": out}) @@ -173,7 +181,10 @@ def build_tools(ctx: ToolContext) -> list[Any]: prefix = _agent_prefix(name) except ValueError as exc: return json.dumps({"error": str(exc)}) - key = prefix + path.lstrip("/") + rel = path.lstrip("/") + if rel == _BUILDER_STATE_FILE: + return json.dumps({"error": f"{_BUILDER_STATE_FILE} is managed by deploys"}) + key = prefix + rel s3.put_object( Bucket=bucket, Key=key, Body=content.encode("utf-8"), ContentType="text/plain; charset=utf-8", @@ -187,7 +198,10 @@ def build_tools(ctx: ToolContext) -> list[Any]: prefix = _agent_prefix(name) except ValueError as exc: return json.dumps({"error": str(exc)}) - key = prefix + path.lstrip("/") + rel = path.lstrip("/") + if rel == _BUILDER_STATE_FILE: + return json.dumps({"error": f"{_BUILDER_STATE_FILE} is managed by deploys"}) + key = prefix + rel try: obj = s3.get_object(Bucket=bucket, Key=key) except ClientError as exc: @@ -246,7 +260,12 @@ def build_tools(ctx: ToolContext) -> list[Any]: }) @tool - async def cp_deploy_tarball(name: str, version: str = "0.1.0", public: bool = True) -> str: + async def cp_deploy_tarball( + name: str, + version: str = "0.1.0", + public: bool = True, + force: bool = False, + ) -> str: """Tar the agent at ``agents//`` and POST it to the control plane's ``/v1/agents/from-tarball`` endpoint on the user's behalf, then BLOCK until the live ``.well-known/agent-card`` reports the @@ -263,6 +282,11 @@ def build_tools(ctx: ToolContext) -> list[Any]: actually accepts. Mismatches cause downstream ``agent 400`` from ``call_agent``. + ``force`` defaults to false so a stale MinIO workspace cannot + silently overwrite source that was deployed from a user's computer + or another tool. Only set ``force=True`` after the user explicitly + accepts replacing the current managed repo contents. + Requires the orchestrator to have forwarded the user's CP JWT (the platform does this automatically when this agent's Card declared ``wants_cp_jwt=True``). @@ -276,6 +300,23 @@ def build_tools(ctx: ToolContext) -> list[Any]: "error": "no CP JWT forwarded — agent declaration missing wants_cp_jwt=True", }) + try: + latest_deploy = await _latest_cp_deployment(settings.cp_url, ctx.cp_jwt, name) + except RuntimeError as exc: + return json.dumps({"error": str(exc)}) + builder_state = _read_builder_state(s3, bucket, name) + drift = _deployment_drift_error( + name, + latest_deploy, + builder_state, + force=force, + ) + if drift is not None: + return json.dumps(drift) + base_head = builder_state.get("repo_head_sha") + if force: + base_head = None + bundle_bytes = _tarball_workspace_dir(s3, bucket, prefix) if not bundle_bytes: return json.dumps({"error": f"no files at agents/{name}/"}) @@ -290,6 +331,8 @@ def build_tools(ctx: ToolContext) -> list[Any]: "entrypoint": entrypoint, "public": "true" if public else "false", "description": "Built by agent-builder.", } + if isinstance(base_head, str) and base_head: + data["base_head_sha"] = base_head try: async with httpx.AsyncClient(timeout=60.0) as c: r = await c.post( @@ -305,6 +348,21 @@ def build_tools(ctx: ToolContext) -> list[Any]: name_ = body.get("name") version_ = body.get("version") url_ = body.get("url") + head_sha = body.get("head_sha") + if isinstance(head_sha, str) and head_sha: + _write_builder_state( + s3, + bucket, + name, + { + "agent": name, + "repo_head_sha": head_sha, + "deployment_id": body.get("deployment_id"), + "version": version_, + "updated_at": int(time.time()), + "source": "agent-builder", + }, + ) # CP returns immediately with status="building". The agent isn't # actually callable until the pod rolls + serves the new card. @@ -320,6 +378,9 @@ def build_tools(ctx: ToolContext) -> list[Any]: "version": version_, "status": body.get("status"), "url": url_, + "head_sha": head_sha, + "deployment_id": body.get("deployment_id"), + "workspace_base_head_sha": base_head, "live": live, } if live: @@ -478,6 +539,7 @@ def _tarball_workspace_dir(s3: Any, bucket: str, prefix: str) -> bytes: gzipped tarball whose entries are relative to ``prefix``.""" paginator = s3.get_paginator("list_objects_v2") buf = io.BytesIO() + added = False with tarfile.open(fileobj=buf, mode="w:gz") as tf: for page in paginator.paginate(Bucket=bucket, Prefix=prefix): for obj in page.get("Contents") or (): @@ -485,12 +547,112 @@ def _tarball_workspace_dir(s3: Any, bucket: str, prefix: str) -> bytes: rel = key[len(prefix):] if not rel: continue + if rel == _BUILDER_STATE_FILE: + continue body = s3.get_object(Bucket=bucket, Key=key)["Body"].read() info = tarfile.TarInfo(name=rel) info.size = len(body) info.mode = 0o644 tf.addfile(info, io.BytesIO(body)) - return buf.getvalue() if buf.getbuffer().nbytes > 0 else b"" + added = True + return buf.getvalue() if added else b"" + + +def _read_builder_state(s3: Any, bucket: str, name: str) -> dict[str, Any]: + try: + obj = s3.get_object(Bucket=bucket, Key=_builder_state_key(name)) + except ClientError as exc: + code = exc.response.get("Error", {}).get("Code") + if code in {"404", "NoSuchKey", "NotFound"}: + return {} + return {} + try: + data = json.loads(obj["Body"].read().decode("utf-8")) + except (TypeError, ValueError, UnicodeDecodeError): + return {} + return data if isinstance(data, dict) else {} + + +def _write_builder_state( + s3: Any, + bucket: str, + name: str, + state: dict[str, Any], +) -> None: + body = json.dumps(state, sort_keys=True, indent=2).encode("utf-8") + s3.put_object( + Bucket=bucket, + Key=_builder_state_key(name), + Body=body, + ContentType="application/json", + ) + + +async def _latest_cp_deployment( + cp_url: str, + cp_jwt: str, + name: str, +) -> dict[str, Any] | None: + try: + async with httpx.AsyncClient(timeout=15.0) as c: + r = await c.get( + f"{cp_url}/v1/agents/{name}/deployments", + headers={"authorization": f"bearer {cp_jwt}"}, + ) + except httpx.HTTPError as exc: + raise RuntimeError(f"cp unreachable: {exc}") from exc + if r.status_code == 404: + return None + if r.status_code >= 400: + raise RuntimeError(f"cp {r.status_code}: {r.text[:1000]}") + body = r.json() or [] + if not isinstance(body, list) or not body: + return None + first = body[0] + return first if isinstance(first, dict) else None + + +def _deployment_drift_error( + name: str, + latest_deployment: dict[str, Any] | None, + builder_state: dict[str, Any], + *, + force: bool = False, +) -> dict[str, Any] | None: + if force or latest_deployment is None: + 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, + "error": "workspace_untracked", + "agent": name, + "message": ( + "This agent already has managed source, but the MinIO " + "workspace has no deploy base marker. Refresh it before " + "deploying, or use force=True only if the user wants to " + "replace the repo." + ), + "current_deployment_id": latest_deployment.get("deploy_id"), + } + return None def _read_entrypoint(bundle: bytes) -> str | None: diff --git a/tests/test_template_init.py b/tests/test_template_init.py index 5c72bd2..1554df0 100644 --- a/tests/test_template_init.py +++ b/tests/test_template_init.py @@ -5,7 +5,10 @@ import tarfile import unittest from agent_builder.tools import ( + _BUILDER_STATE_FILE, + _deployment_drift_error, _render_a2a_init_template, + _tarball_workspace_dir, ) @@ -37,6 +40,49 @@ class TemplateInitTests(unittest.TestCase): self.assertIn("A2AAgent", files["agent.py"]) + def test_deployment_drift_error_blocks_stale_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["current_head_sha"], "abc1234") + self.assertIsNone(err["workspace_base_head_sha"]) + + def test_deployment_drift_error_allows_matching_or_forced_workspace(self) -> None: + latest = {"deploy_id": "dpl_1", "head_sha": "abc1234"} + + self.assertIsNone( + _deployment_drift_error( + "demo-agent", + latest, + {"repo_head_sha": "abc1234"}, + force=False, + ) + ) + self.assertIsNone( + _deployment_drift_error( + "demo-agent", + latest, + {"repo_head_sha": "old9999"}, + force=True, + ) + ) + + def test_tarball_workspace_excludes_builder_state(self) -> None: + prefix = "agents/demo-agent/" + s3 = _FakeS3({ + prefix + "agent.py": b"print('ok')\n", + prefix + _BUILDER_STATE_FILE: b'{"repo_head_sha":"abc1234"}', + }) + + bundle = _tarball_workspace_dir(s3, "bucket", prefix) + + with tarfile.open(fileobj=io.BytesIO(bundle), mode="r:gz") as tf: + self.assertEqual(tf.getnames(), ["agent.py"]) + def _tarball(files: dict[str, str]) -> bytes: buf = io.BytesIO() @@ -47,3 +93,31 @@ def _tarball(files: dict[str, str]) -> bytes: info.size = len(body) tf.addfile(info, io.BytesIO(body)) return buf.getvalue() + + +class _FakeS3: + def __init__(self, objects: dict[str, bytes]) -> None: + self.objects = objects + + def get_paginator(self, name: str) -> "_FakePaginator": + assert name == "list_objects_v2" + return _FakePaginator(self.objects) + + def get_object(self, *, Bucket: str, Key: str) -> dict[str, io.BytesIO]: + return {"Body": io.BytesIO(self.objects[Key])} + + +class _FakePaginator: + def __init__(self, objects: dict[str, bytes]) -> None: + self.objects = objects + + def paginate(self, *, Bucket: str, Prefix: str) -> list[dict[str, object]]: + return [ + { + "Contents": [ + {"Key": key, "Size": len(value)} + for key, value in sorted(self.objects.items()) + if key.startswith(Prefix) + ] + } + ]