GameCult/Huginn
Watch
9
0
Fork
You've already forked Huginn
 0
Code Issues Pull requests 1 Projects Releases Packages Wiki Activity Actions

hold: one decider of what a question holds, read once per launch; a dead Self holder no longer stalls the successor; retire is read

Browse source 
s5 findings of cut-self-session-live:
- blocked-recipe-misses-run-held-spec: eureka_state::hold is the one decider
  (a question holds the item it is raised in, or every claim of the run it is
  raised in; an answer of retire keeps holding the claims still in force).
  queue::on_hold and `huginn blocked` both read it.
- hold-read-not-one-snapshot: on_hold returns a Held snapshot that the request
  carries; claimable, queue and the gate read it, none reads it again.
- dead-self-holder-stalls-successor: launch::slot is the one reader of whether
  a holder lives; claimable asks it, a dead holder is closed by the launch and
  the beat's trace says so.
- retire-answer-reopens: launch::retire_answered resolves the claims of an
  answered-retire question Withdrawn; the beat runs it before reading briefs.
- held-note-threshold-unpinned: a test pins the note for exactly one held brief.

Co-Authored-By: Claude Sonnet 5.5 <noreply@anthropic.com>
This commit is contained in:
Meta Vens 2026-11-01 04:11:38 +01:00
commit 3e48e122af
14 changed files with 1139 additions and 171 deletions
Download patch file Download diff file Expand all files Collapse all files

95
crates/eureka-state/src/bin/huginn/main.rs
View file

