Skip to content

Commit a4beaef

Browse files
authored
Merge pull request #91 from CurtisTechSolutions/feat/stateless-training-batches
feat(training): add stateless batch REST environments
2 parents dfb4e1f + 4bcea94 commit a4beaef

12 files changed

Lines changed: 576 additions & 5 deletions

File tree

‎crates/adapters/gfa-http/src/lib.rs‎

Lines changed: 9 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -11,6 +11,7 @@ mod openapi;
1111
mod opponents;
1212
pub use openapi::openapi_document;
1313
mod stream;
14+
mod training;
1415

1516
pub use stream::LiveUpdates;
1617

@@ -115,6 +116,14 @@ fn build_router(
115116
"/v1/games/{game_id}/positions/validate",
116117
post(validate_position),
117118
)
119+
.route(
120+
"/v1/batch/step",
121+
post(training::batch)
122+
.layer::<_, std::convert::Infallible>(DefaultBodyLimit::max(4 * 1024 * 1024))
123+
.layer(axum::Extension(training::TrainingWorkers(Arc::new(
124+
tokio::sync::Semaphore::new(4),
125+
)))),
126+
)
118127
.route("/v1/matches", post(create).get(history))
119128
.route("/v1/matches/{id}", get(metadata))
120129
.route("/v1/matches/{id}/resign", post(resign))

‎crates/adapters/gfa-http/src/openapi.rs‎

Lines changed: 2 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -3,10 +3,10 @@ use utoipa::OpenApi;
33

44
#[derive(OpenApi)]
55
#[openapi(
6-
info(title = "GamesForAI local API", description = "Single-user local play. Requests require a loopback peer and matching Host and optional Origin. Any local caller can control either seat. Bodies are limited to 64 KiB. Only implemented endpoints are listed."),
6+
info(title = "GamesForAI local API", description = "Single-user local play. Requests require a loopback peer and matching Host and optional Origin. Any local caller can control either seat. Bodies are limited to 64 KiB, except stateless training batches (4 MiB). Only implemented endpoints are listed."),
77
servers((url = "/", description = "This local host")),
88
paths(
9-
super::events::history,
9+
super::events::history, super::training::batch,
1010
super::opponents::catalog, super::opponents::analyze,
1111
super::fork_match, super::simulate, super::resign, super::offer_draw, super::history, super::metadata, super::health, super::games, super::game, super::create, super::state,
1212
super::legal_actions, super::make_move, super::replay, super::validate_position,
Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
use crate::error::HttpError;
2+
use axum::{
3+
extract::{rejection::JsonRejection, State},
4+
Extension, Json,
5+
};
6+
use gfa_api_types::{ApiError, TrainingBatch, TrainingBatchResult};
7+
use gfa_service::GameService;
8+
use std::sync::Arc;
9+
use tokio::sync::Semaphore;
10+
11+
#[derive(Clone)]
12+
pub(crate) struct TrainingWorkers(pub Arc<Semaphore>);
13+
14+
#[utoipa::path(
15+
post, path = "/v1/batch/step", tag = "Training",
16+
request_body = TrainingBatch,
17+
responses(
18+
(status = 200, description = "Ordered independent owner-only training results; no persistence or match IDs", body = TrainingBatchResult),
19+
(status = "default", description = "Batch size, local access, or worker limit failure", body = gfa_api_types::ErrorResponse)
20+
)
21+
)]
22+
pub(crate) async fn batch(
23+
State(service): State<Arc<GameService>>,
24+
Extension(workers): Extension<TrainingWorkers>,
25+
body: Result<Json<TrainingBatch>, JsonRejection>,
26+
) -> Result<Json<TrainingBatchResult>, HttpError> {
27+
let Json(request) = body?;
28+
let permit = workers.0.try_acquire_owned().map_err(|_| {
29+
ApiError::new(
30+
"ENGINE_BUSY",
31+
"All training workers are busy",
32+
"Retry this stateless batch later.",
33+
)
34+
})?;
35+
let result = tokio::task::spawn_blocking(move || {
36+
let _permit = permit;
37+
service.training_batch(request)
38+
})
39+
.await
40+
.map_err(|_| {
41+
ApiError::new(
42+
"ENGINE_UNAVAILABLE",
43+
"Training worker failed",
44+
"Retry this stateless batch.",
45+
)
46+
})??;
47+
Ok(Json(result))
48+
}

