magi/
question_delivery.rs

1//! Admission of operator text: a saved deputy is history, not a listener.
2//! Called under the question update lock as well as when rendering its card.
3use std::path::Path;
4
5use anyhow::Result;
6use jiff::Timestamp;
7use serde::{Deserialize, Serialize};
8
9use crate::ask::{Question, Questions};
10use crate::config::Config;
11
12/// Recovery capabilities of the configuration actually held by the daemon.
13/// Contains no commands, environment values, or credentials.
14#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
15pub struct Recovery {
16    max_deputies: usize,
17    deputy_agents: Vec<String>,
18    deputy_fallback: bool,
19    /// Default timeout used by the running recovery loops.
20    pub answer_timeout: u64,
21}
22
23impl Recovery {
24    /// Snapshot the same configuration and capacity passed to `Deputies`.
25    pub fn new(cfg: Option<&Config>, max_deputies: usize) -> Self {
26        Self {
27            max_deputies,
28            deputy_agents: cfg.map_or_else(Vec::new, |c| {
29                c.agents.iter().map(|a| a.id.clone()).collect()
30            }),
31            deputy_fallback: cfg.is_some_and(|c| c.resolve_roles().is_ok()),
32            answer_timeout: cfg.map_or_else(
33                || Config::default().graph.answer_timeout,
34                |c| c.graph.answer_timeout,
35            ),
36        }
37    }
38
39    fn can_start(&self, agent: &str) -> bool {
40        self.max_deputies > 0
41            && (self.deputy_fallback
42                || (!agent.is_empty() && self.deputy_agents.iter().any(|id| id == agent)))
43    }
44}
45
46/// Stable refusal reasons, independent of the display language.
47#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize)]
48#[serde(rename_all = "snake_case")]
49pub enum UnavailableReason {
50    /// The question is already closed.
51    Settled,
52    /// The deputy has used its start budget.
53    DeputyExhausted,
54    /// Deputy configuration or its agent is unavailable.
55    DeputyDisabled,
56    /// The recipient deadline has passed.
57    Deadline,
58    /// The original agent cannot resume.
59    RecipientUnavailable,
60    /// No live daemon can recover the recipient.
61    DaemonUnavailable,
62}
63
64impl UnavailableReason {
65    /// Existing language-independent API reason code.
66    pub fn code(self) -> &'static str {
67        match self {
68            Self::Settled => "settled",
69            Self::DeputyExhausted => "deputy_exhausted",
70            Self::DeputyDisabled => "deputy_disabled",
71            Self::Deadline => "deadline",
72            Self::RecipientUnavailable => "recipient_unavailable",
73            Self::DaemonUnavailable => "daemon_unavailable",
74        }
75    }
76
77    fn message(self, ja: bool) -> &'static str {
78        let (en, jp) = match self {
79            Self::Settled => (
80                "This question is already settled.",
81                "この質問は終了しています。",
82            ),
83            Self::DeputyExhausted => (
84                "Free-text replies are unavailable: the deputy has used all its starts.",
85                "自由文を送信できません。代理エージェントの起動回数を使い切りました。",
86            ),
87            Self::DeputyDisabled => (
88                "Free-text replies are unavailable: deputies are disabled or their configuration or agent is unavailable.",
89                "自由文を送信できません。代理エージェントが無効か、設定またはエージェントを利用できません。",
90            ),
91            Self::Deadline => (
92                "Free-text replies are unavailable: the recipient's deadline has passed.",
93                "自由文を送信できません。受信先の応答期限を過ぎています。",
94            ),
95            Self::RecipientUnavailable => (
96                "Free-text replies are unavailable: the asking agent has no resumable session.",
97                "自由文を送信できません。質問したエージェントのセッションを再開できません。",
98            ),
99            Self::DaemonUnavailable => (
100                "Free-text replies are unavailable: no running daemon can recover the recipient.",
101                "自由文を送信できません。受信先を復旧するデーモンが稼働していません。",
102            ),
103        };
104        if ja { jp } else { en }
105    }
106}
107
108/// Typed refusal shared by web and command-line ingress.
109#[derive(Debug, thiserror::Error)]
110#[error("question_text_unavailable ({code}): {message}", code = .reason.code())]
111pub struct TextUnavailable {
112    /// Machine-readable refusal reason.
113    pub reason: UnavailableReason,
114    /// Localized explanation for the operator.
115    pub message: String,
116}
117
118/// Current delivery state projected to the question card.
119#[derive(Debug, Clone, Serialize)]
120pub struct Availability {
121    /// `live`, automatically recoverable `waiting`, or `unavailable`.
122    pub state: &'static str,
123    /// Stable reason code, independent of the UI language.
124    pub reason: &'static str,
125    /// Explanation in the question or configured language.
126    pub message: String,
127    #[serde(skip)]
128    unavailable_reason: Option<UnavailableReason>,
129}
130
131impl Availability {
132    /// Whether an operator message can reach a live or recoverable recipient.
133    pub fn accepts_text(&self) -> bool {
134        self.state != "unavailable"
135    }
136}
137
138/// The same heartbeat and upgrade lease readers used by the daemon UI.
139fn recovering(home: &Path, now: Timestamp, cfg: Option<&Config>) -> Option<Recovery> {
140    let daemon = crate::daemon::read_status(home)?;
141    let handover = crate::updater::read_progress(home).is_some_and(|p| {
142        matches!(
143            p.stage,
144            crate::updater::Stage::Parking | crate::updater::Stage::Restarting
145        ) && crate::updater::live_lease(&p, crate::updater::read_lease(home).as_ref(), now)
146            .is_some()
147    });
148    let running = daemon.running(now);
149    let recovery = daemon.question_recovery?;
150    if running {
151        return Some(recovery);
152    }
153    // During handover a successor has not published its capabilities yet.
154    // Do not promise recovery if prospective settings changed in the gap.
155    if handover && cfg.is_some_and(|c| recovery == Recovery::new(Some(c), c.daemon.max_deputies)) {
156        Some(recovery)
157    } else {
158        None
159    }
160}
161
162/// Inspect current leases and the existing deputy/waiter recovery prerequisites.
163pub fn availability(store: &Questions, q: &Question, cfg: Option<&Config>) -> Availability {
164    let now = Timestamp::now();
165    let home = store.root().parent().unwrap_or(store.root());
166    let ja = q.merge_target.as_ref().map_or_else(
167        || cfg.is_some_and(|c| c.graph.language == "ja"),
168        |t| t.language == "ja",
169    );
170    let result = |state, reason, en: &str, jp: &str| Availability {
171        state,
172        reason,
173        unavailable_reason: None,
174        message: if ja { jp } else { en }.to_owned(),
175    };
176    let unavailable = |reason: UnavailableReason| Availability {
177        state: "unavailable",
178        reason: reason.code(),
179        message: reason.message(ja).to_owned(),
180        unavailable_reason: Some(reason),
181    };
182    if !q.status.open() {
183        return unavailable(UnavailableReason::Settled);
184    }
185    if store.read_lease(&q.id).is_some_and(|l| l.fresh(now)) {
186        return result(
187            "live",
188            "lease",
189            if q.waiting_on_agent() {
190                "Waiting for the agent to reply."
191            } else {
192                "An agent is listening."
193            },
194            if q.waiting_on_agent() {
195                "エージェントの返答を待っています。"
196            } else {
197                "エージェントが応答を受け取れます。"
198            },
199        );
200    }
201    let recovery = recovering(home, now, cfg);
202    let timeout = recovery.as_ref().map_or_else(
203        || Config::default().graph.answer_timeout,
204        |r| r.answer_timeout,
205    );
206    if crate::deputy::kind_of(q).is_some() {
207        if q.deputy
208            .as_ref()
209            .is_some_and(|d| d.starts >= crate::deputy::MAX_STARTS)
210        {
211            return unavailable(UnavailableReason::DeputyExhausted);
212        }
213        let Some(recovery) = recovery.as_ref() else {
214            return unavailable(UnavailableReason::DaemonUnavailable);
215        };
216        if !recovery.can_start(crate::deputy::agent_of(q)) {
217            return unavailable(UnavailableReason::DeputyDisabled);
218        }
219        if !crate::deputy::restart_in_time(q, timeout, now) {
220            return unavailable(UnavailableReason::Deadline);
221        }
222    } else {
223        let waiter = crate::waiter::Waiter::new(store.clone(), home.to_path_buf(), cfg.cloned());
224        if q.cwd.is_none() || !waiter.can_resume(q) {
225            return unavailable(UnavailableReason::RecipientUnavailable);
226        }
227        if crate::waiter::decide(q, None, false, timeout, now) == crate::waiter::Action::Expire {
228            return unavailable(UnavailableReason::Deadline);
229        }
230    }
231    if recovery.is_some() {
232        result(
233            "waiting",
234            "recovery",
235            "Waiting for automatic delivery: the recipient is starting, restarting, or waiting for a free slot.",
236            "自動配送待ちです。受信先の起動・再起動、または実行枠の空きを待っています。",
237        )
238    } else {
239        unavailable(UnavailableReason::DaemonUnavailable)
240    }
241}
242
243/// Must run against the latest question inside `Questions::update`. A refusal
244/// returns before changing the thread, answer, or delivery cursor.
245pub fn require(store: &Questions, q: &Question, cfg: Option<&Config>) -> Result<()> {
246    let delivery = availability(store, q, cfg);
247    if let Some(reason) = delivery.unavailable_reason {
248        return Err(TextUnavailable {
249            reason,
250            message: delivery.message,
251        }
252        .into());
253    }
254    Ok(())
255}
magi/question_delivery.rs

magi/
question_delivery.rs
