feat(runtime): run resource schedules during world steps

5 files changed, 245 insertions, 49 deletions

diff --git a/crates/fparkan-runtime/src/lib.rs b/crates/fparkan-runtime/src/lib.rs
index e74976f..180533a 100644
--- a/crates/fparkan-runtime/src/lib.rs
+++ b/crates/fparkan-runtime/src/lib.rs
@@ -1560,27 +1560,36 @@ fn prepare_first_preview_assets<R: ResourceRepository>(
     Ok(MissionAssets::default())
 }
 
-/// Steps headless mode.
+/// Steps the world and due Behavior resource distributors at an explicit queue clock.
+///
+/// Resource updates are prepared before queued world commands and deferred deletion,
+/// and published only after the world step succeeds. This does not yet execute the
+/// separate Control clock, movement, or general mission script handlers.
 ///
 /// # Errors
 ///
-/// Returns [`EngineError`] when the world step fails.
+/// Returns [`EngineError`] when the resource or world step fails; neither stage
+/// publishes partial state on failure.
 pub fn step_headless(
     engine: &mut Engine,
     input: InputSnapshot,
+    queue_time_ms: u32,
 ) -> Result<FrameResult, EngineError> {
-    Ok(FrameResult {
-        snapshot: step(&mut engine.world, &input)?,
-    })
+    let resources = prepare_loaded_mission_resources(engine, queue_time_ms)?;
+    let snapshot = step(&mut engine.world, &input)?;
+    if let Some(resources) = resources {
+        resources.apply(engine);
+    }
+    Ok(FrameResult { snapshot })
 }
 
-/// Steps rendered mode.
+/// Steps rendered mode with the same queue-clock semantics as [`step_headless`].
 ///
 /// # Errors
 ///
