diff --git a/crates/mind-body/src/self_run.rs b/crates/mind-body/src/self_run.rs index e0327ac..13a3698 100644 --- a/crates/mind-body/src/self_run.rs +++ b/crates/mind-body/src/self_run.rs @@ -110,6 +110,43 @@ pub fn start_message(instance: &Slug, run: &PipelineRef, plan_quota: &[PlanQuota ) } +/// The task section of the start message: one section per claim, in claim +/// order. A claimed follow-up gives its id, campaign and `item` verbatim, +/// because the brief lives in the claim and a run that cannot reach its mind +/// would never see its job; a claimed cut spec is named as the task, to be read +/// with `huginn view`; a claim the mind does not hold, or of another kind, is +/// named by id. The claims are bounded by their types, so nothing is cut. A mind +/// that cannot answer is an error: the run is not started on a guessed task. +fn task_section(mind: &dyn MindPort, claims: &[PipelineRef]) -> Result { + let mut task = String::new(); + for claim in claims { + let one = Selection { keys: Some(vec![claim.id.0.clone()]), projection: "document".into(), ..Selection::default() }; + let page = mind.query(&one).with_context(|| format!("the claim `{}` could not be read from the mind", claim.id.0))?; + let PipelinePageItems::Documents(views) = page.items else { bail!("the mind answered headers to a document selection") }; + let id = &claim.id.0; + match views.into_iter().next().map(|view| view.document) { + Some(PipelineDocument::FollowUp(follow_up)) => { + task.push_str(&format!(" + +Task, claimed follow-up `{id}` (campaign `{}`): +{}", follow_up.campaign.0, follow_up.item.0)); + } + Some(PipelineDocument::CutSpec(spec)) => { + task.push_str(&format!(" + +Task, claimed cut spec `{id}` (campaign `{}`): the spec is the task; read it with `huginn view`.", spec.campaign.0)); + } + Some(_) => task.push_str(&format!(" + +The run claims `{id}` ({:?}), which is no brief: its task is unknown.", claim.kind)), + None => task.push_str(&format!(" + +The run claims `{id}`, which the mind does not hold: its task is unknown.")), + } + } + Ok(task) +} + /// What the child is told once, at the context ceiling. pub const FLUSH_MESSAGE: &str = "The context ceiling is reached. Admit to the mind what it lacks, close your run, and end."; @@ -286,13 +323,18 @@ impl fmt::Display for SelfRunOutcome { } } -/// Her one open Self run, read through the launch primitive holder reader. -/// Anything but exactly one run is none: the mind admits at most one Self run of -/// hers in force, and the launch that opens the next closes a dead one. -fn her_run(mind: &dyn MindPort) -> Result> { +/// Her one open Self run and its claims, read through the launch primitive holder +/// reader. Anything but exactly one run is none: the mind admits at most one Self +/// run of hers in force, and the launch that opens the next closes a dead one. +fn her_run(mind: &dyn MindPort) -> Result)>> { let mut held = holders(mind, mind.instance(), RunTurn::SelfRun)?; let (Some(run), None) = (held.pop(), held.pop()) else { return Ok(None) }; - Ok(Some(run.id)) + let one = Selection { keys: Some(vec![run.id.id.0.clone()]), projection: "document".into(), ..Selection::default() }; + let PipelinePageItems::Documents(views) = mind.query(&one)?.items else { bail!("the mind answered headers to a document selection") }; + match views.into_iter().next().map(|view| view.document) { + Some(PipelineDocument::Run(document)) => Ok(Some((run.id, document.claims))), + _ => Err(anyhow!("the run's document could not be read")), + } } /// What the stream said at the end of the run, as the reason of its closure. @@ -375,9 +417,10 @@ fn withdraw_misconfigured(mind: &dyn MindPort, clock: &dyn Clock, run: &Pipeline /// original error) stands, a failed close of a run that ended normally is an /// error, and the launch that opens the next run closes the run left open. pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write) -> Result { - let Some(run) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) }; + let Some((run, claims)) = her_run(mind)? else { return Ok(SelfRunOutcome::NoRun) }; + let task = task_section(mind, &claims)?; let mut heard = Heard::default(); - let ran = lend_and_drive(mind, clock, lender, spawner, run_cap_s, out, &run, &mut heard); + let ran = lend_and_drive(mind, clock, lender, spawner, run_cap_s, out, (&run, &task), &mut heard); let closed = close_if_open(mind, clock, &run, &ran, &heard); if closed.is_err() { let _ = writeln!(out, "the run could not be closed: the next launch closes it"); @@ -390,7 +433,7 @@ pub fn carry(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawne } #[allow(clippy::too_many_arguments, reason = "carry's ports, passed on whole")] -fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write, run: &PipelineRef, heard: &mut Heard) -> Result { +fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, spawner: &dyn Spawner, run_cap_s: u64, out: &mut dyn Write, (run, task): (&PipelineRef, &str), heard: &mut Heard) -> Result { let started_ms = unix_ms(clock.now()); let request = LeaseRequest { expires_at_unix_ms: started_ms.saturating_add(run_cap_s.saturating_mul(1000)) }; let lease = match lender.lend(&request)? { @@ -402,7 +445,7 @@ fn lend_and_drive(mind: &dyn MindPort, clock: &dyn Clock, lender: &dyn Lender, s Lent::Refused(refused) => return close_refused(mind, clock, run, refused.kind), }; let child = spawner.spawn(&claude_args(session_id(run)), lease.token())?; - drive(mind, clock, child, run, (lease.plan_quota(), started_ms), heard, out) + drive(mind, clock, child, (run, task), (lease.plan_quota(), started_ms), heard, out) } /// The most of a result's text that reaches the journal. @@ -440,9 +483,9 @@ fn write_result(line: &str, out: &mut dyn Write) -> std::io::Result<()> { } /// The loop: facts in, one clock read per fact, actions out. -fn drive(mind: &dyn MindPort, clock: &dyn Clock, mut child: Box, run: &PipelineRef, (plan_quota, now_ms): (&[PlanQuota], u64), heard: &mut Heard, out: &mut dyn Write) -> Result { +fn drive(mind: &dyn MindPort, clock: &dyn Clock, mut child: Box, (run, task): (&PipelineRef, &str), (plan_quota, now_ms): (&[PlanQuota], u64), heard: &mut Heard, out: &mut dyn Write) -> Result { let user_line = |text: &str| json!({"type": "user", "message": {"role": "user", "content": text}}).to_string(); - child.write_line(&user_line(&start_message(mind.instance(), run, plan_quota, now_ms))); + child.write_line(&user_line(&format!("{}{task}", start_message(mind.instance(), run, plan_quota, now_ms)))); let mut state = State::default(); let mut now = clock.now(); diff --git a/crates/mind-body/src/self_run/tests.rs b/crates/mind-body/src/self_run/tests.rs index 9390888..291a818 100644 --- a/crates/mind-body/src/self_run/tests.rs +++ b/crates/mind-body/src/self_run/tests.rs @@ -268,12 +268,14 @@ struct Seen<'a> { instance: Slug, doubled: bool, admit_fails: bool, + /// A query for this key fails. + query_fails_on: Option, admits: Cell, } impl<'a> Seen<'a> { fn of(inner: &'a TestMind) -> Self { - Self { inner, instance: inner.instance().clone(), doubled: false, admit_fails: false, admits: Cell::new(0) } + Self { inner, instance: inner.instance().clone(), doubled: false, admit_fails: false, query_fails_on: None, admits: Cell::new(0) } } } @@ -283,6 +285,11 @@ impl MindPort for Seen<'_> { } fn query(&self, selection: &Selection) -> Result { + if let (Some(key), Some(keys)) = (&self.query_fails_on, &selection.keys) + && keys.contains(key) + { + return Err(anyhow!("mind-down-canary")); + } let mut page = self.inner.query(selection)?; if self.doubled && let PipelinePageItems::Headers(items) = &mut page.items @@ -1340,3 +1347,94 @@ result subtype=success is_error=true: Failed to authenticate: OAuth session expi mod lease; mod closure; + +// ---- the task in the start message --------------------------------------- + +fn follow_up_saying(label: &str, item: &str) -> (PipelineDocument, PipelineRef) { + let PipelineDocument::FollowUp(mut document) = follow_up(label, &spec_ref("x")) else { unreachable!() }; + document.item = item.into(); + (PipelineDocument::FollowUp(document), reference(K::FollowUp, &id("follow_up", label))) +} + +fn claiming(claims: &[PipelineRef], documents: Vec) -> TestMind { + let mind = TestMind::seeded(); + mind.committed(vec![spec("x")]); + if !documents.is_empty() { + mind.committed(documents); + } + let PipelineDocument::Run(mut document) = self_run("selfrun-a", "5") else { unreachable!() }; + document.claims = claims.to_vec(); + mind.committed(vec![PipelineDocument::Run(document)]); + mind +} + +#[test] +fn the_start_message_carries_the_claimed_follow_ups_item_and_id_verbatim() { + let (brief, claim) = follow_up_saying("brief-a", "Fix the thing.\nThen report \u{e9}."); + let mind = claiming(std::slice::from_ref(&claim), vec![brief]); + let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false); + let first = &ran.written()[0]; + assert!(first.starts_with(&start_message(&Slug("yggdrasil".into()), &run_ref("selfrun-a"))), "{first}"); + for needle in ["Fix the thing.\nThen report \u{e9}.", claim.id.0.as_str(), "Use the eureka skill."] { + assert!(first.contains(needle), "{needle}: {first}"); + } + assert!(first.contains(&format!("(campaign `{}`)", follow_up_campaign())), "{first}"); +} + +fn follow_up_campaign() -> String { + let PipelineDocument::FollowUp(document) = follow_up("c", &spec_ref("x")) else { unreachable!() }; + document.campaign.0 +} + +#[test] +fn a_claim_that_is_no_brief_is_named_not_dropped() { + // A claimed cut spec is the task: named with its id, to be read with huginn view. + let mind = claiming(&[spec_ref("x")], vec![]); + let ran = carry_with(&Seen::of(&mind), vec![Out(result_line())], Some(0), false); + let first = &ran.written()[0]; + assert!(first.contains(&format!("claimed cut spec `{}`", spec_ref("x").id.0)) && first.contains("huginn view"), "{first}"); + assert!(!first.contains("unknown"), "{first}"); + // A claim the mind does not hold at all. + let absent = reference(K::FollowUp, &id("follow_up", "absent")); + let task = task_section(&mind, std::slice::from_ref(&absent)).unwrap(); + assert!(task.contains(&absent.id.0) && task.contains("does not hold"), "{task}"); + // A document of another kind is named by id and kind. + let other = run_ref("selfrun-a"); + let task = task_section(&mind, std::slice::from_ref(&other)).unwrap(); + assert!(task.contains(&other.id.0) && task.contains("no brief"), "{task}"); +} + +#[test] +fn every_claim_of_sixty_four_full_items_is_named_by_id() { + let big = "\u{e9}".repeat(500); + let docs: Vec<_> = (0..64).map(|n| follow_up_saying(&format!("big-{n}"), &big)).collect(); + let claims: Vec<_> = docs.iter().map(|(_, claim)| claim.clone()).collect(); + let mind = claiming(&claims, docs.into_iter().map(|(document, _)| document).collect()); + let task = task_section(&mind, &claims).unwrap(); + for claim in &claims { + assert!(task.contains(&claim.id.0), "{}", claim.id.0); + } + assert_eq!(task.matches(&big).count(), 64, "every item verbatim"); + assert!(!task.contains("truncated")); + // Claim order is kept. + assert!(task.find(&claims[0].id.0) < task.find(&claims[63].id.0)); +} + +#[test] +fn a_mind_query_failure_on_a_claim_fails_the_run_with_the_cause_and_starts_no_child() { + let (brief, claim) = follow_up_saying("brief-a", "Fix the thing."); + let mind = claiming(std::slice::from_ref(&claim), vec![brief]); + let mut seen = Seen::of(&mind); + seen.query_fails_on = Some(claim.id.0.clone()); + let ran = carry_with(&seen, vec![Out(result_line())], Some(0), false); + let error = format!("{:#}", ran.outcome.as_ref().unwrap_err()); + assert!(error.contains("mind-down-canary") && error.contains(&claim.id.0), "{error}"); + assert!(ran.spawns.is_empty() && ran.events.is_empty() && !ran.all.iter().any(|event| matches!(event, Event::Lent)), "no lease, no child"); + assert!(!ran.out.contains("unknown")); +} + +#[test] +fn a_run_without_claims_gets_the_identity_lines_alone() { + let ran = go(vec![Out(result_line())]); + assert_eq!(ran.written()[0], start_message(&Slug("yggdrasil".into()), &run_ref("selfrun-a"))); +}