feat(telemetry): realtime attested execution pools for AC and Prism - #46
Conversation
Pin Bearer+session ingest, env wiring, and hotkey telemetry-session handshake so the monorepo port fails closed until implemented.
Require GET /v1/execution-pool/live to list in-flight eval runs without score fields for master fan-out.
Port progress validators, hotkey-bound telemetry sessions, progress ingest, ProgressReporter, env injection with mnemonic redaction, and GET /v1/execution-pool/live. Telemetry stays score-free.
Cover schema v5 execution_events, session/events routes, trust boundary (no tier elevation), and live pool surface.
Bump prism-schema.v5, persist execution_events, open hotkey-bound telemetry sessions, ingest monotone events, and expose running jobs on GET /v1/execution-pool/live without scoring or tier elevation.
Require allowlisted progress/session/result and public pool reads, plus GET /v1/pools/executing fan-out with partial-failure and jobs-shape pins.
Preserve Authorization on eval capability POSTs, allow public pool/SSE reads, and add GET /v1/pools/executing that fans out to challenge live pools while stripping score-like fields.
📝 WalkthroughWalkthroughChangesThe PR adds attested telemetry-session handshakes and score-free progress ingestion for Agent Challenge and Prism, best-effort runner reporting, live execution-pool endpoints, and a master proxy aggregator with route allowlisting, concurrent fan-out, response normalization, and score-field redaction. Execution telemetry and pool surfaces
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant Runner
participant AgentChallenge
participant Prism
participant MasterProxy
Runner->>AgentChallenge: Open telemetry session and post progress
AgentChallenge-->>Runner: Progress receipt
Runner->>Prism: Open telemetry session and post execution events
Prism-->>Runner: Inserted or duplicate counts
MasterProxy->>AgentChallenge: Fetch live execution units
MasterProxy->>Prism: Fetch live execution jobs
MasterProxy-->>Runner: Aggregated score-free execution pools
Suggested reviewers: 🚥 Pre-merge checks | ✅ 4 | ❌ 1❌ Failed checks (1 warning)
✅ Passed checks (4 passed)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 16
🧹 Nitpick comments (9)
src/base/master/app_proxy.py (1)
1326-1345: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueConsider routing the fan-out through
client_factory.Constructing
httpx.AsyncClientinline bypasses the injected factory used everywhere else in this app (shared limits, transport, redirect/proxy policy) purely sorespxcan intercept.respxpatches the transport globally, so tests would still pass if the factory-produced client were used — the trade-off is a divergent HTTP config path in the one route that fans out to every challenge. Also worth pinning a connect timeout separately from the 10s total, since a single hung challenge currently holds the aggregate request for the full window.🤖 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 `@src/base/master/app_proxy.py` around lines 1326 - 1345, Update the challenge fan-out around _fetch_challenge_execution_pool to obtain the HTTP client from the existing client_factory instead of constructing an inline httpx.AsyncClient, preserving the factory’s shared transport, limits, and redirect/proxy policy while retaining respx interception in tests. Configure a separate connect timeout alongside the existing total timeout so connection hangs do not consume the entire aggregate request window.tests/unit/test_pools_executing.py (1)
366-419: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winScore-stripping is never actually exercised.
Every mocked upstream body here is already score-free, so
_strip_pool_score_fieldscould be a no-op and the whole suite would still pass. Seed at least one upstream unit withscore/final_score/nestedweightsto prove the aggregator strips rather than merely relays.♻️ Suggested mock change
respx.get(PRISM_LIVE).mock( return_value=httpx.Response( 200, json={ "units": [ { "unit_id": "p1", "status": "executing", "hotkey": "hk1", + "score": 0.91, + "metrics": {"final_score": 0.91, "weights": [1, 2]}, } ] }, ) )🤖 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 `@tests/unit/test_pools_executing.py` around lines 366 - 419, Update test_pools_executing_payload_never_includes_score_fields to include score-related data in at least one mocked upstream unit, such as score, final_score, and nested weights, while retaining the existing forbidden-field assertions. Ensure the response still passes _assert_no_score_fields and contains no forbidden tokens, proving _strip_pool_score_fields removes upstream fields rather than only relaying already-clean payloads.packages/challenges/agent-challenge/src/agent_challenge/evaluation/execution_pool.py (1)
130-147: 🚀 Performance & Scalability | 🔵 TrivialConsider a result cap and short-lived cache for the public pool.
There is no upper bound on the number of units returned, and the master aggregates this endpoint across challenges. A
LIMITplus a few-seconds in-process cache would keep the endpoint cheap under fan-out polling.🤖 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/evaluation/execution_pool.py` around lines 130 - 147, Update the public pool retrieval flow around the EvalRun query and result construction to cap the number of returned units with an explicit LIMIT, and add a short-lived in-process cache for the assembled pool result. Ensure cached responses expire after a few seconds and avoid repeated database work during the cache window while preserving the existing active-phase ordering.packages/challenges/agent-challenge/tests/test_telemetry_session.py (1)
272-336: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winAdd a cross-run session-reuse case.
require_telemetry_sessionrejects a session whoseeval_run_iddiffers from the route, but no test covers it — open a session for run A and POST progress for run B expecting 401invalid_telemetry_session.🤖 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/tests/test_telemetry_session.py` around lines 272 - 336, Add a test alongside test_telemetry_session_expired_then_progress_rejected that opens a telemetry session for one eval run, posts progress to a different eval run using that session ID, and asserts a 401 response with detail code invalid_telemetry_session. Seed both runs and ensure the request uses the second run’s route and identifiers while reusing the first run’s session.packages/challenges/prism/src/prism_challenge/routes.py (2)
448-456: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueMove
import jsonto module scope.Function-local imports here buy nothing; both handlers pay a repeated lookup and the module already imports at the top.
Also applies to: 521-529
🤖 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/prism/src/prism_challenge/routes.py` around lines 448 - 456, Move the json import from the request-body parsing block to module scope in routes.py, and remove the duplicate function-local import in both affected handlers, including the code around the second referenced location. Keep the existing json.loads behavior and validation unchanged.
458-467: 🔒 Security & Privacy | 🔵 Trivial | ⚡ Quick winSilently stripping secret-shaped keys contradicts the docstring; reject instead.
The docstring says a mnemonic "is never accepted", but the handler accepts the request and drops the key — the caller gets a 200 and never learns their wallet material was transmitted. A 422 makes the contract observable and discourages repeats. The loop/break/comprehension can also collapse into one set.
♻️ Proposed rewrite
- # Reject secret material if present as first-class fields (optional hard reject). - for banned in ("mnemonic", "wallet_seed", "private_key", "seed"): - if banned in payload: - # Strip path: ignore banned keys rather than echo; still open session. - payload = { - k: v - for k, v in payload.items() - if k not in {"mnemonic", "wallet_seed", "private_key", "seed"} - } - break + # Fail closed on secret material; never echo, never persist, never accept. + if _BANNED_SECRET_KEYS.intersection(payload): + raise HTTPException( + status.HTTP_422_UNPROCESSABLE_ENTITY, + "secret material is not accepted on telemetry sessions", + )Note that
test_telemetry_session_rejects_mnemonic_and_binds_hotkeyalready allows{200, 422}, so this needs no test change.🤖 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/prism/src/prism_challenge/routes.py` around lines 458 - 467, Update the secret-field validation in the request handler around the banned-key loop to reject payloads containing any of {"mnemonic", "wallet_seed", "private_key", "seed"} with HTTP 422 instead of stripping those keys and opening a session. Replace the loop, break, and payload comprehension with a direct set-intersection check while preserving normal session handling for payloads without banned fields.packages/challenges/prism/src/prism_challenge/repository.py (1)
1630-1671: 🚀 Performance & Scalability | 🔵 Trivial | ⚡ Quick winN+1 query and unbounded result set on the live-pool read.
One extra
SELECTper running job on a public read path, with noLIMIT. A single correlated query (or a window function overexecution_events) collapses this to one round trip, and a cap keeps the response bounded.♻️ Single-query sketch
SELECT j.id, j.submission_id, j.level, j.status, e.event_type, e.sequence, e.task_id, e.payload, e.created_at AS event_created_at FROM eval_jobs j LEFT JOIN execution_events e ON e.id = ( SELECT id FROM execution_events WHERE eval_job_id = j.id ORDER BY sequence DESC LIMIT 1 ) WHERE j.status='running' ORDER BY j.updated_at DESC, j.id LIMIT ?🤖 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/prism/src/prism_challenge/repository.py` around lines 1630 - 1671, Update list_live_execution_pool to fetch running jobs and each job’s latest execution event with one SQL query, using a correlated subquery or window function instead of the per-job execution_events SELECT. Add a bounded LIMIT to the running-job result while preserving jobs without events via the existing latest_event=None behavior and retaining the current response fields.packages/challenges/prism/tests/test_execution_events.py (1)
1-13: 📐 Maintainability & Code Quality | 🔵 Trivial | ⚡ Quick winBoth new Prism telemetry test modules still assert like RED scaffolding. They were written to fail before the implementation existed, but the implementation ships in this same PR, so the deliberately loose assertions (multi-status sets,
if status == 200:branches, shape-agnostic payload parsing) no longer pin any contract and would pass through a real regression.
packages/challenges/prism/tests/test_execution_events.py#L1-L13: refresh the "MUST fail until production code lands" docstring and replace thein {200, 409}/in {200, 422}/ conditional-body assertions with the exact expected status and body.packages/challenges/prism/tests/test_execution_pool_live.py#L214-L236: assert the concrete{"jobs": [{"eval_job_id": ...}]}shape returned byexecution_pool_liverather than accepting a bare list orid/job_idaliases, and do the same for the empty-pool and curve branches.🤖 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/prism/tests/test_execution_events.py` around lines 1 - 13, Refresh the RED-test docstring in packages/challenges/prism/tests/test_execution_events.py:1-13 to describe the implemented behavior, and replace loose status sets and conditional payload checks with the exact expected status codes and response bodies for each telemetry-session and execution-events case. In packages/challenges/prism/tests/test_execution_pool_live.py:214-236, tighten execution_pool_live assertions for populated, empty, and curve responses to require the concrete {"jobs": [{"eval_job_id": ...}]} structure, without accepting bare lists or id/job_id aliases.packages/challenges/prism/tests/test_execution_pool_live.py (1)
28-63: 📐 Maintainability & Code Quality | 🔵 Trivial | 💤 Low valueDuplicate telemetry test helpers across the two new Prism test modules.
_settings,_internal_headers,_sign_bodyand the job-seeding helper are near-identical to those inpackages/challenges/prism/tests/test_execution_events.py. Promoting them toconftest.pykeeps the signing scheme in one place if the canonical message ever changes.🤖 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/prism/tests/test_execution_pool_live.py` around lines 28 - 63, Move the shared test helpers _settings, _internal_headers, _sign_body, and the job-seeding helper from test_execution_pool_live.py and test_execution_events.py into the Prism tests conftest.py. Update both test modules to reuse the centralized helpers, preserving their existing behavior and keeping the canonical signing logic in one place.
🤖 Prompt for all review comments with 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.
Inline comments:
In
`@packages/challenges/agent-challenge/src/agent_challenge/canonical/compose.py`:
- Around line 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.
In
`@packages/challenges/agent-challenge/src/agent_challenge/evaluation/execution_pool.py`:
- Around line 19-23: Consolidate the active-phase definitions by keeping a
single canonical set, preferably ACTIVE_EVAL_PHASES in execution_pool, and
import/reuse it for _ACTIVE_EVAL_PHASES in telemetry_session and _ACTIVE_PHASES
in evaluation.authorization. Remove the duplicated literals and update
references so all three modules always use the shared definition.
- Around line 110-124: The event-loading logic should select only the latest
TaskLogEvent per eval_run_id in SQL rather than fetching every event and
filtering in Python. Update the query in the surrounding execution-pool function
to use a window function or Postgres DISTINCT ON, preserving the existing
sequence/id descending tie-breaker and _latest_event_payload result mapping.
- Around line 93-98: Replace the ineffective top-level pop loop in the
unit-building function with a recursive _strip_score_keys helper that traverses
mappings and lists, removes keys whose lowercase names are in
_SCORE_FIELD_NAMES, and recursively sanitizes nested values. Apply it to the
complete unit after including latest_event so that nested score-shaped keys are
removed everywhere.
In
`@packages/challenges/agent-challenge/src/agent_challenge/evaluation/own_runner/progress_reporter.py`:
- Around line 44-65: Update ProgressReporter.from_env to parse and validate
PROGRESS_BASE_URL_ENV as an absolute URL whose scheme is exactly https and whose
network location is present; return None for invalid, non-HTTPS, or otherwise
unsafe values before constructing the reporter. Preserve normalized
trailing-slash removal for valid URLs and keep the existing token and
required-environment checks unchanged.
In
`@packages/challenges/agent-challenge/src/agent_challenge/evaluation/telemetry_session.py`:
- Around line 246-260: The run-finalization path must close any open telemetry
session when the evaluation reaches a terminal phase. Locate the existing
finalization handler and invoke close_telemetry_session with the run’s session
identifier, preserving the current session lifecycle behavior and avoiding
changes to unrelated telemetry helpers.
- Around line 190-201: Replace the module-level _SESSIONS storage used by the
session-open flow with persistence alongside EvalRun in the existing database
transaction, so sessions are shared across workers and replicas. Update the
corresponding progress/close lookups to read and mutate the persisted session
records, while preserving session expiry and validation behavior; if database
persistence is not available, purge expired records during open and enforce a
per-eval_run_id session cap.
- Around line 155-157: Strengthen session-open validation in the function
containing `_parse_timestamp(out["timestamp"])`: require `hotkey_ss58` to match
the run’s expected signer identity (plan/submission miner hotkey or CVM app
identity), enforce an allowed timestamp skew window, and reject reuse of the
run’s nonce. Preserve the existing signature and token checks while ensuring
each run can be opened only once with a correctly bound, fresh attestation.
In `@packages/challenges/agent-challenge/src/agent_challenge/selfdeploy/eval.py`:
- Around line 275-279: Replace the guest’s use of RUNNER_HOTKEY_MNEMONIC with a
narrowly scoped per-run signing seed or dedicated telemetry hotkey that can only
sign ac-telemetry-session:v1|… payloads. Update the self-deploy secret
preparation around DEFAULT_ALLOWED_ENVS and the corresponding guest signing flow
to pass only this scoped credential, while preserving encrypted handling and
ensuring logs, documentation, and evidence contain only secret names, digests,
or SHAs.
- Around line 470-476: Validate base_url before constructing the environment
mapping that includes EVAL_RUN_TOKEN: require the exact challenge callback URL
https://chain.joinbase.ai/challenges/agent-challenge, rejecting empty values,
non-HTTPS schemes, and foreign hosts. Bind EVAL_PROGRESS_BASE_URL only to this
pinned public origin so the guest cannot send the token elsewhere.
In `@packages/challenges/prism/src/prism_challenge/db.py`:
- Around line 218-227: Add a partial unique index for work-unit-only execution
events in the schema initialization and the _run_migrations path, keyed by
work_unit_id, task_id, and sequence and restricted to rows where eval_job_id IS
NULL. Preserve the existing eval_job_id uniqueness constraint and ensure pre-v5
databases receive the same index.
In `@packages/challenges/prism/src/prism_challenge/repository.py`:
- Around line 1594-1628: Update the execution-event append logic around the
current duplicate probe, max-sequence check, and INSERT so the insert atomically
arbitrates duplicates, including when eval_job_id is NULL. Catch the insert’s
uniqueness/conflict outcome and return "duplicate" instead of allowing an
IntegrityError to escape, while preserving sequence-regression validation and
ensuring NULL eval_job_id duplicates are prevented through the database
operation or equivalent atomic mechanism.
- Around line 1530-1563: Update create_telemetry_session to assign a bounded
expires_at value instead of NULL, using the repository’s existing time and
duration conventions. Update get_telemetry_session to reject or return no
session for rows whose expiry has passed, ensuring ingest_execution_events
cannot use expired session IDs while preserving valid-session behavior.
In `@packages/challenges/prism/src/prism_challenge/routes.py`:
- Around line 409-432: The _verify_telemetry_hotkey_signature helper must
consume each (hotkey, nonce) after successful signature validation, preventing
reuse within the TTL window. Reuse the existing nonces persistence and conflict
behavior from authenticate_miner, and ensure insertion occurs only after
timestamp and signature checks pass while preserving the existing HTTP errors.
- Around line 563-576: Update the event identity handling in the route around
eval_job_id and work_unit_id so the session binding is authoritative: when the
session has either identifier, reject the request if the client-supplied value
is present and differs after string normalization; otherwise use the
session-bound value. Do not allow client-provided identifiers to override the
session or write to another job’s stream, while preserving fallback behavior
when no session binding exists.
In `@src/base/master/app_proxy.py`:
- Around line 730-737: Update the HTTPError handling in the GET
/v1/pools/executing flow to return a generic public error message without
interpolating the exception type or text. Preserve the existing
"upstream_unreachable" error code and empty units response; if diagnostics are
needed, log the exception server-side rather than exposing it in the response.
---
Nitpick comments:
In
`@packages/challenges/agent-challenge/src/agent_challenge/evaluation/execution_pool.py`:
- Around line 130-147: Update the public pool retrieval flow around the EvalRun
query and result construction to cap the number of returned units with an
explicit LIMIT, and add a short-lived in-process cache for the assembled pool
result. Ensure cached responses expire after a few seconds and avoid repeated
database work during the cache window while preserving the existing active-phase
ordering.
In `@packages/challenges/agent-challenge/tests/test_telemetry_session.py`:
- Around line 272-336: Add a test alongside
test_telemetry_session_expired_then_progress_rejected that opens a telemetry
session for one eval run, posts progress to a different eval run using that
session ID, and asserts a 401 response with detail code
invalid_telemetry_session. Seed both runs and ensure the request uses the second
run’s route and identifiers while reusing the first run’s session.
In `@packages/challenges/prism/src/prism_challenge/repository.py`:
- Around line 1630-1671: Update list_live_execution_pool to fetch running jobs
and each job’s latest execution event with one SQL query, using a correlated
subquery or window function instead of the per-job execution_events SELECT. Add
a bounded LIMIT to the running-job result while preserving jobs without events
via the existing latest_event=None behavior and retaining the current response
fields.
In `@packages/challenges/prism/src/prism_challenge/routes.py`:
- Around line 448-456: Move the json import from the request-body parsing block
to module scope in routes.py, and remove the duplicate function-local import in
both affected handlers, including the code around the second referenced
location. Keep the existing json.loads behavior and validation unchanged.
- Around line 458-467: Update the secret-field validation in the request handler
around the banned-key loop to reject payloads containing any of {"mnemonic",
"wallet_seed", "private_key", "seed"} with HTTP 422 instead of stripping those
keys and opening a session. Replace the loop, break, and payload comprehension
with a direct set-intersection check while preserving normal session handling
for payloads without banned fields.
In `@packages/challenges/prism/tests/test_execution_events.py`:
- Around line 1-13: Refresh the RED-test docstring in
packages/challenges/prism/tests/test_execution_events.py:1-13 to describe the
implemented behavior, and replace loose status sets and conditional payload
checks with the exact expected status codes and response bodies for each
telemetry-session and execution-events case. In
packages/challenges/prism/tests/test_execution_pool_live.py:214-236, tighten
execution_pool_live assertions for populated, empty, and curve responses to
require the concrete {"jobs": [{"eval_job_id": ...}]} structure, without
accepting bare lists or id/job_id aliases.
In `@packages/challenges/prism/tests/test_execution_pool_live.py`:
- Around line 28-63: Move the shared test helpers _settings, _internal_headers,
_sign_body, and the job-seeding helper from test_execution_pool_live.py and
test_execution_events.py into the Prism tests conftest.py. Update both test
modules to reuse the centralized helpers, preserving their existing behavior and
keeping the canonical signing logic in one place.
In `@src/base/master/app_proxy.py`:
- Around line 1326-1345: Update the challenge fan-out around
_fetch_challenge_execution_pool to obtain the HTTP client from the existing
client_factory instead of constructing an inline httpx.AsyncClient, preserving
the factory’s shared transport, limits, and redirect/proxy policy while
retaining respx interception in tests. Configure a separate connect timeout
alongside the existing total timeout so connection hangs do not consume the
entire aggregate request window.
In `@tests/unit/test_pools_executing.py`:
- Around line 366-419: Update
test_pools_executing_payload_never_includes_score_fields to include
score-related data in at least one mocked upstream unit, such as score,
final_score, and nested weights, while retaining the existing forbidden-field
assertions. Ensure the response still passes _assert_no_score_fields and
contains no forbidden tokens, proving _strip_pool_score_fields removes upstream
fields rather than only relaying already-clean payloads.
🪄 Autofix (Beta)
Fix all unresolved CodeRabbit comments on this PR:
- Push a commit to this branch (recommended)
- Create a new PR with the fixes
ℹ️ Review info
⚙️ Run configuration
Configuration used: defaults
Review profile: CHILL
Plan: Pro Plus
Run ID: b37914c4-5bee-4737-880f-16a234fdea7d
📒 Files selected for processing (20)
packages/challenges/agent-challenge/src/agent_challenge/api/routes.pypackages/challenges/agent-challenge/src/agent_challenge/canonical/compose.pypackages/challenges/agent-challenge/src/agent_challenge/canonical/eval_wire.pypackages/challenges/agent-challenge/src/agent_challenge/evaluation/execution_pool.pypackages/challenges/agent-challenge/src/agent_challenge/evaluation/own_runner/progress_reporter.pypackages/challenges/agent-challenge/src/agent_challenge/evaluation/telemetry_session.pypackages/challenges/agent-challenge/src/agent_challenge/selfdeploy/cli.pypackages/challenges/agent-challenge/src/agent_challenge/selfdeploy/eval.pypackages/challenges/agent-challenge/tests/test_ac_execution_pool_live.pypackages/challenges/agent-challenge/tests/test_eval_progress_env_wiring.pypackages/challenges/agent-challenge/tests/test_eval_progress_ingest.pypackages/challenges/agent-challenge/tests/test_telemetry_session.pypackages/challenges/prism/src/prism_challenge/db.pypackages/challenges/prism/src/prism_challenge/repository.pypackages/challenges/prism/src/prism_challenge/routes.pypackages/challenges/prism/tests/test_execution_events.pypackages/challenges/prism/tests/test_execution_pool_live.pysrc/base/master/app_proxy.pytests/unit/test_agent_challenge_attested_proxy.pytests/unit/test_pools_executing.py
| #: 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) | ||
| ) |
There was a problem hiding this comment.
🔒 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:
- 1: https://docs.phala.com/phala-cloud/key-management/understanding-onchain-kms
- 2: https://docs.phala.com/phala-cloud/key-management/updating-with-onchain-kms
- 3: https://docs.phala.com/phala-cloud/references/cloud-js-sdk/update-cvm-envs
- 4: https://phalanetwork-1606097b.mintlify.app/phala-cloud/references/cloud-python-sdk/cvm-configuration
- 5: https://docs.phala.com/api-reference/cvms/update-environment-variables
🏁 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 -SRepository: 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" . -SRepository: 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 || trueRepository: 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'
fiRepository: 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 -SRepository: 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.
| # Keep in lockstep with evaluation.authorization._ACTIVE_PHASES and | ||
| # evaluation.telemetry_session._ACTIVE_EVAL_PHASES. | ||
| ACTIVE_EVAL_PHASES: frozenset[str] = frozenset( | ||
| {"eval_prepared", "eval_running", "eval_verifying"} | ||
| ) |
There was a problem hiding this comment.
📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
Three copies of the active-phase set kept in sync by comment only.
ACTIVE_EVAL_PHASES here, _ACTIVE_EVAL_PHASES in telemetry_session.py, and _ACTIVE_PHASES in evaluation/authorization.py must agree or a run becomes invisible in the pool while still accepting telemetry (or vice versa). Define it once and import it in the other two modules.
🤖 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/evaluation/execution_pool.py`
around lines 19 - 23, Consolidate the active-phase definitions by keeping a
single canonical set, preferably ACTIVE_EVAL_PHASES in execution_pool, and
import/reuse it for _ACTIVE_EVAL_PHASES in telemetry_session and _ACTIVE_PHASES
in evaluation.authorization. Remove the duplicated literals and update
references so all three modules always use the shared definition.
| "latest_event": dict(latest_event) if latest_event is not None else None, | ||
| } | ||
| # Defense in depth: never leak score-shaped keys even if a caller mutates. | ||
| for banned in _SCORE_FIELD_NAMES: | ||
| unit.pop(banned, None) | ||
| return unit |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
The score-key scrub is applied where it can't help and skipped where it could.
The pop loop iterates over a dict literal whose keys are fixed and none of which are in _SCORE_FIELD_NAMES, so it is a no-op. Meanwhile latest_event is copied through unscrubbed — currently safe because _latest_event_payload builds a fixed key set, but that is the surface that will grow. Apply the filter recursively to the whole unit (including latest_event) instead.
♻️ Proposed scrub
- # Defense in depth: never leak score-shaped keys even if a caller mutates.
- for banned in _SCORE_FIELD_NAMES:
- unit.pop(banned, None)
- return unit
+ # Defense in depth: strip score-shaped keys at every level, including
+ # latest_event, whose shape may grow over time.
+ return _strip_score_keys(unit)def _strip_score_keys(value: Any) -> Any:
if isinstance(value, Mapping):
return {
k: _strip_score_keys(v)
for k, v in value.items()
if str(k).lower() not in _SCORE_FIELD_NAMES
}
if isinstance(value, list):
return [_strip_score_keys(item) for item in value]
return value📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| "latest_event": dict(latest_event) if latest_event is not None else None, | |
| } | |
| # Defense in depth: never leak score-shaped keys even if a caller mutates. | |
| for banned in _SCORE_FIELD_NAMES: | |
| unit.pop(banned, None) | |
| return unit | |
| "latest_event": dict(latest_event) if latest_event is not None else None, | |
| } | |
| # Defense in depth: strip score-shaped keys at every level, including | |
| # latest_event, whose shape may grow over time. | |
| return _strip_score_keys(unit) | |
| def _strip_score_keys(value: Any) -> Any: | |
| if isinstance(value, Mapping): | |
| return { | |
| k: _strip_score_keys(v) | |
| for k, v in value.items() | |
| if str(k).lower() not in _SCORE_FIELD_NAMES | |
| } | |
| if isinstance(value, list): | |
| return [_strip_score_keys(item) for item in value] | |
| return value |
🤖 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/evaluation/execution_pool.py`
around lines 93 - 98, Replace the ineffective top-level pop loop in the
unit-building function with a recursive _strip_score_keys helper that traverses
mappings and lists, removes keys whose lowercase names are in
_SCORE_FIELD_NAMES, and recursively sanitizes nested values. Apply it to the
complete unit after including latest_event so that nested score-shaped keys are
removed everywhere.
| rows = await session.scalars( | ||
| select(TaskLogEvent) | ||
| .where(TaskLogEvent.submission_id.in_(list(submission_ids))) | ||
| .order_by(TaskLogEvent.sequence.desc(), TaskLogEvent.id.desc()) | ||
| ) | ||
| latest: dict[str, dict[str, Any]] = {} | ||
| for event in rows: | ||
| meta = _metadata_dict(event.metadata_json) | ||
| eval_run_id = meta.get("eval_run_id") | ||
| if not isinstance(eval_run_id, str) or not eval_run_id: | ||
| continue | ||
| if eval_run_id in latest: | ||
| continue | ||
| latest[eval_run_id] = _latest_event_payload(event) | ||
| return latest |
There was a problem hiding this comment.
🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy lift
Unbounded event scan on a public endpoint.
This loads every TaskLogEvent row for all active submissions into memory on each request just to keep the first per eval_run_id. Progress ingestion appends one row per task per sequence, so a few long runs make this thousands of rows per poll on an unauthenticated route.
Push the "latest per run" selection into SQL — a window function (ROW_NUMBER() OVER (PARTITION BY … ORDER BY sequence DESC)) or Postgres DISTINCT ON — or at minimum bound the scan with a LIMIT and an index on (submission_id, sequence DESC).
🤖 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/evaluation/execution_pool.py`
around lines 110 - 124, The event-loading logic should select only the latest
TaskLogEvent per eval_run_id in SQL rather than fetching every event and
filtering in Python. Update the query in the surrounding execution-pool function
to use a window function or Postgres DISTINCT ON, preserving the existing
sequence/id descending tie-breaker and _latest_event_payload result mapping.
| @classmethod | ||
| def from_env(cls, env: Mapping[str, str] | None = None) -> ProgressReporter | None: | ||
| """Build a reporter from injected env, or ``None`` if not configured.""" | ||
|
|
||
| source = os.environ if env is None else env | ||
| base_url = (source.get(PROGRESS_BASE_URL_ENV) or "").strip() | ||
| eval_run_id = (source.get(EVAL_RUN_ID_ENV) or "").strip() | ||
| submission_id = (source.get(EVAL_SUBMISSION_ID_ENV) or "").strip() | ||
| token = (source.get(EVAL_RUN_TOKEN_ENV) or "").strip() | ||
| if not (base_url and eval_run_id and submission_id and token): | ||
| return None | ||
| return cls( | ||
| base_url=base_url.rstrip("/"), | ||
| eval_run_id=eval_run_id, | ||
| submission_id=submission_id, | ||
| token=token, | ||
| timeout_seconds=_parse_timeout(source.get(PROGRESS_TIMEOUT_ENV)), | ||
| ) | ||
|
|
||
| @property | ||
| def url(self) -> str: | ||
| return f"{self.base_url}/evaluation/v1/runs/{self.eval_run_id}/progress" |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Constrain EVAL_PROGRESS_BASE_URL to an https origin before attaching the run token.
from_env accepts any string and url is handed straight to urlopen, which also resolves http://, file:// and ftp://. Since every request carries Authorization: Bearer <EVAL_RUN_TOKEN>, a misconfigured/downgraded base URL leaks the run capability in cleartext (or to an unintended sink). A scheme + netloc check at construction keeps this fail-closed and also answers the SSRF hint from static analysis.
🔒 Proposed scheme guard
+from urllib.parse import urlsplit
+
@@
if not (base_url and eval_run_id and submission_id and token):
return None
+ parts = urlsplit(base_url)
+ if parts.scheme != "https" or not parts.netloc:
+ logger.warning("progress reporter refused non-https base URL")
+ return None
return cls(📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| @classmethod | |
| def from_env(cls, env: Mapping[str, str] | None = None) -> ProgressReporter | None: | |
| """Build a reporter from injected env, or ``None`` if not configured.""" | |
| source = os.environ if env is None else env | |
| base_url = (source.get(PROGRESS_BASE_URL_ENV) or "").strip() | |
| eval_run_id = (source.get(EVAL_RUN_ID_ENV) or "").strip() | |
| submission_id = (source.get(EVAL_SUBMISSION_ID_ENV) or "").strip() | |
| token = (source.get(EVAL_RUN_TOKEN_ENV) or "").strip() | |
| if not (base_url and eval_run_id and submission_id and token): | |
| return None | |
| return cls( | |
| base_url=base_url.rstrip("/"), | |
| eval_run_id=eval_run_id, | |
| submission_id=submission_id, | |
| token=token, | |
| timeout_seconds=_parse_timeout(source.get(PROGRESS_TIMEOUT_ENV)), | |
| ) | |
| @property | |
| def url(self) -> str: | |
| return f"{self.base_url}/evaluation/v1/runs/{self.eval_run_id}/progress" | |
| from urllib.parse import urlsplit | |
| `@classmethod` | |
| def from_env(cls, env: Mapping[str, str] | None = None) -> ProgressReporter | None: | |
| """Build a reporter from injected env, or ``None`` if not configured.""" | |
| source = os.environ if env is None else env | |
| base_url = (source.get(PROGRESS_BASE_URL_ENV) or "").strip() | |
| eval_run_id = (source.get(EVAL_RUN_ID_ENV) or "").strip() | |
| submission_id = (source.get(EVAL_SUBMISSION_ID_ENV) or "").strip() | |
| token = (source.get(EVAL_RUN_TOKEN_ENV) or "").strip() | |
| if not (base_url and eval_run_id and submission_id and token): | |
| return None | |
| parts = urlsplit(base_url) | |
| if parts.scheme != "https" or not parts.netloc: | |
| logger.warning("progress reporter refused non-https base URL") | |
| return None | |
| return cls( | |
| base_url=base_url.rstrip("/"), | |
| eval_run_id=eval_run_id, | |
| submission_id=submission_id, | |
| token=token, | |
| timeout_seconds=_parse_timeout(source.get(PROGRESS_TIMEOUT_ENV)), | |
| ) | |
| `@property` | |
| def url(self) -> str: | |
| return f"{self.base_url}/evaluation/v1/runs/{self.eval_run_id}/progress" |
🤖 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/evaluation/own_runner/progress_reporter.py`
around lines 44 - 65, Update ProgressReporter.from_env to parse and validate
PROGRESS_BASE_URL_ENV as an absolute URL whose scheme is exactly https and whose
network location is present; return None for invalid, non-HTTPS, or otherwise
unsafe values before constructing the reporter. Preserve normalized
trailing-slash removal for valid URLs and keep the existing token and
required-environment checks unchanged.
Source: Linters/SAST tools
| async def create_telemetry_session( | ||
| self, | ||
| *, | ||
| eval_job_id: str | None, | ||
| work_unit_id: str | None, | ||
| instance_id: str, | ||
| hotkey_ss58: str, | ||
| nonce: str, | ||
| ) -> dict[str, str]: | ||
| """Open an attested telemetry session bound to hotkey_ss58 (never stores mnemonic).""" | ||
|
|
||
| session_id = str(uuid4()) | ||
| created = now_iso() | ||
| async with self.database.connect() as conn: | ||
| await conn.execute( | ||
| "INSERT INTO telemetry_sessions(" | ||
| "id, eval_job_id, work_unit_id, instance_id, hotkey_ss58, nonce, " | ||
| "created_at, expires_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", | ||
| ( | ||
| session_id, | ||
| eval_job_id, | ||
| work_unit_id, | ||
| instance_id, | ||
| hotkey_ss58, | ||
| nonce, | ||
| created, | ||
| None, | ||
| ), | ||
| ) | ||
| return { | ||
| "session_id": session_id, | ||
| "hotkey_ss58": hotkey_ss58, | ||
| "instance_id": instance_id, | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Sessions are created with expires_at = NULL and nothing ever expires them.
get_telemetry_session returns the row and ingest_execution_events only checks existence, so an issued session_id is an immortal bearer credential for writing execution events. Set a bounded expiry here and enforce it on read.
🔒 Proposed expiry binding
session_id = str(uuid4())
- created = now_iso()
+ created_dt = datetime.now(UTC)
+ created = created_dt.isoformat()
+ expires_at = (created_dt + timedelta(seconds=TELEMETRY_SESSION_TTL_SECONDS)).isoformat()
@@
created,
- None,
+ expires_at,Then reject expired rows in get_telemetry_session (or in the route) rather than returning them.
🤖 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/prism/src/prism_challenge/repository.py` around lines
1530 - 1563, Update create_telemetry_session to assign a bounded expires_at
value instead of NULL, using the repository’s existing time and duration
conventions. Update get_telemetry_session to reject or return no session for
rows whose expiry has passed, ensuring ingest_execution_events cannot use
expired session IDs while preserving valid-session behavior.
| async with self.database.connect() as conn: | ||
| existing = await conn.execute_fetchall( | ||
| "SELECT id FROM execution_events " | ||
| "WHERE eval_job_id IS ? AND task_id=? AND sequence=?", | ||
| (eval_job_id, task_id, sequence), | ||
| ) | ||
| if existing: | ||
| return "duplicate" | ||
| max_rows = await conn.execute_fetchall( | ||
| "SELECT COALESCE(MAX(sequence), 0) AS sequence FROM execution_events " | ||
| "WHERE eval_job_id IS ? AND task_id=?", | ||
| (eval_job_id, task_id), | ||
| ) | ||
| max_seq = int(list(max_rows)[0]["sequence"]) | ||
| if sequence <= max_seq: | ||
| raise ValueError("execution_event_sequence_regress") | ||
| event_id = str(uuid4()) | ||
| await conn.execute( | ||
| "INSERT INTO execution_events(" | ||
| "id, eval_job_id, work_unit_id, task_id, sequence, event_type, payload, " | ||
| "session_id, hotkey_ss58, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)", | ||
| ( | ||
| event_id, | ||
| eval_job_id, | ||
| work_unit_id, | ||
| task_id, | ||
| sequence, | ||
| event_type, | ||
| dumps(payload), | ||
| session_id, | ||
| hotkey_ss58, | ||
| now_iso(), | ||
| ), | ||
| ) | ||
| return "inserted" |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
Read-then-insert is not atomic; concurrent posts either 500 or duplicate.
The duplicate probe and MAX(sequence) read happen in separate statements from the INSERT, and each database.connect() is its own connection/transaction. Two in-flight batches for the same (eval_job_id, task_id, sequence) can both pass the checks: with a non-NULL eval_job_id the second insert trips the unique constraint and surfaces as an unhandled IntegrityError (500 rather than the intended "duplicate"); with a NULL eval_job_id there is no constraint at all and a duplicate row lands (see the related note on packages/challenges/prism/src/prism_challenge/db.py). Make the insert itself the arbiter.
🛠️ Proposed atomic append
- event_id = str(uuid4())
- await conn.execute(
- "INSERT INTO execution_events("
+ event_id = str(uuid4())
+ cursor = await conn.execute(
+ "INSERT OR IGNORE INTO execution_events("
"id, eval_job_id, work_unit_id, task_id, sequence, event_type, payload, "
"session_id, hotkey_ss58, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)",
@@
)
- return "inserted"
+ return "inserted" if cursor.rowcount else "duplicate"🤖 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/prism/src/prism_challenge/repository.py` around lines
1594 - 1628, Update the execution-event append logic around the current
duplicate probe, max-sequence check, and INSERT so the insert atomically
arbitrates duplicates, including when eval_job_id is NULL. Catch the insert’s
uniqueness/conflict outcome and return "duplicate" instead of allowing an
IntegrityError to escape, while preserving sequence-regression validation and
ensuring NULL eval_job_id duplicates are prevented through the database
operation or equivalent atomic mechanism.
| def _verify_telemetry_hotkey_signature( | ||
| request: Request, | ||
| *, | ||
| hotkey: str, | ||
| nonce: str, | ||
| timestamp: str, | ||
| signature: str, | ||
| body: bytes, | ||
| ) -> None: | ||
| app_settings = request.app.state.settings | ||
| try: | ||
| ts = int(timestamp) | ||
| except ValueError as exc: | ||
| raise HTTPException(status.HTTP_400_BAD_REQUEST, "invalid timestamp") from exc | ||
| if abs(int(datetime.now(UTC).timestamp()) - ts) > app_settings.signature_ttl_seconds: | ||
| raise HTTPException(status.HTTP_401_UNAUTHORIZED, "stale signature") | ||
| message = canonical_submission_message( | ||
| hotkey=hotkey, nonce=nonce, timestamp=timestamp, body=body | ||
| ) | ||
| valid = verify_hotkey_signature(hotkey, message, signature) | ||
| if not valid and app_settings.allow_insecure_signatures: | ||
| valid = verify_dev_signature(app_settings.internal_token(), message, signature) | ||
| if not valid: | ||
| raise HTTPException(status.HTTP_401_UNAUTHORIZED, "invalid signature") |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟠 Major | ⚡ Quick win
Telemetry signatures are replayable within the TTL window; the nonce is never consumed.
authenticate_miner inserts each (hotkey, nonce) into nonces and 409s on reuse, but this helper only checks freshness. The exact same signed session-open body can be replayed until signature_ttl_seconds elapses, minting extra sessions. Persist the nonce the same way the miner path does.
🤖 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/prism/src/prism_challenge/routes.py` around lines 409 -
432, The _verify_telemetry_hotkey_signature helper must consume each (hotkey,
nonce) after successful signature validation, preventing reuse within the TTL
window. Reuse the existing nonces persistence and conflict behavior from
authenticate_miner, and ensure insertion occurs only after timestamp and
signature checks pass while preserving the existing HTTP errors.
| eval_job_id = raw_event.get("eval_job_id") | ||
| work_unit_id = raw_event.get("work_unit_id") | ||
| if eval_job_id is not None: | ||
| eval_job_id = str(eval_job_id) | ||
| else: | ||
| eval_job_id = ( | ||
| str(session["eval_job_id"]) if session.get("eval_job_id") is not None else None | ||
| ) | ||
| if work_unit_id is not None: | ||
| work_unit_id = str(work_unit_id) | ||
| else: | ||
| work_unit_id = ( | ||
| str(session["work_unit_id"]) if session.get("work_unit_id") is not None else None | ||
| ) |
There was a problem hiding this comment.
🔒 Security & Privacy | 🔴 Critical | ⚡ Quick win
A telemetry session can write events onto any other job's stream.
eval_job_id/work_unit_id are taken from the client-supplied event and only fall back to the session's binding. A holder of a session opened for job A can post events tagged with job B, which then surface in GET /execution-pool/live for B and can also poison B's sequence space (a high sequence makes later legitimate events fail with execution_event_sequence_regress). The session binding should be authoritative.
🔒 Proposed fix: reject mismatches instead of trusting the event
- eval_job_id = raw_event.get("eval_job_id")
- work_unit_id = raw_event.get("work_unit_id")
- if eval_job_id is not None:
- eval_job_id = str(eval_job_id)
- else:
- eval_job_id = (
- str(session["eval_job_id"]) if session.get("eval_job_id") is not None else None
- )
- if work_unit_id is not None:
- work_unit_id = str(work_unit_id)
- else:
- work_unit_id = (
- str(session["work_unit_id"]) if session.get("work_unit_id") is not None else None
- )
+ session_job = (
+ str(session["eval_job_id"]) if session.get("eval_job_id") is not None else None
+ )
+ session_unit = (
+ str(session["work_unit_id"]) if session.get("work_unit_id") is not None else None
+ )
+ claimed_job = raw_event.get("eval_job_id")
+ claimed_unit = raw_event.get("work_unit_id")
+ if claimed_job is not None and str(claimed_job) != (session_job or ""):
+ raise HTTPException(status.HTTP_403_FORBIDDEN, "event job is not bound to session")
+ if claimed_unit is not None and str(claimed_unit) != (session_unit or ""):
+ raise HTTPException(status.HTTP_403_FORBIDDEN, "event unit is not bound to session")
+ eval_job_id = session_job
+ work_unit_id = session_unit📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| eval_job_id = raw_event.get("eval_job_id") | |
| work_unit_id = raw_event.get("work_unit_id") | |
| if eval_job_id is not None: | |
| eval_job_id = str(eval_job_id) | |
| else: | |
| eval_job_id = ( | |
| str(session["eval_job_id"]) if session.get("eval_job_id") is not None else None | |
| ) | |
| if work_unit_id is not None: | |
| work_unit_id = str(work_unit_id) | |
| else: | |
| work_unit_id = ( | |
| str(session["work_unit_id"]) if session.get("work_unit_id") is not None else None | |
| ) | |
| session_job = ( | |
| str(session["eval_job_id"]) if session.get("eval_job_id") is not None else None | |
| ) | |
| session_unit = ( | |
| str(session["work_unit_id"]) if session.get("work_unit_id") is not None else None | |
| ) | |
| claimed_job = raw_event.get("eval_job_id") | |
| claimed_unit = raw_event.get("work_unit_id") | |
| if claimed_job is not None and str(claimed_job) != (session_job or ""): | |
| raise HTTPException(status.HTTP_403_FORBIDDEN, "event job is not bound to session") | |
| if claimed_unit is not None and str(claimed_unit) != (session_unit or ""): | |
| raise HTTPException(status.HTTP_403_FORBIDDEN, "event unit is not bound to session") | |
| eval_job_id = session_job | |
| work_unit_id = session_unit |
🤖 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/prism/src/prism_challenge/routes.py` around lines 563 -
576, Update the event identity handling in the route around eval_job_id and
work_unit_id so the session binding is authoritative: when the session has
either identifier, reject the request if the client-supplied value is present
and differs after string normalization; otherwise use the session-bound value.
Do not allow client-provided identifiers to override the session or write to
another job’s stream, while preserving fallback behavior when no session binding
exists.
| except httpx.HTTPError as exc: | ||
| return { | ||
| "units": [], | ||
| "error": { | ||
| "code": "upstream_unreachable", | ||
| "message": f"{type(exc).__name__}: {exc}", | ||
| }, | ||
| } |
There was a problem hiding this comment.
🔒 Security & Privacy | 🟡 Minor | ⚡ Quick win
Don't echo raw upstream exception text on a public route.
f"{type(exc).__name__}: {exc}" surfaces httpx's message, which for connect/timeout/protocol errors commonly embeds the internal target URL (http://127.0.0.1:18081/...). GET /v1/pools/executing is unauthenticated, so this leaks embed topology. Keep the code, drop the interpolated detail (or log it server-side instead).
🛡️ Proposed fix
except httpx.HTTPError as exc:
return {
"units": [],
"error": {
"code": "upstream_unreachable",
- "message": f"{type(exc).__name__}: {exc}",
+ "message": type(exc).__name__,
},
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| except httpx.HTTPError as exc: | |
| return { | |
| "units": [], | |
| "error": { | |
| "code": "upstream_unreachable", | |
| "message": f"{type(exc).__name__}: {exc}", | |
| }, | |
| } | |
| except httpx.HTTPError as exc: | |
| return { | |
| "units": [], | |
| "error": { | |
| "code": "upstream_unreachable", | |
| "message": type(exc).__name__, | |
| }, | |
| } |
🤖 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 `@src/base/master/app_proxy.py` around lines 730 - 737, Update the HTTPError
handling in the GET /v1/pools/executing flow to return a generic public error
message without interpolating the exception type or text. Preserve the existing
"upstream_unreachable" error code and empty units response; if diagnostics are
needed, log the exception server-side rather than exposing it in the response.
2 participants
Summary
GET /v1/pools/executing.Changes
telemetry-session,POST …/progress(Bearer +X-Telemetry-Session),ProgressReporter, env wiring withRUNNER_HOTKEY_MNEMONICredaction,GET /v1/execution-pool/live.prism-schema.v5+execution_events, session/events routes, live pool of RUNNING jobs;constation_okremains sole tier elevation.jobsshape.Test plan
uv run pytest packages/challenges/agent-challenge/tests/test_eval_progress_*.py packages/challenges/agent-challenge/tests/test_telemetry_session.py packages/challenges/agent-challenge/tests/test_ac_execution_pool_live.py packages/challenges/agent-challenge/tests/test_task_event_stream.py -q→ 32 passeduv run pytest packages/challenges/prism/tests/test_execution_events.py packages/challenges/prism/tests/test_execution_pool_live.py -q→ 13 passeduv run pytest tests/unit/test_agent_challenge_attested_proxy.py tests/unit/test_pools_executing.py -q→ 188 passedGET /v1/pools/executingshowed both; teardown to 0 billables (receipts under/tmp/telemetry-live-evidence/w6-live/)/work/baseintelligence/basetree left untouchedCI
Notes
hotkey_ss58+ signature.Summary by CodeRabbit
New Features
Security
Bug Fixes