queue: one owner of the hold; a trip belongs to the run whose end tripped it and its question holds that run's claims
Cut claimable's Mind-wide stop on any follow-up question and queue's inline blocked loop. queue::on_hold is the one reader of what an in-force question holds (a question raised in a run holds that run's claims, anything else holds itself); queue, inbox::derive and the open_and_launch gate read it. Breaker::Trip names the newest ended counted run; the breaker question is raised in that run; the boundary is the newest of a question raised in the item and a run claiming it that a question is raised in, so an answer or withdrawal gives every claim of the run a fresh count. The trip and the held-brief wait are written to the heartbeat trace. Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens
parent
2fb2608249
commit
1b19180555
5 changed files with 217 additions and 70 deletions
|
|
@ -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<PipelineRef>) -
|
|||
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<PipelineRef>) -
|
|||
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<Vec<PipelineRef>> {
|
||||
/// 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<PipelineRef>, 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<Outcome> {
|
|||
let mut passed: Vec<Standing> = 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) => {
|
||||
|
|
|
|||
|
|
@ -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<LaunchOu
|
|||
|
||||
if turn == RunTurn::SelfRun {
|
||||
let open = queue(ports.mind, &instance, &ending)?;
|
||||
if let Some(item) = claims.iter().find(|claim| claim.kind == PipelineKind::CutSpec && !open.contains(claim)) {
|
||||
let held = on_hold(ports.mind)?;
|
||||
if let Some(item) = claims.iter().find(|claim| held.contains(claim) || (claim.kind == PipelineKind::CutSpec && !open.contains(claim))) {
|
||||
return Ok(LaunchOutcome::NotQueued { item: item.clone() });
|
||||
}
|
||||
for item in &claims {
|
||||
if breaker(ports.mind, item, &ending)? == Breaker::Trip {
|
||||
let question = PipelineDocument::Question(breaker_question(item, &label, now));
|
||||
if let Breaker::Trip { run } = breaker(ports.mind, item, &ending)? {
|
||||
let held = claims_of(ports.mind, &run)?.len();
|
||||
let question = PipelineDocument::Question(breaker_question(item, &run, held, &label, now));
|
||||
let id = PipelineRef { kind: PipelineKind::Question, id: key_of(&question)?.as_str().into() };
|
||||
batch.push(question);
|
||||
return Ok(match committed(ports.mind.admit(&agent, &label.0, batch)?) {
|
||||
Ok(()) => 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<Utc>) -> 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<Utc>) -> 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);
|
||||
|
|
|
|||
|
|
@ -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<Vec<PipelineRef>> {
|
||||
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<Vec<PipelineRef>> {
|
||||
let mut held: Vec<PipelineRef> = 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<Vec<PipelineRef>> {
|
||||
let unreported = Selection { cited: Some(Incoming { role: "cut_spec".into(), exists: false }), ..in_force(of_kind(PipelineKind::CutSpec)) };
|
||||
let mut blocked: Vec<PipelineRef> = 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<Breaker> {
|
||||
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<usize> = (0..ran.len()).filter(|index| ran[*index].1).collect();
|
||||
let considered: Vec<usize> = 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::<PipelineRef>::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]
|
||||
|
|
|
|||
|
|
@ -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<Vec<Brief>> {
|
|||
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<String>) -> Result<Vec<Brief>> {
|
||||
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<String>) -> Result<(Vec<Brief>, usize)> {
|
||||
let holds = on_hold(mind)?;
|
||||
let (held, free): (Vec<Brief>, Vec<Brief>) =
|
||||
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
|
||||
|
|
|
|||
|
|
@ -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]
|
||||
|
|
|
|||
Loading…
Reference in a new issue