use super::{Pump, json};
use crate::assistant::{
permission::{Decision, Pending, Request},
wire::Written,
};
use serde_json::Value;
impl Pump {
pub(super) fn permission(&mut self, id: Value, params: &Value) -> anyhow::Result<()> {
let Ok(conversation) = self.permission_session(params) else {
return self
.output
.refuse(id, -32602, "permission names a foreign session");
};
let Ok(budget) = self
.issuer
.words_budget(&conversation, self.seat.harness.content_bytes)
else {
return self
.output
.refuse(id, -32602, "permission_envelope_exceeds_content_bytes");
};
let Ok(request) =
Request::decode(self.session.version, params.clone(), &conversation, budget)
else {
return self.output.refuse(
id,
-32602,
"permission_params_invalid_or_exceed_content_bytes",
);
};
if self.pending.len() >= self.seat.harness.pending_permissions {
return self.output.refuse(id, -32001, "pending_permissions_full");
}
anyhow::ensure!(
!self
.pending
.values()
.any(|pending| pending.id == id && !pending.answer_queued),
"permission_request_id_already_pending"
);
let key = crate::assistant::permission::operation()?;
let cancelled = self.turn.as_ref().is_some_and(|turn| turn.cancelled);
let pending = Pending {
id,
request,
conversation,
draft: None,
cancelled,
holding: !cancelled,
answer_queued: false,
};
if !cancelled {
let issuer = self.issuer.clone();
let job = key.clone();
let words = pending.request.words.clone();
let mut ready = self.approvals.ready.clone();
self.drafts.spawn(async move {
if ready.wait_for(|ready| *ready).await.is_err() {
return (job, Err(anyhow::anyhow!("permission_watch_not_ready")));
}
let result = issuer.hold(job.clone(), words).await;
(job, result)
});
}
self.pending.insert(key.clone(), pending);
if cancelled {
self.answer(&key, Decision::Cancelled, None, None)?;
}
Ok(())
}
pub async fn drafted(
&mut self,
key: &str,
result: anyhow::Result<String>,
) -> anyhow::Result<()> {
let pending = self
.pending
.get_mut(key)
.ok_or_else(|| anyhow::anyhow!("permission_hold_request_unknown"))?;
pending.holding = false;
if pending.cancelled {
self.pending.remove(key);
return Ok(());
}
match result {
Ok(draft) => {
anyhow::ensure!(
!self
.pending
.values()
.any(|pending| pending.draft.as_deref() == Some(&draft)),
"permission_draft_id_reused"
);
let pending = self
.pending
.get_mut(key)
.ok_or_else(|| anyhow::anyhow!("permission_hold_request_unknown"))?;
pending.draft = Some(draft.clone());
let words = pending.request.words.clone();
let conversation = pending.conversation.clone();
self.known_drafts.push_back(draft.clone());
let bound = self.seat.harness.pending_permissions.saturating_mul(2);
if self.known_drafts.len() > bound {
self.known_drafts.pop_front();
}
self.refresh_approvals()?;
self.say_in(&conversation,json!({"kind":"state","state":"waiting_permission","pending_permissions":self.pending.len(),"draft":draft})).await?;
self.say_in(&conversation,json!({"kind":"permission_ask","draft":draft,"url":self.issuer.url(&draft)?,"words":words})).await
}
Err(error) => self.answer(key, Decision::Cancelled, None, Some(&error.to_string())),
}
}
pub(super) async fn decision(
&self,
id: &Value,
change: &Value,
person: &str,
) -> anyhow::Result<()> {
let Some(draft) = change["draft"].as_str() else {
return self.refuse(id, "invalid_decision").await;
};
if !matches!(change["decision"].as_str(), Some("approve" | "refuse")) {
return self.refuse(id, "invalid_decision").await;
}
if !self.known_drafts.iter().any(|known| known == draft)
&& !self
.pending
.values()
.any(|pending| pending.draft.as_deref() == Some(draft))
{
return self.refuse(id, "draft_unknown").await;
}
anyhow::ensure!(!person.is_empty(), "permission_decider_missing");
self.ack(id).await
}
fn refresh_approvals(&self) -> anyhow::Result<()> {
let mut requests = std::collections::BTreeMap::new();
for pending in self
.pending
.values()
.filter(|pending| !pending.cancelled && !pending.answer_queued)
{
if let Some(id) = &pending.draft {
let words = serde_json::to_string(&pending.request.words)?;
requests.insert(
id.clone(),
crate::assistant::approval_watch::Expected {
digest: lys_pass::drafts::request_digest(&words),
},
);
}
}
self.approvals.replace(requests);
Ok(())
}
pub fn approved(
&mut self,
answer: &crate::assistant::approval_watch::Answer,
) -> anyhow::Result<()> {
let key = self
.pending
.iter()
.find(|(_, pending)| {
!pending.cancelled
&& !pending.answer_queued
&& pending.draft.as_deref() == Some(answer.draft.as_str())
})
.map(|(key, _)| key.clone());
if let Some(key) = key {
self.answer(&key, answer.decision, Some(&answer.person), None)?;
} else {
anyhow::ensure!(
self.known_drafts.contains(&answer.draft),
"permission_draft_unknown"
);
}
Ok(())
}
fn answer(
&mut self,
key: &str,
decision: Decision,
person: Option<&str>,
failure: Option<&str>,
) -> anyhow::Result<()> {
let pending = self
.pending
.get_mut(key)
.ok_or_else(|| anyhow::anyhow!("permission_answer_request_unknown"))?;
anyhow::ensure!(!pending.answer_queued, "permission_answer_already_queued");
let (result, decided) = crate::assistant::permission::decide(&pending.request, decision);
self.output
.response_tracked(pending.id.clone(), result.clone(), key.to_owned())?;
pending.answer_queued = true;
let item = json!({"kind":"permission_answered","draft":pending.draft,"decided":if failure.is_some() {"declined"}else{decided},"option_id":result["outcome"]["optionId"],"person":person,"code":failure});
self.answers
.insert(key.to_owned(), (pending.conversation.clone(), item));
if !pending.holding {
self.pending.remove(key);
}
self.refresh_approvals()
}
pub(super) fn cancel_pending(&mut self) -> anyhow::Result<()> {
let keys: Vec<_> = self
.pending
.iter()
.filter(|(_, pending)| !pending.answer_queued)
.map(|(key, _)| key.clone())
.collect();
for key in keys {
if let Some(pending) = self.pending.get_mut(&key) {
pending.cancelled = true;
}
self.answer(&key, Decision::Cancelled, None, None)?;
}
Ok(())
}
pub async fn written(&mut self, written: Written) -> anyhow::Result<()> {
anyhow::ensure!(written.delivered, "permission_answer_not_delivered");
let (conversation, item) = self
.answers
.remove(&written.key)
.ok_or_else(|| anyhow::anyhow!("permission_written_receipt_unknown"))?;
self.say_in(&conversation, item).await?;
self.reset_opening_deadline()?;
if self.answers.is_empty()
&& self.turn.as_ref().is_some_and(|turn| {
turn.idle.is_some()
&& (self.session.version == super::Version::One || turn.acknowledged)
})
{
self.end_turn().await?;
}
Ok(())
}
}