-/// Returns [`EngineError`] when the world step fails.
-pub fn frame(engine: &mut Engine) -> Result<FrameResult, EngineError> {
-    step_headless(engine, InputSnapshot)
+/// Returns [`EngineError`] when the resource or world step fails.
+pub fn frame(engine: &mut Engine, queue_time_ms: u32) -> Result<FrameResult, EngineError> {
+    step_headless(engine, InputSnapshot, queue_time_ms)
 }
 
 /// Returns the loaded mission summary.
@@ -1962,9 +1971,65 @@ pub fn advance_loaded_mission_resources(
     engine: &mut Engine,
     queue_time_ms: u32,
 ) -> Result<usize, EngineError> {
-    let Some(loaded) = engine.loaded.as_ref() else {
+    let Some(resources) = prepare_loaded_mission_resources(engine, queue_time_ms)? else {
         return Ok(0);
     };
+    let scheduled_ticks = resources.scheduled_ticks;
+    resources.apply(engine);
+    Ok(scheduled_ticks)
+}
+
+struct PreparedMissionResources {
+    distributors: Vec<BehaviorDistributorState>,
+    ore_states: Vec<BehaviorOreState>,
+    control_tables: Vec<MissionControlDeviceTableState>,
+    warm_starts: Vec<Option<BehaviorWarmStartState>>,
+    rng_state: u32,
+    scheduled_ticks: usize,
+}
+
+impl PreparedMissionResources {
+    fn apply(self, engine: &mut Engine) {
+        let loaded = engine
+            .loaded
+            .as_mut()
+            .expect("prepared resources require a loaded mission");
+        loaded.behavior_distributors = self.distributors;
+        loaded.behavior_ore_states = self.ore_states;
+        loaded.control_device_tables = self.control_tables;
+        for (bundle, warm_start) in loaded.script_bundles.iter_mut().zip(self.warm_starts) {
+            bundle.behavior_warm_start = warm_start;
+        }
+        engine.behavior_rng_state = self.rng_state;
+    }
+}
+
+fn prepare_loaded_mission_resources(
+    engine: &Engine,
+    queue_time_ms: u32,
+) -> Result<Option<PreparedMissionResources>, EngineError> {
+    let Some(loaded) = engine.loaded.as_ref() else {
+        return Ok(None);
+    };
+    if loaded
+        .behavior_distributors
+        .iter()
+        .filter(|distributor| distributor.super_ai_visible)
+        .all(|distributor| {
+            loaded
+                .script_bundles
+                .iter()
+                .find(|bundle| bundle.clan_index == distributor.clan_index)
+                .and_then(|bundle| bundle.behavior_warm_start)
+                .is_some_and(|schedule| {
+                    schedule.first_time_ms != 0
+                        && schedule.next_deadline_ms != 0
+                        && queue_time_ms <= schedule.next_deadline_ms
+                })
+        })
+    {
+        return Ok(None);
+    }
 
     let mut distributors = loaded.behavior_distributors.clone();
     let mut ore_states = loaded.behavior_ore_states.clone();
@@ -2023,18 +2088,14 @@ pub fn advance_loaded_mission_resources(
         schedule.cleared_words = totals;
     }
 
-    let loaded = engine
-        .loaded
-        .as_mut()
-        .expect("loaded mission was checked above");
-    loaded.behavior_distributors = distributors;
-    loaded.behavior_ore_states = ore_states;
-    loaded.control_device_tables = control_tables;
-    for (bundle, warm_start) in loaded.script_bundles.iter_mut().zip(warm_starts) {
-        bundle.behavior_warm_start = warm_start;
-    }
-    engine.behavior_rng_state = rng_state;
-    Ok(scheduled_ticks)
+    Ok(Some(PreparedMissionResources {
+        distributors,
+        ore_states,
+        control_tables,
+        warm_starts,
+        rng_state,
+        scheduled_ticks,
+    }))
 }
 
 /// Sets native Control global-table selector 1 on rows whose selector-0 value
@@ -5169,7 +5230,7 @@ mod tests {
     #[test]
     fn load_mission_requires_vfs_and_keeps_world_unchanged_on_error() {
         let mut engine = create(EngineServices::default()).expect("engine");
-        let before = step_headless(&mut engine, InputSnapshot).expect("step");
+        let before = step_headless(&mut engine, InputSnapshot, 0).expect("step");
         let err = load_mission(
             &mut engine,
             MissionRequest {
@@ -5180,7 +5241,7 @@ mod tests {
         )
         .expect_err("missing VFS");
         assert!(matches!(err, EngineError::MissingVfs));
-        let after = step_headless(&mut engine, InputSnapshot).expect("step");
+        let after = step_headless(&mut engine, InputSnapshot, 0).expect("step");
         assert_eq!(before.snapshot.objects, after.snapshot.objects);
     }
 
@@ -6911,6 +6972,116 @@ mod tests {
 
     #[test]
     #[ignore = "requires licensed corpus"]
+    fn headless_step_runs_resource_schedule_and_rolls_back_both_error_paths() {
+        let mut engine = create(EngineServices::new(Arc::new(DirectoryVfs::new(
+            licensed_root("IS"),
+        ))))
+        .expect("engine");
+        load_mission(
+            &mut engine,
+            MissionRequest {
+                key: "MISSIONS/Autodemo.00/data.tma".into(),
+                mode: MissionMode::AutoDemo,
+            },
+            100_000,
+        )
+        .expect("licensed mission");
+        let schedules = |engine: &Engine| {
+            loaded_mission_script_bundles(engine)
+                .unwrap()
+                .iter()
+                .map(|bundle| bundle.behavior_warm_start)
+                .collect::<Vec<_>>()
+        };
+        let initial = schedules(&engine);
+        frame(&mut engine, 100_200).expect("deadline equality");
+        assert_eq!(schedules(&engine), initial);
+        let result = step_headless(&mut engine, InputSnapshot, 100_201).expect("due clan 1");
+        assert_eq!(result.snapshot.tick.0, 2);
+        let after = schedules(&engine);
+        assert_eq!(after[0], initial[0]);
+        let updated = after[1].expect("clan 1 schedule");
+        assert_eq!(updated.first_time_ms, 100_201);
+        assert!(f32::from_bits(updated.cleared_words[0]) > 0.0);
+
+        let resources = |engine: &Engine| {
+            let state = engine.loaded.as_ref().unwrap();
+            (
+                state.behavior_distributors.clone(),
+                state.behavior_ore_states.clone(),
+                state.control_device_tables.clone(),
+                schedules(engine),
+                engine.behavior_rng_state,
+            )
+        };
+        let saved_controls = engine
+            .loaded
+            .as_ref()
+            .unwrap()
+            .control_device_tables
+            .clone();
+        let plant = engine
+            .loaded
+            .as_ref()
+            .unwrap()
+            .behavior_distributors
+            .iter()
+            .find(|distributor| distributor.clan_index == 1)
+            .unwrap()
+            .registered_objects
+            .iter()
+            .find(|object| object.class_flags == 0x8000_0010)
+            .unwrap()
+            .handle;
+        engine
+            .loaded
+            .as_mut()
+            .unwrap()
+            .control_device_tables
+            .retain(|table| table.handle != plant);
+        let before = resources(&engine);
+        let world_before = fparkan_world::canonical_state_hash(&engine.world);
+        assert!(matches!(
+            step_headless(&mut engine, InputSnapshot, 100_500),
+            Err(EngineError::BehaviorDistributorTick { clan_index: 1, .. })
+        ));
+        assert_eq!(resources(&engine), before);
+        assert_eq!(
+            fparkan_world::canonical_state_hash(&engine.world),
+            world_before
+        );
+
+        engine.loaded.as_mut().unwrap().control_device_tables = saved_controls;
+        let stale = construct_object(&mut engine.world, ObjectDraft { original_id: None }).unwrap();
+        register_object(&mut engine.world, stale).unwrap();
+        fparkan_world::enqueue(
+            &mut engine.world,
+            fparkan_world::WorldCommand {
+                sequence: 0,
+                target: Some(stale),
+            },
+        )
+        .unwrap();
+        fparkan_world::request_delete(&mut engine.world, stale).unwrap();
+        let before = resources(&engine);
+        let world_before = fparkan_world::canonical_state_hash(&engine.world);
+        assert!(matches!(
+            step_headless(&mut engine, InputSnapshot, 100_500),
+            Err(EngineError::World(_))
+        ));
+        assert_eq!(
+            resources(&engine),
+            before,
+            "a failed world step must not publish staged resource ticks"
+        );
+        assert_eq!(
+            fparkan_world::canonical_state_hash(&engine.world),
+            world_before
+        );
+    }
+
+    #[test]
+    #[ignore = "requires licensed corpus"]
     fn licensed_autodemo_resource_ticks_use_queue_time_and_real_generator_profile() {
         let root = licensed_root("IS");
         let vfs: Arc<dyn Vfs> = Arc::new(DirectoryVfs::new(root));
@@ -7714,7 +7885,7 @@ mod tests {
         )
         .expect("static preview does not require Behavior profile parsing");
         let previous_mission = loaded_mission(&engine).expect("loaded preview").clone();
-        let before = step_headless(&mut engine, InputSnapshot)
+        let before = step_headless(&mut engine, InputSnapshot, 0)
             .expect("step preview world")
             .snapshot
             .objects;
@@ -7726,7 +7897,7 @@ mod tests {
             EngineError::Nres { ref path, .. } if path == "behpsp.res"
         ));
         assert_eq!(loaded_mission(&engine), Some(&previous_mission));
-        let after = step_headless(&mut engine, InputSnapshot)
+        let after = step_headless(&mut engine, InputSnapshot, 0)
             .expect("step retained preview world")
             .snapshot
             .objects;
@@ -7921,7 +8092,7 @@ mod tests {
                 denied,
             });
             let mut engine = create(EngineServices::new(vfs)).expect("engine");
-            let before = step_headless(&mut engine, InputSnapshot).expect("before");
+            let before = step_headless(&mut engine, InputSnapshot, 0).expect("before");
             let err = load_mission(&mut engine, mission, 0).expect_err("load error");
             match denied {
                 DenyRule::Suffix("Land.map") => assert!(matches!(err, EngineError::Vfs { .. })),
@@ -7931,7 +8102,7 @@ mod tests {
                 DenyRule::Suffix(unexpected) => panic!("unexpected deny rule {unexpected}"),
             }
             assert!(loaded_mission(&engine).is_none());
-            let after = step_headless(&mut engine, InputSnapshot).expect("after");
+            let after = step_headless(&mut engine, InputSnapshot, 0).expect("after");
             assert_eq!(before.snapshot.objects, after.snapshot.objects);
         }
     }
@@ -7942,7 +8113,7 @@ mod tests {
         let root = licensed_root("IS");
         let vfs: Arc<dyn Vfs> = Arc::new(DirectoryVfs::new(root));
         let mut engine = create(EngineServices::new(vfs)).expect("engine");
-        let before = step_headless(&mut engine, InputSnapshot).expect("before");
+        let before = step_headless(&mut engine, InputSnapshot, 0).expect("before");
 
         let err = load_mission_with_options(
             &mut engine,
@@ -7967,7 +8138,7 @@ mod tests {
             }
         ));
         assert!(loaded_mission(&engine).is_none());
-        let after = step_headless(&mut engine, InputSnapshot).expect("after");
+        let after = step_headless(&mut engine, InputSnapshot, 0).expect("after");
         assert_eq!(before.snapshot.objects, after.snapshot.objects);
     }
 
@@ -8011,7 +8182,7 @@ mod tests {
             .expect("previous Behavior distributors")
             .to_vec();
         let previous_rng = engine.behavior_rng_state;
-        let before = step_headless(&mut engine, InputSnapshot)
+        let before = step_headless(&mut engine, InputSnapshot, 0)
             .expect("step previous world")
             .snapshot
             .objects;
@@ -8038,7 +8209,7 @@ mod tests {
             Some(previous_distributors.as_slice())
         );
         assert_eq!(engine.behavior_rng_state, previous_rng);
-        let after = step_headless(&mut engine, InputSnapshot)
+        let after = step_headless(&mut engine, InputSnapshot, 0)
             .expect("step retained world")
             .snapshot
             .objects;
@@ -8217,7 +8388,7 @@ mod tests {
         let mut snapshot = None;
         for _ in 0..10_000 {
             snapshot = Some(
-                step_headless(&mut engine, InputSnapshot)
+                step_headless(&mut engine, InputSnapshot, 0)
                     .expect("selected mission deterministic tick")
                     .snapshot,
             );
diff --git a/docs/appendices/knowledge-boundaries.md b/docs/appendices/knowledge-boundaries.md
index 58deb31..00b261c 100644
--- a/docs/appendices/knowledge-boundaries.md
+++ b/docs/appendices/knowledge-boundaries.md
@@ -380,11 +380,14 @@ RVA `0xbcf0` для остальных. Это подтверждает стат
 
 Отдельный API `advance_loaded_mission_resources` принимает явное время очереди
 и обновляет представленные поля `CurrentOre`, строк распределителя и Control
-selector `1`. API не подключён к игровому циклу и не обновляет Control amount,
-профили, special object-state gate или жизненный цикл объектов. Если обязательный
+selector `1`. Тот же расчёт вызывают `step_headless` и `frame` с явным
+clock очереди; подготовка ресурсов и мировой шаг завершаются атомарно.
+Оконное приложение пока использует static preview. Этот путь не обновляет
+Control amount, профили или special object-state gate. Если обязательный
 запрос позднего объекта не поддержан, транзакция возвращает ошибку без применения
 подготовленных изменений. AutoDemo-тест подтверждает связь настоящих профилей с
-этим шагом, но не полный паритет миссии; отдельный aggregate getter Behavior RVA
+этим шагом и проверяет откат при ошибке ресурсов или мировой очереди,
+но не полный паритет миссии; отдельный aggregate getter Behavior RVA
 `0xa490` в oracle не исполнен.
 
 Behavior selectors `0x1002` и `0x1005` используют значения выбранного
diff --git a/docs/appendices/script-vm.md b/docs/appendices/script-vm.md
index 69ea54f..dc42959 100644
--- a/docs/appendices/script-vm.md
+++ b/docs/appendices/script-vm.md
@@ -408,9 +408,11 @@ bundle, но остальные потребители исходного гло
 В Rust также нет значения аргумента в `AI+0x3fc`, recursion guard `AI+0x78`
 или одноразового latch `AI+0x8a8`. Поэтому это связанный начальный поднабор, а
 не полная активация SuperAI. После полной загрузки узкая Rust-реализация
-вызывается явно через `advance_loaded_mission_resources(engine, queue_time_ms)`;
-runtime не читает системную очередь, а приложение пока не вызывает API из
-игрового цикла. Шаг использует разрешённые handles и представленные snapshots
+вызывается из `step_headless(engine, input, queue_time_ms)` и
+`frame(engine, queue_time_ms)`. Отдельный
+`advance_loaded_mission_resources(engine, queue_time_ms)` остаётся доступен
+для прямого ресурсного шага. Runtime не читает системную очередь, а оконное
+приложение пока использует static preview. Шаг использует разрешённые handles и представленные snapshots
 профилей и Control, поэтому не воспроизводит отсутствующие в них callbacks. При
 ошибке ограниченного `Init` Rust отклоняет полную загрузку и сохраняет прежний
 мир и состояние генератора; такая политика отката принадлежит Rust loader-у.
@@ -479,8 +481,14 @@ selector `1` на интерфейсе получателя становится
 только для полного режима; preview остаётся статическим. Явный API
 `advance_loaded_mission_resources` использует очередь времени, расписание,
 профили и представленные Control-таблицы, чтобы исполнить этот ограниченный
-шаг над загруженным состоянием. `step_headless` по-прежнему шагает только
-`World`.
+шаг над загруженным состоянием. `step_headless` и `frame` требуют явный
+`u32` clock и используют ту же подготовку: ресурсные изменения, расписания и
+RNG остаются временными до успешного `World::step`. Ошибка ресурсов не
+продвигает мир; ошибка мировой очереди не публикует ресурсные изменения.
+Игнорируемый тест `headless_step_runs_resource_schedule_and_rolls_back_both_error_paths`
+на AutoDemo проверяет равенство срока, исполнение следующего шага и обе ветви
+отказа, включая отсутствие Control-таблицы у позднего клана и устаревший handle
+в очереди. Это не запускает отдельный clock Control или полную миссионную VM.
 
 ### Добавление и нормализация долей
 
diff --git a/docs/tomes/02-architecture.md b/docs/tomes/02-architecture.md
index 0d0ea11..527d7b2 100644
--- a/docs/tomes/02-architecture.md
+++ b/docs/tomes/02-architecture.md
@@ -363,6 +363,13 @@ CollObject, Control и PhysicalModel, пропуская отсутствующ�
 Статическая цепочка ещё не доказывает исполнение активного Control update на
 миссии. Rust пока сохраняет начальные provider-снимки и не изменяет их этим путём.
 
+В текущем runtime `step_headless` и `frame` получают время очереди явным
+аргументом `u32`. По нему запускается представленный ресурсный планировщик
+Behavior, после чего обрабатываются очередь мира и отложенные удаления.
+Ресурсные изменения публикуются после успешного мирового шага, поэтому ошибка
+не оставляет мир и ресурсы на разных шагах. Это начало расчётного цикла;
+отдельное обновление Control, общая VM и движение ещё требуют подключения.
+
 ## World3D
 
 World3D связывает игровые объекты, события, время, ввод, камеру, сетевые
diff --git a/docs/tomes/06-behavior.md b/docs/tomes/06-behavior.md
index 0ebbbf7..84f4f6e 100644
--- a/docs/tomes/06-behavior.md
+++ b/docs/tomes/06-behavior.md
@@ -202,17 +202,17 @@ SuperAI для записей кланов, затем обновляет Behavi
 
 Это полезное начало сценария, но ещё не целая игровая жизнь AI. Текущий runtime
 не воспроизводит аргумент обратного вызова в `AI+0x3fc`, защиту от повторного
-входа `AI+0x78`, одноразовую метку `AI+0x8a8` или обновления Behavior на игровых
-тиках. При этом один конкретный шаг Behavior — распределение ресурсов — уже
-исполнен на исходной DLL с искусственными объектами. Rust пока не вызывает
-регулярный шаг: игровой мир ещё не предоставляет нужные ресурсные интерфейсы.
+входа `AI+0x78`, одноразовую метку `AI+0x8a8` или полное обновление Behavior на игровых
+тиках. Распределение ресурсов исполнено на исходной DLL с искусственными
+объектами и подключено к Rust `step_headless`: явное время очереди определяет,
+какой клан должен выполнить очередной ресурсный шаг.
 При полной загрузке Rust уже восстанавливает членство объектов в клановых
 списках Behavior и задаёт их начальные доли. Остальные слова записи хранятся
 сырыми: семантика только первого из пяти слов хвоста установлена для
 нормализации.
 Runtime также не подменяет очередь оригинала: вызывающая сторона сама передаёт
 время, а локальный генератор Rust пока учитывает только случайные числа
-восстановленного стартового шага.
+восстановленного стартового шага и ресурсного планировщика.
 
 ### Откуда берутся объекты распределителя
 
@@ -355,9 +355,16 @@ Control selector `1`; неподдержанное свойство, запро�
 арифметическое значение возвращает ошибку. Обновления нескольких кланов, их
 расписания и RNG фиксируются одной транзакцией: если обязательный запрос
 позднего клана не разрешён, ранее подготовленные изменения остаются
-неприменёнными. Приложение пока не вызывает
-этот API из игрового цикла, а динамические callbacks, которые переносят
-количество между Control provider-строками, ещё не моделируются. Поэтому это
+неприменёнными. Обычные `step_headless(engine, input, queue_time_ms)` и
+`frame(engine, queue_time_ms)` запускают этот расчёт до обработки очереди мира.
+Они сначала готовят ресурсные изменения, затем выполняют шаг мира и публикуют
+ресурсное состояние после его успеха. Ошибка любой стадии сохраняет прежние
+ресурсы; ошибка подготовки также не продвигает мировой tick. До срока
+планировщика большие таблицы ресурсов не копируются.
+
+Оконное приложение пока запускает static preview и не вызывает этот расчётный
+путь. Динамические callbacks, которые переносят количество между Control
+provider-строками, ещё не моделируются. Поэтому это
 исполняемый загруженный resource step, а не полный симулятор миссии.
 
 Для настоящих ресурсов AutoDemo тест проходит несколько временных точек: