feat(landmine): port prod hotpatch landmines into main

docker/master-entrypoint.sh

@@ -208,7 +208,24 @@ start_embedded_challenges() {
     "PRISM_DOCKER_ENABLED=${PRISM_DOCKER_ENABLED:-false}"
     "PRISM_WORKER_PLANE__ENABLED=${PRISM_WORKER_PLANE__ENABLED:-false}"
     "PRISM_DOCKER_BACKEND=${PRISM_DOCKER_BACKEND:-cli}"
+    # HOTPATCH allowlist: OpenRouter plagiarism + worker plane.
+# PROD POLICY: Prism eval never runs on master — CPU_REEXEC must stay false;
+# miners supply Lium pods; admission_requires_worker must stay true.
+    "PRISM_WORKER_PLANE__CPU_REEXEC_TEST_MODE=${PRISM_WORKER_PLANE__CPU_REEXEC_TEST_MODE:-false}"
+    "PRISM_WORKER_PLANE__ADMISSION_REQUIRES_WORKER=${PRISM_WORKER_PLANE__ADMISSION_REQUIRES_WORKER:-true}"
+    "PRISM_WORKER_PLANE__MASTER_BASE_URL=${PRISM_WORKER_PLANE__MASTER_BASE_URL:-http://127.0.0.1:8081}"
+    "PRISM_PLAGIARISM_LLM_ENABLED=${PRISM_PLAGIARISM_LLM_ENABLED:-false}"
+    "PRISM_PLAGIARISM_LLM_REQUIRED=${PRISM_PLAGIARISM_LLM_REQUIRED:-false}"
+    "PRISM_OPENROUTER_API_KEY_FILE=${PRISM_OPENROUTER_API_KEY_FILE:-/run/secrets/openrouter_api_key}"
+    "PRISM_OPENROUTER_BASE_URL=${PRISM_OPENROUTER_BASE_URL:-https://openrouter.ai/api/v1}"
+    "PRISM_OPENROUTER_MODEL=${PRISM_OPENROUTER_MODEL:-x-ai/grok-4.5}"
+    "PRISM_ALLOW_INSECURE_SIGNATURES=${PRISM_ALLOW_INSECURE_SIGNATURES:-false}"
+    "PRISM_CONSTATION_BASE_URL=${PRISM_CONSTATION_BASE_URL:-http://127.0.0.1:8081}"
   )
+  # Optional constation token only when parent set it (avoid empty unknown noise)
+  if [[ -n "${PRISM_CONSTATION_INTERNAL_TOKEN:-}" ]]; then
+    prism_env+=("PRISM_CONSTATION_INTERNAL_TOKEN=${PRISM_CONSTATION_INTERNAL_TOKEN}")
+  fi
   if [[ -n "${py_path}" ]]; then
     prism_env+=("PYTHONPATH=${py_path}")
   fi

packages/challenges/prism/src/prism_challenge/app.py

@@ -21,7 +21,11 @@
     work_unit_to_payload,
 )
 from .db import Database
