1use std::pin::pin;23use anyhow::bail;4use futures::stream::Peekable;5use futures::StreamExt as _;6use remowt_ui_prompt::Prompter;7use tokio::io::{AsyncRead, AsyncWrite, AsyncWriteExt as _};8use tokio::select;9use tokio_util::codec::{FramedRead, LinesCodec};1011pub async fn run_conversation<R, W, P>(reader: R, mut writer: W, prompt: P) -> anyhow::Result<()>12where13 R: AsyncRead,14 W: AsyncWrite + Unpin,15 P: Prompter,16{17 let mut lines = pin!(FramedRead::new(reader, LinesCodec::new()).peekable());1819 while let Some(line) = lines.next().await {20 let line = line?;21 let res = if let Some(prompt_text) = line.strip_prefix("PAM_PROMPT_ECHO_OFF ") {22 prompt.prompt_text(false, prompt_text, "", &[]).await?23 } else if let Some(prompt_text) = line.strip_prefix("PAM_PROMPT_ECHO_ON ") {24 prompt.prompt_text(true, prompt_text, "", &[]).await?25 } else if let Some(msg_text) = line.strip_prefix("PAM_ERROR_MSG ") {26 prompt.display_text(true, msg_text, &[]).await?;27 String::new()28 } else if let Some(msg_text) = line.strip_prefix("PAM_TEXT_INFO ") {29 select! {30 _ = Peekable::peek(lines.as_mut()) => {},31 r = prompt.display_text(false, msg_text, &[]) => {r?}32 }33 String::new()34 } else if line == "SUCCESS" {35 return Ok(());36 } else if line == "FAILURE" {37 bail!("helper reported failure")38 } else {39 bail!("unknown agent request: {line}")40 };4142 if res.contains('\n') {43 bail!("response should not include newline")44 }4546 writer.write_all(res.as_bytes()).await?;47 writer.write_all(b"\n").await?;48 }49 bail!("agent finished unexpectedly")50}