diff --git a/crates/mind-body/src/heartbeat.rs b/crates/mind-body/src/heartbeat.rs index 3e34836..013d9b2 100644 --- a/crates/mind-body/src/heartbeat.rs +++ b/crates/mind-body/src/heartbeat.rs @@ -13,8 +13,10 @@ //! Self holds briefs (`inbox::derive`) is a Self run claiming them, in place of //! her Persona turn: a session that ended with a brief open, or a brief admitted //! after its flush, is never an orphan, and a Self start takes the beat like any -//! launch. A brief with an open question raised in it (the repetition breaker's -//! trip) is not claimed until that question is answered. Nothing is metered here: +//! launch. A brief an open question holds (raised in it, or in a run that claims it: +//! the repetition breaker's trip holds the claims of the run it belongs to) is +//! neither claimed nor told until that question is answered or withdrawn; other +//! briefs are. Nothing is metered here: //! no meter and no model call. //! //! The heartbeat owns `heartbeat.cc` and is its only writer; Persona units read @@ -40,7 +42,7 @@ use huginn_mind::{PipelineDocumentSummary, PipelineFacts, PipelinePageItems}; use crate::chat_client::{Chat, ChatDelivery}; use crate::control::{ControlSource, Effective, FileSource, HeldReason, SwarmEffective, read_effective, read_envelope, set_mode_world_readable}; use crate::launch::{Clock, LaunchOutcome, LaunchRequest, Launcher, Ports, holders, open_and_launch}; -use crate::queue::{MindPort, any_of, citing, headers, in_force, of_kind}; +use crate::queue::{MindPort, any_of, citing, claims_of, headers, in_force, of_kind}; use crate::self_run::inbox; use crate::tempo::{self, Standing}; @@ -565,6 +567,10 @@ fn launch(organs: &Organs, row: &Row, turn: RunTurn, claims: Vec) - Ok(match open_and_launch(&ports, request) { Ok(LaunchOutcome::Launched { run }) => Launched::Run(label_of(&run)), Ok(LaunchOutcome::Busy { .. }) => Launched::Busy, + Ok(LaunchOutcome::Tripped { run, question }) => { + let held = claims_of(row.seat.mind, &run).map_or(0, |claims| claims.len()); + Launched::Refused(format!("tripped: {} holds the {held} claims of {}", question.id.0, run.id.0)) + } Ok(_) => Launched::Refused("launch refused".into()), Err(error) => Launched::Refused(format!("launch failed: {error}")), }) @@ -574,27 +580,24 @@ fn launch(organs: &Organs, row: &Row, turn: RunTurn, claims: Vec) - const BRIEFS_PER_SELF_RUN: usize = 64; /// The briefs a Self run would claim for this Mind: those in force that no live -/// Self run holds. None while a question is open in a brief of hers. A Self run -/// claims up to `BRIEFS_PER_SELF_RUN` briefs and every run it ends empty counts -/// against all of them, so one cause trips them together; a trip is the Mind's -/// stop, one question for the cause, not one per brief. It lasts until the -/// operator answers or withdraws that question, and the count starts afresh. -fn claimable(row: &Row) -> Result> { +/// Self run holds and no open question holds, and how many an open question +/// holds. A Self run claims up to `BRIEFS_PER_SELF_RUN` briefs and every run it +/// ends empty counts against all of them, so one cause trips them together; a +/// trip holds the claims of its run, not the Mind, until the operator answers +/// or withdraws its question, and the count starts afresh. `inbox::derive` +/// reads `queue::on_hold`; nothing here decides what is held. +fn claimable(row: &Row) -> Result<(Vec, usize)> { let mind = row.seat.mind; if !holders(mind, row.slug(), RunTurn::SelfRun)?.is_empty() { - return Ok(Vec::new()); + return Ok((Vec::new(), 0)); } - let tripped = headers(mind, &in_force(of_kind(PipelineKind::Question)))? - .into_iter() - .any(|header| matches!(header.facts, PipelineFacts::Question { raised_in: Some(item), .. } if item.kind == PipelineKind::FollowUp)); - if tripped { - return Ok(Vec::new()); - } - Ok(inbox::derive(mind, &HashSet::new())? + let (free, held) = inbox::derive_counting_held(mind, &HashSet::new())?; + let claims = free .into_iter() .map(|brief| PipelineRef { kind: PipelineKind::FollowUp, id: brief.id.as_str().into() }) .take(BRIEFS_PER_SELF_RUN) - .collect()) + .collect(); + Ok((claims, held)) } /// One poll. @@ -700,10 +703,13 @@ pub fn poll(organs: &Organs, config: &Config) -> Result { let mut passed: Vec = standings.clone(); while due && let Some(winner) = tempo::pick(&passed, beat_s) { // Her Self holds briefs: the beat is a Self run for them; otherwise a Persona turn. - let claims = claimable(&rows[winner]).unwrap_or_else(|error| { + let (claims, held) = claimable(&rows[winner]).unwrap_or_else(|error| { rows[winner].note = format!("briefs: {error}"); - Vec::new() + (Vec::new(), 0) }); + if held > 0 { + rows[winner].note = format!("{held} briefs held by open questions"); + } let turn = if claims.is_empty() { RunTurn::PersonaTurn } else { RunTurn::SelfRun }; match launch(organs, &rows[winner], turn, claims)? { Launched::Run(label) => { diff --git a/crates/mind-body/src/launch.rs b/crates/mind-body/src/launch.rs index c4828f6..cd99fc2 100644 --- a/crates/mind-body/src/launch.rs +++ b/crates/mind-body/src/launch.rs @@ -3,8 +3,9 @@ //! `open_and_launch` is the whole path, in this order: (0) the in-force runs of //! hers of the request's turn are read, a live one (its unit active, or //! admitted within `START_GRACE_S`) answers Busy and a dead one is to be closed -//! Recorded; (1) every cut-spec claim must be in the queue; (2) no claim may -//! trip the repetition breaker; (3) the run is admitted, in the one batch that +//! Recorded; (1) every cut-spec claim must be in the queue; (2) no claim may be held by an open question, and none may +//! trip the repetition breaker (a trip raises one question in the run whose end +//! tripped it, which holds that run's claims); (3) the run is admitted, in the one batch that //! closes the dead holders, so liveness is decided by the in-force record alone //! and the unit table is only the evidence for a close; (4) the unit is started, //! only after that batch commits. The run is the grant (target invariant @@ -28,7 +29,7 @@ use eureka_pipeline::{ use huginn_mind::{MindRefusal, PipelineAdmissionOutcome, PipelineDocumentSummary, PipelineFacts}; use crate::control::Grant; -use crate::queue::{Breaker, MindPort, any_of, breaker, citing, headers, in_force, of_kind, queue}; +use crate::queue::{Breaker, MindPort, any_of, breaker, citing, claims_of, headers, in_force, of_kind, on_hold, queue}; /// How long a run just admitted may have no active unit before its slot is /// treated as dead: `systemctl start --no-block` returns before the unit runs. @@ -245,8 +246,9 @@ pub enum LaunchOutcome { NotQueued { item: PipelineRef, }, + /// The trip belongs to `run`, the run whose end tripped it; `question` is raised in it. Tripped { - item: PipelineRef, + run: PipelineRef, question: PipelineRef, }, /// The live holder of the request's turn, from step 0 or from admission. @@ -344,16 +346,18 @@ pub fn open_and_launch(ports: &Ports, request: LaunchRequest) -> Result LaunchOutcome::Tripped { item: item.clone(), question: id }, + Ok(()) => LaunchOutcome::Tripped { run, question: id }, Err(declined) => LaunchOutcome::Refused(declined), }); } @@ -429,22 +433,24 @@ fn breaker_label(label: &Label) -> Label { Label(format!("breaker-{}", label.0)) } -/// The question a tripped item raises, in the item's own campaign. -fn breaker_question(item: &PipelineRef, label: &Label, now: DateTime) -> PipelineQuestion { +/// The question a trip raises, in the campaign of the claim that tripped: one +/// per trip, raised in `run` and holding the `held` claims it claimed. +fn breaker_question(claim: &PipelineRef, run: &PipelineRef, held: usize, label: &Label, now: DateTime) -> PipelineQuestion { PipelineQuestion { - campaign: Slug(root_of(item).to_string()), + campaign: Slug(root_of(claim).to_string()), label: breaker_label(label), - title: "Repeated runs on one item admitted nothing".into(), - question: - "The last runs on this item left nothing that cites it, so it launches no more. Answering this reopens the item and starts a fresh count." - .into(), + title: "Repeated runs on the same work admitted nothing".into(), + question: format!( + "Run {} and the runs before it left nothing that cites the work it held (claims held: {held}), so none of it launches again. Answering or withdrawing this reopens every claim of that run with a fresh count.", + run.id.0 + ), options: vec![ - QuestionOption { label: "reopen".into(), text: "Rule on the item, which reopens it.".into() }, - QuestionOption { label: "retire".into(), text: "Resolve the item so it is no longer worked.".into() }, + QuestionOption { label: "reopen".into(), text: "Answer to reopen every claim of the run with a fresh count.".into() }, + QuestionOption { label: "retire".into(), text: "Resolve the claims so they are no longer worked, and answer with this option.".into() }, ], recommended: "reopen".into(), depends: vec![], - raised_in: Some(item.clone()), + raised_in: Some(run.clone()), asked_on: date(now), } } @@ -853,12 +859,13 @@ mod tests { let before = rig.runs(); let outcome = rig.launch(RunTurn::SelfRun, &[spec_ref("x")]); let question = id("question", "breaker-mind-20261009T131500000Z"); - assert_eq!(outcome, LaunchOutcome::Tripped { item: spec_ref("x"), question: reference(K::Question, &question) }); + assert_eq!(outcome, LaunchOutcome::Tripped { run: run_ref("h2"), question: reference(K::Question, &question) }); assert_eq!((rig.runs(), rig.started()), (before, 0)); assert_eq!(rig.mind.ids(K::Question), vec![question.clone()]); let view = rig.mind.view(K::Question, &question).unwrap(); let PipelineDocument::Question(asked) = &view.document else { panic!("not a question") }; - assert_eq!((asked.raised_in.clone(), asked.campaign.0.as_str()), (Some(spec_ref("x")), CAMPAIGN)); + assert_eq!((asked.raised_in.clone(), asked.campaign.0.as_str()), (Some(run_ref("h2")), CAMPAIGN), "raised in the run whose end tripped it"); + assert!(asked.question.contains(&run_ref("h2").id.0) && asked.question.contains("claims held: 1"), "{}", asked.question); assert_eq!(view.admission.provenance.session.0, "mind-20261009T131500000Z"); // Open, the question blocks the item. assert_eq!(rig.launch(RunTurn::SelfRun, &[spec_ref("x")]), LaunchOutcome::NotQueued { item: spec_ref("x") }); @@ -918,9 +925,29 @@ mod tests { let outcome = rig.launch(RunTurn::SelfRun, &[spec_ref("x")]); assert!(matches!(outcome, LaunchOutcome::Tripped { .. }), "{outcome:?}"); assert!(!rig.in_force("dead"), "the trip's batch closes the dead run too"); + let question = rig.mind.ids(K::Question).pop().expect("the trip asked"); + let PipelineDocument::Question(asked) = rig.mind.view(K::Question, &question).unwrap().document else { panic!("not a question") }; + assert_eq!(asked.raised_in, Some(run_ref("dead")), "the dead holder, closed in the same batch, is the run the question is raised in"); assert_eq!((rig.runs(), rig.started()), (3, 0)); } + #[test] + fn the_gate_refuses_a_claim_an_open_question_holds_whatever_its_kind() { + let rig = Rig::new(); + rig.mind.committed(vec![spec("x"), follow_up("brief", &spec_ref("x"))]); + let brief = reference(K::FollowUp, &id("follow_up", "brief")); + // A question raised in a run holds the follow-up that run claimed, though the follow-up is no cut spec. + rig.mind.committed(vec![run("failed", RunTurn::SelfRun, RunOperator::Mind, &[brief.clone(), spec_ref("x")])]); + rig.mind.committed(vec![close(run_ref("failed"), recorded())]); + rig.mind.committed(vec![question_in("fork", &run_ref("failed"))]); + let runs = rig.runs(); + assert_eq!(rig.launch(RunTurn::SelfRun, std::slice::from_ref(&brief)), LaunchOutcome::NotQueued { item: brief.clone() }); + assert_eq!(rig.launch(RunTurn::SelfRun, &[spec_ref("x")]), LaunchOutcome::NotQueued { item: spec_ref("x") }, "a cut spec held by a run-raised question"); + assert_eq!((rig.runs(), rig.started()), (runs, 0), "nothing admitted, nothing started"); + rig.mind.committed(vec![close(reference(K::Question, &id("question", "fork")), withdrawn())]); + assert!(matches!(rig.launch(RunTurn::SelfRun, &[brief]), LaunchOutcome::Launched { .. }), "withdrawn, the hold lifts"); + } + /// The one source of every drawn case: splitmix64 from a fixed seed, so a /// failure names a case that runs again. struct Draw(u64); diff --git a/crates/mind-body/src/queue.rs b/crates/mind-body/src/queue.rs index 526c6b4..bf6e69f 100644 --- a/crates/mind-body/src/queue.rs +++ b/crates/mind-body/src/queue.rs @@ -162,27 +162,55 @@ pub(crate) fn live_claims(mind: &dyn MindPort, instance: &Slug, ending: &[Pipeli Ok(claims) } +/// The claims of the run `run`, resolved or not (a citation outlives its +/// resolution). Empty when the mind has no such run. +pub(crate) fn claims_of(mind: &dyn MindPort, run: &PipelineRef) -> Result> { + let one = Selection { keys: Some(vec![run.id.0.clone()]), ..of_kind(PipelineKind::Run) }; + Ok(headers(mind, &one)? + .into_iter() + .flat_map(|header| match header.facts { + PipelineFacts::Run { claims, .. } => claims, + _ => Vec::new(), + }) + .collect()) +} + +/// The items an in-force question holds: the one owner of the stop. A question +/// raised in a run holds every claim of that run, the set that failed together; +/// a question raised in anything else holds that item. The question's text is +/// display only, and a held item is released by answering or withdrawing it. +pub fn on_hold(mind: &dyn MindPort) -> Result> { + let mut held: Vec = Vec::new(); + for header in headers(mind, &in_force(of_kind(PipelineKind::Question)))? { + if let PipelineFacts::Question { raised_in: Some(item), .. } = header.facts { + if item.kind == PipelineKind::Run { + held.extend(claims_of(mind, &item)?); + } else { + held.push(item); + } + } + } + Ok(held) +} + /// What may be worked, oldest first: the cut specs in force that no report -/// answers, that no open question blocks, and that no live run claims. A run in +/// answers, that no open question holds, and that no live run claims. A run in /// `ending` is one the caller closes in the same admission that opens the next, /// so it claims nothing here. pub fn queue(mind: &dyn MindPort, instance: &Slug, ending: &[PipelineRef]) -> Result> { let unreported = Selection { cited: Some(Incoming { role: "cut_spec".into(), exists: false }), ..in_force(of_kind(PipelineKind::CutSpec)) }; - let mut blocked: Vec = Vec::new(); - for header in headers(mind, &in_force(of_kind(PipelineKind::Question)))? { - if let PipelineFacts::Question { raised_in: Some(item), .. } = header.facts { - blocked.push(item); - } - } + let blocked = on_hold(mind)?; let claimed = live_claims(mind, instance, ending)?; Ok(headers(mind, &unreported)?.into_iter().map(|header| header.id).filter(|spec| !blocked.contains(spec) && !claimed.contains(spec)).collect()) } /// Whether an item may still be launched on. -#[derive(Clone, Copy, Debug, PartialEq, Eq)] +#[derive(Clone, Debug, PartialEq, Eq)] pub enum Breaker { Clear, - Trip, + /// Tripped, and the trip belongs to `run`: the newest ended counted run on + /// the item, the run whose end completed the streak or the share. + Trip { run: PipelineRef }, } /// The repetition breaker, derived from the documents that cite the item. @@ -196,29 +224,34 @@ pub enum Breaker { /// withdrawn run never ran and is no run here. A counted run is ended when /// resolved or listed in `ending`, and is empty when no document of the set /// other than a run was admitted from its own admission up to the next counted -/// run's (or to now). Trip when the last `CONSECUTIVE` ended runs are empty, or +/// run's (or to now). Trip, naming the newest ended counted run, when the last `CONSECUTIVE` ended runs are empty, or /// `EMPTY_SHARE` of the last `WINDOW`. What else the mind admitted, such as a /// Persona turn's run or an unrelated ruling, cites nothing here and counts for /// nothing. pub fn breaker(mind: &dyn MindPort, item: &PipelineRef, ending: &[PipelineRef]) -> Result { let mut set = headers(mind, &Selection { cites: Some(citing(item, None)), ..Selection::default() })?; set.sort_by_key(|header| header.admission.ordinal); - let boundary = set + let asked_in_item = set .iter() .filter(|header| matches!(&header.facts, PipelineFacts::Question { raised_in: Some(raised), .. } if raised == item)) - .map(|header| header.admission.ordinal) - .max(); + .map(|header| header.admission.ordinal); + let asked_in_run = Selection { + cites: Some(citing(item, Some("claims"))), + cited: Some(Incoming { role: "raised_in".into(), exists: true }), + ..of_kind(PipelineKind::Run) + }; + let boundary = asked_in_item.chain(headers(mind, &asked_in_run)?.into_iter().map(|run| run.admission.ordinal)).max(); let after = |header: &&PipelineDocumentSummary| boundary.is_none_or(|boundary| header.admission.ordinal > boundary); let withdrawn = |run: &PipelineDocumentSummary| matches!(run.status, PipelineStatusSummary::Resolved { outcome: ResolutionOutcome::Withdrawn { .. }, .. }); - // (admission ordinal, ended) of each counted run, oldest first. - let ran: Vec<(u64, bool)> = set + // (admission ordinal, ended, ref) of each counted run, oldest first. + let ran: Vec<(u64, bool, PipelineRef)> = set .iter() .filter(after) .filter(|header| { matches!(header.facts, PipelineFacts::Run { turn: RunTurn::SelfRun, operated_by: RunOperator::Mind, .. }) && !withdrawn(header) }) - .map(|run| (run.admission.ordinal, matches!(run.status, PipelineStatusSummary::Resolved { .. }) || ending.contains(&run.id))) + .map(|run| (run.admission.ordinal, matches!(run.status, PipelineStatusSummary::Resolved { .. }) || ending.contains(&run.id), run.id.clone())) .collect(); let ended: Vec = (0..ran.len()).filter(|index| ran[*index].1).collect(); let considered: Vec = ended.iter().rev().take(WINDOW).copied().collect(); @@ -235,7 +268,10 @@ pub fn breaker(mind: &dyn MindPort, item: &PipelineRef, ending: &[PipelineRef]) .collect(); let consecutive = empty.len() >= CONSECUTIVE && empty[..CONSECUTIVE].iter().all(|empty| *empty); let share = empty.iter().filter(|empty| **empty).count() >= EMPTY_SHARE; - Ok(if consecutive || share { Breaker::Trip } else { Breaker::Clear }) + Ok(match ended.last() { + Some(newest) if consecutive || share => Breaker::Trip { run: ran[*newest].2.clone() }, + _ => Breaker::Clear, + }) } #[cfg(test)] @@ -262,7 +298,7 @@ mod tests { fn tripped(runs: &[bool]) -> bool { let (mind, item) = worked(runs); - breaker(&mind, &item, &[]).unwrap() == Breaker::Trip + matches!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip { .. }) } #[test] @@ -307,7 +343,7 @@ mod tests { mind.committed(vec![close(run_ref("third"), recorded())]); breaker(&mind, &item, &[]).unwrap() }; - assert_eq!(after_a_run_by(RunOperator::Mind), Breaker::Trip, "hers, the third empty one"); + assert!(matches!(after_a_run_by(RunOperator::Mind), Breaker::Trip { .. }), "hers, the third empty one"); assert_eq!(after_a_run_by(RunOperator::Operator), Breaker::Clear, "his, counted for nothing"); } @@ -362,7 +398,7 @@ mod tests { ended_run(&mind, &item, "h", n, false); mind.committed(vec![close(run_ref(&format!("p{n}")), recorded())]); } - assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip); + assert!(matches!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip { .. })); } #[test] @@ -370,7 +406,7 @@ mod tests { let (mind, item) = worked(&[E, E]); mind.committed(vec![run("dying", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item))]); assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Clear, "open, it is not yet ended"); - assert_eq!(breaker(&mind, &item, &[run_ref("dying")]).unwrap(), Breaker::Trip); + assert!(matches!(breaker(&mind, &item, &[run_ref("dying")]).unwrap(), Breaker::Trip { .. })); let instance = Slug(INSTANCE.into()); assert!(queue(&mind, &instance, &[]).unwrap().iter().all(|spec| *spec != item), "it claims the item"); assert!(queue(&mind, &instance, &[run_ref("dying")]).unwrap().contains(&item), "an ending run claims nothing"); @@ -387,7 +423,7 @@ mod tests { assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Clear, "the tenth empty run, but nine of them are before the question"); ended_run(&mind, &item, "g", 1, false); ended_run(&mind, &item, "g", 2, false); - assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip, "three empty runs after it"); + assert!(matches!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip { .. }), "three empty runs after it"); } #[test] @@ -403,7 +439,7 @@ mod tests { ended_run(&mind, &item, "h", 2, false); breaker(&mind, &item, &[]).unwrap() }; - assert_eq!(withdrawn_between(false), Breaker::Trip, "the withdrawn run is neither counted nor does it break the streak"); + assert!(matches!(withdrawn_between(false), Breaker::Trip { .. }), "the withdrawn run is neither counted nor does it break the streak"); assert_eq!(withdrawn_between(true), Breaker::Clear, "a document after the withdrawn run still lands in the first run's window"); } @@ -417,7 +453,7 @@ mod tests { let (mind, item) = worked(&[E, E, E]); mind.committed(vec![run("live", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item)), follow_up("with-next", &item)]); - assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip, "the next run's batch is not the previous run's window"); + assert!(matches!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip { .. }), "the next run's batch is not the previous run's window"); } /// The adapter over the wire vocabulary, answered by a real mind. @@ -469,7 +505,7 @@ mod tests { history_as(&mind, &spec_ref("x"), "x", &[E, E]); history_as(&mind, &spec_ref("y"), "y", &[E, E, E]); assert_eq!(breaker(&mind, &spec_ref("x"), &[]).unwrap(), Breaker::Clear); - assert_eq!(breaker(&mind, &spec_ref("y"), &[]).unwrap(), Breaker::Trip); + assert!(matches!(breaker(&mind, &spec_ref("y"), &[]).unwrap(), Breaker::Trip { .. })); } #[test] @@ -502,6 +538,72 @@ mod tests { assert_eq!(queue(&mind, &other, &[]).unwrap().len(), 3, "claims are read for the instance asked"); } + #[test] + fn a_trip_names_the_newest_ended_counted_run() { + let (mind, item) = worked(&[E, E, E, E]); + assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip { run: run_ref("h3") }); + mind.committed(vec![run("live", RunTurn::SelfRun, RunOperator::Mind, std::slice::from_ref(&item))]); + assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Trip { run: run_ref("h3") }, "a run still open is not ended"); + assert_eq!(breaker(&mind, &item, &[run_ref("live")]).unwrap(), Breaker::Trip { run: run_ref("live") }, "a run in ending is"); + } + + /// Two items claimed together by every run of a history; the last one ended empty. + fn bundle() -> (TestMind, PipelineRef, PipelineRef) { + let mind = TestMind::seeded(); + mind.committed(vec![spec("a"), spec("b")]); + let (a, b) = (spec_ref("a"), spec_ref("b")); + for n in 0..3 { + mind.committed(vec![run(&format!("f{n}"), RunTurn::SelfRun, RunOperator::Mind, &[a.clone(), b.clone()])]); + mind.committed(vec![close(run_ref(&format!("f{n}")), recorded())]); + } + (mind, a, b) + } + + #[test] + fn a_question_raised_in_a_run_holds_every_claim_of_it_until_answered_or_withdrawn() { + for withdraw in [false, true] { + let (mind, a, b) = bundle(); + assert!(on_hold(&mind).unwrap().is_empty()); + mind.committed(vec![question_in("fork", &run_ref("f2"))]); + let held = on_hold(&mind).unwrap(); + assert!(held.contains(&a) && held.contains(&b) && held.len() == 2, "{held:?}: the claims of the resolved run"); + assert_eq!(queue(&mind, &Slug(INSTANCE.into()), &[]).unwrap(), Vec::::new(), "queue reads the same hold"); + let question = id("question", "fork"); + if withdraw { + mind.committed(vec![close(reference(PipelineKind::Question, &question), withdrawn())]); + } else { + mind.committed(vec![ruling("answer", Some((&question, "A")))]); + } + assert!(on_hold(&mind).unwrap().is_empty(), "answered or withdrawn, nothing is held (withdrawn: {withdraw})"); + } + } + + #[test] + fn a_question_raised_in_anything_else_holds_that_item_alone() { + let (mind, a, b) = bundle(); + mind.committed(vec![question_in("fork", &a)]); + assert_eq!(on_hold(&mind).unwrap(), vec![a], "the legacy shape: raised in the item, it holds the item and no claimant"); + assert!(!on_hold(&mind).unwrap().contains(&b)); + } + + #[test] + fn answering_a_question_in_the_run_gives_every_claim_of_it_a_fresh_count() { + let (mind, a, b) = bundle(); + assert_eq!(breaker(&mind, &a, &[]).unwrap(), Breaker::Trip { run: run_ref("f2") }); + assert_eq!(breaker(&mind, &b, &[]).unwrap(), Breaker::Trip { run: run_ref("f2") }); + mind.committed(vec![question_in("fork", &run_ref("f2"))]); + mind.committed(vec![ruling("answer", Some((&id("question", "fork"), "A")))]); + for item in [&a, &b] { + assert_eq!(breaker(&mind, item, &[]).unwrap(), Breaker::Clear, "counts only runs after f2"); + } + for n in 0..3 { + mind.committed(vec![run(&format!("g{n}"), RunTurn::SelfRun, RunOperator::Mind, &[a.clone(), b.clone()])]); + mind.committed(vec![close(run_ref(&format!("g{n}")), recorded())]); + let expected = if n == 2 { Breaker::Trip { run: run_ref("g2") } } else { Breaker::Clear }; + assert_eq!(breaker(&mind, &b, &[]).unwrap(), expected, "b has no question of its own and trips after three fresh empty runs, never before"); + } + } + /// A daemon that is not there: the failure names no input, endpoint or /// instance. #[test] diff --git a/crates/mind-body/src/self_run/inbox.rs b/crates/mind-body/src/self_run/inbox.rs index 61bad3b..9744502 100644 --- a/crates/mind-body/src/self_run/inbox.rs +++ b/crates/mind-body/src/self_run/inbox.rs @@ -18,7 +18,7 @@ use eureka_pipeline::PipelineKind; use huginn_mind::{PipelineFacts, PipelineStatusSummary}; use crate::persona; -use crate::queue::{MindPort, any_of, headers, in_force, of_kind}; +use crate::queue::{MindPort, any_of, headers, in_force, of_kind, on_hold}; /// The owner a follow-up names when it is a brief for her Self. pub const OWNER: &str = "self-run"; @@ -65,9 +65,19 @@ fn open_briefs(mind: &dyn MindPort) -> Result> { Ok(briefs) } -/// Her briefs in force that the session has not been told. +/// Her briefs in force that the session has not been told and no open question +/// holds: a held brief is neither claimed nor told until its question is +/// answered or withdrawn. pub fn derive(mind: &dyn MindPort, told: &HashSet) -> Result> { - Ok(open_briefs(mind)?.into_iter().filter(|brief| !told.contains(&brief.id)).collect()) + Ok(derive_counting_held(mind, told)?.0) +} + +/// `derive`, and how many of her briefs in force an open question holds. +pub fn derive_counting_held(mind: &dyn MindPort, told: &HashSet) -> Result<(Vec, usize)> { + let holds = on_hold(mind)?; + let (held, free): (Vec, Vec) = + open_briefs(mind)?.into_iter().partition(|brief| holds.iter().any(|item| item.kind == PipelineKind::FollowUp && item.id.0 == brief.id)); + Ok((free.into_iter().filter(|brief| !told.contains(&brief.id)).collect(), held.len())) } /// Her briefs in force that a closed run claimed, with that run's id: a list for diff --git a/crates/mind-body/tests/public_surface.txt b/crates/mind-body/tests/public_surface.txt index 02f8d4d..a53aa33 100644 --- a/crates/mind-body/tests/public_surface.txt +++ b/crates/mind-body/tests/public_surface.txt @@ -114,7 +114,7 @@ mind_body::heartbeat: struct PersonaId # [derive (Clone , Debug , PartialEq , Eq mind_body::heartbeat: struct PollTrace # [derive (Clone , Debug , PartialEq , Eq , DatabaseEntry)] pub-fields[at: DateTime < Utc >, tempo_s: u32, instances: Vec < InstanceTrace >, launches: Vec < String >] mind_body::heartbeat: struct Seat< 'a > pub-fields[who: PersonaId, mind: & 'a dyn MindPort, chat: & 'a dyn Chat] mind_body::launch: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Declined { Admission (MindRefusal) , Conflict , AlreadyAdmitted , StartFailed , GrantForOtherInstance , PersonaRunTakesNoClaims , } -mind_body::launch: # [derive (Clone , Debug , PartialEq , Eq)] pub enum LaunchOutcome { Launched { run : PipelineRef , } , NotQueued { item : PipelineRef , } , Tripped { item : PipelineRef , question : PipelineRef , } , Busy { run : PipelineRef , } , Refused (Declined) , } +mind_body::launch: # [derive (Clone , Debug , PartialEq , Eq)] pub enum LaunchOutcome { Launched { run : PipelineRef , } , NotQueued { item : PipelineRef , } , Tripped { run : PipelineRef , question : PipelineRef , } , Busy { run : PipelineRef , } , Refused (Declined) , } mind_body::launch: const START_GRACE_S : i64 mind_body::launch: fn open_and_launch (ports : & Ports , request : LaunchRequest) -> Result < LaunchOutcome > mind_body::launch: fn run_label (now : DateTime < Utc >) -> Label @@ -195,11 +195,12 @@ mind_body::persona: struct StartSelfRun # [derive (Clone , Debug , PartialEq , E mind_body::persona: struct TraceItem # [derive (Clone , Debug , PartialEq , Eq , Serialize , Deserialize)] pub-fields[provenance: String, room: String, text: String] mind_body::persona: struct TurnRequest # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[caller: String, request_id: String, instructions: String, input: String, output_schema_json: String] mind_body::persona: struct TurnTrace # [derive (Clone , Debug , PartialEq , Eq , DatabaseEntry)] pub-fields[run: String, turn: u32, at: DateTime < Utc >, inputs: Vec < TraceItem >, outputs: Vec < TraceItem >, outcome: String] -mind_body::queue: # [derive (Clone , Copy , Debug , PartialEq , Eq)] pub enum Breaker { Clear , Trip , } +mind_body::queue: # [derive (Clone , Debug , PartialEq , Eq)] pub enum Breaker { Clear , Trip { run : PipelineRef , } , } mind_body::queue: const CONSECUTIVE : usize mind_body::queue: const EMPTY_SHARE : usize mind_body::queue: const WINDOW : usize mind_body::queue: fn breaker (mind : & dyn MindPort , item : & PipelineRef , ending : & [PipelineRef]) -> Result < Breaker > +mind_body::queue: fn on_hold (mind : & dyn MindPort) -> Result < Vec < PipelineRef > > mind_body::queue: fn queue (mind : & dyn MindPort , instance : & Slug , ending : & [PipelineRef]) -> Result < Vec < PipelineRef > > mind_body::queue: impl HuginnMind < 'static > mind_body::queue: impl HuginnMind < 'static >: fn new (client : HuginnClient) -> Self @@ -259,6 +260,7 @@ mind_body::self_run::decide: impl State mind_body::self_run::inbox: const OWNER : & str mind_body::self_run::inbox: fn backlog (mind : & dyn MindPort) -> Result < Vec < (Brief , String) > > mind_body::self_run::inbox: fn derive (mind : & dyn MindPort , told : & HashSet < String >) -> Result < Vec < Brief > > +mind_body::self_run::inbox: fn derive_counting_held (mind : & dyn MindPort , told : & HashSet < String >) -> Result < (Vec < Brief > , usize) > mind_body::self_run::inbox: fn render (briefs : & [Brief] , plan : & str) -> String mind_body::self_run::inbox: fn resolving_sentence (id : & str) -> String mind_body::self_run::inbox: struct Brief # [derive (Clone , Debug , PartialEq , Eq)] pub-fields[id: String, campaign: String, repo: String, item: String, ordinal: u64]