From 2dabb85009f8e1f9c5d571cd5e920d085e53d7b2 Mon Sep 17 00:00:00 2001
From: Meta Vens <rwvens@hotmail.com>
Date: Sun, 11 Oct 2026 15:36:31 +0200
Subject: [PATCH] Verse run: every member of a tick's frozen cover ends
 committed or refused, by subject

Refusal names its subject. absorb reads GroupedRun.refused (LeftBeforeSelection
-> OpportunityGone, ScopeChanged -> ScopeMoved, WorldRefused -> WorldRejected)
and records an interrupted singleton's member; refreshed returns the frozen
members it found no opportunity for; a cell-level refusal records every member
of the cell. TickReport carries the refused subjects' labels and Display prints
"refused: cell N <label> <reason>". Unbuilt before this verification; no commit
here is known not to build.

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
---
 crates/ghostlight/src/verse/run.rs       | 122 ++++++++-----
 crates/ghostlight/src/verse/run/tests.rs | 219 ++++++++++++++++++++---
 2 files changed, 275 insertions(+), 66 deletions(-)

diff --git a/crates/ghostlight/src/verse/run.rs b/crates/ghostlight/src/verse/run.rs
index 8fe9ba1..6bef91b 100644
--- a/crates/ghostlight/src/verse/run.rs
+++ b/crates/ghostlight/src/verse/run.rs
@@ -23,7 +23,7 @@
 
 use super::calls::{RecordingInferencePort, canonical};
 use crate::controllers::{
-    CellRun, ControllerError, ControllerPendingReason, ControllerRunner, InferenceFaultClass,
+    CellRun, ConstituentRefusalReason, ControllerError, ControllerPendingReason, ControllerRunner, InferenceFaultClass,
     NarrativeRun, OperationalRun, SubmissionDisposition,
 };
 use crate::cover::{
@@ -37,7 +37,7 @@ use crate::{
 use chrono::Utc;
 use cultcache_rs::{CacheBackingStore, CultCacheEnvelope, OwnedRedbMessagePackBackingStore};
 use serde::{Deserialize, Serialize};
-use std::{ops::ControlFlow, path::Path, sync::Mutex};
+use std::{collections::BTreeMap, ops::ControlFlow, path::Path, sync::Mutex};
 use thiserror::Error;
 
 /// The run record document's schema.
@@ -108,6 +108,9 @@ pub enum RefusalReason {
 pub struct Refusal {
     /// The cell's position in the tick's cover.
     pub cell: u32,
+    /// The frozen member that committed nothing. Every member of a closed
+    /// tick's cells is in `committed` or here, once.
+    pub subject: SubjectId,
     pub reason: RefusalReason,
 }
 
@@ -227,6 +230,9 @@ pub struct TickReport {
     pub checkpoint: TickCheckpoint,
     /// Every act the tick committed, from the operator log.
     pub acts: Vec<OperatorEvent>,
+    /// The labels of the subjects the checkpoint's refusals name, from the
+    /// snapshot the tick opened on.
+    pub labels: BTreeMap<SubjectId, String>,
     pub spent: u64,
     pub remaining: u64,
 }
@@ -283,7 +289,8 @@ impl std::fmt::Display for TickReport {
                 RefusalReason::OpportunityGone => "opportunity_gone",
                 RefusalReason::ScopeMoved => "scope_moved",
             };
-            writeln!(f, "  refused: cell {} {reason}", refusal.cell)?;
+            let label = self.labels.get(&refusal.subject).map_or("-", String::as_str);
+            writeln!(f, "  refused: cell {} {label} {reason}", refusal.cell)?;
         }
         writeln!(
             f,
@@ -634,10 +641,12 @@ fn command_of(submission: &SubmissionDisposition) -> CommandId {
 }
 
 /// Folds one finished cell into the tick. Returns the code of the pending state
-/// that ends the run, when the cell did not finish.
+/// that ends the run, when the cell did not finish. A pending cell's unreached
+/// members are not refusals: the tick stays open and they resume.
 fn absorb(
     run: CellRun,
-    cell: u32,
+    cell: &Cell,
+    position: u32,
     committed: &mut Vec<CommandId>,
     refusals: &mut Vec<Refusal>,
 ) -> Option<&'static str> {
@@ -648,10 +657,7 @@ fn absorb(
         }
         CellRun::Narrative(NarrativeRun::Pending(pending)) => Some(pending_code(pending.reason())),
         CellRun::Narrative(NarrativeRun::Interrupted(_)) => {
-            refusals.push(Refusal {
-                cell,
-                reason: RefusalReason::Interrupted,
-            });
+            refusals.extend(refuse_all(cell, position, RefusalReason::Interrupted));
             None
         }
         CellRun::Operational(OperationalRun::Completed(decision)) => {
@@ -663,32 +669,55 @@ fn absorb(
         }
         CellRun::Grouped(run) => {
             committed.extend(run.submissions.iter().map(|entry| command_of(&entry.submission)));
+            refusals.extend(run.refused.iter().map(|refused| Refusal {
+                cell: position,
+                subject: refused.subject,
+                reason: match refused.reason {
+                    ConstituentRefusalReason::LeftBeforeSelection => RefusalReason::OpportunityGone,
+                    ConstituentRefusalReason::ScopeChanged => RefusalReason::ScopeMoved,
+                    ConstituentRefusalReason::WorldRefused => RefusalReason::WorldRejected,
+                },
+            }));
             run.pending.map(pending_code)
         }
     }
 }
 
+/// One refusal per member of the cell.
+fn refuse_all(cell: &Cell, position: u32, reason: RefusalReason) -> impl Iterator<Item = Refusal> {
+    cell.members().iter().map(move |member| Refusal {
+        cell: position,
+        subject: member.subject,
+        reason,
+    })
+}
+
 /// The frozen cell with each member bound to the opportunity the world holds
-/// for that subject now. The partition is fixed when the tick opens; an
-/// opportunity is the world's current binding for a subject, and an earlier
-/// cell's act can move it (a word spoken in a shared room changes what its
-/// listeners know). A subject that no longer holds an opportunity leaves the
-/// cell, and an empty cell is gone.
-fn refreshed(frozen: &FrozenCell, tick: TickIndex, fresh: &[DecisionOpportunity]) -> Option<Cell> {
-    let members: Vec<Constituent> = frozen
-        .members
-        .iter()
-        .filter_map(|subject| {
-            fresh
-                .iter()
-                .find(|opportunity| opportunity.scope.subject_id == *subject)
-                .map(|opportunity| Constituent {
-                    subject: *subject,
-                    opportunity: opportunity.clone(),
-                })
-        })
-        .collect();
-    if frozen.group {
+/// for that subject now, and the frozen members that hold none. The partition
+/// is fixed when the tick opens; an opportunity is the world's current binding
+/// for a subject, and an earlier cell's act can move it (a word spoken in a
+/// shared room changes what its listeners know). A subject that no longer holds
+/// an opportunity leaves the cell, and an empty cell is gone.
+fn refreshed(
+    frozen: &FrozenCell,
+    tick: TickIndex,
+    fresh: &[DecisionOpportunity],
+) -> (Option<Cell>, Vec<SubjectId>) {
+    let mut gone = Vec::new();
+    let mut members: Vec<Constituent> = Vec::new();
+    for subject in &frozen.members {
+        match fresh
+            .iter()
+            .find(|opportunity| opportunity.scope.subject_id == *subject)
+        {
+            Some(opportunity) => members.push(Constituent {
+                subject: *subject,
+                opportunity: opportunity.clone(),
+            }),
+            None => gone.push(*subject),
+        }
+    }
+    let cell = if frozen.group {
         (!members.is_empty()).then(|| Cell::Group {
             id: frozen.id,
             tick,
@@ -700,7 +729,8 @@ fn refreshed(frozen: &FrozenCell, tick: TickIndex, fresh: &[DecisionOpportunity]
             tick,
             member,
         })
-    }
+    };
+    (cell, gone)
 }
 
 enum TickEnd {
@@ -880,16 +910,18 @@ async fn run_tick(
                 break;
             }
         };
-        let Some(cell) = refreshed(frozen, index, &fresh) else {
-            refusals.push(Refusal {
-                cell: position,
-                reason: RefusalReason::OpportunityGone,
-            });
+        let (cell, gone) = refreshed(frozen, index, &fresh);
+        refusals.extend(gone.into_iter().map(|subject| Refusal {
+            cell: position,
+            subject,
+            reason: RefusalReason::OpportunityGone,
+        }));
+        let Some(cell) = cell else {
             continue;
         };
         match runner.run_cell(&cell).await {
             Ok(run) => {
-                if let Some(code) = absorb(run, position, &mut committed, &mut refusals) {
+                if let Some(code) = absorb(run, &cell, position, &mut committed, &mut refusals) {
                     open = Some(failed(code));
                     break;
                 }
@@ -903,10 +935,9 @@ async fn run_tick(
                     open = Some(failed(code));
                     break;
                 }
-                CellEnd::Refused(reason) => refusals.push(Refusal {
-                    cell: position,
-                    reason,
-                }),
+                CellEnd::Refused(reason) => {
+                    refusals.extend(refuse_all(&cell, position, reason));
+                }
             },
         }
     }
@@ -938,6 +969,16 @@ async fn run_tick(
             Vec::new()
         }
     };
+    let labels = refusals
+        .iter()
+        .filter_map(|refusal| {
+            snapshot
+                .subjects
+                .iter()
+                .find(|subject| subject.id == refusal.subject)
+                .map(|subject| (refusal.subject, subject.label.clone()))
+        })
+        .collect();
     let report = TickReport {
         checkpoint: TickCheckpoint {
             tick: ordinal,
@@ -948,6 +989,7 @@ async fn run_tick(
             tokens: spent.saturating_sub(spent_before),
         },
         acts,
+        labels,
         spent,
         remaining: port.cap().saturating_sub(spent),
     };
diff --git a/crates/ghostlight/src/verse/run/tests.rs b/crates/ghostlight/src/verse/run/tests.rs
index e4bfe4e..b4a19ee 100644
--- a/crates/ghostlight/src/verse/run/tests.rs
+++ b/crates/ghostlight/src/verse/run/tests.rs
@@ -531,7 +531,7 @@ async fn resumed_after_both_cells_committed_elsewhere(reverse: bool) {
     };
     for cell in order {
         let fresh = verse.mailbox.opportunities().await.unwrap();
-        let bound = refreshed(&FrozenCell::of(cell), TickIndex(0), &fresh).unwrap();
+        let bound = refreshed(&FrozenCell::of(cell), TickIndex(0), &fresh).0.unwrap();
         runner.run_cell(&bound).await.unwrap();
     }
     assert_eq!(verse.acts().await, 2);
@@ -706,12 +706,19 @@ async fn a_cell_the_world_refuses_is_a_refusal_and_the_tick_closes() {
     assert_eq!(checkpoint.refusals.len(), 1);
     assert_eq!(checkpoint.refusals[0].reason, RefusalReason::WorldRejected);
     assert!(checkpoint.refusals[0].cell < 2);
+    assert_eq!(checkpoint.refusals[0].subject, verse.subject("Adrasteia").await);
     assert_eq!(verse.now().await, u64::from(SPAN));
     assert_eq!(verse.acts().await, 1);
 }
 
 #[test]
 fn the_operator_reads_refusals_and_how_a_run_ended() {
+    let [one, two, three, four] = [(); 4].map(|()| crate::SubjectId::issue());
+    let refusal = |cell, subject, reason| Refusal {
+        cell,
+        subject,
+        reason,
+    };
     let report = TickReport {
         checkpoint: TickCheckpoint {
             tick: 4,
@@ -719,26 +726,20 @@ fn the_operator_reads_refusals_and_how_a_run_ended() {
             cells: 3,
             committed: Vec::new(),
             refusals: vec![
-                Refusal {
-                    cell: 1,
-                    reason: RefusalReason::WorldRejected,
-                },
-                Refusal {
-                    cell: 2,
-                    reason: RefusalReason::OpportunityGone,
-                },
-                Refusal {
-                    cell: 0,
-                    reason: RefusalReason::Interrupted,
-                },
-                Refusal {
-                    cell: 5,
-                    reason: RefusalReason::ScopeMoved,
-                },
+                refusal(1, one, RefusalReason::WorldRejected),
+                refusal(2, two, RefusalReason::OpportunityGone),
+                refusal(0, three, RefusalReason::Interrupted),
+                refusal(5, four, RefusalReason::ScopeMoved),
             ],
             tokens: 7,
         },
         acts: Vec::new(),
+        labels: BTreeMap::from([
+            (one, "Ada".to_owned()),
+            (two, "Brann".to_owned()),
+            (three, "Cato".to_owned()),
+            (four, "Dura".to_owned()),
+        ]),
         spent: 40,
         remaining: 60,
     };
@@ -746,10 +747,10 @@ fn the_operator_reads_refusals_and_how_a_run_ended() {
         report.to_string(),
         concat!(
             "tick 4 open: 3 cells, 0 acts, 4 refused\n",
-            "  refused: cell 1 world_rejected\n",
-            "  refused: cell 2 opportunity_gone\n",
-            "  refused: cell 0 interrupted\n",
-            "  refused: cell 5 scope_moved\n",
+            "  refused: cell 1 Ada world_rejected\n",
+            "  refused: cell 2 Brann opportunity_gone\n",
+            "  refused: cell 0 Cato interrupted\n",
+            "  refused: cell 5 Dura scope_moved\n",
             "  tokens: 7 this tick, 40 spent, 60 remaining\n",
         )
     );
@@ -895,7 +896,7 @@ async fn a_cell_is_run_on_what_the_world_holds_for_its_members_now() {
             panic!("two subjects under two cells are singletons")
         };
         let frozen = FrozenCell::of(cell);
-        let Some(Cell::Singleton { member, .. }) = refreshed(&frozen, TickIndex(0), &moved) else {
+        let (Some(Cell::Singleton { member, .. }), gone) = refreshed(&frozen, TickIndex(0), &moved) else {
             panic!("a subject that still has an opportunity keeps its cell")
         };
         assert_ne!(member.opportunity, held.opportunity);
@@ -903,12 +904,15 @@ async fn a_cell_is_run_on_what_the_world_holds_for_its_members_now() {
         // The member is bound to its own subject's opportunity, not a neighbour's.
         assert_eq!(member.subject, held.subject);
         assert_eq!(member.opportunity.scope.subject_id, member.subject);
+        assert!(gone.is_empty());
     }
     let frozen_group = FrozenCell::of(grouped);
-    let Some(Cell::Group { members: now, .. }) = refreshed(&frozen_group, TickIndex(0), &moved)
+    let (Some(Cell::Group { members: now, .. }), gone) =
+        refreshed(&frozen_group, TickIndex(0), &moved)
     else {
         panic!("a group keeps its members")
     };
+    assert!(gone.is_empty());
     assert_eq!(now.len(), 2);
     assert!(now.iter().all(|member| moved.contains(&member.opportunity)));
     assert!(
@@ -918,7 +922,8 @@ async fn a_cell_is_run_on_what_the_world_holds_for_its_members_now() {
 
     // A subject with no opportunity leaves its cell, and an empty cell is gone.
     let one = &opportunities[..1];
-    let Some(Cell::Group { members: left, .. }) = refreshed(&frozen_group, TickIndex(0), one)
+    let (Some(Cell::Group { members: left, .. }), left_out) =
+        refreshed(&frozen_group, TickIndex(0), one)
     else {
         panic!("one member remains")
     };
@@ -927,10 +932,13 @@ async fn a_cell_is_run_on_what_the_world_holds_for_its_members_now() {
     let gone = singles
         .cells
         .iter()
-        .filter(|cell| refreshed(&FrozenCell::of(cell), TickIndex(0), one).is_none())
+        .filter(|cell| refreshed(&FrozenCell::of(cell), TickIndex(0), one).0.is_none())
         .count();
     assert_eq!(gone, 1);
-    assert!(refreshed(&frozen_group, TickIndex(0), &[]).is_none());
+    let (nothing, all_gone) = refreshed(&frozen_group, TickIndex(0), &[]);
+    assert!(nothing.is_none());
+    assert_eq!(all_gone, frozen_group.members);
+    assert_eq!(left_out, vec![opportunities[1].scope.subject_id]);
 }
 
 #[test]
@@ -1792,3 +1800,162 @@ fn two_racing_opens_of_one_tick_cannot_both_succeed() {
         assert_eq!(opened, 1, "ordinal {ordinal} was opened {opened} times");
     }
 }
+
+
+/// A reply for the one-cell cover: every constituent the request offers
+/// speaks, so the grouped cell gets its one inference answered for both.
+fn grouped_reply(request: &str) -> Vec<InferenceEvent> {
+    let handles: Vec<usize> = (0..8)
+        .filter(|i| request.contains(&format!("c{i}__speak")))
+        .collect();
+    assert!(!handles.is_empty(), "a one-cell cover offers grouped handles");
+    handles
+        .into_iter()
+        .map(|i| InferenceEvent::ToolCall {
+            call_id: format!("act{i}"),
+            name: format!("c{i}__speak"),
+            arguments: json!({ "text": format!("Member {i} speaks.") }).to_string(),
+        })
+        .collect()
+}
+
+fn one_cell(ticks: u32) -> VerseRunConfig {
+    VerseRunConfig {
+        ticks,
+        tick_span: TickMinutes::new(SPAN).unwrap(),
+        cover: CoverBudget {
+            cells: 1,
+            constituent_cap: 8,
+            urgency_slots: 0,
+        },
+    }
+}
+
+/// Runs the grouped cover for `ticks`, returning the verse, the outcome and
+/// every report the operator saw.
+async fn grouped_run(ticks: u32) -> (tempfile::TempDir, Harness, RunOutcome, Vec<TickReport>) {
+    let (dir, verse) = Harness::fresh().await;
+    let mut reports = Vec::new();
+    let outcome = verse
+        .run_under(
+            ScriptedPort::new(grouped_reply),
+            u64::MAX,
+            one_cell(ticks),
+            |report| {
+                reports.push(report.clone());
+                ControlFlow::Continue(())
+            },
+        )
+        .await;
+    (dir, verse, outcome, reports)
+}
+
+/// Every member of every frozen cell of a closed tick is, once, either a
+/// subject whose command id the checkpoint holds or a subject it refused.
+async fn assert_every_member_accounted_for(verse: &Harness, outcome: &RunOutcome) {
+    let world = verse.mailbox.snapshot().await.unwrap().world_id;
+    let closed: Vec<_> = outcome.record.checkpoints.iter().filter(|c| c.closed).collect();
+    assert!(!closed.is_empty(), "a run with no closed tick proves nothing");
+    for checkpoint in closed {
+        let tick = verse.runs.tick(world, checkpoint.tick).unwrap().unwrap();
+        let mut members = 0;
+        for cell in &tick.cells {
+            for subject in &cell.members {
+                members += 1;
+                let command = CommandId::for_cell_constituent(
+                    world,
+                    cell.id,
+                    *subject,
+                    TickIndex(checkpoint.tick),
+                );
+                let committed = checkpoint.committed.iter().filter(|c| **c == command).count();
+                let refused = checkpoint
+                    .refusals
+                    .iter()
+                    .filter(|r| r.subject == *subject)
+                    .count();
+                assert_eq!(committed + refused, 1, "tick {} member", checkpoint.tick);
+            }
+        }
+        assert_eq!(checkpoint.committed.len() + checkpoint.refusals.len(), members);
+    }
+}
+
+#[tokio::test]
+async fn a_grouped_cell_records_every_member_it_could_not_commit() {
+    let (_dir, verse, outcome, reports) = grouped_run(2).await;
+    assert_eq!(outcome.record.status, RunStatus::Completed);
+    assert_eq!(reports.len(), 2);
+    let world = verse.mailbox.snapshot().await.unwrap();
+    for (checkpoint, report) in outcome.record.checkpoints.iter().zip(&reports) {
+        assert!(checkpoint.closed);
+        assert_eq!(checkpoint.cells, 1);
+        assert_eq!(checkpoint.committed.len(), 1);
+        let [refusal] = checkpoint.refusals.as_slice() else {
+            panic!("one member of two is refused")
+        };
+        assert_eq!(refusal.reason, RefusalReason::ScopeMoved);
+        let label = &world
+            .subjects
+            .iter()
+            .find(|subject| subject.id == refusal.subject)
+            .unwrap()
+            .label;
+        assert!(
+            report
+                .to_string()
+                .contains(&format!("refused: cell 0 {label} scope_moved")),
+            "the operator reads who was refused"
+        );
+    }
+    assert_every_member_accounted_for(&verse, &outcome).await;
+}
+
+#[tokio::test]
+async fn every_member_of_a_closed_tick_is_committed_or_refused_once() {
+    // Singletons, all committing.
+    let (_dir, verse) = Harness::fresh().await;
+    let outcome = verse.run_quietly(ScriptedPort::speaking(), u64::MAX, 2).await;
+    assert_every_member_accounted_for(&verse, &outcome).await;
+
+    // Singletons, one refused by the world.
+    let (_dir, verse) = Harness::fresh().await;
+    let nothing = id_text(CommandId::new());
+    let port = ScriptedPort::new(move |request| {
+        if who(request) == "Adrasteia" {
+            vec![InferenceEvent::ToolCall {
+                call_id: "act".into(),
+                name: "attack".into(),
+                arguments: json!({ "target": nothing, "text": "Adrasteia strikes." }).to_string(),
+            }]
+        } else {
+            vec![speak_call("Brannoch speaks.")]
+        }
+    });
+    let outcome = verse.run_quietly(port, u64::MAX, 2).await;
+    assert!(outcome.record.checkpoints.iter().all(|c| c.refusals.len() == 1));
+    assert_every_member_accounted_for(&verse, &outcome).await;
+
+    // One grouped cell.
+    let (_dir, verse, outcome, _) = grouped_run(2).await;
+    assert_every_member_accounted_for(&verse, &outcome).await;
+}
+
+#[tokio::test]
+async fn a_refused_grouped_member_acts_next_tick() {
+    let (_dir, verse, outcome, _) = grouped_run(2).await;
+    let world = verse.mailbox.snapshot().await.unwrap().world_id;
+    let [first, second] = outcome.record.checkpoints.as_slice() else {
+        panic!("two ticks, two checkpoints")
+    };
+    let refused_first = first.refusals[0].subject;
+    let tick = verse.runs.tick(world, second.tick).unwrap().unwrap();
+    let again = CommandId::for_cell_constituent(
+        world,
+        tick.cells[0].id,
+        refused_first,
+        TickIndex(second.tick),
+    );
+    assert!(second.committed.contains(&again));
+    assert_ne!(second.refusals[0].subject, refused_first);
+}