fix(raw-weights): rebuild expired pending and surface rejection reasons

packages/challenges/agent-challenge/src/agent_challenge/evaluation/raw_weight_push.py

@@ -92,6 +92,56 @@ class PushAttemptResult:
     error: str | None = None
 
 
+def _as_utc(value: datetime) -> datetime:
+    if value.tzinfo is None:
+        return value.replace(tzinfo=UTC)
+    return value.astimezone(UTC)
+
+
+def _pending_still_fresh(
+    payload: RawWeightPushRequest,
+    *,
+    now: datetime,
+    skew_seconds: int = 30,
+) -> bool:
+    """Mirror master receipt policy lower/upper bound for local pending reuse."""
+
+    expires = _as_utc(payload.expires_at)
+    receipt = _as_utc(now)
+    return receipt < (expires + timedelta(seconds=int(skew_seconds)))
+
+
+def _safe_response_detail(response: httpx.Response) -> str:
+    """Extract a short rejection detail without logging secrets or full bodies."""
+
+    try:
+        data = response.json()
+    except Exception:  # noqa: BLE001
+        return ""
+    if not isinstance(data, dict):
+        return ""
+    detail = data.get("detail")
+    if isinstance(detail, str):
+        return detail[:300]
+    if isinstance(detail, list):
+        # FastAPI validation error shape — field locs only.
+        parts: list[str] = []
+        for item in detail[:5]:
+            if isinstance(item, dict):
+                loc = ".".join(str(x) for x in item.get("loc", ()))
+                if loc:
+                    parts.append(loc)
+        return ",".join(parts)[:300]
+    return ""
+
+
+def _is_all_zero_weights(weights: Mapping[str, float]) -> bool:
+    return bool(weights) and all(float(value) <= 0.0 for value in weights.values())
+
+
+NON_RETRYABLE_PUSH_STATUSES = frozenset({409, 413, 415, 422})
+
+
 class RawWeightPushLedger(Base):
     """Singleton SQLAlchemy row for durable attempt/ack checkpointing.
 
@@ -196,6 +246,22 @@ async def record_pending(
             row.updated_at = attempted_at
             await session.commit()
 
+    async def clear_pending(self) -> None:
+        """Drop a non-retryable pending snapshot so the next push rebuilds fresh bytes."""
+
+        async with self.database.session() as session:
+            row = await self._get_row(session)
+            if row is None:
+                return
+            row.pending_epoch = None
+            row.pending_revision = None
+            row.pending_payload_digest = None
+            row.pending_canonical_payload = None
+            row.pending_nonce = None
+            row.pending_attempted_at = None
+            row.updated_at = datetime.now(UTC).isoformat()
+            await session.commit()
+
     async def acknowledge(
         self,
         *,
@@ -315,6 +381,8 @@ def _build_payload(
         }
         if not body["weights"]:
             raise ValueError("empty weight map has no push surface")
+        if _is_all_zero_weights(body["weights"]):
+            raise ValueError("all-zero weight map has no push surface")
         digest = RawWeightPushRequest.compute_digest(body)
         body["payload_digest"] = digest
         payload = RawWeightPushRequest.model_validate(body)
@@ -375,6 +443,19 @@ async def push_once(
                 pending_bytes = str(pending["canonical_payload"]).encode("utf-8")
                 payload = RawWeightPushRequest.model_validate_json(pending_bytes)
                 raw_bytes = payload.canonical_bytes()
+                if not _pending_still_fresh(payload, now=now):
+                    logger.info(
+                        "raw weight pending expired; rebuilding",
+                        extra={
+                            "epoch": payload.epoch,
+                            "revision": payload.revision,
+                            "digest": payload.payload_digest[:12],
+                        },
+                    )
+                    await self.store.clear_pending()
+                    pending = None
+                    payload = None
+                    raw_bytes = None
             except Exception:  # noqa: BLE001 - corrupt pending is rebuilt
                 pending = None
                 payload = None
@@ -402,6 +483,16 @@ async def push_once(
                     cursor_advanced=False,
                     error="empty weights",
                 )
+            if _is_all_zero_weights(cleaned):
+                return PushAttemptResult(
+                    status="skipped_zero",
+                    epoch=0,
+                    revision=0,
+                    payload_digest="",
+                    snapshot_id=None,
+                    cursor_advanced=False,
+                    error="all-zero weight map",
+                )
             # Host-trust path: never silently push unattested scores as if TEE-backed.
             _log_unattested_weight_push_if_no_phala(cleaned)
             resolved_epoch = (
@@ -477,14 +568,30 @@ async def push_once(
                 error=f"status={response.status_code}",
             )
         if response.status_code not in {200, 201}:
+            detail = _safe_response_detail(response)
+            logger.info(
+                "raw weight push rejected",
+                extra={
+                    "status": response.status_code,
+                    "detail": detail,
+                    "epoch": payload.epoch,
+                    "revision": payload.revision,
+                    "digest": payload.payload_digest[:12],
+                },
+            )
+            if response.status_code in NON_RETRYABLE_PUSH_STATUSES:
+                await self.store.clear_pending()
+            err = f"status={response.status_code}"
+            if detail:
+                err = f"{err} detail={detail}"
             return PushAttemptResult(
                 status="rejected",
                 epoch=payload.epoch,
                 revision=payload.revision,
                 payload_digest=payload.payload_digest,
                 snapshot_id=None,
                 cursor_advanced=False,
-                error=f"status={response.status_code}",
+                error=err,
             )
         try:
             ack = RawWeightPushAcknowledgement.model_validate(response.json())

packages/challenges/agent-challenge/tests/test_ac_raw_weight_push.py

@@ -277,3 +277,99 @@ def handler(request: httpx.Request) -> httpx.Response:
     for record in caplog.records:
         assert TOKEN not in str(record.__dict__)
     await http.aclose()
+
+
+@pytest.mark.asyncio
+async def test_expired_pending_is_rebuilt_not_retried(database: Database) -> None:
+    clock = FakeClock()
+    seen: list[str] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        parsed = RawWeightPushRequest.model_validate_json(request.content)
+        seen.append(parsed.nonce)
+        return httpx.Response(
+            200,
+            json={
+                "protocol_version": "1.0",
+                "challenge_slug": SLUG,
+                "epoch": parsed.epoch,
+                "revision": parsed.revision,
+                "snapshot_id": "snap-ac-rebuilt",
+                "payload_digest": parsed.payload_digest,
+                "accepted": True,
+                "idempotent": False,
+            },
+        )
+
+    http = httpx.AsyncClient(
+        transport=httpx.MockTransport(handler),
+        base_url="http://master.test",
+    )
+    client = RawWeightPushClient(
+        database=database,
+        challenge_slug=SLUG,
+        master_base_url="http://master.test",
+        shared_token=TOKEN,
+        now_fn=clock.now,
+        http_client=http,
+        epoch_fn=lambda: 99,
+        freshness_seconds=60,
+    )
+    await client.init()
+    old = clock.now() - timedelta(hours=12)
+    stale_payload, stale_raw = client._build_payload(
+        weights={WINNER: 1.0},
+        epoch=1,
+        revision=1,
+        nonce="ac-stale-nonce",
+        now=old,
+    )
+    await client.store.record_pending(
+        epoch=stale_payload.epoch,
+        revision=stale_payload.revision,
+        payload_digest=stale_payload.payload_digest,
+        canonical_payload=stale_raw.decode("utf-8"),
+        nonce=stale_payload.nonce,
+        attempted_at=old.isoformat(),
+    )
+    with activate_role(Role.CHALLENGE):
+        result = await client.push_once(weights={WINNER: 1.0}, epoch=99, reuse_pending=True)
+    assert result.status == "acknowledged"
+    assert len(seen) == 1
+    assert seen[0] != "ac-stale-nonce"
+    assert await client.store.get_pending() is None
+    await http.aclose()
+
+
+@pytest.mark.asyncio
+async def test_all_zero_weights_skipped_locally(database: Database) -> None:
+    clock = FakeClock()
+    calls: list[httpx.Request] = []
+
+    def handler(request: httpx.Request) -> httpx.Response:
+        calls.append(request)
+        return httpx.Response(500)
+
+    http = httpx.AsyncClient(
+        transport=httpx.MockTransport(handler),
+        base_url="http://master.test",
+    )
+    client = RawWeightPushClient(
+        database=database,
+        challenge_slug=SLUG,
+        master_base_url="http://master.test",
+        shared_token=TOKEN,
+        now_fn=clock.now,
+        http_client=http,
+        epoch_fn=lambda: 1,
+    )
+    await client.init()
+    with activate_role(Role.CHALLENGE):
+        result = await client.push_once(
+            weights={"5GziQCcRpN8NCJktX343brnfuVe3w6gUYieeStXPD1Dag2At": 0.0},
+            epoch=1,
+        )
+    assert result.cursor_advanced is False
+    assert result.status in {"skipped_empty", "skipped_zero"}
+    assert calls == []
+    await http.aclose()