feat(environments): add SQS-Kubernetes backend for remote EKS sandbox execution

[rmfan]

Motivation

Pier needs to drive remote EKS sandboxes over the same SQS bridge that
Harbor uses today (see the LLM360/agent-dist infrastructure). Right
now, pier ships Docker / Modal / Daytona backends only — running against
a hosted EKS cluster requires patching in Harbor's environment module,
which pulls in the entire Harbor runtime and its abstractions.

This PR ships a pier-native SQSKubernetesEnvironment that speaks
directly to the existing docker_k8s_consumer Deployment on EKS,
without any dependency on the harbor Python package. The consumer
(agent-dist sandbox/docker_k8s_consumer.py) is unchanged.

Design

Three implementation options were considered:

• Option A — direct import via pier's import_path mechanism.
  Users would declare import_path="harbor.environments.sqs_kubernetes:SQSKubernetesEnvironment"
  in TrialEnvironmentConfig. Zero pier code. Rejected: silently
  drops pier-specific features (default_user, network_allowlist,
  agent_install_spec, capabilities/resource_capabilities, filtered
  egress, Windows/agent-install validators) because Harbor's class
  subclasses Harbor's BaseEnvironment, not pier's. Only suitable as a
  short-lived proof-of-life spike.

• Option B — pier-native port. (THIS PR.)
  A fresh implementation of SQSKubernetesEnvironment that subclasses
  pier's BaseEnvironment directly, honours all pier abstractions, and
  registers as a first-class environment in _ENVIRONMENT_REGISTRY.
  ~2200 LoC of duplication with Harbor's client-side wire layer, but
  the file mirrors Harbor's structure closely so a future consolidation
  is straightforward. No runtime harbor dependency.

• Option C — shared wire library.
  Extract the pure SQS/S3 wire layer (message framing, three-tier
  encoding, S3 fallback, response-queue polling, retry policy, error-code
  vocabulary) into a new sandbox_sqs_client package shipped from
  agent-dist. Both Harbor and pier would subclass their own BaseEnvironment
  and delegate wire I/O to the shared library. Best long-term shape —
  single wire implementation, both frameworks pick up upstream fixes for
  free. Requires a packaging refactor of Harbor plus a coordinated
  release; deferred to a follow-up PR (see "Notes" below).

This PR ships Option B. Diverges from Harbor by:

• Subclassing pier's BaseEnvironment and honouring its abstract
  methods (_validate_definition, start, stop, exec,
  upload_file, upload_dir, download_file, download_dir).
• Exposing pier's capabilities / resource_capabilities properties
  (mounted=True because stop() pulls each volume mount before
  deletion; gpus=False, filtered_egress=False, windows=False).
• Using pier's ExecResult (no error_code field) — workload failures
  are synthesised into a normal ExecResult with an appropriate exit
  code (137 for OOM, 1 for DISK_FULL), so the agent loop sees a
  regular "user command failed" event and proceeds.
• Honouring pier's default_user, _merge_env, _resolve_user, and
  task_env_config.workdir on every exec() call.
• Plumbing pier's resource-enforcement policies (cpu_enforcement_policy,
  memory_enforcement_policy, request/limit values) into the
  containers/create body via Cpu / CpuRequest / Memory /
  MemoryRequest fields for the consumer to honour.
• Rejecting Windows tasks at _validate_definition time (EKS Linux
  nodegroups only).

No runtime harbor dependency — patterns reused conceptually; wire
protocol (SQS message shape, error-code vocabulary, S3 fallback) is
duplicated here.

Large-payload handling (load-bearing)

The 256 KiB SQS hard limit means any body approaching that size has to
fall back to S3. This port implements the same three-tier approach
Harbor uses:

1. Empty / small body → inline UTF-8 in the content field.
2. Body over _COMPRESSION_THRESHOLD (250 KiB) → zlib + base64 into
   content, with compress=true.
3. Encoded body over s3_threshold (default 200 KiB) → uploaded to S3,
   with s3_key in place of content. Never silently falls back to
   inline on S3 failure — a distinct S3PayloadError is raised so we
   don't hit MessageTooLong later (that was the bug agent-dist PR #212
   fixed for the analogous Path A code).

