A record may leave its lists some time after reaching a state: leaves_after

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01GT37Z1Xtfd9pUuQtMTg6Yt

src/aggregates/mod.rs

@@ -51,6 +51,35 @@ pub struct AggregateSchema {
     /// no answer in six weeks has to go on without one, and nothing in
     /// a building is going to press that button.
     pub after_for_state: HashMap<String, AfterSpec>,
+    /// How long a record stays in the list once it has entered a
+    /// state, when that is not for ever: a card taken off the front
+    /// page leaves the desk's list a little later, instead of standing
+    /// there greyed out. The event log keeps it.
+    pub leaves_for_state: HashMap<String, LeavesSpec>,
+}
+
+/// `leaves_after: { minutes: 10 }` on a state: how long a record that
+/// has entered it stays in its bucket's list. Days, hours and minutes
+/// add up; the sweep that takes it out runs every minute.
+#[derive(Clone, Debug, Default, PartialEq, serde::Serialize, serde::Deserialize)]
+#[serde(deny_unknown_fields)]
+pub struct LeavesSpec {
+    #[serde(default, skip_serializing_if = "is_zero")]
+    pub days: u32,
+    #[serde(default, skip_serializing_if = "is_zero")]
+    pub hours: u32,
+    #[serde(default, skip_serializing_if = "is_zero")]
+    pub minutes: u32,
+}
+
+fn is_zero(n: &u32) -> bool {
+    *n == 0
+}
+
+impl LeavesSpec {
+    pub fn millis(&self) -> i64 {
+        ((i64::from(self.days) * 24 + i64::from(self.hours)) * 60 + i64::from(self.minutes)) * 60 * 1000
+    }
 }
 
 /// A mail to send when a record enters a state. `to` is `submitter`
@@ -125,6 +154,8 @@ struct RawState {
     grants: Option<String>,
     #[serde(default)]
     after: Option<AfterSpec>,
+    #[serde(default)]
+    leaves_after: Option<LeavesSpec>,
 }
 
 #[derive(Debug, serde::Deserialize)]
@@ -197,7 +228,14 @@ pub fn parse_aggregates_yaml(raw: &str) -> anyhow::Result<HashMap<String, Aggreg
         let mut mail_for_state = HashMap::new();
         let mut grants_for_state = HashMap::new();
         let mut after_for_state = HashMap::new();
+        let mut leaves_for_state = HashMap::new();
         for (state, raw_state) in raw_schema.states {
+            if let Some(leaves) = raw_state.leaves_after {
+                if leaves.millis() == 0 {
+                    anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: leaves_after needs some days, hours or minutes");
+                }
+                leaves_for_state.insert(state.clone(), leaves);
+            }
             if let Some(group) = raw_state.grants {
                 if group.trim().is_empty() {
                     anyhow::bail!("aggregates.yaml: bucket {bucket:?}: state {state:?}: grants needs a group");
@@ -245,6 +283,7 @@ pub fn parse_aggregates_yaml(raw: &str) -> anyhow::Result<HashMap<String, Aggreg
                 mail_for_state,
                 grants_for_state,
                 after_for_state,
+                leaves_for_state,
             },
         );
     }

src/deadlines.rs

@@ -37,6 +37,55 @@ pub async fn run(state: AppState) {
     }
 }
 
+/// Records leaving their list: once a record has been in a state with
+/// `leaves_after` for that long, it is taken out of the bucket. Not a
+/// move - nothing is decided, no mail goes - and the event log keeps
+/// the record. Every minute, over the buckets that use it and no
+/// others.
+pub async fn run_leaving(state: AppState) {
+    let mut tick = tokio::time::interval(std::time::Duration::from_secs(60));
+    loop {
+        tick.tick().await;
+        if let Err(e) = sweep_leaving(&state).await {
+            tracing::warn!(error = %e, "leaving sweep failed");
+        }
+    }
+}
+
+/// Whether a record has stayed out its time in the state it is in.
+pub fn has_left(schema: &crate::aggregates::AggregateSchema, answer: &Answer, now_ms: i64) -> bool {
+    schema
+        .leaves_for_state
+        .get(&answer.state)
+        .is_some_and(|leaves| now_ms >= waiting_since(answer) + leaves.millis())
+}
+
+pub async fn sweep_leaving(state: &AppState) -> anyhow::Result<()> {
+    let now_ms = chrono::Utc::now().timestamp_millis();
+    let aggregates = state.aggregates.load();
+    for schema in aggregates.values().filter(|s| !s.leaves_for_state.is_empty()) {
+        let Ok(store) = state.jetstream.get_key_value(&schema.bucket).await else { continue };
+        let keys: Vec<String> = {
+            use futures::TryStreamExt;
+            match store.keys().await {
+                Ok(keys) => keys.try_collect().await.unwrap_or_default(),
+                Err(_) => continue,
+            }
+        };
+        for id in keys {
+            let Ok(Some(bytes)) = store.get(&id).await else { continue };
+            let Ok(answer) = serde_json::from_slice::<Answer>(&bytes) else { continue };
+            if has_left(schema, &answer, now_ms) {
+                match store.delete(&id).await {
+                    Ok(()) => tracing::info!(bucket = %schema.bucket, id, state = %answer.state, "a record left its list"),
+                    Err(e) => tracing::warn!(bucket = %schema.bucket, id, "a record could not leave its list: {e}"),
+                }
+            }
+        }
+    }
+    Ok(())
+}
+
 /// When a record entered the state it is in now. A record nobody has
 /// moved has been waiting since it was sent.
 fn waiting_since(answer: &Answer) -> i64 {
@@ -150,6 +199,21 @@ aggregates:
         }
     }
 
