Route player waits through strategic world time
This commit is contained in:
parent
a4122e91d4
commit
31300cd84f
4 changed files with 277 additions and 70 deletions
|
|
@ -883,7 +883,7 @@ fn execute(
|
|||
for clock in campaign.clocks.values_mut() {
|
||||
clock.progress = clock.progress.saturating_add(1).min(clock.threshold);
|
||||
}
|
||||
let tick_number = campaign.away_ticks_processed.saturating_add(1);
|
||||
let tick_number = campaign.strategic_tick_count.saturating_add(1);
|
||||
let tick_events = match resolved_plan.or(plan) {
|
||||
Some(plan) => apply_strategic_tick_plan(&mut campaign, plan)?,
|
||||
None => deterministic_strategic_tick(&mut campaign, tick_number),
|
||||
|
|
@ -907,8 +907,15 @@ fn execute(
|
|||
}
|
||||
let event_ids = tick_events.iter().map(|event| event.id.clone()).collect();
|
||||
campaign.events.extend(tick_events);
|
||||
campaign.away_ticks_processed = tick_number.min(8);
|
||||
campaign.pending_ticks = campaign.pending_ticks.saturating_sub(1);
|
||||
if source == TickSource::PlayerWait {
|
||||
campaign.last_player_activity = Utc::now();
|
||||
campaign.away_ticks_processed = 0;
|
||||
campaign.pending_ticks = 0;
|
||||
} else {
|
||||
campaign.away_ticks_processed =
|
||||
campaign.away_ticks_processed.saturating_add(1).min(8);
|
||||
campaign.pending_ticks = campaign.pending_ticks.saturating_sub(1);
|
||||
}
|
||||
commit_strategic_tick(
|
||||
store,
|
||||
row,
|
||||
|
|
@ -2650,7 +2657,7 @@ fn rebase_member_migration(
|
|||
|
||||
fn deterministic_strategic_tick(
|
||||
campaign: &mut Campaign,
|
||||
tick_number: u8,
|
||||
tick_number: u64,
|
||||
) -> Vec<crate::domain::Event> {
|
||||
let mut events = Vec::new();
|
||||
for institution in campaign.institutions.values_mut() {
|
||||
|
|
|
|||
|
|
@ -87,15 +87,20 @@ struct AppState {
|
|||
|
||||
struct LiveTurnGuard {
|
||||
counter: Arc<AtomicUsize>,
|
||||
mesh: MeshPublisher,
|
||||
_commit_read: Option<OwnedRwLockReadGuard<()>>,
|
||||
}
|
||||
impl LiveTurnGuard {
|
||||
async fn enter(state: &AppState) -> Self {
|
||||
let mut guard = Self {
|
||||
counter: state.live_turns.clone(),
|
||||
mesh: state.mesh.clone(),
|
||||
_commit_read: None,
|
||||
};
|
||||
guard.counter.fetch_add(1, Ordering::SeqCst);
|
||||
let pressure = guard.counter.fetch_add(1, Ordering::SeqCst) + 1;
|
||||
if let Err(error) = guard.mesh.publish_live_turn_pressure(pressure) {
|
||||
tracing::warn!(%error, pressure, "live-turn pressure CultMesh publication failed");
|
||||
}
|
||||
state.live_turn_started.notify_waiters();
|
||||
guard._commit_read = Some(state.live_commit_gate.clone().read_owned().await);
|
||||
guard
|
||||
|
|
@ -103,7 +108,11 @@ impl LiveTurnGuard {
|
|||
}
|
||||
impl Drop for LiveTurnGuard {
|
||||
fn drop(&mut self) {
|
||||
self.counter.fetch_sub(1, Ordering::SeqCst);
|
||||
let previous = self.counter.fetch_sub(1, Ordering::SeqCst);
|
||||
let pressure = previous.saturating_sub(1);
|
||||
if let Err(error) = self.mesh.publish_live_turn_pressure(pressure) {
|
||||
tracing::warn!(%error, pressure, "live-turn pressure CultMesh release failed");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -4238,6 +4247,80 @@ async fn command(
|
|||
)
|
||||
.into_response();
|
||||
}
|
||||
if let WorldCommand::Wait {
|
||||
expected_revision,
|
||||
minutes,
|
||||
} = &command
|
||||
{
|
||||
let campaign = match load_campaign(&runtime.store) {
|
||||
Ok(value) => value,
|
||||
Err(error) => {
|
||||
return (
|
||||
StatusCode::INTERNAL_SERVER_ERROR,
|
||||
Json(ErrorBody {
|
||||
error: error.to_string(),
|
||||
}),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
};
|
||||
if *expected_revision != campaign.revision {
|
||||
return (
|
||||
StatusCode::CONFLICT,
|
||||
Json(ErrorBody {
|
||||
error: format!(
|
||||
"stale revision: expected {expected_revision}, actual {}",
|
||||
campaign.revision
|
||||
),
|
||||
}),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
let strategic_minutes = u32::from(campaign.tick_hours).saturating_mul(60);
|
||||
if *minutes > strategic_minutes {
|
||||
return (
|
||||
StatusCode::UNPROCESSABLE_ENTITY,
|
||||
Json(ErrorBody {
|
||||
error: format!(
|
||||
"one wait may cover at most one strategic horizon ({strategic_minutes} minutes); wait again after the world advances"
|
||||
),
|
||||
}),
|
||||
)
|
||||
.into_response();
|
||||
}
|
||||
if *minutes == strategic_minutes {
|
||||
return match advance_one_strategic_tick(
|
||||
&state,
|
||||
&runtime,
|
||||
campaign,
|
||||
ghostlight_dungeon::domain::TickSource::PlayerWait,
|
||||
false,
|
||||
)
|
||||
.await
|
||||
{
|
||||
Ok(Some(result)) => {
|
||||
if let Err(error) = refresh_mesh(&state).await {
|
||||
tracing::warn!(%error, "player strategic wait CultMesh publication failed");
|
||||
}
|
||||
Json(player_command_projection(&result, None)).into_response()
|
||||
}
|
||||
Ok(None) => (
|
||||
StatusCode::SERVICE_UNAVAILABLE,
|
||||
Json(ErrorBody {
|
||||
error: "strategic waiting yielded without advancing the world".into(),
|
||||
}),
|
||||
)
|
||||
.into_response(),
|
||||
Err(error) => (
|
||||
StatusCode::UNPROCESSABLE_ENTITY,
|
||||
Json(ErrorBody {
|
||||
error: error.to_string(),
|
||||
}),
|
||||
)
|
||||
.into_response(),
|
||||
};
|
||||
}
|
||||
}
|
||||
if let WorldCommand::Assess {
|
||||
intent, proposal, ..
|
||||
} = &mut command
|
||||
|
|
@ -4550,10 +4633,14 @@ async fn command(
|
|||
&result,
|
||||
CommandResult::Committed { .. } | CommandResult::Created { .. }
|
||||
) {
|
||||
let narration = publish_latest_narration(&state, &runtime)
|
||||
.await
|
||||
.ok()
|
||||
.flatten();
|
||||
let narration = if should_react {
|
||||
publish_latest_narration(&state, &runtime)
|
||||
.await
|
||||
.ok()
|
||||
.flatten()
|
||||
} else {
|
||||
None
|
||||
};
|
||||
if let Err(error) = refresh_mesh(&state).await {
|
||||
tracing::warn!(%error, "post-command CultMesh publication failed");
|
||||
}
|
||||
|
|
@ -5220,39 +5307,41 @@ async fn process_due_ticks(
|
|||
if campaign.away_ticks_processed >= target {
|
||||
return Ok(());
|
||||
}
|
||||
let has_simulatable_agency = campaign
|
||||
.agency_profiles
|
||||
.values()
|
||||
.any(|profile| profile.active_leaf && profile.simulation_eligible);
|
||||
if !has_simulatable_agency {
|
||||
if yield_to_live_turns && state.live_turns.load(Ordering::SeqCst) > 0 {
|
||||
return Ok(());
|
||||
}
|
||||
let _background_commit = if yield_to_live_turns {
|
||||
match state.live_commit_gate.clone().try_write_owned() {
|
||||
Ok(guard) => Some(guard),
|
||||
Err(_) => return Ok(()),
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
if yield_to_live_turns && state.live_turns.load(Ordering::SeqCst) > 0 {
|
||||
return Ok(());
|
||||
}
|
||||
runtime
|
||||
.kernel
|
||||
.command(WorldCommand::AdvanceStrategicTick {
|
||||
expected_revision: campaign.revision,
|
||||
source: source.clone(),
|
||||
plan: None,
|
||||
model_receipt_hash: None,
|
||||
resolution_wave: None,
|
||||
})
|
||||
.await?;
|
||||
continue;
|
||||
}
|
||||
let Some(model) = &state.model else {
|
||||
if advance_one_strategic_tick(
|
||||
state,
|
||||
runtime,
|
||||
campaign,
|
||||
source.clone(),
|
||||
yield_to_live_turns,
|
||||
)
|
||||
.await?
|
||||
.is_none()
|
||||
{
|
||||
return Ok(());
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
async fn advance_one_strategic_tick(
|
||||
state: &AppState,
|
||||
runtime: &CampaignRuntime,
|
||||
campaign: Campaign,
|
||||
source: ghostlight_dungeon::domain::TickSource,
|
||||
yield_to_live_turns: bool,
|
||||
) -> anyhow::Result<Option<CommandResult>> {
|
||||
if yield_to_live_turns && state.live_turns.load(Ordering::SeqCst) > 0 {
|
||||
return Ok(None);
|
||||
}
|
||||
let has_simulatable_agency = campaign
|
||||
.agency_profiles
|
||||
.values()
|
||||
.any(|profile| profile.active_leaf && profile.simulation_eligible);
|
||||
let (model_receipt_hash, resolution_wave) = if has_simulatable_agency {
|
||||
let Some(model) = &state.model else {
|
||||
if yield_to_live_turns {
|
||||
return Ok(None);
|
||||
}
|
||||
anyhow::bail!("strategic waiting requires the model provider");
|
||||
};
|
||||
let permit = Arc::new(SnapshotPermit::new_resolution(
|
||||
runtime.store.clone(),
|
||||
|
|
@ -5275,11 +5364,8 @@ async fn process_due_ticks(
|
|||
)
|
||||
.await?
|
||||
else {
|
||||
return Ok(());
|
||||
return Ok(None);
|
||||
};
|
||||
if yield_to_live_turns && state.live_turns.load(Ordering::SeqCst) > 0 {
|
||||
return Ok(());
|
||||
}
|
||||
for stage in &output.stages {
|
||||
match runtime
|
||||
.store
|
||||
|
|
@ -5299,31 +5385,36 @@ async fn process_due_ticks(
|
|||
}
|
||||
}
|
||||
}
|
||||
if yield_to_live_turns && state.live_turns.load(Ordering::SeqCst) > 0 {
|
||||
return Ok(());
|
||||
}
|
||||
let _background_commit = if yield_to_live_turns {
|
||||
match state.live_commit_gate.clone().try_write_owned() {
|
||||
Ok(guard) => Some(guard),
|
||||
Err(_) => return Ok(()),
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
if yield_to_live_turns && state.live_turns.load(Ordering::SeqCst) > 0 {
|
||||
return Ok(());
|
||||
}
|
||||
runtime
|
||||
.kernel
|
||||
.command(WorldCommand::AdvanceStrategicTick {
|
||||
expected_revision: campaign.revision,
|
||||
source: source.clone(),
|
||||
plan: None,
|
||||
model_receipt_hash: Some(output.aggregate_receipt_hash),
|
||||
resolution_wave: Some(output.wave),
|
||||
})
|
||||
.await?;
|
||||
(Some(output.aggregate_receipt_hash), Some(output.wave))
|
||||
} else {
|
||||
(None, None)
|
||||
};
|
||||
if yield_to_live_turns && state.live_turns.load(Ordering::SeqCst) > 0 {
|
||||
return Ok(None);
|
||||
}
|
||||
let _background_commit = if yield_to_live_turns {
|
||||
match state.live_commit_gate.clone().try_write_owned() {
|
||||
Ok(guard) => Some(guard),
|
||||
Err(_) => return Ok(None),
|
||||
}
|
||||
} else {
|
||||
None
|
||||
};
|
||||
if yield_to_live_turns && state.live_turns.load(Ordering::SeqCst) > 0 {
|
||||
return Ok(None);
|
||||
}
|
||||
runtime
|
||||
.kernel
|
||||
.command(WorldCommand::AdvanceStrategicTick {
|
||||
expected_revision: campaign.revision,
|
||||
source,
|
||||
plan: None,
|
||||
model_receipt_hash,
|
||||
resolution_wave,
|
||||
})
|
||||
.await
|
||||
.map(Some)
|
||||
.map_err(Into::into)
|
||||
}
|
||||
|
||||
async fn await_background_work<T>(
|
||||
|
|
@ -6008,6 +6099,10 @@ mod tests {
|
|||
async fn interactive_model_work_cancels_background_inference_and_excludes_its_commit_gate() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let state = empty_app_state(dir.path());
|
||||
state
|
||||
.mesh
|
||||
.publish_snapshot(&[], &[], &state.model_status, 0)
|
||||
.unwrap();
|
||||
let trigger_state = state.clone();
|
||||
let (entered_tx, entered_rx) = tokio::sync::oneshot::channel();
|
||||
let (release_tx, release_rx) = tokio::sync::oneshot::channel();
|
||||
|
|
@ -6029,11 +6124,19 @@ mod tests {
|
|||
assert!(result.is_none());
|
||||
entered_rx.await.unwrap();
|
||||
assert_eq!(state.live_turns.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(
|
||||
state.mesh.health().unwrap()["scheduler"]["live_turn_pressure"],
|
||||
1
|
||||
);
|
||||
assert!(state.live_commit_gate.clone().try_write_owned().is_err());
|
||||
|
||||
let _ = release_tx.send(());
|
||||
trigger.await.unwrap();
|
||||
assert_eq!(state.live_turns.load(Ordering::SeqCst), 0);
|
||||
assert_eq!(
|
||||
state.mesh.health().unwrap()["scheduler"]["live_turn_pressure"],
|
||||
0
|
||||
);
|
||||
assert!(state.live_commit_gate.clone().try_write_owned().is_ok());
|
||||
}
|
||||
|
||||
|
|
@ -6139,6 +6242,54 @@ mod tests {
|
|||
assert_eq!(advanced.actors["player"], original_player);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn player_strategic_wait_uses_the_same_tick_commit_and_resets_away_accounting() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
let state = empty_app_state(dir.path());
|
||||
let mut campaign = seed("Player wait world");
|
||||
let original_time = campaign.world_time;
|
||||
campaign.last_player_activity = chrono::Utc::now() - chrono::Duration::hours(4);
|
||||
campaign.away_ticks_processed = 3;
|
||||
campaign.pending_ticks = 2;
|
||||
campaign.clocks.insert(
|
||||
"pressure".into(),
|
||||
ghostlight_dungeon::domain::WorldClock {
|
||||
id: "pressure".into(),
|
||||
label: "Pressure rises".into(),
|
||||
progress: 0,
|
||||
threshold: 4,
|
||||
consequence: "the settlement must choose".into(),
|
||||
},
|
||||
);
|
||||
let runtime = state
|
||||
.registry
|
||||
.create(campaign.clone(), vec![], vec![])
|
||||
.await
|
||||
.unwrap();
|
||||
|
||||
let result = advance_one_strategic_tick(
|
||||
&state,
|
||||
&runtime,
|
||||
campaign,
|
||||
ghostlight_dungeon::domain::TickSource::PlayerWait,
|
||||
false,
|
||||
)
|
||||
.await
|
||||
.unwrap();
|
||||
assert!(result.is_some());
|
||||
|
||||
let advanced = load_campaign(&runtime.store).unwrap();
|
||||
assert_eq!(
|
||||
advanced.world_time,
|
||||
original_time + chrono::Duration::hours(i64::from(advanced.tick_hours))
|
||||
);
|
||||
assert_eq!(advanced.clocks["pressure"].progress, 1);
|
||||
assert_eq!(advanced.strategic_tick_count, 1);
|
||||
assert_eq!(advanced.away_ticks_processed, 0);
|
||||
assert_eq!(advanced.pending_ticks, 0);
|
||||
assert!(advanced.last_player_activity > chrono::Utc::now() - chrono::Duration::minutes(1));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn campaign_selection_is_derived_from_exact_membership() {
|
||||
let dir = tempfile::tempdir().unwrap();
|
||||
|
|
|
|||
|
|
@ -450,6 +450,31 @@ impl MeshPublisher {
|
|||
.map(|record| record.value)
|
||||
}
|
||||
|
||||
pub fn publish_live_turn_pressure(&self, live_turn_pressure: usize) -> Result<Value> {
|
||||
let mut node = self
|
||||
.node
|
||||
.lock()
|
||||
.map_err(|_| anyhow::anyhow!("mesh lock poisoned"))?;
|
||||
let mut health = node.get_required::<ServiceHealthRecord>(HEALTH_KEY)?.value;
|
||||
health["scheduler"]["live_turn_pressure"] = json!(live_turn_pressure);
|
||||
health["updatedAtUtc"] = json!(Utc::now().to_rfc3339());
|
||||
let mut remote_messages = Vec::new();
|
||||
self.put_and_stage(
|
||||
&mut node,
|
||||
HEALTH_KEY,
|
||||
&ServiceHealthRecord {
|
||||
value: health.clone(),
|
||||
},
|
||||
&mut remote_messages,
|
||||
)?;
|
||||
node.flush()?;
|
||||
drop(node);
|
||||
if let Some(remote) = &self.remote {
|
||||
remote.enqueue(remote_messages);
|
||||
}
|
||||
Ok(health)
|
||||
}
|
||||
|
||||
pub fn operator_surface(&self, campaign_id: uuid::Uuid) -> Result<Value> {
|
||||
self.node
|
||||
.lock()
|
||||
|
|
@ -671,6 +696,23 @@ mod tests {
|
|||
assert_eq!(reopened.health().unwrap(), written);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn live_turn_pressure_updates_the_canonical_health_document_immediately() {
|
||||
let temp = tempdir().unwrap();
|
||||
let publisher = MeshPublisher::open(temp.path().join("mesh.cc"), None).unwrap();
|
||||
publisher
|
||||
.publish_snapshot(&[], &[], &fixture_model_status(), 0)
|
||||
.unwrap();
|
||||
|
||||
let pressured = publisher.publish_live_turn_pressure(3).unwrap();
|
||||
assert_eq!(pressured["scheduler"]["live_turn_pressure"], 3);
|
||||
assert_eq!(publisher.health().unwrap(), pressured);
|
||||
|
||||
let idle = publisher.publish_live_turn_pressure(0).unwrap();
|
||||
assert_eq!(idle["scheduler"]["live_turn_pressure"], 0);
|
||||
assert_eq!(publisher.health().unwrap(), idle);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn unavailable_odin_cannot_take_away_the_local_projection() {
|
||||
let temp = tempdir().unwrap();
|
||||
|
|
|
|||
|
|
@ -91,6 +91,7 @@ pub fn player_surface_for_actor(
|
|||
.as_ref()
|
||||
.map(|cover| cover.effective_budget)
|
||||
.unwrap_or(campaign.resolution_policy.active_cell_budget);
|
||||
let strategic_wait_minutes = u32::from(campaign.tick_hours).saturating_mul(60);
|
||||
let mut children = vec![
|
||||
json!({"id":"dungeon.status","kind":"card","props":{"title":format!("{} · revision {} · {}",campaign.name,campaign.revision,campaign.world_time)},"children":[]}),
|
||||
json!({"id":"dungeon.location","kind":"card","props":{"title":format!("{} · {}",location.name,player.name)},"children":[]}),
|
||||
|
|
@ -115,7 +116,7 @@ pub fn player_surface_for_actor(
|
|||
json!({"expected_revision":campaign.revision}),
|
||||
&["description", "intended_effect"],
|
||||
),
|
||||
json!({"id":"dungeon.wait.minutes","kind":"control.input.number","props":{"label":"Wait minutes","value":30,"min":1,"max":1440},"stateBindings":[local_draft("minutes","number")],"children":[]}),
|
||||
json!({"id":"dungeon.wait.minutes","kind":"control.input.number","props":{"label":format!("Wait minutes ({} max)",strategic_wait_minutes),"value":30_u32.min(strategic_wait_minutes),"min":1,"max":strategic_wait_minutes},"stateBindings":[local_draft("minutes","number")],"children":[]}),
|
||||
command_control(
|
||||
"dungeon.wait",
|
||||
"Wait",
|
||||
|
|
@ -164,10 +165,15 @@ pub fn player_surface_for_actor(
|
|||
if location.routes.is_empty() {
|
||||
children.push(json!({"id":"dungeon.travel.none","kind":"text","props":{"value":"No compiled route leaves this location yet."},"children":[]}));
|
||||
} else {
|
||||
let default_destination_id = location
|
||||
.routes
|
||||
.keys()
|
||||
.next()
|
||||
.expect("non-empty routes have a first destination");
|
||||
children.push(json!({
|
||||
"id":"dungeon.travel.destination",
|
||||
"kind":"control.select",
|
||||
"props":{"label":"Group destination"},
|
||||
"props":{"label":"Group destination","value":default_destination_id},
|
||||
"stateBindings":[local_draft("destination_location_id","choice")],
|
||||
"children":location.routes.iter().filter_map(|(destination_id,route)|campaign.locations.get(destination_id).map(|destination|json!({
|
||||
"id":format!("dungeon.travel.option.{destination_id}"),
|
||||
|
|
@ -599,6 +605,7 @@ mod tests {
|
|||
assert!(encoded.contains(&format!("\"bindingName\":\"{binding}\"")));
|
||||
}
|
||||
assert!(encoded.contains("governance.travel.propose"));
|
||||
assert!(encoded.contains("\"value\":\"harbor\""));
|
||||
assert!(encoded.contains("world.destination.compile"));
|
||||
assert!(encoded.contains("campaign.entry"));
|
||||
assert!(!encoded.contains("payload.fields"));
|
||||
|
|
|
|||
Loading…
Reference in a new issue