Response direction: when the consumer's response carries s3_key, the
client downloads from S3 before returning bytes to the caller.
Streaming exec (multiple STREAM chunks bounded at ~1 KiB each) does not
use S3 — verified by the existing chunked streaming path.

Per-agent egress allowlist (agent_process_env wiring)

Wires pier's BaseEnvironment.network_allowlist through to the cluster's
egress-proxy so agent-facing processes get scoped internet access without
blocking verifier / setup commands. Matches the server-side design in
agent-dist#215 (design doc)
implemented in agent-dist#216
(consumer + egress-proxy).

Flow:

1. start() — if not task_env_config.allow_internet and
   network_allowlist.domains is non-empty, adds
   network_allowlist={"domains":[…]} to the containers/create SQS body.
   Both preconditions guarantee zero wire-body change on tasks that don't
   opt in.
2. Consumer (PR #216) mints a 128-bit token, writes
   token -> {domains, sandbox_id} into the shared egress-allowlist
   ConfigMap, stamps the token as a pod annotation, and returns
   egress_token in the create response body.
3. start() captures the token on self._egress_token. An older consumer
   that doesn't understand network_allowlist returns no egress_token
   → self._egress_token stays None and everything downstream degrades
   to a no-op (unknown fields in the create body are ignored server-side).
4. agent_process_env(env) — no-op unless self._egress_token is set;
   with a token, injects
   HTTP_PROXY / http_proxy / HTTPS_PROXY / https_proxy = http://<token>:@egress-proxy.sandbox-proxy.svc.cluster.local:3128
   plus a NO_PROXY / no_proxy covering in-cluster + loopback. Caller-
   supplied env wins on key collision (mirrors docker / modal backends).
   Egress-proxy service address is configurable via an egress_proxy_url=
   constructor kwarg for non-default deploys.
5. stop() — resets self._egress_token = None unconditionally so any
   post-stop agent_process_env call returns env unchanged (the consumer
   also removes the ConfigMap entry on containers/delete).

capabilities.filtered_egress is set to True. The declaration is static
because the code path exists in every version of this class; the runtime
no-op on pre-PR-216 clusters is a data-plane concern (server returns no
token → we don't inject), not a capability claim.

Registration

Registered in _ENVIRONMENT_REGISTRY as EnvironmentType.SQS_KUBERNETES
("sqs_kubernetes"). Usable via pier run --env sqs_kubernetes ... or
via TrialEnvironmentConfig(type=EnvironmentType.SQS_KUBERNETES, kwargs=...).

Prerequisites (per cluster)

The consumer side must be deployed and reachable — see the LLM360/agent-dist
CLAUDE.md for cluster provisioning. Callers need:

• SQS queue — e.g. docker-requests-sandbox-eks-coding, passed as
  sqs_queue_url (full URL).
• S3 bucket — for large-payload fallback and build contexts, e.g.
  sqs-message-queue-large-objects-475108760152-eu-west-1-an.
• ECR registry URL — e.g. 123.dkr.ecr.eu-west-1.amazonaws.com/sandbox.
• AWS credentials — via boto3 credential chain (env vars, profile,
  or IRSA), optionally overridden via s3_access_key_id /
  s3_secret_access_key kwargs.

Optional kwargs:

• metrics_bridge_queue — trial-level metrics push.
• pre_install_commands — idempotent per-task setup commands run at
  create time.
• ephemeral_storage_limit — K8s resource quantity for heavy pip
  installs (e.g. 20Gi).
• slurm_user / slurm_job_id — stamped as pod labels for provenance
  (auto-fills from SLURM_JOB_USER / USER / SLURM_JOB_ID env vars).
• repo_url / repo_commit / repo_dest — server-side git clone in
  the consumer's BuildKit init container.
• create_max_attempts (default 8) — retry budget for retryable
  containers/create errors (CLUSTER_FULL, INTERNAL_ERROR, EVICTED).

Test plan

Unit tests (included)

46 tests in tests/environments/test_sqs_kubernetes.py:

• _validate_definition rejects missing / bad sqs_queue_url,
  missing s3_bucket, missing registry_url, missing
  environment_dir, and Windows tasks; happy-path passes.
• Large-payload handling (per the review focus):
  • 150 KiB body → inline, content populated, s3_key absent, no
    boto3 put_object call.
  • 250 KiB uncompressible → S3 upload, s3_key set to
    sqs-k8s-payloads/<uuid>, content empty, put_object called
    with correct bucket / key / body.
  • 2 MiB body → S3.
  • Response with s3_key → downloaded from S3 before returning.
  • Full roundtrip: 2 MiB request via S3, 3 MiB response via S3.
• S3 upload / download failure surfaces as S3PayloadError (distinct
  from SqsConsumerError) so callers can't confuse the two.
• Consumer-error parsing trusts body["retryable"] as the single
  source of truth; missing / malformed bodies default to
  retryable=False (fail-closed).
• Workload errors (is_workload=True) are synthesised into an
  ExecResult (OOM → return_code=137, DISK_FULL → 1) — never raised.
• capabilities property returns the correct shape (mounted=True,
  gpus=False, filtered_egress=False, windows=False, docker_compose=False).
• resource_capabilities returns both request+limit flags for CPU and
  memory.
• exec() signature matches pier's BaseEnvironment.exec exactly
  (command, cwd, env, timeout_sec, user).
• Factory registry contains EnvironmentType.SQS_KUBERNETES → correct
  module + class name.
• Chunk decoding (text vs binary content types), image-name /
  family sanitisers, Dockerfile COPY parsing.

All 46 pass; full pier suite (pytest tests/) still passes (140 total).
Ruff clean.

Manual smoke test (not in this PR)

Against a live consumer on sandbox-eks-coding:

pier run -p <task> \
  --env sqs_kubernetes \
  --env-kwargs '{"sqs_queue_url":"https://sqs.eu-west-1.amazonaws.com/475108760152/docker-requests-sandbox-eks-coding","s3_bucket":"sqs-message-queue-large-objects-475108760152-eu-west-1-an","registry_url":"475108760152.dkr.ecr.eu-west-1.amazonaws.com/sandbox"}'

Requires:

• AWS credentials that can sqs:SendMessage / sqs:ReceiveMessage on
  the docker-requests queue, and s3:GetObject / s3:PutObject on the
  S3 bucket.
• The consumer Deployment running (kubectl -n sandbox-proxy get pods -l app=docker-k8s-consumer).

Notes

• Long-term plan is Option C from the design section above — extract
  a shared sandbox_sqs_client wire library so Harbor and pier stop
  duplicating the SQS/S3 encoder, the retry/backoff logic, and the
  error-code vocabulary. This PR ships Option B first for immediate
  unblock; the duplication is contained to one file that closely
  mirrors Harbor's structure to keep future consolidation easy.
• Container-id lifecycle mirrors Harbor exactly: a fresh uuid per
  retry attempt (no 409 Conflict on retry), and every id the consumer
  may have acted on is tracked so stop() can send delayed DELETE
  messages for orphans (avoids sandbox-pod pile-up on CLUSTER_FULL /
  build-quota storms).
• The shared response queue is process-scoped and refcounted — 50
  pollers + 20 async delete workers per process, torn down when the
  last instance's stop() returns. Matches Harbor's pattern so a
  process running many trials in parallel amortises SQS TLS setup.

🤖 Generated with Claude Code

https://claude.ai/code/session_01WwJe2MmnF7s5ntM31tN38q

E2E Test Results (added post-merge-review)

scripts/e2e_sqs_kubernetes.py drives this env end-to-end against the shared prod
sandbox-eks cluster (queue docker-requests, S3 bucket
sqs-message-queue-large-objects-475108760152-eu-west-1-an). Every sandbox is
labelled slurm_user=pier-e2e-test-2026-08-21 and a try/finally sweep deletes
orphans via kubectl -n sandbox-proxy delete pods -l slurm_user=....

Test-run: 2026-08-21, PR head c9166d3 (was dca59d4) — image bash:latest
(busybox has no bash; pier's _sqs_exec_run hardcodes bash -c, so alpine /
busybox always give rc=127).

Test	Result	Duration	Evidence
1_happy_path	PASS	5.7s	create 3.3s, exec1 0.34s, exec2 0.40s, upload 0.37s, download 0.39s; sha256 roundtrip OK; HTTP_PROXY=http://egress-proxy.sandbox-proxy.svc:3128 auto-injected; OPENAI_BASE_URL unset on prod
2_large_payload	PASS	11.4s	2 MiB upload via S3 fallback (_upload_to_s3 spy fired once, key sqs-k8s-payloads/<uuid>); 2 MiB + 3 MiB roundtrips byte-perfect; bogus-bucket raises S3PayloadError("upload", ..., NoSuchBucket)
3_stress	PASS	690s	99/100 lifecycles (99.0%); 1 failure exec2 rc=-1 (STREAM timeout, no error_code); create p50 2.91s / p95 3.77s / p99 7.77s; exec p50 0.50s / p95 0.61s / p99 0.85s
4_network_allowlist	PASS with finding	5.6s	See finding below — pier does NOT plumb network_allowlist through; test still passes because behaviour is documented, not asserted

Update (commit 9ebb278): Test 4 now asserts allowlist enforcement
end-to-end (curl httpbin.org expects 2xx, curl example.com expects
403) when the target cluster is running the server-side changes from
agent-dist#216. On clusters still running the pre-PR-216 consumer, the
create response omits egress_token; the test detects that
(self._egress_token is None), emits a WARN, and falls back to the
previous "documented as no-op" behaviour so it still passes. Re-run
against the target cluster once PR #216 is deployed to see the
enforcement path in the summary.

Prod-cluster impact: peak 20 concurrent sandbox pods, 102 total lifecycles
across the four tests, ~110 SQS response-queue creations (14 leaked and were
swept), ~5 S3 objects in sqs-k8s-payloads/ (each ~2 MiB). Estimated cost:
~$0.05 (SQS + S3 + a few pod-seconds on nodes we already run).

Orphan check: kubectl get pods -l slurm_user=pier-e2e-test-2026-08-21 = 0
survivors after each run (orphans_found: 0 in the JSON summary).

Findings (not blockers)

• network_allowlist is a no-op in SQSKubernetesEnvironment.
  grep -n network_allowlist src/pier/environments/sqs_kubernetes.py matches
  only the filtered_egress=False capability line. The attribute set by
  BaseEnvironment.__init__ is ignored — sandbox pod egress is governed
  entirely by the cluster-wide egress-proxy env vars auto-injected by
  k8s_pod_backend.build_pod_spec(). This matches Harbor's current behaviour
  (Harbor also does not wire the attribute), so no regression. Worth
  documenting in the class docstring — happy to follow up.
• 1 exec rc=-1 under 20-way concurrent load. No error_code set,
  suggests STREAM timeout rather than a consumer-side terminal error. Pier's
  outer exec() retry loop caught it in prod usage; the E2E script bypasses
  that by calling _sqs_exec_run via exec() directly, so the retry did
  happen (retryable=True on infra errors) but the underlying STREAM_END
  didn't arrive within _KEEPALIVE_ROLLING_TIMEOUT_SEC. Hypothesis: shared
  cluster contention rather than a code bug. Verification would need a
  reproducer under a quiescent cluster + consumer log capture.

To reproduce

uv sync
uv run python scripts/e2e_sqs_kubernetes.py \
  --queue-url https://sqs.eu-west-1.amazonaws.com/475108760152/docker-requests \
  --s3-bucket sqs-message-queue-large-objects-475108760152-eu-west-1-an \
  --registry-url 475108760152.dkr.ecr.eu-west-1.amazonaws.com/sandbox \
  --concurrency 20 --iterations 5 --max-total 100 \
  --json-out e2e-summary.json

Requires AWS creds in the boto3 chain and kubectl configured for the
sandbox-eks context (used only by the post-run orphan sweep).