@ -22,13 +22,17 @@ mod render;
mod schema;
mod trouble;
use std::cell::{Cell, RefCell};
use std::io::Write;
use std::process::ExitCode;
use std::sync::OnceLock;
use clap::{CommandFactory, FromArgMatches, Parser, Subcommand};
use cultnet_rs::{PROJECTION_DOCUMENT, Selection};
use huginn_mind::eureka_pipeline::{Line, PipelineDocument, PipelineRef, Short, Slug, pipeline_key};
use eureka_state::hold;
use huginn_mind::eureka_pipeline::{
Line, PipelineDocument, PipelineKind, PipelineRef, Short, Slug, pipeline_key,
};
use huginn_mind::{
BATCH_MAX, Faculty, HuginnMindRequest, HuginnMindResponse, MindRefusal, PipelineAdmissionBatch, PipelineAdmissionOutcome,
PipelinePageItems, PipelineProvenance, PipelineSelectionPage, SemanticQuery, SourceQuery,
@ -182,7 +186,15 @@ fn run(settings: &Settings, command: Command) -> Result<u8, Trouble> {
let semantic = near.map(|text| SemanticQuery { text: Line(text), top_k: top_k.unwrap_or(DEFAULT_TOP_K) });
query(settings, selection, semantic, None, documents)
}
Command::Sources { text, repo, top_k, json } => sources(settings, text, repo, top_k, json),
Command::Sources {
text,
repo,
top_k,
json,
} => sources(settings, text, repo, top_k, json),
Command::Recipe(Recipe::Blocked { spec_id, page }) => {
blocked(settings, &spec_id, page.cursor, page.documents)
}
Command::Recipe(recipe) => {
let ask = recipe.ask()?;
query(settings, door::selection_value(&ask.selection)?, ask.semantic, ask.page.cursor, ask.page.documents)
@ -264,9 +276,82 @@ fn query(
}
}
fn sources(settings: &Settings, text: String, repo: Option<String>, top_k: Option<u32>, json: bool) -> Result<u8, Trouble> {
let query = SourceQuery { text: Line(text), top_k: top_k.unwrap_or(DEFAULT_TOP_K), repo: repo.map(Short) };
match settings.ask(|instance| HuginnMindRequest::SearchSources { instance: instance.clone(), query })? {
/// The open questions that hold a spec, and the answered-`retire` ones whose
/// claims are not yet resolved: `eureka_state::hold`'s answer, the one the
/// heartbeat's gate reads, printed as a selection of those questions.
fn blocked(
settings: &Settings,
spec_id: &str,
cursor: Option<String>,
documents: bool,
) -> Result<u8, Trouble> {
let spec = PipelineRef {
kind: door::kind_of(spec_id)?,
id: Short(spec_id.to_string()),
};
let failure: RefCell<Option<Trouble>> = RefCell::new(None);
let as_of = Cell::new(0);
let ask = |selection: &Selection| -> anyhow::Result<PipelineSelectionPage> {
let selection = selection.clone();
match settings.ask(|instance| HuginnMindRequest::Query {
instance: instance.clone(),
selection,
semantic: None,
}) {
Ok(HuginnMindResponse::Query(page)) => {
as_of.set(page.as_of);
Ok(page)
}
Ok(_) => Err(anyhow::anyhow!(
"the mind answered a query with something else"
)),
Err(trouble) => {
*failure.borrow_mut() = Some(trouble);
Err(anyhow::anyhow!("the mind could not be asked"))
}
}
};
let holding = match hold::holds(&ask) {
Ok(holds) => holds,
Err(error) => {
return Err(failure
.take()
.unwrap_or_else(|| Trouble::Internal(error.to_string())));
}
};
let keys: Vec<String> = holding
.iter()
.filter(|held| held.items.contains(&spec))
.map(|held| held.question.id.id.0.clone())
.collect();
if keys.is_empty() {
say(format_args!("matched 0 as_of {} next -", as_of.get()));
return Ok(0);
}
let selection = Selection {
schemas: Some(vec![PipelineKind::Question.type_id().to_string()]),
keys: Some(keys),
..Selection::default()
};
query(settings, selection, None, cursor, documents)
}
fn sources(
settings: &Settings,
text: String,
repo: Option<String>,
top_k: Option<u32>,
json: bool,
) -> Result<u8, Trouble> {
let query = SourceQuery {
text: Line(text),
top_k: top_k.unwrap_or(DEFAULT_TOP_K),
repo: repo.map(Short),
};
match settings.ask(|instance| HuginnMindRequest::SearchSources {
instance: instance.clone(),
query,
})? {
HuginnMindResponse::Sources(found) => {
if json {
say(serde_json::to_string_pretty(&found).map_err(|error| Trouble::Internal(error.to_string()))?);

33
crates/eureka-state/src/bin/huginn/recipes.rs
View file

@ -1,6 +1,8 @@
//! The campaign progress view as named subcommands. Each recipe is exactly the
//! selection `references/campaign-state.md` gives (Eureka 278b881, "Recipes"),
//! built per call and never cached. Each one's `--help` prints its selection.
//! The one exception is `blocked`, which asks what a question holds and so
//! reads through `eureka_state::hold`, the decider the heartbeat's gate reads.
use clap::{Args, Subcommand};
use huginn_mind::SemanticQuery;
@ -66,7 +68,7 @@ pub enum Recipe {
#[command(flatten)]
page: Paging,
},
/// Open questions raised in a spec: the spec waits on the operator.
/// Questions that hold a spec, answered-retire ones included: the spec waits on the operator.
Blocked {
spec_id: String,
#[command(flatten)]
@ -125,7 +127,12 @@ impl Recipe {
Self::Open { campaigns, page } => (open(&campaigns), None, page),
Self::Findings { campaign, blocking, page } => (findings(&campaign, blocking), None, page),
Self::Unreported { campaign, page } => (unreported(&campaign), None, page),
Self::Blocked { spec_id, page } => (blocked(&spec_id), None, page),
// Several selections, not one: `main` answers it through `eureka_state::hold`.
Self::Blocked { .. } => {
return Err(Trouble::Internal(
"blocked is answered by the hold, not by one selection".into(),
));
}
Self::Unjudged { campaign, page } => (unjudged(&campaign), None, page),
Self::History { id, page } => (history(door::kind_of(&id)?.type_id(), &id), None, page),
Self::Cut { campaign, label, page } => (cut(&campaign, &label), None, page),
@ -198,14 +205,6 @@ fn unreported(campaign: &str) -> Value {
})
}
fn blocked(spec_id: &str) -> Value {
json!({
"schemas": [schema(PipelineKind::Question)],
"fields": [in_force()],
"cites": { "target": { "schemaId": schema(PipelineKind::CutSpec), "recordKey": spec_id }, "role": "raised_in" },
})
}
fn unjudged(campaign: &str) -> Value {
json!({
"schemas": [schema(PipelineKind::CutReport)],
@ -257,6 +256,18 @@ fn gaps() -> Value {
})
}
/// `blocked` is not one selection: it reads the in-force questions and, for each,
/// what it holds, by `eureka_state::hold`, the decider the heartbeat's gate reads.
fn blocked_help() -> String {
let questions = json!({ "schemas": [schema(PipelineKind::Question)], "fields": [in_force()] });
format!(
"selection (first of several):
{}
A question holds the item it is raised in, or every claim of the run it is raised in; one answered `retire` keeps holding the claims still in force. The recipe prints the questions that hold <id>.",
serde_json::to_string_pretty(&questions).unwrap_or_default()
)
}
/// Each recipe's `--help` addendum: its selection with placeholders in the
/// arguments' places. The same builders as execution, so help cannot drift
/// from what runs.
@ -276,7 +287,7 @@ pub fn help() -> Vec<(&'static str, String)> {
("open", selection(open(&[c()]))),
("findings", findings_help),
("unreported", selection(unreported(&c()))),
("blocked", selection(blocked("<id>"))),
("blocked", blocked_help()),
("unjudged", selection(unjudged(&c()))),
("history", selection(history("epiphany.pipeline.<kind>.v2", "<id>"))),
("cut", selection(cut(&c(), "<label>"))),

173
crates/eureka-state/src/hold.rs Normal file
View file

@ -0,0 +1,173 @@
//! What an open question holds: the one decider, read by the heartbeat's gate
//! and by `huginn blocked`, so the two cannot disagree.
//!
//! 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. A
//! question answered with the option `retire` keeps holding the claims of the
//! run it was raised in for as long as they are in force: the answer has said
//! they are not to be worked, and whoever resolves them is the one writer of
//! that. Answering or withdrawing a question with any other choice lifts its
//! hold. The question's text is display only.
//!
//! Nothing here keeps state. Every answer is read through `ask`, one typed
//! selection at a time, so a caller with a daemon and a caller with a
//! temporary mind read the same way.
use anyhow::{Result, bail};
use cultnet_rs::{Citation, FieldPredicate, Incoming, RecordRef, Selection};
use huginn_mind::eureka_pipeline::{PipelineKind, PipelineRef};
use huginn_mind::{
PipelineDocumentSummary, PipelineFacts, PipelinePageItems, PipelineSelectionPage,
PipelineStatusSummary,
};
/// The option label of a breaker question whose answer retires the claims it holds.
pub const RETIRE: &str = "retire";
/// One page of a selection, from whoever reads the mind.
pub type Ask<'a> = dyn Fn(&Selection) -> Result<PipelineSelectionPage> + 'a;
/// A question and the items it holds.
#[derive(Clone, Debug, PartialEq, Eq)]
pub struct Hold {
pub question: PipelineDocumentSummary,
pub items: Vec<PipelineRef>,
/// The question is answered `retire`: the items are held until they are resolved.
pub retired: bool,
}
/// The documents that cite `target`, under `role` or any.
pub fn citing(target: &PipelineRef, role: Option<&str>) -> Citation {
Citation {
target: RecordRef::new(target.kind.type_id(), target.id.0.clone()),
role: role.map(Into::into),
}
}
pub fn of_kind(kind: PipelineKind) -> Selection {
Selection {
schemas: Some(vec![kind.type_id().to_string()]),
..Selection::default()
}
}
pub fn any_of(mut selection: Selection, index: &str, values: &[&str]) -> Selection {
selection
.fields
.get_or_insert_with(Vec::new)
.push(FieldPredicate {
index: index.into(),
op: "any_of".into(),
values: Some(values.iter().map(|value| value.to_string()).collect()),
number: None,
});
selection
}
pub fn in_force(selection: Selection) -> Selection {
any_of(selection, "in_force", &["true"])
}
/// Every page of a selection, in the mind's order. The one paging loop.
pub fn headers(ask: &Ask, selection: &Selection) -> Result<Vec<PipelineDocumentSummary>> {
let mut all = Vec::new();
let mut asking = selection.clone();
loop {
let page = ask(&asking)?;
let PipelinePageItems::Headers(items) = page.items else {
bail!("the mind answered documents to a header selection")
};
all.extend(items);
match page.next {
Some(cursor) => asking.cursor = Some(cursor),
None => return Ok(all),
}
}
}
/// The claims of the run `run`, resolved or not (a citation outlives its
/// resolution). Empty when the mind has no such run.
pub fn claims_of(ask: &Ask, run: &PipelineRef) -> Result<Vec<PipelineRef>> {
let one = Selection {
keys: Some(vec![run.id.0.clone()]),
..of_kind(PipelineKind::Run)
};
Ok(headers(ask, &one)?
.into_iter()
.flat_map(|header| match header.facts {
PipelineFacts::Run { claims, .. } => claims,
_ => Vec::new(),
})
.collect())
}
/// Every question holding anything: the open ones, each with what it holds,
/// then the answered-`retire` ones whose claims are still in force.
pub fn holds(ask: &Ask) -> Result<Vec<Hold>> {
let mut found = Vec::new();
for question in headers(ask, &in_force(of_kind(PipelineKind::Question)))? {
if let PipelineFacts::Question {
raised_in: Some(at),
..
} = &question.facts
{
let items = if at.kind == PipelineKind::Run {
claims_of(ask, at)?
} else {
vec![at.clone()]
};
found.push(Hold {
question,
items,
retired: false,
});
}
}
let asked_in = Selection {
cited: Some(Incoming {
role: "raised_in".into(),
exists: true,
}),
..of_kind(PipelineKind::Run)
};
for run in headers(ask, &asked_in)? {
let questions = Selection {
cites: Some(citing(&run.id, Some("raised_in"))),
..of_kind(PipelineKind::Question)
};
for question in headers(ask, &questions)? {
if matches!(question.status, PipelineStatusSummary::InForce)
|| !answered_retire(ask, &question.id)?
{
continue;
}
let mut items = Vec::new();
for claim in claims_of(ask, &run.id)? {
let live = Selection {
keys: Some(vec![claim.id.0.clone()]),
..in_force(of_kind(claim.kind))
};
if !headers(ask, &live)?.is_empty() {
items.push(claim);
}
}
if !items.is_empty() {
found.push(Hold {
question,
items,
retired: true,
});
}
}
}
Ok(found)
}
/// Whether a ruling answers `question` with the option `retire`.
fn answered_retire(ask: &Ask, question: &PipelineRef) -> Result<bool> {
let answers = Selection {
cites: Some(citing(question, Some("answers"))),
..of_kind(PipelineKind::Ruling)
};
Ok(headers(ask, &answers)?.iter().any(|ruling| matches!(&ruling.facts, PipelineFacts::Ruling { choice: Some(choice), .. } if choice.0 == RETIRE)))
}

2
crates/eureka-state/src/lib.rs
View file

@ -30,6 +30,8 @@ use huginn_mind::eureka_pipeline::Slug;
use huginn_mind::{HuginnMindRequest, HuginnMindResponse, MAX_DEFERRED_BODY_BYTES, PipelineProvenance};
use serde_json::Value;
pub mod hold;
/// The correlation key of the one request a session carries.
const MESSAGE_ID: &str = "eureka-state-call";
const POLL: Duration = Duration::from_millis(2);

147
crates/eureka-state/tests/cli.rs
View file

@ -11,9 +11,11 @@ use std::sync::{Arc, Mutex};
use cultnet_rs::Selection;
use huginn_mind::envelope::{OperationFailure, decode_request, encode_failure, encode_response};
use huginn_mind::eureka_pipeline::{
CodeLocation, CommitRange, CutVerification, Date, Evidence, EvidenceKind, MutationRecord, OrgRepo, PipelineCutReport,
PipelineCutSpec, PipelineDocument, PipelineKind, PipelineRef, PipelineRuling, Promise, ReportCommit, RulingAuthority, Short,
StructuralDelta, pipeline_key,
CodeLocation, CommitRange, CutVerification, Date, Evidence, EvidenceKind, MutationRecord,
OrgRepo, PipelineCutReport, PipelineCutSpec, PipelineDocument, PipelineKind, PipelineQuestion,
PipelineRef, PipelineResolution, PipelineRuling, PipelineRun, Promise, QuestionOption,
ReportCommit, ResolutionOutcome, RulingAuthority, RunOperator, RunTurn, Short, StructuralDelta,
pipeline_key,
};
use huginn_mind::{
AdmissionFacts, HuginnMindRequest, HuginnMindResponse, IndexStatus, MindRefusal, MindStatus, PersonaStatus, PipelineAdmissionOutcome, PipelineDocumentView,
@ -747,17 +749,6 @@ fn the_recipes_send_exactly_the_pages_selections() {
"cited": { "role": "cut_spec", "exists": false },
}),
),
(
vec!["blocked", "eureka-substrate:cut_spec:cut-x.r1"],
json!({
"schemas": [schema("question")],
"fields": [in_force],
"cites": {
"target": { "schemaId": schema("cut_spec"), "recordKey": "eureka-substrate:cut_spec:cut-x.r1" },
"role": "raised_in",
},
}),
),
(
vec!["unjudged", c],
json!({
@ -802,6 +793,134 @@ fn the_recipes_send_exactly_the_pages_selections() {
}
}
/// `huginn blocked` answers with the hold the heartbeat's gate reads: a question
/// raised in a run holds every claim of that run, and an answer of `retire`
/// keeps holding the claims until they are resolved.
#[test]
fn blocked_names_the_questions_that_hold_a_spec_as_the_gate_reads_them() {
let server = live();
let dir = tempfile::tempdir().unwrap();
let admit = |name: &str, documents: Value| {
let file = write(&dir, name, &documents);
let ran = at(server.addr, &admit_args(&file), None);
assert_eq!(ran.code, 0, "{name}: {}", ran.err);
};
let value = |document: PipelineDocument| serde_json::to_value(document).unwrap();
let spec_id = |cut: &str| format!("{CAMPAIGN}:cut_spec:cut-{cut}.r1");
let held = |cut: &str| at(server.addr, &["blocked", &spec_id(cut)], None);
admit("campaign.json", json!([campaign_json()]));
admit("specs.json", json!([spec("A"), spec("B")]));
let claim = PipelineRef {
kind: PipelineKind::CutSpec,
id: Short(spec_id("A")),
};
admit(
"run.json",
json!([value(PipelineDocument::Run(PipelineRun {
instance: slug(INSTANCE),
label: "failed".into(),
operated_by: RunOperator::Mind,
turn: RunTurn::SelfRun,
host: "test".into(),
started_on: Date("2026-10-10".into()),
budget_usd: "0".into(),
claims: vec![claim.clone()],
campaigns: vec![slug(CAMPAIGN)],
}))]),
);
let run = PipelineRef {
kind: PipelineKind::Run,
id: Short(format!("{INSTANCE}:run:failed")),
};
let resolution = |subject: &PipelineRef, outcome: ResolutionOutcome| {
value(PipelineDocument::Resolution(PipelineResolution {
subject: subject.clone(),
sequence: 1,
outcome,
rationale: "So.".into(),
resolved_on: Date("2026-10-10".into()),
}))
};
admit(
"ended.json",
json!([resolution(
&run,
ResolutionOutcome::Recorded {
reason: "unit not active".into()
}
)]),
);
assert!(
held("A").out.starts_with("matched 0 "),
"no question yet: {}",
held("A").out
);
let question = format!("{CAMPAIGN}:question:fork");
admit(
"question.json",
json!([value(PipelineDocument::Question(PipelineQuestion {
campaign: slug(CAMPAIGN),
label: "fork".into(),
title: "A fork".into(),
question: "Which?".into(),
options: ["reopen", "retire"]
.iter()
.map(|name| QuestionOption {
label: (*name).into(),
text: "an option".into()
})
.collect(),
recommended: "reopen".into(),
depends: vec![],
raised_in: Some(run),
asked_on: Date("2026-10-10".into()),
}))]),
);
let shown = held("A");
assert_eq!(shown.code, 0, "{}", shown.err);
assert!(
shown.out.starts_with(&format!("{question} InForce ")) && shown.out.contains("matched 1 "),
"a question raised in the run holds its claim: {}",
shown.out
);
assert!(
held("B").out.starts_with("matched 0 "),
"and not a spec the run never claimed: {}",
held("B").out
);
let PipelineDocument::Ruling(mut answer) = ruling(None) else {
unreachable!()
};
(answer.answers, answer.choice) = (Some(Short(question.clone())), Some("retire".into()));
admit(
"answer.json",
json!([value(PipelineDocument::Ruling(answer))]),
);
let shown = held("A");
assert!(
shown.out.starts_with(&format!("{question} Resolved")) && shown.out.contains("matched 1 "),
"retired and not yet resolved, it still holds: {}",
shown.out
);
admit(
"withdrawn.json",
json!([resolution(
&claim,
ResolutionOutcome::Withdrawn {
reason: "retired".into()
}
)]),
);
assert!(
held("A").out.starts_with("matched 0 "),
"the claim resolved, nothing holds it: {}",
held("A").out
);
}
#[test]
fn paging_and_query_flags_shape_the_selection() {
let (selection, semantic) = sent(&["gaps", "--cursor", "abc", "--documents"]);

2
crates/mind-body/src/control/tests.rs
View file

@ -131,7 +131,7 @@ mod grant {
fn launch_as(mind_instance: &str, grant: Grant, turn: RunTurn) -> Result<LaunchOutcome> {
let mind = SilentMind(slug(mind_instance));
let ports = Ports { mind: &mind, launcher: &NoUnits, clock: &Now, host: "host" };
open_and_launch(&ports, LaunchRequest { turn, claims: vec![], agent: "agent".into(), grant })
open_and_launch(&ports, LaunchRequest { turn, claims: vec![], agent: "agent".into(), grant, held: crate::queue::Held::default() })
}
#[test]

135
crates/mind-body/src/heartbeat.rs
View file

@ -41,8 +41,8 @@ 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, claims_of, headers, of_kind};
use crate::launch::{Clock, LaunchOutcome, LaunchRequest, Launcher, Ports, holders, open_and_launch, retire_answered, slot};
use crate::queue::{Held, MindPort, any_of, citing, claims_of, headers, of_kind, on_hold};
use crate::self_run::inbox;
use crate::tempo::{self, Standing};
@ -557,13 +557,36 @@ enum Launched {
Refused(String),
}
fn ports<'a>(organs: &Organs<'a>, mind: &'a dyn MindPort) -> Ports<'a> {
Ports {
mind,
launcher: organs.launcher,
clock: organs.clock,
host: organs.host,
}
}
/// The one way the heartbeat starts a run: the same brake read and the same
/// opener for a Persona turn (a beat, an operator mention, a Self run that ended)
/// and for the Self run a beat gives to the briefs her Self holds.
fn launch(organs: &Organs, row: &Row, turn: RunTurn, claims: Vec<PipelineRef>) -> Result<Launched> {
let Effective::Released(grant) = organs.control.effective(row.slug()) else { return Ok(Launched::Refused("the brake is held".into())) };
let ports = Ports { mind: row.seat.mind, launcher: organs.launcher, clock: organs.clock, host: organs.host };
let request = LaunchRequest { turn, claims, agent: AGENT.into(), grant };
fn launch(
organs: &Organs,
row: &Row,
turn: RunTurn,
claims: Vec<PipelineRef>,
held: Held,
) -> Result<Launched> {
let Effective::Released(grant) = organs.control.effective(row.slug()) else {
return Ok(Launched::Refused("the brake is held".into()));
};
let request = LaunchRequest {
turn,
claims,
agent: AGENT.into(),
grant,
held,
};
let ports = ports(organs, row.seat.mind);
Ok(match open_and_launch(&ports, request) {
Ok(LaunchOutcome::Launched { run }) => Launched::Run(label_of(&run)),
Ok(LaunchOutcome::Busy { .. }) => Launched::Busy,
@ -579,25 +602,56 @@ fn launch(organs: &Organs, row: &Row, turn: RunTurn, claims: Vec<PipelineRef>) -
/// The most briefs one Self run claims.
const BRIEFS_PER_SELF_RUN: usize = 64;
/// What a Self run would claim for this Mind, read at one instant.
struct Claimable {
/// The briefs in force that no live Self run holds and no open question holds.
claims: Vec<PipelineRef>,
/// How many briefs an open question holds.
held: usize,
/// The hold they were chosen by, which the launch consults and reads no second time.
hold: Held,
/// Self runs in force whose unit is gone: the launch closes them.
dead: Vec<PipelineRef>,
}
/// The briefs a Self run would claim for this Mind: those in force that no live
/// 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)> {
/// Self run holds and no 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. Whether a holder lives is `launch::slot`'s answer
/// (a dead one is closed by the launch, so it does not stop the successor), and
/// what is held is `queue::on_hold`'s, read once here and carried into the
/// launch; nothing here decides either.
fn claimable(organs: &Organs, row: &Row) -> Result<Claimable> {
let mind = row.seat.mind;
if !holders(mind, row.slug(), RunTurn::SelfRun)?.is_empty() {
return Ok((Vec::new(), 0));
let hold = on_hold(mind)?;
let found = slot(
&ports(organs, mind),
row.slug(),
RunTurn::SelfRun,
organs.clock.now(),
)?;
if found.live.is_some() {
return Ok(Claimable {
claims: Vec::new(),
held: 0,
hold,
dead: Vec::new(),
});
}
let (free, held) = inbox::derive_counting_held(mind, &HashSet::new())?;
let (free, held) = inbox::derive_counting_held(mind, &hold, &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();
Ok((claims, held))
Ok(Claimable {
claims,
held,
hold,
dead: found.dead,
})
}
/// One poll.
@ -664,7 +718,7 @@ pub fn poll(organs: &Organs, config: &Config) -> Result<Outcome> {
if mentioned.is_none() && !self_ended {
continue;
}
match launch(organs, row, RunTurn::PersonaTurn, vec![])? {
match launch(organs, row, RunTurn::PersonaTurn, vec![], Held::default())? {
Launched::Run(label) => {
if let Some(key) = mentioned {
state.handled_mentions.push(key);
@ -703,15 +757,44 @@ 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, held) = claimable(&rows[winner]).unwrap_or_else(|error| {
rows[winner].note = format!("briefs: {error}");
(Vec::new(), 0)
});
if held > 0 {
rows[winner].note = format!("{held} briefs held by open questions");
// An answer of retire is written before the briefs are read, so a retired brief is resolved, not claimed.
if let Err(error) = retire_answered(rows[winner].seat.mind, AGENT, now) {
rows[winner].note = format!("retire: {error}");
}
let turn = if claims.is_empty() { RunTurn::PersonaTurn } else { RunTurn::SelfRun };
match launch(organs, &rows[winner], turn, claims)? {
let Claimable {
claims,
held,
hold,
dead,
} = claimable(organs, &rows[winner]).unwrap_or_else(|error| {
rows[winner].note = format!("briefs: {error}");
Claimable {
claims: Vec::new(),
held: 0,
hold: Held::default(),
dead: Vec::new(),
}
});
let mut notes: Vec<String> = Vec::new();
if held > 0 {
notes.push(format!("{held} briefs held by open questions"));
}
if !claims.is_empty() && !dead.is_empty() {
let ids: Vec<&str> = dead.iter().map(|run| run.id.0.as_str()).collect();
notes.push(format!(
"closing the Self run whose unit is gone: {}",
ids.join(", ")
));
}
if !notes.is_empty() {
rows[winner].note = notes.join("; ");
}
let turn = if claims.is_empty() {
RunTurn::PersonaTurn
} else {
RunTurn::SelfRun
};
match launch(organs, &rows[winner], turn, claims, hold)? {
Launched::Run(label) => {
state.last_beat_at = now;
rows[winner].busy = turn == RunTurn::PersonaTurn;

141
crates/mind-body/src/heartbeat/tests.rs
View file

@ -1100,3 +1100,144 @@ fn a_beat_with_no_held_brief_writes_no_held_note() {
let notes: Vec<String> = rig.store.traces.borrow().iter().flat_map(|trace| trace.instances.iter().map(|instance| instance.note.clone())).collect();
assert!(notes.iter().all(|note| !note.contains("held")), "{notes:?}");
}
/// Her Self run in force whose unit is gone, claiming `claims`, admitted `ago` seconds before now.
fn crashed_self_run(rig: &Rig, claims: &[PipelineRef], ago: i64) {
rig.minds[ANN].admit_at(rig.now() - Duration::seconds(ago));
rig.minds[ANN].committed(vec![run_in(
"ann",
"crashed",
RunTurn::SelfRun,
RunOperator::Mind,
claims,
)]);
}
/// Polls on, a beat apart, leaving every run to its unit: nothing here ends one.
fn polls(rig: &Rig, count: usize) {
for _ in 0..count {
rig.advance(1000);
rig.poll();
}
}
fn notes_of(rig: &Rig) -> Vec<String> {
rig.store
.traces
.borrow()
.iter()
.flat_map(|trace| trace.instances.iter().map(|instance| instance.note.clone()))
.collect()
}
#[test]
fn a_self_run_left_in_force_by_a_crash_is_closed_by_the_next_beat_and_the_successor_launches() {
let (rig, source) = rig_with_a_persona_turn();
brief_by(&rig, ANN, crate::persona::AGENT, "orphan", &source);
crashed_self_run(&rig, &[brief_ref("orphan")], 3600);
assert_eq!(self_runs(&rig, ANN).len(), 1);
polls(&rig, 8);
let runs = self_runs(&rig, ANN);
assert_eq!(runs.len(), 2, "one successor, once");
assert!(
matches!(
rig.minds[ANN]
.view(K::Run, "ann:run:crashed")
.unwrap()
.status,
huginn_mind::PipelineStatus::Resolved { .. }
),
"the dead holder was closed by the launch"
);
assert_eq!(
claims_of_run(&runs[1]),
vec![brief_ref("orphan").id.0],
"claiming the brief the dead run left"
);
assert!(
notes_of(&rig)
.contains(&"closing the Self run whose unit is gone: ann:run:crashed".to_string()),
"{:?}",
notes_of(&rig)
);
}
#[test]
fn a_self_run_whose_unit_may_still_be_starting_or_is_running_gets_no_successor() {
let (rig, source) = rig_with_a_persona_turn();
brief_by(&rig, ANN, crate::persona::AGENT, "orphan", &source);
crashed_self_run(&rig, &[brief_ref("orphan")], 30);
rig.advance(10);
rig.poll();
assert_eq!(
self_runs(&rig, ANN).len(),
1,
"inside the start grace the unit may still be starting"
);
rig.launcher
.revive_as(&Slug("ann".into()), RunTurn::SelfRun);
rig.advance(3600);
rig.poll();
polls(&rig, 4);
assert_eq!(
self_runs(&rig, ANN).len(),
1,
"a run whose unit is active holds the briefs however old it is"
);
assert!(matches!(
rig.minds[ANN]
.view(K::Run, "ann:run:crashed")
.unwrap()
.status,
huginn_mind::PipelineStatus::InForce
));
}
#[test]
fn one_held_brief_says_so_in_the_trace() {
let (rig, source) = rig_with_a_persona_turn();
brief_by(&rig, ANN, crate::persona::AGENT, "solo", &source);
fail_bundle(&rig, &[brief_ref("solo")]);
beats_until(&rig, 12, || {
notes_of(&rig).contains(&"1 briefs held by open questions".to_string())
});
assert!(
notes_of(&rig).contains(&"1 briefs held by open questions".to_string()),
"{:?}",
notes_of(&rig)
);
}
#[test]
fn an_answer_of_retire_resolves_the_bundle_and_nothing_relaunches() {
let (rig, source) = rig_with_a_persona_turn();
let items = failed_bundle(&rig, &source);
beats(&rig, 3);
let question = rig.minds[ANN]
.ids(K::Question)
.pop()
.expect("the trip asked");
assert_eq!(self_runs(&rig, ANN).len(), 3);
rig.minds[ANN].committed(vec![crate::testkit::ruling(
"answer",
Some((&question, "retire")),
)]);
beats(&rig, 6);
assert_eq!(
self_runs(&rig, ANN).len(),
3,
"no Self run after the answer: retire does not relaunch"
);
for item in &items {
let view = rig.minds[ANN].view(K::FollowUp, &item.id.0).unwrap();
assert!(
matches!(view.status, huginn_mind::PipelineStatus::Resolved { record, .. } if matches!(record.outcome, eureka_pipeline::ResolutionOutcome::Withdrawn { .. })),
"{item:?} is resolved Withdrawn"
);
}
assert_eq!(
rig.minds[ANN].ids(K::Question).len(),
1,
"and no second question"
);
}

263
crates/mind-body/src/launch.rs
View file

@ -1,9 +1,9 @@
//! The one way her organs open a run and start its unit.
//!
//! `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 be held by an open question, and none may
//! hers of the request's turn are read through `slot`, which tells a live one
//! (its unit active, or admitted within `START_GRACE_S`: it answers Busy) from a
//! dead one (to be closed Recorded); (1) every cut-spec claim must be in the queue; (2) no claim may be held by an open question (the `Held` snapshot the request carries), 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
@ -26,10 +26,16 @@ use eureka_pipeline::{
Date, Label, PipelineDocument, PipelineKind, PipelineQuestion, PipelineRef, PipelineResolution, PipelineRun, QuestionOption, ResolutionOutcome,
RunOperator, RunTurn, Slug, pipeline_key,
};
use huginn_mind::{MindRefusal, PipelineAdmissionOutcome, PipelineDocumentSummary, PipelineFacts};
use huginn_mind::{
BATCH_MAX, MindRefusal, PipelineAdmissionOutcome, PipelineDocumentSummary, PipelineFacts,
};
use crate::control::Grant;
use crate::queue::{Breaker, MindPort, any_of, breaker, citing, claims_of, headers, in_force, of_kind, on_hold, queue};
use crate::queue::{
Breaker, Held, MindPort, any_of, breaker, citing, claims_of, headers, holds, in_force, of_kind,
queue,
};
use eureka_state::hold::RETIRE;
/// 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.
@ -221,6 +227,10 @@ pub struct LaunchRequest {
/// The reading of a released instance brake and a released swarm brake.
/// Taken by value, so one reading opens at most one run.
pub grant: Grant,
/// The hold the caller chose its claims by, read once with `on_hold`: the
/// gate refuses against it and reads it no second time. A Persona turn
/// claims nothing, so it carries the empty one.
pub held: Held,
}
/// Why admission did not open the run: its own refusal, or one of the two
@ -320,8 +330,43 @@ pub(crate) fn holders(mind: &dyn MindPort, instance: &Slug, turn: RunTurn) -> Re
.collect())
}
/// The holders of a turn's slot, told apart: the live one, if any (its unit
/// active, or admitted within `START_GRACE_S`), and the dead, which the next
/// launch closes. The one reader of whether a holder lives: step 0 and the
/// heartbeat's claim of briefs both ask here.
pub(crate) struct Slot {
pub live: Option<PipelineRef>,
pub dead: Vec<PipelineRef>,
}
pub(crate) fn slot(
ports: &Ports,
instance: &Slug,
turn: RunTurn,
now: DateTime<Utc>,
) -> Result<Slot> {
let mut found = Slot {
live: None,
dead: Vec::new(),
};
for holder in holders(ports.mind, instance, turn)? {
if ports.launcher.alive(instance, turn) || !past_grace(&holder.admission.admitted_at, now) {
found.live = Some(holder.id);
return Ok(found);
}
found.dead.push(holder.id);
}
Ok(found)
}
pub fn open_and_launch(ports: &Ports, request: LaunchRequest) -> Result<LaunchOutcome> {
let LaunchRequest { turn, claims, agent, grant } = request;
let LaunchRequest {
turn,
claims,
agent,
grant,
held,
} = request;
let instance = ports.mind.instance().clone();
if grant.instance() != &instance {
return Ok(LaunchOutcome::Refused(Declined::GrantForOtherInstance));
@ -336,25 +381,31 @@ pub fn open_and_launch(ports: &Ports, request: LaunchRequest) -> Result<LaunchOu
// in the batch that admits whatever comes next.
let mut batch: Vec<PipelineDocument> = Vec::new();
let mut ending: Vec<PipelineRef> = Vec::new();
for holder in holders(ports.mind, &instance, turn)? {
if ports.launcher.alive(&instance, turn) || !past_grace(&holder.admission.admitted_at, now) {
return Ok(LaunchOutcome::Busy { run: holder.id });
}
batch.push(close_dead(ports.mind, &holder.id, now)?);
ending.push(holder.id);
let found = slot(ports, &instance, turn, now)?;
if let Some(run) = found.live {
return Ok(LaunchOutcome::Busy { run });
}
for dead in found.dead {
batch.push(close_dead(ports.mind, &dead, now)?);
ending.push(dead);
}
if turn == RunTurn::SelfRun {
let open = queue(ports.mind, &instance, &ending)?;
let held = on_hold(ports.mind)?;
if let Some(item) = claims.iter().find(|claim| held.contains(claim) || (claim.kind == PipelineKind::CutSpec && !open.contains(claim))) {
let open = queue(ports.mind, &instance, &ending, &held)?;
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 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() };
let count = claims_of(ports.mind, &run)?.len();
let question =
PipelineDocument::Question(breaker_question(item, &run, count, &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 { run, question: id },
@ -427,6 +478,34 @@ fn committed(outcome: PipelineAdmissionOutcome) -> Result<(), Declined> {
}
}
/// Resolves, Withdrawn, every item an answer `retire` keeps held that is still
/// in force, in the admissions of one batch each: the one writer of a
/// retirement, so that a retired claim leaves the follow-up lists and the queue
/// instead of waiting held for ever. Returns how many it resolved.
pub fn retire_answered(mind: &dyn MindPort, agent: &str, now: DateTime<Utc>) -> Result<usize> {
let retired: Vec<PipelineRef> = holds(mind)?
.into_iter()
.filter(|hold| hold.retired)
.flat_map(|hold| hold.items)
.collect();
let label = run_label(now);
let mut resolved = 0;
for chunk in retired.chunks(BATCH_MAX) {
let mut batch = Vec::new();
for item in chunk {
let outcome = ResolutionOutcome::Withdrawn {
reason: "retired by the answer to a breaker question".into(),
};
batch.push(closure(mind, item, now, outcome, "The operator answered the breaker question for the run that claimed this item with retire, so it is no longer worked.")?);
}
match committed(mind.admit(agent, &format!("retire-{}", label.0), batch)?) {
Ok(()) => resolved += chunk.len(),
Err(_) => return Err(anyhow!("the retired claims could not be resolved")),
}
}
Ok(resolved)
}
/// `breaker-<run label>`: one question per trip, since each trip is a launch
/// with a label of its own.
fn breaker_label(label: &Label) -> Label {
@ -441,14 +520,14 @@ fn breaker_question(claim: &PipelineRef, run: &PipelineRef, held: usize, label:
label: breaker_label(label),
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 {} and the runs before it left nothing that cites the work it held (claims held: {held}), so none of it launches again. Answering with reopen, or withdrawing this, reopens every claim of that run with a fresh count; answering with retire resolves every claim of it Withdrawn.",
run.id.0
)
.as_str()
.into(),
options: vec![
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() },
QuestionOption { label: RETIRE.into(), text: "Answer to retire every claim of the run: each is resolved Withdrawn and none is worked again.".into() },
],
recommended: "reopen".into(),
depends: vec![],
@ -489,7 +568,16 @@ mod tests {
fn try_launch_through(&self, mind: &dyn MindPort, turn: RunTurn, claims: &[PipelineRef]) -> Result<LaunchOutcome> {
let ports = Ports { mind, launcher: &self.launcher, clock: &self.clock, host: "yggdrasil-host" };
let grant = Grant::for_test(Slug(INSTANCE.into()));
open_and_launch(&ports, LaunchRequest { turn, claims: claims.to_vec(), agent: "agent-x".into(), grant })
open_and_launch(
&ports,
LaunchRequest {
turn,
claims: claims.to_vec(),
agent: "agent-x".into(),
grant,
held: crate::queue::on_hold(mind)?,
},
)
}
fn at(&self, milliseconds: i64) {
@ -950,6 +1038,128 @@ mod tests {
assert!(matches!(rig.launch(RunTurn::SelfRun, &[brief]), LaunchOutcome::Launched { .. }), "withdrawn, the hold lifts");
}
/// A mind where a question lands the moment the hold has been read once.
struct Landing<'a> {
inner: &'a TestMind,
question: RefCell<Option<Vec<PipelineDocument>>>,
hold_reads: Cell<u32>,
}
impl MindPort for Landing<'_> {
fn instance(&self) -> &Slug {
self.inner.instance()
}
fn query(&self, selection: &Selection) -> Result<huginn_mind::PipelineSelectionPage> {
let answer = self.inner.query(selection);
let reads_the_hold = selection.schemas.as_deref()
== Some(&[K::Question.type_id().to_string()][..])
&& selection
.fields
.iter()
.flatten()
.any(|predicate| predicate.index == "in_force");
if reads_the_hold {
self.hold_reads.set(self.hold_reads.get() + 1);
if let Some(documents) = self.question.take() {
self.inner.committed(documents);
}
}
answer
}
fn admit(
&self,
agent: &str,
session: &str,
documents: Vec<PipelineDocument>,
) -> Result<Outcome> {
self.inner.admit(agent, session, documents)
}
}
#[test]
fn a_launch_reads_the_hold_once_and_a_question_landing_after_it_changes_no_answer_within_the_launch()
{
let rig = Rig::new();
rig.mind.committed(vec![spec("x")]);
let landing = Landing {
inner: &rig.mind,
question: RefCell::new(Some(vec![question_in("fork", &spec_ref("x"))])),
hold_reads: Cell::new(0),
};
// The request's snapshot is read first (the question lands behind it); the queue and the gate use that one.
let outcome = rig.launch_through(&landing, RunTurn::SelfRun, &[spec_ref("x")]);
assert!(
matches!(outcome, LaunchOutcome::Launched { .. }),
"{outcome:?}: queue and gate answered from one snapshot"
);
assert_eq!(
landing.hold_reads.get(),
1,
"the hold was read once for the whole launch"
);
assert_eq!(
on_hold(&rig.mind).unwrap().items(),
[spec_ref("x")],
"the question did land"
);
}
#[test]
fn retiring_resolves_every_claim_an_answer_of_retire_holds_once() {
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"));
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 question = id("question", "fork");
assert_eq!(
retire_answered(&rig.mind, "agent-x", rig.clock.now()).unwrap(),
0,
"an open question retires nothing"
);
rig.mind
.committed(vec![ruling("answer", Some((&question, "reopen")))]);
assert_eq!(
retire_answered(&rig.mind, "agent-x", rig.clock.now()).unwrap(),
0,
"reopen retires nothing"
);
rig.mind
.committed(vec![question_in("fork2", &run_ref("failed"))]);
rig.mind.committed(vec![ruling(
"answer2",
Some((&id("question", "fork2"), "retire")),
)]);
assert_eq!(
retire_answered(&rig.mind, "agent-x", rig.clock.now()).unwrap(),
2
);
for item in [&brief, &spec_ref("x")] {
let view = rig.mind.view(item.kind, &item.id.0).unwrap();
assert!(
matches!(view.status, PipelineStatus::Resolved { record, .. } if matches!(record.outcome, ResolutionOutcome::Withdrawn { .. })),
"{item:?}"
);
}
assert_eq!(
retire_answered(&rig.mind, "agent-x", rig.clock.now()).unwrap(),
0,
"resolved, nothing is left to retire"
);
assert!(rig.mind.ids(K::Run).len() == 1, "and nothing launched");
}
/// The one source of every drawn case: splitmix64 from a fixed seed, so a
/// failure names a case that runs again.
struct Draw(u64);
@ -1074,7 +1284,16 @@ mod tests {
let (launcher, clock) = (Recording::default(), Fixed(Cell::new(at(0))));
let ports = Ports { mind, launcher: &launcher, clock: &clock, host: "yggdrasil-host" };
let grant = Grant::for_test(Slug(grant.into()));
let outcome = open_and_launch(&ports, LaunchRequest { turn, claims, agent: "agent-x".into(), grant });
let outcome = open_and_launch(
&ports,
LaunchRequest {
turn,
claims,
agent: "agent-x".into(),
grant,
held: Held::default(),
},
);
assert!(launcher.started.borrow().is_empty());
outcome
}

4
crates/mind-body/src/persona.rs
View file

@ -59,7 +59,7 @@ use crate::connector::render_plan;
use crate::control::{ControlSource, Effective, HeldReason, derive};
use crate::heartbeat::{PollTrace, TempoView};
use crate::launch::{Clock, LaunchOutcome, LaunchRequest, Launcher, Ports, holders, open_and_launch};
use crate::queue::{MindPort, any_of, citing, headers, in_force, live_claims, of_kind};
use crate::queue::{MindPort, any_of, citing, headers, in_force, live_claims, of_kind, on_hold};
use crate::recall;
/// The admitting agent's name on everything this organ admits.
@ -955,7 +955,7 @@ fn start_self_run(
}
let Effective::Released(grant) = read(control, swarm, organs.mind.instance()) else { return Ok(Err(Failure::SelfRun)) };
let ports = Ports { mind: organs.mind, launcher: organs.launcher, clock: organs.clock, host: organs.host };
let request = LaunchRequest { turn: RunTurn::SelfRun, claims: vec![claim], agent: AGENT.into(), grant };
let request = LaunchRequest { turn: RunTurn::SelfRun, claims: vec![claim], agent: AGENT.into(), grant, held: on_hold(organs.mind)? };
Ok(match open_and_launch(&ports, request)? {
LaunchOutcome::Launched { run } | LaunchOutcome::Busy { run } => Ok(run),
_ => Err(Failure::SelfRun),

2
crates/mind-body/src/persona/tests.rs
View file

@ -227,7 +227,7 @@ impl Rig {
let effective = read(&FileSource::new(self.store()), &FileSource::new(self.swarm_store()), self.mind.instance());
if let Effective::Released(grant) = effective {
let ports = Ports { mind: &self.mind, launcher: &Recording::default(), clock: &self.clock, host: "yggdrasil-host" };
let request = LaunchRequest { turn: RunTurn::PersonaTurn, claims: vec![], agent: "mind-heartbeat".into(), grant };
let request = LaunchRequest { turn: RunTurn::PersonaTurn, claims: vec![], agent: "mind-heartbeat".into(), grant, held: crate::queue::Held::default() };
assert!(matches!(open_and_launch(&ports, request).unwrap(), LaunchOutcome::Launched { .. }));
}
}

296
crates/mind-body/src/queue.rs
View file

@ -7,12 +7,17 @@
//! record (target invariant run-is-the-grant).
use anyhow::{Result, anyhow, bail};
use cultnet_rs::{Citation, FieldPredicate, Incoming, RecordRef, Selection};
use eureka_pipeline::{Line, PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome, RunOperator, RunTurn, Slug};
use cultnet_rs::{Incoming, Selection};
use eureka_pipeline::{
Line, PipelineDocument, PipelineKind, PipelineRef, ResolutionOutcome, RunOperator, RunTurn,
Slug,
};
use eureka_state::hold::{self, Hold};
use eureka_state::{ClientError, HuginnClient};
use huginn_mind::{
Faculty, HuginnMindRequest, HuginnMindResponse, PipelineAdmissionBatch, PipelineAdmissionOutcome, PipelineDocumentSummary, PipelineFacts,
PipelinePageItems, PipelineProvenance, PipelineSelectionPage, PipelineStatusSummary, SemanticQuery,
Faculty, HuginnMindRequest, HuginnMindResponse, PipelineAdmissionBatch,
PipelineAdmissionOutcome, PipelineDocumentSummary, PipelineFacts, PipelineProvenance,
PipelineSelectionPage, PipelineStatusSummary, SemanticQuery,
};
/// The last `CONSECUTIVE` ended runs over an item, all empty, trip its breaker.
@ -109,42 +114,14 @@ fn client_error_code(error: &ClientError) -> &'static str {
}
}
/// The documents that cite `target`, under `role` or any.
pub(crate) fn citing(target: &PipelineRef, role: Option<&str>) -> Citation {
Citation { target: RecordRef::new(target.kind.type_id(), target.id.0.clone()), role: role.map(Into::into) }
}
pub(crate) use eureka_state::hold::{any_of, citing, in_force, of_kind};
pub(crate) fn of_kind(kind: PipelineKind) -> Selection {
Selection { schemas: Some(vec![kind.type_id().to_string()]), ..Selection::default() }
}
pub(crate) fn any_of(mut selection: Selection, index: &str, values: &[&str]) -> Selection {
selection.fields.get_or_insert_with(Vec::new).push(FieldPredicate {
index: index.into(),
op: "any_of".into(),
values: Some(values.iter().map(|value| value.to_string()).collect()),
number: None,
});
selection
}
pub(crate) fn in_force(selection: Selection) -> Selection {
any_of(selection, "in_force", &["true"])
}
/// Every page of a selection, in the mind's order. The one paging loop.
pub(crate) fn headers(mind: &dyn MindPort, selection: &Selection) -> Result<Vec<PipelineDocumentSummary>> {
let mut all = Vec::new();
let mut asking = selection.clone();
loop {
let page = mind.query(&asking)?;
let PipelinePageItems::Headers(items) = page.items else { bail!("the mind answered documents to a header selection") };
all.extend(items);
match page.next {
Some(cursor) => asking.cursor = Some(cursor),
None => return Ok(all),
}
}
/// Every page of a selection, in the mind's order.
pub(crate) fn headers(
mind: &dyn MindPort,
selection: &Selection,
) -> Result<Vec<PipelineDocumentSummary>> {
hold::headers(&|asking| mind.query(asking), selection)
}
/// The claims of the runs of `instance` that are still in force and not in
@ -165,43 +142,64 @@ pub(crate) fn live_claims(mind: &dyn MindPort, instance: &Slug, ending: &[Pipeli
/// 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())
hold::claims_of(&|asking| mind.query(asking), run)
}
/// 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);
}
}
/// Every question holding anything and what it holds: `eureka_state::hold`,
/// the one decider, which `huginn blocked` reads too.
pub fn holds(mind: &dyn MindPort) -> Result<Vec<Hold>> {
hold::holds(&|asking| mind.query(asking))
}
/// A snapshot of the items an open question holds, read once by whoever
/// launches and passed to everything that launch consults, so the gate, the
/// queue and the claims they chose cannot answer differently within one launch.
#[derive(Clone, Debug, Default, PartialEq, Eq)]
pub struct Held(Vec<PipelineRef>);
impl Held {
pub fn contains(&self, item: &PipelineRef) -> bool {
self.0.contains(item)
}
Ok(held)
pub fn items(&self) -> &[PipelineRef] {
&self.0
}
}
/// The items an in-force question holds, read now: the one owner of the stop.
pub fn on_hold(mind: &dyn MindPort) -> Result<Held> {
Ok(Held(
holds(mind)?
.into_iter()
.flat_map(|hold| hold.items)
.collect(),
))
}
/// What may be worked, oldest first: the cut specs in force that no report
/// 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 blocked = on_hold(mind)?;
pub fn queue(
mind: &dyn MindPort,
instance: &Slug,
ending: &[PipelineRef],
held: &Held,
) -> Result<Vec<PipelineRef>> {
let unreported = Selection {
cited: Some(Incoming {
role: "cut_spec".into(),
exists: false,
}),
..in_force(of_kind(PipelineKind::CutSpec))
};
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())
Ok(headers(mind, &unreported)?
.into_iter()
.map(|header| header.id)
.filter(|spec| !held.contains(spec) && !claimed.contains(spec))
.collect())
}
/// Whether an item may still be launched on.
@ -282,6 +280,15 @@ mod tests {
use eureka_pipeline::{RunOperator, RunTurn};
use super::*;
/// `queue` against a hold read now.
fn queued(
mind: &dyn MindPort,
instance: &Slug,
ending: &[PipelineRef],
) -> Result<Vec<PipelineRef>> {
queue(mind, instance, ending, &on_hold(mind)?)
}
use crate::testkit::*;
/// A mind holding the spec `x` and a history of ended runs on it.
@ -408,8 +415,19 @@ mod tests {
assert_eq!(breaker(&mind, &item, &[]).unwrap(), Breaker::Clear, "open, it is not yet ended");
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");
assert!(
queued(&mind, &instance, &[])
.unwrap()
.iter()
.all(|spec| *spec != item),
"it claims the item"
);
assert!(
queued(&mind, &instance, &[run_ref("dying")])
.unwrap()
.contains(&item),
"an ending run claims nothing"
);
}
#[test]
@ -468,7 +486,32 @@ mod tests {
assert!(matches!(outcome, PipelineAdmissionOutcome::Committed { .. }), "{outcome:?}");
let facts = mind.view(PipelineKind::Run, &run_ref("p").id.0).unwrap().admission.provenance;
assert_eq!(
(facts.faculty, facts.agent.0.as_str(), facts.session.0.as_str(), facts.tool.0.as_str()),
queued(&port, port.instance(), &[]).unwrap(),
vec![spec_ref("x")]
);
let outcome = port
.admit(
"agent-x",
"session-y",
vec![run("p", RunTurn::PersonaTurn, RunOperator::Mind, &[])],
)
.unwrap();
assert!(
matches!(outcome, PipelineAdmissionOutcome::Committed { .. }),
"{outcome:?}"
);
let facts = mind
.view(PipelineKind::Run, &run_ref("p").id.0)
.unwrap()
.admission
.provenance;
assert_eq!(
(
facts.faculty,
facts.agent.0.as_str(),
facts.session.0.as_str(),
facts.tool.0.as_str()
),
(Faculty::SelfFaculty, "agent-x", "session-y", "mind-body")
);
let refused = port.query(&any_of(of_kind(PipelineKind::Ruling), "turn", &["canary-turn-5d2e"])).unwrap_err();
@ -523,19 +566,43 @@ mod tests {
let mind = TestMind::seeded();
mind.committed(vec![spec("a"), spec("b"), spec("c"), spec("d"), spec("e")]);
let instance = Slug(INSTANCE.into());
let all: Vec<PipelineRef> = ["a", "b", "c", "d", "e"].iter().map(|cut| spec_ref(cut)).collect();
assert_eq!(queue(&mind, &instance, &[]).unwrap(), all);
let all: Vec<PipelineRef> = ["a", "b", "c", "d", "e"]
.iter()
.map(|cut| spec_ref(cut))
.collect();
assert_eq!(queued(&mind, &instance, &[]).unwrap(), all);
mind.committed(vec![report("b")]);
mind.committed(vec![question_in("fork", &spec_ref("c"))]);
mind.committed(vec![run("held", RunTurn::SelfRun, RunOperator::Operator, &[spec_ref("d")])]);
assert_eq!(queue(&mind, &instance, &[]).unwrap(), vec![spec_ref("a"), spec_ref("e")]);
mind.committed(vec![run(
"held",
RunTurn::SelfRun,
RunOperator::Operator,
&[spec_ref("d")],
)]);
assert_eq!(
queued(&mind, &instance, &[]).unwrap(),
vec![spec_ref("a"), spec_ref("e")]
);
mind.committed(vec![close(run_ref("held"), recorded())]);
assert_eq!(queue(&mind, &instance, &[]).unwrap(), vec![spec_ref("a"), spec_ref("d"), spec_ref("e")], "a closed run claims nothing");
assert_eq!(
queued(&mind, &instance, &[]).unwrap(),
vec![spec_ref("a"), spec_ref("d"), spec_ref("e")],
"a closed run claims nothing"
);
let other = Slug("another-instance".into());
mind.committed(vec![run("again", RunTurn::SelfRun, RunOperator::Operator, &[spec_ref("a")])]);
assert_eq!(queue(&mind, &other, &[]).unwrap().len(), 3, "claims are read for the instance asked");
mind.committed(vec![run(
"again",
RunTurn::SelfRun,
RunOperator::Operator,
&[spec_ref("a")],
)]);
assert_eq!(
queued(&mind, &other, &[]).unwrap().len(),
3,
"claims are read for the instance asked"
);
}
#[test]
@ -563,26 +630,85 @@ mod tests {
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());
assert!(on_hold(&mind).unwrap().items().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");
assert!(
held.contains(&a) && held.contains(&b) && held.items().len() == 2,
"{held:?}: the claims of the resolved run"
);
assert_eq!(
queued(&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})");
assert!(
on_hold(&mind).unwrap().items().is_empty(),
"answered or withdrawn, nothing is held (withdrawn: {withdraw})"
);
}
}
#[test]
fn a_question_answered_retire_keeps_holding_the_claims_that_are_still_in_force() {
let (mind, a, b) = bundle();
mind.committed(vec![question_in("fork", &run_ref("f2"))]);
let question = id("question", "fork");
mind.committed(vec![ruling(
"answer",
Some((&question, eureka_state::hold::RETIRE)),
)]);
let held = on_hold(&mind).unwrap();
assert!(
held.contains(&a) && held.contains(&b) && held.items().len() == 2,
"{held:?}: retire holds what the answer retired"
);
let retired: Vec<bool> = holds(&mind)
.unwrap()
.iter()
.map(|hold| hold.retired)
.collect();
assert_eq!(retired, [true]);
mind.committed(vec![close(a.clone(), withdrawn())]);
assert_eq!(
on_hold(&mind).unwrap().items(),
[b.clone()],
"a resolved claim drops out"
);
mind.committed(vec![close(b, withdrawn())]);
assert!(
on_hold(&mind).unwrap().items().is_empty(),
"all resolved, nothing is held"
);
}
#[test]
fn a_question_answered_with_another_choice_holds_nothing() {
let (mind, _, _) = bundle();
mind.committed(vec![question_in("fork", &run_ref("f2"))]);
mind.committed(vec![ruling(
"answer",
Some((&id("question", "fork"), "reopen")),
)]);
assert!(on_hold(&mind).unwrap().items().is_empty());
assert!(holds(&mind).unwrap().is_empty());
}
#[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_eq!(
on_hold(&mind).unwrap().items(),
[a],
"the legacy shape: raised in the item, it holds the item and no claimant"
);
assert!(!on_hold(&mind).unwrap().contains(&b));
}
@ -619,8 +745,12 @@ mod tests {
fn an_unreachable_mind_names_no_input() {
let canary = "canary-instance-7f3a91";
let endpoint: SocketAddr = "127.0.0.1:9".parse().unwrap();
let port = HuginnMind::new(HuginnClient::new(endpoint, Slug(canary.into()), Duration::from_millis(300)));
let error = queue(&port, &Slug(canary.into()), &[]).unwrap_err();
let port = HuginnMind::new(HuginnClient::new(
endpoint,
Slug(canary.into()),
Duration::from_millis(300),
));
let error = queued(&port, &Slug(canary.into()), &[]).unwrap_err();
let shown = format!("{error:#}");
assert!(shown.contains("unavailable"), "{shown}");
assert!(!shown.contains(canary) && !shown.contains("127.0.0.1"), "{shown}");

14
crates/mind-body/src/self_run/inbox.rs
View file

@ -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, on_hold};
use crate::queue::{Held, 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";
@ -67,16 +67,16 @@ fn open_briefs(mind: &dyn MindPort) -> Result<Vec<Brief>> {
/// 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.
/// answered or withdrawn. Reads the hold once.
pub fn derive(mind: &dyn MindPort, told: &HashSet<String>) -> Result<Vec<Brief>> {
Ok(derive_counting_held(mind, told)?.0)
Ok(derive_counting_held(mind, &on_hold(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)?;
/// `derive` against a hold the caller already read, and how many of her briefs
/// in force that hold keeps back.
pub fn derive_counting_held(mind: &dyn MindPort, holds: &Held, told: &HashSet<String>) -> Result<(Vec<Brief>, usize)> {
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));
open_briefs(mind)?.into_iter().partition(|brief| holds.items().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()))
}

9
crates/mind-body/src/testkit.rs
View file

@ -56,7 +56,12 @@ impl Recording {
}
pub(crate) fn revive(&self, turn: RunTurn) {
self.running.borrow_mut().push((Slug(INSTANCE.into()), turn));
self.revive_as(&Slug(INSTANCE.into()), turn);
}
/// `revive` for a named instance.
pub(crate) fn revive_as(&self, instance: &Slug, turn: RunTurn) {
self.running.borrow_mut().push((instance.clone(), turn));
}
}
@ -366,7 +371,7 @@ pub(crate) fn question_in(label: &str, item: &PipelineRef) -> PipelineDocument {
label: label.into(),
title: "A fork".into(),
question: "Which?".into(),
options: ["A", "B"].iter().map(|name| QuestionOption { label: (*name).into(), text: "an option".into() }).collect(),
options: ["A", "B", "reopen", "retire"].iter().map(|name| QuestionOption { label: (*name).into(), text: "an option".into() }).collect(),
recommended: "A".into(),
depends: vec![],
raised_in: Some(item.clone()),