made_core/entities/
deliberation.rs

1//! [`Deliberation`] aggregate.
2//!
3//! The Deliberation is the central aggregate root of MADE.
4//! It owns the lifecycle of one deliberation from proposal generation
5//! through peer review to scoring and completion. State transitions
6//! are explicit and protected so no caller can place the aggregate in
7//! an inconsistent shape.
8//!
9//! Phase graph (linear):
10//!
11//! ```text
12//! Proposing -> Revising -> Validating -> Scoring -> Completed
13//! ```
14//!
15//! Transitions are one-way. `Revising` accepts many `revise_proposal`
16//! calls so the use-case layer can run multiple peer-review rounds
17//! (critique → revise) without adding externally-observable phases.
18//! Methods reject operations that do not match the current phase.
19
20use std::collections::BTreeMap;
21
22use serde::{Deserialize, Serialize};
23use time::OffsetDateTime;
24
25use crate::entities::{DeliberationPhase, Proposal, RankedOutcome, ValidationOutcome};
26use crate::error::DomainError;
27use crate::value_objects::{DurationMs, ProposalContent, ProposalId, Rounds, Specialty, TaskId};
28
29/// Aggregate root: one deliberation over one task.
30#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
31pub struct Deliberation {
32    task_id: TaskId,
33    specialty: Specialty,
34    rounds_budget: Rounds,
35    phase: DeliberationPhase,
36
37    proposals: BTreeMap<ProposalId, Proposal>,
38    outcomes: BTreeMap<ProposalId, ValidationOutcome>,
39    ranking: Vec<ProposalId>,
40
41    #[serde(with = "time::serde::rfc3339")]
42    started_at: OffsetDateTime,
43    #[serde(with = "time::serde::rfc3339::option")]
44    completed_at: Option<OffsetDateTime>,
45}
46
47impl Deliberation {
48    #[must_use]
49    pub fn start(
50        task_id: TaskId,
51        specialty: Specialty,
52        rounds_budget: Rounds,
53        now: OffsetDateTime,
54    ) -> Self {
55        Self {
56            task_id,
57            specialty,
58            rounds_budget,
59            phase: DeliberationPhase::Proposing,
60            proposals: BTreeMap::new(),
61            outcomes: BTreeMap::new(),
62            ranking: Vec::new(),
63            started_at: now,
64            completed_at: None,
65        }
66    }
67
68    #[must_use]
69    pub fn task_id(&self) -> &TaskId {
70        &self.task_id
71    }
72    #[must_use]
73    pub fn specialty(&self) -> &Specialty {
74        &self.specialty
75    }
76    #[must_use]
77    pub fn rounds_budget(&self) -> Rounds {
78        self.rounds_budget
79    }
80    #[must_use]
81    pub fn phase(&self) -> DeliberationPhase {
82        self.phase
83    }
84    #[must_use]
85    pub fn proposals(&self) -> &BTreeMap<ProposalId, Proposal> {
86        &self.proposals
87    }
88    #[must_use]
89    pub fn outcomes(&self) -> &BTreeMap<ProposalId, ValidationOutcome> {
90        &self.outcomes
91    }
92    #[must_use]
93    pub fn started_at(&self) -> OffsetDateTime {
94        self.started_at
95    }
96    #[must_use]
97    pub fn completed_at(&self) -> Option<OffsetDateTime> {
98        self.completed_at
99    }
100
101    /// Add a new proposal. Only allowed while `Proposing`. Duplicate
102    /// proposal ids are rejected.
103    pub fn add_proposal(&mut self, proposal: Proposal) -> Result<(), DomainError> {
104        self.require_phase(DeliberationPhase::Proposing)?;
105        if self.proposals.contains_key(proposal.id()) {
106            return Err(DomainError::AlreadyExists {
107                what: "deliberation.proposal",
108            });
109        }
110        self.proposals.insert(proposal.id().clone(), proposal);
111        Ok(())
112    }
113
114    /// Revise an existing proposal. Only allowed in the `Revising` phase.
115    pub fn revise_proposal(
116        &mut self,
117        proposal_id: &ProposalId,
118        new_content: impl Into<ProposalContent>,
119        now: OffsetDateTime,
120    ) -> Result<(), DomainError> {
121        self.require_phase(DeliberationPhase::Revising)?;
122        let proposal = self
123            .proposals
124            .get_mut(proposal_id)
125            .ok_or(DomainError::NotFound {
126                what: "deliberation.proposal",
127            })?;
128        proposal.revise(new_content, now)
129    }
130
131    /// Attach a validation outcome for a proposal. Only allowed in
132    /// `Validating`. Every proposal must receive exactly one outcome
133    /// before advancing to `Scoring`.
134    pub fn attach_outcome(
135        &mut self,
136        proposal_id: &ProposalId,
137        outcome: ValidationOutcome,
138    ) -> Result<(), DomainError> {
139        self.require_phase(DeliberationPhase::Validating)?;
140        if !self.proposals.contains_key(proposal_id) {
141            return Err(DomainError::NotFound {
142                what: "deliberation.proposal",
143            });
144        }
145        if self.outcomes.contains_key(proposal_id) {
146            return Err(DomainError::AlreadyExists {
147                what: "deliberation.outcome",
148            });
149        }
150        self.outcomes.insert(proposal_id.clone(), outcome);
151        Ok(())
152    }
153
154    /// Advance to the next phase, enforcing the preconditions of the
155    /// transition:
156    ///
157    /// - `Proposing -> Revising`: at least one proposal present.
158    /// - `Validating -> Scoring`: every proposal has an outcome.
159    /// - Other transitions are unconditional.
160    #[allow(unknown_lints, clippy::collapsible_match)] // collapsible_match added in clippy 1.95; rust 1.90 toolchain doesn't know it
161    pub fn advance(&mut self) -> Result<DeliberationPhase, DomainError> {
162        let next = self.phase.next().ok_or(DomainError::InvalidTransition {
163            from: "Completed",
164            to: "Completed",
165        })?;
166
167        match (self.phase, next) {
168            (DeliberationPhase::Proposing, DeliberationPhase::Revising) => {
169                if self.proposals.is_empty() {
170                    return Err(DomainError::InvariantViolated {
171                        reason: "cannot leave Proposing without proposals",
172                    });
173                }
174            }
175            (DeliberationPhase::Validating, DeliberationPhase::Scoring) => {
176                if self.outcomes.len() != self.proposals.len() {
177                    return Err(DomainError::InvariantViolated {
178                        reason: "every proposal must have an outcome before Scoring",
179                    });
180                }
181            }
182            _ => {}
183        }
184
185        self.phase = next;
186        Ok(self.phase)
187    }
188
189    /// Compute the ranking and mark the deliberation complete. Only
190    /// allowed from `Scoring`. The winning proposal gets rank 0; ties
191    /// are broken by proposal id to keep the ordering deterministic.
192    pub fn complete(&mut self, now: OffsetDateTime) -> Result<Vec<RankedOutcome>, DomainError> {
193        self.require_phase(DeliberationPhase::Scoring)?;
194
195        let mut ranked =
196            self.materialize_ranked_tuples(&self.proposals.keys().cloned().collect::<Vec<_>>())?;
197
198        ranked.sort_by(|a, b| b.2.score().cmp(&a.2.score()).then_with(|| a.0.cmp(&b.0)));
199
200        self.ranking = ranked.iter().map(|(id, _, _)| id.clone()).collect();
201        self.phase = DeliberationPhase::Completed;
202        self.completed_at = Some(now);
203
204        Ok(ranked
205            .into_iter()
206            .enumerate()
207            .map(|(i, (_, proposal, outcome))| {
208                RankedOutcome::new(proposal, outcome, u32::try_from(i).unwrap_or(u32::MAX))
209            })
210            .collect())
211    }
212
213    /// Reorder the final ranking after completion while preserving the
214    /// same proposal set. This lets the application layer impose a
215    /// deterministic post-scoring preference (for example, valid
216    /// structured outputs before invalid ones) without mutating
217    /// proposals or outcomes.
218    pub fn reprioritize(
219        &mut self,
220        ranking: Vec<ProposalId>,
221    ) -> Result<Vec<RankedOutcome>, DomainError> {
222        self.require_phase(DeliberationPhase::Completed)?;
223        let current: std::collections::BTreeSet<_> = self.ranking.iter().cloned().collect();
224        let proposed: std::collections::BTreeSet<_> = ranking.iter().cloned().collect();
225        if ranking.len() != self.ranking.len() || current != proposed {
226            return Err(DomainError::InvariantViolated {
227                reason: "reprioritized ranking must contain every completed proposal exactly once",
228            });
229        }
230
231        let ranked = self.materialize_ranked_tuples(&ranking)?;
232        self.ranking = ranking;
233        Ok(ranked
234            .into_iter()
235            .enumerate()
236            .map(|(i, (_, proposal, outcome))| {
237                RankedOutcome::new(proposal, outcome, u32::try_from(i).unwrap_or(u32::MAX))
238            })
239            .collect())
240    }
241
242    /// Total duration from start to completion, when completed.
243    #[must_use]
244    pub fn duration(&self) -> Option<DurationMs> {
245        self.completed_at.map(|end| {
246            let delta = end - self.started_at;
247            let millis = delta.whole_milliseconds();
248            let bounded = u64::try_from(millis).unwrap_or(0);
249            DurationMs::from_millis(bounded)
250        })
251    }
252
253    #[must_use]
254    pub fn ranking(&self) -> &[ProposalId] {
255        &self.ranking
256    }
257
258    /// Reconstruct [`RankedOutcome`]s from the persisted ranking + the
259    /// stored proposals and outcomes. Useful when an upstream caller
260    /// (e.g. `RunCouncilDecisionUseCase` in Warn mode) needs to read a
261    /// completed deliberation back from a repository without re-running
262    /// the algorithm.
263    ///
264    /// Fails with [`DomainError::InvalidTransition`] if the deliberation
265    /// is not yet in the `Completed` phase.
266    pub fn ranked_outcomes(&self) -> Result<Vec<RankedOutcome>, DomainError> {
267        self.require_phase(DeliberationPhase::Completed)?;
268        Ok(self
269            .materialize_ranked_tuples(&self.ranking)?
270            .into_iter()
271            .enumerate()
272            .map(|(i, (_, proposal, outcome))| {
273                RankedOutcome::new(proposal, outcome, u32::try_from(i).unwrap_or(u32::MAX))
274            })
275            .collect())
276    }
277
278    fn require_phase(&self, expected: DeliberationPhase) -> Result<(), DomainError> {
279        if self.phase == expected {
280            Ok(())
281        } else {
282            Err(DomainError::InvalidTransition {
283                from: self.phase.name(),
284                to: expected.name(),
285            })
286        }
287    }
288
289    fn materialize_ranked_tuples(
290        &self,
291        ranking: &[ProposalId],
292    ) -> Result<Vec<(ProposalId, Proposal, ValidationOutcome)>, DomainError> {
293        ranking
294            .iter()
295            .map(|id| {
296                let proposal =
297                    self.proposals
298                        .get(id)
299                        .cloned()
300                        .ok_or(DomainError::InvariantViolated {
301                            reason: "missing proposal in ranking",
302                        })?;
303                let outcome =
304                    self.outcomes
305                        .get(id)
306                        .cloned()
307                        .ok_or(DomainError::InvariantViolated {
308                            reason: "missing outcome at Scoring",
309                        })?;
310                Ok::<_, DomainError>((id.clone(), proposal, outcome))
311            })
312            .collect()
313    }
314}
315
316#[cfg(test)]
317mod tests {
318    use super::*;
319    use crate::entities::ValidatorReport;
320    use crate::value_objects::{AgentId, Attributes, Score, TaskId};
321    use time::macros::datetime;
322
323    fn now() -> OffsetDateTime {
324        datetime!(2026-04-15 12:00:00 UTC)
325    }
326
327    fn specialty() -> Specialty {
328        Specialty::new("triage").unwrap()
329    }
330
331    fn start() -> Deliberation {
332        Deliberation::start(
333            TaskId::new("t1").unwrap(),
334            specialty(),
335            Rounds::default(),
336            now(),
337        )
338    }
339
340    fn proposal(id: &str, content: &str) -> Proposal {
341        Proposal::new(
342            ProposalId::new(id).unwrap(),
343            AgentId::new("a").unwrap(),
344            specialty(),
345            content,
346            Attributes::empty(),
347            now(),
348        )
349        .unwrap()
350    }
351
352    fn outcome(score: f64) -> ValidationOutcome {
353        ValidationOutcome::new(
354            Score::new(score).unwrap(),
355            vec![ValidatorReport::new("x", true, "", Attributes::empty()).unwrap()],
356        )
357    }
358
359    #[test]
360    fn starts_in_proposing() {
361        let d = start();
362        assert_eq!(d.phase(), DeliberationPhase::Proposing);
363        assert!(d.proposals().is_empty());
364        assert!(d.completed_at().is_none());
365    }
366
367    #[test]
368    fn proposals_only_accepted_while_proposing() {
369        let mut d = start();
370        d.add_proposal(proposal("p1", "x")).unwrap();
371        d.advance().unwrap(); // Revising
372        let err = d.add_proposal(proposal("p2", "y")).unwrap_err();
373        assert!(matches!(err, DomainError::InvalidTransition { .. }));
374    }
375
376    #[test]
377    fn duplicate_proposal_id_is_rejected() {
378        let mut d = start();
379        d.add_proposal(proposal("p1", "x")).unwrap();
380        assert!(matches!(
381            d.add_proposal(proposal("p1", "y")).unwrap_err(),
382            DomainError::AlreadyExists { .. }
383        ));
384    }
385
386    #[test]
387    fn cannot_leave_proposing_without_proposals() {
388        let mut d = start();
389        assert!(matches!(
390            d.advance().unwrap_err(),
391            DomainError::InvariantViolated { .. }
392        ));
393        assert_eq!(d.phase(), DeliberationPhase::Proposing);
394    }
395
396    #[test]
397    fn revise_only_allowed_while_revising() {
398        let mut d = start();
399        d.add_proposal(proposal("p1", "x")).unwrap();
400        assert!(matches!(
401            d.revise_proposal(&ProposalId::new("p1").unwrap(), "y", now())
402                .unwrap_err(),
403            DomainError::InvalidTransition { .. }
404        ));
405        d.advance().unwrap(); // Revising
406        d.revise_proposal(&ProposalId::new("p1").unwrap(), "y", now())
407            .unwrap();
408        assert_eq!(
409            d.proposals()
410                .get(&ProposalId::new("p1").unwrap())
411                .unwrap()
412                .content(),
413            "y"
414        );
415    }
416
417    #[test]
418    fn cannot_enter_scoring_with_missing_outcomes() {
419        let mut d = start();
420        d.add_proposal(proposal("p1", "x")).unwrap();
421        d.add_proposal(proposal("p2", "y")).unwrap();
422        for _ in 0..2 {
423            d.advance().unwrap();
424        }
425        // Now in Validating. Attach only one outcome.
426        d.attach_outcome(&ProposalId::new("p1").unwrap(), outcome(0.9))
427            .unwrap();
428        assert!(matches!(
429            d.advance().unwrap_err(),
430            DomainError::InvariantViolated { .. }
431        ));
432    }
433
434    #[test]
435    fn duplicate_outcome_is_rejected() {
436        let mut d = start();
437        d.add_proposal(proposal("p1", "x")).unwrap();
438        for _ in 0..2 {
439            d.advance().unwrap();
440        }
441        d.attach_outcome(&ProposalId::new("p1").unwrap(), outcome(0.8))
442            .unwrap();
443        assert!(matches!(
444            d.attach_outcome(&ProposalId::new("p1").unwrap(), outcome(0.9))
445                .unwrap_err(),
446            DomainError::AlreadyExists { .. }
447        ));
448    }
449
450    #[test]
451    fn complete_ranks_descending_by_score() {
452        let mut d = start();
453        d.add_proposal(proposal("p1", "a")).unwrap();
454        d.add_proposal(proposal("p2", "b")).unwrap();
455        d.add_proposal(proposal("p3", "c")).unwrap();
456        for _ in 0..2 {
457            d.advance().unwrap();
458        }
459        d.attach_outcome(&ProposalId::new("p1").unwrap(), outcome(0.5))
460            .unwrap();
461        d.attach_outcome(&ProposalId::new("p2").unwrap(), outcome(0.9))
462            .unwrap();
463        d.attach_outcome(&ProposalId::new("p3").unwrap(), outcome(0.7))
464            .unwrap();
465        d.advance().unwrap(); // Scoring
466
467        let ranked = d.complete(datetime!(2026-04-15 12:00:01 UTC)).unwrap();
468        assert_eq!(d.phase(), DeliberationPhase::Completed);
469        assert_eq!(ranked[0].rank(), 0);
470        assert_eq!(ranked[0].proposal().id().as_str(), "p2");
471        assert_eq!(ranked[1].proposal().id().as_str(), "p3");
472        assert_eq!(ranked[2].proposal().id().as_str(), "p1");
473    }
474
475    #[test]
476    fn reprioritize_reorders_completed_ranking() {
477        let mut d = start();
478        d.add_proposal(proposal("p1", "a")).unwrap();
479        d.add_proposal(proposal("p2", "b")).unwrap();
480        for _ in 0..2 {
481            d.advance().unwrap();
482        }
483        d.attach_outcome(&ProposalId::new("p1").unwrap(), outcome(0.9))
484            .unwrap();
485        d.attach_outcome(&ProposalId::new("p2").unwrap(), outcome(0.1))
486            .unwrap();
487        d.advance().unwrap();
488        d.complete(now()).unwrap();
489
490        let reprioritized = d
491            .reprioritize(vec![
492                ProposalId::new("p2").unwrap(),
493                ProposalId::new("p1").unwrap(),
494            ])
495            .unwrap();
496        assert_eq!(reprioritized[0].proposal().id().as_str(), "p2");
497        assert_eq!(reprioritized[1].proposal().id().as_str(), "p1");
498        assert_eq!(d.ranking()[0].as_str(), "p2");
499    }
500
501    #[test]
502    fn ties_are_broken_by_proposal_id() {
503        let mut d = start();
504        d.add_proposal(proposal("p2", "a")).unwrap();
505        d.add_proposal(proposal("p1", "b")).unwrap();
506        for _ in 0..2 {
507            d.advance().unwrap();
508        }
509        d.attach_outcome(&ProposalId::new("p1").unwrap(), outcome(0.7))
510            .unwrap();
511        d.attach_outcome(&ProposalId::new("p2").unwrap(), outcome(0.7))
512            .unwrap();
513        d.advance().unwrap();
514
515        let ranked = d.complete(now()).unwrap();
516        assert_eq!(ranked[0].proposal().id().as_str(), "p1");
517        assert_eq!(ranked[1].proposal().id().as_str(), "p2");
518    }
519
520    #[test]
521    fn complete_only_allowed_from_scoring() {
522        let mut d = start();
523        d.add_proposal(proposal("p1", "x")).unwrap();
524        assert!(matches!(
525            d.complete(now()).unwrap_err(),
526            DomainError::InvalidTransition { .. }
527        ));
528    }
529
530    #[test]
531    fn completed_deliberation_has_duration() {
532        let mut d = start();
533        d.add_proposal(proposal("p1", "x")).unwrap();
534        for _ in 0..2 {
535            d.advance().unwrap();
536        }
537        d.attach_outcome(&ProposalId::new("p1").unwrap(), outcome(0.5))
538            .unwrap();
539        d.advance().unwrap();
540        d.complete(datetime!(2026-04-15 12:00:00.750 UTC)).unwrap();
541
542        assert_eq!(d.duration().unwrap().get(), 750);
543    }
544
545    #[test]
546    fn cannot_advance_past_completed() {
547        let mut d = start();
548        d.add_proposal(proposal("p1", "x")).unwrap();
549        for _ in 0..2 {
550            d.advance().unwrap();
551        }
552        d.attach_outcome(&ProposalId::new("p1").unwrap(), outcome(0.5))
553            .unwrap();
554        d.advance().unwrap();
555        d.complete(now()).unwrap();
556        assert!(matches!(
557            d.advance().unwrap_err(),
558            DomainError::InvalidTransition { .. }
559        ));
560    }
561}
made_core/entities/deliberation.rs

made_core/entities/
deliberation.rs