‎crates/apps/gfa-server/src/lib.rs‎

Lines changed: 3 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -289,3 +289,6 @@ mod mcp_live_tests;
289289

290290
#[cfg(test)]
291291
mod uci_cap_tests;
292+
293+
#[cfg(test)]
294+
mod training_tests;

‎crates/apps/gfa-server/src/openapi_tests.rs‎

Lines changed: 2 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -61,6 +61,7 @@ async fn served_openapi_matches_real_requests_responses_and_conditional_formats(
6161
("/v1/games/{game_id}/simulate", "post"),
6262
("/v1/games/{game_id}/opponents", "get"),
6363
("/v1/analysis", "post"),
64+
("/v1/batch/step", "post"),
6465
("/v1/matches", "post"),
6566
("/v1/matches", "get"),
6667
("/v1/matches/{id}", "get"),
@@ -81,7 +82,7 @@ async fn served_openapi_matches_real_requests_responses_and_conditional_formats(
8182
"missing {method} {path}"
8283
);
8384
}
84-
assert_eq!(doc["paths"].as_object().map(serde_json::Map::len), Some(21));
85+
assert_eq!(doc["paths"].as_object().map(serde_json::Map::len), Some(22));
8586
let game_info = &doc["paths"]["/v1/games/{game_id}/info"]["get"];
8687
assert!(game_info["responses"]["200"]["content"]["text/markdown"].is_object());
8788
assert!(game_info["responses"]["304"].get("content").is_none());
Lines changed: 207 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,207 @@
1+
use super::{
2+
tests::{call, config, fixture},
3+
*,
4+
};
5+
use axum::http::StatusCode;
6+
use serde_json::{json, Value};
7+
8+
type TestResult = Result<(), ServerError>;
9+
10+
async fn batch(app: &Application, operations: Value) -> Result<Value, ServerError> {
11+
let (status, body) = call(
12+
&app.router,
13+
"POST",
14+
"/v1/batch/step",
15+
json!({"operations": operations}),
16+
None,
17+
)
18+
.await?;
19+
assert_eq!(status, StatusCode::OK, "{body}");
20+
Ok(body)
21+
}
22+
23+
#[tokio::test]
24+
async fn training_batches_create_step_reset_branch_and_never_persist() -> TestResult {
25+
let dir = tempfile::tempdir()?;
26+
let app = fixture(&config(&dir)).await?;
27+
let initial = batch(
28+
&app,
29+
json!([
30+
{"op":"create","game_id":"tictactoe","seed":42,"seat":0},
31+
{"op":"create","game_id":"connect4","seed":43,"seat":1},
32+
{"op":"create","game_id":"missing","seed":0,"seat":0}
33+
]),
34+
)
35+
.await?;
36+
assert_eq!(initial["results"][0]["status"], "ok");
37+
assert_eq!(
38+
initial["results"][1]["frame"]["action_mask"],
39+
json!(vec![false; 7])
40+
);
41+
assert_eq!(initial["results"][2]["error"]["code"], "UNKNOWN_GAME");
42+
let checkpoint = initial["results"][0]["checkpoint"].clone();
43+
let request = json!([
44+
{"op":"step","checkpoint":checkpoint,"seat":0,"action":4},
45+
{"op":"step","checkpoint":checkpoint,"seat":0,"action":0},
46+
{"op":"step","checkpoint":checkpoint,"seat":1,"action":4}
47+
]);
48+
let first = batch(&app, request.clone()).await?;
49+
assert_eq!(first, batch(&app, request).await?);
50+
assert_eq!(first["results"][0]["frame"]["to_act"], json!([1]));
51+
assert_eq!(first["results"][0]["frame"]["rewards"], json!([0.0, 0.0]));
52+
assert_ne!(
53+
first["results"][0]["checkpoint"],
54+
first["results"][1]["checkpoint"]
55+
);
56+
assert_eq!(first["results"][2]["status"], "error");
57+
let moved = &first["results"][0]["checkpoint"];
58+
let frames = batch(
59+
&app,
60+
json!([
61+
{"op":"observe","checkpoint":moved,"seat":1},
62+
{"op":"observe","checkpoint":moved,"seat":9},
63+
{"op":"reset","checkpoint":moved,"seed":42,"seat":0},
64+
{"op":"observe","checkpoint":moved}
65+
]),
66+
)
67+
.await?;
68+
assert_eq!(
69+
frames["results"][0]["frame"]["legal_actions"]
70+
.as_array()
71+
.map(Vec::len),
72+
Some(8)
73+
);
74+
assert_eq!(frames["results"][1]["status"], "error");
75+
assert_eq!(frames["results"][2]["checkpoint"], checkpoint);
76+
assert!(frames["results"][3]["frame"]["seat"].is_null());
77+
assert_eq!(frames["results"][3]["frame"]["legal_actions"], json!([]));
78+
assert_eq!(
79+
frames["results"][3]["frame"]["action_mask"],
80+
json!(vec![false; 9])
81+
);
82+
assert!(frames["results"][0]["frame"].get("checkpoint").is_none());
83+
let (_, history) = call(&app.router, "GET", "/v1/matches", Value::Null, None).await?;
84+
assert_eq!(history["matches"], json!([]));
85+
app.store.close().await;
86+
Ok(())
87+
}
88+
89+
#[tokio::test]
90+
async fn batch_rewards_endings_and_invalid_checkpoints_match_native_rules() -> TestResult {
91+
let dir = tempfile::tempdir()?;
92+
let app = fixture(&config(&dir)).await?;
93+
let created = batch(
94+
&app,
95+
json!([
96+
{"op":"create","game_id":"tictactoe","seed":0,"seat":0},
97+
{"op":"create","game_id":"chess","config":{"max_plies":1},"seed":0,"seat":0}
98+
]),
99+
)
100+
.await?;
101+
let mut checkpoint = created["results"][0]["checkpoint"].clone();
102+
for (turn, action) in [0, 3, 1, 4, 2].into_iter().enumerate() {
103+
let result = batch(
104+
&app,
105+
json!([
106+
{"op":"step","checkpoint":checkpoint,"seat":turn%2,"action":action}
107+
]),
108+
)
109+
.await?;
110+
assert_eq!(result["results"][0]["status"], "ok");
111+
if turn == 4 {
112+
assert_eq!(result["results"][0]["frame"]["terminated"], true);
113+
assert_eq!(result["results"][0]["frame"]["rewards"], json!([1.0, -1.0]));
114+
}
115+
checkpoint = result["results"][0]["checkpoint"].clone();
116+
}
117+
let ended = batch(
118+
&app,
119+
json!([
120+
{"op":"step","checkpoint":checkpoint,"seat":1,"action":8}
121+
]),
122+
)
123+
.await?;
124+
assert_eq!(ended["results"][0]["error"]["code"], "MATCH_FINISHED");
125+
let chess = &created["results"][1];
126+
let capped = batch(
127+
&app,
128+
json!([{
129+
"op":"step","checkpoint":chess["checkpoint"],"seat":0,
130+
"action":chess["frame"]["legal_actions"][0]["index"]
131+
}]),
132+
)
133+
.await?;
134+
assert_eq!(capped["results"][0]["frame"]["truncated"], true);
135+
assert_eq!(capped["results"][0]["frame"]["terminated"], false);
136+
checkpoint["engine_version"] = json!("incompatible");
137+
let bad = batch(
138+
&app,
139+
json!([
140+
{"op":"observe","checkpoint":checkpoint,"seat":0}
141+
]),
142+
)
143+
.await?;
144+
assert_eq!(bad["results"][0]["error"]["code"], "INVALID_POSITION");
145+
app.store.close().await;
146+
Ok(())
147+
}
148+
149+
#[tokio::test]
150+
async fn batch_limits_and_route_body_override_are_enforced() -> TestResult {
151+
let dir = tempfile::tempdir()?;
152+
let app = fixture(&config(&dir)).await?;
153+
let create = json!({"op":"create","game_id":"tictactoe","seed":0,"seat":0});
154+
for operations in [vec![], vec![create.clone(); 65]] {
155+
let (status, body) = call(
156+
&app.router,
157+
"POST",
158+
"/v1/batch/step",
159+
json!({"operations":operations}),
160+
None,
161+
)
162+
.await?;
163+
assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
164+
assert_eq!(body["error"]["code"], "BATCH_LIMIT");
165+
}
166+
// Training accepts more than the standard route's 64 KiB.
167+
let result = batch(
168+
&app,
169+
json!([{
170+
"op":"create","game_id":"tictactoe","seed":0,"seat":0,
171+
"config":{"unknown": "x".repeat(70000)}
172+
}]),
173+
)
174+
.await?;
175+
assert_eq!(result["results"][0]["error"]["code"], "INVALID_CONFIG");
176+
let (status, _) = call(
177+
&app.router,
178+
"POST",
179+
"/v1/batch/step",
180+
json!({
181+
"operations":[{"op":"create","game_id":"tictactoe","seed":0,"seat":0,
182+
"config":{"unknown":"x".repeat(4 * 1024 * 1024)}}]
183+
}),
184+
None,
185+
)
186+
.await?;
187+
assert_eq!(status, StatusCode::PAYLOAD_TOO_LARGE);
188+
let (status, _) = call(&app.router, "POST", "/v1/batch/step", json!({
189+
"operations":[{"op":"create","game_id":"tictactoe","seed":0,"seat":0,"match_id":"private"}]
190+
}), None).await?;
191+
assert_eq!(status, StatusCode::UNPROCESSABLE_ENTITY);
192+
193+
// Direct trusted service calls also bound output even without HTTP input limits.
194+
let request = serde_json::from_value(json!({"operations":vec![json!({
195+
"op":"create","game_id":"x".repeat(200000),"seed":0,"seat":0
196+
});64]}))?;
197+
let result = app.service.training_batch(request)?;
198+
assert_eq!(result.results.len(), 64);
199+
assert!(serde_json::to_vec(&result)?.len() <= 8 * 1024 * 1024);
200+
assert!(serde_json::to_value(result)?["results"]
201+
.as_array()
202+
.ok_or("results")?
203+
.iter()
204+
.any(|row| row["error"]["code"] == "BATCH_LIMIT"));
205+
app.store.close().await;
206+
Ok(())
207+
}

