From a60d4f65cd9ea2a0d46b5409d5f2dcbb0ba94c2e Mon Sep 17 00:00:00 2001 From: Sonic Shih Date: Sun, 26 Jul 2026 12:37:31 +0800 Subject: [PATCH 1/4] feat(research): persist verified catalog partitions --- .../polymarket-btc-5m.example.json | 2 +- .../polymarket-sol-5m.example.json | 2 +- .../src/polymarket_evidence/catalog.rs | 344 +++++++++- .../src/event_cohort_partition.rs | 593 +++++++++++++++++- .../crates/ploy-research/src/lib.rs | 4 +- .../ploy-research/src/prediction_loop_fs.rs | 40 +- 6 files changed, 971 insertions(+), 14 deletions(-) diff --git a/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json b/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json index aae126720..82b91ad5b 100644 --- a/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json +++ b/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json @@ -15,7 +15,7 @@ "horizon": "5m", "time_cohort_boundary_ms": 0, "prompt_snapshot_id": "sha256:2b55ba0e724dfc9f5a040911e397058a09c6229e42064e4998739d617b368dcb", - "search_policy_snapshot_id": "sha256:61def2b8a6dd911d47dc068282dac61631418332a12cc55bb98131c883b13ab4", + "search_policy_snapshot_id": "sha256:7b59f5f43dcbc4f5956cc9b78ccd5635ef7405578099d85859fe64590e170067", "search_budget": { "max_candidates": 6, "max_llm_calls": 2, diff --git a/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json b/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json index f9a0db56c..c95f60585 100644 --- a/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json +++ b/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json @@ -15,7 +15,7 @@ "horizon": "5m", "time_cohort_boundary_ms": 0, "prompt_snapshot_id": "sha256:0816ebccf4c75ee6bdcfe315b253c84ddf1808a38687ebe3b924b87bd72a52a9", - "search_policy_snapshot_id": "sha256:61def2b8a6dd911d47dc068282dac61631418332a12cc55bb98131c883b13ab4", + "search_policy_snapshot_id": "sha256:7b59f5f43dcbc4f5956cc9b78ccd5635ef7405578099d85859fe64590e170067", "search_budget": { "max_candidates": 6, "max_llm_calls": 2, diff --git a/rust_hft/prediction-markets/crates/ploy-market-data/src/polymarket_evidence/catalog.rs b/rust_hft/prediction-markets/crates/ploy-market-data/src/polymarket_evidence/catalog.rs index e7b3cedb7..33ed5ff98 100644 --- a/rust_hft/prediction-markets/crates/ploy-market-data/src/polymarket_evidence/catalog.rs +++ b/rust_hft/prediction-markets/crates/ploy-market-data/src/polymarket_evidence/catalog.rs @@ -7,6 +7,7 @@ use super::{ use anyhow::{bail, ensure, Result}; use chrono::{DateTime, Utc}; use serde::{Deserialize, Serialize}; +use serde_json::Value; use sha2::{Digest, Sha256}; use std::collections::BTreeMap; use std::path::Path; @@ -16,7 +17,8 @@ const BTC_5M_SECS: i64 = 300; /// Immutable identity of the verifier that classified an evidence receipt. /// Paths and mutable version strings are deliberately not accepted as identities. -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct PolymarketCatalogVerifier { source_sha256: String, binary_sha256: String, @@ -50,13 +52,13 @@ impl PolymarketCatalogVerifier { } } -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum PolymarketResearchTask { Btc5mBacktest, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum PolymarketCatalogReceiptState { Ready, @@ -64,7 +66,7 @@ pub enum PolymarketCatalogReceiptState { Rejected, } -#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize)] +#[derive(Debug, Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)] #[serde(rename_all = "snake_case")] pub enum PolymarketCatalogReason { EvidenceVerificationFailed, @@ -75,7 +77,8 @@ pub enum PolymarketCatalogReason { UnsupportedProduct, } -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct PolymarketEvidenceAvailability { pub contract: Option>, pub books: Option>, @@ -215,6 +218,262 @@ impl PolymarketReadyEventCatalog { pub fn receipts(&self) -> impl Iterator { self.receipts.values() } + + /// Restore only #319's authenticated Ready receipts. The caller supplies + /// JSON values so this module can retain control of the receipt identity + /// reconstruction and re-hash every persisted record before admission. + pub fn from_persisted_ready_receipts(receipts: Vec) -> Result { + let mut catalog = Self::default(); + for value in receipts { + let persisted: PersistedReadyCatalogReceipt = + serde_json::from_value(value).map_err(|error| { + anyhow::anyhow!("parse persisted ready catalog receipt: {error}") + })?; + let receipt = persisted.into_receipt()?; + let digest = receipt.receipt_sha256.clone(); + ensure!( + !catalog.receipts.contains_key(&digest), + "persisted ready catalog contains duplicate receipt_sha256 {digest}" + ); + ensure!( + !catalog + .receipts + .values() + .any(|existing| existing.market_id == receipt.market_id), + "persisted ready catalog contains duplicate market_id {}", + receipt.market_id + ); + catalog.receipts.insert(digest, receipt); + } + Ok(catalog) + } + + /// Bound variable-length Ready receipt fields before a caller serializes an + /// artifact. This preserves the caller's fixed byte budget without first + /// allocating a canonical representation of unbounded producer text. + pub fn validate_ready_artifact_bounds( + &self, + max_entries: usize, + max_text_bytes: usize, + ) -> Result<()> { + let mut ready_count = 0_usize; + for receipt in self.receipts.values() { + if receipt.state != PolymarketCatalogReceiptState::Ready { + continue; + } + ready_count = ready_count + .checked_add(1) + .ok_or_else(|| anyhow::anyhow!("Ready catalog count overflow"))?; + ensure!( + ready_count <= max_entries, + "ready-event catalog exceeds the bounded entry count" + ); + for (field, value) in [ + ("market_id", receipt.market_id.as_str()), + ( + "up_token_id", + receipt.up_token_id.as_deref().unwrap_or_default(), + ), + ( + "down_token_id", + receipt.down_token_id.as_deref().unwrap_or_default(), + ), + ] { + ensure!( + value.len() <= max_text_bytes, + "ready catalog {field} exceeds {max_text_bytes} bytes" + ); + } + } + Ok(()) + } +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedReadyCatalogReceipt { + receipt_sha256: String, + market_id: String, + content_sha256: String, + manifest_sha256: String, + qualification_sha256: String, + success_sha256: Option, + verifier: PolymarketCatalogVerifier, + event_start: Option>, + event_end: Option>, + up_token_id: Option, + down_token_id: Option, + sequence: Option, + coverage: Option, + trade_completion: Option, + availability: Option, + state: PolymarketCatalogReceiptState, + reasons: Vec, + supported_tasks: Vec, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedEvidenceSequence { + start: u64, + end: u64, + gaps: u64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedSurfaceCoverage { + up_book: u64, + down_book: u64, + trades: u64, + reference: u64, + settlement: u64, +} + +#[derive(Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedTradeCompletion { + trade_count: u64, + trade_record_ids_sha256: String, +} + +impl PersistedReadyCatalogReceipt { + fn into_receipt(self) -> Result { + ensure!( + self.state == PolymarketCatalogReceiptState::Ready, + "persisted catalog receipt {} is not Ready", + self.receipt_sha256 + ); + ensure!( + self.reasons.is_empty(), + "persisted Ready catalog receipt {} has rejection reasons", + self.receipt_sha256 + ); + ensure!( + self.supported_tasks == [PolymarketResearchTask::Btc5mBacktest], + "persisted Ready catalog receipt {} has unsupported tasks", + self.receipt_sha256 + ); + ensure!(!self.market_id.trim().is_empty() && self.market_id.trim() == self.market_id); + for (label, digest) in [ + ("receipt", &self.receipt_sha256), + ("content", &self.content_sha256), + ("manifest", &self.manifest_sha256), + ("qualification", &self.qualification_sha256), + ] { + ensure!(is_sha256(digest), "persisted {label} digest is invalid"); + } + let success_sha256 = self + .success_sha256 + .as_deref() + .ok_or_else(|| anyhow::anyhow!("persisted Ready receipt is missing success_sha256"))?; + ensure!( + is_sha256(success_sha256), + "persisted success digest is invalid" + ); + let verifier = PolymarketCatalogVerifier::new( + self.verifier.source_sha256, + self.verifier.binary_sha256, + self.verifier.configuration_sha256, + self.verifier.policy_sha256, + )?; + let event_start = self + .event_start + .ok_or_else(|| anyhow::anyhow!("persisted Ready receipt is missing event_start"))?; + let event_end = self + .event_end + .ok_or_else(|| anyhow::anyhow!("persisted Ready receipt is missing event_end"))?; + ensure!( + event_start < event_end, + "persisted Ready receipt has an invalid event window" + ); + let up_token_id = self + .up_token_id + .filter(|value| !value.is_empty()) + .ok_or_else(|| anyhow::anyhow!("persisted Ready receipt is missing Up token"))?; + let down_token_id = self + .down_token_id + .filter(|value| !value.is_empty()) + .ok_or_else(|| anyhow::anyhow!("persisted Ready receipt is missing Down token"))?; + ensure!( + up_token_id != down_token_id, + "persisted Ready receipt has identical tokens" + ); + let sequence = self + .sequence + .ok_or_else(|| anyhow::anyhow!("persisted Ready receipt is missing sequence"))?; + ensure!( + sequence.start <= sequence.end && sequence.gaps == 0, + "persisted Ready receipt has invalid sequence" + ); + let coverage = self + .coverage + .ok_or_else(|| anyhow::anyhow!("persisted Ready receipt is missing coverage"))?; + ensure!( + coverage.up_book > 0 + && coverage.down_book > 0 + && coverage.trades > 0 + && coverage.reference > 0 + && coverage.settlement > 0, + "persisted Ready receipt has incomplete coverage" + ); + let trade_completion = self.trade_completion.ok_or_else(|| { + anyhow::anyhow!("persisted Ready receipt is missing trade completion") + })?; + ensure!( + trade_completion.trade_count > 0 + && is_sha256(&trade_completion.trade_record_ids_sha256), + "persisted Ready receipt has invalid trade completion" + ); + let availability = self + .availability + .ok_or_else(|| anyhow::anyhow!("persisted Ready receipt is missing availability"))?; + let settlement = availability.settlement.ok_or_else(|| { + anyhow::anyhow!("persisted Ready receipt is missing settlement availability") + })?; + ensure!( + settlement >= event_end, + "persisted Ready receipt settlement predates event end" + ); + let receipt = PolymarketCatalogReceipt { + receipt_sha256: self.receipt_sha256, + market_id: self.market_id, + content_sha256: self.content_sha256, + manifest_sha256: self.manifest_sha256, + qualification_sha256: self.qualification_sha256, + success_sha256: self.success_sha256, + verifier, + event_start: Some(event_start), + event_end: Some(event_end), + up_token_id: Some(up_token_id), + down_token_id: Some(down_token_id), + sequence: Some(PolymarketEvidenceSequence { + start: sequence.start, + end: sequence.end, + gaps: sequence.gaps, + }), + coverage: Some(PolymarketCandidateSurfaceCoverage { + up_book: coverage.up_book, + down_book: coverage.down_book, + trades: coverage.trades, + reference: coverage.reference, + settlement: coverage.settlement, + }), + trade_completion: Some(PolymarketEvidenceTradeCompletion { + trade_count: trade_completion.trade_count, + trade_record_ids_sha256: trade_completion.trade_record_ids_sha256, + }), + availability: Some(availability), + state: self.state, + reasons: self.reasons, + supported_tasks: self.supported_tasks, + }; + ensure!( + receipt_digest(&receipt)? == receipt.receipt_sha256, + "persisted Ready receipt digest does not match its canonical content" + ); + Ok(receipt) + } } #[derive(Deserialize)] @@ -879,4 +1138,79 @@ mod tests { 1 ); } + + #[test] + fn persisted_ready_catalog_rejects_non_ready_extra_and_rehashed_receipts() { + let rows = verified_tests::valid_rows(); + let (_temp, triplet) = verified_tests::candidate_triplet(&rows); + let ready_path = triplet + .data + .parent() + .unwrap() + .join("qualification-ready.json"); + let ready_anchor = qualification(&ready_path, &triplet, "BTCUSDT", "up-token"); + let mut catalog = PolymarketReadyEventCatalog::default(); + catalog + .verify_and_append( + "market-1", + &triplet, + &tests::trust(&triplet), + &ready_path, + &ready_anchor, + PolymarketCatalogVerifier::new( + "d".repeat(64), + "e".repeat(64), + "f".repeat(64), + "a".repeat(64), + ) + .unwrap(), + ) + .unwrap(); + let receipt = serde_json::to_value(catalog.receipts().next().unwrap()).unwrap(); + assert_eq!( + PolymarketReadyEventCatalog::from_persisted_ready_receipts(vec![receipt.clone()]) + .unwrap() + .receipts() + .count(), + 1 + ); + + let mut non_ready = receipt.clone(); + non_ready["state"] = json!("partial"); + assert!( + PolymarketReadyEventCatalog::from_persisted_ready_receipts(vec![non_ready]) + .expect_err("non-Ready receipts cannot enter persisted ready catalog") + .to_string() + .contains("not Ready") + ); + + let mut extra = receipt.clone(); + extra["unexpected"] = json!(true); + assert!( + PolymarketReadyEventCatalog::from_persisted_ready_receipts(vec![extra]) + .expect_err("extra receipt fields must fail closed") + .to_string() + .contains("unknown field") + ); + + let mut rehashed = receipt; + rehashed["market_id"] = json!("market-rewritten"); + assert!( + PolymarketReadyEventCatalog::from_persisted_ready_receipts(vec![rehashed]) + .expect_err("receipt content cannot move behind a fixed receipt identity") + .to_string() + .contains("digest does not match") + ); + + let mut oversized_receipt = catalog.receipts().next().unwrap().clone(); + oversized_receipt.market_id = "x".repeat(33); + oversized_receipt.receipt_sha256 = receipt_digest(&oversized_receipt).unwrap(); + let mut oversized_catalog = PolymarketReadyEventCatalog::default(); + oversized_catalog.append(oversized_receipt).unwrap(); + assert!(oversized_catalog + .validate_ready_artifact_bounds(1, 32) + .expect_err("artifact serialization must reject oversized producer text") + .to_string() + .contains("market_id exceeds")); + } } diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs b/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs index 1becfc56d..587fcdb04 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs @@ -1,18 +1,28 @@ -use std::collections::BTreeSet; +use std::{collections::BTreeSet, path::Path}; use ploy_market_data::polymarket_evidence::{ PolymarketCatalogReceipt, PolymarketCatalogReceiptState, PolymarketReadyEventCatalog, }; -use serde::Serialize; +use serde::{Deserialize, Serialize}; +use serde_json::Value; use crate::{ prediction_loop::{current_prediction_policy_snapshot_id, validate_sha256_id}, - prediction_loop_fs::{canonical_json_bytes, sha256_hex}, + prediction_loop_fs::{ + canonical_json_bytes, read_verified_artifact_bounded, sha256_hex, + write_content_addressed_json, ArtifactRef, + }, }; pub const EVENT_COHORT_PARTITION_VERSION: &str = "event_cohort_partition.v3"; +pub const CATALOG_PARTITION_ARTIFACT_VERSION: &str = "catalog_partition_artifact.v1"; +const MAX_CATALOG_PARTITION_ARTIFACT_BYTES: usize = 8 * 1024 * 1024; +const MAX_CATALOG_PARTITION_ARTIFACT_PATH_BYTES: usize = 1_024; +const MAX_READY_CATALOG_ENTRIES: usize = 512; +const MAX_READY_CATALOG_TEXT_BYTES: usize = 2_048; -#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] pub struct EventCohortReadyEntry { receipt_sha256: String, market_id: String, @@ -21,6 +31,48 @@ pub struct EventCohortReadyEntry { settlement_available_at_ms: i64, } +/// Immutable path-and-digest reference to the canonical artifact. The path is +/// only a locator; readback verifies that its filename and bytes match this ID. +#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +pub struct CatalogPartitionArtifactRef { + path: String, + artifact_sha256: String, + payload_sha256: String, +} + +impl CatalogPartitionArtifactRef { + pub fn path(&self) -> &str { + &self.path + } + + pub fn artifact_sha256(&self) -> &str { + &self.artifact_sha256 + } + + pub fn payload_sha256(&self) -> &str { + &self.payload_sha256 + } +} + +/// The only successful #365 readback. Its fields have already passed fresh +/// canonical byte, receipt, partition, policy, and membership validation. +#[derive(Debug)] +pub struct ValidatedCatalogPartitionArtifact { + catalog: PolymarketReadyEventCatalog, + partition: EventCohortPartition, +} + +impl ValidatedCatalogPartitionArtifact { + pub fn catalog(&self) -> &PolymarketReadyEventCatalog { + &self.catalog + } + + pub fn partition(&self) -> &EventCohortPartition { + &self.partition + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct EventCohortPartition { schema_version: &'static str, @@ -46,6 +98,346 @@ struct EventCohortPartitionPayload<'a> { held_out_market_ids: &'a [String], } +#[derive(Serialize)] +struct CatalogPartitionArtifactPayload<'a> { + schema_version: &'static str, + policy_snapshot_id: &'a str, + catalog_receipts: Vec<&'a PolymarketCatalogReceipt>, + partition: PersistedEventCohortPartition, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct CatalogPartitionArtifactEnvelope { + schema_version: String, + payload_sha256: String, + payload: PersistedCatalogPartitionArtifactPayload, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedCatalogPartitionArtifactPayload { + schema_version: String, + policy_snapshot_id: String, + catalog_receipts: Vec, + partition: PersistedEventCohortPartition, +} + +#[derive(Serialize, Deserialize)] +#[serde(deny_unknown_fields)] +struct PersistedEventCohortPartition { + schema_version: String, + ready_entries: Vec, + common_time_boundary_ms: i64, + label_availability_cutoff_ms: i64, + causal_projection_policy_id: String, + train_market_ids: Vec, + crossing_excluded_market_ids: Vec, + held_out_market_ids: Vec, + digest: String, +} + +impl From<&EventCohortPartition> for PersistedEventCohortPartition { + fn from(value: &EventCohortPartition) -> Self { + Self { + schema_version: value.schema_version.to_string(), + ready_entries: value.ready_entries.clone(), + common_time_boundary_ms: value.common_time_boundary_ms, + label_availability_cutoff_ms: value.label_availability_cutoff_ms, + causal_projection_policy_id: value.causal_projection_policy_id.clone(), + train_market_ids: value.train_market_ids.clone(), + crossing_excluded_market_ids: value.crossing_excluded_market_ids.clone(), + held_out_market_ids: value.held_out_market_ids.clone(), + digest: value.digest.clone(), + } + } +} + +impl PersistedEventCohortPartition { + fn into_partition(self) -> Result { + if self.schema_version != EVENT_COHORT_PARTITION_VERSION { + return Err(format!( + "unsupported event cohort partition schema {}", + self.schema_version + )); + } + if self.common_time_boundary_ms <= 0 { + return Err("common time boundary must be positive".into()); + } + validate_sha256_id( + &self.causal_projection_policy_id, + "causal projection policy identity", + )?; + if self.causal_projection_policy_id != current_prediction_policy_snapshot_id() { + return Err("persisted partition has stale causal projection policy identity".into()); + } + validate_sha256_id(&self.digest, "event cohort partition digest")?; + let mut receipt_ids = BTreeSet::new(); + let mut market_ids = BTreeSet::new(); + let mut expected_train = Vec::new(); + let mut expected_crossing = Vec::new(); + let mut expected_held_out = Vec::new(); + for entry in &self.ready_entries { + validate_lower_sha256(&entry.receipt_sha256, "catalog receipt_sha256")?; + validate_market_id(&entry.market_id)?; + if entry.reference_path_start_ms >= entry.reference_path_end_ms { + return Err(format!( + "ready market {} has an invalid reference path", + entry.market_id + )); + } + if entry.settlement_available_at_ms < entry.reference_path_end_ms { + return Err(format!( + "ready market {} settlement label predates event end", + entry.market_id + )); + } + if !receipt_ids.insert(&entry.receipt_sha256) { + return Err(format!( + "persisted partition contains duplicate receipt_sha256 {}", + entry.receipt_sha256 + )); + } + if !market_ids.insert(&entry.market_id) { + return Err(format!( + "persisted partition contains duplicate market_id {}", + entry.market_id + )); + } + if entry.reference_path_end_ms < self.common_time_boundary_ms { + expected_train.push(entry.market_id.clone()); + } else if entry.reference_path_start_ms >= self.common_time_boundary_ms { + expected_held_out.push(entry.market_id.clone()); + } else { + expected_crossing.push(entry.market_id.clone()); + } + } + if self.train_market_ids != expected_train + || self.crossing_excluded_market_ids != expected_crossing + || self.held_out_market_ids != expected_held_out + { + return Err("persisted partition assignments do not match its ready entries".into()); + } + let expected_label_cutoff = self + .ready_entries + .iter() + .filter(|entry| entry.reference_path_start_ms >= self.common_time_boundary_ms) + .map(|entry| entry.reference_path_start_ms) + .min() + .unwrap_or(self.common_time_boundary_ms); + if self.label_availability_cutoff_ms != expected_label_cutoff { + return Err("persisted partition label availability cutoff is invalid".into()); + } + if self.ready_entries.iter().any(|entry| { + entry.reference_path_end_ms < self.common_time_boundary_ms + && entry.settlement_available_at_ms >= expected_label_cutoff + }) { + return Err( + "persisted partition includes a training label unavailable by cutoff".into(), + ); + } + let payload = EventCohortPartitionPayload { + schema_version: EVENT_COHORT_PARTITION_VERSION, + ready_entries: &self.ready_entries, + common_time_boundary_ms: self.common_time_boundary_ms, + label_availability_cutoff_ms: self.label_availability_cutoff_ms, + causal_projection_policy_id: &self.causal_projection_policy_id, + train_market_ids: &self.train_market_ids, + crossing_excluded_market_ids: &self.crossing_excluded_market_ids, + held_out_market_ids: &self.held_out_market_ids, + }; + let digest = format!("sha256:{}", sha256_hex(&canonical_json_bytes(&payload)?)); + if digest != self.digest { + return Err("persisted partition digest does not match canonical content".into()); + } + Ok(EventCohortPartition { + schema_version: EVENT_COHORT_PARTITION_VERSION, + ready_entries: self.ready_entries, + common_time_boundary_ms: self.common_time_boundary_ms, + label_availability_cutoff_ms: self.label_availability_cutoff_ms, + causal_projection_policy_id: self.causal_projection_policy_id, + train_market_ids: self.train_market_ids, + crossing_excluded_market_ids: self.crossing_excluded_market_ids, + held_out_market_ids: self.held_out_market_ids, + digest: self.digest, + }) + } +} + +/// Persist #319's already-authenticated Ready catalog and #322's already +/// derived partition. This never constructs or re-splits a partition. +pub fn write_catalog_partition_artifact( + output_root: &Path, + directory: &Path, + catalog: &PolymarketReadyEventCatalog, + partition: &EventCohortPartition, +) -> Result { + validate_catalog_partition_write_bounds(catalog, partition)?; + validate_catalog_partition_membership(catalog, partition)?; + let payload = CatalogPartitionArtifactPayload { + schema_version: CATALOG_PARTITION_ARTIFACT_VERSION, + policy_snapshot_id: partition.causal_projection_policy_id(), + catalog_receipts: catalog + .receipts() + .filter(|receipt| receipt.state == PolymarketCatalogReceiptState::Ready) + .collect(), + partition: PersistedEventCohortPartition::from(partition), + }; + let payload_sha256 = format!("sha256:{}", sha256_hex(&canonical_json_bytes(&payload)?)); + let envelope = CatalogPartitionArtifactEnvelope { + schema_version: CATALOG_PARTITION_ARTIFACT_VERSION.to_string(), + payload_sha256: payload_sha256.clone(), + payload: PersistedCatalogPartitionArtifactPayload { + schema_version: payload.schema_version.to_string(), + policy_snapshot_id: payload.policy_snapshot_id.to_string(), + catalog_receipts: payload + .catalog_receipts + .into_iter() + .map(serde_json::to_value) + .collect::, _>>() + .map_err(|error| format!("serialize catalog receipt: {error}"))?, + partition: payload.partition, + }, + }; + let bytes = canonical_json_bytes(&envelope)?; + if bytes.len() > MAX_CATALOG_PARTITION_ARTIFACT_BYTES { + return Err(format!( + "catalog partition artifact exceeds {MAX_CATALOG_PARTITION_ARTIFACT_BYTES} bytes" + )); + } + let artifact = + write_content_addressed_json(output_root, directory, "catalog-partition", &envelope)?; + Ok(CatalogPartitionArtifactRef { + path: artifact.path, + artifact_sha256: format!("sha256:{}", artifact.sha256), + payload_sha256, + }) +} + +/// Freshly verify a bounded, canonical artifact before returning usable inputs. +pub fn read_catalog_partition_artifact( + output_root: &Path, + artifact: &CatalogPartitionArtifactRef, +) -> Result { + if artifact.path.len() > MAX_CATALOG_PARTITION_ARTIFACT_PATH_BYTES { + return Err("catalog partition artifact path exceeds the bounded length".into()); + } + validate_sha256_id( + &artifact.artifact_sha256, + "catalog partition artifact identity", + )?; + validate_sha256_id( + &artifact.payload_sha256, + "catalog partition payload identity", + )?; + let raw_digest = artifact + .artifact_sha256 + .strip_prefix("sha256:") + .expect("validate_sha256_id accepts only sha256 IDs"); + let bytes = read_verified_artifact_bounded( + output_root, + &ArtifactRef { + path: artifact.path.clone(), + sha256: raw_digest.to_string(), + }, + MAX_CATALOG_PARTITION_ARTIFACT_BYTES, + )?; + let envelope: CatalogPartitionArtifactEnvelope = serde_json::from_slice(&bytes) + .map_err(|error| format!("parse catalog partition artifact: {error}"))?; + if canonical_json_bytes(&envelope)? != bytes { + return Err("catalog partition artifact is not canonical JSON".into()); + } + if envelope.schema_version != CATALOG_PARTITION_ARTIFACT_VERSION + || envelope.payload.schema_version != CATALOG_PARTITION_ARTIFACT_VERSION + { + return Err("unsupported catalog partition artifact schema".into()); + } + if envelope.payload_sha256 != artifact.payload_sha256 { + return Err("catalog partition artifact payload identity mismatches reference".into()); + } + let expected_payload_sha256 = format!( + "sha256:{}", + sha256_hex(&canonical_json_bytes(&envelope.payload)?) + ); + if expected_payload_sha256 != envelope.payload_sha256 { + return Err( + "catalog partition artifact payload digest does not match canonical content".into(), + ); + } + let PersistedCatalogPartitionArtifactPayload { + schema_version: _, + policy_snapshot_id, + catalog_receipts, + partition, + } = envelope.payload; + let catalog = PolymarketReadyEventCatalog::from_persisted_ready_receipts(catalog_receipts) + .map_err(|error| format!("validate persisted ready catalog: {error}"))?; + let partition = partition.into_partition()?; + if policy_snapshot_id != partition.causal_projection_policy_id() { + return Err("artifact policy identity does not match partition".into()); + } + if policy_snapshot_id != current_prediction_policy_snapshot_id() { + return Err("catalog partition artifact has stale policy identity".into()); + } + validate_catalog_partition_membership(&catalog, &partition)?; + Ok(ValidatedCatalogPartitionArtifact { catalog, partition }) +} + +fn validate_catalog_partition_membership( + catalog: &PolymarketReadyEventCatalog, + partition: &EventCohortPartition, +) -> Result<(), String> { + if partition.causal_projection_policy_id() != current_prediction_policy_snapshot_id() { + return Err("partition has stale causal projection policy identity".into()); + } + let ready = catalog + .receipts() + .filter(|receipt| receipt.state == PolymarketCatalogReceiptState::Ready) + .collect::>(); + if ready.len() != partition.ready_entries.len() { + return Err("partition does not contain every Ready catalog receipt".into()); + } + for (receipt, entry) in ready.into_iter().zip(&partition.ready_entries) { + let settlement = receipt + .availability + .as_ref() + .and_then(|availability| availability.settlement) + .ok_or_else(|| { + format!( + "ready market {} is missing settlement availability", + receipt.market_id + ) + })?; + if receipt.receipt_sha256 != entry.receipt_sha256 + || receipt.market_id != entry.market_id + || receipt.event_start.map(|value| value.timestamp_millis()) + != Some(entry.reference_path_start_ms) + || receipt.event_end.map(|value| value.timestamp_millis()) + != Some(entry.reference_path_end_ms) + || settlement.timestamp_millis() != entry.settlement_available_at_ms + { + return Err(format!( + "partition entry does not match Ready catalog receipt {}", + receipt.market_id + )); + } + } + Ok(()) +} + +fn validate_catalog_partition_write_bounds( + catalog: &PolymarketReadyEventCatalog, + partition: &EventCohortPartition, +) -> Result<(), String> { + catalog + .validate_ready_artifact_bounds(MAX_READY_CATALOG_ENTRIES, MAX_READY_CATALOG_TEXT_BYTES) + .map_err(|error| format!("validate ready catalog artifact bounds: {error}"))?; + if partition.ready_entries.len() > MAX_READY_CATALOG_ENTRIES { + return Err("event cohort partition exceeds the bounded Ready entry count".into()); + } + Ok(()) +} + impl EventCohortPartition { /// Derive the common partition from #319's complete ordered Ready catalog. /// No downstream snapshot enters this identity. @@ -282,7 +674,15 @@ mod tests { PolymarketEvidenceAvailability, PolymarketReadyEventCatalog, PolymarketResearchTask, }; - use super::EventCohortPartition; + use crate::prediction_loop_fs::{ + canonical_json_bytes, sha256_hex, write_content_addressed_json, + }; + + use super::{ + read_catalog_partition_artifact, write_catalog_partition_artifact, + CatalogPartitionArtifactEnvelope, CatalogPartitionArtifactRef, EventCohortPartition, + EventCohortPartitionPayload, EventCohortReadyEntry, EVENT_COHORT_PARTITION_VERSION, + }; fn ready_receipt( receipt_sha256: char, @@ -324,6 +724,26 @@ mod tests { } } + fn persisted_ready_catalog_fixture() -> PolymarketReadyEventCatalog { + let receipt = serde_json::from_str(r#"{ + "availability":{"books":"2026-07-17T05:30:02Z","contract":"2026-07-17T05:29:59Z","references":"2026-07-17T05:29:57Z","settlement":"2026-07-17T05:35:02Z","trades":"2026-07-17T05:30:05Z"}, + "content_sha256":"7dc38e6a4930c7ec840787fe24eb9256bc297c7243e5fe38177aff4bf2a6fd8c", + "coverage":{"down_book":1,"reference":1,"settlement":1,"trades":1,"up_book":1}, + "down_token_id":"down-token","event_end":"2026-07-17T05:35:00Z","event_start":"2026-07-17T05:30:00Z", + "manifest_sha256":"9fd05772d126b0c7e6f1fbe68595637562166051784f8bc5e0a5b6e8e9b8aef4","market_id":"market-1", + "qualification_sha256":"52233ca3d83364aaeccb3460a4716a095ab645422773751c0a348b111d9db4ce","reasons":[], + "receipt_sha256":"ccc6bdb04b333e261629f34ca0df72cc6ea09b14a27f925c8a7e8c035a3bfa67", + "sequence":{"end":7,"gaps":0,"start":1},"state":"ready", + "success_sha256":"9c483a286640fbc5e213f782ad33a31d83d4ba4c66868b62a8f5fcc39e1e27c0", + "supported_tasks":["btc5m_backtest"], + "trade_completion":{"trade_count":1,"trade_record_ids_sha256":"984aa561be4dd28b3c6638ed6d3369837e828613e8677a1c5a469ef3866f1c5b"}, + "up_token_id":"up-token", + "verifier":{"binary_sha256":"eeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeeee","configuration_sha256":"ffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffffff","policy_sha256":"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa","source_sha256":"dddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddddd"} + }"#).expect("parse verified #319 receipt fixture"); + PolymarketReadyEventCatalog::from_persisted_ready_receipts(vec![receipt]) + .expect("fixture is a verified Ready catalog receipt") + } + #[test] fn ready_catalog_assigns_each_market_once_and_excludes_crossing_reference_paths() { let held_out = ready_receipt('a', "held-out", 2_000, 2_500); @@ -418,6 +838,169 @@ mod tests { assert_eq!(partition.common_time_boundary_ms(), 1_000); } + #[test] + fn persisted_catalog_partition_artifact_round_trips_only_after_content_addressed_readback() { + let root = std::env::temp_dir().join(format!( + "ploy-catalog-partition-artifact-{}-{}", + std::process::id(), + uuid::Uuid::new_v4() + )); + let catalog = PolymarketReadyEventCatalog::default(); + let partition = EventCohortPartition::from_ready_catalog(&catalog, 1_000).unwrap(); + + let artifact = + write_catalog_partition_artifact(&root, &root.join("evidence"), &catalog, &partition) + .expect("persist canonical catalog and partition"); + let restored = read_catalog_partition_artifact(&root, &artifact) + .expect("fresh readback must validate the persisted artifact"); + + assert_eq!(restored.partition().digest(), partition.digest()); + assert_eq!(restored.catalog().receipts().count(), 0); + std::fs::remove_dir_all(root).expect("remove artifact fixture"); + } + + #[test] + fn persisted_catalog_partition_artifact_round_trips_a_verified_ready_receipt() { + let root = std::env::temp_dir().join(format!( + "ploy-catalog-partition-artifact-ready-{}-{}", + std::process::id(), + uuid::Uuid::new_v4() + )); + let catalog = persisted_ready_catalog_fixture(); + let end = catalog + .receipts() + .next() + .unwrap() + .event_end + .unwrap() + .timestamp_millis(); + let partition = EventCohortPartition::from_ready_catalog(&catalog, end + 3_000).unwrap(); + let artifact = + write_catalog_partition_artifact(&root, &root.join("evidence"), &catalog, &partition) + .unwrap(); + + let restored = read_catalog_partition_artifact(&root, &artifact).unwrap(); + let receipt = restored.catalog().receipts().next().unwrap(); + assert_eq!( + receipt.receipt_sha256, + "ccc6bdb04b333e261629f34ca0df72cc6ea09b14a27f925c8a7e8c035a3bfa67" + ); + assert_eq!(receipt.market_id, "market-1"); + assert_eq!( + restored.partition().ready_entries()[0].receipt_sha256(), + receipt.receipt_sha256 + ); + assert_eq!(restored.partition().digest(), partition.digest()); + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn persisted_catalog_partition_artifact_rejects_corruption_mutable_paths_and_missing_partition() + { + let root = std::env::temp_dir().join(format!( + "ploy-catalog-partition-artifact-rejection-{}-{}", + std::process::id(), + uuid::Uuid::new_v4() + )); + let catalog = PolymarketReadyEventCatalog::default(); + let partition = EventCohortPartition::from_ready_catalog(&catalog, 1_000).unwrap(); + let artifact = + write_catalog_partition_artifact(&root, &root.join("evidence"), &catalog, &partition) + .unwrap(); + + let mut mutable_path = artifact.clone(); + mutable_path.path = "mutable.json".to_string(); + assert!(read_catalog_partition_artifact(&root, &mutable_path).is_err()); + + let mut missing_partition: serde_json::Value = + serde_json::from_slice(&std::fs::read(root.join(artifact.path())).unwrap()).unwrap(); + missing_partition["payload"] + .as_object_mut() + .unwrap() + .remove("partition"); + let missing = write_content_addressed_json( + &root, + &root.join("evidence"), + "catalog-partition", + &missing_partition, + ) + .unwrap(); + let missing_ref = CatalogPartitionArtifactRef { + path: missing.path, + artifact_sha256: format!("sha256:{}", missing.sha256), + payload_sha256: artifact.payload_sha256.clone(), + }; + assert!(read_catalog_partition_artifact(&root, &missing_ref) + .expect_err("missing partition must fail closed") + .contains("missing field `partition`")); + + std::fs::write(root.join(artifact.path()), b"corrupt").unwrap(); + assert!(read_catalog_partition_artifact(&root, &artifact) + .expect_err("tampered bytes must fail before parse") + .contains("hash mismatch")); + std::fs::remove_dir_all(root).unwrap(); + } + + #[test] + fn persisted_catalog_partition_artifact_rejects_catalog_partition_membership_mismatch() { + let root = std::env::temp_dir().join(format!( + "ploy-catalog-partition-artifact-membership-{}-{}", + std::process::id(), + uuid::Uuid::new_v4() + )); + let catalog = PolymarketReadyEventCatalog::default(); + let partition = EventCohortPartition::from_ready_catalog(&catalog, 1_000).unwrap(); + let artifact = + write_catalog_partition_artifact(&root, &root.join("evidence"), &catalog, &partition) + .unwrap(); + let mut envelope: CatalogPartitionArtifactEnvelope = + serde_json::from_slice(&std::fs::read(root.join(artifact.path())).unwrap()).unwrap(); + let persisted = &mut envelope.payload.partition; + persisted.ready_entries.push(EventCohortReadyEntry { + receipt_sha256: "a".repeat(64), + market_id: "missing-from-catalog".to_string(), + reference_path_start_ms: 2_000, + reference_path_end_ms: 2_500, + settlement_available_at_ms: 2_500, + }); + persisted.label_availability_cutoff_ms = 2_000; + persisted.held_out_market_ids = vec!["missing-from-catalog".to_string()]; + let partition_payload = EventCohortPartitionPayload { + schema_version: EVENT_COHORT_PARTITION_VERSION, + ready_entries: &persisted.ready_entries, + common_time_boundary_ms: persisted.common_time_boundary_ms, + label_availability_cutoff_ms: persisted.label_availability_cutoff_ms, + causal_projection_policy_id: &persisted.causal_projection_policy_id, + train_market_ids: &persisted.train_market_ids, + crossing_excluded_market_ids: &persisted.crossing_excluded_market_ids, + held_out_market_ids: &persisted.held_out_market_ids, + }; + persisted.digest = format!( + "sha256:{}", + sha256_hex(&canonical_json_bytes(&partition_payload).unwrap()) + ); + envelope.payload_sha256 = format!( + "sha256:{}", + sha256_hex(&canonical_json_bytes(&envelope.payload).unwrap()) + ); + let mismatched = write_content_addressed_json( + &root, + &root.join("evidence"), + "catalog-partition", + &envelope, + ) + .unwrap(); + let mismatched_ref = CatalogPartitionArtifactRef { + path: mismatched.path, + artifact_sha256: format!("sha256:{}", mismatched.sha256), + payload_sha256: envelope.payload_sha256, + }; + assert!(read_catalog_partition_artifact(&root, &mismatched_ref) + .expect_err("partition cannot add a receipt absent from the catalog") + .contains("does not contain every Ready catalog receipt")); + std::fs::remove_dir_all(root).unwrap(); + } + #[test] fn training_label_unavailable_by_the_cutoff_rejects_the_partition() { let mut train = ready_receipt('a', "late-label", 100, 999); diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/lib.rs b/rust_hft/prediction-markets/crates/ploy-research/src/lib.rs index b21d7044a..b8cca84e1 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/lib.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/lib.rs @@ -54,7 +54,9 @@ pub use deribit::{ DeribitFeatureLoadResult, }; pub use event_cohort_partition::{ - EventCohortPartition, EventCohortReadyEntry, EVENT_COHORT_PARTITION_VERSION, + read_catalog_partition_artifact, write_catalog_partition_artifact, CatalogPartitionArtifactRef, + EventCohortPartition, EventCohortReadyEntry, ValidatedCatalogPartitionArtifact, + CATALOG_PARTITION_ARTIFACT_VERSION, EVENT_COHORT_PARTITION_VERSION, }; pub use event_ml::{ build_event_ml_strategy_handoff, build_walk_forward_report, canonical_event_ml_architecture, diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop_fs.rs b/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop_fs.rs index 18d5d3225..49f04c4bd 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop_fs.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop_fs.rs @@ -1,5 +1,5 @@ use std::fs::{self, File, OpenOptions}; -use std::io::Write; +use std::io::{Read, Write}; use std::path::{Component, Path, PathBuf}; use fs2::FileExt; @@ -320,6 +320,44 @@ pub(crate) fn verify_artifact( Ok(path) } +/// Read a verified content-addressed artifact without allowing an unbounded +/// allocation from a caller-controlled file. +pub(crate) fn read_verified_artifact_bounded( + output_root: &Path, + artifact: &ArtifactRef, + max_bytes: usize, +) -> Result, String> { + let path = artifact_path(output_root, artifact)?; + reject_symlink_components(output_root, &path)?; + let file = File::open(&path) + .map_err(|error| format!("open referenced evidence {}: {error}", path.display()))?; + let mut body = Vec::with_capacity(max_bytes.saturating_add(1).min(64 * 1024)); + file.take((max_bytes as u64).saturating_add(1)) + .read_to_end(&mut body) + .map_err(|error| format!("read referenced evidence {}: {error}", path.display()))?; + if body.len() > max_bytes { + return Err(format!( + "referenced evidence exceeds {max_bytes} bytes: {}", + path.display() + )); + } + let digest = sha256_hex(&body); + if digest != artifact.sha256 { + return Err(format!("evidence hash mismatch for {}", path.display())); + } + let file_name = path + .file_stem() + .and_then(|value| value.to_str()) + .unwrap_or_default(); + if !file_name.ends_with(&artifact.sha256) { + return Err(format!( + "evidence filename is not content-addressed: {}", + path.display() + )); + } + Ok(body) +} + fn reject_symlink_components(output_root: &Path, path: &Path) -> Result<(), String> { let root_metadata = fs::symlink_metadata(output_root) .map_err(|error| format!("inspect evidence root {}: {error}", output_root.display()))?; From 74ea6536c2b5c327dcc2e17d36b345a86d1020af Mon Sep 17 00:00:00 2001 From: Sonic Shih Date: Sun, 26 Jul 2026 12:48:12 +0800 Subject: [PATCH 2/4] test(research): use secure artifact temp directories --- .../polymarket-btc-5m.example.json | 2 +- .../polymarket-sol-5m.example.json | 2 +- .../src/event_cohort_partition.rs | 102 +++++++++--------- 3 files changed, 53 insertions(+), 53 deletions(-) diff --git a/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json b/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json index 82b91ad5b..dbe8accec 100644 --- a/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json +++ b/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json @@ -15,7 +15,7 @@ "horizon": "5m", "time_cohort_boundary_ms": 0, "prompt_snapshot_id": "sha256:2b55ba0e724dfc9f5a040911e397058a09c6229e42064e4998739d617b368dcb", - "search_policy_snapshot_id": "sha256:7b59f5f43dcbc4f5956cc9b78ccd5635ef7405578099d85859fe64590e170067", + "search_policy_snapshot_id": "sha256:4ca4577e38da9497f31add4367016276f86ae46d0ea779279744386baceb1bcf", "search_budget": { "max_candidates": 6, "max_llm_calls": 2, diff --git a/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json b/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json index c95f60585..f9ba9942d 100644 --- a/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json +++ b/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json @@ -15,7 +15,7 @@ "horizon": "5m", "time_cohort_boundary_ms": 0, "prompt_snapshot_id": "sha256:0816ebccf4c75ee6bdcfe315b253c84ddf1808a38687ebe3b924b87bd72a52a9", - "search_policy_snapshot_id": "sha256:7b59f5f43dcbc4f5956cc9b78ccd5635ef7405578099d85859fe64590e170067", + "search_policy_snapshot_id": "sha256:4ca4577e38da9497f31add4367016276f86ae46d0ea779279744386baceb1bcf", "search_budget": { "max_candidates": 6, "max_llm_calls": 2, diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs b/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs index 587fcdb04..977d1b539 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs @@ -840,32 +840,27 @@ mod tests { #[test] fn persisted_catalog_partition_artifact_round_trips_only_after_content_addressed_readback() { - let root = std::env::temp_dir().join(format!( - "ploy-catalog-partition-artifact-{}-{}", - std::process::id(), - uuid::Uuid::new_v4() - )); + let root = tempfile::tempdir().unwrap(); let catalog = PolymarketReadyEventCatalog::default(); let partition = EventCohortPartition::from_ready_catalog(&catalog, 1_000).unwrap(); - let artifact = - write_catalog_partition_artifact(&root, &root.join("evidence"), &catalog, &partition) - .expect("persist canonical catalog and partition"); - let restored = read_catalog_partition_artifact(&root, &artifact) + let artifact = write_catalog_partition_artifact( + root.path(), + &root.path().join("evidence"), + &catalog, + &partition, + ) + .expect("persist canonical catalog and partition"); + let restored = read_catalog_partition_artifact(root.path(), &artifact) .expect("fresh readback must validate the persisted artifact"); assert_eq!(restored.partition().digest(), partition.digest()); assert_eq!(restored.catalog().receipts().count(), 0); - std::fs::remove_dir_all(root).expect("remove artifact fixture"); } #[test] fn persisted_catalog_partition_artifact_round_trips_a_verified_ready_receipt() { - let root = std::env::temp_dir().join(format!( - "ploy-catalog-partition-artifact-ready-{}-{}", - std::process::id(), - uuid::Uuid::new_v4() - )); + let root = tempfile::tempdir().unwrap(); let catalog = persisted_ready_catalog_fixture(); let end = catalog .receipts() @@ -875,11 +870,15 @@ mod tests { .unwrap() .timestamp_millis(); let partition = EventCohortPartition::from_ready_catalog(&catalog, end + 3_000).unwrap(); - let artifact = - write_catalog_partition_artifact(&root, &root.join("evidence"), &catalog, &partition) - .unwrap(); + let artifact = write_catalog_partition_artifact( + root.path(), + &root.path().join("evidence"), + &catalog, + &partition, + ) + .unwrap(); - let restored = read_catalog_partition_artifact(&root, &artifact).unwrap(); + let restored = read_catalog_partition_artifact(root.path(), &artifact).unwrap(); let receipt = restored.catalog().receipts().next().unwrap(); assert_eq!( receipt.receipt_sha256, @@ -891,36 +890,36 @@ mod tests { receipt.receipt_sha256 ); assert_eq!(restored.partition().digest(), partition.digest()); - std::fs::remove_dir_all(root).unwrap(); } #[test] fn persisted_catalog_partition_artifact_rejects_corruption_mutable_paths_and_missing_partition() { - let root = std::env::temp_dir().join(format!( - "ploy-catalog-partition-artifact-rejection-{}-{}", - std::process::id(), - uuid::Uuid::new_v4() - )); + let root = tempfile::tempdir().unwrap(); let catalog = PolymarketReadyEventCatalog::default(); let partition = EventCohortPartition::from_ready_catalog(&catalog, 1_000).unwrap(); - let artifact = - write_catalog_partition_artifact(&root, &root.join("evidence"), &catalog, &partition) - .unwrap(); + let artifact = write_catalog_partition_artifact( + root.path(), + &root.path().join("evidence"), + &catalog, + &partition, + ) + .unwrap(); let mut mutable_path = artifact.clone(); mutable_path.path = "mutable.json".to_string(); - assert!(read_catalog_partition_artifact(&root, &mutable_path).is_err()); + assert!(read_catalog_partition_artifact(root.path(), &mutable_path).is_err()); let mut missing_partition: serde_json::Value = - serde_json::from_slice(&std::fs::read(root.join(artifact.path())).unwrap()).unwrap(); + serde_json::from_slice(&std::fs::read(root.path().join(artifact.path())).unwrap()) + .unwrap(); missing_partition["payload"] .as_object_mut() .unwrap() .remove("partition"); let missing = write_content_addressed_json( - &root, - &root.join("evidence"), + root.path(), + &root.path().join("evidence"), "catalog-partition", &missing_partition, ) @@ -930,31 +929,31 @@ mod tests { artifact_sha256: format!("sha256:{}", missing.sha256), payload_sha256: artifact.payload_sha256.clone(), }; - assert!(read_catalog_partition_artifact(&root, &missing_ref) + assert!(read_catalog_partition_artifact(root.path(), &missing_ref) .expect_err("missing partition must fail closed") .contains("missing field `partition`")); - std::fs::write(root.join(artifact.path()), b"corrupt").unwrap(); - assert!(read_catalog_partition_artifact(&root, &artifact) + std::fs::write(root.path().join(artifact.path()), b"corrupt").unwrap(); + assert!(read_catalog_partition_artifact(root.path(), &artifact) .expect_err("tampered bytes must fail before parse") .contains("hash mismatch")); - std::fs::remove_dir_all(root).unwrap(); } #[test] fn persisted_catalog_partition_artifact_rejects_catalog_partition_membership_mismatch() { - let root = std::env::temp_dir().join(format!( - "ploy-catalog-partition-artifact-membership-{}-{}", - std::process::id(), - uuid::Uuid::new_v4() - )); + let root = tempfile::tempdir().unwrap(); let catalog = PolymarketReadyEventCatalog::default(); let partition = EventCohortPartition::from_ready_catalog(&catalog, 1_000).unwrap(); - let artifact = - write_catalog_partition_artifact(&root, &root.join("evidence"), &catalog, &partition) - .unwrap(); + let artifact = write_catalog_partition_artifact( + root.path(), + &root.path().join("evidence"), + &catalog, + &partition, + ) + .unwrap(); let mut envelope: CatalogPartitionArtifactEnvelope = - serde_json::from_slice(&std::fs::read(root.join(artifact.path())).unwrap()).unwrap(); + serde_json::from_slice(&std::fs::read(root.path().join(artifact.path())).unwrap()) + .unwrap(); let persisted = &mut envelope.payload.partition; persisted.ready_entries.push(EventCohortReadyEntry { receipt_sha256: "a".repeat(64), @@ -984,8 +983,8 @@ mod tests { sha256_hex(&canonical_json_bytes(&envelope.payload).unwrap()) ); let mismatched = write_content_addressed_json( - &root, - &root.join("evidence"), + root.path(), + &root.path().join("evidence"), "catalog-partition", &envelope, ) @@ -995,10 +994,11 @@ mod tests { artifact_sha256: format!("sha256:{}", mismatched.sha256), payload_sha256: envelope.payload_sha256, }; - assert!(read_catalog_partition_artifact(&root, &mismatched_ref) - .expect_err("partition cannot add a receipt absent from the catalog") - .contains("does not contain every Ready catalog receipt")); - std::fs::remove_dir_all(root).unwrap(); + assert!( + read_catalog_partition_artifact(root.path(), &mismatched_ref) + .expect_err("partition cannot add a receipt absent from the catalog") + .contains("does not contain every Ready catalog receipt") + ); } #[test] From a4813d28e3046aea758a5db8cbc7dfe5e9e04cef Mon Sep 17 00:00:00 2001 From: Sonic Shih Date: Sun, 26 Jul 2026 12:57:23 +0800 Subject: [PATCH 3/4] fix(research): harden catalog artifact bounds --- .../polymarket-btc-5m.example.json | 2 +- .../polymarket-sol-5m.example.json | 2 +- .../src/event_cohort_partition.rs | 134 +++++++++++++++++- .../ploy-research/src/prediction_loop.rs | 7 +- .../ploy-research/src/prediction_loop_fs.rs | 64 +++++++++ 5 files changed, 202 insertions(+), 7 deletions(-) diff --git a/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json b/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json index dbe8accec..b67bc9077 100644 --- a/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json +++ b/rust_hft/prediction-markets/config/research_missions/polymarket-btc-5m.example.json @@ -15,7 +15,7 @@ "horizon": "5m", "time_cohort_boundary_ms": 0, "prompt_snapshot_id": "sha256:2b55ba0e724dfc9f5a040911e397058a09c6229e42064e4998739d617b368dcb", - "search_policy_snapshot_id": "sha256:4ca4577e38da9497f31add4367016276f86ae46d0ea779279744386baceb1bcf", + "search_policy_snapshot_id": "sha256:baf00ceead225e1f03f9eea831f664e8f296d44f045a14d6f07aff2da685abac", "search_budget": { "max_candidates": 6, "max_llm_calls": 2, diff --git a/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json b/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json index f9ba9942d..f59f7d28d 100644 --- a/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json +++ b/rust_hft/prediction-markets/config/research_missions/polymarket-sol-5m.example.json @@ -15,7 +15,7 @@ "horizon": "5m", "time_cohort_boundary_ms": 0, "prompt_snapshot_id": "sha256:0816ebccf4c75ee6bdcfe315b253c84ddf1808a38687ebe3b924b87bd72a52a9", - "search_policy_snapshot_id": "sha256:4ca4577e38da9497f31add4367016276f86ae46d0ea779279744386baceb1bcf", + "search_policy_snapshot_id": "sha256:baf00ceead225e1f03f9eea831f664e8f296d44f045a14d6f07aff2da685abac", "search_budget": { "max_candidates": 6, "max_llm_calls": 2, diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs b/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs index 977d1b539..9c11d00cf 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/event_cohort_partition.rs @@ -9,7 +9,7 @@ use serde_json::Value; use crate::{ prediction_loop::{current_prediction_policy_snapshot_id, validate_sha256_id}, prediction_loop_fs::{ - canonical_json_bytes, read_verified_artifact_bounded, sha256_hex, + canonical_json_bytes, read_verified_artifact_bounded, relative_path, sha256_hex, write_content_addressed_json, ArtifactRef, }, }; @@ -305,8 +305,14 @@ pub fn write_catalog_partition_artifact( "catalog partition artifact exceeds {MAX_CATALOG_PARTITION_ARTIFACT_BYTES} bytes" )); } + let expected_path = relative_path( + output_root, + &directory.join(format!("catalog-partition-{}.json", sha256_hex(&bytes))), + )?; + validate_catalog_partition_artifact_path(&expected_path)?; let artifact = write_content_addressed_json(output_root, directory, "catalog-partition", &envelope)?; + debug_assert_eq!(artifact.path, expected_path); Ok(CatalogPartitionArtifactRef { path: artifact.path, artifact_sha256: format!("sha256:{}", artifact.sha256), @@ -319,9 +325,7 @@ pub fn read_catalog_partition_artifact( output_root: &Path, artifact: &CatalogPartitionArtifactRef, ) -> Result { - if artifact.path.len() > MAX_CATALOG_PARTITION_ARTIFACT_PATH_BYTES { - return Err("catalog partition artifact path exceeds the bounded length".into()); - } + validate_catalog_partition_artifact_path(&artifact.path)?; validate_sha256_id( &artifact.artifact_sha256, "catalog partition artifact identity", @@ -370,9 +374,16 @@ pub fn read_catalog_partition_artifact( catalog_receipts, partition, } = envelope.payload; + if catalog_receipts.len() > MAX_READY_CATALOG_ENTRIES { + return Err("persisted ready catalog exceeds the bounded entry count".into()); + } + if partition.ready_entries.len() > MAX_READY_CATALOG_ENTRIES { + return Err("event cohort partition exceeds the bounded Ready entry count".into()); + } let catalog = PolymarketReadyEventCatalog::from_persisted_ready_receipts(catalog_receipts) .map_err(|error| format!("validate persisted ready catalog: {error}"))?; let partition = partition.into_partition()?; + validate_catalog_partition_write_bounds(&catalog, &partition)?; if policy_snapshot_id != partition.causal_projection_policy_id() { return Err("artifact policy identity does not match partition".into()); } @@ -383,6 +394,13 @@ pub fn read_catalog_partition_artifact( Ok(ValidatedCatalogPartitionArtifact { catalog, partition }) } +fn validate_catalog_partition_artifact_path(path: &str) -> Result<(), String> { + if path.len() > MAX_CATALOG_PARTITION_ARTIFACT_PATH_BYTES { + return Err("catalog partition artifact path exceeds the bounded length".into()); + } + Ok(()) +} + fn validate_catalog_partition_membership( catalog: &PolymarketReadyEventCatalog, partition: &EventCohortPartition, @@ -682,6 +700,7 @@ mod tests { read_catalog_partition_artifact, write_catalog_partition_artifact, CatalogPartitionArtifactEnvelope, CatalogPartitionArtifactRef, EventCohortPartition, EventCohortPartitionPayload, EventCohortReadyEntry, EVENT_COHORT_PARTITION_VERSION, + MAX_READY_CATALOG_ENTRIES, }; fn ready_receipt( @@ -1001,6 +1020,113 @@ mod tests { ); } + #[test] + fn persisted_catalog_partition_artifact_rejects_correctly_hashed_oversized_entries() { + let root = tempfile::tempdir().unwrap(); + let catalog = persisted_ready_catalog_fixture(); + let end = catalog + .receipts() + .next() + .unwrap() + .event_end + .unwrap() + .timestamp_millis(); + let partition = EventCohortPartition::from_ready_catalog(&catalog, end + 3_000).unwrap(); + let artifact = write_catalog_partition_artifact( + root.path(), + &root.path().join("evidence"), + &catalog, + &partition, + ) + .unwrap(); + let envelope_bytes = std::fs::read(root.path().join(artifact.path())).unwrap(); + + let mut oversized_catalog: CatalogPartitionArtifactEnvelope = + serde_json::from_slice(&envelope_bytes).unwrap(); + let receipt = oversized_catalog.payload.catalog_receipts[0].clone(); + oversized_catalog.payload.catalog_receipts = vec![receipt; MAX_READY_CATALOG_ENTRIES + 1]; + oversized_catalog.payload_sha256 = format!( + "sha256:{}", + sha256_hex(&canonical_json_bytes(&oversized_catalog.payload).unwrap()) + ); + let oversized_catalog_ref = write_content_addressed_json( + root.path(), + &root.path().join("evidence"), + "catalog-partition", + &oversized_catalog, + ) + .unwrap(); + let oversized_catalog_ref = CatalogPartitionArtifactRef { + path: oversized_catalog_ref.path, + artifact_sha256: format!("sha256:{}", oversized_catalog_ref.sha256), + payload_sha256: oversized_catalog.payload_sha256, + }; + assert!( + read_catalog_partition_artifact(root.path(), &oversized_catalog_ref) + .expect_err("an oversized canonical catalog must fail before receipt admission") + .contains("persisted ready catalog exceeds the bounded entry count") + ); + + let mut oversized_partition: CatalogPartitionArtifactEnvelope = + serde_json::from_slice(&envelope_bytes).unwrap(); + let entry = oversized_partition.payload.partition.ready_entries[0].clone(); + oversized_partition.payload.partition.ready_entries = + vec![entry; MAX_READY_CATALOG_ENTRIES + 1]; + let persisted = &mut oversized_partition.payload.partition; + let partition_payload = EventCohortPartitionPayload { + schema_version: EVENT_COHORT_PARTITION_VERSION, + ready_entries: &persisted.ready_entries, + common_time_boundary_ms: persisted.common_time_boundary_ms, + label_availability_cutoff_ms: persisted.label_availability_cutoff_ms, + causal_projection_policy_id: &persisted.causal_projection_policy_id, + train_market_ids: &persisted.train_market_ids, + crossing_excluded_market_ids: &persisted.crossing_excluded_market_ids, + held_out_market_ids: &persisted.held_out_market_ids, + }; + persisted.digest = format!( + "sha256:{}", + sha256_hex(&canonical_json_bytes(&partition_payload).unwrap()) + ); + oversized_partition.payload_sha256 = format!( + "sha256:{}", + sha256_hex(&canonical_json_bytes(&oversized_partition.payload).unwrap()) + ); + let oversized_partition_ref = write_content_addressed_json( + root.path(), + &root.path().join("evidence"), + "catalog-partition", + &oversized_partition, + ) + .unwrap(); + let oversized_partition_ref = CatalogPartitionArtifactRef { + path: oversized_partition_ref.path, + artifact_sha256: format!("sha256:{}", oversized_partition_ref.sha256), + payload_sha256: oversized_partition.payload_sha256, + }; + assert!( + read_catalog_partition_artifact(root.path(), &oversized_partition_ref) + .expect_err( + "an oversized canonical partition must fail before membership admission" + ) + .contains("event cohort partition exceeds the bounded Ready entry count") + ); + } + + #[test] + fn catalog_partition_writer_rejects_an_oversized_relative_artifact_path() { + let root = tempfile::tempdir().unwrap(); + let catalog = PolymarketReadyEventCatalog::default(); + let partition = EventCohortPartition::from_ready_catalog(&catalog, 1_000).unwrap(); + let directory = root.path().join(vec!["nested"; 200].join("/")); + + assert!( + write_catalog_partition_artifact(root.path(), &directory, &catalog, &partition) + .expect_err("the writer must reject an artifact reference it cannot safely return") + .contains("catalog partition artifact path exceeds the bounded length") + ); + assert!(!directory.exists()); + } + #[test] fn training_label_unavailable_by_the_cutoff_rejects_the_partition() { let mut train = ready_receipt('a', "late-label", 100, 999); diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop.rs b/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop.rs index cfa9bf6ac..0f3c200fb 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop.rs @@ -234,7 +234,7 @@ pub fn current_prediction_policy_snapshot_id() -> String { format!("sha256:{:x}", digest.finalize()) } -fn prediction_policy_sources() -> [(&'static str, &'static [u8]); 40] { +fn prediction_policy_sources() -> [(&'static str, &'static [u8]); 41] { [ ( "crates/ploy-research/src/autofactor.rs", @@ -293,6 +293,10 @@ fn prediction_policy_sources() -> [(&'static str, &'static [u8]); 40] { "crates/ploy-market-data/src/polymarket_evidence/set.rs", include_bytes!("../../ploy-market-data/src/polymarket_evidence/set.rs"), ), + ( + "crates/ploy-market-data/src/polymarket_evidence/catalog.rs", + include_bytes!("../../ploy-market-data/src/polymarket_evidence/catalog.rs"), + ), ( "crates/ploy-research/src/prediction_loop.rs", include_bytes!("prediction_loop.rs"), @@ -4167,6 +4171,7 @@ mod tests { "crates/ploy-market-data/src/polymarket_evidence/wire.rs", "crates/ploy-market-data/src/polymarket_evidence/verified.rs", "crates/ploy-market-data/src/polymarket_evidence/set.rs", + "crates/ploy-market-data/src/polymarket_evidence/catalog.rs", ] { assert!(paths.contains(&path)); } diff --git a/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop_fs.rs b/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop_fs.rs index 49f04c4bd..02bcb2ef7 100644 --- a/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop_fs.rs +++ b/rust_hft/prediction-markets/crates/ploy-research/src/prediction_loop_fs.rs @@ -193,6 +193,7 @@ fn write_content_addressed( extension: &str, body: &[u8], ) -> Result { + validate_artifact_write_directory(output_root, directory)?; create_dir_all_durable(directory, "evidence")?; reject_symlink_components(output_root, directory)?; let digest = sha256_hex(body); @@ -246,6 +247,52 @@ fn write_content_addressed( }) } +fn validate_artifact_write_directory(output_root: &Path, directory: &Path) -> Result<(), String> { + match fs::symlink_metadata(output_root) { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(format!( + "evidence root must not be a symlink: {}", + output_root.display() + )); + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(format!( + "inspect evidence root {}: {error}", + output_root.display() + )); + } + } + let relative = directory + .strip_prefix(output_root) + .map_err(|_| format!("artifact path {} escapes output root", directory.display()))?; + let mut current = output_root.to_path_buf(); + for component in relative.components() { + let Component::Normal(component) = component else { + return Err(format!("unsafe artifact path {}", directory.display())); + }; + current.push(component); + match fs::symlink_metadata(¤t) { + Ok(metadata) if metadata.file_type().is_symlink() => { + return Err(format!( + "evidence path contains symlink component {}", + current.display() + )); + } + Ok(_) => {} + Err(error) if error.kind() == std::io::ErrorKind::NotFound => {} + Err(error) => { + return Err(format!( + "inspect evidence path {}: {error}", + current.display() + )); + } + } + } + Ok(()) +} + pub(crate) fn atomic_write_json(path: &Path, value: &T) -> Result<(), String> { let parent = path .parent() @@ -428,6 +475,23 @@ pub(crate) fn next_attempt_dir(parent: &Path) -> Result { mod tests { use super::*; + #[test] + fn content_addressed_writer_rejects_parent_directory_escape_before_creation() { + let root = tempfile::tempdir().unwrap(); + let output_root = root.path().join("output"); + let escaped_directory = output_root.join("inside/../../escaped"); + + assert!(write_content_addressed_json( + &output_root, + &escaped_directory, + "record", + &serde_json::json!({"a": 1}), + ) + .expect_err("a writer directory must not escape its output root") + .contains("unsafe artifact path")); + assert!(!escaped_directory.exists()); + } + #[test] fn content_addressed_artifact_rejects_tampering_and_escape() { let root = std::env::temp_dir().join(format!( From 5561e42b3dc6b7b680a49365543f90c2a38e7fea Mon Sep 17 00:00:00 2001 From: Sonic Shih Date: Sun, 26 Jul 2026 13:02:01 +0800 Subject: [PATCH 4/4] docs(research): track catalog partition artifact --- rust_hft/prediction-markets/tasks/todo.md | 2 ++ 1 file changed, 2 insertions(+) diff --git a/rust_hft/prediction-markets/tasks/todo.md b/rust_hft/prediction-markets/tasks/todo.md index ffc964dd8..6ec90636d 100644 --- a/rust_hft/prediction-markets/tasks/todo.md +++ b/rust_hft/prediction-markets/tasks/todo.md @@ -48,6 +48,8 @@ Migrate the imported PLOY compatibility code into Monday's canonical market-fami - [x] Bind immutable Polymarket evidence to external content and manifest SHA-256 anchors before semantic consumption. - [x] Bind selected-event trades to an event-local collector completion proof from raw tape through sealed research evidence. - [x] Project verified Polymarket evidence into availability-safe research carriers without replaying discovery metadata or settlement labels before observation. +- [x] Persist and independently verify the immutable Ready catalog plus event + cohort partition artifact without reconstructing the partition (#365). - [ ] Wire the verified-artifact ResearchSnapshot adapter into the snapshot CLI and complete its cloud alpha-harness E2E. - [ ] Project causally valid in-event Chainlink ticks without allowing them to replace the pre-open five-minute strike (#304). - [x] Promote snapshot, prediction LoopRun, and event evaluator to precompiled