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>
This commit is contained in:
Meta Vens 2026-10-11 15:36:31 +02:00
commit 2dabb85009
2 changed files with 275 additions and 66 deletions

View file

@ -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),
};

View file

@ -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);
}