-from .evaluator.checkpoint_intake import CheckpointIntakeError, CheckpointIntakeService
+from .evaluator.checkpoint_intake import (
+    CheckpointIntakeError,
+    CheckpointIntakeService,
+    CheckpointPublishError,
+)
 from .evaluator.checkpoint_publisher import (
     CheckpointPublisher,
     HuggingFaceCheckpointPublisher,
@@ -292,7 +296,20 @@ async def publish_checkpoint(
             )
         except CheckpointIntakeError as exc:
             raise HTTPException(status.HTTP_400_BAD_REQUEST, str(exc)) from exc
+        except CheckpointPublishError as exc:
+            # Publisher failed: no checkpoint_ref recorded; surface last_error without tokens.
+            raise HTTPException(
+                status.HTTP_502_BAD_GATEWAY,
+                detail={
+                    "status": "failed",
+                    "error": str(exc),
+                    "submission_id": exc.submission_id,
+                    "repo_id": exc.repo_id,
+                    "checkpoint_ref": None,
+                },
+            ) from exc
         return {
+            "status": "success",
             "checkpoint_ref": published.checkpoint_ref,
             "repo_id": published.repo_id,
             "revision": published.revision,

packages/challenges/prism/src/prism_challenge/config.py

@@ -10,6 +10,8 @@
 from pydantic import AliasChoices, BaseModel, Field
 from pydantic_settings import SettingsConfigDict
 
+from .evaluator.checkpoint_publisher import DEFAULT_CHECKPOINT_REPO_ID
+
 _DATA_TMP_ARTIFACT_ROOT = Path("/data/tmp/prism-eval-artifacts")
 _TMP_ARTIFACT_ROOT = Path("/tmp/prism-eval-artifacts")
 
@@ -397,7 +399,7 @@ def _known_environment_names(cls) -> set[str]:
         ),
     )
     checkpoint_repo_id: str = Field(
-        default="baseintelligence/prism-checkpoints",
+        default=DEFAULT_CHECKPOINT_REPO_ID,
         validation_alias=AliasChoices("PRISM_CHECKPOINT_REPO_ID", "PRISM_HF_CHECKPOINT_REPO_ID"),
     )
     subnet_rules_json: str | None = None
