From d5a93339384a5e3d83cc7c9262a99ef4dd66a14d Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Mon, 13 Jul 2026 12:20:48 +0000 Subject: [PATCH] a2a-source-edit: write agent.py --- agent.py | 383 +++++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 383 insertions(+) create mode 100644 agent.py diff --git a/agent.py b/agent.py new file mode 100644 index 0000000..f68f8ce --- /dev/null +++ b/agent.py @@ -0,0 +1,383 @@ +"""Deterministic release validation agent for customer-integration-engineer. + +This agent validates a customer-integration-engineer release without editing the +validated source. It is intentionally local-logic only: no LLM credentials, +provider keys, code-editor-agent delegation, or cosmetic mutation paths. +""" +from __future__ import annotations + +import json +import re +from typing import Any +from urllib.parse import urlparse + +from pydantic import BaseModel, Field + +import a2a_pack as a2a +from a2a_pack import ( + A2AAgent, + NoAuth, + Pricing, + Resources, + RunContext, + WorkspaceAccess, + WorkspaceMode, +) + + +EXPECTED_AGENT_NAME = "customer-integration-engineer" +EXPECTED_LIVE_VERSION = "0.1.5" +EXPECTED_PUBLIC = False +EXPECTED_WORKSPACE_MODES = {"read_only", "read_write_overlay"} +EXPECTED_SKILL_COUNT = 5 +MAX_ALLOWED_IMPROVEMENT_ITERATIONS = 0 +SECRET_KEY_RE = re.compile(r"(api[_-]?key|token|secret|password|authorization|bearer)", re.I) +SECRET_VALUE_RE = re.compile( + r"(?i)(sk-[A-Za-z0-9_-]{12,}|xox[baprs]-[A-Za-z0-9-]{12,}|gh[pousr]_[A-Za-z0-9_]{20,}|bearer\s+[A-Za-z0-9._-]{16,})" +) + + +class CustomerIntegrationEngineerConfig(BaseModel): + """Static release-validation policy.""" + + target_agent_name: str = EXPECTED_AGENT_NAME + target_version: str = EXPECTED_LIVE_VERSION + required_skill_count: int = EXPECTED_SKILL_COUNT + + +class CardValidationInput(BaseModel): + """Bounded card details copied from the live Agent Card or registry view.""" + + card: dict[str, Any] = Field(description="Live Agent Card JSON for the target release.") + expected_version: str = Field(default=EXPECTED_LIVE_VERSION, max_length=32) + expected_skill_count: int = Field(default=EXPECTED_SKILL_COUNT, ge=1, le=20) + require_private: bool = True + + +class SmokeCallInput(BaseModel): + """Structured result from an Agent Studio delegated live smoke invocation.""" + + skill_name: str = Field(min_length=1, max_length=80) + input_args: dict[str, Any] = Field(default_factory=dict) + result: dict[str, Any] = Field(default_factory=dict) + status_code: int = Field(default=200, ge=100, le=599) + used_workspace_delegation: bool = True + delegated_workspace_mode: str = Field(default="read_only", max_length=64) + + +class ReviewerFinding(BaseModel): + severity: str = Field(description="critical, warning, info, or pass", max_length=32) + area: str = Field(max_length=120) + message: str = Field(max_length=1000) + + +class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, NoAuth]): + name = EXPECTED_AGENT_NAME + description = ( + "Validates the customer-integration-engineer release end to end while " + "preserving source unless a critical defect remains." + ) + version = "0.1.0" + + config_model = CustomerIntegrationEngineerConfig + auth_model = NoAuth + + pricing = Pricing( + price_per_call_usd=0.0, + caller_pays_llm=False, + notes="Deterministic validation checks only; no LLM credential required.", + ) + resources = Resources(cpu="250m", memory="512Mi", max_runtime_seconds=300) + workspace_access = WorkspaceAccess.dynamic( + max_files=64, + allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), + require_reason=False, + ) + tools_used = ("a2a-pack", "workspace-delegation-review") + + @a2a.tool( + description="Validate the live Agent Card version, privacy, five public skills, and workspace safety controls.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def validate_live_card( + self, + ctx: RunContext[NoAuth], + payload: CardValidationInput, + ) -> dict[str, Any]: + await ctx.emit_progress("validating live card contract") + card = payload.card + skills = _skills(card) + workspace = _workspace_access(card) + runtime = _runtime(card) + checks = [ + _check("agent_name", card.get("name") == self.config.target_agent_name, f"expected {self.config.target_agent_name}"), + _check("version", str(card.get("version")) == payload.expected_version, f"expected live version {payload.expected_version}"), + _check("skill_count", len(skills) == payload.expected_skill_count, f"expected {payload.expected_skill_count} skills, got {len(skills)}"), + _check("private_exposure", _is_private(card) if payload.require_private else True, "agent must remain private"), + _check("workspace_enabled", bool(workspace.get("enabled")), "WorkspaceAccess.dynamic must be enabled"), + _check("workspace_modes", EXPECTED_WORKSPACE_MODES <= set(_mode_values(workspace.get("allowed_modes"))), "requires read_only and read_write_overlay"), + _check("no_llm_required", not _runtime_requires_caller_llm(runtime), "validator/smoke path must not require caller LLM"), + ] + return _result("validate_live_card", checks, extra={"skills": [_skill_name(s) for s in skills]}) + + @a2a.tool( + description="Confirm the primary public skill has bounded typed inputs and no credential-shaped parameters.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def validate_primary_skill_schema( + self, + ctx: RunContext[NoAuth], + payload: CardValidationInput, + primary_skill_name: str = "", + ) -> dict[str, Any]: + await ctx.emit_progress("validating primary skill schema") + skills = _skills(payload.card) + primary = _select_skill(skills, primary_skill_name) + schema = _input_schema(primary) if primary else {} + properties = schema.get("properties") if isinstance(schema, dict) else {} + required = schema.get("required") if isinstance(schema, dict) else [] + property_names = sorted(properties) if isinstance(properties, dict) else [] + checks = [ + _check("primary_skill_present", primary is not None, "primary skill must exist"), + _check("schema_is_object", isinstance(schema, dict) and schema.get("type") == "object", "input_schema must be a JSON object"), + _check("bounded_property_count", isinstance(properties, dict) and 1 <= len(properties) <= 12, "primary skill should expose 1-12 typed inputs"), + _check("additional_properties_false", schema.get("additionalProperties") is False, "schema should reject unknown inputs"), + _check("all_inputs_typed", _all_properties_typed(properties), "each input property must declare a type or structured schema"), + _check("no_credential_inputs", not any(SECRET_KEY_RE.search(name) for name in property_names), "public inputs must not request secrets"), + _check("required_is_bounded", isinstance(required, list) and len(required) <= len(property_names), "required list must be bounded"), + ] + return _result( + "validate_primary_skill_schema", + checks, + extra={"primary_skill": _skill_name(primary) if primary else None, "input_names": property_names}, + ) + + @a2a.tool( + description="Review source-change policy for critical-only edits, no code-editor-agent use, and version bump discipline.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def review_source_change_policy( + self, + ctx: RunContext[NoAuth], + mutated_source: bool, + critical_defect_remaining: bool, + used_code_editor_agent: bool, + improvement_iterations: int = 0, + previous_version: str = EXPECTED_LIVE_VERSION, + resulting_version: str = EXPECTED_LIVE_VERSION, + changed_files: list[str] | None = None, + ) -> dict[str, Any]: + await ctx.emit_progress("reviewing source mutation policy") + files = changed_files or [] + checks = [ + _check("no_code_editor_agent", not used_code_editor_agent, "code-editor-agent must not be used"), + _check("iteration_limit", 0 <= improvement_iterations <= MAX_ALLOWED_IMPROVEMENT_ITERATIONS, "at most 0 improvement iterations allowed"), + _check("critical_only_mutation", (not mutated_source) or critical_defect_remaining, "source may change only when a critical defect remains"), + _check("version_bumped_on_mutation", (not mutated_source) or previous_version != resulting_version, "version must change after real code-editor mutation"), + _check("no_cosmetic_files", not _looks_cosmetic_only(files), "avoid cosmetic-only source changes"), + ] + return _result("review_source_change_policy", checks, extra={"changed_files": files}) + + @a2a.tool( + description="Validate a live smoke-call result from Agent Studio workspace delegation and redact any secret-like values.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def validate_smoke_call( + self, + ctx: RunContext[NoAuth], + smoke: SmokeCallInput, + ) -> dict[str, Any]: + await ctx.emit_progress("validating smoke-call result") + secret_paths = _secret_paths(smoke.result) + checks = [ + _check("http_success", 200 <= smoke.status_code < 300, f"status_code={smoke.status_code}"), + _check("structured_result", isinstance(smoke.result, dict) and bool(smoke.result), "smoke result must be a non-empty JSON object"), + _check("workspace_delegated", smoke.used_workspace_delegation, "smoke must use Agent Studio workspace delegation"), + _check("delegation_mode_safe", smoke.delegated_workspace_mode in EXPECTED_WORKSPACE_MODES, "delegated mode must be read_only or read_write_overlay"), + _check("no_secrets_returned", not secret_paths, "result must not contain secret-shaped keys or values"), + ] + return _result( + "validate_smoke_call", + checks, + extra={"skill_name": smoke.skill_name, "secret_like_paths": secret_paths}, + ) + + @a2a.tool( + description="Combine validation outputs and reviewer findings into a final go/no-go release report.", + timeout_seconds=60, + idempotent=True, + cost_class="cheap", + ) + async def final_release_report( + self, + ctx: RunContext[NoAuth], + validation_results: list[dict[str, Any]], + reviewer_findings: list[ReviewerFinding] | None = None, + warnings_to_accept: list[str] | None = None, + ) -> dict[str, Any]: + await ctx.emit_progress("building final release report") + findings = reviewer_findings or [] + warnings = warnings_to_accept or [] + criticals = [f.model_dump() for f in findings if f.severity.lower() == "critical"] + failed_checks = [ + {"result_index": i, "check": check} + for i, result in enumerate(validation_results) + for check in result.get("checks", []) + if isinstance(check, dict) and not check.get("ok") + ] + residual_warnings = [f.model_dump() for f in findings if f.severity.lower() == "warning"] + all_warnings_accounted = len(residual_warnings) == 0 or bool(warnings) + checks = [ + _check("zero_critical_findings", not criticals, "reviewer must report zero critical findings"), + _check("all_validation_checks_pass", not failed_checks, "all prior validation checks must pass"), + _check("warnings_patched_or_listed", all_warnings_accounted, "warnings must be patched or listed as residual risks"), + ] + status = "pass" if all(c["ok"] for c in checks) else "fail" + return { + "ok": status == "pass", + "status": status, + "target_agent": self.config.target_agent_name, + "target_version": self.config.target_version, + "checks": checks, + "critical_findings": criticals, + "failed_checks": failed_checks, + "residual_risks": warnings or [f["message"] for f in residual_warnings], + "secrets_included": False, + } + + +def _runtime(card: dict[str, Any]) -> dict[str, Any]: + value = card.get("runtime") + return value if isinstance(value, dict) else {} + + +def _workspace_access(card: dict[str, Any]) -> dict[str, Any]: + for key in ("workspace_access", "workspaceAccess", "workspace"): + value = card.get(key) + if isinstance(value, dict): + return value + runtime = _runtime(card) + for key in ("workspace_access", "workspaceAccess", "workspace"): + value = runtime.get(key) + if isinstance(value, dict): + return value + return {} + + +def _skills(card: dict[str, Any]) -> list[dict[str, Any]]: + value = card.get("skills") or card.get("tools") or [] + return [item for item in value if isinstance(item, dict)] if isinstance(value, list) else [] + + +def _skill_name(skill: dict[str, Any] | None) -> str: + if not isinstance(skill, dict): + return "" + return str(skill.get("name") or skill.get("id") or "") + + +def _select_skill(skills: list[dict[str, Any]], name: str) -> dict[str, Any] | None: + if name: + return next((skill for skill in skills if _skill_name(skill) == name), None) + return skills[0] if skills else None + + +def _input_schema(skill: dict[str, Any] | None) -> dict[str, Any]: + if not isinstance(skill, dict): + return {} + for key in ("input_schema", "inputSchema", "parameters"): + value = skill.get(key) + if isinstance(value, dict): + return value + return {} + + +def _mode_values(value: Any) -> list[str]: + if isinstance(value, list): + raw = value + elif isinstance(value, tuple): + raw = list(value) + elif isinstance(value, str): + raw = [value] + else: + raw = [] + return [str(item.get("value") if isinstance(item, dict) else item).lower() for item in raw] + + +def _is_private(card: dict[str, Any]) -> bool: + expose = card.get("expose") + if isinstance(expose, dict) and "public" in expose: + return expose.get("public") is False + visibility = str(card.get("visibility") or card.get("access") or "").lower() + if visibility in {"private", "unlisted"}: + return True + public = card.get("public") + return public is False if isinstance(public, bool) else True + + +def _runtime_requires_caller_llm(runtime: dict[str, Any]) -> bool: + provisioning = str(runtime.get("llm_provisioning") or runtime.get("llmProvisioning") or "").lower() + pricing = runtime.get("pricing") if isinstance(runtime.get("pricing"), dict) else {} + return provisioning in {"platform", "caller_provided", "platform_or_caller_provided"} and bool(pricing.get("caller_pays_llm")) + + +def _all_properties_typed(properties: Any) -> bool: + if not isinstance(properties, dict) or not properties: + return False + for schema in properties.values(): + if not isinstance(schema, dict): + return False + if not any(key in schema for key in ("type", "anyOf", "oneOf", "allOf", "$ref")): + return False + return True + + +def _looks_cosmetic_only(paths: list[str]) -> bool: + if not paths: + return False + non_cosmetic_exts = {".py", ".yaml", ".yml", ".toml", ".json", ".txt"} + cosmetic_markers = {"readme", "docs/", ".md"} + lowered = [p.lower() for p in paths] + return all(any(marker in p for marker in cosmetic_markers) for p in lowered) and not any( + any(p.endswith(ext) for ext in non_cosmetic_exts - {".txt"}) for p in lowered + ) + + +def _secret_paths(value: Any, prefix: str = "$") -> list[str]: + found: list[str] = [] + if isinstance(value, dict): + for key, item in value.items(): + path = f"{prefix}.{key}" + if SECRET_KEY_RE.search(str(key)): + found.append(path) + found.extend(_secret_paths(item, path)) + elif isinstance(value, list): + for i, item in enumerate(value[:100]): + found.extend(_secret_paths(item, f"{prefix}[{i}]")) + elif isinstance(value, str) and SECRET_VALUE_RE.search(value): + found.append(prefix) + return sorted(set(found)) + + +def _check(name: str, ok: bool, detail: str) -> dict[str, Any]: + return {"name": name, "ok": bool(ok), "detail": detail} + + +def _result(name: str, checks: list[dict[str, Any]], *, extra: dict[str, Any] | None = None) -> dict[str, Any]: + ok = all(item["ok"] for item in checks) + payload = {"ok": ok, "status": "pass" if ok else "fail", "checkset": name, "checks": checks, "secrets_included": False} + if extra: + payload.update(extra) + return json.loads(json.dumps(payload, default=str)) + + +def _is_safe_url(url: str) -> bool: + parsed = urlparse(url) + return parsed.scheme in {"https", "http"} and bool(parsed.netloc)