From 0764842f6bd0d78ce2cfd00b13ffb3097ddd3f8c Mon Sep 17 00:00:00 2001 From: Meta Vens Date: Wed, 16 Sep 2026 19:19:34 +0200 Subject: [PATCH] Return a session panic from the sweep instead of unwinding it (L1 Cut 4, c4.s1.f4) A panic inside one elaboration session used to resume-unwind out of sweep, abort its siblings mid-inference, and end the runtime's elaboration driver task, whose handle is dropped: elaboration stopped silently until restart. Now each session holds its permit inside a SessionPermit guard whose Drop raises the sweep's stop before the permit is released, on every exit but a clean return. A quarantine-class error disarms nothing, and a panicking task never reaches the line that disarms it, so no session starts after a panic, by the same path that stops sessions after a quarantine-class error. The join loop turns a task that did not return into ControllerError::Serialization("an elaboration session did not return: ...") and returns it after every started sibling has finished. The class is quarantine-class on purpose: sweep returns only quarantine-class errors and debug-logs the rest, so this keeps that one rule, and the driver's handling of any returned error stays exactly as it is (debug log). Whether that error quarantines cognition is L1.f1's, and is not decided here. Serialization is the variant the elaboration lane already uses for its local invariants. elaborate_world takes its interval, as drive_cover_tick does, so a test can drive it at one millisecond; production passes ELABORATION_SWEEP_INTERVAL unchanged. Tests: a_panicking_session_lets_its_sibling_finish (the held sibling commits in the panicking sweep, the error is returned, permits are free, and a second sweep resumes and commits the panicked session); no_session_starts_after_a_panic; and in Dungeon the_elaboration_driver_survives_a_panicking_sweep, which waits for a second sweep's listing and inference against the driver task ending. Co-Authored-By: Claude Opus 5 --- crates/ghostlight-dungeon/src/runtime.rs | 122 ++++++++++++++++++++- crates/ghostlight/src/controllers.rs | 130 +++++++++++++++++++++++ crates/ghostlight/src/elaboration.rs | 42 ++++++-- 3 files changed, 283 insertions(+), 11 deletions(-) diff --git a/crates/ghostlight-dungeon/src/runtime.rs b/crates/ghostlight-dungeon/src/runtime.rs index 33620a4..6dd9f82 100644 --- a/crates/ghostlight-dungeon/src/runtime.rs +++ b/crates/ghostlight-dungeon/src/runtime.rs @@ -407,7 +407,7 @@ pub(crate) async fn run(state_root_binding: Option) -> anyhow::Result<( }; tokio::spawn(maintain_mesh_projection(state.clone())); tokio::spawn(drive_cover_tick(state.clone(), configured_tick_interval())); - tokio::spawn(elaborate_world(state.clone())); + tokio::spawn(elaborate_world(state.clone(), ELABORATION_SWEEP_INTERVAL)); let app = app_router(state.clone(), web_root); tracing::info!(address = %bound_address, "Ghostlight Dungeon world owner serving"); let server = axum::serve( @@ -1831,8 +1831,8 @@ const ELABORATION_SWEEP_INTERVAL: Duration = Duration::from_secs(300); /// all, which is the config gate the runtime already has: no mode flag joins it. /// Sweeps never overlap, because this loop awaits one before it ticks the next, /// so one sweep's claim set is the only set of sessions alive. -async fn elaborate_world(state: AppState) { - let mut interval = tokio::time::interval(ELABORATION_SWEEP_INTERVAL); +async fn elaborate_world(state: AppState, interval: Duration) { + let mut interval = tokio::time::interval(interval); interval.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Skip); interval.tick().await; loop { @@ -3594,6 +3594,122 @@ mod tests { assert_eq!(state.elaboration_permits.available_permits(), elaboration); } + /// Panics on its first `infer` call. Every later call announces itself on + /// `again` and answers with text. + struct PanicOnceInferencePort { + first: AtomicBool, + again: Semaphore, + } + + #[async_trait::async_trait] + impl InferencePort for PanicOnceInferencePort { + fn prepare(&self, request: InferenceRequest) -> Result { + PreparedInference::prepare("ghostlight-controller-test", 4_102_444_800_000, request) + } + + async fn infer( + &self, + _request: PreparedInference, + ) -> Result { + if self.first.swap(false, Ordering::SeqCst) { + panic!("the fixture port panics inside the first elaboration session"); + } + self.again.add_permits(1); + Ok(InferenceOutput::new( + vec![InferenceEvent::Text("The driver fixture speaks.".into())], + "sha256:panic-once-port", + )) + } + } + + /// `AlwaysFreshWorkStore`, announcing each in-flight listing: one per sweep. + struct SweepCountingWorkStore { + listed: Semaphore, + } + + #[async_trait::async_trait] + impl ControllerWorkStore for SweepCountingWorkStore { + async fn lookup( + &self, + command_id: CommandId, + ) -> Result { + AlwaysFreshWorkStore.lookup(command_id).await + } + + async fn persist( + &self, + work: &ControllerWork, + ) -> Result { + AlwaysFreshWorkStore.persist(work).await + } + + async fn custody_probe(&self) -> Result { + AlwaysFreshWorkStore.custody_probe().await + } + + async fn elaboration_in_flight( + &self, + ) -> Result, ControllerWorkStoreError> { + self.listed.add_permits(1); + AlwaysFreshWorkStore.elaboration_in_flight().await + } + } + + /// The elaboration driver outlives a sweep whose session panicked. The + /// driver runs at a one-millisecond interval over a world with a deficit, + /// under an elaboration pool of one, so a sweep that stops after a panic + /// has no sibling session. The first sweep's session panics; the driver + /// then reads a second listing, which only a second sweep does, and that + /// sweep's session reaches the provider. If the driver task ends first, + /// the test fails on that instead of waiting. + #[tokio::test] + async fn the_elaboration_driver_survives_a_panicking_sweep() { + let fixture = fixture().await; + two_cell_world( + &fixture.state, + &fixture.cookie, + BTreeMap::from([(SubjectKind::Person, 4)]), + vec![CreateJurisdictionIntent { + handle: "sere".into(), + label: "The Low Sere".into(), + permille: 1000, + }], + true, + ) + .await; + let port = Arc::new(PanicOnceInferencePort { + first: AtomicBool::new(true), + again: Semaphore::new(0), + }); + let store = Arc::new(SweepCountingWorkStore { + listed: Semaphore::new(0), + }); + let mut state = fixture.state.clone(); + state.controllers = Some(Arc::new( + ControllerRunner::open( + state.world.clone(), + port.clone(), + store.clone(), + test_controller_models(), + ) + .expect("the fixture ports open"), + )); + state.elaboration_permits = Arc::new(Semaphore::new(1)); + + let mut driver = tokio::spawn(elaborate_world(state.clone(), Duration::from_millis(1))); + tokio::select! { + listed = store.listed.acquire_many(2) => listed.unwrap().forget(), + ended = &mut driver => panic!("the elaboration driver ended: {ended:?}"), + } + assert!(!port.first.load(Ordering::SeqCst), "the first sweep's session never panicked"); + tokio::select! { + again = port.again.acquire() => again.unwrap().forget(), + ended = &mut driver => panic!("the elaboration driver ended: {ended:?}"), + } + assert!(!state.controller_quarantined.load(Ordering::SeqCst)); + driver.abort(); + } + /// Raises `ControllerError::requires_quarantine` on every call. Used to /// prove the tick driver's quarantine edge rather than any cognition /// outcome. diff --git a/crates/ghostlight/src/controllers.rs b/crates/ghostlight/src/controllers.rs index de19a35..10432f7 100644 --- a/crates/ghostlight/src/controllers.rs +++ b/crates/ghostlight/src/controllers.rs @@ -11489,6 +11489,136 @@ mod tests { task.await.unwrap(); } + /// Holds the first session on each of two dead ends in the provider + /// boundary together, then panics the one on the Unwalked Road while the + /// Far Road's waits for the test. Every later call answers its road with a + /// shed. + struct PanickingPort { + both: tokio::sync::Barrier, + panicking: tokio::sync::Semaphore, + release: tokio::sync::Semaphore, + first_unwalked: AtomicBool, + first_far: AtomicBool, + roads: Box Result + Send + Sync>, + calls: std::sync::atomic::AtomicUsize, + } + + #[async_trait] + impl InferencePort for PanickingPort { + fn prepare(&self, request: InferenceRequest) -> Result { + PreparedInference::prepare("ghostlight-controller-test", 4_102_444_800_000, request) + } + + async fn infer( + &self, + request: PreparedInference, + ) -> Result { + let call = self.calls.fetch_add(1, Ordering::SeqCst); + if prompt_of(&request).contains("The Unwalked Road") { + if self.first_unwalked.swap(false, Ordering::SeqCst) { + self.both.wait().await; + self.panicking.add_permits(1); + panic!("the fixture port panics inside one session"); + } + } else if self.first_far.swap(false, Ordering::SeqCst) { + self.both.wait().await; + self.release.acquire().await.unwrap().forget(); + } + (self.roads)(call, &request) + } + } + + /// A session that panics does not take its sweep down with it. Two + /// sessions are in the provider boundary together; one panics while the + /// other is still held there. The held session finishes its inference and + /// commits in the same sweep, the sweep returns a quarantine-class error + /// naming the panic, and every permit is free. A second sweep then runs + /// normally: it resumes the panicked session's row and commits it. + #[tokio::test] + async fn a_panicking_session_lets_its_sibling_finish() { + let (_directory, mailbox, task, _commons, roads) = dead_end_world( + crate::tests::stock_weights(), + &["The Unwalked Road", "The Far Road"], + Roots::Commons, + 0, + ) + .await; + let port = Arc::new(PanickingPort { + both: tokio::sync::Barrier::new(2), + panicking: tokio::sync::Semaphore::new(0), + release: tokio::sync::Semaphore::new(0), + first_unwalked: AtomicBool::new(true), + first_far: AtomicBool::new(true), + roads: Box::new(shed_on_the_named_road(vec![ + ("The Unwalked Road", roads[0]), + ("The Far Road", roads[1]), + ])), + calls: std::sync::atomic::AtomicUsize::new(0), + }); + let permits = pool(2); + let store = fresh_store(); + let runner = sweep_runner(&mailbox, port.clone(), store.clone()); + let sweep = tokio::spawn({ + let runner = runner.clone(); + let permits = permits.clone(); + async move { runner.sweep(permits).await } + }); + port.panicking.acquire().await.unwrap().forget(); + port.release.add_permits(1); + let Err(error) = sweep.await.expect("a session's panic ended the sweep") else { + panic!("a panicking session did not end the sweep with an error"); + }; + assert!(error.requires_quarantine(), "{error:?}"); + assert!(error.to_string().contains("panic"), "{error}"); + let snapshot = mailbox.snapshot().await.unwrap(); + assert!( + place_labelled(&snapshot, "The Shed on The Far Road").is_some(), + "the held sibling did not commit in the panicking sweep" + ); + assert_eq!(snapshot.boundaries.len(), 1); + assert_eq!(port.calls.load(Ordering::SeqCst), 2); + assert_eq!(permits.available_permits(), 2, "a permit outlived its session"); + + runner.sweep(permits.clone()).await.expect("the sweep after a panic did not run"); + assert_eq!(port.calls.load(Ordering::SeqCst), 3); + let snapshot = mailbox.snapshot().await.unwrap(); + assert!(snapshot.boundaries.is_empty()); + assert!(place_labelled(&snapshot, "The Shed on The Unwalked Road").is_some()); + assert_eq!(store.work.lock().unwrap().len(), 2, "the resumed session opened a row"); + assert_eq!(permits.available_permits(), 2); + drop(runner); + drop(mailbox); + task.await.unwrap(); + } + + /// No session starts after one has panicked. Under a pool of one the + /// second entry waits for the first session's permit; the first session + /// panics, and the second never reaches the provider boundary. + #[tokio::test] + async fn no_session_starts_after_a_panic() { + let (_directory, mailbox, task, _commons, _roads) = dead_end_world( + crate::tests::stock_weights(), + &["The Unwalked Road", "The Far Road"], + Roots::Commons, + 0, + ) + .await; + let port = Arc::new(SweepPort::new(|_, _| { + panic!("the fixture port panics inside every session") + })); + let permits = pool(1); + let runner = sweep_runner(&mailbox, port.clone(), fresh_store()); + let Err(error) = runner.sweep(permits.clone()).await else { + panic!("a panicking session did not end the sweep with an error"); + }; + assert!(error.requires_quarantine(), "{error:?}"); + assert_eq!(port.calls(), 1, "a session started after a panic"); + assert_eq!(permits.available_permits(), 1); + drop(runner); + drop(mailbox); + task.await.unwrap(); + } + /// Probe 3's deficit half, through the sweep. Sweep one's deficit session /// is refused; a clock tick moves the ancestry, and with it the id a fresh /// derivation would give the deficit. Sweep two finds the refused session by diff --git a/crates/ghostlight/src/elaboration.rs b/crates/ghostlight/src/elaboration.rs index 534eef0..9dc48dd 100644 --- a/crates/ghostlight/src/elaboration.rs +++ b/crates/ghostlight/src/elaboration.rs @@ -639,7 +639,9 @@ impl ElaborationRunner { /// /// Every spawned session finishes before this returns. A quarantine-class /// error is returned after that, and no session starts after one has - /// returned such an error; any other error ends only its own session. + /// returned such an error. A session that panics is that same error: its + /// siblings finish, nothing starts after it, and it is returned. Any other + /// error ends only its own session. pub async fn sweep(&self, permits: Arc) -> Result<(), ControllerError> { let snapshot = self.mailbox.snapshot().await.map_err(snapshot_error)?; if snapshot.phase != WorldPhase::Active { @@ -666,14 +668,16 @@ impl ElaborationRunner { let snapshot = snapshot.clone(); let stopped = stopped.clone(); running.spawn(async move { + let mut held = SessionPermit { + _permit: permit, + stopped, + stop: true, + }; let outcome = runner.step_session(command_id, session, &snapshot).await; - if outcome + held.stop = outcome .as_ref() - .is_err_and(ControllerError::requires_quarantine) - { - stopped.store(true, Ordering::SeqCst); - } - drop(permit); + .is_err_and(ControllerError::requires_quarantine); + drop(held); outcome }); } @@ -687,7 +691,11 @@ impl ElaborationRunner { Ok(Err(error)) => { tracing::debug!(%error, "elaboration step did not admit a patch"); } - Err(error) => std::panic::resume_unwind(error.into_panic()), + Err(error) => { + quarantine.get_or_insert(ControllerError::Serialization(format!( + "an elaboration session did not return: {error}" + ))); + } } } quarantine.map_or(Ok(()), Err) @@ -717,6 +725,24 @@ impl ElaborationRunner { } } +/// One session's permit. Raises the sweep's stop before the permit is released +/// on every exit but a clean return: a quarantine-class error, or a session that +/// never returned because it panicked. `drop` runs before the fields drop, so +/// the sweep waiting on this permit sees the stop when it wakes. +struct SessionPermit { + _permit: tokio::sync::OwnedSemaphorePermit, + stopped: Arc, + stop: bool, +} + +impl Drop for SessionPermit { + fn drop(&mut self) { + if self.stop { + self.stopped.store(true, Ordering::SeqCst); + } + } +} + fn snapshot_error(error: MailboxError) -> ControllerError { ControllerError::Snapshot(error) }