Skip to content

feat(telemetry): realtime attested execution pools for AC and Prism #46

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
echobt merged 7 commits into from
Jul 27, 2026
Merged

feat(telemetry): realtime attested execution pools for AC and Prism #46

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
7 commits
Select commit Hold shift + click to select a range
c00d030
test(agent-challenge): add contracts for attested progress telemetry
echobt Jul 27, 2026
cd90e88
test(agent-challenge): add contracts for live execution pool
echobt Jul 27, 2026
a21e7ba
feat(agent-challenge): add attested progress and live pool telemetry
echobt Jul 27, 2026
e3fb72c
test(prism): add contracts for execution telemetry
echobt Jul 27, 2026
bbe5420
feat(prism): add execution events session and live pool
echobt Jul 27, 2026
616e45c
test(master): add contracts for eval telemetry proxy and pools
echobt Jul 27, 2026
016bff7
feat(master): allowlist eval telemetry and aggregate live pools
echobt Jul 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
221 changes: 221 additions & 0 deletions packages/challenges/agent-challenge/src/agent_challenge/api/routes.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -77,6 +77,12 @@
process_direct_eval_result,
validate_result_bounds,
)
from ..evaluation.execution_pool import list_live_execution_units
from ..evaluation.progress import (
MAX_PROGRESS_BODY_BYTES,
EvalProgressError,
process_eval_progress,
)
from ..evaluation.replay_audit import (
REPLAY_AUDIT_LABEL,
AggregationSpec,
Expand All @@ -102,6 +108,12 @@
redact_secrets,
redact_task_event_message,
)
from ..evaluation.telemetry_session import (
TELEMETRY_SESSION_HEADER,
TelemetrySessionError,
open_telemetry_session,
require_telemetry_session,
)
from ..evaluation.terminal_bench import TERMINAL_BENCH_EVALUATOR
from ..evaluation.validator_executor import (
finalize_job_if_complete,
Expand Down Expand Up @@ -1853,6 +1865,215 @@ async def receive_direct_eval_result(
)


@router.post(
"/evaluation/v1/runs/{eval_run_id}/telemetry-session",
)
async def open_eval_telemetry_session(
eval_run_id: str,
http_request: Request,
session: DatabaseSession,
authorization: Annotated[str | None, Header()] = None,
) -> JSONResponse:
"""Open a hotkey-attested telemetry session for mid-run progress posts.

Auth mirrors the final result route (Bearer EVAL_RUN_TOKEN). Body carries
hotkey_ss58 + signature only — never a mnemonic. Progress POSTs must present
the issued session id via X-Telemetry-Session.
"""

if not settings.attested_review_enabled or not settings.phala_attestation_enabled:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "attested_eval_disabled"},
)
prefix = "Bearer "
token = (
authorization[len(prefix) :]
if isinstance(authorization, str) and authorization.startswith(prefix)
else None
)
content_type = http_request.headers.get("content-type", "").split(";", 1)[0].strip().lower()
if content_type != "application/json":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"code": "telemetry_session_media_invalid"},
)
run = await session.scalar(select(EvalRun).where(EvalRun.eval_run_id == eval_run_id))
if run is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "eval_run_unknown"},
)
if not authenticate_eval_token(run, token):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"code": "invalid_eval_token"},
)
body = await _read_bounded_result_body(
http_request,
max_bytes=MAX_PROGRESS_BODY_BYTES,
)
try:
try:
parsed = json.loads(body.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise TelemetrySessionError(
"telemetry session body is not valid JSON",
code="invalid_telemetry_session",
) from exc
if not isinstance(parsed, dict):
raise TelemetrySessionError(
"telemetry session body must be a JSON object",
code="invalid_telemetry_session",
)
opened = open_telemetry_session(
eval_run_id=eval_run_id,
eval_run_phase=run.phase,
body=parsed,
)
except TelemetrySessionError as exc:
if exc.code == "eval_run_terminal":
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail={"code": exc.code},
) from exc
if exc.code in {
"invalid_telemetry_signature",
"invalid_telemetry_session",
"telemetry_signature_invalid",
}:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"code": exc.code},
) from exc
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={"code": exc.code},
) from exc
return JSONResponse(status_code=status.HTTP_200_OK, content=opened)


