initial: test-helper agent for e2e platform feature coverage

Two skills:
  - echo(text): trivial sub-second response. Used by approval-mode tests
    that need a fast handoff target.
  - try_scope_expansion(reason, read_patterns, ttl_seconds, mode):
    deliberately calls ctx.request_scope() with caller-supplied args so
    e2e tests can exercise the orchestrator's scope-negotiation flow
    end-to-end without inventing a real workload that happens to need
    extra files.

Standard A2A scaffold: Dockerfile FROM a2a-pack-base, /invoke/{skill} +
/.well-known/agent-card served by the SDK's serve/asgi.py adapter, deploy/
manifests for k8s, Gitea Actions workflow that builds + bumps the image
SHA.

Public, low resources (50m CPU / 128Mi mem). Not exposed via the public
ingress in any meaningful way — only the in-cluster Service is consulted
by the orchestrator's call_agent.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>

.gitea/workflows/build.yml

@@ -0,0 +1,35 @@
+name: build
+on:
+  push:
+    branches: [main]
+    paths-ignore:
+      - 'deploy/**'
+
+jobs:
+  build:
+    runs-on: ubuntu-latest
+    steps:
+      - uses: actions/checkout@v4
+        with:
+          fetch-depth: 0
+
+      - name: build image
+        run: |
+          IMG=registry.a2acloud.io/agents/test-helper
+          docker build -t "$IMG:$GITHUB_SHA" -t "$IMG:latest" .
+          docker push "$IMG:$GITHUB_SHA"
+          docker push "$IMG:latest"
+
+      - name: bump deploy manifest
+        run: |
+          IMG=registry.a2acloud.io/agents/test-helper
+          sed -i "s|image: $IMG:.*|image: $IMG:$GITHUB_SHA|" deploy/20-deployment.yaml
+          git config user.email "ci@a2a.local"
+          git config user.name "ci"
+          git add deploy/20-deployment.yaml
+          if git diff --staged --quiet; then
+            echo "no manifest changes"
+          else
+            git commit -m "ci: bump image to $GITHUB_SHA"
+            git push "http://gitea_admin:${{ secrets.ADMIN_PW }}@gitea-http.gitea.svc.cluster.local:3000/gitea_admin/test-helper.git" HEAD:main
+          fi

.gitignore

@@ -0,0 +1,5 @@
+.env
+__pycache__/
+*.pyc
+.pytest_cache/
+.venv/

Dockerfile

@@ -0,0 +1,14 @@
+FROM registry.a2acloud.io/a2a/a2a-pack-base:latest
+
+WORKDIR /app
+
+COPY requirements.txt .
+RUN pip install --no-cache-dir -r requirements.txt
+
+COPY . .
+
+ENV A2A_ENTRYPOINT=agent:TestHelperAgent
+ENV PORT=8000
+EXPOSE 8000
+
+CMD a2a run --entrypoint "$A2A_ENTRYPOINT" --host 0.0.0.0 --port 8000

a2a.yaml

@@ -0,0 +1,6 @@
+name: test-helper
+version: 0.1.0
+entrypoint: agent:TestHelperAgent
+description: Platform feature exerciser for e2e tests.
+expose:
+  public: true

agent.py

