From 7cfd32b464cc3a384f0c44e5092f81cd0f7391c7 Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Sat, 14 Nov 2026 22:38:39 +0100 Subject: [PATCH 1/3] feat: let a task wait on an outside event, and hold a task whose effect is not settled A session does not wait for an outside event. The holder gives the task up with `wait_for`, and the task is blocked until a tap or a person reports the event. An agent cannot report an event. A holder records an external action before it makes the action. While an action is not settled, no session can claim the task, also after the lease expires, and the holder cannot complete the task. The holder settles its action while its claim is in force. After that, only a person or a tap settles it. A task also has the fields `parent`, `scores`, `implements`, `reversibility`, `artifacts_touched` and `policy_context`. The service sets `policy_context`, and it refuses a `parent` that the session cannot read. Both ledger backends project the new events through one function. The memory server has the tools `wait_for`, `effect_begun` and `effect_settled`. The command line has `amem task wait`, `arrived`, `effect-begin` and `effect-settle`. Act 4 of the demo shows both mechanisms. Each new guard has a test that fails without the guard: the check in `claimable`, the completion with an action that is not settled, the refusal of an agent that reports an event, the settlement by a second agent, the `policy_context` stamp and the `parent` check. Co-Authored-By: Claude Fable 5.2 Claude-Session: https://claude.ai/code/session_01D5DyJRpu5ned5y3d92U7Ts --- PLAN.md | 2 +- README.md | 9 +- crates/ledger/src/lib.rs | 574 +++++++++++++++++++++++++++++++++- crates/ledger/src/postgres.rs | 39 +++ crates/ledger/src/sqlite.rs | 29 ++ crates/memory/src/main.rs | 116 +++++++ crates/server/src/holds.rs | 293 +++++++++++++++++ crates/server/src/lib.rs | 27 ++ crates/server/src/mcp.rs | 95 ++++++ docs/demo-narrative.md | 19 +- docs/terms.md | 3 + local/README.md | 38 ++- scripts/demo/4-fabric.sh | 30 ++ 13 files changed, 1248 insertions(+), 26 deletions(-) create mode 100644 crates/server/src/holds.rs diff --git a/PLAN.md b/PLAN.md index 608d78fa..fd27f367 100644 --- a/PLAN.md +++ b/PLAN.md @@ -68,7 +68,7 @@ Landed so far: |---|---|---| | 0 | #69 | Nothing | | 1 | #71 | Nothing | -| 2 | #73, and attempt keys for a claim and a completion in the pull request after #74 | An attempt key on the other operations; an uncertain external effect blocking reassignment; waiting on an outside event as a state; the record's remaining fields (`parent`, `scores`, `implements`, `reversibility`, `artifacts_touched`, `policy_context`) | +| 2 | #73, attempt keys for a claim and a completion in #77, and waiting, unsettled effects and the remaining fields in this pull request | An attempt key on the other operations; a connector that serialises external actions and reconciles them; taps that report the events tasks wait on; rules that read `scores`, `implements`, `reversibility` and `artifacts_touched` | | 3 | the pull request after #73 | A suite that gates a version of a definition; the data scope enforced; a definition required by default, once the system's own filings name one; tools offered a few at boot and the rest by search | What stays unbuilt after these, and stays on README's list: federation across domains, authentication of people, logprob-based drift, and accelerator time, which only a gateway can count. So do the items on that list that need another machine or an outside service more than they need code: the limits on re-enactment, harness settings and decomposition templates in the platform domain, bytes stored and resource limits on the sandbox, the authoring of agent definitions, the search tool against its live provider, a harness under Seatbelt on a Mac, and the Gofannon bridge against a live deployment, whose agents make model calls that the egress does not meter. diff --git a/README.md b/README.md index f704619b..98b69a61 100644 --- a/README.md +++ b/README.md @@ -110,10 +110,11 @@ two weeks. It is also the reason that none of it is the production shape. - CPU or memory limits on the sandbox - **The task record, in part.** A claim has a generation, and the ledger refuses a holder that gives an earlier one. A task can be `blocked` on a task that it depends on. - A claim and a completion can carry an attempt key. These parts are not built: an - attempt key for the other operations, a block on a new claim - while an external effect is uncertain, and a `blocked` state for a task that waits on an - outside event. The generation does not prove who the caller is. + A claim and a completion can carry an attempt key. A holder can give a task up until an + outside event comes, and a task with an external action that is not settled goes to no + other session. These parts are not built: an attempt key for the other operations, a + connector that makes external actions in sequence and settles them, and taps that + report the events. The generation does not prove who the caller is. - **Task definitions, in part, and autonomy levels.** A task can name a task definition. It then takes the checks of the definition, and its holder can call only the tools that the definition declares. Triage requires a definition only when the grants say so, diff --git a/crates/ledger/src/lib.rs b/crates/ledger/src/lib.rs index e5c635d9..3f0f1c25 100644 --- a/crates/ledger/src/lib.rs +++ b/crates/ledger/src/lib.rs @@ -198,6 +198,85 @@ pub struct TaskSpec { /// The checks that say this task is done, its own or those its definition gave it. #[serde(default, skip_serializing_if = "Vec::is_empty")] pub acceptance: Vec, + /// The task this one was split from. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub parent: Option, + /// What triage scored it, beside the one number the queue is ordered by. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub scores: Option, + /// The decisions this task carries out: entries of the record. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub implements: Vec, + /// Whether what the task does can be undone. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub reversibility: Option, + /// The owned paths the task changes, as globs. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub artifacts_touched: Vec, + /// The policy in force when the task was filed. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub policy_context: Vec, +} + +/// What triage scored a task: the parts its rank is made of. +#[derive(Clone, Debug, Default, PartialEq, Eq, Serialize, Deserialize)] +pub struct Scores { + /// What it is worth done. + #[serde(default)] + pub value: i64, + /// How soon it loses that worth. + #[serde(default)] + pub urgency: i64, + /// What it costs to do. + #[serde(default)] + pub effort: i64, + /// The tasks that wait on it. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub unblocks: Vec, +} + +/// Whether what a task does can be undone. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "lowercase")] +pub enum Reversibility { + /// It can. + Reversible, + /// It cannot. + Irreversible, +} + +impl std::str::FromStr for Reversibility { + type Err = String; + + fn from_str(s: &str) -> Result { + match s { + "reversible" => Ok(Reversibility::Reversible), + "irreversible" => Ok(Reversibility::Irreversible), + other => Err(format!("{other:?} is not reversible or irreversible")), + } + } +} + +impl std::str::FromStr for Outcome { + type Err = String; + + fn from_str(s: &str) -> Result { + match s { + "applied" => Ok(Outcome::Applied), + "not_applied" => Ok(Outcome::NotApplied), + other => Err(format!("{other:?} is not applied or not_applied")), + } + } +} + +/// How an external effect turned out, as whoever reconciled it found the destination. +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize, Deserialize)] +#[serde(rename_all = "snake_case")] +pub enum Outcome { + /// The destination holds the effect. + Applied, + /// The destination does not hold it, so the action may be made again. + NotApplied, } /// The fields of a task that the ledger keeps and never queries by, as one JSON column. @@ -207,6 +286,64 @@ struct Extra { definition: Option, #[serde(default, skip_serializing_if = "Vec::is_empty")] acceptance: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + parent: Option, + #[serde(default, skip_serializing_if = "Option::is_none")] + scores: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + implements: Vec, + #[serde(default, skip_serializing_if = "Option::is_none")] + reversibility: Option, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + artifacts_touched: Vec, + #[serde(default, skip_serializing_if = "Vec::is_empty")] + policy_context: Vec, + /// The outside event the task waits on. Not of the spec: events set and clear it. + #[serde(default, skip_serializing_if = "Option::is_none")] + waits_on: Option, + /// The external actions begun under a claim and not yet settled, by action id. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + effects: Vec, +} + +impl Extra { + fn parse(id: &str, text: Option<&str>) -> Result { + match text { + // a row from before the column has none, and reads as a task with nothing more + None | Some("") => Ok(Extra::default()), + Some(text) => serde_json::from_str(text).map_err(|_| Error::BadRow { + id: id.to_string(), + what: format!("extra {text:?}"), + }), + } + } +} + +/// The `extra` column of task `id` after `event`, when the event changes it: the event +/// a task waits on, and the effects not settled. Both backends project through this. +pub(crate) fn extra_after( + id: &str, + extra: Option<&str>, + event: &Event, +) -> Result, Error> { + let mut e = Extra::parse(id, extra)?; + match event { + Event::Waiting { on } => e.waits_on = Some(on.clone()), + Event::Arrived { .. } => e.waits_on = None, + Event::EffectBegun { action } => { + if !e.effects.contains(action) { + e.effects.push(action.clone()); + } + } + Event::EffectSettled { action, .. } => e.effects.retain(|a| a != action), + _ => return Ok(None), + } + serde_json::to_string(&e) + .map(Some) + .map_err(|err| Error::BadRow { + id: id.to_string(), + what: format!("extra serialises: {err}"), + }) } impl TaskSpec { @@ -215,6 +352,14 @@ impl TaskSpec { let extra = Extra { definition: self.definition.clone(), acceptance: self.acceptance.clone(), + parent: self.parent.clone(), + scores: self.scores.clone(), + implements: self.implements.clone(), + reversibility: self.reversibility, + artifacts_touched: self.artifacts_touched.clone(), + policy_context: self.policy_context.clone(), + waits_on: None, + effects: Vec::new(), }; serde_json::to_string(&extra).unwrap_or_default() } @@ -251,6 +396,29 @@ pub struct Task { /// Where its handoff was posted back upstream, once a tap has done so. #[serde(default)] pub posted_to: Option, + /// The outside event it waits on: a check that runs, a reply, a change that has not + /// landed. Nobody may claim it until that event arrives. + #[serde(default, skip_serializing_if = "Option::is_none")] + pub waiting_on: Option, + /// The external actions begun under a claim of it whose outcome nobody has recorded. + /// Nobody may claim it while there is one: a lease that ran out does not say whether + /// the destination holds the effect. + #[serde(default, skip_serializing_if = "Vec::is_empty")] + pub effects: Vec, +} + +impl Task { + /// Why nobody may claim this task whatever its lease says, when something holds it + /// back: the event it waits on, or an effect nobody has settled. + #[must_use] + pub fn held_back(&self) -> Option { + if let Some(on) = &self.waiting_on { + return Some(format!("waits on {on}")); + } + self.effects + .first() + .map(|a| format!("effect {a} is not settled")) + } } /// What an erasure did to an entry's promotion records and approval tasks. @@ -385,7 +553,8 @@ type Clock = Box OffsetDateTime + Send + Sync>; enum Event { Created { id: String, - spec: TaskSpec, + /// Boxed: a spec is several times the size of any other event. + spec: Box, }, Admitted, /// Admitted with a dependency that is not done: not claimable until each is. @@ -443,6 +612,26 @@ enum Event { reason: String, superseded_by: Option, }, + /// The holder gave the task up until an outside event comes: nobody's to claim + /// until [`Event::Arrived`]. + Waiting { + on: String, + }, + /// The outside event a task waited on came. + Arrived { + event: String, + }, + /// The holder is about to make an external action. Until [`Event::EffectSettled`] + /// names the same action, nobody else may claim the task. + EffectBegun { + action: String, + }, + /// Someone looked at the destination and recorded what became of an action. + EffectSettled { + action: String, + outcome: Outcome, + by: String, + }, /// The handoff of a completed task was posted back where the task came from. Posted { to: String, @@ -480,6 +669,10 @@ impl Event { Event::HandoffLeft { .. } => "handoff_left", Event::Completed { .. } => "completed", Event::Invalidated { .. } => "invalidated", + Event::Waiting { .. } => "waiting", + Event::Arrived { .. } => "arrived", + Event::EffectBegun { .. } => "effect_begun", + Event::EffectSettled { .. } => "effect_settled", Event::Posted { .. } => "posted", Event::Rescored { .. } => "rescored", Event::Promoted { .. } => "promoted", @@ -687,14 +880,7 @@ impl RawTask { }; let requirements = json_list(&self.requirements, "requirements")?; let depends_on = json_list(&self.depends_on, "depends_on")?; - // a row from before the column has none, and reads as a task with nothing more - let extra: Extra = match self.extra.as_deref() { - None | Some("") => Extra::default(), - Some(text) => serde_json::from_str(text).map_err(|_| Error::BadRow { - id: id.clone(), - what: format!("extra {text:?}"), - })?, - }; + let extra = Extra::parse(&id, self.extra.as_deref())?; Ok(Task { spec: TaskSpec { domain: self.domain, @@ -709,6 +895,12 @@ impl RawTask { priority: self.priority, definition: extra.definition, acceptance: extra.acceptance, + parent: extra.parent, + scores: extra.scores, + implements: extra.implements, + reversibility: extra.reversibility, + artifacts_touched: extra.artifacts_touched, + policy_context: extra.policy_context, }, created: parse_time(&id, &self.created)?, status: Status::parse(&id, &self.status)?, @@ -720,6 +912,8 @@ impl RawTask { reason: self.reason, superseded_by: self.superseded_by, posted_to: self.posted_to, + waiting_on: extra.waits_on, + effects: extra.effects, id, }) } @@ -743,6 +937,7 @@ fn claimable(task: &Task, now: OffsetDateTime, manifest: &[String], handling: La let state = matches!(task.status, Status::Open | Status::Blocked) || (task.status == Status::Claimed && !lease_live); state + && task.held_back().is_none() && task.spec.classification <= handling && task.spec.requirements.iter().all(|r| manifest.contains(r)) } @@ -926,7 +1121,7 @@ impl Ledger { let now = self.now(); let event = Event::Created { id: id.clone(), - spec, + spec: Box::new(spec), }; self.append_at(&id, now, &event).await?; Ok(Created { @@ -962,7 +1157,8 @@ impl Ledger { } self.append(id, &Event::Admitted).await?; let waits_on = self.not_done(&task.spec.depends_on).await?; - if !waits_on.is_empty() { + // a task sent back to triage while it waited on an event waits on it still + if !waits_on.is_empty() || task.waiting_on.is_some() { self.append(id, &Event::Blocked { on: waits_on }).await?; } self.get(id).await @@ -1230,6 +1426,7 @@ impl Ledger { op: "complete", }); } + Self::effects_settled(&task)?; let completed = Event::Completed { handoff: handoff.to_string(), attempt: attempt.map(str::to_string), @@ -1285,6 +1482,7 @@ impl Ledger { let held = task.status == Status::Claimed && task.lease_until.is_some_and(|t| t >= now); let resumed_it = continuation.spec.origin.as_deref() == Some(continues(id).as_str()) && continuation.status == Status::Done; + Self::effects_settled(&task)?; if task.status.is_terminal() || held || !resumed_it { return Err(Error::WrongState { id: id.to_string(), @@ -1335,6 +1533,177 @@ impl Ledger { self.get(id).await } + /// A task with an effect nobody has settled is not done, whoever says so: whether + /// the destination holds the effect is part of what done means. + fn effects_settled(task: &Task) -> Result<(), Error> { + if task.effects.is_empty() { + return Ok(()); + } + Err(Error::WrongState { + id: task.id.clone(), + status: task.status, + op: "complete while an effect of it is not settled", + }) + } + + /// Give a task up until an outside event comes: a check that runs, a reply, a change + /// that has not landed. The task is blocked and nobody's to claim until + /// [`Ledger::arrived`] is told of `on`; the holder's session can end meanwhile. A + /// handoff says where the next holder starts. + /// + /// # Errors + /// + /// [`Error::NotFound`], [`Error::NotHolder`], [`Error::Fenced`], or + /// [`Error::WrongState`]. + pub async fn wait_for( + &mut self, + id: &str, + by: Held<'_>, + on: &str, + handoff: Option<&str>, + ) -> Result { + let now = self.now(); + let task = self.get(id).await?; + if task.status != Status::Claimed { + return Err(Error::WrongState { + id: id.to_string(), + status: task.status, + op: "wait", + }); + } + if let Some(handoff) = handoff { + let left = Event::HandoffLeft { + handoff: handoff.to_string(), + }; + self.append_held(id, now, by, &left).await?; + } + let waiting = Event::Waiting { on: on.to_string() }; + self.append_held(id, now, by, &waiting).await?; + self.get(id).await + } + + /// An outside event came: every task of the domain that waited on it waits on it no + /// more, and is open again once its dependencies are done. The tasks that waited. + /// + /// # Errors + /// + /// A database error. + pub async fn arrived(&mut self, domain: &str, event: &str) -> Result, Error> { + let mut woken = Vec::new(); + // whatever its state says: a task sent back to triage while it waited waits still + for status in [Status::Blocked, Status::Proposed, Status::Open] { + for t in self.list(domain, status).await? { + if t.waiting_on.as_deref() != Some(event) { + continue; + } + let came = Event::Arrived { + event: event.to_string(), + }; + self.append(&t.id, &came).await?; + if status == Status::Blocked && self.not_done(&t.spec.depends_on).await?.is_empty() + { + let by = event.to_string(); + self.append(&t.id, &Event::Unblocked { by }).await?; + } + woken.push(self.get(&t.id).await?); + } + } + Ok(woken) + } + + /// Record that the holder is about to make an external action, before it makes it. + /// From here until [`Ledger::effect_settled`] names the action, nobody else may + /// claim the task: a lease that runs out does not say whether the destination holds + /// the effect. A retry of the action carries the same id, and is the same effect. + /// + /// # Errors + /// + /// [`Error::NotFound`], [`Error::NotHolder`], or [`Error::Fenced`]: the claim is + /// checked in the transaction that records the effect. + pub async fn effect_begun( + &mut self, + id: &str, + by: Held<'_>, + action: &str, + ) -> Result { + let now = self.now(); + let begun = Event::EffectBegun { + action: action.to_string(), + }; + self.append_held(id, now, by, &begun).await?; + self.get(id).await + } + + /// Record what became of an external action, as `by` found the destination. With + /// `held`, it is the holder that says so, and the claim is checked in the + /// transaction that records it. Without, it is someone who reconciles after the + /// claim: whether `by` may is the caller's rule, and the ledger records who did. + /// Asked again for an action already settled, it changes nothing. + /// + /// # Errors + /// + /// [`Error::NotFound`]; [`Error::WrongState`] for an action the task never began; + /// with `held`, [`Error::NotHolder`] or [`Error::Fenced`]. + pub async fn effect_settled( + &mut self, + id: &str, + action: &str, + outcome: Outcome, + (by, held): (&str, Option>), + ) -> Result { + let task = self.get(id).await?; + if task.effects.iter().any(|a| a == action) { + let settled = Event::EffectSettled { + action: action.to_string(), + outcome, + by: by.to_string(), + }; + match held { + Some(held) => { + let now = self.now(); + self.append_held(id, now, held, &settled).await?; + } + None => self.append(id, &settled).await?, + } + return self.get(id).await; + } + if self.settlement(id, action).await?.is_some() { + return Ok(task); + } + Err(Error::WrongState { + id: id.to_string(), + status: task.status, + op: "settle an effect the task did not begin", + }) + } + + /// How an action of a task was settled, and by whom, when it was. + /// + /// # Errors + /// + /// A database error. + pub async fn settlement( + &self, + id: &str, + action: &str, + ) -> Result, Error> { + let payloads = on_backend!(&self.backend, db => db.payloads_of(id, "effect_settled"))?; + let mut last = None; + for p in payloads { + if let Event::EffectSettled { + action: a, + outcome, + by, + } = Event::parse(id, &p)? + { + if a == action { + last = Some((outcome, by)); + } + } + } + Ok(last) + } + /// Leave a handoff on a task after the claim it was written under is over: the /// lease ran out, the task may be held again, and what the earlier holder learned is /// still worth the next one's reading. It changes nothing of the task. `by` names the @@ -1967,6 +2336,10 @@ mod tests { two_claimers_on_two_connections_get_two_tasks, a_domain_claims_only_its_own_queue, a_rescore_moves_the_queue_and_keeps_the_filed_priority, + a_task_that_waits_on_an_event_is_nobody_s_until_the_event_comes, + an_effect_not_settled_holds_the_task_past_its_lease, + a_settled_effect_gives_the_task_back_and_fences_who_began_it, + a_task_keeps_the_rest_of_its_record, a_task_waits_on_its_continuation, a_task_closes_with_its_continuation, a_continuation_given_up_on_gives_its_task_back, @@ -3975,10 +4348,10 @@ mod tests { let id = dm_core::id::new_task_id(domain); let created = Event::Created { id: id.clone(), - spec: TaskSpec { + spec: Box::new(TaskSpec { domain: domain.into(), ..spec("Old") - }, + }), }; let mut events = vec![created]; if admit { @@ -4239,6 +4612,181 @@ mod tests { assert_eq!(attr(3, "memory.domain"), Some(Value::from("dom-a"))); } + async fn a_task_that_waits_on_an_event_is_nobody_s_until_the_event_comes(b: &Site) { + let (mut l, now) = b.ledger().await; + let t = admitted(&mut l, spec("Open the change and wait for its check")).await; + let c = claim(&mut l, &t.id, "s1", mins(30)).await.unwrap(); + let by = held("s1", c.generation); + let parked = l + .wait_for(&t.id, by, "check:42", Some("mem-a-h1")) + .await + .unwrap(); + it_is_parked(&parked); + // its dependencies are done and no lease is live, and still nobody gets it + assert!(next(&mut l, "s2", mins(30)).await.is_none()); + *now.lock().unwrap() = T0 + mins(90); + nobody_gets_it(&mut l, &t.id).await; + it_waits_through_a_replay(&mut l, &t.id, by).await; + only_its_event_wakes_it(&mut l, &t).await; + the_woken_task_is_claimed_again(&mut l, &t).await; + } + + fn it_is_parked(parked: &Task) { + assert_eq!(parked.status, Status::Blocked); + assert_eq!(parked.waiting_on.as_deref(), Some("check:42")); + assert_eq!(parked.holder, None); + assert_eq!(parked.handoff.as_deref(), Some("mem-a-h1")); + assert_eq!(parked.held_back().as_deref(), Some("waits on check:42")); + } + + async fn it_waits_through_a_replay(l: &mut Ledger, id: &str, by: Held<'_>) { + // who gave it up holds it no more + assert!(is_not_holder(&l.heartbeat(id, by, mins(30)).await)); + // a replay reads the same from the events + l.rebuild().await.unwrap(); + let replayed = l.get(id).await.unwrap(); + assert_eq!(replayed.waiting_on.as_deref(), Some("check:42")); + assert!(next(l, "s2", mins(30)).await.is_none()); + } + + /// Neither the next claim nor a claim by name gets the task. + async fn nobody_gets_it(l: &mut Ledger, id: &str) { + assert!(next(l, "s2", mins(30)).await.is_none()); + assert!(is_wrong_state(&claim(l, id, "s2", mins(30)).await)); + } + + async fn only_its_event_wakes_it(l: &mut Ledger, t: &Task) { + assert!(l.arrived("dom-a", "check:41").await.unwrap().is_empty()); + assert!(l.arrived("dom-b", "check:42").await.unwrap().is_empty()); + assert!(next(l, "s2", mins(30)).await.is_none()); + let woken = l.arrived("dom-a", "check:42").await.unwrap(); + assert_eq!(woken.len(), 1); + assert_eq!(woken[0].id, t.id); + assert_eq!(woken[0].status, Status::Open); + assert_eq!(woken[0].waiting_on, None); + } + + async fn the_woken_task_is_claimed_again(l: &mut Ledger, t: &Task) { + // told twice, the event wakes nothing more + assert!(l.arrived("dom-a", "check:42").await.unwrap().is_empty()); + let c = next(l, "s2", mins(30)).await.unwrap(); + assert_eq!(c.task.id, t.id); + assert_eq!(c.generation, 2); + assert_eq!(c.task.handoff.as_deref(), Some("mem-a-h1")); + assert!(is_fenced( + &l.wait_for(&t.id, held("s1", 1), "check:43", None).await, + 1, + 2 + )); + } + + async fn an_effect_not_settled_holds_the_task_past_its_lease(b: &Site) { + let (mut l, now) = b.ledger().await; + let t = admitted(&mut l, spec("Post the release note")).await; + let c = claim(&mut l, &t.id, "s1", mins(30)).await.unwrap(); + let by = held("s1", c.generation); + l.effect_begun(&t.id, by, "act-1").await.unwrap(); + // the retry of the action is the same effect + let again = l.effect_begun(&t.id, by, "act-1").await.unwrap(); + assert_eq!(again.effects, vec!["act-1".to_string()]); + // its holder cannot call it done while nobody knows what the destination holds + assert!(is_wrong_state(&l.complete(&t.id, by, "mem-a-h").await)); + // and says what it holds itself while its claim is in force + l.effect_begun(&t.id, by, "act-0").await.unwrap(); + let own = l + .effect_settled(&t.id, "act-0", Outcome::Applied, ("s1", Some(by))) + .await + .unwrap(); + assert_eq!(own.effects, vec!["act-1".to_string()]); + *now.lock().unwrap() = T0 + mins(31); + // the lease ran out, and that alone gives the task to nobody + nobody_gets_it(&mut l, &t.id).await; + the_effect_holds_through_a_replay(&mut l, &t.id, by).await; + } + + async fn the_effect_holds_through_a_replay(l: &mut Ledger, id: &str, by: Held<'_>) { + let held_back = l.get(id).await.unwrap().held_back(); + assert_eq!(held_back.as_deref(), Some("effect act-1 is not settled")); + // nor may the one who lost the lease begin another + assert!(is_not_holder(&l.effect_begun(id, by, "act-2").await)); + l.rebuild().await.unwrap(); + assert_eq!(l.get(id).await.unwrap().effects, vec!["act-1"]); + assert!(next(l, "s2", mins(30)).await.is_none()); + } + + async fn a_settled_effect_gives_the_task_back_and_fences_who_began_it(b: &Site) { + let (mut l, now) = b.ledger().await; + let t = admitted(&mut l, spec("Post the release note")).await; + let c = claim(&mut l, &t.id, "s1", mins(30)).await.unwrap(); + let by = held("s1", c.generation); + l.effect_begun(&t.id, by, "act-1").await.unwrap(); + *now.lock().unwrap() = T0 + mins(31); + not_settled_by_a_wrong_name_or_a_stale_holder(&mut l, &t.id, by).await; + assert_eq!(l.settlement(&t.id, "act-1").await.unwrap(), None); + let settled = l + .effect_settled(&t.id, "act-1", Outcome::NotApplied, ("human:a:x", None)) + .await + .unwrap(); + assert!(settled.effects.is_empty()); + assert_eq!( + l.settlement(&t.id, "act-1").await.unwrap(), + Some((Outcome::NotApplied, "human:a:x".to_string())) + ); + settled_once_and_claimed_again(&mut l, &t, by).await; + } + + async fn not_settled_by_a_wrong_name_or_a_stale_holder(l: &mut Ledger, id: &str, by: Held<'_>) { + let unknown = l + .effect_settled(id, "act-9", Outcome::Applied, ("human:a:x", None)) + .await; + assert!(is_wrong_state(&unknown)); + // who lost the lease does not say what became of it + let stale = l + .effect_settled(id, "act-1", Outcome::Applied, ("s1", Some(by))) + .await; + assert!(is_not_holder(&stale)); + } + + async fn settled_once_and_claimed_again(l: &mut Ledger, t: &Task, by: Held<'_>) { + // said again, it is the same settlement and no second event + let events = l.event_count().await.unwrap(); + l.effect_settled(&t.id, "act-1", Outcome::Applied, ("human:a:y", None)) + .await + .unwrap(); + assert_eq!(l.event_count().await.unwrap(), events); + let c2 = next(l, "s2", mins(30)).await.unwrap(); + assert_eq!((c2.task.id.as_str(), c2.generation), (t.id.as_str(), 2)); + // the delayed request of the first holder is refused where the effect is recorded + assert!(is_fenced(&l.effect_begun(&t.id, by, "act-1").await, 1, 2)); + assert!(l.get(&t.id).await.unwrap().effects.is_empty()); + } + + async fn a_task_keeps_the_rest_of_its_record(b: &Site) { + let (mut l, _) = b.ledger().await; + let mut s = spec("Carry out the decision"); + s.parent = Some("task-a-1".into()); + s.scores = Some(Scores { + value: 8, + urgency: 5, + effort: 3, + unblocks: vec!["task-a-9".into()], + }); + s.implements = vec!["mem-a-0847".into()]; + s.reversibility = Some(Reversibility::Irreversible); + s.artifacts_touched = vec!["docs/**".into()]; + s.policy_context = vec!["pol-L1-007".into()]; + let filed = l.create(s.clone()).await.unwrap().task; + assert_eq!(filed.spec, s); + drop(l); + let l = b.open().await; + assert_eq!(l.get(&filed.id).await.unwrap().spec, s); + // a task without them says nothing of them + let plain = serde_json::to_value(spec("Plain")).unwrap(); + for absent in ["parent", "scores", "implements", "reversibility"] { + assert!(plain.get(absent).is_none(), "{absent}"); + } + } + async fn a_rescore_moves_the_queue_and_keeps_the_filed_priority(b: &Site) { let (mut l, _) = b.ledger().await; let first = admitted(&mut l, spec("First")).await; diff --git a/crates/ledger/src/postgres.rs b/crates/ledger/src/postgres.rs index 6d68de06..add7c926 100644 --- a/crates/ledger/src/postgres.rs +++ b/crates/ledger/src/postgres.rs @@ -1336,6 +1336,41 @@ async fn insert_promotion( } /// Update the projection for one event. Replay calls this and nothing else. +/// The projection of the events that hold a task back from a claim, or stop doing so: +/// what it waits on and which effects are not settled live in `extra`. +async fn project_hold( + conn: &C, + s: &str, + id: &str, + event: &Event, +) -> Result<(), Error> { + let row = conn + .query_opt( + &format!("SELECT extra FROM {s}.tasks WHERE id = $1 FOR UPDATE"), + &[&id], + ) + .await?; + let extra: Option = row.map(|r| r.try_get(0)).transpose()?.flatten(); + if let Some(next) = crate::extra_after(id, extra.as_deref(), event)? { + conn.execute( + &format!("UPDATE {s}.tasks SET extra = $2 WHERE id = $1"), + &[&id, &next], + ) + .await?; + } + if matches!(event, Event::Waiting { .. }) { + conn.execute( + &format!( + "UPDATE {s}.tasks SET status = 'blocked', holder = NULL, lease_until = NULL + WHERE id = $1" + ), + &[&id], + ) + .await?; + } + Ok(()) +} + async fn project( conn: &C, s: &str, @@ -1391,6 +1426,10 @@ async fn project( lease_until = NULL WHERE id = $1", vec![&id, reason, superseded_by], ), + Event::Waiting { .. } + | Event::Arrived { .. } + | Event::EffectBegun { .. } + | Event::EffectSettled { .. } => return project_hold(conn, s, id, event).await, Event::Posted { to } => ("SET posted_to = $2 WHERE id = $1", vec![&id, to]), Event::Rescored { priority, .. } => { ("SET priority = $2 WHERE id = $1", vec![&id, priority]) diff --git a/crates/ledger/src/sqlite.rs b/crates/ledger/src/sqlite.rs index f0f9b1d5..26f5f72a 100644 --- a/crates/ledger/src/sqlite.rs +++ b/crates/ledger/src/sqlite.rs @@ -883,6 +883,31 @@ fn project_waiting(conn: &Connection, id: &str, event: &Event) -> Result<(), Err Ok(()) } +/// The projection of the events that hold a task back from a claim, or stop doing so: +/// what it waits on and which effects are not settled live in `extra`. +fn project_hold(conn: &Connection, id: &str, event: &Event) -> Result<(), Error> { + let extra: Option = conn + .query_row("SELECT extra FROM tasks WHERE id = ?1", params![id], |r| { + r.get(0) + }) + .optional()? + .flatten(); + if let Some(next) = crate::extra_after(id, extra.as_deref(), event)? { + conn.execute( + "UPDATE tasks SET extra = ?2 WHERE id = ?1", + params![id, next], + )?; + } + if matches!(event, Event::Waiting { .. }) { + conn.execute( + "UPDATE tasks SET status = 'blocked', holder = NULL, lease_until = NULL + WHERE id = ?1", + params![id], + )?; + } + Ok(()) +} + /// Update the projection for one event. Replay calls this and nothing else. fn project(conn: &Connection, id: &str, at: OffsetDateTime, event: &Event) -> Result<(), Error> { match event { @@ -942,6 +967,10 @@ fn project(conn: &Connection, id: &str, at: OffsetDateTime, event: &Event) -> Re params![id, reason, superseded_by], )?; } + Event::Waiting { .. } + | Event::Arrived { .. } + | Event::EffectBegun { .. } + | Event::EffectSettled { .. } => project_hold(conn, id, event)?, Event::Posted { to } => { conn.execute( "UPDATE tasks SET posted_to = ?2 WHERE id = ?1", diff --git a/crates/memory/src/main.rs b/crates/memory/src/main.rs index 0653862e..baf73f5c 100644 --- a/crates/memory/src/main.rs +++ b/crates/memory/src/main.rs @@ -1124,6 +1124,21 @@ enum TaskOp { /// takes those of its definition. #[arg(long = "accept")] acceptance: Vec, + /// The task this one is split from. + #[arg(long)] + parent: Option, + /// A decision this task carries out, as an entry id; repeat for more. + #[arg(long = "implements")] + implements: Vec, + /// Whether what the task does can be undone: `reversible` or `irreversible`. + #[arg(long)] + reversibility: Option, + /// An owned path the task changes, as a glob; repeat for more. + #[arg(long = "touches")] + artifacts_touched: Vec, + /// What triage scored it: `value,urgency,effort`, three whole numbers. + #[arg(long)] + scores: Option, }, /// Admit a proposed task. Admit { id: String }, @@ -1167,6 +1182,44 @@ enum TaskOp { #[arg(long)] handoff: Option, }, + /// Give a held task up until an outside event comes. Nobody may claim it until + /// `task arrived` names the event. + Wait { + id: String, + /// The name of the event, one token. + on: String, + /// The generation of the claim. + #[arg(long)] + generation: u64, + #[arg(long)] + handoff: Option, + }, + /// Report that an outside event came: the tasks that waited on it are open again. + /// A tap or a person reports it, not an agent. + Arrived { + event: String, + #[arg(long)] + domain: Option, + }, + /// Record an external action before it is made. Until it is settled, the task + /// goes to nobody else, whatever its lease says. + EffectBegin { + id: String, + /// A stable id for the action, one token; a retry names the same id. + action: String, + /// The generation of the claim. + #[arg(long)] + generation: u64, + }, + /// Record what became of an external action: `applied` or `not_applied`. The + /// holder names the generation of its claim; a person who reconciles names none. + EffectSettle { + id: String, + action: String, + outcome: dm_ledger::Outcome, + #[arg(long)] + generation: Option, + }, /// Leave a handoff on a task after the claim of that generation is over. LateHandoff { id: String, @@ -4899,6 +4952,53 @@ async fn held_task(m: &Memory, op: TaskOp) -> Fallible { } } +/// What holds a task back from a claim though nobody holds it, and what ends that. +async fn holds_cmd(m: &Memory, op: TaskOp) -> Fallible { + match op { + TaskOp::Wait { + id, + on, + generation, + handoff, + } => json( + &m.wait_for(&id, Some(generation), &on, handoff.as_deref()) + .await?, + ), + TaskOp::Arrived { event, domain } => { + json(&m.event_arrived(domain.as_deref(), &event).await?) + } + TaskOp::EffectBegin { + id, + action, + generation, + } => json(&m.effect_begun(&id, Some(generation), &action).await?), + TaskOp::EffectSettle { + id, + action, + outcome, + generation, + } => json(&m.effect_settled(&id, generation, &action, outcome).await?), + _ => Ok(()), + } +} + +/// The `--scores value,urgency,effort` of `task create`. +fn scores_of(text: &str) -> Result> { + let parts: Vec = text + .split(',') + .map(|n| n.trim().parse::()) + .collect::>()?; + let [value, urgency, effort] = parts[..] else { + return Err("scores are value,urgency,effort: three whole numbers".into()); + }; + Ok(dm_ledger::Scores { + value, + urgency, + effort, + unblocks: Vec::new(), + }) +} + /// `amem task rescore` and `amem task stats`: the fabric's ranking and its telemetry. async fn fabric_cmd(m: &Memory, op: TaskOp) -> Fallible { match op { @@ -4919,7 +5019,13 @@ async fn task(m: &Memory, op: TaskOp) -> Fallible { priority, definition, acceptance, + parent, + implements, + reversibility, + artifacts_touched, + scores, } => { + let scores = scores.as_deref().map(scores_of).transpose()?; let c = m .create_task(TaskSpec { domain: domain.unwrap_or_default(), @@ -4934,6 +5040,12 @@ async fn task(m: &Memory, op: TaskOp) -> Fallible { priority, definition, acceptance, + parent, + scores, + implements, + reversibility, + artifacts_touched, + ..TaskSpec::default() }) .await?; json(&c.task) @@ -4957,6 +5069,10 @@ async fn task(m: &Memory, op: TaskOp) -> Fallible { | TaskOp::Complete { .. } | TaskOp::Release { .. } | TaskOp::LateHandoff { .. }) => held_task(m, op).await, + op @ (TaskOp::Wait { .. } + | TaskOp::Arrived { .. } + | TaskOp::EffectBegin { .. } + | TaskOp::EffectSettle { .. }) => holds_cmd(m, op).await, TaskOp::Invalidate { id, reason } => json(&m.invalidate(&id, &reason, None).await?), TaskOp::Tree { id } => json(&m.task_tree(&id).await?), TaskOp::Retry { id } => json(&m.retry_task(&id).await?), diff --git a/crates/server/src/holds.rs b/crates/server/src/holds.rs new file mode 100644 index 00000000..599accdb --- /dev/null +++ b/crates/server/src/holds.rs @@ -0,0 +1,293 @@ +//! What holds a task back from a claim though nobody holds it: an outside event it +//! waits on, and an external effect whose outcome nobody has recorded. A session does +//! not wait for an event, and a lease that ran out does not say what a destination holds. + +use crate::{lock, Error, Memory}; +use dm_ledger::{Outcome, Task}; +use dm_policy::decide::Class; + +/// A name a caller gives an event or an action: one token, so it reads the same in a +/// log line, a command and a tool call. +fn token(what: &str, name: &str, most: usize) -> Result<(), Error> { + let plain = !name.is_empty() + && name.chars().count() <= most + && name.chars().all(|c| c.is_ascii_graphic()); + if plain { + Ok(()) + } else { + Err(Error::Denied(format!( + "{what} is 1 to {most} printable ASCII characters without a space" + ))) + } +} + +impl Memory { + /// Give a held task up until an outside event comes: a check that runs, a reply, a + /// change that has not landed. Nobody may claim it until a tap or a person reports + /// the event with [`Memory::event_arrived`]. The handoff, when given, must exist + /// and be readable here; the next holder starts from it. + /// + /// # Errors + /// + /// [`Error::Denied`] for an event name that is not one token; a ledger error, + /// including a lease already lost; a task outside the grants. + pub async fn wait_for( + &self, + id: &str, + generation: Option, + on: &str, + handoff: Option<&str>, + ) -> Result { + token("an event name", on, 200)?; + if let Some(h) = handoff { + self.get(h).await?; + } + let mut l = self.ledger.lock().await; + self.check_task(&l.get(id).await?)?; + let by = self.under(self.claim_of(id, generation)?); + let t = l.wait_for(id, by, on, handoff).await?; + lock(&self.claims).remove(id); + self.observe_task(&t); + Ok(t) + } + + /// Report that an outside event came: each task of the domain that waited on it is + /// open again, for the service to offer. A tap reports it, or a person. An agent + /// does not: it would be the agent that says its own wait is over. The tasks this + /// session may see are returned, each read. + /// + /// # Errors + /// + /// [`Error::Denied`] for an agent, or an event name that is not one token; an + /// ungranted domain; a ledger error. + pub async fn event_arrived( + &self, + domain: Option<&str>, + event: &str, + ) -> Result, Error> { + token("an event name", event, 200)?; + if Class::of(&self.author, &self.grants.owner) == Class::Agent { + return Err(Error::Denied( + "an agent does not report the event a task waits on: a tap or a person does".into(), + )); + } + let domain = self.resolve_domain(domain)?; + let woken = self.ledger.lock().await.arrived(&domain, event).await?; + self.read_tasks(woken) + } + + /// Record that this session is about to make an external action under its claim, + /// before it makes it. Until the action is settled, the task goes to nobody else, + /// whatever its lease says. A retry of the action names the same id. + /// + /// # Errors + /// + /// [`Error::Denied`] for an action id that is not one token; a ledger error, + /// including a lease already lost or a claim made since; a task outside the grants. + pub async fn effect_begun( + &self, + id: &str, + generation: Option, + action: &str, + ) -> Result { + token("an action id", action, 64)?; + let mut l = self.ledger.lock().await; + self.check_task(&l.get(id).await?)?; + let by = self.under(self.claim_of(id, generation)?); + let t = l.effect_begun(id, by, action).await?; + self.observe_task(&t); + Ok(t) + } + + /// Record what became of an external action. The holder says so while its claim is + /// in force: it had the destination's answer. After that, a person or a tap that + /// looked at the destination says so; an agent that lost its claim does not, since + /// it cannot tell a lost answer from an action that was not made. + /// + /// # Errors + /// + /// [`Error::Denied`] for an action id that is not one token; a ledger error: for an + /// agent without the claim in force, the task's holder check; for an action the + /// task did not begin, a wrong state; a task outside the grants. + pub async fn effect_settled( + &self, + id: &str, + generation: Option, + action: &str, + outcome: Outcome, + ) -> Result { + token("an action id", action, 64)?; + let mut l = self.ledger.lock().await; + self.check_task(&l.get(id).await?)?; + let reconciles = Class::of(&self.author, &self.grants.owner) != Class::Agent; + let own = lock(&self.claims).get(id).copied(); + // a person or a tap that made no claim here reconciles; everyone else speaks + // under a claim, and the ledger checks it + let held = if reconciles && own.is_none() { + None + } else { + Some(self.under(self.claim_of(id, generation)?)) + }; + let t = l + .effect_settled(id, action, outcome, (&self.author, held)) + .await?; + self.observe_task(&t); + Ok(t) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use dm_ledger::{Status, TaskSpec}; + use std::path::Path; + + fn mem(dir: &Path) -> Memory { + Memory::init(dir, "human:test:me", "dom-me") + .unwrap() + .without_read_log() + } + + fn as_author(dir: &Path, author: &str) -> Memory { + Memory::open(dir) + .unwrap() + .with_author(author) + .without_read_log() + } + + /// A task the owner filed and admitted, claimed by `agent:a`; that session and the id. + async fn held(owner: &Memory, dir: &Path, title: &str) -> (Memory, String) { + let task = owner + .create_task(TaskSpec { + title: title.into(), + body: "Do it.\n".into(), + ..TaskSpec::default() + }) + .await + .unwrap() + .task; + owner.admit(&task.id).await.unwrap(); + let agent = as_author(dir, "agent:a"); + agent + .claim(&task.id, time::Duration::minutes(5)) + .await + .unwrap(); + (agent, task.id) + } + + fn denied(r: &Result, saying: &str) -> bool { + matches!(r, Err(Error::Denied(why)) if why.contains(saying)) + } + + #[tokio::test] + async fn a_task_that_waits_is_offered_again_when_a_tap_reports_the_event() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let (agent, id) = held(&owner, dir.path(), "Open the change and wait").await; + let spaced = agent.wait_for(&id, None, "check 42", None).await; + assert!(denied(&spaced, "an event name")); + let waiting = agent.wait_for(&id, None, "check:42", None).await.unwrap(); + assert_eq!(waiting.status, Status::Blocked); + assert_eq!(waiting.waiting_on.as_deref(), Some("check:42")); + // the session that gave it up holds it no more, and nobody else gets it + assert!(agent + .heartbeat(&id, None, time::Duration::minutes(5)) + .await + .is_err()); + let other = as_author(dir.path(), "agent:b"); + assert!(other.claim(&id, time::Duration::minutes(5)).await.is_err()); + // an agent does not say that its own wait is over + let by_agent = other.event_arrived(None, "check:42").await; + assert!(denied(&by_agent, "a tap or a person")); + assert_eq!(owner.get_task(&id).await.unwrap().status, Status::Blocked); + let tap = as_author(dir.path(), "tap:checks"); + assert!(tap + .event_arrived(None, "check:41") + .await + .unwrap() + .is_empty()); + let woken = tap.event_arrived(None, "check:42").await.unwrap(); + assert_eq!(woken.len(), 1); + assert_eq!( + (woken[0].id.as_str(), woken[0].status), + (id.as_str(), Status::Open) + ); + let again = other.claim(&id, time::Duration::minutes(5)).await.unwrap(); + assert_eq!(again.generation, 2); + } + + #[tokio::test] + async fn an_effect_is_settled_by_its_holder_or_by_a_person_and_by_no_other_agent() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let (agent, id) = held(&owner, dir.path(), "Post the release note").await; + let spaced = agent.effect_begun(&id, None, "act 1").await; + assert!(denied(&spaced, "an action id")); + let begun = agent.effect_begun(&id, None, "act-1").await.unwrap(); + assert_eq!(begun.effects, vec!["act-1".to_string()]); + // another agent's session made no claim, and names the holder's generation + let other = as_author(dir.path(), "agent:b"); + let unheld = other + .effect_settled(&id, None, "act-1", Outcome::Applied) + .await; + assert!(matches!(unheld, Err(Error::Ledger(_))), "{unheld:?}"); + let named = other + .effect_settled(&id, Some(begun.generation), "act-1", Outcome::Applied) + .await; + assert!(matches!(named, Err(Error::Ledger(_))), "{named:?}"); + assert_eq!(owner.get_task(&id).await.unwrap().effects.len(), 1); + // the holder had the destination's answer + let own = agent + .effect_settled(&id, None, "act-1", Outcome::Applied) + .await + .unwrap(); + assert!(own.effects.is_empty()); + // and a person reconciles one whose holder is gone + agent.effect_begun(&id, None, "act-2").await.unwrap(); + let person = as_author(dir.path(), "human:test:me"); + let reconciled = person + .effect_settled(&id, None, "act-2", Outcome::NotApplied) + .await + .unwrap(); + assert!(reconciled.effects.is_empty()); + } + + #[tokio::test] + async fn the_service_says_which_policy_a_task_was_filed_under_and_checks_its_parent() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let filed = owner + .create_task(TaskSpec { + title: "Carry out the decision".into(), + body: "Do it.\n".into(), + policy_context: vec!["pol-made-up".into()], + implements: vec!["mem-me-0847".into()], + ..TaskSpec::default() + }) + .await + .unwrap() + .task; + assert_eq!(filed.spec.policy_context, Vec::::new()); + assert_eq!(filed.spec.implements, vec!["mem-me-0847".to_string()]); + let orphan = owner + .create_task(TaskSpec { + title: "A part of nothing".into(), + body: "Do it.\n".into(), + parent: Some("task-me-nowhere".into()), + ..TaskSpec::default() + }) + .await; + assert!(orphan.is_err()); + let part = owner + .create_task(TaskSpec { + title: "A part".into(), + body: "Do it.\n".into(), + parent: Some(filed.id.clone()), + ..TaskSpec::default() + }) + .await + .unwrap() + .task; + assert_eq!(part.spec.parent, Some(filed.id)); + } +} diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index 41ef88ef..f3984e6f 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -37,6 +37,7 @@ use tokio::sync::Mutex as AsyncMutex; pub mod bundle; pub mod contexts; pub mod fabric; +mod holds; pub mod log; pub mod mcp; pub mod platform; @@ -2159,6 +2160,11 @@ impl Memory { filing: Filing, parent: Option<&Task>, ) -> Result { + // a parent the filer names is a task it can read; one it cannot is not found. + // The task a session holds is its parent without the asking + if let (Some(named), None) = (&spec.parent, parent) { + self.check_task(&l.get(named).await?)?; + } let mut spec = self.stamp(spec, filing, parent)?; // what a session files while it holds a task is that task's work: it runs under // the same definition, and is checked the same way, unless it names its own @@ -2205,14 +2211,31 @@ impl Memory { )?; spec.created_by = self.author.clone(); spec.classification = spec.classification.max(self.label()); + // the policy in force is the service's to say, not the filer's + spec.policy_context = self.policy_context(); // a session files no higher than the task it holds: work it delegates serves // that task, and rank is not something a worker can raise if let (Filing::Held, Some(p)) = (filing, parent) { spec.priority = spec.priority.min(p.spec.priority); + spec.parent = Some(p.id.clone()); } Ok(spec) } + /// The verified policy entries of the record now, by id: what a task filed now is + /// filed under. + fn policy_context(&self) -> Vec { + let policy = self.policy(); + let mut ids: Vec = policy + .entries() + .iter() + .filter(|e| e.tier == dm_core::TrustTier::Verified) + .map(|e| e.entry_id.clone()) + .collect(); + ids.sort(); + ids + } + /// Admit a proposed task. On a laptop the owner is the triage. /// /// # Errors @@ -2324,6 +2347,8 @@ impl Memory { priority: parent.spec.priority, definition: parent.spec.definition.clone(), acceptance: parent.spec.acceptance.clone(), + parent: Some(parent.id.clone()), + policy_context: self.policy_context(), ..TaskSpec::default() } } @@ -2504,6 +2529,8 @@ impl Memory { priority: parent.spec.priority, definition: parent.spec.definition.clone(), acceptance: parent.spec.acceptance.clone(), + policy_context: self.policy_context(), + ..TaskSpec::default() }) .await?; let continuation = self diff --git a/crates/server/src/mcp.rs b/crates/server/src/mcp.rs index 9b11c191..3d7c3953 100644 --- a/crates/server/src/mcp.rs +++ b/crates/server/src/mcp.rs @@ -343,6 +343,52 @@ pub struct CreateTaskParams { /// Checks that say this task is done, in place of its definition's. #[serde(default)] pub acceptance: Vec, + /// The task this one is split from, when it is not the task you hold. + #[serde(default)] + pub parent: Option, + /// The decisions this task carries out, as entry ids. + #[serde(default)] + pub implements: Vec, + /// Whether what the task does can be undone: `reversible` or `irreversible`. + #[serde(default)] + pub reversibility: Option, + /// The owned paths the task changes, as globs. + #[serde(default)] + pub artifacts_touched: Vec, +} + +/// Parameters of `wait_for`. +#[derive(Deserialize, schemars::JsonSchema, Default)] +pub struct WaitParams { + /// The task id. + pub id: String, + /// The generation of your claim. Optional in the session that made the claim. + #[serde(default)] + pub generation: Option, + /// The name of the outside event the task waits on, one token, as the tap that + /// reports it will name it. + pub on: String, + /// A handoff entry you wrote about what you did before waiting; the next claimant + /// starts from it. Optional. + #[serde(default)] + pub handoff: Option, +} + +/// Parameters of `effect_begun` and `effect_settled`. +#[derive(Deserialize, schemars::JsonSchema, Default)] +pub struct EffectParams { + /// The task id. + pub id: String, + /// The generation of your claim. Optional in the session that made the claim. + #[serde(default)] + pub generation: Option, + /// A stable id for the external action, one token of at most 64 characters. A retry + /// of the action names the same id. + pub action: String, + /// For `effect_settled`: `applied` when the destination holds the effect, + /// `not_applied` when it does not. + #[serde(default)] + pub outcome: Option, } /// Parameters that name a task. @@ -852,6 +898,11 @@ impl McpServer { priority: p.priority.unwrap_or(0), definition: p.definition, acceptance: p.acceptance, + parent: p.parent, + implements: p.implements, + reversibility: parse("reversibility", p.reversibility)?, + artifacts_touched: p.artifacts_touched, + ..dm_ledger::TaskSpec::default() }; let c = self.memory.create_task(spec).await.map_err(mcp_err)?; // what a task that stays proposed lacks goes back to whoever filed it @@ -976,6 +1027,50 @@ impl McpServer { ok(json!({ "task": t })) } + #[tool( + description = "Give a held task up until an outside event comes: a check that runs, a reply, a change that has not landed. Do not wait in the session. The task is blocked, and nobody may claim it, until a tap or a person reports the event; then it is offered again. Pass the id of a handoff you wrote so the next claimant starts where you stopped." + )] + async fn wait_for(&self, p: Parameters) -> Result { + let p = p.0; + let t = self + .memory + .wait_for(&p.id, p.generation, &p.on, p.handoff.as_deref()) + .await + .map_err(mcp_err)?; + ok(json!({ "task": t })) + } + + #[tool( + description = "Record that you are about to make an external action under your claim, before you make it. Until the action is settled, the task goes to nobody else, even when your lease runs out. A retry of the action names the same action id." + )] + async fn effect_begun(&self, p: Parameters) -> Result { + let p = p.0; + let t = self + .memory + .effect_begun(&p.id, p.generation, &p.action) + .await + .map_err(mcp_err)?; + ok(json!({ "task": t })) + } + + #[tool( + description = "Record what became of an external action you began: `applied` when the destination holds the effect, `not_applied` when it does not. Only while your claim is in force. After that a person reconciles it: you cannot tell a lost answer from an action that was not made." + )] + async fn effect_settled( + &self, + p: Parameters, + ) -> Result { + let p = p.0; + let outcome = parse::("outcome", p.outcome)? + .ok_or_else(|| McpError::invalid_params("outcome: applied or not_applied", None))?; + let t = self + .memory + .effect_settled(&p.id, p.generation, &p.action, outcome) + .await + .map_err(mcp_err)?; + ok(json!({ "task": t })) + } + #[tool( description = "Leave a handoff on a task after your claim on it is over: the lease ran out, or the task was claimed again. Write the handoff first with `write`. It changes nothing of the task; the next holder reads it with get_task." )] diff --git a/docs/demo-narrative.md b/docs/demo-narrative.md index c4411941..3b68c49c 100644 --- a/docs/demo-narrative.md +++ b/docs/demo-narrative.md @@ -226,12 +226,21 @@ ledger refuses it, and gives the two generations. The holder's name does not hel two holders have the same name. The first holder leaves what it learned as a late handoff, which the next holder reads. -**4.4 The rescore.** Each move shows its reason. A task with a higher priority runs +**4.4 Waiting, and an effect that nobody has settled.** The holder of a task gives the +task up until a check completes. The task shows as blocked, and it names the event that +it waits on. The next claim gets a different task, although this task has the highest +priority. Then the event comes, and the task is open again. The next holder records an +external action, a push, before it makes the push. Its lease of one second expires. The +ledger still gives the task to nobody, because nobody knows if the push arrived. A person +looks at the destination and settles the action as not applied. Only then does the next +claim get the task. + +**4.5 The rescore.** Each move shows its reason. A task with a higher priority runs first. One curation task moved from 0 to 10. The reason: curation had done 17% of the completed work, less than its guaranteed 20%. The curation tasks filed at -1 went down to -2, because this period's strategy gives curation a lower weight. -**4.5 Delegation.** The demo asks Claude Code to compare two decisions with subagents. +**4.6 Delegation.** The demo asks Claude Code to compare two decisions with subagents. Claude Code calls its Agent tool twice, once for each subagent. The system stops each call before it runs. It files the subagent's prompt on the ledger as a task. With that task, it files a follow-up task that continues the comparison after the subagent's task @@ -242,14 +251,14 @@ back on the board. The system files what Claude Code wrote as a handoff. The tas there. Nobody may claim it while its follow-up task is open. Thus nobody does the work twice. -**4.6 The delegated work.** Codex on qwen3.8-27b takes each task and files a report. +**4.7 The delegated work.** Codex on qwen3.8-27b takes each task and files a report. -**4.7 Results come back.** A new Claude Code session claims the follow-up task, works on +**4.8 Results come back.** A new Claude Code session claims the follow-up task, works on it with both reports, and completes it. When the follow-up task completes, the original task "Compare two decisions" also closes, with the same handoff. The reason: the follow-up task was the rest of its work. The last line shows it done. -**4.8 Telemetry.** Tasks by kind and status. "gate" means the held writes that wait for +**4.9 Telemetry.** Tasks by kind and status. "gate" means the held writes that wait for you. "promote" means the approvals to verified. ## Act 5 · Watching it diff --git a/docs/terms.md b/docs/terms.md index 6e39dbd0..00af1dbe 100644 --- a/docs/terms.md +++ b/docs/terms.md @@ -63,6 +63,9 @@ term here in the same commit as the first document that needs it. | generation | The number of a claim of a task. Each claim gets the next number. A heartbeat, a completion and a release carry it, and the ledger refuses an earlier one | attempt (for a generation), fencing token | | late handoff | A handoff that a holder leaves on a task after its claim is over | | | attempt key | A key that a caller makes for one claim or one completion. The same key asked again gives the same result, and the ledger does the operation one time | idempotency key, request id | +| outside event | Something that occurs outside the system and that a task waits on: a check that completes, a reply, a change that lands. A tap or a person reports it by name | signal, trigger | +| external action | A change that a holder makes outside the record under its claim. The holder records it on the task before it makes it | side effect | +| settle | To record what became of an external action: the destination holds it, or does not | reconcile, resolve | | task definition | The record in the platform domain for one kind of work: its purpose, its tools, its acceptance checks and its outputs. A task names one as `@` | task type, task template | | quarantine | `state/quarantine`, where the policy holds a write that it refused, for the owner to release or drop | | | owner | The person who holds the root and its grants | user (for the owner), admin | diff --git a/local/README.md b/local/README.md index 985153f4..5f152335 100644 --- a/local/README.md +++ b/local/README.md @@ -624,9 +624,41 @@ completion. The ledger looks for the key in the transaction that makes the claim. The key is in the event of the claim or the completion, so the ledger finds it again after a rebuild. A command has no session, so `amem task` has no attempt key. -Not built yet: an attempt key for the other operations, a block on a new claim while an -external effect is uncertain, and a `blocked` state for a task that waits on an outside -event. +A session does not wait for an outside event, for example a check that runs or a reply. +The holder gives the task up with `wait_for` (`amem task wait `), and names +the event with one token. The task is then `blocked`, and its `waiting_on` field gives +the event. No session can claim it. A tap or a person reports the event with +`amem task arrived `. Each task of the domain that waited on that event is then +`open` again. `amem` refuses an agent that reports an event, because the agent would +decide that its own wait is over. + +A holder records an external action before it makes the action: `effect_begun` +(`amem task effect-begin `). The action id is one token of at most 64 +characters, and a retry of the action gives the same id. The ledger checks the claim in +the transaction that records the action, so it refuses a holder of an earlier claim. +The task's `effects` field lists the actions that are not settled. While the list is not +empty: + +- no session can claim the task, also after the lease expires; +- the holder cannot complete the task. + +`effect_settled` (`amem task effect-settle applied|not_applied`) records +what the destination holds. The holder can settle its action while its claim is in force. +After that, only a person or a tap can settle it. An agent that lost its claim cannot, +because it cannot know if the destination got the action. + +A task also has these fields: `parent`, `scores`, `implements`, `reversibility`, +`artifacts_touched` and `policy_context`. `amem task create` and `create_task` take them, +but not `policy_context`, and `create_task` does not take `scores`. `amem` sets +`policy_context` to the ids of the verified policy entries at the time of the filing. It +sets `parent` for a task that a session files while it holds a task. `amem` refuses a +`parent` that the session cannot read. The fields are a record: no rule reads +`scores`, `implements`, `reversibility` or `artifacts_touched`. + +Not built yet: an attempt key for the other operations, and a connector that puts +external actions in sequence and settles them. `amem` does not make the external action, +so the holder must record it. The taps do not report events yet: a person does, or a +script that runs `amem task arrived`. ## Task definitions: `platform task` diff --git a/scripts/demo/4-fabric.sh b/scripts/demo/4-fabric.sh index f7cb615b..e830b643 100755 --- a/scripts/demo/4-fabric.sh +++ b/scripts/demo/4-fabric.sh @@ -54,6 +54,36 @@ a task late-handoff "$fenced" "$note" --generation "$first" >/dev/null echo " refused. Its handoff is left on the task for the holder now: late-handoff, generation $first" a task release "$fenced" --generation "$second" >/dev/null a task invalidate "$fenced" --reason "obsolete: shown" >/dev/null +say "Waiting, and an effect nobody has settled" "A session does not wait for an outside event. It gives the task up until the event comes, and the tap's event puts the task back on offer. A holder records an external action before it makes it. When its lease then runs out, nobody knows if the destination holds the effect, so the ledger gives the task to nobody until a person settles it." +# claim the next task and say whether it is $1; anything else goes back to the queue +offer() { + got=$(a task claim-next --domain dom-demo --lease-seconds "${2:-600}") + gid=$(printf '%s' "$got" | j "d['id'] if d else ''") + ggen=$(printf '%s' "$got" | j "d['generation'] if d else ''") + if [ "$gid" = "$1" ]; then + echo " claim-next gave the task, at generation $ggen" + elif [ -n "$gid" ]; then + a task release "$gid" --generation "$ggen" >/dev/null + echo " claim-next gave another task, not this one, though it ranks first" + else + echo " claim-next gave nothing" + fi +} +waits=$(a task create "Land the fixture change" --domain dom-demo --priority 60 --reversibility irreversible --touches "fixtures/**" --body "Open the change, then wait for its check." | j "d['id']") +a task admit "$waits" >/dev/null +offer "$waits" +a task wait "$waits" check:fixtures-42 --generation "$ggen" | j "f\" given up by its holder: {d['status']}, waiting on {d['waiting_on']}\"" +offer "$waits" +a task arrived check:fixtures-42 --domain dom-demo | j "f\" the event check:fixtures-42 came: {len(d)} task is {d[0]['status']} again\"" +offer "$waits" 1 +a task effect-begin "$waits" push-fixtures-1 --generation "$ggen" | j "f\" the holder records the action before it makes it: effects {d['effects']}\"" +sleep 2 +echo " the lease of one second ran out, and the holder never said what became of the push" +offer "$waits" +a task effect-settle "$waits" push-fixtures-1 not_applied | j "f\" a person looked at the destination and settled it as not applied: effects {d.get('effects', [])}\"" +offer "$waits" +a task release "$waits" --generation "$ggen" >/dev/null +a task invalidate "$waits" --reason "obsolete: shown" >/dev/null say "The rescore" "Each cycle the queue is ranked again; every move is a ledger event with its reason, and curation's share of the work is held to its floor." a task rescore | j "f\" rubric {d['rubric']}; curation's share of the completed work {d['maintenance_share']:.0%} against a floor of {d['floor']:.0%}\n\" + '\n'.join(f\" {c['from']:>3} -> {c['to']:<3} {c['title']:<40} {c['why']}\" for c in d['changed'])" From 37601fac9060bbb062605f84b0eaef14e8e86572 Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Sun, 15 Nov 2026 08:49:51 +0100 Subject: [PATCH 2/3] fix: hold the work of a task with an unsettled effect under each name The adversarial pass found nine defects. This commit fixes each of them. 1. An unsettled effect held only the claims of its own task. A different agent got the work with `retry_task` or `fan_out`, or closed the task with `invalidate`. Now the ledger files no retry, no continuation and no subagent task of such a task. No session can claim a task that continues or retries it, on SQLite and on PostgreSQL. The service does not invalidate it, and `fan_out` does not close a continuation that has such an effect. 2. A done continuation did not close a task with an unsettled effect, and a later settlement put the work on offer again. Now the settlement of the last effect closes the task with the handoff of that continuation. 3. `wait_for`, `effect_begun` and the reason of `invalidate` put the words of a session on a task below its session label. The service now refuses them. 4. The check of a named `parent` and the `policy_context` stamp had no test that failed without them. Each has one now. 5. A waiting task that goes back to triage had no test. It has one now, on both backends. 6. A second settlement with a different outcome gave no error. The ledger now refuses it, and the record keeps the first outcome. 7. A refused claim of a held task had a wrong reason on the record. The reason now names the event or the effect. 8. `amem work` and `amem gofannon work` stopped on the refusal to complete a task with an unsettled effect. They now release the task with a handoff. The gofannon test showed the defect before the fix. 9. The doc comment of `project` was above `project_hold` in the PostgreSQL backend. Each function has its own comment now. Each assertion failed with its guard removed, one mutation for each run. The ledger tests ran on SQLite and PostgreSQL: - no check in `create`, no `task:` in the origins, no walk up the chain: `no_work_is_filed_from_a_task_whose_effect_is_not_settled` reported "retries: gave Ok(Created ...)". The service test `a_task_with_an_unsettled_effect_is_not_retried_fanned_out_or_closed` failed too. - no `resumes_unsettled` in `claim_next` or in `claim`, for each backend: `a_task_that_resumes_one_with_an_unsettled_effect_is_nobody_s` reported "assertion failed: next(...).is_none()" or "is_wrong_state(&claim(...))". - no effect check in `invalidate`: `a_task_with_an_unsettled_effect_is_not_retried_fanned_out_or_closed`. - no effect check before a fold: `a_continuation_with_an_unsettled_effect_is_not_folded_into_a_new_one`. - no close after a settlement: `a_task_closes_with_its_done_continuation_when_its_last_effect_is_settled` reported left Claimed, right Done. - no label check in `effect_begun`, `wait_for` or `invalidate`: `a_session_that_read_above_a_task_writes_no_words_of_its_own_on_it` failed at the assertion for that call. - no `check_task` on the named parent: `a_filing_names_no_parent_that_the_session_may_not_read`. - an empty `policy_context`, and no stamp: `a_filing_names_the_verified_policy_of_the_record` reported left [] and left ["pol-made-up"]. - no waiting clause in `admit`, and `arrived` that reads only blocked tasks: `a_task_sent_back_to_triage_while_it_waits_still_waits` reported left Open, right Blocked, and one woken task where two were expected. - a different outcome accepted: `a_settled_effect_gives_the_task_back_and_fences_who_began_it`. - no reason from the hold: `a_task_that_waits_on_an_event_is_nobody_s_until_the_event_comes`, `an_effect_not_settled_holds_the_task_past_its_lease` and the test of the resumed task reported the old reason. - a report that completes: `a_report_does_not_complete_a_task_with_an_action_that_nobody_settled`. - a gofannon run that completes: `a_run_that_left_an_action_not_settled_gives_the_task_back_with_its_result` reported "cannot complete while an effect of it is not settled". These limits are open, and `local/README.md` and `PLAN.md` name them. The hold follows the `origin` of a task, so a new task with the same text and no origin is not held. The classification of a task does not go up, so a session that read above its task records no effect and no wait on it. A task under a lease in force stays with its holder when its continuation is done. Co-Authored-By: Claude Fable 5.2 Claude-Session: https://claude.ai/code/session_01D5DyJRpu5ned5y3d92U7Ts --- PLAN.md | 2 +- crates/ledger/src/lib.rs | 416 ++++++++++++++++++++++++++++++++-- crates/ledger/src/postgres.rs | 30 ++- crates/ledger/src/sqlite.rs | 25 ++ crates/memory/src/gofannon.rs | 59 ++++- crates/memory/src/work.rs | 51 ++++- crates/server/src/holds.rs | 248 +++++++++++++++++++- crates/server/src/lib.rs | 87 +++++-- local/README.md | 38 +++- 9 files changed, 895 insertions(+), 61 deletions(-) diff --git a/PLAN.md b/PLAN.md index 60d59a89..ff87103f 100644 --- a/PLAN.md +++ b/PLAN.md @@ -68,7 +68,7 @@ Landed so far: |---|---|---| | 0 | #69 | Nothing | | 1 | #71 | Nothing | -| 2 | #73, attempt keys for a claim and a completion in #77, and waiting, unsettled effects and the remaining fields in this pull request | An attempt key on the other operations; a connector that serialises external actions and reconciles them; taps that report the events tasks wait on; rules that read `scores`, `implements`, `reversibility` and `artifacts_touched` | +| 2 | #73, attempt keys for a claim and a completion in #77, and waiting, unsettled effects and the remaining fields in this pull request | An attempt key on the other operations; a connector that serialises external actions and reconciles them; taps that report the events tasks wait on; rules that read `scores`, `implements`, `reversibility` and `artifacts_touched`; the hold of an unsettled effect follows a task's `origin` only, so the same work filed anew under no origin is not held; a task's classification does not rise to its holder's label, so a session that has read above its task records no effect and no wait on it; a task held under a live lease when its continuation finishes is left to its holder, and is on offer again if that lease runs out | | 3 | #74 | A suite that gates a version of a definition; the data scope enforced; a definition required by default, once the system's own filings name one; tools offered a few at boot and the rest by search | | 6 | #78 | `check_policy` selecting rules by the action; the unused-rule check over one review period; the approver of an exception checked against the layer breached; a new version of a rule keeping its place under the ceiling; a decision that survives an enforced module that fails; the author and the date of an imported rule not trusted; a session that reloads the policy when a rule changes; a verified rule that stays in force while its next version waits for review; a denial that names no policy counted for the module that made it, in the rates and the lint | diff --git a/crates/ledger/src/lib.rs b/crates/ledger/src/lib.rs index 3f0f1c25..6d956ac1 100644 --- a/crates/ledger/src/lib.rs +++ b/crates/ledger/src/lib.rs @@ -926,6 +926,24 @@ pub fn continues(parent: &str) -> String { format!("continues:{parent}") } +/// The task whose work `task` is under another name: the one it continues, or the one +/// it retries. An effect of that task that nobody has settled holds this one too. +pub(crate) fn resumed(task: &Task) -> Option<&str> { + let origin = task.spec.origin.as_deref()?; + origin + .strip_prefix("continues:") + .or_else(|| origin.strip_prefix("retries:")) +} + +/// The task a filing with this origin is filed from: the one it continues, retries, or +/// is a part of. +fn filed_from_of(origin: Option<&str>) -> Option<&str> { + let origin = origin?; + ["continues:", "retries:", "task:"] + .iter() + .find_map(|p| origin.strip_prefix(p)) +} + /// Whether a session may claim `task` now: it is admitted, or claimed under a lease /// that has run out; the session's manifest carries every requirement; and the /// session's handling covers the task's classification. Whether its dependencies are @@ -1106,7 +1124,8 @@ impl Ledger { /// /// # Errors /// - /// A database error. + /// [`Error::WrongState`], naming the task, when the origin files this one from a + /// task whose work is held by an effect that nobody has settled; a database error. pub async fn create(&mut self, spec: TaskSpec) -> Result { if let (Some(o), Some(u)) = (&spec.origin, &spec.upstream_event) { let existing = on_backend!(&self.backend, db => db.find_origin(&spec.domain, o, u))?; @@ -1117,6 +1136,17 @@ impl Ledger { }); } } + // the work of a task with an effect that nobody has settled goes to nobody else, + // under another task's name either: no continuation, retry or part of it is filed + if let Some(from) = filed_from_of(spec.origin.as_deref()) { + if let Some(unsettled) = self.unsettled_above(from).await? { + return Err(Error::WrongState { + id: unsettled.id, + status: unsettled.status, + op: "hand its work on while an effect of it is not settled", + }); + } + } let id = dm_core::id::new_task_id(&spec.domain); let now = self.now(); let event = Event::Created { @@ -1130,6 +1160,29 @@ impl Ledger { }) } + /// The task whose unsettled effect holds the work of `id`: `id` itself, or a task + /// that `id` continues or retries, however far up. A task this ledger does not hold + /// ends the walk. + async fn unsettled_above(&self, id: &str) -> Result, Error> { + let mut seen = std::collections::HashSet::new(); + let mut up = Some(id.to_string()); + while let Some(id) = up.take() { + if !seen.insert(id.clone()) { + break; + } + let task = match self.get(&id).await { + Ok(t) => t, + Err(Error::NotFound(_)) => break, + Err(e) => return Err(e), + }; + if !task.effects.is_empty() { + return Ok(Some(task)); + } + up = resumed(&task).map(str::to_string); + } + Ok(None) + } + /// Admit a proposed task, making it claimable. /// /// # Errors @@ -1356,6 +1409,11 @@ impl Ledger { ) } else if task.status == Status::Proposed { "the task was not admitted".to_string() + } else if let Some(held_back) = self.why_held_back(&task).await? { + format!( + "after the claim of generation {}, {held_back}", + task.generation + ) } else { // claimable by someone, and still not this session's to take format!( @@ -1374,6 +1432,28 @@ impl Ledger { self.append_at(id, now, &refused).await } + /// What holds `task` back from every claim, in words for its record: the event it + /// waits on, an effect of it that nobody has settled, or such an effect of the task + /// whose work it resumes. + async fn why_held_back(&self, task: &Task) -> Result, Error> { + if let Some(on) = &task.waiting_on { + return Ok(Some(format!("the task waited on {on}"))); + } + if let Some(action) = task.effects.first() { + return Ok(Some(format!("effect {action} of the task was not settled"))); + } + let Some(up) = resumed(task) else { + return Ok(None); + }; + Ok(self.unsettled_above(up).await?.map(|t| { + format!( + "effect {} of {}, whose work the task resumes, was not settled", + t.effects.first().map_or("", String::as_str), + t.id + ) + })) + } + /// Renew a lease. Refused when the caller does not hold the task or the lease has /// already expired, in which case the task is back in the pool and the caller must /// stop. Returns the task so a caller can see an invalidation. @@ -1638,12 +1718,15 @@ impl Ledger { /// `held`, it is the holder that says so, and the claim is checked in the /// transaction that records it. Without, it is someone who reconciles after the /// claim: whether `by` may is the caller's rule, and the ledger records who did. - /// Asked again for an action already settled, it changes nothing. + /// Asked again for an action already settled with the same outcome, it changes + /// nothing. With the last effect settled, a task whose continuation finished + /// meanwhile closes, as [`Ledger::close_resumed`] closes it. /// /// # Errors /// - /// [`Error::NotFound`]; [`Error::WrongState`] for an action the task never began; - /// with `held`, [`Error::NotHolder`] or [`Error::Fenced`]. + /// [`Error::NotFound`]; [`Error::WrongState`] for an action the task never began, + /// or one already settled with the other outcome; with `held`, + /// [`Error::NotHolder`] or [`Error::Fenced`]. pub async fn effect_settled( &mut self, id: &str, @@ -1652,31 +1735,91 @@ impl Ledger { (by, held): (&str, Option>), ) -> Result { let task = self.get(id).await?; - if task.effects.iter().any(|a| a == action) { - let settled = Event::EffectSettled { - action: action.to_string(), - outcome, - by: by.to_string(), - }; - match held { - Some(held) => { - let now = self.now(); - self.append_held(id, now, held, &settled).await?; - } - None => self.append(id, &settled).await?, - } - return self.get(id).await; + if !task.effects.iter().any(|a| a == action) { + return self.settled_before(task, action, outcome).await; } - if self.settlement(id, action).await?.is_some() { - return Ok(task); + let settled = Event::EffectSettled { + action: action.to_string(), + outcome, + by: by.to_string(), + }; + match held { + Some(held) => { + let now = self.now(); + self.append_held(id, now, held, &settled).await?; + } + None => self.append(id, &settled).await?, } + self.close_after_settlement(id).await?; + self.get(id).await + } + + /// The answer for an action that is not pending. Settled before with this outcome, + /// it is the same settlement, and the task comes back as it is. Settled with the + /// other outcome, the record stays as it is and the caller is told: an answer of + /// "taken" would say that the record now holds what it does not. + async fn settled_before( + &self, + task: Task, + action: &str, + outcome: Outcome, + ) -> Result { + let op = match self.settlement(&task.id, action).await? { + Some((was, _)) if was == outcome => return Ok(task), + Some(_) => "settle with another outcome an effect that is settled", + None => "settle an effect the task did not begin", + }; Err(Error::WrongState { - id: id.to_string(), + id: task.id, status: task.status, - op: "settle an effect the task did not begin", + op, }) } + /// A completion of a continuation leaves the task it resumed open while an effect of + /// that task is not settled. With the last effect settled, the task closes with the + /// continuation that finished it. Otherwise its work would be on offer a second time. + async fn close_after_settlement(&mut self, id: &str) -> Result<(), Error> { + let task = self.get(id).await?; + if !task.effects.is_empty() || task.status.is_terminal() { + return Ok(()); + } + let marker = continues(id); + let finished = self.filed_from(id).await?.into_iter().rev().find(|t| { + t.status == Status::Done && t.spec.origin.as_deref() == Some(marker.as_str()) + }); + if let Some(continuation) = finished { + self.close_resumed(&continuation).await; + } + Ok(()) + } + + /// A continuation that is done finishes the task it resumed, and that one the task + /// it resumed in turn: up the chain, with the continuation's handoff, while each is + /// open, nobody holds it and every effect of it is settled. The chain stops at the + /// first task that does not close. + pub async fn close_resumed(&mut self, done: &Task) { + let Some(handoff) = done.handoff.clone() else { + return; + }; + let mut by = done.clone(); + while let Some(parent) = by + .spec + .origin + .as_deref() + .and_then(|o| o.strip_prefix("continues:")) + .map(str::to_string) + { + let Ok(closed) = self + .complete_by_continuation(&parent, &by.id, &handoff) + .await + else { + break; + }; + by = closed; + } + } + /// How an action of a task was settled, and by whom, when it was. /// /// # Errors @@ -2340,6 +2483,10 @@ mod tests { an_effect_not_settled_holds_the_task_past_its_lease, a_settled_effect_gives_the_task_back_and_fences_who_began_it, a_task_keeps_the_rest_of_its_record, + a_task_sent_back_to_triage_while_it_waits_still_waits, + no_work_is_filed_from_a_task_whose_effect_is_not_settled, + a_task_that_resumes_one_with_an_unsettled_effect_is_nobody_s, + a_task_closes_with_its_done_continuation_when_its_last_effect_is_settled, a_task_waits_on_its_continuation, a_task_closes_with_its_continuation, a_continuation_given_up_on_gives_its_task_back, @@ -4626,6 +4773,14 @@ mod tests { assert!(next(&mut l, "s2", mins(30)).await.is_none()); *now.lock().unwrap() = T0 + mins(90); nobody_gets_it(&mut l, &t.id).await; + // and its record says what the claim lost to + assert_eq!( + last_refusal(&l, &t.id).await, + ( + "s2".to_string(), + "after the claim of generation 1, the task waited on check:42".to_string() + ) + ); it_waits_through_a_replay(&mut l, &t.id, by).await; only_its_event_wakes_it(&mut l, &t).await; the_woken_task_is_claimed_again(&mut l, &t).await; @@ -4655,6 +4810,11 @@ mod tests { assert!(is_wrong_state(&claim(l, id, "s2", mins(30)).await)); } + /// The last refused claim on the task's record: who asked, and why not. + async fn last_refusal(l: &Ledger, id: &str) -> (String, String) { + l.refused_claims(id).await.unwrap().pop().unwrap() + } + async fn only_its_event_wakes_it(l: &mut Ledger, t: &Task) { assert!(l.arrived("dom-a", "check:41").await.unwrap().is_empty()); assert!(l.arrived("dom-b", "check:42").await.unwrap().is_empty()); @@ -4699,11 +4859,20 @@ mod tests { .unwrap(); assert_eq!(own.effects, vec!["act-1".to_string()]); *now.lock().unwrap() = T0 + mins(31); - // the lease ran out, and that alone gives the task to nobody - nobody_gets_it(&mut l, &t.id).await; + the_lease_that_ran_out_gives_it_to_nobody(&mut l, &t.id).await; the_effect_holds_through_a_replay(&mut l, &t.id, by).await; } + /// The lease ran out, and that alone gives the task to nobody. Its record says what + /// the claim lost to. + async fn the_lease_that_ran_out_gives_it_to_nobody(l: &mut Ledger, id: &str) { + nobody_gets_it(l, id).await; + assert_eq!( + last_refusal(l, id).await.1, + "after the claim of generation 1, effect act-1 of the task was not settled" + ); + } + async fn the_effect_holds_through_a_replay(l: &mut Ledger, id: &str, by: Held<'_>) { let held_back = l.get(id).await.unwrap().held_back(); assert_eq!(held_back.as_deref(), Some("effect act-1 is not settled")); @@ -4750,10 +4919,19 @@ mod tests { async fn settled_once_and_claimed_again(l: &mut Ledger, t: &Task, by: Held<'_>) { // said again, it is the same settlement and no second event let events = l.event_count().await.unwrap(); - l.effect_settled(&t.id, "act-1", Outcome::Applied, ("human:a:y", None)) + l.effect_settled(&t.id, "act-1", Outcome::NotApplied, ("human:a:y", None)) .await .unwrap(); + // said again with the other outcome, it is refused: the record keeps the first + let other = l + .effect_settled(&t.id, "act-1", Outcome::Applied, ("human:a:y", None)) + .await; + assert!(is_wrong_state(&other), "{other:?}"); assert_eq!(l.event_count().await.unwrap(), events); + assert_eq!( + l.settlement(&t.id, "act-1").await.unwrap(), + Some((Outcome::NotApplied, "human:a:x".to_string())) + ); let c2 = next(l, "s2", mins(30)).await.unwrap(); assert_eq!((c2.task.id.as_str(), c2.generation), (t.id.as_str(), 2)); // the delayed request of the first holder is refused where the effect is recorded @@ -4761,6 +4939,194 @@ mod tests { assert!(l.get(&t.id).await.unwrap().effects.is_empty()); } + /// A task that `s1` claimed and gave up for `check:42`, then sent back to triage. No + /// call sends a waiting task back today: a task is claimed only when each task it + /// depends on is done, and a done task is not invalidated. The events can hold it + /// all the same, so the test appends the event. + async fn parked_and_sent_back(l: &mut Ledger, title: &str) -> Task { + let t = admitted(l, spec(title)).await; + let c = claim(l, &t.id, "s1", mins(30)).await.unwrap(); + l.wait_for(&t.id, held("s1", c.generation), "check:42", None) + .await + .unwrap(); + let because = "task-a-gone".to_string(); + l.append(&t.id, &Event::Retriaged { because }) + .await + .unwrap(); + let back = l.get(&t.id).await.unwrap(); + assert_eq!(back.status, Status::Proposed); + assert_eq!(back.waiting_on.as_deref(), Some("check:42")); + back + } + + async fn a_task_sent_back_to_triage_while_it_waits_still_waits(b: &Site) { + let (mut l, _) = b.ledger().await; + // admitted again before the event: it shows as blocked, and it is nobody's + let first = parked_and_sent_back(&mut l, "Admitted before the event").await; + let again = l.admit(&first.id).await.unwrap(); + assert_eq!(again.status, Status::Blocked); + assert_eq!(again.waiting_on.as_deref(), Some("check:42")); + nobody_gets_it(&mut l, &first.id).await; + // still in triage when the event comes: the event reaches it there + let second = parked_and_sent_back(&mut l, "Admitted after the event").await; + // the blocked task first, then the proposed one: the order in which arrived looks + let woken = l.arrived("dom-a", "check:42").await.unwrap(); + let seen: Vec<_> = woken + .iter() + .map(|t| (t.id.as_str(), t.status, t.waiting_on.as_deref())) + .collect(); + assert_eq!( + seen, + vec![ + (first.id.as_str(), Status::Open, None), + (second.id.as_str(), Status::Proposed, None) + ] + ); + assert_eq!(l.admit(&second.id).await.unwrap().status, Status::Open); + for id in [&first.id, &second.id] { + let c = claim(&mut l, id, "s2", mins(30)).await.unwrap(); + assert_eq!(c.generation, 2); + } + } + + /// A task filed from `from` under `kind`: `continues`, `retries` or `task`. + fn filed_from(kind: &str, from: &str, title: &str) -> TaskSpec { + TaskSpec { + origin: Some(format!("{kind}:{from}")), + ..spec(title) + } + } + + async fn no_work_is_filed_from_a_task_whose_effect_is_not_settled(b: &Site) { + let (mut l, _) = b.ledger().await; + let t = admitted(&mut l, spec("Post the release note")).await; + let c = claim(&mut l, &t.id, "s1", mins(30)).await.unwrap(); + let by = held("s1", c.generation); + let resumes = l + .create(filed_from("continues", &t.id, "Resume")) + .await + .unwrap() + .task; + l.effect_begun(&t.id, by, "act-1").await.unwrap(); + l.release(&t.id, by, None).await.unwrap(); + nothing_is_filed_from(&mut l, &t.id, &resumes.id).await; + // a task filed from nothing this ledger holds is filed as before + l.create(filed_from("task", "task-a-nowhere", "Elsewhere")) + .await + .unwrap(); + l.effect_settled(&t.id, "act-1", Outcome::NotApplied, ("human:a:x", None)) + .await + .unwrap(); + let retry = l.create(filed_from("retries", &t.id, "Again")).await; + assert_eq!(retry.unwrap().task.status, Status::Proposed); + } + + /// No retry, no continuation and no part of `id` is filed, nor a retry of the task + /// that resumes it. Each refusal names `id`, and none leaves an event. + async fn nothing_is_filed_from(l: &mut Ledger, id: &str, resumes: &str) { + let before = l.event_count().await.unwrap(); + for (kind, from) in [ + ("retries", id), + ("continues", id), + ("task", id), + ("retries", resumes), + ] { + let filed = l.create(filed_from(kind, from, "Again")).await; + assert!( + matches!(&filed, Err(Error::WrongState { id: held, .. }) if held == id), + "{kind}:{from} gave {filed:?}" + ); + } + assert_eq!(l.event_count().await.unwrap(), before); + } + + async fn a_task_that_resumes_one_with_an_unsettled_effect_is_nobody_s(b: &Site) { + let (mut l, _) = b.ledger().await; + let t = admitted(&mut l, spec("Post the release note")).await; + let c = claim(&mut l, &t.id, "s1", mins(30)).await.unwrap(); + let by = held("s1", c.generation); + // filed before the effect began, so the filing was not refused + let resumes = admitted(&mut l, filed_from("continues", &t.id, "Resume")).await; + let again = admitted(&mut l, filed_from("retries", &resumes.id, "Again")).await; + l.effect_begun(&t.id, by, "act-1").await.unwrap(); + l.release(&t.id, by, None).await.unwrap(); + neither_is_claimed(&mut l, &t.id, [&resumes.id, &again.id]).await; + l.effect_settled(&t.id, "act-1", Outcome::Applied, ("human:a:x", None)) + .await + .unwrap(); + let c2 = claim(&mut l, &resumes.id, "s2", mins(30)).await.unwrap(); + assert_eq!(c2.generation, 1); + } + + /// Nobody gets the task that resumes `id`, or the retry of that task, by the next + /// claim or by name. The record of the first says which effect the claim lost to. + async fn neither_is_claimed(l: &mut Ledger, id: &str, resuming: [&String; 2]) { + assert!(next(l, "s2", mins(30)).await.is_none()); + for r in resuming { + assert!(is_wrong_state(&claim(l, r, "s2", mins(30)).await)); + } + assert_eq!( + last_refusal(l, resuming[0]).await.1, + format!( + "after the claim of generation 0, effect act-1 of {id}, whose work the task resumes, was not settled" + ) + ); + } + + async fn a_task_closes_with_its_done_continuation_when_its_last_effect_is_settled(b: &Site) { + let (mut l, now) = b.ledger().await; + // a root that was given up, the task that resumes it, and the task that resumes that + let root = given_up(&mut l, "Ship the release").await; + let t = admitted(&mut l, filed_from("continues", &root.id, "Resume")).await; + let c = claim(&mut l, &t.id, "s1", mins(30)).await.unwrap(); + let resumes = admitted(&mut l, filed_from("continues", &t.id, "Resume")).await; + let c2 = claim(&mut l, &resumes.id, "s2", mins(120)).await.unwrap(); + // its holder begins an effect while the continuation runs, and its lease runs out + l.effect_begun(&t.id, held("s1", c.generation), "act-1") + .await + .unwrap(); + *now.lock().unwrap() = T0 + mins(31); + let done = l + .complete(&resumes.id, held("s2", c2.generation), "mem-a-h2") + .await + .unwrap(); + l.close_resumed(&done).await; + the_chain_stays_open_and_nobody_s(&mut l, &t.id, &root.id).await; + the_settlement_closes_the_chain(&mut l, &t.id, &root.id).await; + } + + /// Not done while nobody knows what the destination holds, and nobody's meanwhile. + async fn the_chain_stays_open_and_nobody_s(l: &mut Ledger, id: &str, root: &str) { + let left = get_all(l, &[id, root]).await; + assert_eq!(left[0].status, Status::Claimed); + assert_eq!(left[1].status, Status::Open); + assert!(next(l, "s3", mins(30)).await.is_none()); + } + + /// A task that `s0` claimed and gave back. + async fn given_up(l: &mut Ledger, title: &str) -> Task { + let t = admitted(l, spec(title)).await; + let c = claim(l, &t.id, "s0", mins(30)).await.unwrap(); + l.release(&t.id, held("s0", c.generation), None) + .await + .unwrap() + } + + /// With `act-1` of `id` settled, `id` and `root` are done with the handoff of the + /// continuation. The work was finished once, so nothing is on offer again. + async fn the_settlement_closes_the_chain(l: &mut Ledger, id: &str, root: &str) { + let settled = l + .effect_settled(id, "act-1", Outcome::Applied, ("human:a:x", None)) + .await + .unwrap(); + assert_eq!(settled.status, Status::Done); + assert_eq!(settled.handoff.as_deref(), Some("mem-a-h2")); + let root = l.get(root).await.unwrap(); + assert_eq!(root.status, Status::Done); + assert_eq!(root.handoff.as_deref(), Some("mem-a-h2")); + assert!(next(l, "s3", mins(30)).await.is_none()); + } + async fn a_task_keeps_the_rest_of_its_record(b: &Site) { let (mut l, _) = b.ledger().await; let mut s = spec("Carry out the decision"); diff --git a/crates/ledger/src/postgres.rs b/crates/ledger/src/postgres.rs index add7c926..f0148d59 100644 --- a/crates/ledger/src/postgres.rs +++ b/crates/ledger/src/postgres.rs @@ -726,6 +726,7 @@ impl Db { if !claimable(&task, now, manifest, handling) || !deps_complete(&tx, &task.spec.depends_on).await? || awaits_continuation(&tx, &s, &task.id).await? + || resumes_unsettled(&tx, &task).await? { return Err(Error::WrongState { id: id.to_string(), @@ -942,6 +943,7 @@ async fn take( if !claimable(&task, now, manifest, handling) || !deps_complete(tx, &task.spec.depends_on).await? || awaits_continuation(tx, s, &task.id).await? + || resumes_unsettled(tx, &task).await? { return Ok(false); } @@ -1087,6 +1089,32 @@ async fn awaits_continuation(conn: &C, s: &str, id: &str) -> R .is_some()) } +/// Whether the work of `task` is held by an effect that nobody has settled: an effect of +/// the task it continues or retries, however far up and in whichever domain. Such an +/// effect holds the work under each of its names, so the task is nobody's to claim. +async fn resumes_unsettled(conn: &C, task: &Task) -> Result { + let mut seen = std::collections::HashSet::new(); + let mut up = crate::resumed(task).map(str::to_string); + while let Some(id) = up.take() { + if !seen.insert(id.clone()) { + break; + } + let Some(s) = schema_of_task_in(conn, &id).await? else { + break; + }; + let above = match get_in(conn, &s, &id).await { + Ok(t) => t, + Err(Error::NotFound(_)) => break, + Err(e) => return Err(e), + }; + if !above.effects.is_empty() { + return Ok(true); + } + up = crate::resumed(&above).map(str::to_string); + } + Ok(false) +} + /// Drop one domain's projection and replay its events into it. async fn rebuild_domain(conn: &C, domain: &str, s: &str) -> Result { // a schema an older binary made after this process connected has no promotions table @@ -1335,7 +1363,6 @@ async fn insert_promotion( Ok(()) } -/// Update the projection for one event. Replay calls this and nothing else. /// The projection of the events that hold a task back from a claim, or stop doing so: /// what it waits on and which effects are not settled live in `extra`. async fn project_hold( @@ -1371,6 +1398,7 @@ async fn project_hold( Ok(()) } +/// Update the projection for one event. Replay calls this and nothing else. async fn project( conn: &C, s: &str, diff --git a/crates/ledger/src/sqlite.rs b/crates/ledger/src/sqlite.rs index 26f5f72a..8b7633a6 100644 --- a/crates/ledger/src/sqlite.rs +++ b/crates/ledger/src/sqlite.rs @@ -435,6 +435,7 @@ impl Db { if !claimable(&task, now, manifest, handling) || !deps_complete(&tx, &task.spec.depends_on)? || awaits_continuation(&tx, &task)? + || resumes_unsettled(&tx, &task)? { continue; } @@ -470,6 +471,7 @@ impl Db { if !claimable(&task, now, manifest, handling) || !deps_complete(&tx, &task.spec.depends_on)? || awaits_continuation(&tx, &task)? + || resumes_unsettled(&tx, &task)? { return Err(Error::WrongState { id: id.to_string(), @@ -794,6 +796,29 @@ fn awaits_continuation(conn: &Connection, task: &Task) -> Result { .is_some()) } +/// Whether the work of `task` is held by an effect that nobody has settled: an effect of +/// the task it continues or retries, however far up. Such an effect holds the work +/// under each of its names, so the task is nobody's to claim. +fn resumes_unsettled(conn: &Connection, task: &Task) -> Result { + let mut seen = std::collections::HashSet::new(); + let mut up = crate::resumed(task).map(str::to_string); + while let Some(id) = up.take() { + if !seen.insert(id.clone()) { + break; + } + let above = match get_in(conn, &id) { + Ok(t) => t, + Err(Error::NotFound(_)) => break, + Err(e) => return Err(e), + }; + if !above.effects.is_empty() { + return Ok(true); + } + up = crate::resumed(&above).map(str::to_string); + } + Ok(false) +} + /// Append the event and project it, in the caller's transaction. fn apply(conn: &Connection, id: &str, at: OffsetDateTime, event: &Event) -> Result<(), Error> { conn.execute( diff --git a/crates/memory/src/gofannon.rs b/crates/memory/src/gofannon.rs index d1d81fa2..78770efc 100644 --- a/crates/memory/src/gofannon.rs +++ b/crates/memory/src/gofannon.rs @@ -191,7 +191,8 @@ async fn run(m: &Arc, plan: &Plan, task: &dm_client::Task) -> Result>, refuse: bool, + /// An external action the run begins under the task's claim and does not settle. + begins: Option<&'static str>, } impl Tools for Scripted { fn call( &self, - _m: Arc, + m: Arc, _tool: String, _config: String, args: String, @@ -271,6 +288,7 @@ mod tests { c.len() }; let refuse = self.refuse; + let begins = self.begins; let answered = |v: Value| { Ok(Answered { answer: v.to_string(), @@ -282,7 +300,14 @@ mod tests { Some("start") if refuse => { Err("127.0.0.1: not a host the gofannon tool is granted".into()) } - Some("start") => answered(json!({"run_id": "run-1"})), + Some("start") => { + if let (Some(action), Some(id)) = + (begins, args["input"]["task_id"].as_str()) + { + m.effect_begun(id, None, action).await.unwrap(); + } + answered(json!({"run_id": "run-1"})) + } _ if n == 2 => answered(json!({"run_id": "run-1", "status": "running"})), _ => answered( json!({"run_id": "run-1", "status": "completed", "result": {"label": "bug"}}), @@ -293,6 +318,14 @@ mod tests { } fn root(dir: &Path, refuse: bool) -> (Arc, Arc) { + root_with(dir, refuse, None) + } + + fn root_with( + dir: &Path, + refuse: bool, + begins: Option<&'static str>, + ) -> (Arc, Arc) { Memory::init(dir, "human:test:me", "dom-me").unwrap(); let path = dir.join(dm_server::GRANTS_FILE); let mut g = dm_policy::Grants::load(&path).unwrap(); @@ -311,6 +344,7 @@ mod tests { let s = Arc::new(Scripted { calls: Mutex::new(Vec::new()), refuse, + begins, }); m.set_tools(s.clone()); (m, s) @@ -391,4 +425,21 @@ mod tests { h.body ); } + + /// The ledger refuses to complete a task with an action that nobody has settled. The + /// worker does not stop on that refusal: the task goes back with the run's result. + #[tokio::test] + async fn a_run_that_left_an_action_not_settled_gives_the_task_back_with_its_result() { + let dir = tempfile::tempdir().unwrap(); + let (m, _) = root_with(dir.path(), false, Some("post-1")); + let owner = Memory::open(dir.path()).unwrap(); + let id = filed(&owner).await; + let worked = work(&m, &plan()).await.map_err(|e| e.to_string()); + assert_eq!(worked, Ok(1)); + let t = owner.get_task(&id).await.unwrap(); + assert_eq!(t.status, dm_ledger::Status::Open); + assert_eq!(t.effects, vec!["post-1".to_string()]); + let h = owner.get(&t.handoff.unwrap()).await.unwrap(); + assert!(h.body.contains("\"label\": \"bug\""), "{}", h.body); + } } diff --git a/crates/memory/src/work.rs b/crates/memory/src/work.rs index 2c6a392a..3aa0fc4f 100644 --- a/crates/memory/src/work.rs +++ b/crates/memory/src/work.rs @@ -430,7 +430,7 @@ async fn settle( ); Ok(0) } - "claimed" if delegated_report(task, code, &output) => { + "claimed" if completes_with_report(task, &now, code, &output) => { let handoff = client .write_handoff( &format!("Report: {}", task.title), @@ -453,10 +453,19 @@ async fn settle( } "claimed" => { let tail = tail_of(&output, 6000); + let held = unsettled(&now) + .map(|action| { + format!( + " External action `{action}` of the task is not settled: nobody can \ + claim the task until a person or a tap settles it \ + (`amem task effect-settle`)." + ) + }) + .unwrap_or_default(); let body = format!( "# {title}\n\n`{harness}` stopped without completing task {id} (exit {code}). \ The task is back in the pool; what follows is what it printed, so the next \ - claimant can start from here.\n\n## Output\n\n{tail}\n", + claimant can start from here.{held}\n\n## Output\n\n{tail}\n", title = task.title, harness = harness.binary(), id = task.id, @@ -496,6 +505,20 @@ fn delegated_report(task: &Task, code: i32, output: &str) -> bool { .is_some_and(|o| o.starts_with("task:")) } +/// Whether a run's report completes its task: the report of a delegated task, `task` as +/// it was claimed, when `now`, the task as it is, has no external action that nobody +/// has settled. +fn completes_with_report(task: &Task, now: &Task, code: i32, output: &str) -> bool { + delegated_report(task, code, output) && unsettled(now).is_none() +} + +/// The first external action of a task that nobody has settled, when there is one. The +/// ledger refuses to complete such a task, whatever its run reported: the task goes +/// back with a handoff, and waits for whoever settles the action. +pub(crate) fn unsettled(task: &Task) -> Option<&str> { + task.raw.get("effects")?.as_array()?.first()?.as_str() +} + fn tail_of(s: &str, chars: usize) -> String { let s = s.trim(); if s.is_empty() { @@ -809,6 +832,30 @@ mod tests { assert!(!delegated_report(&task(None), 0, "done")); } + #[test] + fn a_report_does_not_complete_a_task_with_an_action_that_nobody_settled() { + let task = |effects: serde_json::Value| Task { + id: "task-a-1".into(), + title: "t".into(), + body: "b".into(), + domain: "dom-a".into(), + status: "claimed".into(), + depends_on: Vec::new(), + requirements: Vec::new(), + handoff: None, + raw: serde_json::json!({ "origin": "task:task-a-0", "effects": effects }), + }; + let claimed = task(serde_json::Value::Null); + let report = "DR-19 is proposed."; + let settled = task(serde_json::json!([])); + assert!(completes_with_report(&claimed, &claimed, 0, report)); + assert!(completes_with_report(&claimed, &settled, 0, report)); + // the ledger would refuse the completion: the task goes back with a handoff + let begun = task(serde_json::json!(["push-1", "push-2"])); + assert_eq!(unsettled(&begun), Some("push-1")); + assert!(!completes_with_report(&claimed, &begun, 0, report)); + } + use super::*; #[test] diff --git a/crates/server/src/holds.rs b/crates/server/src/holds.rs index 599accdb..6f6f1751 100644 --- a/crates/server/src/holds.rs +++ b/crates/server/src/holds.rs @@ -21,7 +21,28 @@ fn token(what: &str, name: &str, most: usize) -> Result<(), Error> { } } +/// Why a task is not closed, folded or handed on: an effect of it that nobody has settled. +pub(crate) fn unsettled(task: &Task) -> Option { + task.effects + .first() + .map(|a| format!("effect {a} of {} is not settled", task.id)) +} + impl Memory { + /// What a session writes on a task in its own words, every session that may read + /// the task reads. A session whose label is above the task's classification has + /// read what those sessions may not, so it writes no such words there. A filing + /// rises to the session's label instead; a task that is filed stays where it is. + pub(crate) fn writes_within(&self, task: &Task, what: &str) -> Result<(), Error> { + if self.label() <= task.spec.classification { + return Ok(()); + } + Err(Error::Denied(format!( + "this session has read above the classification of {}: it does not write {what} on that task", + task.id + ))) + } + /// Give a held task up until an outside event comes: a check that runs, a reply, a /// change that has not landed. Nobody may claim it until a tap or a person reports /// the event with [`Memory::event_arrived`]. The handoff, when given, must exist @@ -29,8 +50,9 @@ impl Memory { /// /// # Errors /// - /// [`Error::Denied`] for an event name that is not one token; a ledger error, - /// including a lease already lost; a task outside the grants. + /// [`Error::Denied`] for an event name that is not one token, or from a session + /// whose label is above the task's classification; a ledger error, including a + /// lease already lost; a task outside the grants. pub async fn wait_for( &self, id: &str, @@ -43,7 +65,9 @@ impl Memory { self.get(h).await?; } let mut l = self.ledger.lock().await; - self.check_task(&l.get(id).await?)?; + let task = l.get(id).await?; + self.check_task(&task)?; + self.writes_within(&task, "an event name")?; let by = self.under(self.claim_of(id, generation)?); let t = l.wait_for(id, by, on, handoff).await?; lock(&self.claims).remove(id); @@ -82,8 +106,9 @@ impl Memory { /// /// # Errors /// - /// [`Error::Denied`] for an action id that is not one token; a ledger error, - /// including a lease already lost or a claim made since; a task outside the grants. + /// [`Error::Denied`] for an action id that is not one token, or from a session + /// whose label is above the task's classification; a ledger error, including a + /// lease already lost or a claim made since; a task outside the grants. pub async fn effect_begun( &self, id: &str, @@ -92,7 +117,9 @@ impl Memory { ) -> Result { token("an action id", action, 64)?; let mut l = self.ledger.lock().await; - self.check_task(&l.get(id).await?)?; + let task = l.get(id).await?; + self.check_task(&task)?; + self.writes_within(&task, "an action id")?; let by = self.under(self.claim_of(id, generation)?); let t = l.effect_begun(id, by, action).await?; self.observe_task(&t); @@ -139,6 +166,7 @@ impl Memory { #[cfg(test)] mod tests { use super::*; + use dm_core::Label; use dm_ledger::{Status, TaskSpec}; use std::path::Path; @@ -252,6 +280,214 @@ mod tests { assert!(reconciled.effects.is_empty()); } + fn not_handed_on(r: &Result, id: &str) -> bool { + matches!( + r, + Err(Error::Ledger(dm_ledger::Error::WrongState { id: held, op, .. })) + if held == id && op.contains("hand its work on") + ) + } + + fn part() -> Vec<(String, String)> { + vec![("A part".to_string(), "Do the part.\n".to_string())] + } + + #[tokio::test] + async fn a_task_with_an_unsettled_effect_is_not_retried_fanned_out_or_closed() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let (agent, id) = held(&owner, dir.path(), "Push the release").await; + agent.effect_begun(&id, None, "push-1").await.unwrap(); + agent.release(&id, None, None).await.unwrap(); + // another agent does not get the work under the task's name, or under another + let other = as_author(dir.path(), "agent:b"); + assert!(other.claim(&id, time::Duration::minutes(5)).await.is_err()); + let retried = other.retry_task(&id).await; + assert!(not_handed_on(&retried, &id), "{retried:?}"); + let fanned = other.fan_out(&id, &part()).await; + assert!(not_handed_on(&fanned, &id), "{fanned:?}"); + // and nobody closes it while nobody knows what the destination holds + for who in [&other, &owner] { + let closed = who.invalidate(&id, "obsolete", None).await; + assert!(denied(&closed, "effect push-1"), "{closed:?}"); + } + let tree = owner.task_tree(&id).await.unwrap(); + assert_eq!(tree.task.status, Status::Open); + assert_eq!(tree.task.effects, vec!["push-1".to_string()]); + assert!(tree.children.is_empty()); + // a person looked at the destination: the work is on offer again + owner + .effect_settled(&id, None, "push-1", Outcome::NotApplied) + .await + .unwrap(); + the_retry_holds_its_task_open_in_turn(&owner, &other, &id).await; + } + + /// The retry of `id` begins an effect and gives up: `id` is not closed over it. + async fn the_retry_holds_its_task_open_in_turn(owner: &Memory, other: &Memory, id: &str) { + let retry = retried_with_an_effect(other, id).await; + let closed = owner.invalidate(id, "obsolete", None).await; + assert!(denied(&closed, "effect push-2"), "{closed:?}"); + assert_eq!(owner.get_task(id).await.unwrap().status, Status::Open); + owner + .effect_settled(&retry, None, "push-2", Outcome::Applied) + .await + .unwrap(); + let closed = owner.invalidate(id, "obsolete", None).await.unwrap(); + assert_eq!(closed.status, Status::Invalidated); + let below = owner.get_task(&retry).await.unwrap(); + assert_eq!(below.status, Status::Invalidated); + } + + /// A retry of `id` that `other` filed, claimed and gave up with `push-2` begun. + async fn retried_with_an_effect(other: &Memory, id: &str) -> String { + let retry = other.retry_task(id).await.unwrap(); + assert_eq!(retry.status, Status::Open); + other + .claim(&retry.id, time::Duration::minutes(5)) + .await + .unwrap(); + other.effect_begun(&retry.id, None, "push-2").await.unwrap(); + other.release(&retry.id, None, None).await.unwrap(); + retry.id + } + + #[tokio::test] + async fn a_continuation_with_an_unsettled_effect_is_not_folded_into_a_new_one() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let (agent, id) = held(&owner, dir.path(), "Push the release").await; + let resumes = owner + .create_task(TaskSpec { + title: "Resume: Push the release".into(), + body: "Do it.\n".into(), + origin: Some(dm_ledger::continues(&id)), + ..TaskSpec::default() + }) + .await + .unwrap() + .task; + owner.admit(&resumes.id).await.unwrap(); + let other = as_author(dir.path(), "agent:b"); + other + .claim(&resumes.id, time::Duration::minutes(5)) + .await + .unwrap(); + other + .effect_begun(&resumes.id, None, "push-1") + .await + .unwrap(); + other.release(&resumes.id, None, None).await.unwrap(); + // a new fan-out would close the continuation and give its work to another + let fanned = agent.fan_out(&id, &part()).await; + assert!(denied(&fanned, "effect push-1"), "{fanned:?}"); + let tree = owner.task_tree(&id).await.unwrap(); + assert_eq!(tree.children.len(), 1); + assert_eq!(tree.children[0].task.status, Status::Open); + owner + .effect_settled(&resumes.id, None, "push-1", Outcome::Applied) + .await + .unwrap(); + let fanned = agent.fan_out(&id, &part()).await.unwrap(); + assert_eq!(fanned.children.len(), 1); + let folded = owner.get_task(&resumes.id).await.unwrap(); + assert_eq!(folded.status, Status::Invalidated); + } + + /// A session writes its own words on a task only at or below the task's + /// classification: a session that reads the task later reads those words. + #[tokio::test] + async fn a_session_that_read_above_a_task_writes_no_words_of_its_own_on_it() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let (agent, id) = held(&owner, dir.path(), "Announce the launch").await; + let secret = owner + .write(crate::WriteRequest { + body: "The launch is CODENAME-BLUEFIN, in May.\n".into(), + classification: Some(Label::Restricted), + ..crate::WriteRequest::default() + }) + .await + .unwrap() + .id; + agent.get(&secret).await.unwrap(); + assert_eq!(agent.label(), Label::Restricted); + let begun = agent.effect_begun(&id, None, "bluefin-may").await; + assert!(denied(&begun, "read above"), "{begun:?}"); + let waits = agent + .wait_for(&id, None, "launch:CODENAME-BLUEFIN", None) + .await; + assert!(denied(&waits, "read above"), "{waits:?}"); + let closed = agent.invalidate(&id, "BLUEFIN moved", None).await; + assert!(denied(&closed, "read above"), "{closed:?}"); + // a session below reads the task, and none of those words + let reader = as_author(dir.path(), "agent:c").with_handling_cap(Label::Internal); + let seen = reader.get_task(&id).await.unwrap(); + assert_eq!(seen.spec.classification, Label::Public); + assert_eq!((seen.status, seen.reason), (Status::Claimed, None)); + assert_eq!((seen.waiting_on, seen.effects), (None, Vec::new())); + // on a task at its label, the session writes them + let high = owner + .create_task(TaskSpec { + title: "Brief the launch team".into(), + body: "Do it.\n".into(), + classification: Label::Restricted, + ..TaskSpec::default() + }) + .await + .unwrap() + .task; + owner.admit(&high.id).await.unwrap(); + agent + .claim(&high.id, time::Duration::minutes(5)) + .await + .unwrap(); + let begun = agent.effect_begun(&high.id, None, "bluefin-may").await; + assert_eq!(begun.unwrap().effects, vec!["bluefin-may".to_string()]); + } + + /// A parent that the session may not read is refused as one that does not exist. + #[tokio::test] + async fn a_filing_names_no_parent_that_the_session_may_not_read() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let hidden = owner + .create_task(TaskSpec { + title: "Brief the launch team".into(), + body: "Do it.\n".into(), + classification: Label::Restricted, + ..TaskSpec::default() + }) + .await + .unwrap() + .task; + let capped = as_author(dir.path(), "agent:c").with_handling_cap(Label::Internal); + let mut answers = Vec::new(); + for parent in [hidden.id.as_str(), "task-me-nowhere"] { + let filed = capped + .create_task(TaskSpec { + title: "A part".into(), + body: "Do it.\n".into(), + parent: Some(parent.to_string()), + ..TaskSpec::default() + }) + .await; + match filed { + Err(Error::Ledger(dm_ledger::Error::NotFound(named))) => { + assert_eq!(named, parent); + answers.push("not found"); + } + other => panic!("{parent}: {other:?}"), + } + } + assert_eq!(answers.len(), 2); + assert_eq!(capped.label(), Label::Public); + for st in [Status::Proposed, Status::Open] { + let listed = owner.tasks(None, st).await.unwrap(); + assert!(listed.iter().all(|t| t.spec.parent.is_none()), "{listed:?}"); + } + } + #[tokio::test] async fn the_service_says_which_policy_a_task_was_filed_under_and_checks_its_parent() { let dir = tempfile::tempdir().unwrap(); diff --git a/crates/server/src/lib.rs b/crates/server/src/lib.rs index fe64c563..7d68c701 100644 --- a/crates/server/src/lib.rs +++ b/crates/server/src/lib.rs @@ -2451,10 +2451,12 @@ impl Memory { /// /// # Errors /// - /// A parent outside the grants or above the handling, or a ledger error; - /// [`Error::Denied`] when the parent has an open continuation this session may not - /// read. That check runs before the items are matched, so the refusal is the same - /// whatever the items are. + /// A parent outside the grants or above the handling, or a ledger error: the ledger + /// files no part and no continuation of a task whose work is held by an effect that + /// nobody has settled. [`Error::Denied`] when the parent has an open continuation + /// this session may not read, or one with an effect that is not settled. Those + /// checks run before the items are matched, so the refusal is the same whatever the + /// items are. pub async fn fan_out(&self, parent: &str, items: &[(String, String)]) -> Result { let mut l = self.ledger.lock().await; let parent = l.get(parent).await?; @@ -2656,6 +2658,13 @@ impl Memory { .into(), )); } + // a continuation that began an effect is not closed as folded while nobody + // knows what the destination holds: its work would go to the new one + if let Some(why) = holds::unsettled(&t) { + return Err(Error::Denied(format!( + "{why}; settle the effect, then fan out again" + ))); + } ids.extend(t.spec.depends_on.iter().cloned()); folded.push(t.id); } @@ -2727,7 +2736,8 @@ impl Memory { /// /// # Errors /// - /// A task outside the grants, or a ledger error. + /// A task outside the grants, or a ledger error: the ledger files no retry of a + /// task whose work is held by an effect that nobody has settled. pub async fn retry_task(&self, id: &str) -> Result { let mut l = self.ledger.lock().await; let root = l.get(id).await?; @@ -2933,19 +2943,9 @@ impl Memory { }; lock(&self.claims).remove(id); // a continuation finishes the task it resumed, and that one the task it resumed in - // turn: up the chain, while each is open and nobody holds it - let mut by = done.clone(); - while let Some(parent) = by - .spec - .origin - .as_deref() - .and_then(|o| o.strip_prefix("continues:")) - { - let Ok(closed) = l.complete_by_continuation(parent, &by.id, handoff).await else { - break; - }; - by = closed; - } + // turn: up the chain, while each is open, nobody holds it and its effects are + // settled. One left open for an effect closes when the effect is settled + l.close_resumed(&done).await; self.observe_task(&done); Ok(done) } @@ -3019,10 +3019,18 @@ impl Memory { /// not its to close: the call is refused before anything moves, and the refusal /// names nothing of that work. /// + /// A task with an effect that nobody has settled is not closed, and neither is a + /// task with such a task open below it: what the destination holds is not known, + /// and a closed task says that the question is over. Whoever looked at the + /// destination settles the effect first. The reason is the session's own words on + /// the task, so a session that has read above the task's classification is refused. + /// /// # Errors /// /// A ledger error, or a task outside the grants; [`Error::Denied`] when an open task - /// filed from it, however far down, is above the session's handling. + /// filed from it, however far down, is above the session's handling, when the task + /// or such an open task has an effect that is not settled, or when the session's + /// label is above the task's classification. pub async fn invalidate( &self, id: &str, @@ -3030,9 +3038,12 @@ impl Memory { superseded_by: Option<&str>, ) -> Result { let mut l = self.ledger.lock().await; - self.check_task(&l.get(id).await?)?; + let named = l.get(id).await?; + self.check_task(&named)?; + self.writes_within(&named, "a reason")?; let handling = self.handling()?; let mut open = Vec::new(); + let mut unsettled = holds::unsettled(&named); let mut pending = l.filed_from(id).await?; while let Some(t) = pending.pop() { pending.extend(l.filed_from(&t.id).await?); @@ -3045,8 +3056,16 @@ impl Memory { .into(), )); } + unsettled = unsettled.or_else(|| holds::unsettled(&t)); open.push(t.id); } + // after the walk: the refusal for work this session may not read comes first, + // and is the same whatever that work holds + if let Some(why) = unsettled { + return Err(Error::Denied(format!( + "{why}; settle the effect, then invalidate" + ))); + } let task = l.invalidate(id, reason, superseded_by).await?; for t in open { l.invalidate( @@ -6602,6 +6621,34 @@ mod tests { .id } + /// A task is filed under the verified policy entries of the record, whatever its + /// filer says, and so are the part and the continuation that a fan-out files. + #[tokio::test] + async fn a_filing_names_the_verified_policy_of_the_record() { + let dir = tempfile::tempdir().unwrap(); + let owner = mem(dir.path()); + let agent = Memory::open(dir.path()).unwrap().with_author("agent:a"); + let rule = policy_entry(&owner, RULE).await; + // a rule that nobody verified is not in force, and no filing names it + policy_entry(&owner, &rule_text("pol-L2-091", 2, "")).await; + verify(&owner, &agent, &rule).await; + let fresh = Memory::open(dir.path()).unwrap(); + let filed = fresh + .create_task(TaskSpec { + policy_context: vec!["pol-made-up".into()], + ..tspec("Carry out the decision", 0, None, &[], &[]) + }) + .await + .unwrap() + .task; + assert_eq!(filed.spec.policy_context, vec![rule.clone()]); + let part = [("A part".to_string(), "Do the part.\n".to_string())]; + let out = fresh.fan_out(&filed.id, &part).await.unwrap(); + assert_eq!(out.children[0].spec.policy_context, vec![rule.clone()]); + let resumes = out.continuation.unwrap(); + assert_eq!(resumes.spec.policy_context, vec![rule]); + } + /// An agent supersedes the newest decision of the record's one chain. async fn agent_supersedes(root: &Path, of: &str) -> Result { let agent = Memory::open(root).unwrap().with_author("agent:a"); diff --git a/local/README.md b/local/README.md index df299783..c2e903b5 100644 --- a/local/README.md +++ b/local/README.md @@ -729,12 +729,36 @@ The task's `effects` field lists the actions that are not settled. While the lis empty: - no session can claim the task, also after the lease expires; -- the holder cannot complete the task. +- the holder cannot complete the task; +- the ledger files no retry, no continuation and no subagent task of the task, so + `retry_task` and `fan_out` fail; +- no session can claim a task that continues or retries the task, however far down; +- `amem` does not invalidate the task, or a task that has such a task open below it; +- `fan_out` does not close an open continuation that has such an action. + +The work of the task thus goes to no other session, under the name of the task or under +a different name. A refused claim goes on the record of the task with its reason. The +reason names the event that the task waited on, or the action that was not settled. `effect_settled` (`amem task effect-settle applied|not_applied`) records what the destination holds. The holder can settle its action while its claim is in force. After that, only a person or a tap can settle it. An agent that lost its claim cannot, -because it cannot know if the destination got the action. +because it cannot know if the destination got the action. The same outcome for a settled +action changes nothing. The ledger refuses a different outcome for a settled action, and +the record keeps the first outcome. + +A continuation can be done while its task has an action that is not settled. The task +then stays open, and no session can claim it. When the last action is settled, the ledger +closes the task with the handoff of that continuation. + +An event name, an action id and the reason of an invalidation are words of the session. +Each session that can read the task reads them. Thus `amem` refuses `wait_for`, +`effect_begun` and `invalidate_task` from a session whose session label is above the +classification of the task. + +`amem work` and `amem gofannon work` do not complete a task that has an action that is +not settled. They release the task with a handoff. With `amem work`, the handoff names +the action. A task also has these fields: `parent`, `scores`, `implements`, `reversibility`, `artifacts_touched` and `policy_context`. `amem task create` and `create_task` take them, @@ -749,6 +773,16 @@ external actions in sequence and settles them. `amem` does not make the external so the holder must record it. The taps do not report events yet: a person does, or a script that runs `amem task arrived`. +These limits are open: + +- The ledger knows the work of a task only from the `origin` of a task filed from it. A + new task with the same text and no such `origin` is not held. +- `amem` does not raise the classification of a task. A session that read above the + classification of its task cannot record an action or a wait on that task. +- A continuation can be done while a holder has its task under a lease in force. The + ledger does not close that task, because the holder completes it. If the lease expires + first, the task is on offer again. + ## Task definitions: `platform task` A task definition is the record for one kind of work. It is an artifact in the platform From 2fe833945071a7fc5d7ba8cb1573981005704fb1 Mon Sep 17 00:00:00 2001 From: Sander Striker Date: Sun, 15 Nov 2026 09:20:30 +0100 Subject: [PATCH 3/3] docs: name the pull request of package 2 in the plan The row of package 2 in "Landed so far" said "this pull request". It now names #81. Co-Authored-By: Claude Fable 5.2 Claude-Session: https://claude.ai/code/session_01D5DyJRpu5ned5y3d92U7Ts --- PLAN.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/PLAN.md b/PLAN.md index ff87103f..d79d94b7 100644 --- a/PLAN.md +++ b/PLAN.md @@ -68,7 +68,7 @@ Landed so far: |---|---|---| | 0 | #69 | Nothing | | 1 | #71 | Nothing | -| 2 | #73, attempt keys for a claim and a completion in #77, and waiting, unsettled effects and the remaining fields in this pull request | An attempt key on the other operations; a connector that serialises external actions and reconciles them; taps that report the events tasks wait on; rules that read `scores`, `implements`, `reversibility` and `artifacts_touched`; the hold of an unsettled effect follows a task's `origin` only, so the same work filed anew under no origin is not held; a task's classification does not rise to its holder's label, so a session that has read above its task records no effect and no wait on it; a task held under a live lease when its continuation finishes is left to its holder, and is on offer again if that lease runs out | +| 2 | #73, attempt keys for a claim and a completion in #77, and waiting, unsettled effects and the remaining fields in #81 | An attempt key on the other operations; a connector that serialises external actions and reconciles them; taps that report the events tasks wait on; rules that read `scores`, `implements`, `reversibility` and `artifacts_touched`; the hold of an unsettled effect follows a task's `origin` only, so the same work filed anew under no origin is not held; a task's classification does not rise to its holder's label, so a session that has read above its task records no effect and no wait on it; a task held under a live lease when its continuation finishes is left to its holder, and is on offer again if that lease runs out | | 3 | #74 | A suite that gates a version of a definition; the data scope enforced; a definition required by default, once the system's own filings name one; tools offered a few at boot and the rest by search | | 6 | #78 | `check_policy` selecting rules by the action; the unused-rule check over one review period; the approver of an exception checked against the layer breached; a new version of a rule keeping its place under the ceiling; a decision that survives an enforced module that fails; the author and the date of an imported rule not trusted; a session that reloads the policy when a rule changes; a verified rule that stays in force while its next version waits for review; a denial that names no policy counted for the module that made it, in the rates and the lint |