@router.post(
"/evaluation/v1/runs/{eval_run_id}/progress",
)
async def receive_eval_progress(
eval_run_id: str,
http_request: Request,
session: DatabaseSession,
authorization: Annotated[str | None, Header()] = None,
x_telemetry_session: Annotated[str | None, Header(alias="X-Telemetry-Session")] = None,
) -> JSONResponse:
"""Receive one mid-run task progress event from the canonical Eval CVM.

Auth: Bearer EVAL_RUN_TOKEN (same as result) plus X-Telemetry-Session from a
prior telemetry-session open. Events land in TaskLogEvent for the SSE feed.
This route never mutates scores — POST .../result remains the only score path.
"""

if not settings.attested_review_enabled or not settings.phala_attestation_enabled:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "attested_eval_disabled"},
)
prefix = "Bearer "
token = (
authorization[len(prefix) :]
if isinstance(authorization, str) and authorization.startswith(prefix)
else None
)
content_type = http_request.headers.get("content-type", "").split(";", 1)[0].strip().lower()
if content_type != "application/json":
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail={"code": "progress_media_invalid"},
)
run = await session.scalar(
select(EvalRun).where(EvalRun.eval_run_id == eval_run_id).with_for_update()
)
if run is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail={"code": "eval_run_unknown"},
)
if not authenticate_eval_token(run, token):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"code": "invalid_eval_token"},
)
# Prefer explicit Header param; fall back to raw header for alias robustness.
session_header = x_telemetry_session or http_request.headers.get(TELEMETRY_SESSION_HEADER)
try:
require_telemetry_session(session_header, eval_run_id=eval_run_id)
except TelemetrySessionError as exc:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail={"code": exc.code},
) from exc
body = await _read_bounded_result_body(
http_request,
max_bytes=MAX_PROGRESS_BODY_BYTES,
)
try:
# Progress is observability-only (optional float progress). Do not use
# parse_json_object / canonical_json_v1 which forbid floats — the final
# result route remains the only canonical attested body path.
try:
parsed = json.loads(body.decode("utf-8"))
except (UnicodeDecodeError, json.JSONDecodeError) as exc:
raise EvalProgressError(
"progress body is not valid JSON",
code="progress_invalid",
) from exc
if not isinstance(parsed, dict):
raise EvalProgressError(
"progress body must be a JSON object",
code="progress_invalid",
)
request_body = parsed
if request_body.get("eval_run_id") != eval_run_id:
raise EvalProgressError(
"progress run does not match route",
code="progress_run_mismatch",
)
receipt, created = await process_eval_progress(
session,
run=run,
progress_request=request_body,
)
await session.commit()
except EvalProgressError as exc:
await session.rollback()
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={"code": exc.code},
) from exc
except ValueError as exc:
await session.rollback()
code = exc.code if isinstance(exc, EvalProgressError) else "progress_invalid"
if code == "result_too_large":
raise HTTPException(
status_code=status.HTTP_413_CONTENT_TOO_LARGE,
detail={"code": code},
) from exc
raise HTTPException(
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
detail={"code": code},
) from exc
return JSONResponse(
status_code=status.HTTP_202_ACCEPTED if created else status.HTTP_200_OK,
content=receipt,
)


@public_route(tags=["execution"])
@router.get("/v1/execution-pool/live")
async def execution_pool_live(session: DatabaseSession) -> dict[str, list[dict[str, object]]]:
"""In-flight EvalRun units with latest TaskLogEvent. Empty pool is honest []."""

units = await list_live_execution_units(session)
return {"units": units}


@router.post("/submissions/{submission_id}/review/deployed")
async def acknowledge_submission_review_deployment(
submission_id: int,
Expand Down
29 changes: 28 additions & 1 deletion packages/challenges/agent-challenge/src/agent_challenge/canonical/compose.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -148,11 +148,36 @@
KEY_RELEASE_TLS_CA_ENV,
"LLM_COST_LIMIT",
"EVAL_RUN_TOKEN",
# Mid-run progress reporter (observability only; never score-bearing).
"EVAL_PROGRESS_BASE_URL",
"EVAL_RUN_ID",
"EVAL_SUBMISSION_ID",
# Local CVM signing secret name only — value never leaves the guest / never hits master APIs.
"RUNNER_HOTKEY_MNEMONIC",
# Measured OpenRouter (eval agent inside measured CVM only when product allows).
# Never Base gateway; keys stay miner/session encrypted_env on attested guests.
"OPENROUTER_API_KEY",
)


#: Env names that may appear in encrypted_env / runner config but are NOT part of
#: the measured compose_hash pin (``04011776…``). Progress/telemetry is optional
#: observability injected at deploy; baking the names into allowed_envs would
#: rotate every historical pin. ``encrypt_eval_secrets`` still accepts these via
#: ``DEFAULT_ALLOWED_ENVS``; compose generation for eval pins uses
#: :data:`MEASURED_ALLOWED_ENVS`.
PROGRESS_OPTIONAL_ENVS: tuple[str, ...] = (
"EVAL_PROGRESS_BASE_URL",
"EVAL_RUN_ID",
"EVAL_SUBMISSION_ID",
"RUNNER_HOTKEY_MNEMONIC",
)

#: Pin-stable allowlist used when rendering measured app-compose bytes.
MEASURED_ALLOWED_ENVS: tuple[str, ...] = tuple(
name for name in DEFAULT_ALLOWED_ENVS if name not in set(PROGRESS_OPTIONAL_ENVS)
)
Comment on lines +163 to +179

@coderabbitai coderabbitai Bot Jul 27, 2026

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

🔒 Security & Privacy | 🟠 Major | 🏗️ Heavy lift

🧩 Analysis chain

🌐 Web query:

Phala Cloud CVM encrypted_env env_keys must be listed in app-compose allowed_envs

💡 Result:

In Phala Cloud, when using a CVM (Confidential Virtual Machine) with on-chain Key Management Service (KMS), the list of environment variable keys provided in env_keys (during an environment update) must correspond to the allowed_envs whitelist defined in the application's app-compose.json configuration [1][2]. The app-compose.json is a canonicalized JSON manifest that defines your application's settings, including the allowed_envs field—a whitelist of environment variable names that your workload is permitted to read [1]. Because allowed_envs is a component of the app-compose.json file, any modification to this list (adding or removing an environment variable key) changes the manifest's contents, thereby generating a new compose_hash [1][2]. This process necessitates a specific two-phase update flow: 1. Phase 1: You submit the new encrypted_env and the updated env_keys list [3]. If these keys differ from the previously registered whitelist or if the on-chain configuration requires verification, the API will return a compose_hash [3][4]. 2. On-chain Registration: You must use this compose_hash to call the addComposeHash function on the blockchain to register the new configuration [1][2]. 3. Phase 2: Once registered, you retry the environment update request, providing both the compose_hash and the resulting transaction_hash as proof of on-chain registration [5][3]. If you update the encrypted values of environment variables without changing the keys themselves, the allowed_envs (and thus the compose_hash) remain unchanged, and the update can typically proceed without the two-phase on-chain registration flow [2].

Citations:

🏁 Script executed:

#!/bin/bash
set -euo pipefail

TARGET packages/challenges/agent-challenge/src/agent_challenge/canonical/compose.py

echo "== line counts =="
wc -l "$TARGET"

echo "== relevant constants and generated snippets =="
nl -ba "$TARGET" | sed -n '140,190p;350,405p'

echo "== env_keys/default progress references =="
rg -n "DEFAULT_ALLOWED_ENVS|MEASURED_ALLOWED_ENVS|PROGRESS_OPTIONAL_ENVS|encrypt_eval_secrets|generate_app_compose|env_keys|allowed_envs|EVAL_PROGRESS_BASE_URL|EVAL_RUN_TOKEN|ProgressReporter|ProgressReporter\.from_env|EVAL_RUN_ID|EVAL_SUBMISSION_ID|RUNNER_HOTKEY_MNEMONIC" packages/challenges/agent-challenge/src -S