‎crates/core/gfa-core/src/env.rs‎

Lines changed: 4 additions & 2 deletions
Original file line numberDiff line numberDiff line change
@@ -21,8 +21,10 @@ impl<G: Game> Clone for Env<G> {
2121
}
2222

2323
/// Versioned full-state checkpoint, including private information and engine RNG.
24-
/// Never return this object through a player's observation or a public API.
25-
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize)]
24+
/// Never return it through a player/spectator projection or a ranked-match API.
25+
/// A trusted training host may return caller-owned checkpoints only to their owner.
26+
#[cfg_attr(feature = "openapi", derive(utoipa::ToSchema))]
27+
#[derive(Clone, Debug, PartialEq, Serialize, Deserialize, crate::schemars::JsonSchema)]
2628
#[serde(deny_unknown_fields)]
2729
pub struct EnvSnapshot {
2830
/// Snapshot format version.

‎crates/domain/gfa-api-types/src/lib.rs‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -312,3 +312,8 @@ pub use seats::{OpponentReply, Seat};
312312

313313
mod events;
314314
pub use events::{EventData, EventPage, EventsQuery, RecordedEvent};
315+
316+
mod training;
317+
pub use training::{
318+
TrainingBatch, TrainingBatchResult, TrainingFrame, TrainingOperation, TrainingResult,
319+
};

0 commit comments

Comments
 (0)