a2a-source-edit: remove agent.py

2026-07-13 12:20:39 +00:00
parent 96021d17c1
commit cec48420c0
1 changed files with 0 additions and 383 deletions
--- a/agent.py
+++ b/agent.py
@@ -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)