Repository: BaseIntelligence/base

Length of output: 203

🏁 Script executed:

#!/bin/bash
set -euo pipefail

TARGET="packages/challenges/agent-challenge/src/agent_challenge/canonical/compose.py"

echo "== line counts =="
wc -l "$TARGET"

echo "== relevant constants and generated snippets =="
nl -ba "$TARGET" | sed -n '140,190p;350,405p'

echo "== env_keys/default progress references =="
rg -n "DEFAULT_ALLOWED_ENVS|MEASURED_ALLOWED_ENVS|PROGRESS_OPTIONAL_ENVS|encrypt_eval_secrets|generate_app_compose|env_keys|allowed_envs|EVAL_PROGRESS_BASE_URL|EVAL_RUN_TOKEN|ProgressReporter|ProgressReporter\.from_env|EVAL_RUN_ID|EVAL_SUBMISSION_ID|RUNNER_HOTKEY_MNEMONIC" packages/challenges/agent-challenge/src -S

echo "== candidate file references across repo =="
rg -n "eval_progress_base_url|progress_base_url|EVAL_PROGRESS_BASE_URL|ProgressReporter|EVAL_RUN_TOKEN|allowed_envs|env_keys|generate_app_compose|encrypt_eval_secrets" . -S

Repository: BaseIntelligence/base

Length of output: 347

🏁 Script executed:

#!/bin/bash
set -euo pipefail

TARGET="packages/challenges/agent-challenge/src/agent_challenge/canonical/compose.py"

echo "== line counts =="
wc -l "$TARGET"

echo "== relevant constants and generated snippets =="
awk '{printf "%6d\t", NR; print}' "$TARGET" | sed -n '140,190p;350,405p'

echo "== env_keys/default progress references in agent challenge src =="
rg -n "DEFAULT_ALLOWED_ENVS|MEASURED_ALLOWED_ENVS|PROGRESS_OPTIONAL_ENVS|encrypt_eval_secrets|generate_app_compose|env_keys|allowed_envs|EVAL_PROGRESS_BASE_URL|EVAL_RUN_TOKEN|ProgressReporter|ProgressReporter\.from_env|EVAL_RUN_ID|EVAL_SUBMISSION_ID|RUNNER_HOTKEY_MNEMONIC" packages/challenges/agent-challenge/src -S || true

echo "== repo-wide references =="
rg -n "eval_progress_base_url|progress_base_url|EVAL_PROGRESS_BASE_URL|ProgressReporter|EVAL_RUN_TOKEN|allowed_envs|env_keys|generate_app_compose|encrypt_eval_secrets" . -S || true

Repository: BaseIntelligence/base

Length of output: 50377

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== canonical compose render/generate relevant section =="
awk '{printf "%6d\t", NR; print}' packages/challenges/agent-challenge/src/agent_challenge/canonical/compose.py | sed -n '320,490p'

echo "== selfdeploy/eval encrypt/prepare relevant sections =="
awk '{printf "%6d\t", NR; print}' packages/challenges/agent-challenge/src/agent_challenge/selfdeploy/eval.py | sed -n '40,90p;220,335p;450,485p'

echo "== selfdeploy/api route/create request relevant env_keys =="
awk '{printf "%6d\t", NR; print}' packages/challenges/agent-challenge/src/agent_challenge/api/routes.py | sed -n '720,765p;4845,4905p' | sed -n '1,80p'

echo "== selfdeploy docs allow/env_keys section =="
if [ -f packages/challenges/agent-challenge/docs/miner/self-deploy.md ]; then
  awk '{printf "%6d\t", NR; print}' packages/challenges/agent-challenge/docs/miner/self-deploy.md | sed -n '100,170p;350,380p'
fi

Repository: BaseIntelligence/base

Length of output: 27029

🏁 Script executed:

#!/bin/bash
set -euo pipefail

