From cec48420c06e1877ec4ddc34855a886595819a19 Mon Sep 17 00:00:00 2001 From: a2a-cloud Date: Mon, 13 Jul 2026 12:20:39 +0000 Subject: [PATCH] a2a-source-edit: remove agent.py --- agent.py | 383 ------------------------------------------------------- 1 file changed, 383 deletions(-) delete mode 100644 agent.py diff --git a/agent.py b/agent.py deleted file mode 100644 index f68f8ce..0000000 --- a/agent.py +++ /dev/null @@ -1,383 +0,0 @@ -"""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)