diff --git a/crates/huginn-mind/src/admission.rs b/crates/huginn-mind/src/admission.rs index 148568b..d6c965a 100644 --- a/crates/huginn-mind/src/admission.rs +++ b/crates/huginn-mind/src/admission.rs @@ -522,19 +522,21 @@ fn check(docs: &Docs, staged: &Staged, mind: &Slug) -> Result<(), MindRefusal> { // be in force, and no other run in force may hold it. A run already in // the image under this key, byte for byte, is a replay and owes its // claims no fresh freedom, so the replay still answers AlreadyAdmitted - // after the work it claimed has moved on. + // after the work it claimed has moved on. One live run of hers + // per instance and turn (ruling one-live-self-run) is decided by the same + // rule. D::Run(run) => { let replay = docs.in_image(K::Run, key) == Some(&staged.document); - run_claims_rule(docs, key, run, replay) + run_rule(docs, key, run, replay) } D::FollowUp(_) | D::Instance(_) => Ok(()), } } -/// The one rule for a run holding claims, whether it is opened or reinstated -/// by withdrawing its close: every claim in force (a replay of an admitted run -/// owes none) and held by no other live run. -fn run_claims_rule(docs: &Docs, run_key: &str, run: &PipelineRun, replay: bool) -> Result<(), MindRefusal> { +/// The one rule for a run opening or reinstated by withdrawing its close: every +/// claim in force (a replay of an admitted run owes none) and held by no other +/// live run, then at most one live run of hers per instance and turn. +fn run_rule(docs: &Docs, run_key: &str, run: &PipelineRun, replay: bool) -> Result<(), MindRefusal> { for claim in &run.claims { if !replay && !docs.in_force(claim.kind, &claim.id.0) { return Err(MindRefusal::CitesResolvedDocument { kind: claim.kind, id: claim.id.0.clone() }); @@ -543,6 +545,9 @@ fn run_claims_rule(docs: &Docs, run_key: &str, run: &PipelineRun, replay: bool) return Err(MindRefusal::AlreadyClaimed { item: claim.id.0.clone(), run: holder.into() }); } } + if !replay && let Some(holder) = docs.live_holder(run_key, run) { + return Err(MindRefusal::AlreadyLive { run: holder.into() }); + } Ok(()) } @@ -683,14 +688,15 @@ fn resolution_rule(docs: &Docs, resolution: &PipelineResolution) -> Result<(), M }); } // Withdrawing a run's closure puts the run back in force, and a run in - // force holds its claims: refuse while another run in force holds one. + // force holds its claims and its turn's slot: refuse while another run in force + // holds one of the claims or another run of hers of its turn is live. if matches!(resolution.outcome, ResolutionOutcome::Withdrawn { .. }) && subject_kind == K::Resolution && let Some(D::Resolution(reinstating)) = docs.find(K::Resolution, subject_id) && reinstating.subject.kind == K::Run && let Some(D::Run(run)) = docs.find(K::Run, &reinstating.subject.id.0) { - run_claims_rule(docs, &reinstating.subject.id.0, run, false)?; + run_rule(docs, &reinstating.subject.id.0, run, false)?; } for (_, referent) in outcome_citations(&resolution.outcome) { if !docs.in_force(referent.kind, &referent.id.0) { @@ -2003,6 +2009,94 @@ mod tests { assert_eq!(refusal(admit(&mut reopened, vec![run("d", &[spec_ref("1")])])), claimed); } + /// one-live-self-run at admission: a second run of hers of one turn is + /// AlreadyLive naming the holder, whether the holder is claimless or + /// claiming, and a claim the holder holds is still AlreadyClaimed first. + #[test] + fn a_second_live_run_of_hers_of_one_turn_is_already_live() { + let mut mind = with_specs(); + committed(admit(&mut mind, vec![mind_run("a", RunTurn::SelfRun, &[])])); + let live_a = MindRefusal::AlreadyLive { run: run_key("a") }; + assert_eq!(refusal(admit(&mut mind, vec![mind_run("b", RunTurn::SelfRun, &[])])), live_a, "a claimless holder"); + assert_eq!(refusal(admit(&mut mind, vec![mind_run("b", RunTurn::SelfRun, &[spec_ref("1")])])), live_a); + committed(admit(&mut mind, vec![recorded("a")])); + committed(admit(&mut mind, vec![mind_run("b", RunTurn::SelfRun, &[spec_ref("1")])])); + let live_b = MindRefusal::AlreadyLive { run: run_key("b") }; + assert_eq!(refusal(admit(&mut mind, vec![mind_run("c", RunTurn::SelfRun, &[spec_ref("2")])])), live_b, "a claiming holder"); + let claimed = MindRefusal::AlreadyClaimed { item: spec_ref("1").id.0, run: run_key("b") }; + assert_eq!(refusal(admit(&mut mind, vec![mind_run("c", RunTurn::SelfRun, &[spec_ref("1")])])), claimed, "claims answer first"); + } + + /// The rule counts only in-force runs of hers of the same turn: an + /// operator's run, a run of the other turn and a run after the holder is + /// Recorded are admitted, and one live PersonaTurn and one live SelfRun of + /// hers coexist. + #[test] + fn an_operators_run_another_turn_and_a_run_after_the_holder_closed_are_admitted() { + let mut mind = with_specs(); + committed(admit(&mut mind, vec![mind_run("a", RunTurn::SelfRun, &[])])); + committed(admit(&mut mind, vec![run("op1", &[])])); + committed(admit(&mut mind, vec![run("op2", &[])])); + committed(admit(&mut mind, vec![mind_run("p", RunTurn::PersonaTurn, &[])])); + let live_p = MindRefusal::AlreadyLive { run: run_key("p") }; + assert_eq!(refusal(admit(&mut mind, vec![mind_run("p2", RunTurn::PersonaTurn, &[])])), live_p); + committed(admit(&mut mind, vec![recorded("a")])); + committed(admit(&mut mind, vec![mind_run("b", RunTurn::SelfRun, &[])])); + committed(admit(&mut mind, vec![recorded("b")])); + committed(admit(&mut mind, vec![mind_run("c", RunTurn::SelfRun, &[])])); + + let mut operator_only = with_specs(); + committed(admit(&mut operator_only, vec![run("op", &[])])); + committed(admit(&mut operator_only, vec![mind_run("a", RunTurn::SelfRun, &[])])); + } + + /// Two runs of hers of one turn in one batch hold nothing: the batch is + /// checked against itself, refused whole, and nothing is written. + #[test] + fn two_live_runs_of_hers_in_one_batch_are_already_live_and_nothing_commits() { + let mut mind = with_specs(); + let refused = refusal(admit(&mut mind, vec![mind_run("a", RunTurn::SelfRun, &[]), mind_run("b", RunTurn::SelfRun, &[])])); + assert!(matches!(refused, MindRefusal::AlreadyLive { .. }), "{refused:?}"); + assert!(mind.envelope(K::Run, &run_key("a")).is_none() && mind.envelope(K::Run, &run_key("b")).is_none()); + } + + /// Withdrawing the close of a run of hers reinstates it, and a reinstated + /// run is a live one: refused while another run of its turn is live, + /// allowed when none is. + #[test] + fn reinstating_a_run_of_hers_over_a_live_holder_is_already_live() { + let closure = r(K::Resolution, &format!("{INSTANCE}:resolution:run.a.n1")); + let mut mind = with_specs(); + committed(admit(&mut mind, vec![mind_run("a", RunTurn::SelfRun, &[])])); + committed(admit(&mut mind, vec![recorded("a")])); + committed(admit(&mut mind, vec![mind_run("b", RunTurn::SelfRun, &[])])); + assert_eq!( + refusal(admit(&mut mind, vec![resolution(closure.clone(), withdrawn())])), + MindRefusal::AlreadyLive { run: run_key("b") } + ); + committed(admit(&mut mind, vec![recorded("b")])); + committed(admit(&mut mind, vec![resolution(closure, withdrawn())])); + assert_eq!( + refusal(admit(&mut mind, vec![mind_run("c", RunTurn::SelfRun, &[])])), + MindRefusal::AlreadyLive { run: run_key("a") }, + "the reinstated run holds the slot" + ); + } + + /// A retry of an admitted run answers AlreadyAdmitted even when another + /// run of its turn has since become the live one. + #[test] + fn replaying_an_admitted_run_of_hers_after_another_took_the_turn_is_already_admitted() { + let mut mind = with_specs(); + let (first, _) = committed(admit(&mut mind, vec![mind_run("a", RunTurn::SelfRun, &[])])); + committed(admit(&mut mind, vec![recorded("a")])); + committed(admit(&mut mind, vec![mind_run("b", RunTurn::SelfRun, &[])])); + assert_eq!( + admit(&mut mind, vec![mind_run("a", RunTurn::SelfRun, &[])]), + PipelineAdmissionOutcome::AlreadyAdmitted { receipt_id: first } + ); + } + /// Two openers on one claim in one batch: the claim is held by neither, so /// the batch is refused whole and nothing is written. One opener at a time /// is `a_second_run_on_a_live_claim_is_already_claimed`. diff --git a/crates/huginn-mind/src/docs.rs b/crates/huginn-mind/src/docs.rs index a3bae9a..2427338 100644 --- a/crates/huginn-mind/src/docs.rs +++ b/crates/huginn-mind/src/docs.rs @@ -7,8 +7,8 @@ use cultcache_rs::CultCacheEnvelope; use eureka_pipeline::{ - OrgRepo, PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, PipelineStewardship, ResolutionOutcome, - Short, Slug, + OrgRepo, PipelineDocument, PipelineKind, PipelineRef, PipelineResolution, PipelineRun, PipelineStewardship, ResolutionOutcome, + RunOperator, Short, Slug, }; use schemars::JsonSchema; use serde::{Deserialize, Serialize}; @@ -229,6 +229,29 @@ impl Docs { }) } + /// The in-force run of hers, other than `run_key`, that holds the slot of + /// the turn `run` is of: ruling one-live-self-run's one live run per + /// instance and turn, asked by a run's opening and by the withdrawal that + /// would put a closed run back in force. An operator's run neither holds + /// the slot nor is refused for it. + pub(crate) fn live_holder(&self, run_key: &str, run: &PipelineRun) -> Option<&str> { + if run.operated_by == RunOperator::Operator { + return None; + } + self.of_kind(PipelineKind::Run).find_map(|(other_key, document)| match document { + PipelineDocument::Run(other) + if other_key != run_key + && other.instance == run.instance + && other.turn == run.turn + && other.operated_by == RunOperator::Mind + && self.in_force(PipelineKind::Run, other_key) => + { + Some(other_key) + } + _ => None, + }) + } + /// The stewardship of `(mind, repo)` a hand-off acts on: the one whose /// withdrawal already carries `hand_off_key`, when image or batch holds /// that withdrawal, so a replay derives the same withdrawal again and A9 diff --git a/crates/huginn-mind/src/fixtures.rs b/crates/huginn-mind/src/fixtures.rs index 7dde8ca..100f32c 100644 --- a/crates/huginn-mind/src/fixtures.rs +++ b/crates/huginn-mind/src/fixtures.rs @@ -327,13 +327,23 @@ pub(crate) fn hand_off(from: &str, to: &str, repo_name: &str, documents: &[&str] }) } -/// A Self run of this mind that claims `claims`. +/// An operator's Self run on this mind that claims `claims`, which the one-live +/// rule never counts. pub(crate) fn run(label: &str, claims: &[PipelineRef]) -> PipelineDocument { + run_of(label, RunOperator::Operator, RunTurn::SelfRun, claims) +} + +/// A run of hers (operated_by Mind) of `turn` that claims `claims`. +pub(crate) fn mind_run(label: &str, turn: RunTurn, claims: &[PipelineRef]) -> PipelineDocument { + run_of(label, RunOperator::Mind, turn, claims) +} + +fn run_of(label: &str, operated_by: RunOperator, turn: RunTurn, claims: &[PipelineRef]) -> PipelineDocument { PipelineDocument::Run(PipelineRun { instance: slug(INSTANCE), label: l(label), - operated_by: RunOperator::Mind, - turn: RunTurn::SelfRun, + operated_by, + turn, host: s("yggdrasil"), started_on: date(), budget_usd: s("5"), diff --git a/crates/huginn-mind/src/query.rs b/crates/huginn-mind/src/query.rs index c74bedb..8debedb 100644 --- a/crates/huginn-mind/src/query.rs +++ b/crates/huginn-mind/src/query.rs @@ -1293,7 +1293,7 @@ mod tests { let facts = headers(&page).iter().map(|header| header.facts.clone()).collect::>(); assert!(facts.iter().any(|fact| matches!( fact, - PipelineFacts::Run { label, turn: RunTurn::SelfRun, operated_by: RunOperator::Mind, claims, .. } + PipelineFacts::Run { label, turn: RunTurn::SelfRun, operated_by: RunOperator::Operator, claims, .. } if label.0 == "a" && claims.len() == 1 )), "{facts:?}"); } diff --git a/crates/huginn-mind/src/refusal.rs b/crates/huginn-mind/src/refusal.rs index c771744..192558d 100644 --- a/crates/huginn-mind/src/refusal.rs +++ b/crates/huginn-mind/src/refusal.rs @@ -58,6 +58,10 @@ pub enum MindRefusal { /// A run claims work another run in force already holds: `item` is the /// claimed document's id and `run` the id of the run that holds it. AlreadyClaimed { item: String, run: String }, + /// A run of hers opens, or is reinstated, while another in-force run of + /// hers of the same instance and turn is live: `run` is the holder's id + /// (ruling one-live-self-run). + AlreadyLive { run: String }, /// The one refusal a mind never raises: the answer exceeds the largest /// body the organ will deliver by any plane, one send or a deferred body. /// It lives here because a refusal rides the response schema and there is diff --git a/schemas/cultnet/huginn.mind_response.v1.schema.json b/schemas/cultnet/huginn.mind_response.v1.schema.json index 4102b55..2dbe66b 100644 --- a/schemas/cultnet/huginn.mind_response.v1.schema.json +++ b/schemas/cultnet/huginn.mind_response.v1.schema.json @@ -1375,6 +1375,27 @@ "AlreadyClaimed" ] }, + { + "description": "A run of hers opens, or is reinstated, while another in-force run of\nhers of the same instance and turn is live: `run` is the holder's id\n(ruling one-live-self-run).", + "type": "object", + "properties": { + "AlreadyLive": { + "type": "object", + "properties": { + "run": { + "type": "string" + } + }, + "required": [ + "run" + ] + } + }, + "additionalProperties": false, + "required": [ + "AlreadyLive" + ] + }, { "description": "The one refusal a mind never raises: the answer exceeds the largest\nbody the organ will deliver by any plane, one send or a deferred body.\nIt lives here because a refusal rides the response schema and there is\none of those, so a client reads it the way it reads every other refusal\nrather than parsing a second vocabulary. `bytes` is the encoded answer\n(the payload, not its envelope) and `limit` is the largest body the\ndaemon defers, both so a caller can narrow its own request; nothing is\ntruncated or paginated on its behalf. The daemon owns it and its\ndeferral bound decides it; no rule of the mind is involved.", "type": "object",