echo "== selfdeploy/cli eval prepare/deploy create/encrypted section =="
awk '{printf "%6d\t", NR; print}' packages/challenges/agent-challenge/src/agent_challenge/selfdeploy/cli.py | sed -n '650,720p;730,760p;770,810p' || true

echo "== selfdeploy/cli dry-run/env_keys display section =="
awk '{printf "%6d\t", NR; print}' packages/challenges/agent-challenge/src/agent_challenge/selfdeploy/cli.py | sed -n '460,510p' || true

echo "== relevant eval progress wiring tests =="
awk '{printf "%6d\t", NR; print}' packages/challenges/agent-challenge/tests/test_eval_progress_env_wiring.py | sed -n '1,220p' || true

echo "== selfdeploy ordered trust hardening progress allowed/env_keys section =="
awk '{printf "%6d\t", NR; print}' packages/challenges/agent-challenge/tests/test_selfdeploy_ordered_trust_hardening.py | sed -n '200,270p;580,615p' || true

echo "== direct selfdeploy module references to phala api/env_vars/cvms =="
rg -n "EnvVar|env_vars|POST /cvms|env_keys=|encrypted_env|build_eval_progress|encrypt_eval_secrets|eval_deploy|encrypted.env_keys|env_keys" packages/challenges/agent-challenge/src/agent_challenge/selfdeploy -S

Repository: BaseIntelligence/base

Length of output: 36969

Align Phala env_keys with the measured allowed_envs before sending deploy secrets.

EVAL_PROGRESS_BASE_URL, EVAL_RUN_ID, EVAL_SUBMISSION_ID, and RUNNER_HOTKEY_MNEMONIC are encrypted as env_keys, but the default measured compose and EVAL_ALLOWED_ENVS use MEASURED_ALLOWED_ENVS, which drops those names. Phala CVM env_keys must match the compose allowed_envs; otherwise the runtime binding can fail or these telemetry/signing inputs won’t be injected to the CVM. Bind the progress inputs to a measured/compose-bound configuration instead of keeping an operator-shipped URL outside compose_hash.

🤖 Prompt for AI Agents 
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.

In `@packages/challenges/agent-challenge/src/agent_challenge/canonical/compose.py`
around lines 163 - 179, Align Phala deploy secret env_keys with the measured
compose allowlist by updating the relevant configuration and deployment flow
around MEASURED_ALLOWED_ENVS and EVAL_ALLOWED_ENVS. Ensure
EVAL_PROGRESS_BASE_URL, EVAL_RUN_ID, EVAL_SUBMISSION_ID, and
RUNNER_HOTKEY_MNEMONIC are included in the compose-bound measured configuration
and used consistently for env_keys, while removing any operator-shipped progress
URL that remains outside compose_hash.


_DIGEST_PIN_RE = re.compile(r"@sha256:[0-9a-f]{64}$")
_DIGEST_REF_RE = re.compile(r"^(?:sha256:)?[0-9a-f]{64}$")

Expand Down Expand Up @@ -355,7 +380,7 @@ def generate_app_compose(
orchestrator_image: str,
name: str = DEFAULT_APP_NAME,
command: Sequence[str] | None = None,
allowed_envs: Sequence[str] = DEFAULT_ALLOWED_ENVS,
allowed_envs: Sequence[str] = MEASURED_ALLOWED_ENVS,
key_release_url: str | None = None,
attestation_enabled: bool = True,
job_dir: str = DEFAULT_JOB_DIR,
Expand Down Expand Up @@ -496,6 +521,8 @@ def app_compose_hash(compose: Mapping[str, Any]) -> str:
"APP_COMPOSE_MANIFEST_VERSION",
"APP_COMPOSE_RUNNER",
"DEFAULT_ALLOWED_ENVS",
"MEASURED_ALLOWED_ENVS",
"PROGRESS_OPTIONAL_ENVS",
"DEFAULT_APP_NAME",
"DEFAULT_CACHE_ROOT",
"DEFAULT_DIGEST_MANIFEST",
Expand Down
Loading
Loading