@@ -0,0 +1,100 @@
+"""test-helper agent — exercises platform features for end-to-end tests.
+
+Not a user-facing agent. Lives on the platform so e2e tests can target
+predictable, fast behaviour without bringing the chart pipeline along
+for every assertion.
+
+Skills:
+  - ``echo``: trivial fast response. For approval-mode tests that just
+    need *some* handoff to pause on.
+  - ``try_scope_expansion``: deliberately calls ``ctx.request_scope()``
+    so the orchestrator's scope-negotiation path can be exercised.
+"""
+from __future__ import annotations
+
+import json
+
+from pydantic import BaseModel
+
+from a2a_pack import A2AAgent, NoAuth, RunContext, skill
+from a2a_pack.context import ScopeDenied
+
+
+class TestHelperConfig(BaseModel):
+    pass
+
+
+class TestHelperAgent(A2AAgent[TestHelperConfig, NoAuth]):
+    name = "test-helper"
+    description = (
+        "Platform feature exerciser for e2e tests: cheap echo + a deliberate "
+        "ctx.request_scope() caller."
+    )
+    version = "0.1.0"
+
+    config_model = TestHelperConfig
+    auth_model = NoAuth
+    tools_used = ("e2e",)
+
+    @skill(
+        description=(
+            "Echo the text back. Trivial, sub-second response — used by "
+            "tests that need to exercise the handoff/approval mechanics "
+            "without paying for chart rendering."
+        ),
+        tags=["e2e", "echo"],
+    )
+    async def echo(self, ctx: RunContext[NoAuth], text: str = "ping") -> dict:
+        await ctx.emit_progress(f"echoing {len(text)} chars")
+        bucket = getattr(ctx.workspace, "bucket", None)
+        return {"ok": True, "echoed": text, "bucket": bucket}
+
+    @skill(
+        description=(
+            "Call ctx.request_scope() with the caller-supplied read patterns "
+            "+ ttl, then report the platform's response. Lets tests verify "
+            "the orchestrator's scope-negotiation flow end-to-end without "
+            "needing a real workload that happens to need extra files."
+        ),
+        tags=["e2e", "scope-request"],
+        allow_scope_expansion=True,
+    )
+    async def try_scope_expansion(
+        self,
+        ctx: RunContext[NoAuth],
+        reason: str = "e2e test of scope negotiation",
+        read_patterns: list[str] = ["reference/**"],
+        ttl_seconds: int = 60,
+        mode: str = "read_only",
+    ) -> dict:
+        original = list(getattr(ctx.workspace, "allow_patterns", ()))
+        await ctx.emit_progress(
+            f"requesting scope expansion: read={read_patterns} ttl={ttl_seconds}s"
+        )
+        try:
+            grant = await ctx.request_scope(
+                reason=reason,
+                read=read_patterns,
+                ttl_seconds=ttl_seconds,
+                mode=mode,
+            )
+        except ScopeDenied as exc:
+            await ctx.emit_progress(f"scope denied: {exc}")
+            return {
+                "ok": False,
+                "denied": True,
+                "reason": str(exc),
+                "original_allow": original,
+            }
+
+        new_allow = list(getattr(ctx.workspace, "allow_patterns", ()))
+        await ctx.emit_progress(f"scope granted: now {new_allow}")
+        return {
+            "ok": True,
+            "original_allow": original,
+            "new_allow": new_allow,
+            "new_grant_id": grant.grant_id,
+            "new_mode": grant.mode if isinstance(grant.mode, str)
+                        else grant.mode.value,
+            "new_ttl_seconds": grant.expires_at - grant.issued_at,
+        }

deploy/20-deployment.yaml

@@ -0,0 +1,70 @@
+apiVersion: apps/v1
+kind: Deployment
+metadata:
+  name: test-helper
+  namespace: agents
+  labels:
+    app: test-helper
+    a2a/managed-by: control-plane
+spec:
+  replicas: 1
+  strategy:
+    type: Recreate
+  selector:
+    matchLabels:
+      app: test-helper
+  template:
+    metadata:
+      labels:
+        app: test-helper
+    spec:
+      containers:
+        - name: agent
+          image: registry.a2acloud.io/agents/test-helper:placeholder
+          imagePullPolicy: Always
+          ports:
+            - containerPort: 8000
+              name: http
+          readinessProbe:
+            httpGet: {path: /healthz, port: 8000}
+            initialDelaySeconds: 2
+            periodSeconds: 5
+          livenessProbe:
+            httpGet: {path: /healthz, port: 8000}
+            initialDelaySeconds: 10
+            periodSeconds: 15
+          resources:
+            requests: {cpu: 50m, memory: 128Mi}
+            limits: {cpu: 200m, memory: 256Mi}
+---
+apiVersion: v1
+kind: Service
+metadata:
+  name: test-helper
+  namespace: agents
+spec:
+  type: ClusterIP
+  selector:
+    app: test-helper
+  ports:
+    - name: http
+      port: 80
+      targetPort: 8000
+---
+apiVersion: networking.k8s.io/v1
+kind: Ingress
+metadata:
+  name: test-helper
+  namespace: agents
+spec:
+  rules:
+    - host: test-helper.88-99-219-120.nip.io
+      http:
+        paths:
+          - path: /
+            pathType: Prefix
+            backend:
+              service:
+                name: test-helper
+                port:
+                  number: 80

requirements.txt

@@ -0,0 +1 @@
+pydantic>=2.6