+    #[test]
+    fn a_record_leaves_its_list_when_its_time_in_the_state_is_up() {
+        let s = schema(
+            "aggregates:\n  - bucket: people\n    initial: open\n    states:\n      open: { event: shown }\n      gone: { event: gone, leaves_after: { minutes: 10 } }\n    transitions:\n      open: [gone]\n",
+        );
+        let t = 1_000_000_000_000;
+        let mut answer = waiting("gone", t - 60 * 60 * 1000);
+        answer.decided_ms = Some(t);
+        assert!(!has_left(&s, &answer, t + 9 * 60 * 1000), "measured from entering the state, not from sending");
+        assert!(has_left(&s, &answer, t + 10 * 60 * 1000));
+        // A state without it keeps its records.
+        assert!(!has_left(&s, &waiting("open", 0), t));
+        assert!(crate::aggregates::parse_aggregates_yaml("aggregates:\n  - bucket: p\n    initial: open\n    states:\n      open: { event: o, leaves_after: {} }\n    transitions: {}\n").is_err());
+    }
+
     #[test]
     fn a_deadline_comes_due_the_day_it_says_and_not_before() {
         let s = schema(HEARINGS);

src/main.rs

@@ -132,6 +132,7 @@ async fn serve() -> anyhow::Result<()> {
     // windows (see announce.rs).
     tokio::spawn(portal::announce::run(state.clone()));
     tokio::spawn(portal::deadlines::run(state.clone()));
+    tokio::spawn(portal::deadlines::run_leaving(state.clone()));
     // The accounts the content mandates: the first person in, and
     // everyone a page names as responsible, at start and whenever the
     // content changes (seed.rs).

src/trainer/build.rs

@@ -440,6 +440,12 @@ pub struct PlanNeed {
     /// own it.
     #[serde(default)]
     pub one_per_person: bool,
+    /// How long a record stays in the lists once it reaches a stage,
+    /// when not for ever: `{ sluttet: { minutes: 10 } }` takes a card
+    /// off the desk's list ten minutes after it came off the front
+    /// page, instead of leaving it there greyed out.
+    #[serde(default)]
+    pub leaves_after: BTreeMap<String, crate::aggregates::LeavesSpec>,
     pub collects: Vec<Field>,
     /// How many required fields this person will tolerate: what the
     /// business said about them, not what the form happens to ask.
@@ -763,6 +769,12 @@ impl Plan {
                     bail!(at(&format!("deadline on {state:?} goes to {:?}, which is not a move the state machine has from there", d.to)));
                 }
             }
+            if let Some(bad) = need.leaves_after.keys().find(|st| !need.stages.contains(st)) {
+                bail!(at(&format!("leaves_after names {bad:?}, which is not one of its stages")));
+            }
+            if need.leaves_after.values().any(|l| l.millis() == 0) {
+                bail!(at("leaves_after needs some days, hours or minutes"));
+            }
             if need.one_per_person && need.enters_as.is_none() {
                 bail!(at("one_per_person keeps a card for each signed-in person, and the need has no enters_as to sign them in"));
             }
@@ -2059,6 +2071,9 @@ pub fn render(plan: &Plan, words: &BTreeMap<String, Value>) -> anyhow::Result<BT
                     if let Some(d) = need.deadlines.get(st) {
                         state["after"] = json!({ "days": d.days, "to": d.to });
                     }
+                    if let Some(leaves) = need.leaves_after.get(st) {
+                        state["leaves_after"] = json!(leaves);
+                    }
                     (st.clone(), state)
                 }).collect::<Map<_, _>>(),
                 "transitions": machine(need),
@@ -2720,7 +2735,7 @@ mod tests {
             stages: v(stages), done_when: v(done), success: v(success), skippable: v(skippable),
             moves_by: BTreeMap::new(), watched_by: vec![],
             events: BTreeMap::new(), attended_by: None, mails: BTreeMap::new(), stages_text: BTreeMap::new(), turned_down_from: vec![], reached_from: BTreeMap::new(), requires: String::new(), followed_by: vec![],
-            grants: BTreeMap::new(), deadlines: BTreeMap::new(), escalate: BTreeMap::new(), public_states: vec![], presents: false, paused: false, one_per_person: false,
+            grants: BTreeMap::new(), deadlines: BTreeMap::new(), escalate: BTreeMap::new(), public_states: vec![], presents: false, paused: false, one_per_person: false, leaves_after: BTreeMap::new(),
             collects: vec![], max_fields: 3,
         }
     }

src/trainer/plan.rs

@@ -196,6 +196,11 @@ rest. Each is one line in the plan, and the builder does the wiring:
   their own, which the form opens with and which sending again
   replaces - a card on the front page they edit themselves. A picture
   left unchosen stays as it was.
+- **Let finished things go.** `leaves_after: { <stage>: { minutes: 10 } }`
+  takes a record out of every list that long after it reaches that
+  stage (days, hours and minutes add up). A card taken off the front
+  page need not stand greyed out on the desk for ever; the history
+  keeps it.
 - **Pause what you cannot take on.** `paused: true` on a need keeps
   its card on the page, saying what the business does and that it is
   not taking new ones right now, with no form and a button that does