bobseares-2/customer-integration-engineer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

restore: customer integration agent 0.1.5

Browse Source
This commit is contained in:
bobseares@gmail.com
2026-07-13 09:29:26 -03:00
parent 1e2bee5df9
commit d9313ef6e8
7 changed files with 1381 additions and 385 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
.a2a-builder-state.json
View File

@@ -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"
}

6
a2a.yaml
View File

@@ -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

1387
agent.py
View File

File diff suppressed because it is too large Load Diff

4
requirements.txt
View File

@@ -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

64
tests/test_agent.py
View File

@@ -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

235
tests/test_customer_integration_engineer.py
View File

@@ -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

62
tests/test_runtime_safety.py
View File

@@ -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