From d9313ef6e863c5a524b1eed7bcc1eb4eef75d217 Mon Sep 17 00:00:00 2001 From: a2a-cloud operator Date: Mon, 13 Jul 2026 09:29:26 -0300 Subject: [PATCH] restore: customer integration agent 0.1.5 --- .a2a-builder-state.json | 8 - a2a.yaml | 6 +- agent.py | 1387 +++++++++++++++---- requirements.txt | 4 + tests/test_agent.py | 64 - tests/test_customer_integration_engineer.py | 235 +++- tests/test_runtime_safety.py | 62 +- 7 files changed, 1381 insertions(+), 385 deletions(-) delete mode 100644 .a2a-builder-state.json delete mode 100644 tests/test_agent.py diff --git a/.a2a-builder-state.json b/.a2a-builder-state.json deleted file mode 100644 index 251cf95..0000000 --- a/.a2a-builder-state.json +++ /dev/null @@ -1,8 +0,0 @@ -{ - "agent": "customer-integration-engineer", - "deployment_id": "dpl_91fd10b692b84f88", - "repo_head_sha": "4f3a5fbbee1abe61138a1b801c6e6924a961f799", - "source": "agent-builder", - "updated_at": 1783945337, - "version": "0.1.0" -} \ No newline at end of file diff --git a/a2a.yaml b/a2a.yaml index fda0cb7..aac123c 100644 --- a/a2a.yaml +++ b/a2a.yaml @@ -1,10 +1,10 @@ name: customer-integration-engineer -version: 0.1.0 +version: 0.1.5 entrypoint: agent:CustomerIntegrationEngineer expose: public: false runtime: resources: - cpu: "250m" + cpu: "500m" memory: 512Mi - max_runtime_seconds: 300 + max_runtime_seconds: 600 diff --git a/agent.py b/agent.py index f68f8ce..17a07c0 100644 --- a/agent.py +++ b/agent.py @@ -1,21 +1,30 @@ -"""Deterministic release validation agent for customer-integration-engineer. +"""Production-safe deterministic customer integration engineering agent. -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. +This agent guides an integration from requirements through generated workspace +artifacts and sandbox-style acceptance evidence. It deliberately does not call +an LLM, read provider credentials, execute arbitrary shell, deploy, merge, or +mutate customer systems. All customer credentials stay behind caller-managed +secret references. """ from __future__ import annotations +import asyncio +import hashlib +import ipaddress import json import re -from typing import Any +import socket +import time +from datetime import UTC, datetime +from typing import Any, Literal from urllib.parse import urlparse -from pydantic import BaseModel, Field - import a2a_pack as a2a from a2a_pack import ( A2AAgent, + ConsumerSetup, + ConsumerSetupField, + EgressPolicy, NoAuth, Pricing, Resources, @@ -23,361 +32,1139 @@ from a2a_pack import ( WorkspaceAccess, WorkspaceMode, ) +from a2a_pack.context import AgentEvent +from pydantic import BaseModel, ConfigDict, Field, field_validator, model_validator -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,})" -) +MAX_JSON_BYTES = 256_000 +MAX_TEXT_BYTES = 128_000 +APPROVAL_PHRASE = "I_APPROVE_THE_LISTED_STEPS" +OUTPUT_ROOT = "outputs/integrations" +SECRET_WORDS = ("secret", "token", "password", "passwd", "apikey", "api_key", "authorization", "bearer") -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 +def _tool_input_schema(request_model: type[BaseModel]) -> dict[str, Any]: + return { + "type": "object", + "required": ["request"], + "additionalProperties": False, + "properties": {"request": request_model.model_json_schema()}, + } -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 StrictModel(BaseModel): + model_config = ConfigDict(extra="forbid", str_strip_whitespace=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 AuditEntry(StrictModel): + ts: str + action: str + integration_id: str + details: dict[str, Any] = Field(default_factory=dict) -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 ArtifactRecord(StrictModel): + path: str + sha256: str + size_bytes: int = Field(ge=0) + mime_type: str + + +class SkillStatus(StrictModel): + status: Literal["ok", "needs_setup", "blocked", "failed"] + integration_id: str + dry_run: bool + audit: list[AuditEntry] = Field(default_factory=list) + warnings: list[str] = Field(default_factory=list) + residual_risks: list[str] = Field(default_factory=list) + + +class SystemSpec(StrictModel): + name: str = Field(min_length=1, max_length=80) + kind: Literal[ + "rest_api", + "webhook", + "database", + "object_storage", + "observability", + "gitea", + "kubernetes", + "argo_cd", + "queue", + "other", + ] + base_url: str | None = Field(default=None, max_length=2048) + environment: str = Field(default="sandbox", max_length=64) + owner: str | None = Field(default=None, max_length=120) + notes: str | None = Field(default=None, max_length=2000) + + @field_validator("base_url") + @classmethod + def _safe_url_text(cls, value: str | None) -> str | None: + if value is None or not value: + return None + parsed = urlparse(value) + if parsed.scheme not in {"https", "http"} or not parsed.netloc: + raise ValueError("base_url must be an absolute http(s) URL") + return value + + +class AuthMethod(StrictModel): + system_name: str = Field(min_length=1, max_length=80) + auth_type: Literal["none", "api_key", "bearer_token", "oauth2_client_credentials", "hmac_signature", "mtls", "basic", "oidc"] + secret_ref: str | None = Field( + default=None, + pattern=r"^[A-Z][A-Z0-9_]{2,127}$", + description="Caller-managed secret reference name only; never a raw secret value.", + ) + header_name: str | None = Field(default=None, max_length=80) + token_url: str | None = Field(default=None, max_length=2048) + scopes: list[str] = Field(default_factory=list, max_length=20) + + @model_validator(mode="after") + def _require_secret_ref_for_secret_auth(self) -> "AuthMethod": + if self.auth_type != "none" and not self.secret_ref: + raise ValueError("secret_ref is required for non-none auth methods") + _reject_raw_secret(self.secret_ref or "") + return self + + +class FieldMapping(StrictModel): + source: str = Field(min_length=1, max_length=160) + target: str = Field(min_length=1, max_length=160) + transform: str = Field(default="copy", max_length=500) + required: bool = True + pii: bool = False + + +class DataContract(StrictModel): + name: str = Field(min_length=1, max_length=120) + direction: Literal["inbound", "outbound", "bidirectional"] + content_type: str = Field(default="application/json", max_length=120) + schema_ref: str | None = Field(default=None, max_length=500) + fields: list[FieldMapping] = Field(default_factory=list, max_length=200) + example_payload: dict[str, Any] | None = None + + @field_validator("example_payload") + @classmethod + def _bounded_example(cls, value: dict[str, Any] | None) -> dict[str, Any] | None: + if value is not None: + _ensure_bounded_json(value, MAX_JSON_BYTES) + return value + + +class RateLimitSpec(StrictModel): + system_name: str = Field(min_length=1, max_length=80) + max_requests: int = Field(ge=1, le=1_000_000) + per_seconds: int = Field(ge=1, le=86_400) + burst: int | None = Field(default=None, ge=1, le=1_000_000) + retry_after_header: str | None = Field(default="Retry-After", max_length=80) + + +class EnvironmentSpec(StrictModel): + name: str = Field(min_length=1, max_length=64) + purpose: Literal["dev", "test", "sandbox", "staging", "prod"] + endpoint_refs: list[str] = Field(default_factory=list, max_length=20) + object_storage_ref: str | None = Field(default=None, max_length=128) + observability_ref: str | None = Field(default=None, max_length=128) + mutable: bool = False + + +class ComplianceConstraint(StrictModel): + name: str = Field(min_length=1, max_length=120) + category: Literal["pii", "phi", "pci", "soc2", "gdpr", "hipaa", "data_residency", "retention", "audit", "other"] + requirement: str = Field(min_length=1, max_length=2000) + + +class SuccessCriterion(StrictModel): + name: str = Field(min_length=1, max_length=120) + measurement: str = Field(min_length=1, max_length=500) + target: str = Field(min_length=1, max_length=200) + + +class IntegrationRequirements(StrictModel): + integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$") + title: str = Field(min_length=1, max_length=160) + systems: list[SystemSpec] = Field(min_length=1, max_length=12) + auth_methods: list[AuthMethod] = Field(default_factory=list, max_length=12) + data_contracts: list[DataContract] = Field(default_factory=list, max_length=20) + rate_limits: list[RateLimitSpec] = Field(default_factory=list, max_length=20) + environments: list[EnvironmentSpec] = Field(default_factory=list, max_length=12) + compliance_constraints: list[ComplianceConstraint] = Field(default_factory=list, max_length=30) + success_criteria: list[SuccessCriterion] = Field(min_length=1, max_length=20) + external_host_allowlist: list[str] = Field(default_factory=list, max_length=50) + repository_write_allowlist: list[str] = Field(default_factory=list, max_length=20) + notes: str | None = Field(default=None, max_length=4000) + dry_run: bool = True + + @field_validator("external_host_allowlist") + @classmethod + def _clean_hosts(cls, value: list[str]) -> list[str]: + return [_validate_hostname(host) for host in value] + + @field_validator("repository_write_allowlist") + @classmethod + def _clean_repos(cls, value: list[str]) -> list[str]: + out: list[str] = [] + for repo in value: + clean = repo.strip() + if not re.fullmatch(r"[A-Za-z0-9_.-]{1,80}/[A-Za-z0-9_.-]{1,120}", clean): + raise ValueError("repository allowlist entries must look like owner/repo") + out.append(clean) + return out + + +class DesignIntegrationRequest(StrictModel): + requirements: IntegrationRequirements + create_artifacts: bool = True + dry_run: bool = True + + +class DesignIntegrationResult(SkillStatus): + architecture_path: str | None = None + mapping_path: str | None = None + plan_digest: str + artifacts: list[ArtifactRecord] = Field(default_factory=list) + architecture_summary: str + field_mapping_count: int = Field(ge=0) + required_setup: list[str] = Field(default_factory=list) + + +class GenerateConnectorRequest(StrictModel): + requirements: IntegrationRequirements + connector_language: Literal["python", "typescript"] = "python" + include_tests: bool = True + dry_run: bool = True + external_host_allowlist: list[str] = Field(default_factory=list, max_length=50) + repository_write_allowlist: list[str] = Field(default_factory=list, max_length=20) + + @field_validator("external_host_allowlist") + @classmethod + def _clean_hosts(cls, value: list[str]) -> list[str]: + return [_validate_hostname(host) for host in value] + + +class GenerateConnectorResult(SkillStatus): + connector_path: str | None = None + test_path: str | None = None + plan_digest: str + artifacts: list[ArtifactRecord] = Field(default_factory=list) + generated_files: list[str] = Field(default_factory=list) + blocked_actions: list[str] = Field(default_factory=list) + + +class ContractFixture(StrictModel): + name: str = Field(min_length=1, max_length=120) + payload: dict[str, Any] + should_pass: bool = True + + @field_validator("payload") + @classmethod + def _bounded_payload(cls, value: dict[str, Any]) -> Any: + _ensure_bounded_json(value, MAX_JSON_BYTES) + return value + + +class ContractArtifact(StrictModel): + name: str = Field(min_length=1, max_length=120) + kind: Literal["json_schema", "openapi_3", "webhook"] + schema_doc: dict[str, Any] + fixtures: list[ContractFixture] = Field(default_factory=list, max_length=50) + signature_header: str | None = Field(default=None, max_length=120) + secret_ref: str | None = Field(default=None, pattern=r"^[A-Z][A-Z0-9_]{2,127}$") + + @field_validator("schema_doc") + @classmethod + def _bounded_schema(cls, value: dict[str, Any]) -> Any: + _ensure_bounded_json(value, MAX_JSON_BYTES) + return value + + +class ValidateContractRequest(StrictModel): + integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$") + contracts: list[ContractArtifact] = Field(min_length=1, max_length=20) + fail_on_drift: bool = True + dry_run: bool = True + + +class ContractCheck(StrictModel): + name: str + kind: str + status: Literal["pass", "fail", "warn"] + findings: list[str] = Field(default_factory=list) + fixture_results: list[dict[str, Any]] = Field(default_factory=list) + + +class ValidateContractResult(SkillStatus): + contract_report_path: str | None = None + artifacts: list[ArtifactRecord] = Field(default_factory=list) + checks: list[ContractCheck] = Field(default_factory=list) + drift_detected: bool = False + + +class EndpointTest(StrictModel): + name: str = Field(min_length=1, max_length=120) + method: Literal["GET", "POST", "PUT", "PATCH", "DELETE"] = "POST" + url: str = Field(max_length=2048) + expected_status: int = Field(default=200, ge=100, le=599) + request_json: dict[str, Any] | None = None + required_secret_refs: list[str] = Field(default_factory=list, max_length=10) + idempotency_key: str | None = Field(default=None, max_length=160) + + @field_validator("request_json") + @classmethod + def _bounded_request(cls, value: dict[str, Any] | None) -> dict[str, Any] | None: + if value is not None: + _ensure_bounded_json(value, MAX_JSON_BYTES) + return value + + @field_validator("required_secret_refs") + @classmethod + def _secret_refs_only(cls, value: list[str]) -> list[str]: + for ref in value: + if not re.fullmatch(r"[A-Z][A-Z0-9_]{2,127}", ref): + raise ValueError("secret refs must be uppercase reference names") + _reject_raw_secret(ref) + return value + + +class RunAcceptanceRequest(StrictModel): + integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$") + endpoint_tests: list[EndpointTest] = Field(default_factory=list, max_length=20) + contract_fixtures: list[ContractArtifact] = Field(default_factory=list, max_length=20) + external_host_allowlist: list[str] = Field(default_factory=list, max_length=50) + allow_network: bool = False + max_retries: int = Field(default=1, ge=0, le=3) + timeout_seconds: float = Field(default=8.0, ge=1.0, le=30.0) + dry_run: bool = True + + @field_validator("external_host_allowlist") + @classmethod + def _clean_hosts(cls, value: list[str]) -> list[str]: + return [_validate_hostname(host) for host in value] + + +class AcceptanceStepResult(StrictModel): + name: str + status: Literal["pass", "fail", "skipped", "blocked"] + attempts: int = Field(ge=0) + detail: str + idempotency_key: str | None = None + + +class RunAcceptanceResult(SkillStatus): + acceptance_report_path: str | None = None + artifacts: list[ArtifactRecord] = Field(default_factory=list) + step_results: list[AcceptanceStepResult] = Field(default_factory=list) + contract_results: list[ContractCheck] = Field(default_factory=list) + network_calls_made: int = Field(ge=0) + + +class HandoffSection(StrictModel): + heading: str = Field(min_length=1, max_length=120) + body: str = Field(min_length=1, max_length=8000) + + @field_validator("body") + @classmethod + def _redact_body(cls, value: str) -> str: + return redact_secrets(value) + + +class PrepareHandoffRequest(StrictModel): + integration_id: str = Field(pattern=r"^[a-z][a-z0-9-]{2,63}$") + title: str = Field(min_length=1, max_length=160) + rollout_steps: list[str] = Field(min_length=1, max_length=30) + rollback_steps: list[str] = Field(min_length=1, max_length=30) + monitoring_checks: list[str] = Field(min_length=1, max_length=30) + operator_notes: list[HandoffSection] = Field(default_factory=list, max_length=20) + delivery_actions: list[str] = Field(default_factory=list, max_length=20) + plan_digest: str | None = Field(default=None, max_length=128) + approval_phrase: str | None = Field(default=None, max_length=64) + approve_delivery: bool = False + dry_run: bool = True + + +class PrepareHandoffResult(SkillStatus): + handoff_path: str | None = None + plan_digest: str + artifacts: list[ArtifactRecord] = Field(default_factory=list) + approval_required: bool + approval_valid: bool + delivery_executed: bool = False + + +class CustomerIntegrationEngineerConfig(StrictModel): + default_dry_run: bool = True + + +_DESIGN_OUTPUT_SCHEMA = DesignIntegrationResult.model_json_schema() +_GENERATE_OUTPUT_SCHEMA = GenerateConnectorResult.model_json_schema() +_VALIDATE_OUTPUT_SCHEMA = ValidateContractResult.model_json_schema() +_ACCEPTANCE_OUTPUT_SCHEMA = RunAcceptanceResult.model_json_schema() +_HANDOFF_OUTPUT_SCHEMA = PrepareHandoffResult.model_json_schema() class CustomerIntegrationEngineer(A2AAgent[CustomerIntegrationEngineerConfig, NoAuth]): - name = EXPECTED_AGENT_NAME + name = "customer-integration-engineer" description = ( - "Validates the customer-integration-engineer release end to end while " - "preserving source unless a critical defect remains." + "Guides and validates customer integrations from requirements through a tested handoff while " + "keeping credentials and production changes under customer control." ) - version = "0.1.0" + version = "0.1.5" 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.", + notes="Deterministic no-LLM integration engineering; callers keep credentials and production changes under their control.", ) - resources = Resources(cpu="250m", memory="512Mi", max_runtime_seconds=300) + resources = Resources(cpu="500m", memory="512Mi", max_runtime_seconds=600) + egress = EgressPolicy(deny_internet_by_default=True) workspace_access = WorkspaceAccess.dynamic( - max_files=64, + max_files=256, allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), - require_reason=False, + require_reason=True, + deny_patterns=("**/.env", "**/*secret*", "**/*token*", "**/id_rsa", "**/.git/**"), + max_total_size_bytes=20 * 1024 * 1024, + ) + tools_used = ("pydantic", "jsonschema", "httpx") + consumer_setup = ConsumerSetup.from_fields( + ConsumerSetupField.config("GITEA_REPOSITORY", label="Gitea repository", description="Optional owner/repo target controlled by the caller.", required=False), + ConsumerSetupField.secret("GITEA_TOKEN", label="Gitea token", description="Optional caller-managed token reference; never returned or stored.", required=False), + ConsumerSetupField.config("API_BASE_URL", label="API base URL", description="Optional endpoint used only when allowlisted and dry_run=false.", required=False, input_type="url"), + ConsumerSetupField.secret("API_AUTH_TOKEN", label="API auth token", description="Optional caller-managed API credential reference.", required=False), + ConsumerSetupField.config("OBJECT_STORAGE_REF", label="Object storage reference", description="Caller-owned object storage config reference.", required=False), + ConsumerSetupField.config("OBSERVABILITY_REF", label="Observability reference", description="Caller-owned logs/metrics/traces reference.", required=False), + ConsumerSetupField.config("KUBERNETES_CONTEXT_REF", label="Kubernetes context reference", description="Optional caller-owned Kubernetes context reference.", required=False), + ConsumerSetupField.config("ARGOCD_APP_REF", label="Argo CD app reference", description="Optional caller-owned Argo CD app reference.", required=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="Capture requirements, produce architecture and field mapping artifacts", input_schema=_tool_input_schema(DesignIntegrationRequest), timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic") + async def design_integration(self, ctx: RunContext[NoAuth], request: DesignIntegrationRequest) -> DesignIntegrationResult: + req = request.requirements.model_copy( + update={"dry_run": _effective_dry_run(self, request.dry_run, request.requirements.dry_run)} + ) + audit = [_audit("design_started", req.integration_id, {"dry_run": req.dry_run})] + await _emit(ctx, audit[-1]) + warnings = _requirements_warnings(req) + required_setup = _required_setup(req) + digest = _plan_digest({"phase": "design", "requirements": req.model_dump(mode="json")}) + architecture = _render_architecture(req, digest, warnings, required_setup) + mapping = _build_mapping(req) + artifacts: list[ArtifactRecord] = [] + arch_path = mapping_path = None + if request.create_artifacts: + base = _integration_base(req.integration_id) + arch_path = f"{base}/architecture.md" + mapping_path = f"{base}/mapping.json" + artifacts.append(await _write_workspace_text(ctx, arch_path, architecture, "text/markdown")) + artifacts.append(await _write_workspace_json(ctx, mapping_path, mapping)) + audit.append(_audit("design_completed", req.integration_id, {"artifacts": [a.path for a in artifacts]})) + await _emit(ctx, audit[-1]) + return DesignIntegrationResult( + status="ok", + integration_id=req.integration_id, + dry_run=req.dry_run, + audit=audit, + warnings=warnings, + residual_risks=_residual_risks(req), + architecture_path=arch_path, + mapping_path=mapping_path, + plan_digest=digest, + artifacts=artifacts, + architecture_summary=f"{req.title}: {len(req.systems)} systems, {len(req.data_contracts)} contracts, {len(req.success_criteria)} success criteria.", + field_mapping_count=sum(len(c.fields) for c in req.data_contracts), + required_setup=required_setup, ) - @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="Generate a minimal workspace-contained connector scaffold with no deployment or repository mutation", input_schema=_tool_input_schema(GenerateConnectorRequest), timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic") + async def generate_connector(self, ctx: RunContext[NoAuth], request: GenerateConnectorRequest) -> GenerateConnectorResult: + req = request.requirements.model_copy( + update={"dry_run": _effective_dry_run(self, request.dry_run, request.requirements.dry_run)} + ) + audit = [_audit("connector_generation_started", req.integration_id, {"language": request.connector_language, "dry_run": req.dry_run})] + await _emit(ctx, audit[-1]) + warnings = _requirements_warnings(req) + blocked_actions: list[str] = [] + if request.repository_write_allowlist or req.repository_write_allowlist: + blocked_actions.append("repository writes are never performed; generated source is written only under outputs/integrations/{integration_id}/") + allowlist = sorted(set(req.external_host_allowlist + request.external_host_allowlist)) + digest = _plan_digest({"phase": "generate_connector", "requirements": req.model_dump(mode="json"), "language": request.connector_language}) + base = _integration_base(req.integration_id) + artifacts: list[ArtifactRecord] = [] + generated_files: list[str] = [] + if request.connector_language == "python": + connector_path = f"{base}/connector/python/connector.py" + test_path = f"{base}/connector/python/test_connector.py" if request.include_tests else None + source = _python_connector(req, allowlist, digest) + artifacts.append(await _write_workspace_text(ctx, connector_path, source, "text/x-python")) + generated_files.append(connector_path) + if test_path: + test_source = _python_connector_test(req) + artifacts.append(await _write_workspace_text(ctx, test_path, test_source, "text/x-python")) + generated_files.append(test_path) + else: + connector_path = f"{base}/connector/typescript/connector.ts" + test_path = f"{base}/connector/typescript/connector.test.ts" if request.include_tests else None + source = _typescript_connector(req, allowlist, digest) + artifacts.append(await _write_workspace_text(ctx, connector_path, source, "text/typescript")) + generated_files.append(connector_path) + if test_path: + test_source = _typescript_connector_test(req) + artifacts.append(await _write_workspace_text(ctx, test_path, test_source, "text/typescript")) + generated_files.append(test_path) + audit.append(_audit("connector_generation_completed", req.integration_id, {"files": generated_files, "blocked_actions": blocked_actions})) + await _emit(ctx, audit[-1]) + status: Literal["ok", "needs_setup", "blocked", "failed"] = "blocked" if blocked_actions and not req.dry_run else "ok" + return GenerateConnectorResult( + status=status, + integration_id=req.integration_id, + dry_run=req.dry_run, + audit=audit, + warnings=warnings, + residual_risks=_residual_risks(req), + connector_path=connector_path, + test_path=test_path, + plan_digest=digest, + artifacts=artifacts, + generated_files=generated_files, + blocked_actions=blocked_actions, ) - @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, + @a2a.tool(description="Validate OpenAPI, JSON Schema, and webhook contracts with deterministic fixtures", input_schema=_tool_input_schema(ValidateContractRequest), timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic") + async def validate_contract(self, ctx: RunContext[NoAuth], request: ValidateContractRequest) -> ValidateContractResult: + dry_run = _effective_dry_run(self, request.dry_run) + audit = [_audit("contract_validation_started", request.integration_id, {"contracts": len(request.contracts), "dry_run": dry_run})] + await _emit(ctx, audit[-1]) + checks = [_validate_one_contract(contract) for contract in request.contracts] + drift = any(check.status == "fail" for check in checks) + report = { + "integration_id": request.integration_id, + "status": "fail" if drift and request.fail_on_drift else "pass", + "dry_run": dry_run, + "generated_at": _now(), + "checks": [c.model_dump(mode="json") for c in checks], + "secret_handling": "secret_ref values only; raw credentials are rejected and redacted", } + path = f"{_integration_base(request.integration_id)}/contract-report.json" + artifact = await _write_workspace_json(ctx, path, report) + audit.append(_audit("contract_validation_completed", request.integration_id, {"drift_detected": drift, "report": path})) + await _emit(ctx, audit[-1]) + return ValidateContractResult( + status="failed" if drift and request.fail_on_drift else "ok", + integration_id=request.integration_id, + dry_run=dry_run, + audit=audit, + warnings=[] if not drift else ["contract drift detected; inspect contract-report.json before handoff"], + residual_risks=[] if not drift else ["fixtures are deterministic and bounded; live provider behavior still requires caller-controlled staging verification"], + contract_report_path=path, + artifacts=[artifact], + checks=checks, + drift_detected=drift, + ) + + @a2a.tool(description="Run dry-run or allowlisted acceptance checks using endpoint metadata and secret references", input_schema=_tool_input_schema(RunAcceptanceRequest), timeout_seconds=180, idempotent=True, max_retries=0, cost_class="deterministic") + async def run_acceptance(self, ctx: RunContext[NoAuth], request: RunAcceptanceRequest) -> RunAcceptanceResult: + dry_run = _effective_dry_run(self, request.dry_run) + live_network_block = _live_network_unavailable_reason(self) + audit = [_audit("acceptance_started", request.integration_id, {"endpoint_tests": len(request.endpoint_tests), "dry_run": dry_run, "allow_network": request.allow_network})] + await _emit(ctx, audit[-1]) + contract_results = [_validate_one_contract(contract) for contract in request.contract_fixtures] + step_results: list[AcceptanceStepResult] = [] + network_calls = 0 + warnings: list[str] = [] + for test in request.endpoint_tests: + idem = test.idempotency_key or _plan_digest({"integration_id": request.integration_id, "test": test.model_dump(mode="json")})[:32] + if dry_run or not request.allow_network: + validation = _validate_endpoint_metadata(test.url, request.external_host_allowlist) + if validation: + step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=validation, idempotency_key=idem)) + warnings.append(validation) + else: + step_results.append(AcceptanceStepResult(name=test.name, status="skipped", attempts=0, detail="dry_run/default safe mode: endpoint metadata was validated but no DNS lookup or external request was sent", idempotency_key=idem)) + continue + metadata_validation = _validate_endpoint_metadata(test.url, request.external_host_allowlist) + if metadata_validation: + step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=metadata_validation, idempotency_key=idem)) + warnings.append(metadata_validation) + continue + if live_network_block: + step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=live_network_block, idempotency_key=idem)) + warnings.append(live_network_block) + continue + validation = await _validate_endpoint_target(test.url, request.external_host_allowlist) + if validation: + step_results.append(AcceptanceStepResult(name=test.name, status="blocked", attempts=0, detail=validation, idempotency_key=idem)) + warnings.append(validation) + continue + result = await _call_endpoint_with_retries(test, request.max_retries, request.timeout_seconds, idem) + network_calls += result.attempts + step_results.append(result) + report = { + "integration_id": request.integration_id, + "dry_run": dry_run, + "allow_network": request.allow_network, + "network_calls_made": network_calls, + "generated_at": _now(), + "endpoint_results": [s.model_dump(mode="json") for s in step_results], + "contract_results": [c.model_dump(mode="json") for c in contract_results], + "safety": { + "ssrf_protection": "scheme, host allowlist, DNS/IP public-routability, and redirect revalidation enforced", + "secrets": "only secret references accepted; no raw values stored", + "payload_limits_bytes": MAX_JSON_BYTES, + "timeouts_seconds": request.timeout_seconds, + }, + } + path = f"{_integration_base(request.integration_id)}/acceptance-report.json" + artifact = await _write_workspace_json(ctx, path, report) + failed = any(s.status in {"fail", "blocked"} for s in step_results) or any(c.status == "fail" for c in contract_results) + audit.append(_audit("acceptance_completed", request.integration_id, {"report": path, "network_calls_made": network_calls, "failed_or_blocked": failed})) + await _emit(ctx, audit[-1]) + return RunAcceptanceResult( + status="blocked" if any(s.status == "blocked" for s in step_results) else "failed" if failed else "ok", + integration_id=request.integration_id, + dry_run=dry_run, + audit=audit, + warnings=sorted(set(warnings)), + residual_risks=["No production deployment, merge, or environment mutation is performed by this agent."], + acceptance_report_path=path, + artifacts=[artifact], + step_results=step_results, + contract_results=contract_results, + network_calls_made=network_calls, + ) + + @a2a.tool(description="Prepare rollout, rollback, monitoring, and operator handoff documentation with approval-gated delivery planning", input_schema=_tool_input_schema(PrepareHandoffRequest), timeout_seconds=120, idempotent=True, max_retries=0, cost_class="deterministic") + async def prepare_handoff(self, ctx: RunContext[NoAuth], request: PrepareHandoffRequest) -> PrepareHandoffResult: + dry_run = _effective_dry_run(self, request.dry_run) + audit = [_audit("handoff_started", request.integration_id, {"dry_run": dry_run, "delivery_actions": len(request.delivery_actions)})] + await _emit(ctx, audit[-1]) + plan_material = { + "integration_id": request.integration_id, + "rollout_steps": request.rollout_steps, + "rollback_steps": request.rollback_steps, + "monitoring_checks": request.monitoring_checks, + "delivery_actions": request.delivery_actions, + } + digest = _plan_digest(plan_material) + approval_required = bool(request.delivery_actions) + approval_valid = (not approval_required) or ( + request.approve_delivery + and request.approval_phrase == APPROVAL_PHRASE + and request.plan_digest == digest + ) + warnings: list[str] = [] + if approval_required and not approval_valid: + warnings.append(f"delivery actions require approval_phrase={APPROVAL_PHRASE!r} and plan_digest={digest}") + if approval_required and approval_valid: + warnings.append("approval validated for the listed plan, but this agent still does not deploy, merge, or modify customer systems") + handoff_request = request.model_copy(update={"dry_run": dry_run}) + handoff = _render_handoff(handoff_request, digest, approval_required, approval_valid) + path = f"{_integration_base(request.integration_id)}/handoff.md" + artifact = await _write_workspace_text(ctx, path, handoff, "text/markdown") + audit.append(_audit("handoff_completed", request.integration_id, {"handoff": path, "approval_required": approval_required, "approval_valid": approval_valid})) + await _emit(ctx, audit[-1]) + return PrepareHandoffResult( + status="blocked" if approval_required and not approval_valid else "ok", + integration_id=request.integration_id, + dry_run=dry_run, + audit=audit, + warnings=warnings, + residual_risks=["Operators must execute rollout or rollback in caller-controlled systems outside this agent."], + handoff_path=path, + plan_digest=digest, + artifacts=[artifact], + approval_required=approval_required, + approval_valid=approval_valid, + delivery_executed=False, + ) -def _runtime(card: dict[str, Any]) -> dict[str, Any]: - value = card.get("runtime") - return value if isinstance(value, dict) else {} +def _now() -> str: + return datetime.now(UTC).isoformat() -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 _effective_dry_run(agent: CustomerIntegrationEngineer, *values: bool) -> bool: + return bool(agent.config.default_dry_run or any(values)) -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 _live_network_unavailable_reason(agent: CustomerIntegrationEngineer) -> str | None: + if getattr(agent.egress, "deny_internet_by_default", False): + return ( + "live network checks are disabled by this runtime because egress denies internet by default; " + "acceptance can validate metadata in dry_run mode but cannot contact external endpoints" + ) + return None -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 _audit(action: str, integration_id: str, details: dict[str, Any] | None = None) -> AuditEntry: + return AuditEntry(ts=_now(), action=action, integration_id=integration_id, details=redact_obj(details 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 +async def _emit(ctx: RunContext[NoAuth], entry: AuditEntry) -> None: + await ctx.emit_event(AgentEvent(kind="audit", payload=entry.model_dump(mode="json"))) + await ctx.emit_progress(f"{entry.action}: {entry.integration_id}") -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 _integration_base(integration_id: str) -> str: + if not re.fullmatch(r"[a-z][a-z0-9-]{2,63}", integration_id): + raise ValueError("invalid integration_id") + return f"{OUTPUT_ROOT}/{integration_id}" -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] +def _safe_output_path(path: str) -> str: + clean = path.replace("\\", "/").strip("/") + if not clean.startswith(f"{OUTPUT_ROOT}/"): + raise ValueError("writes must stay under outputs/integrations/{integration_id}/") + if ".." in clean.split("/") or clean.startswith("/"): + raise ValueError("unsafe output path") + return clean + + +async def _write_workspace_text(ctx: RunContext[NoAuth], path: str, text: str, mime_type: str) -> ArtifactRecord: + clean = _safe_output_path(path) + data = redact_secrets(text).encode("utf-8") + if len(data) > MAX_TEXT_BYTES * 4: + raise ValueError("artifact exceeds bounded text size") + ws = ctx.workspace + if hasattr(ws, "is_writable_output") and ws.is_writable_output(clean): # type: ignore[attr-defined] + if hasattr(ws, "write_bytes"): + ws.write_bytes(clean, data) # type: ignore[attr-defined] + else: + view = await ws.open_view(purpose=f"write {clean}", hints=[clean], max_files=1, mode=WorkspaceMode.READ_WRITE_OVERLAY, reason="Persist integration engineering output under outputs/integrations") + await view.write(clean, data) else: - raw = [] - return [str(item.get("value") if isinstance(item, dict) else item).lower() for item in raw] + raise PermissionError("workspace grant does not allow writes under outputs/integrations") + artifact_name = clean.replace("/", "__") + ref = await ctx.write_artifact(artifact_name, data, mime_type) + await ctx.emit_artifact(ref) + return ArtifactRecord(path=clean, sha256=hashlib.sha256(data).hexdigest(), size_bytes=len(data), mime_type=mime_type) -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 +async def _write_workspace_json(ctx: RunContext[NoAuth], path: str, payload: dict[str, Any]) -> ArtifactRecord: + safe_payload = redact_obj(payload) + _ensure_bounded_json(safe_payload, MAX_JSON_BYTES) + return await _write_workspace_text(ctx, path, json.dumps(safe_payload, indent=2, sort_keys=True) + "\n", "application/json") -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 _ensure_bounded_json(value: Any, max_bytes: int) -> None: + data = json.dumps(value, default=str, sort_keys=True).encode("utf-8") + if len(data) > max_bytes: + raise ValueError(f"JSON payload exceeds {max_bytes} bytes") -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 _reject_raw_secret(value: str) -> None: + if not value: + return + if any(marker in value.lower() for marker in ("sk-", "ghp_", "xoxb-", "-----begin", "bearer ")): + raise ValueError("raw credential-looking values are not allowed; use a secret_ref") -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 redact_secrets(text: str) -> str: + redacted = text + patterns = [ + r"(?i)(authorization\s*[:=]\s*bearer\s+)[A-Za-z0-9._~+/=-]+", + r"(?i)(api[_-]?key\s*[:=]\s*)[A-Za-z0-9._~+/=-]{8,}", + r"(?i)(token\s*[:=]\s*)[A-Za-z0-9._~+/=-]{8,}", + r"(?i)(password\s*[:=]\s*)[^\s,;]+", + r"sk-[A-Za-z0-9]{12,}", + r"ghp_[A-Za-z0-9]{12,}", + ] + for pattern in patterns: + redacted = re.sub(pattern, _redact_match, redacted) + return redacted -def _secret_paths(value: Any, prefix: str = "$") -> list[str]: - found: list[str] = [] +def _redact_match(match: re.Match[str]) -> str: + return (match.group(1) if match.lastindex else "") + "[REDACTED]" + + +def _jsonschema_error_path(error: Any) -> Any: + return error.path + + +def redact_obj(value: Any) -> Any: if isinstance(value, dict): + out: dict[str, Any] = {} 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)) + if any(word in str(key).lower() for word in SECRET_WORDS): + out[key] = "[REDACTED_REF]" if isinstance(item, str) else "[REDACTED]" + else: + out[key] = redact_obj(item) + return out + if isinstance(value, list): + return [redact_obj(item) for item in value] + if isinstance(value, str): + return redact_secrets(value) + return value -def _check(name: str, ok: bool, detail: str) -> dict[str, Any]: - return {"name": name, "ok": bool(ok), "detail": detail} +def _plan_digest(payload: dict[str, Any]) -> str: + safe = redact_obj(payload) + encoded = json.dumps(safe, sort_keys=True, separators=(",", ":"), default=str).encode("utf-8") + return hashlib.sha256(encoded).hexdigest() -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 _requirements_warnings(req: IntegrationRequirements) -> list[str]: + warnings: list[str] = [] + if not req.auth_methods: + warnings.append("no auth methods supplied; confirm this is intentional") + if not req.data_contracts: + warnings.append("no data contracts supplied; mapping and validation will be limited") + if not req.rate_limits: + warnings.append("no rate limits supplied; connector will use conservative defaults") + if any(env.purpose == "prod" and env.mutable for env in req.environments): + warnings.append("production environment marked mutable; this agent will still not mutate it") + return warnings -def _is_safe_url(url: str) -> bool: +def _residual_risks(req: IntegrationRequirements) -> list[str]: + risks = [ + "Generated connector is minimal and must be reviewed by customer operators before production use.", + "Provider-specific authentication and observability wiring stay under caller-controlled setup references.", + ] + if req.dry_run: + risks.append("dry_run=true: acceptance evidence is deterministic and non-mutating, not proof of live provider behavior.") + return risks + + +def _required_setup(req: IntegrationRequirements) -> list[str]: + setup = {auth.secret_ref for auth in req.auth_methods if auth.secret_ref} + for env in req.environments: + for ref in (env.object_storage_ref, env.observability_ref): + if ref: + setup.add(ref) + return sorted(setup) + + +def _build_mapping(req: IntegrationRequirements) -> Any: + return { + "integration_id": req.integration_id, + "title": req.title, + "contracts": [ + { + "name": c.name, + "direction": c.direction, + "content_type": c.content_type, + "schema_ref": c.schema_ref, + "fields": [f.model_dump(mode="json") for f in c.fields], + } + for c in req.data_contracts + ], + "pii_fields": [f"{c.name}.{field.target}" for c in req.data_contracts for field in c.fields if field.pii], + } + + +def _render_architecture(req: IntegrationRequirements, digest: str, warnings: list[str], setup: list[str]) -> str: + systems = "\n".join(f"- **{s.name}** ({s.kind}, {s.environment}) {s.base_url or '(endpoint configured by caller)'}" for s in req.systems) + auth = "\n".join(f"- {a.system_name}: {a.auth_type}, secret_ref={a.secret_ref or 'none'}" for a in req.auth_methods) or "- No auth methods supplied." + contracts = "\n".join(f"- {c.name}: {c.direction}, {len(c.fields)} mapped fields, schema_ref={c.schema_ref or 'inline/fixture'}" for c in req.data_contracts) or "- No contracts supplied." + rates = "\n".join(f"- {r.system_name}: {r.max_requests}/{r.per_seconds}s burst={r.burst or r.max_requests}" for r in req.rate_limits) or "- Conservative default: bounded retries with idempotency keys." + compliance = "\n".join(f"- {c.category}: {c.name} — {c.requirement}" for c in req.compliance_constraints) or "- No explicit compliance constraints supplied." + success = "\n".join(f"- {s.name}: {s.measurement} target={s.target}" for s in req.success_criteria) + warning_text = "\n".join(f"- {w}" for w in warnings) or "- None" + setup_text = "\n".join(f"- {s}" for s in setup) or "- None" + return f"""# Integration Architecture: {req.title} + +Integration ID: `{req.integration_id}` +Plan digest: `{digest}` +Dry run: `{req.dry_run}` + +## Systems +{systems} + +## Authentication and Secret References +{auth} + +Raw credentials are neither accepted nor stored. Operators configure referenced secrets outside this agent. + +## Data Contracts and Field Mapping +{contracts} + +See `mapping.json` for exact field-level mapping. + +## Rate Limits and Retries +{rates} + +All mutating calls must carry idempotency keys. Retries are bounded and honor provider retry metadata. + +## Environments +{chr(10).join(f"- {e.name}: {e.purpose}, mutable={e.mutable}" for e in req.environments) or '- No environments supplied.'} + +## Compliance Constraints +{compliance} + +## Success Criteria +{success} + +## Required Caller Setup +{setup_text} + +## Safety Controls +- Default dry_run=true. +- External requests require explicit host allowlist, DNS/IP validation, redirect revalidation, payload limits, and timeouts. +- Repository writes beyond outputs, deployments, merges, and environment mutations are not performed. +- Delivery actions require `{APPROVAL_PHRASE}` bound to a plan digest and still remain operator-executed outside this agent. + +## Warnings +{warning_text} +""" + + +def _python_connector(req: IntegrationRequirements, allowlist: list[str], digest: str) -> str: + return f'''"""Minimal connector scaffold for {req.integration_id}. +Generated by customer-integration-engineer. Review before use. +Plan digest: {digest} +""" +from __future__ import annotations + +import hashlib +import ipaddress +import json +import socket +from urllib.parse import urlparse + +import httpx + +ALLOWED_HOSTS = {allowlist!r} +MAX_BYTES = {MAX_JSON_BYTES} +TIMEOUT_SECONDS = 8.0 + + +def validate_url(url: str) -> str: parsed = urlparse(url) - return parsed.scheme in {"https", "http"} and bool(parsed.netloc) + if parsed.scheme != "https" or not parsed.hostname: + raise ValueError("connector only permits absolute https URLs") + host = parsed.hostname.lower() + if host not in ALLOWED_HOSTS: + raise ValueError(f"host {{host!r}} is not allowlisted") + for info in socket.getaddrinfo(host, parsed.port or 443, type=socket.SOCK_STREAM): + ip = ipaddress.ip_address(info[4][0]) + if ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified: + raise ValueError(f"host {{host!r}} resolved to unsafe address") + return url + + +def idempotency_key(operation: str, payload: dict) -> str: + material = json.dumps({{"operation": operation, "payload": payload}}, sort_keys=True).encode("utf-8") + return hashlib.sha256(material).hexdigest() + + +async def send_json(url: str, payload: dict, *, secret_ref: str | None = None, dry_run: bool = True) -> dict: + body = json.dumps(payload).encode("utf-8") + if len(body) > MAX_BYTES: + raise ValueError("payload too large") + safe_url = validate_url(url) + key = idempotency_key("send_json", payload) + if dry_run: + return {{"status": "skipped", "reason": "dry_run", "idempotency_key": key, "url": safe_url}} + # Resolve secret_ref outside this scaffold using your approved secret manager. + headers = {{"Idempotency-Key": key, "Content-Type": "application/json"}} + async with httpx.AsyncClient(timeout=TIMEOUT_SECONDS, follow_redirects=False) as client: + response = await client.post(safe_url, content=body, headers=headers) + return {{"status_code": response.status_code, "idempotency_key": key}} +''' + + +def _python_connector_test(req: IntegrationRequirements) -> str: + return f'''from connector import idempotency_key + + +def test_idempotency_key_stable(): + payload = {{"integration_id": "{req.integration_id}", "event": "synthetic"}} + assert idempotency_key("send_json", payload) == idempotency_key("send_json", payload) +''' + + +def _typescript_connector(req: IntegrationRequirements, allowlist: list[str], digest: str) -> str: + hosts = json.dumps(allowlist) + return f'''// Minimal connector scaffold for {req.integration_id}. Plan digest: {digest} +const ALLOWED_HOSTS = new Set({hosts}); +const MAX_BYTES = {MAX_JSON_BYTES}; + +export function validateUrl(raw: string): URL {{ + const url = new URL(raw); + if (url.protocol !== "https:") throw new Error("connector only permits https URLs"); + if (!ALLOWED_HOSTS.has(url.hostname.toLowerCase())) throw new Error("host is not allowlisted"); + return url; +}} + +export async function sendJson(rawUrl: string, payload: unknown, options: {{ dryRun?: boolean, secretRef?: string }} = {{}}) {{ + const url = validateUrl(rawUrl); + const body = JSON.stringify(payload); + if (new TextEncoder().encode(body).length > MAX_BYTES) throw new Error("payload too large"); + const idem = await crypto.subtle.digest("SHA-256", new TextEncoder().encode(body)); + const idempotencyKey = Array.from(new Uint8Array(idem)).map(b => b.toString(16).padStart(2, "0")).join(""); + if (options.dryRun !== false) return {{ status: "skipped", reason: "dry_run", idempotencyKey, url: url.toString() }}; + const response = await fetch(url, {{ method: "POST", body, headers: {{ "Content-Type": "application/json", "Idempotency-Key": idempotencyKey }} }}); + return {{ statusCode: response.status, idempotencyKey }}; +}} +''' + + +def _typescript_connector_test(req: IntegrationRequirements) -> str: + return f'''import {{ validateUrl }} from "./connector"; + +test("allowlist rejects unknown host for {req.integration_id}", () => {{ + expect(() => validateUrl("https://example.invalid/webhook")).toThrow(); +}}); +''' + + +def _validate_one_contract(contract: ContractArtifact) -> ContractCheck: + findings: list[str] = [] + fixture_results: list[dict[str, Any]] = [] + try: + if contract.kind == "json_schema": + from jsonschema import Draft202012Validator + Draft202012Validator.check_schema(contract.schema_doc) + validator = Draft202012Validator(contract.schema_doc) + for fixture in contract.fixtures: + errors = sorted(validator.iter_errors(fixture.payload), key=_jsonschema_error_path) + passed = not errors + expected = fixture.should_pass + fixture_results.append({"fixture": fixture.name, "passed": passed, "expected": expected, "errors": [e.message for e in errors[:5]]}) + if passed != expected: + findings.append(f"fixture {fixture.name!r} expected should_pass={expected} but got {passed}") + elif contract.kind == "openapi_3": + version = str(contract.schema_doc.get("openapi", "")) + if not version.startswith("3."): + findings.append("OpenAPI document must declare openapi: 3.x") + if not isinstance(contract.schema_doc.get("paths"), dict) or not contract.schema_doc.get("paths"): + findings.append("OpenAPI document must include non-empty paths") + if "servers" in contract.schema_doc and not isinstance(contract.schema_doc["servers"], list): + findings.append("OpenAPI servers must be a list") + elif contract.kind == "webhook": + if not contract.signature_header: + findings.append("webhook contract requires signature_header") + if not contract.secret_ref: + findings.append("webhook contract requires secret_ref, not a raw secret") + event_type = contract.schema_doc.get("event_type") or contract.schema_doc.get("type") + if not event_type: + findings.append("webhook schema_doc should include event_type or type") + for fixture in contract.fixtures: + has_event = "event" in fixture.payload or "type" in fixture.payload + passed = bool(has_event) + fixture_results.append({"fixture": fixture.name, "passed": passed, "expected": fixture.should_pass, "errors": [] if passed else ["missing event/type"]}) + if passed != fixture.should_pass: + findings.append(f"fixture {fixture.name!r} webhook expectation mismatch") + except Exception as exc: # noqa: BLE001 + findings.append(f"contract validation error: {type(exc).__name__}: {exc}") + status: Literal["pass", "fail", "warn"] = "fail" if findings else "pass" + return ContractCheck(name=contract.name, kind=contract.kind, status=status, findings=findings, fixture_results=fixture_results) + + +def _validate_hostname(host: str) -> str: + clean = host.strip().lower().rstrip(".") + if not clean or len(clean) > 253: + raise ValueError("invalid hostname") + if clean in {"localhost", "metadata.google.internal"} or clean.endswith(".local") or clean.endswith(".internal"): + raise ValueError("internal hostnames are not allowed") + try: + ip = ipaddress.ip_address(clean) + except ValueError: + if not re.fullmatch(r"[a-z0-9.-]+", clean) or ".." in clean: + raise ValueError("hostname contains invalid characters") + return clean + if _unsafe_ip(ip): + raise ValueError("private, loopback, link-local, multicast, reserved, or unspecified IPs are not allowed") + return clean + + +def _unsafe_ip(ip: ipaddress._BaseAddress) -> bool: + return bool(ip.is_private or ip.is_loopback or ip.is_link_local or ip.is_multicast or ip.is_reserved or ip.is_unspecified) + + +def _validate_endpoint_metadata(url: str, allowlist: list[str]) -> str | None: + parsed = urlparse(url) + if parsed.scheme != "https" or not parsed.hostname: + return "endpoint URL must be absolute https with a hostname" + host = parsed.hostname.lower().rstrip(".") + if host not in set(allowlist): + return f"host {host!r} is not in external_host_allowlist" + try: + _validate_hostname(host) + except ValueError as exc: + return str(exc) + return None + + +async def _validate_endpoint_target(url: str, allowlist: list[str]) -> str | None: + metadata_error = _validate_endpoint_metadata(url, allowlist) + if metadata_error: + return metadata_error + parsed = urlparse(url) + host = parsed.hostname.lower().rstrip(".") # type: ignore[union-attr] + try: + infos = await asyncio.to_thread(socket.getaddrinfo, host, parsed.port or 443, type=socket.SOCK_STREAM) + except socket.gaierror as exc: + return f"DNS resolution failed for {host!r}: {exc}" + for info in infos: + ip = ipaddress.ip_address(info[4][0]) + if _unsafe_ip(ip): + return f"host {host!r} resolves to unsafe IP {ip}" + return None + + +async def _call_endpoint_with_retries(test: EndpointTest, max_retries: int, timeout_seconds: float, idem: str) -> AcceptanceStepResult: + import httpx + + attempts = 0 + last_detail = "not attempted" + headers = {"Idempotency-Key": idem, "Content-Type": "application/json"} + for attempt in range(max_retries + 1): + attempts = attempt + 1 + try: + async with httpx.AsyncClient(timeout=timeout_seconds, follow_redirects=False) as client: + response = await client.request(test.method, test.url, json=test.request_json, headers=headers) + if 300 <= response.status_code < 400: + location = response.headers.get("location", "") + return AcceptanceStepResult(name=test.name, status="blocked", attempts=attempts, detail=f"redirect blocked pending revalidation: {location[:200]}", idempotency_key=idem) + if response.status_code == test.expected_status: + return AcceptanceStepResult(name=test.name, status="pass", attempts=attempts, detail=f"received expected status {response.status_code}", idempotency_key=idem) + last_detail = f"expected {test.expected_status}, got {response.status_code}" + if response.status_code not in {408, 409, 425, 429, 500, 502, 503, 504}: + break + await asyncio.sleep(min(2.0, 0.25 * (attempt + 1))) + except Exception as exc: # noqa: BLE001 + last_detail = f"{type(exc).__name__}: {exc}" + await asyncio.sleep(min(2.0, 0.25 * (attempt + 1))) + return AcceptanceStepResult(name=test.name, status="fail", attempts=attempts, detail=redact_secrets(last_detail), idempotency_key=idem) + + +def _render_handoff(request: PrepareHandoffRequest, digest: str, approval_required: bool, approval_valid: bool) -> str: + operator_sections = "\n\n".join(f"## {s.heading}\n{s.body}" for s in request.operator_notes) or "## Operator Notes\nNo additional operator notes supplied." + delivery = "\n".join(f"- {item}" for item in request.delivery_actions) or "- No delivery actions requested." + return f"""# Integration Handoff: {request.title} + +Integration ID: `{request.integration_id}` +Plan digest: `{digest}` +Dry run: `{request.dry_run}` + +## Rollout Plan +{chr(10).join(f"{i+1}. {redact_secrets(step)}" for i, step in enumerate(request.rollout_steps))} + +## Rollback Plan +{chr(10).join(f"{i+1}. {redact_secrets(step)}" for i, step in enumerate(request.rollback_steps))} + +## Monitoring and Alerting Checks +{chr(10).join(f"- {redact_secrets(check)}" for check in request.monitoring_checks)} + +{operator_sections} + +## Delivery Actions Requiring Operator Control +{delivery} + +Approval required: `{approval_required}` +Approval valid for this exact digest: `{approval_valid}` + +This agent does not deploy, merge, modify customer systems, or execute delivery actions. Operators must perform approved delivery steps in their own controlled systems. + +## Evidence Checklist +- Review `architecture.md` and `mapping.json`. +- Review connector source under `connector/`. +- Review `contract-report.json`. +- Review `acceptance-report.json`. +- Confirm all secret references are configured in caller-owned systems. +""" diff --git a/requirements.txt b/requirements.txt index 3a86bcd..21a2e22 100644 --- a/requirements.txt +++ b/requirements.txt @@ -1 +1,5 @@ # a2a-pack is auto-installed by the deploy build. +jsonschema>=4.22 +httpx>=0.27 +pytest>=8.0 +pytest-asyncio>=0.23 diff --git a/tests/test_agent.py b/tests/test_agent.py deleted file mode 100644 index ce65e6d..0000000 --- a/tests/test_agent.py +++ /dev/null @@ -1,64 +0,0 @@ -import pytest - -from a2a_pack import NoAuth -from a2a_pack.context import LocalRunContext - -from agent import CustomerIntegrationEngineer, SmokeCallInput - - -@pytest.mark.asyncio -async def test_source_policy_rejects_code_editor_agent(): - agent = CustomerIntegrationEngineer() - ctx = LocalRunContext(auth=NoAuth()) - - result = await agent.review_source_change_policy( - ctx, - mutated_source=False, - critical_defect_remaining=False, - used_code_editor_agent=True, - improvement_iterations=0, - ) - - assert result["ok"] is False - failed = {check["name"] for check in result["checks"] if not check["ok"]} - assert "no_code_editor_agent" in failed - - -@pytest.mark.asyncio -async def test_smoke_call_rejects_secret_like_result(): - agent = CustomerIntegrationEngineer() - ctx = LocalRunContext(auth=NoAuth()) - - result = await agent.validate_smoke_call( - ctx, - SmokeCallInput( - skill_name="validate_live_card", - input_args={"bounded": True}, - result={"ok": True, "api_key": "sk-testsecret1234567890"}, - status_code=200, - used_workspace_delegation=True, - delegated_workspace_mode="read_only", - ), - ) - - assert result["ok"] is False - failed = {check["name"] for check in result["checks"] if not check["ok"]} - assert "no_secrets_returned" in failed - assert "$.api_key" in result["secret_like_paths"] - - -@pytest.mark.asyncio -async def test_final_report_passes_with_clean_validations(): - agent = CustomerIntegrationEngineer() - ctx = LocalRunContext(auth=NoAuth()) - - result = await agent.final_release_report( - ctx, - validation_results=[{"checks": [{"name": "sample", "ok": True}]}], - reviewer_findings=[], - warnings_to_accept=[], - ) - - assert result["ok"] is True - assert result["critical_findings"] == [] - assert result["secrets_included"] is False diff --git a/tests/test_customer_integration_engineer.py b/tests/test_customer_integration_engineer.py index 12551b7..2f1c092 100644 --- a/tests/test_customer_integration_engineer.py +++ b/tests/test_customer_integration_engineer.py @@ -1,5 +1,234 @@ -"""Compatibility placeholder for the current deterministic release validator.""" +from __future__ import annotations + +import pytest +from a2a_pack import LocalRunContext, LocalWorkspaceClient, NoAuth, WorkspaceAccess, WorkspaceMode + +from agent import ( + APPROVAL_PHRASE, + AuthMethod, + ContractArtifact, + ContractFixture, + CustomerIntegrationEngineer, + DesignIntegrationRequest, + EndpointTest, + FieldMapping, + GenerateConnectorRequest, + IntegrationRequirements, + PrepareHandoffRequest, + RunAcceptanceRequest, + SuccessCriterion, + SystemSpec, + ValidateContractRequest, + _plan_digest, + redact_obj, +) -def test_current_validator_replaces_stale_contract(): - assert True +def workspace(): + return LocalWorkspaceClient( + {}, + access=WorkspaceAccess.dynamic( + max_files=256, + allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), + require_reason=False, + ), + ) + + +def ctx(ws=None): + return LocalRunContext(auth=NoAuth(), workspace=ws or workspace(), task_id="test") + + +def requirements() -> IntegrationRequirements: + return IntegrationRequirements( + integration_id="webhook-demo", + title="Synthetic webhook integration", + systems=[SystemSpec(name="Source", kind="webhook", base_url="https://hooks.example.com/events")], + auth_methods=[AuthMethod(system_name="Source", auth_type="hmac_signature", secret_ref="WEBHOOK_SECRET", header_name="X-Signature")], + data_contracts=[ + { + "name": "event", + "direction": "inbound", + "fields": [FieldMapping(source="event.id", target="id"), FieldMapping(source="event.email", target="email", pii=True)], + "example_payload": {"event": {"id": "evt_1", "email": "a@example.com"}}, + } + ], + rate_limits=[{"system_name": "Source", "max_requests": 60, "per_seconds": 60}], + environments=[{"name": "sandbox", "purpose": "sandbox", "endpoint_refs": ["API_BASE_URL"], "observability_ref": "OBSERVABILITY_REF"}], + compliance_constraints=[{"name": "PII minimization", "category": "pii", "requirement": "Mask email in logs."}], + success_criteria=[SuccessCriterion(name="Webhook accepted", measurement="2xx response", target="100% for fixtures")], + external_host_allowlist=["hooks.example.com"], + dry_run=True, + ) + + +@pytest.mark.asyncio +async def test_synthetic_webhook_outputs_expected_files(): + agent = CustomerIntegrationEngineer() + ws = workspace() + run_ctx = ctx(ws) + req = requirements() + + design = await agent.design_integration(run_ctx, DesignIntegrationRequest(requirements=req)) + connector = await agent.generate_connector(run_ctx, GenerateConnectorRequest(requirements=req)) + contract = await agent.validate_contract( + run_ctx, + ValidateContractRequest( + integration_id=req.integration_id, + contracts=[ + ContractArtifact( + name="webhook", + kind="webhook", + schema_doc={"event_type": "customer.created"}, + signature_header="X-Signature", + secret_ref="WEBHOOK_SECRET", + fixtures=[ContractFixture(name="ok", payload={"event": "customer.created"})], + ) + ], + ), + ) + acceptance = await agent.run_acceptance( + run_ctx, + RunAcceptanceRequest( + integration_id=req.integration_id, + endpoint_tests=[EndpointTest(name="dry-run webhook", url="https://hooks.example.com/events")], + contract_fixtures=[], + external_host_allowlist=["hooks.example.com"], + dry_run=True, + ), + ) + handoff = await agent.prepare_handoff( + run_ctx, + PrepareHandoffRequest( + integration_id=req.integration_id, + title=req.title, + rollout_steps=["Enable webhook in sandbox"], + rollback_steps=["Disable webhook"], + monitoring_checks=["Alert on 5xx rate > 1%"], + ), + ) + + expected = { + "outputs/integrations/webhook-demo/architecture.md", + "outputs/integrations/webhook-demo/mapping.json", + "outputs/integrations/webhook-demo/connector/python/connector.py", + "outputs/integrations/webhook-demo/connector/python/test_connector.py", + "outputs/integrations/webhook-demo/contract-report.json", + "outputs/integrations/webhook-demo/acceptance-report.json", + "outputs/integrations/webhook-demo/handoff.md", + } + assert expected.issubset(set(ws.iter_paths())) + assert design.status == connector.status == contract.status == acceptance.status == handoff.status == "ok" + assert acceptance.network_calls_made == 0 + assert "outputs__integrations__webhook-demo__architecture.md" in run_ctx.artifacts + assert "architecture.md" not in run_ctx.artifacts + + +def test_auth_redaction_and_raw_secret_rejection(): + redacted = redact_obj({"Authorization": "Bearer super-secret-token", "nested": {"api_key": "sk-test123456789"}}) + assert "super-secret-token" not in str(redacted) + assert "sk-test" not in str(redacted) + with pytest.raises(ValueError): + AuthMethod(system_name="API", auth_type="bearer_token", secret_ref="sk-live-raw-secret") + + +@pytest.mark.asyncio +async def test_schema_drift_fails_json_schema_fixture(): + agent = CustomerIntegrationEngineer() + result = await agent.validate_contract( + ctx(), + ValidateContractRequest( + integration_id="schema-demo", + contracts=[ + ContractArtifact( + name="customer", + kind="json_schema", + schema_doc={"type": "object", "required": ["id"], "properties": {"id": {"type": "string"}}, "additionalProperties": False}, + fixtures=[ContractFixture(name="missing id", payload={"email": "a@example.com"}, should_pass=True)], + ) + ], + ), + ) + assert result.status == "failed" + assert result.drift_detected is True + assert result.checks[0].status == "fail" + + +@pytest.mark.asyncio +async def test_ssrf_blocks_private_and_unallowlisted_hosts(): + agent = CustomerIntegrationEngineer() + result = await agent.run_acceptance( + ctx(), + RunAcceptanceRequest( + integration_id="ssrf-demo", + endpoint_tests=[ + EndpointTest(name="private", url="https://127.0.0.1/hook"), + EndpointTest(name="unlisted", url="https://evil.example/hook"), + ], + external_host_allowlist=["api.example.com"], + dry_run=True, + ), + ) + assert result.status == "blocked" + assert all(step.status == "blocked" for step in result.step_results) + + +@pytest.mark.asyncio +async def test_approval_bound_to_plan_digest(): + agent = CustomerIntegrationEngineer() + request = PrepareHandoffRequest( + integration_id="approval-demo", + title="Approval demo", + rollout_steps=["create release branch"], + rollback_steps=["revert release branch"], + monitoring_checks=["check error rate"], + delivery_actions=["open pull request"], + approval_phrase=APPROVAL_PHRASE, + approve_delivery=True, + plan_digest="wrong", + ) + blocked = await agent.prepare_handoff(ctx(), request) + assert blocked.status == "blocked" + digest = _plan_digest({ + "integration_id": request.integration_id, + "rollout_steps": request.rollout_steps, + "rollback_steps": request.rollback_steps, + "monitoring_checks": request.monitoring_checks, + "delivery_actions": request.delivery_actions, + }) + approved = await agent.prepare_handoff(ctx(), request.model_copy(update={"plan_digest": digest})) + assert approved.status == "ok" + assert approved.delivery_executed is False + + +@pytest.mark.asyncio +async def test_idempotency_stable_in_acceptance_dry_run(): + agent = CustomerIntegrationEngineer() + request = RunAcceptanceRequest( + integration_id="idempotency-demo", + endpoint_tests=[EndpointTest(name="hook", url="https://hooks.example.com/events", request_json={"x": 1})], + external_host_allowlist=["hooks.example.com"], + dry_run=True, + ) + first = await agent.run_acceptance(ctx(), request) + second = await agent.run_acceptance(ctx(), request) + assert first.step_results[0].idempotency_key == second.step_results[0].idempotency_key + + +@pytest.mark.asyncio +async def test_retry_metadata_without_network_by_default(): + agent = CustomerIntegrationEngineer() + result = await agent.run_acceptance( + ctx(), + RunAcceptanceRequest( + integration_id="retry-demo", + endpoint_tests=[EndpointTest(name="safe", url="https://hooks.example.com/events")], + external_host_allowlist=["hooks.example.com"], + max_retries=3, + allow_network=False, + dry_run=True, + ), + ) + assert result.step_results[0].status == "skipped" + assert result.step_results[0].attempts == 0 + assert result.network_calls_made == 0 diff --git a/tests/test_runtime_safety.py b/tests/test_runtime_safety.py index e495027..f02d473 100644 --- a/tests/test_runtime_safety.py +++ b/tests/test_runtime_safety.py @@ -1,12 +1,60 @@ -"""Runtime safety checks for the current deterministic release validator.""" +from __future__ import annotations -from agent import CustomerIntegrationEngineer +import pytest +from a2a_pack import LocalRunContext, LocalWorkspaceClient, NoAuth, WorkspaceAccess, WorkspaceMode + +from agent import CustomerIntegrationEngineer, EndpointTest, RunAcceptanceRequest -def test_no_llm_required_by_pricing(): - assert CustomerIntegrationEngineer.pricing.caller_pays_llm is False +def workspace(): + return LocalWorkspaceClient( + {}, + access=WorkspaceAccess.dynamic( + max_files=256, + allowed_modes=(WorkspaceMode.READ_ONLY, WorkspaceMode.READ_WRITE_OVERLAY), + require_reason=False, + ), + ) -def test_workspace_modes_declared(): - modes = {mode.value for mode in CustomerIntegrationEngineer.workspace_access.allowed_modes} - assert {"read_only", "read_write_overlay"} <= modes +def ctx(): + return LocalRunContext(auth=NoAuth(), workspace=workspace(), task_id="test") + + +@pytest.mark.asyncio +async def test_default_dry_run_forces_skip_even_if_request_allows_network(): + agent = CustomerIntegrationEngineer(config={"default_dry_run": True}) + result = await agent.run_acceptance( + ctx(), + RunAcceptanceRequest( + integration_id="forced-dry-run", + endpoint_tests=[EndpointTest(name="hook", url="https://hooks.example.com/events")], + external_host_allowlist=["hooks.example.com"], + allow_network=True, + dry_run=False, + ), + ) + assert result.status == "ok" + assert result.dry_run is True + assert result.network_calls_made == 0 + assert result.step_results[0].status == "skipped" + + +@pytest.mark.asyncio +async def test_live_network_blocked_by_runtime_egress_policy(): + agent = CustomerIntegrationEngineer(config={"default_dry_run": False}) + result = await agent.run_acceptance( + ctx(), + RunAcceptanceRequest( + integration_id="live-blocked", + endpoint_tests=[EndpointTest(name="hook", url="https://hooks.example.com/events")], + external_host_allowlist=["hooks.example.com"], + allow_network=True, + dry_run=False, + ), + ) + assert result.status == "blocked" + assert result.dry_run is False + assert result.network_calls_made == 0 + assert result.step_results[0].status == "blocked" + assert "egress denies internet by default" in result.step_results[0].detail