@@ -455,9 +457,7 @@ def _known_environment_names(cls) -> set[str]:
     openrouter_api_key_file: str | None = Field(
         default="/run/secrets/openrouter_api_key",
         repr=False,
-        validation_alias=AliasChoices(
-            "PRISM_OPENROUTER_API_KEY_FILE", "OPENROUTER_API_KEY_FILE"
-        ),
+        validation_alias=AliasChoices("PRISM_OPENROUTER_API_KEY_FILE", "OPENROUTER_API_KEY_FILE"),
     )
     openrouter_base_url: str = Field(
         default="https://openrouter.ai/api/v1",

packages/challenges/prism/src/prism_challenge/evaluator/checkpoint_intake.py

@@ -8,16 +8,21 @@
 This module owns ONLY the master-side intake/publish/record step. The hotkey-signed, permit-gated
 HTTP endpoint is wired in :mod:`prism_challenge.app`; the validator-side cadence + push client lives
 in :mod:`prism_challenge.evaluator.checkpoint_push`.
+
+Observability: every publish attempt updates :attr:`CheckpointIntakeService.last_status` /
+``last_error`` / ``last_checkpoint_ref`` and emits a structured log line with ``repo_id`` +
+``submission_id`` (never tokens). Failed publishes never record a ``checkpoint_ref``.
 """
 
 from __future__ import annotations
 
 import asyncio
+import logging
 from collections.abc import Mapping
 from dataclasses import dataclass
 from pathlib import Path
 from tempfile import TemporaryDirectory
-from typing import Protocol
+from typing import Literal, Protocol
 
 from .checkpoint_publisher import (
     CheckpointPublisher,
@@ -27,11 +32,24 @@
 )
 from .checkpoints import resolve_checkpoint_artifact_path
 
+logger = logging.getLogger(__name__)
+
+PublishStatus = Literal["success", "failed"]
+
 
 class CheckpointIntakeError(ValueError):
     """Raised when an uploaded checkpoint payload is malformed (no files / unsafe path)."""
 
 
+class CheckpointPublishError(RuntimeError):
+    """Raised when the publisher fails; no ``checkpoint_ref`` is recorded."""
+
+    def __init__(self, message: str, *, submission_id: str, repo_id: str) -> None:
+        super().__init__(message)
+        self.submission_id = submission_id
+        self.repo_id = repo_id
+
+
 class SupportsRecordCheckpoint(Protocol):
     """The slice of :class:`~prism_challenge.repository.PrismRepository` this service needs."""
 
@@ -46,12 +64,48 @@ async def record_published_checkpoint(
     ) -> None: ...
 
 
+@dataclass(frozen=True)
+class CheckpointIntakeResult:
+    """Observable outcome of one master-side publish attempt."""
+
+    status: PublishStatus
+    submission_id: str
+    repo_id: str
+    checkpoint_ref: str | None
+    revision: str | None
+    files: tuple[str, ...]
+    last_error: str | None = None
+
+    @property
+    def ok(self) -> bool:
+        return self.status == "success" and bool(self.checkpoint_ref)
+
+
+def is_training_publish_complete(
+    *,
+    hf_token_configured: bool,
+    checkpoint_ref: str | None,
+) -> bool:
+    """Whether training completion may be marked fully successful.
+
+    Fail-closed when HF is configured: a published ``checkpoint_ref`` is required.
+    When HF is disabled (dev / mock offline path), missing ref is allowed.
+    """
+    if checkpoint_ref:
+        return True
+    return not hf_token_configured
+
+
 @dataclass
 class CheckpointIntakeService:
     """Receive a pushed checkpoint, publish it via the publisher, and record the public ref."""
 
     publisher: CheckpointPublisher
     repository: SupportsRecordCheckpoint
+    last_status: PublishStatus | None = None
+    last_error: str | None = None
+    last_checkpoint_ref: str | None = None
+    last_result: CheckpointIntakeResult | None = None
 
     async def publish(
         self,
@@ -67,28 +121,76 @@ async def publish(
 
         The (mock) publisher upload runs off the event loop. Only AFTER a successful publish is the
         ``checkpoint_ref`` recorded on the assignment, so a failed publish records nothing.
+        On failure, :attr:`last_error` / :attr:`last_status` are updated and
+        :class:`CheckpointPublishError` is raised (HTTP layer maps it to a non-2xx response).
         """
         if not files:
             raise CheckpointIntakeError("checkpoint upload must contain at least one file")
         names = tuple(sorted(files))
         resolved_revision = revision or revision_for(submission_id, attempt, names)
-        published = await asyncio.to_thread(
-            self._publish_files,
-            submission_id=submission_id,
-            attempt=attempt,
-            names=names,
-            files=files,
-            revision=resolved_revision,
-        )
+        repo_id = getattr(self.publisher, "repo_id", "") or ""
+        try:
+            published = await asyncio.to_thread(
+                self._publish_files,
+                submission_id=submission_id,
+                attempt=attempt,
+                names=names,
+                files=files,
+                revision=resolved_revision,
+            )
+        except CheckpointIntakeError:
+            raise
+        except Exception as exc:
+            err = _safe_error_message(exc)
+            result = CheckpointIntakeResult(
+                status="failed",
+                submission_id=submission_id,
+                repo_id=repo_id,
+                checkpoint_ref=None,
+                revision=resolved_revision,
+                files=names,
+                last_error=err,
+            )
+            self._record_outcome(result)
+            logger.error(
+                "checkpoint publish failed submission_id=%s repo_id=%s error=%s",
+                submission_id,
+                repo_id,
+                err,
+            )
+            raise CheckpointPublishError(err, submission_id=submission_id, repo_id=repo_id) from exc
+
         await self.repository.record_published_checkpoint(
             submission_id=submission_id,
             attempt=attempt,
             validator_hotkey=validator_hotkey,
             checkpoint_ref=published.checkpoint_ref,
             arch_hash=arch_hash,
         )
+        result = CheckpointIntakeResult(
+            status="success",
+            submission_id=submission_id,
+            repo_id=published.repo_id,
+            checkpoint_ref=published.checkpoint_ref,
+            revision=published.revision,
+            files=tuple(published.files),
+            last_error=None,
+        )
+        self._record_outcome(result)
+        logger.info(
+            "checkpoint publish success submission_id=%s repo_id=%s checkpoint_ref=%s",
+            submission_id,
+            published.repo_id,
+            published.checkpoint_ref,
+        )
         return published
 
+    def _record_outcome(self, result: CheckpointIntakeResult) -> None:
+        self.last_result = result
+        self.last_status = result.status
+        self.last_error = result.last_error
+        self.last_checkpoint_ref = result.checkpoint_ref
+
     def _publish_files(
         self,
         *,
@@ -113,3 +215,17 @@ def _publish_files(
                 revision=revision,
             )
             return self.publisher.publish(upload)
+
+
+def _safe_error_message(exc: BaseException) -> str:
+    """Human-readable error without secret-shaped values (tokens never logged)."""
+    name = type(exc).__name__
+    text = str(exc).strip() or name
+    # Hard-cap length; strip common secret-bearing substrings if a caller ever embeds them.
+    lowered = text.lower()
+    for marker in ("hf_token", "authorization:", "bearer ", "api_key=", "token="):
+        if marker in lowered:
+            return f"{name}: <redacted>"
+    if len(text) > 500:
+        text = text[:500] + "…"
+    return f"{name}: {text}" if name not in text else text

packages/challenges/prism/src/prism_challenge/evaluator/checkpoint_publisher.py

@@ -21,7 +21,7 @@
 
 from .checkpoints import resolve_checkpoint_artifact_path
 
-DEFAULT_CHECKPOINT_REPO_ID = "baseintelligence/prism-checkpoints"
+DEFAULT_CHECKPOINT_REPO_ID = "BaseIntelligence/top-prism-architecture"
 
 
 @dataclass(frozen=True)
@@ -115,6 +115,53 @@ def call_count(self) -> int:
         return len(self.uploads)
 
 
+@dataclass
+class DisabledCheckpointPublisher:
+    """No-op publisher: Prism HF checkpoint upload is intentionally OFF on this host.
+
+    publish() records the call but never contacts HuggingFace. download() raises so
+    resume-from-public-checkpoint is unavailable while upload is disabled.
+    """
+
+    repo_id: str = DEFAULT_CHECKPOINT_REPO_ID
+    uploads: list = field(default_factory=list)
+    published: list = field(default_factory=list)
+
+    def publish(self, upload: CheckpointUpload) -> PublishedCheckpoint:
+        self.uploads.append(upload)
+        result = PublishedCheckpoint(
+            checkpoint_ref=checkpoint_ref_for(self.repo_id, upload.revision),
+            repo_id=self.repo_id,
+            revision=upload.revision,
+            files=tuple(upload.files),
+        )
+        self.published.append(result)
+        return result
+
+    def download(self, checkpoint_ref: str, dest_dir: Path) -> Path:
+        raise RuntimeError(
+            "prism_checkpoint_upload_disabled: download unavailable while "
+            "PRISM_CHECKPOINT_UPLOAD_ENABLED is false"
+        )
+
+
+def publisher_from_env(
+    *,
+    repo_id: str | None = None,
+    token: str | None = None,
+) -> CheckpointPublisher:
+    """Return HF publisher only when upload is explicitly enabled; else disabled no-op."""
+    import os
+
+    enabled = (os.environ.get("PRISM_CHECKPOINT_UPLOAD_ENABLED") or "false").strip().lower()
+    if enabled in ("1", "true", "yes", "on"):
+        return HuggingFaceCheckpointPublisher(
+            repo_id=repo_id or DEFAULT_CHECKPOINT_REPO_ID,
+            token=token,
+        )
+    return DisabledCheckpointPublisher(repo_id=repo_id or DEFAULT_CHECKPOINT_REPO_ID)
+
+
 class HuggingFaceCheckpointPublisher:
     """Deploy-time publisher backed by ``huggingface_hub`` (imported lazily; mocked in tests).
 
@@ -143,9 +190,17 @@ def _hf_api(self) -> Any:
         return self._api
 
     def publish(self, upload: CheckpointUpload) -> PublishedCheckpoint:
+        import os
+
+        enabled = (os.environ.get("PRISM_CHECKPOINT_UPLOAD_ENABLED") or "false").strip().lower()
+        if enabled not in ("1", "true", "yes", "on"):
+            raise RuntimeError(
+                "prism_checkpoint_upload_disabled: refusing HuggingFace upload "
+                "(set PRISM_CHECKPOINT_UPLOAD_ENABLED=true to re-enable)"
+            )
         files = _read_checkpoint_files(upload)
         api = self._hf_api()
-        api.create_repo(repo_id=self.repo_id, repo_type="model", exist_ok=True, private=True)
+        api.create_repo(repo_id=self.repo_id, repo_type="model", exist_ok=True, private=False)
         for name in files:
             source = resolve_checkpoint_artifact_path(upload.checkpoint_dir, name)
             api.upload_file(