#![forbid(unsafe_code)]
#![warn(missing_docs)]
pub mod edge;
pub use edge::{EdgeAdapter, IngressDirective, Priority, build_attribution};
pub mod edge_credentials;
#[cfg(feature = "test-util")]
pub mod test_util {
use super::{
AssertedAttribution, Attribution, EdgeCredentials, Message, build_asserted_attribution,
};
use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;
#[must_use]
pub fn signed_envelope(
creds: &EdgeCredentials,
conversation_id: &str,
exec_id: &str,
messages: &[Message],
caller: Option<ExternalIdentity>,
participants: Vec<ExternalIdentity>,
) -> AssertedAttribution {
build_asserted_attribution(
creds,
conversation_id,
exec_id,
messages,
&Attribution {
caller,
participants,
},
)
}
}
pub use edge_credentials::{
CredentialError, EdgeCredentials, EdgeCredentialsError, edge_credentials_from_env_or_fail,
};
pub use polyc_crypto::sensitive::Sensitive;
#[must_use]
pub fn expose_nonempty(secret: Option<&Sensitive<String>>) -> Option<&str> {
secret
.map(Sensitive::expose)
.map(String::as_str)
.filter(|s| !s.is_empty())
}
#[doc(hidden)]
pub fn assert_redacted(value: &impl core::fmt::Debug, must_not_contain: &[&str]) {
let debug = format!("{value:?}");
assert!(
debug.contains("Sensitive(<redacted>)"),
"expected a Sensitive redaction marker in Debug output, got {debug}"
);
for raw in must_not_contain {
assert!(
!debug.contains(raw),
"Debug output leaked a raw secret value {raw:?}: {debug}"
);
}
}
use std::sync::Arc;
use chrono::{DateTime, Datelike as _, FixedOffset, Utc};
use connectrpc::client::{ClientConfig, HttpClient};
use futures::Stream;
use polyc_agent::text_message;
use polyc_proto::proto::polychrome::agent::v1::{
AgentEnd, AgentRequest, AgentServiceClient, AgentStart,
ApprovalPreview as WireAgentApprovalPreview,
ApprovalPreviewFire as WireAgentApprovalPreviewFire, AssertedAttribution, ClassifyRequest,
CompactionReason as WireCompactionReason, ContextCompacted,
IngressDirective as WireIngressDirective, InterruptRequest, Message, ParticipantMessage,
PendingApproval as WireAgentPendingApproval, PendingQuestion as WireAgentPendingQuestion,
QuestionOption as WireAgentQuestionOption, TurnFailureKind as WireTurnFailureKind, Verdict,
agent_response, content, tool_call_content,
};
use polyc_proto::proto::polychrome::approval::v1::{
ApprovalPreview as WireListApprovalPreview, ApprovalPreviewFire as WireListApprovalPreviewFire,
ApprovalResponseRequest, ApprovalServiceClient, ListPendingRequest, PendingApprovalEntry,
};
use polyc_proto::proto::polychrome::credential::v1::{
AddCredentialKeyRequest, CredentialKeySummary as WireCredentialKeySummary,
CredentialKeyVerifier as WireCredentialKeyVerifier, CredentialServiceClient,
CredentialSummary as WireCredentialSummary, EnrollCredentialRequest,
KeyState as WireCredentialKeyState, ListCredentialsRequest, RetireCredentialKeyRequest,
RevokeCredentialRequest,
};
use polyc_proto::proto::polychrome::ops::v1::{
AckRequest, DecideRequest, NotificationServiceClient, OperatorMailboxServiceClient,
PollPendingRequest, SubscribeRequest, decide_reply, ops_action_view, upgrade_outcome,
};
use polyc_proto::proto::polychrome::persona::v1::{
AdminInviteRequest, AttestVerifiedEmailOutcome, AttestVerifiedEmailRequest, AutoLinkOutcome,
AutoLinkRequest, CompleteLinkRequest, DescribeRequest, LinkOutcome, PersonaServiceClient,
RebuildUsageRollupsRequest, SetIncognitoRequest, StartDeepLinkRequest, StartLinkRequest,
};
use polyc_proto::proto::polychrome::question::v1::{
Decline as WireDecline, ListPendingQuestionsRequest, PendingQuestionEntry,
QuestionAnswerRequest, QuestionOptionEntry, QuestionServiceClient,
SelectOption as WireSelectOption, question_answer_request::Answer as WireAnswer,
};
use polyc_proto::proto::polychrome::routine::v1::{FireRoutineRequest, RoutineServiceClient};
#[derive(Debug, thiserror::Error)]
pub enum DialError {
#[error("invalid agent address {addr:?}: {source}")]
InvalidAddress {
addr: String,
#[source]
source: http::uri::InvalidUri,
},
#[error("tls setup failed for agent address: {0}")]
Tls(String),
#[error(transparent)]
Connect(#[from] connectrpc::ConnectError),
#[error("bearer credential is not a valid header value: {0}")]
InvalidBearer(String),
}
impl DialError {
#[must_use]
pub const fn code(&self) -> Option<connectrpc::ErrorCode> {
match self {
Self::Connect(e) => Some(e.code),
Self::InvalidAddress { .. } | Self::Tls(_) | Self::InvalidBearer(_) => None,
}
}
#[must_use]
pub const fn is_retryable(&self) -> bool {
matches!(
self.code(),
Some(
connectrpc::ErrorCode::Unavailable
| connectrpc::ErrorCode::DeadlineExceeded
| connectrpc::ErrorCode::ResourceExhausted
| connectrpc::ErrorCode::Aborted
)
)
}
#[must_use]
pub const fn is_deadline_exceeded(&self) -> bool {
matches!(self.code(), Some(connectrpc::ErrorCode::DeadlineExceeded))
}
}
const AGENT_DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);
const CONTROL_DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
fn http_client_for(uri: &http::Uri) -> Result<HttpClient, DialError> {
if uri.scheme_str() == Some("https") {
use rustls_platform_verifier::ConfigVerifierExt;
let tls = rustls::ClientConfig::with_platform_verifier()
.map_err(|e| DialError::Tls(e.to_string()))?;
Ok(HttpClient::builder()
.connect_timeout(CONNECT_TIMEOUT)
.with_tls(std::sync::Arc::new(tls)))
} else {
Ok(HttpClient::builder()
.connect_timeout(CONNECT_TIMEOUT)
.plaintext())
}
}
fn bearer_header_value(bearer: &str) -> Result<http::HeaderValue, DialError> {
let mut bearer_value = http::HeaderValue::from_str(&format!("Bearer {bearer}"))
.map_err(|source| DialError::InvalidBearer(source.to_string()))?;
bearer_value.set_sensitive(true);
Ok(bearer_value)
}
fn bearer_header(bearer: &str) -> Result<http::HeaderMap, DialError> {
let mut headers = http::HeaderMap::new();
headers.insert(http::header::AUTHORIZATION, bearer_header_value(bearer)?);
Ok(headers)
}
fn build_control_client<C>(
addr: &str,
bearer: Option<&str>,
build_client: impl FnOnce(HttpClient, ClientConfig) -> C,
) -> Result<Arc<C>, DialError> {
let uri = addr
.parse::<http::Uri>()
.map_err(|source| DialError::InvalidAddress {
addr: addr.to_owned(),
source,
})?;
let http = http_client_for(&uri)?;
let mut config = ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT);
if let Some(bearer) = bearer {
config = config.with_default_headers(bearer_header(bearer)?);
}
Ok(Arc::new(build_client(http, config)))
}
fn traced_options() -> connectrpc::client::CallOptions {
let mut headers = http::HeaderMap::new();
polyc_runtime::propagation::inject_current_span_into(&mut headers);
connectrpc::client::CallOptions::default()
.with_headers(headers.into_iter().filter_map(|(n, v)| n.map(|n| (n, v))))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CompactionReason {
Summarized,
Truncated,
}
impl CompactionReason {
#[must_use]
pub fn notice_headline(self, summarized_messages: u32) -> String {
match self {
Self::Summarized => {
let plural = if summarized_messages == 1 { "" } else { "s" };
format!(
"🧠 Summarized {summarized_messages} earlier message{plural} to keep the \
conversation manageable"
)
}
Self::Truncated => {
"✂️ Trimmed earlier tool output to keep the conversation manageable".to_owned()
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TurnEvent {
ContextCompacted {
reason: CompactionReason,
summarized_messages: u32,
summary_preview: String,
},
TextDelta(String),
ToolStarted {
name: String,
},
ApprovalPending {
request_id: String,
tool_name: String,
title: String,
args_json: String,
reason: String,
resolve_token: String,
preview: Option<ApprovalPreview>,
},
QuestionPending {
call_id: String,
index: u32,
header: String,
question: String,
options: Vec<QuestionOptionPrompt>,
args_json: String,
answer_token: String,
already_surfaced: bool,
},
HandoffStarted {
child_agent_id: String,
reason: String,
},
InviteDelivery {
target_user_id: String,
code: String,
inviter_display: String,
},
WalletLinkPrompt {
link_url: Option<String>,
renewal: bool,
},
PersonaCredentialPrompt {
link_url: Option<String>,
},
TurnFailed {
kind: TurnFailureKind,
message: String,
},
Done,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TurnFailureKind {
RateLimit,
Timeout,
Unavailable,
Auth,
BadRequest,
Other,
}
impl TurnFailureKind {
#[must_use]
pub const fn is_retryable(self) -> bool {
matches!(self, Self::RateLimit | Self::Timeout | Self::Unavailable)
}
}
#[derive(Clone)]
pub struct AgentDialer {
client: Arc<AgentServiceClient<HttpClient>>,
credentials: Option<Arc<EdgeCredentials>>,
}
impl AgentDialer {
pub fn new(addr: &str) -> Result<Self, DialError> {
let uri = addr
.parse::<http::Uri>()
.map_err(|source| DialError::InvalidAddress {
addr: addr.to_owned(),
source,
})?;
let http = http_client_for(&uri)?;
let config = ClientConfig::new(uri).with_default_timeout(AGENT_DIAL_TIMEOUT);
let client = AgentServiceClient::new(http, config);
Ok(Self {
client: Arc::new(client),
credentials: None,
})
}
pub fn with_credentials(addr: &str, creds: EdgeCredentials) -> Result<Self, DialError> {
let uri = addr
.parse::<http::Uri>()
.map_err(|source| DialError::InvalidAddress {
addr: addr.to_owned(),
source,
})?;
let http = http_client_for(&uri)?;
let headers = bearer_header(creds.bearer())?;
let config = ClientConfig::new(uri)
.with_default_timeout(AGENT_DIAL_TIMEOUT)
.with_default_headers(headers);
let client = AgentServiceClient::new(http, config);
Ok(Self {
client: Arc::new(client),
credentials: Some(Arc::new(creds)),
})
}
pub fn approval_dialer(addr: &str) -> Result<ApprovalDialer, DialError> {
ApprovalDialer::new(addr)
}
pub fn approval_dialer_with_credentials(
&self,
addr: &str,
) -> Result<ApprovalDialer, DialError> {
self.credentials.as_ref().map_or_else(
|| ApprovalDialer::new(addr),
|creds| ApprovalDialer::with_credentials(addr, Arc::clone(creds)),
)
}
pub fn question_dialer(addr: &str) -> Result<QuestionDialer, DialError> {
QuestionDialer::new(addr)
}
pub async fn run_turn(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
) -> Result<String, DialError> {
self.run_turn_with(conversation_id, exec_id, user_text, Attribution::default())
.await
}
pub async fn run_turn_with(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
attribution: Attribution,
) -> Result<String, DialError> {
Ok(self
.run_turn_with_approvals(
conversation_id,
exec_id,
user_text,
attribution,
IngressDirective::default(),
)
.await?
.reply)
}
pub async fn run_turn_with_approvals(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
attribution: Attribution,
ingress_directive: IngressDirective,
) -> Result<BufferedTurn, DialError> {
let request = build_request(
conversation_id,
exec_id,
vec![text_message("user", user_text)],
None,
&attribution,
false,
ingress_directive,
"",
self.credentials.as_deref(),
);
self.run_turn_buffered_request(request).await
}
pub async fn run_routine_turn(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
attribution: Attribution,
ingress_directive: IngressDirective,
occurrence: &str,
) -> Result<BufferedTurn, DialError> {
let request = build_request(
conversation_id,
exec_id,
vec![text_message("user", user_text)],
None,
&attribution,
true,
ingress_directive,
occurrence,
self.credentials.as_deref(),
);
self.run_turn_buffered_request(request).await
}
async fn run_turn_buffered_request(
&self,
request: AgentRequest,
) -> Result<BufferedTurn, DialError> {
let mut stream = self
.client
.connect_with_options(request, traced_options())
.await?;
let mut text_parts: Vec<String> = Vec::new();
let mut scaffolding: Vec<String> = Vec::new();
let mut wallet_link_prompt = WalletLinkPrompt::None;
let mut persona_credential_prompt: Option<String> = None;
let mut pending_approvals: Vec<PendingApprovalPrompt> = Vec::new();
let mut pending_questions: Vec<PendingQuestionPrompt> = Vec::new();
while let Some(view) = stream.message().await? {
let response = view.to_owned_message();
match response.r#type {
Some(agent_response::Type::Outputs(outputs)) => {
for msg in outputs.messages {
aggregate_output_message(msg, &mut text_parts, &mut scaffolding);
}
}
Some(agent_response::Type::End(end)) => {
if let Some(prompt) = end.wallet_link_prompt.into_option() {
let renewal = prompt.renewal;
let link_url =
(!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
wallet_link_prompt = WalletLinkPrompt::Present { link_url, renewal };
}
if let Some(prompt) = end.persona_credential_prompt.into_option() {
let link_url =
(!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
persona_credential_prompt = link_url;
}
pending_approvals = end
.pending_approvals
.into_iter()
.map(PendingApprovalPrompt::from)
.collect();
pending_questions = end
.pending_questions
.into_iter()
.map(PendingQuestionPrompt::from)
.collect();
}
Some(agent_response::Type::Compacted(_)) | None => {}
}
}
Ok(BufferedTurn {
reply: finalize_buffered_reply(
&text_parts,
&scaffolding,
wallet_link_prompt,
persona_credential_prompt.as_deref(),
),
pending_approvals,
pending_questions,
})
}
pub async fn run_turn_streaming(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
self.run_turn_streaming_messages(
conversation_id,
exec_id,
vec![text_message("user", user_text)],
)
.await
}
pub async fn run_turn_streaming_messages(
&self,
conversation_id: &str,
exec_id: &str,
messages: Vec<Message>,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
self.run_turn_streaming_messages_with(
conversation_id,
exec_id,
messages,
None,
Attribution::default(),
IngressDirective::default(),
)
.await
}
pub async fn run_turn_streaming_messages_with(
&self,
conversation_id: &str,
exec_id: &str,
messages: Vec<Message>,
payment_receipt: Option<PaymentReceipt>,
attribution: Attribution,
ingress_directive: IngressDirective,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
let request = build_request(
conversation_id,
exec_id,
messages,
payment_receipt,
&attribution,
false,
ingress_directive,
"",
self.credentials.as_deref(),
);
self.run_turn_streaming_request(request).await
}
async fn run_turn_streaming_request(
&self,
request: AgentRequest,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
let mut stream = self
.client
.connect_with_options(request, traced_options())
.await?;
Ok(async_stream::try_stream! {
let mut ended = false;
while let Some(view) = stream.message().await? {
let response = view.to_owned_message();
match response.r#type {
Some(agent_response::Type::Compacted(c)) => {
yield event_from_compacted(*c);
}
Some(agent_response::Type::Outputs(outputs)) => {
for msg in outputs.messages {
if let Some(event) = message_to_event(msg) {
yield event;
}
}
}
Some(agent_response::Type::End(end)) => {
for event in events_from_end(*end) {
yield event;
}
ended = true;
}
None => {}
}
}
if !ended {
yield TurnEvent::Done;
}
})
}
pub async fn should_respond(
&self,
conversation_id: &str,
bot_name: &str,
surface: &str,
transcript: Vec<ParticipantMessage>,
) -> Result<bool, DialError> {
let request = ClassifyRequest {
conversation_id: conversation_id.to_owned(),
bot_name: bot_name.to_owned(),
surface: surface.to_owned(),
transcript,
..Default::default()
};
let resp = self
.client
.classify_with_options(request, traced_options())
.await?
.into_owned();
Ok(resp.verdict.to_i32() == Verdict::VERDICT_RESPOND as i32)
}
pub async fn interrupt(&self, conversation_id: &str) -> Result<bool, DialError> {
let request = InterruptRequest {
conversation_id: conversation_id.to_owned(),
..Default::default()
};
let resp = self
.client
.interrupt_with_options(request, traced_options())
.await?
.into_owned();
Ok(resp.interrupted)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApprovalChoice {
Approve,
ApproveForSession,
Deny,
Abort,
Defer,
}
impl ApprovalChoice {
#[must_use]
pub const fn approved(self) -> bool {
matches!(self, Self::Approve | Self::ApproveForSession)
}
#[must_use]
pub const fn approved_for_session(self) -> bool {
matches!(self, Self::ApproveForSession)
}
#[must_use]
pub const fn is_abort(self) -> bool {
matches!(self, Self::Abort)
}
#[must_use]
pub const fn is_defer(self) -> bool {
matches!(self, Self::Defer)
}
}
#[must_use]
pub fn approval_decided_text(label: &str, choice: ApprovalChoice, decider: &str) -> String {
match choice {
ApprovalChoice::ApproveForSession => format!(
"✅ Approved by {decider} — running \"{label}\"… (won't ask again this session)"
),
ApprovalChoice::Approve => format!("✅ Approved by {decider} — running \"{label}\"…"),
ApprovalChoice::Deny => format!("🚫 Denied by {decider} — \"{label}\" was not run."),
ApprovalChoice::Abort => {
format!("🛑 Aborted by {decider} — \"{label}\" was not run; the turn was stopped.")
}
ApprovalChoice::Defer => {
format!("↩️ Sent back by {decider} — \"{label}\" is still waiting for a decision.")
}
}
}
#[must_use]
pub fn approval_completed_text(label: &str, decider: &str, success: bool) -> String {
if success {
format!("✅ Approved by {decider} — \"{label}\" is done.")
} else {
format!("✅ Approved by {decider} — \"{label}\" ran but hit an error.")
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApprovalOutcome {
pub persisted: bool,
pub signature_hex: String,
pub signed_by_hex: String,
}
impl From<polyc_proto::proto::polychrome::approval::v1::ApprovalResponseReply> for ApprovalOutcome {
fn from(reply: polyc_proto::proto::polychrome::approval::v1::ApprovalResponseReply) -> Self {
Self {
persisted: reply.persisted,
signature_hex: reply.signature_hex,
signed_by_hex: reply.signed_by_hex,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingApproval {
pub request_id: String,
pub tool_name: String,
pub args_json: String,
pub reason: String,
pub resolve_token: String,
pub preview: Option<ApprovalPreview>,
pub already_surfaced: bool,
}
impl From<PendingApprovalEntry> for PendingApproval {
fn from(p: PendingApprovalEntry) -> Self {
Self {
request_id: p.request_id,
tool_name: p.tool_name,
args_json: p.args_json,
reason: p.reason,
resolve_token: p.resolve_token,
preview: p.preview.into_option().map(Into::into),
already_surfaced: p.already_surfaced,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApprovalPreviewFire {
pub local_time: String,
pub utc_time: String,
pub zone_label: String,
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct ApprovalPreview {
pub prompt_text: String,
pub next_fires: Vec<ApprovalPreviewFire>,
pub zone_name: String,
pub zone_is_fallback: bool,
pub cadence: String,
}
impl From<WireAgentApprovalPreviewFire> for ApprovalPreviewFire {
fn from(f: WireAgentApprovalPreviewFire) -> Self {
Self {
local_time: f.local_time,
utc_time: f.utc_time,
zone_label: f.zone_label,
}
}
}
impl From<WireListApprovalPreviewFire> for ApprovalPreviewFire {
fn from(f: WireListApprovalPreviewFire) -> Self {
Self {
local_time: f.local_time,
utc_time: f.utc_time,
zone_label: f.zone_label,
}
}
}
impl From<WireAgentApprovalPreview> for ApprovalPreview {
fn from(p: WireAgentApprovalPreview) -> Self {
Self {
prompt_text: p.prompt_text,
next_fires: p.next_fires.into_iter().map(Into::into).collect(),
zone_name: p.zone_name,
zone_is_fallback: p.zone_is_fallback,
cadence: p.cadence,
}
}
}
impl From<WireListApprovalPreview> for ApprovalPreview {
fn from(p: WireListApprovalPreview) -> Self {
Self {
prompt_text: p.prompt_text,
next_fires: p.next_fires.into_iter().map(Into::into).collect(),
zone_name: p.zone_name,
zone_is_fallback: p.zone_is_fallback,
cadence: p.cadence,
}
}
}
#[must_use]
pub fn approval_preview_text(preview: &ApprovalPreview, now: DateTime<Utc>) -> String {
let mut lines = vec![
"This routine will run:".to_owned(),
String::new(),
preview.prompt_text.clone(),
];
if !preview.cadence.is_empty() {
lines.push(String::new());
lines.push(format!("Runs {}", preview.cadence));
}
lines.push(String::new());
if preview.next_fires.is_empty() {
lines.push(
"This schedule has no upcoming runs, so approving it won't run anything.".to_owned(),
);
} else {
lines.push(run_list_title(preview.next_fires.len()));
lines.push(FENCE.to_owned());
lines.extend(run_rows(&preview.next_fires, now));
lines.push(FENCE.to_owned());
}
if preview.zone_is_fallback {
lines.push(String::new());
lines.push(format!(
"We don't know your time zone yet, so these times are in the routine's own zone ({}).",
preview.zone_name
));
}
lines.join("\n")
}
const FENCE: &str = "```";
fn run_list_title(count: usize) -> String {
if count == 1 {
"Next run".to_owned()
} else {
format!("Next {count} runs")
}
}
struct RunRow {
day: String,
time: String,
relative: String,
}
fn run_rows(fires: &[ApprovalPreviewFire], now: DateTime<Utc>) -> Vec<String> {
let rows: Vec<RunRow> = fires.iter().map(|fire| run_cells(fire, now)).collect();
let day_width = rows
.iter()
.map(|row| row.day.chars().count())
.max()
.unwrap_or(0);
let time_width = rows
.iter()
.map(|row| row.time.chars().count())
.max()
.unwrap_or(0);
rows.iter()
.map(|row| {
format!(
" {:day_width$} {:time_width$} {}",
row.day, row.time, row.relative
)
})
.collect()
}
fn run_cells(fire: &ApprovalPreviewFire, now: DateTime<Utc>) -> RunRow {
let Ok(local) = DateTime::parse_from_rfc3339(&fire.local_time) else {
return RunRow {
day: fire.local_time.clone(),
time: String::new(),
relative: String::new(),
};
};
let now_local = now.with_timezone(local.offset());
let day = if local.year() == now_local.year() {
local.format("%a %b %e").to_string()
} else {
local.format("%a %b %e, %Y").to_string()
};
let mut time = local.format("%-I:%M %p").to_string();
if !fire.zone_label.is_empty() {
time.push(' ');
time.push_str(&fire.zone_label);
}
RunRow {
day,
time,
relative: relative_day(local, now_local),
}
}
fn relative_day(local: DateTime<FixedOffset>, now_local: DateTime<FixedOffset>) -> String {
if local < now_local {
return "passed".to_owned();
}
match (local.date_naive() - now_local.date_naive()).num_days() {
0 => "today".to_owned(),
1 => "tomorrow".to_owned(),
days => format!("in {days} days"),
}
}
#[derive(Clone)]
pub struct ApprovalDialer {
client: Arc<ApprovalServiceClient<HttpClient>>,
credentials: Option<Arc<EdgeCredentials>>,
}
impl ApprovalDialer {
pub fn new(addr: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, None, ApprovalServiceClient::new)?,
credentials: None,
})
}
pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, Some(bearer), ApprovalServiceClient::new)?,
credentials: None,
})
}
pub fn with_credentials(
addr: &str,
credentials: Arc<EdgeCredentials>,
) -> Result<Self, DialError> {
let client =
build_control_client(addr, Some(credentials.bearer()), ApprovalServiceClient::new)?;
Ok(Self {
client,
credentials: Some(credentials),
})
}
#[allow(clippy::too_many_arguments)] pub async fn respond(
&self,
request_id: &str,
choice: ApprovalChoice,
reason: &str,
conversation_id: &str,
modified_args_json: &str,
injected_context: &str,
resolve_token: &str,
responder: Option<ExternalIdentity>,
) -> Result<ApprovalOutcome, DialError> {
let request = self.respond_request(
request_id,
choice,
reason,
conversation_id,
modified_args_json,
injected_context,
resolve_token,
responder,
);
let reply = self
.client
.respond_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.into())
}
#[allow(clippy::too_many_arguments)]
fn respond_request(
&self,
request_id: &str,
choice: ApprovalChoice,
reason: &str,
conversation_id: &str,
modified_args_json: &str,
injected_context: &str,
resolve_token: &str,
responder: Option<ExternalIdentity>,
) -> ApprovalResponseRequest {
use polyc_proto::proto::polychrome::approval::v1::{
Approve, Defer, Deny, approval_response_request::Decision,
};
let decision = if choice.is_defer() {
Decision::Defer(Box::new(Defer {
reason: reason.to_owned(),
__buffa_unknown_fields: buffa::UnknownFields::default(),
}))
} else if choice.approved() {
Decision::Approve(Box::new(Approve {
modified_args_json: modified_args_json.to_owned(),
injected_context: injected_context.to_owned(),
reason: reason.to_owned(),
approved_for_session: choice.approved_for_session(),
__buffa_unknown_fields: buffa::UnknownFields::default(),
}))
} else {
Decision::Deny(Box::new(Deny {
reason: reason.to_owned(),
abort: choice.is_abort(),
__buffa_unknown_fields: buffa::UnknownFields::default(),
}))
};
let mut request = ApprovalResponseRequest {
request_id: request_id.to_owned(),
conversation_id: conversation_id.to_owned(),
decision: Some(decision),
resolve_token: resolve_token.to_owned(),
asserted_approval: buffa::MessageField::none(),
..Default::default()
};
if let (Some(responder), Some(creds)) = (responder, self.credentials.as_deref()) {
creds.attach_approval_assertion(&mut request, responder);
}
request
}
pub async fn list_pending(
&self,
conversation_id: &str,
) -> Result<Vec<PendingApproval>, DialError> {
let mut pending = Vec::new();
let mut page_token = String::new();
loop {
let request = ListPendingRequest {
conversation_id: conversation_id.to_owned(),
page_token: page_token.clone(),
..Default::default()
};
let reply = self
.client
.list_pending_with_options(request, traced_options())
.await?
.into_owned();
pending.extend(reply.pending.into_iter().map(PendingApproval::from));
if reply.next_page_token.is_empty() {
break;
}
page_token = reply.next_page_token;
}
Ok(pending)
}
pub async fn excise_taint(
&self,
conversation_id: &str,
positions: &[u64],
all_quarantined: bool,
source_only: bool,
reason: &str,
actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
) -> Result<ExcisionOutcome, DialError> {
use polyc_proto::proto::polychrome::approval::v1::ExciseTaintRequest;
let request = ExciseTaintRequest {
conversation_id: conversation_id.to_owned(),
positions: positions.to_vec(),
all_quarantined,
source_only,
reason: reason.to_owned(),
actor: buffa::MessageField::some(actor),
..Default::default()
};
let reply = self
.client
.excise_taint_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.into())
}
pub async fn replay_conversation(
&self,
conversation_id: &str,
from: Option<usize>,
to: Option<usize>,
over: Option<ReplayOverrideSpec>,
actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
) -> Result<ReplayReport, DialError> {
use polyc_proto::proto::polychrome::approval::v1::{
ReplayConversationRequest, ReplayOverride,
};
let override_msg = over.map(ReplayOverride::from);
let to_bound = |v: Option<usize>| v.and_then(|n| i64::try_from(n).ok()).unwrap_or(-1);
let request = ReplayConversationRequest {
conversation_id: conversation_id.to_owned(),
from: to_bound(from),
to: to_bound(to),
r#override: override_msg
.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
actor: buffa::MessageField::some(actor),
..Default::default()
};
let reply = self
.client
.replay_conversation_with_options(request, traced_options())
.await?
.into_owned();
let turns = reply
.turns
.into_iter()
.map(ReplayTurnVerdict::from)
.collect();
Ok(ReplayReport {
turns,
all_match: reply.all_match,
})
}
pub async fn verify_conversation(
&self,
conversation_id: &str,
actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
) -> Result<VerificationOutcome, DialError> {
use polyc_proto::proto::polychrome::approval::v1::VerifyConversationRequest;
let request = VerifyConversationRequest {
conversation_id: conversation_id.to_owned(),
actor: buffa::MessageField::some(actor),
..Default::default()
};
let reply = self
.client
.verify_conversation_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.into())
}
pub async fn repair_conversation(
&self,
conversation_id: &str,
actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
operation_id: &str,
) -> Result<Vec<u64>, DialError> {
use polyc_proto::proto::polychrome::approval::v1::RepairConversationRequest;
let request = RepairConversationRequest {
conversation_id: conversation_id.to_owned(),
actor: buffa::MessageField::some(actor),
operation_id: operation_id.to_owned(),
..Default::default()
};
let reply = self
.client
.repair_conversation_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.quarantined_positions)
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReplayOverrideSpec {
Completion {
turn: u32,
text: String,
},
ToolResult {
turn: u32,
index: u32,
result_json: String,
},
}
impl From<ReplayOverrideSpec> for polyc_proto::proto::polychrome::approval::v1::ReplayOverride {
fn from(spec: ReplayOverrideSpec) -> Self {
use polyc_proto::proto::polychrome::approval::v1::{
ReplayCompletionOverride, ReplayToolResultOverride, replay_override::Kind,
};
let kind = match spec {
ReplayOverrideSpec::Completion { turn, text } => {
Kind::Completion(Box::new(ReplayCompletionOverride {
turn,
text,
__buffa_unknown_fields: buffa::UnknownFields::default(),
}))
}
ReplayOverrideSpec::ToolResult {
turn,
index,
result_json,
} => Kind::ToolResult(Box::new(ReplayToolResultOverride {
turn,
index,
result_json,
__buffa_unknown_fields: buffa::UnknownFields::default(),
})),
};
Self {
kind: Some(kind),
__buffa_unknown_fields: buffa::UnknownFields::default(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReplayTurnOutcome {
Match,
Diverged {
field: String,
detail: String,
},
Unreplayable {
reason: String,
},
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReplayTurnVerdict {
pub turn_index: u32,
pub outcome: ReplayTurnOutcome,
}
impl From<polyc_proto::proto::polychrome::approval::v1::ReplayTurnVerdict> for ReplayTurnVerdict {
fn from(v: polyc_proto::proto::polychrome::approval::v1::ReplayTurnVerdict) -> Self {
use polyc_proto::proto::polychrome::approval::v1::ReplayOutcome;
let outcome = match v.outcome.as_known() {
Some(ReplayOutcome::Match) => ReplayTurnOutcome::Match,
Some(ReplayOutcome::Diverged) => ReplayTurnOutcome::Diverged {
field: v.field,
detail: v.detail,
},
Some(ReplayOutcome::Unreplayable) => {
ReplayTurnOutcome::Unreplayable { reason: v.detail }
}
Some(ReplayOutcome::Unspecified) | None => ReplayTurnOutcome::Unreplayable {
reason: "the control plane returned an unknown replay outcome".to_owned(),
},
};
Self {
turn_index: v.turn_index,
outcome,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ReplayReport {
pub turns: Vec<ReplayTurnVerdict>,
pub all_match: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ExcisionOutcome {
pub persisted: bool,
pub excised_positions: Vec<u64>,
pub signature_hex: String,
pub signed_by_hex: String,
}
impl From<polyc_proto::proto::polychrome::approval::v1::ExciseTaintReply> for ExcisionOutcome {
fn from(reply: polyc_proto::proto::polychrome::approval::v1::ExciseTaintReply) -> Self {
Self {
persisted: reply.persisted,
excised_positions: reply.excised_positions,
signature_hex: reply.signature_hex,
signed_by_hex: reply.signed_by_hex,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct VerificationOutcome {
pub verified: bool,
pub violation: String,
pub event_count: u64,
}
impl From<polyc_proto::proto::polychrome::approval::v1::VerifyConversationReply>
for VerificationOutcome
{
fn from(reply: polyc_proto::proto::polychrome::approval::v1::VerifyConversationReply) -> Self {
Self {
verified: reply.verified,
violation: reply.violation,
event_count: reply.event_count,
}
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct UsageRollupsRebuilt {
pub conversations_scanned: u32,
pub personas_rebuilt: u32,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StartedLink {
pub code: String,
pub expires_at_ms: u64,
pub persona_id: String,
}
impl From<polyc_proto::proto::polychrome::persona::v1::StartLinkReply> for StartedLink {
fn from(reply: polyc_proto::proto::polychrome::persona::v1::StartLinkReply) -> Self {
Self {
code: reply.code,
expires_at_ms: reply.expires_at_ms,
persona_id: reply.persona_id,
}
}
}
impl From<polyc_proto::proto::polychrome::persona::v1::AdminInviteReply> for StartedLink {
fn from(reply: polyc_proto::proto::polychrome::persona::v1::AdminInviteReply) -> Self {
Self {
code: reply.code,
expires_at_ms: reply.expires_at_ms,
persona_id: reply.persona_id,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StartedDeepLink {
pub token: String,
pub expires_at_ms: u64,
pub persona_id: String,
}
impl From<polyc_proto::proto::polychrome::persona::v1::StartDeepLinkReply> for StartedDeepLink {
fn from(reply: polyc_proto::proto::polychrome::persona::v1::StartDeepLinkReply) -> Self {
Self {
token: reply.token,
expires_at_ms: reply.expires_at_ms,
persona_id: reply.persona_id,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum LinkCeremony {
Linked {
persona_id: String,
},
AlreadyLinked {
persona_id: String,
},
InvalidOrExpired,
Throttled,
Failed,
EstablishedPersona,
}
impl LinkCeremony {
#[must_use]
pub const fn user_message(&self) -> &'static str {
match self {
Self::Linked { .. } => {
"✅ Linked — this account now shares one Polychrome persona with your other channels."
}
Self::AlreadyLinked { .. } => {
"✅ Already linked — this account was already on that persona."
}
Self::InvalidOrExpired => {
"That link is invalid or expired. Start a fresh one from your other channel and try again."
}
Self::Throttled => "Too many attempts — wait a few minutes, then try again.",
Self::Failed => "That link failed to complete. Start a fresh one and try again.",
Self::EstablishedPersona => {
"This account already has its own established persona. Start the link from this account instead, then complete it on the other channel."
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum AttestedEmail {
Linked {
persona_id: String,
},
AlreadyLinked {
persona_id: String,
},
EstablishedPersona,
Blocked,
CapHalted,
Failed,
}
#[must_use]
pub const fn invite_error_message(err: &DialError) -> &'static str {
match err.code() {
Some(ErrorCode::PermissionDenied) => ADMIN_ONLY_INVITE,
Some(ErrorCode::ResourceExhausted) => {
"That's a lot of invites in a row — wait a couple of minutes, then try again."
}
_ => "I couldn't create the invite right now — try again in a moment.",
}
}
pub const ADMIN_ONLY_INVITE: &str = "Only admins can send invites.";
#[must_use]
pub fn invite_opener(inviter: &str) -> String {
if inviter.trim().is_empty() {
"🎟️ You're invited to Polychrome.".to_owned()
} else {
format!("🎟️ You're invited. {inviter} set you up with access.")
}
}
#[must_use]
pub fn invite_sent_ack(target_display: &str) -> String {
format!("📬 Invite sent to {target_display} by DM. It expires in 60 minutes.")
}
#[must_use]
pub const fn incognito_error_message(err: &DialError) -> &'static str {
match err.code() {
Some(ErrorCode::PermissionDenied) => {
"I couldn't toggle incognito — that only works for a conversation you're part of."
}
_ => "I couldn't toggle incognito — try again.",
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum QuestionChoice {
SelectOption(u32),
Decline,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QuestionOutcome {
pub persisted: bool,
pub signature_hex: String,
pub signed_by_hex: String,
}
impl From<polyc_proto::proto::polychrome::question::v1::QuestionAnswerReply> for QuestionOutcome {
fn from(reply: polyc_proto::proto::polychrome::question::v1::QuestionAnswerReply) -> Self {
Self {
persisted: reply.persisted,
signature_hex: reply.signature_hex,
signed_by_hex: reply.signed_by_hex,
}
}
}
#[derive(Clone)]
pub struct QuestionDialer {
client: Arc<QuestionServiceClient<HttpClient>>,
}
impl QuestionDialer {
pub fn new(addr: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, None, QuestionServiceClient::new)?,
})
}
pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, Some(bearer), QuestionServiceClient::new)?,
})
}
pub async fn respond(
&self,
call_id: &str,
index: u32,
choice: QuestionChoice,
conversation_id: &str,
answer_token: &str,
responder: Option<ExternalIdentity>,
) -> Result<QuestionOutcome, DialError> {
let answer = match choice {
QuestionChoice::SelectOption(index) => {
WireAnswer::SelectOption(Box::new(WireSelectOption {
index,
__buffa_unknown_fields: buffa::UnknownFields::default(),
}))
}
QuestionChoice::Decline => WireAnswer::Decline(Box::new(WireDecline {
__buffa_unknown_fields: buffa::UnknownFields::default(),
})),
};
let request = QuestionAnswerRequest {
call_id: call_id.to_owned(),
index,
conversation_id: conversation_id.to_owned(),
answer: Some(answer),
answer_token: answer_token.to_owned(),
responder: responder.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
..Default::default()
};
let reply = self
.client
.respond_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.into())
}
pub async fn list_pending(
&self,
conversation_id: &str,
) -> Result<Vec<PendingQuestionPrompt>, DialError> {
let mut pending = Vec::new();
let mut page_token = String::new();
loop {
let request = ListPendingQuestionsRequest {
conversation_id: conversation_id.to_owned(),
page_token: page_token.clone(),
..Default::default()
};
let reply = self
.client
.list_pending_with_options(request, traced_options())
.await?
.into_owned();
pending.extend(reply.pending.into_iter().map(PendingQuestionPrompt::from));
if reply.next_page_token.is_empty() {
break;
}
page_token = reply.next_page_token;
}
Ok(pending)
}
}
#[derive(Clone)]
pub struct PersonaDialer {
client: Arc<PersonaServiceClient<HttpClient>>,
}
impl PersonaDialer {
pub fn new(addr: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, None, PersonaServiceClient::new)?,
})
}
pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, Some(bearer), PersonaServiceClient::new)?,
})
}
pub fn new_admin(addr: &str, bearer: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, Some(bearer), PersonaServiceClient::new)?,
})
}
pub async fn rebuild_usage_rollups(
&self,
actor: ExternalIdentity,
) -> Result<UsageRollupsRebuilt, DialError> {
let request = RebuildUsageRollupsRequest {
actor: buffa::MessageField::some(actor),
..Default::default()
};
let reply = self
.client
.rebuild_usage_rollups_with_options(request, traced_options())
.await?
.into_owned();
Ok(UsageRollupsRebuilt {
conversations_scanned: reply.conversations_scanned,
personas_rebuilt: reply.personas_rebuilt,
})
}
pub async fn start_link(&self, identity: ExternalIdentity) -> Result<StartedLink, DialError> {
let request = StartLinkRequest {
identity: buffa::MessageField::some(identity),
..Default::default()
};
let reply = self
.client
.start_link_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.into())
}
pub async fn start_deeplink(
&self,
identity: ExternalIdentity,
) -> Result<StartedDeepLink, DialError> {
let request = StartDeepLinkRequest {
identity: buffa::MessageField::some(identity),
..Default::default()
};
let reply = self
.client
.start_deep_link_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.into())
}
pub async fn admin_invite(
&self,
actor: ExternalIdentity,
target: ExternalIdentity,
) -> Result<StartedLink, DialError> {
let request = AdminInviteRequest {
actor: buffa::MessageField::some(actor),
target: buffa::MessageField::some(target),
..Default::default()
};
let reply = self
.client
.admin_invite_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.into())
}
pub async fn set_incognito(
&self,
actor: ExternalIdentity,
conversation_id: &str,
on: bool,
) -> Result<bool, DialError> {
let request = SetIncognitoRequest {
actor: buffa::MessageField::some(actor),
conversation_id: conversation_id.to_owned(),
on,
..Default::default()
};
let reply = self
.client
.set_incognito_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.on)
}
pub async fn auto_link(
&self,
identity: ExternalIdentity,
asserted_email: &str,
basis: &str,
) -> Result<Option<String>, DialError> {
let request = AutoLinkRequest {
identity: buffa::MessageField::some(identity),
asserted_email: asserted_email.to_owned(),
basis: basis.to_owned(),
..Default::default()
};
let reply = self
.client
.auto_link_with_options(request, traced_options())
.await?
.into_owned();
Ok(match reply.outcome.as_known() {
Some(AutoLinkOutcome::Linked) => Some(reply.persona_id),
_ => None,
})
}
pub async fn attest_verified_email(
&self,
identity: ExternalIdentity,
verified_email: &str,
) -> Result<AttestedEmail, DialError> {
let request = AttestVerifiedEmailRequest {
identity: buffa::MessageField::some(identity),
verified_email: verified_email.to_owned(),
..Default::default()
};
let reply = self
.client
.attest_verified_email_with_options(request, traced_options())
.await?
.into_owned();
Ok(match reply.outcome.as_known() {
Some(AttestVerifiedEmailOutcome::Linked) => AttestedEmail::Linked {
persona_id: reply.persona_id,
},
Some(AttestVerifiedEmailOutcome::AlreadyLinked) => AttestedEmail::AlreadyLinked {
persona_id: reply.persona_id,
},
Some(AttestVerifiedEmailOutcome::EstablishedPersona) => {
AttestedEmail::EstablishedPersona
}
Some(AttestVerifiedEmailOutcome::Blocked) => AttestedEmail::Blocked,
Some(AttestVerifiedEmailOutcome::CapHalted) => AttestedEmail::CapHalted,
Some(AttestVerifiedEmailOutcome::Unspecified) | None => AttestedEmail::Failed,
})
}
pub async fn complete_link(
&self,
code: &str,
identity: ExternalIdentity,
) -> Result<LinkCeremony, DialError> {
let request = CompleteLinkRequest {
code: code.to_owned(),
identity: buffa::MessageField::some(identity),
..Default::default()
};
let reply = self
.client
.complete_link_with_options(request, traced_options())
.await?
.into_owned();
Ok(match reply.outcome.as_known() {
Some(LinkOutcome::Linked) => LinkCeremony::Linked {
persona_id: reply.persona_id,
},
Some(LinkOutcome::AlreadyLinked) => LinkCeremony::AlreadyLinked {
persona_id: reply.persona_id,
},
Some(LinkOutcome::InvalidCode | LinkOutcome::Expired) => LinkCeremony::InvalidOrExpired,
Some(LinkOutcome::Throttled) => LinkCeremony::Throttled,
Some(LinkOutcome::EstablishedPersona) => LinkCeremony::EstablishedPersona,
Some(LinkOutcome::Unspecified) | None => LinkCeremony::Failed,
})
}
pub async fn describe(&self, identity: ExternalIdentity) -> Result<PersonaView, DialError> {
let request = DescribeRequest {
identity: buffa::MessageField::some(identity),
..Default::default()
};
let reply = self
.client
.describe_with_options(request, traced_options())
.await?
.into_owned();
let Some(profile) = reply.profile.into_option() else {
return Ok(PersonaView {
persona_id: reply.persona_id,
..Default::default()
});
};
Ok(PersonaView {
persona_id: reply.persona_id,
status: profile.status,
identities: profile
.identities
.into_iter()
.map(LinkedIdentity::from)
.collect(),
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LinkedIdentity {
pub provider: String,
pub external_id: String,
pub display_name: String,
}
impl From<ExternalIdentity> for LinkedIdentity {
fn from(id: ExternalIdentity) -> Self {
Self {
provider: id.provider,
external_id: id.external_id,
display_name: id.display_name,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PersonaView {
pub persona_id: String,
pub status: String,
pub identities: Vec<LinkedIdentity>,
}
pub use polyc_proto::proto::polychrome::agent::v1::Message as TurnMessage;
pub use polyc_proto::proto::polychrome::agent::v1::ParticipantMessage as GateMessage;
pub use polyc_proto::proto::polychrome::agent::v1::PaymentReceipt;
pub use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;
pub use connectrpc::ErrorCode;
#[derive(Debug, Clone, Default)]
pub struct Attribution {
pub caller: Option<ExternalIdentity>,
pub participants: Vec<ExternalIdentity>,
}
#[must_use]
pub fn attributed_message(speaker: &str, text: &str) -> TurnMessage {
text_message("user", &format!("{speaker}: {text}"))
}
#[must_use]
pub fn user_message(text: &str) -> TurnMessage {
text_message("user", text)
}
#[must_use]
pub fn namespaced_id(namespace: &str, native_id: &str) -> String {
format!("{namespace}:{native_id}")
}
#[must_use]
pub fn hashed_conversation_id(namespace: uuid::Uuid, parts: &[&str]) -> String {
let joined = parts.join(":");
uuid::Uuid::new_v5(&namespace, joined.as_bytes())
.hyphenated()
.to_string()
}
#[must_use]
pub fn framed_conversation_id(namespace: uuid::Uuid, parts: &[&str]) -> String {
let mut framed = String::new();
for p in parts {
framed.push_str(&p.len().to_string());
framed.push(':');
framed.push_str(p);
}
uuid::Uuid::new_v5(&namespace, framed.as_bytes())
.hyphenated()
.to_string()
}
impl From<IngressDirective> for WireIngressDirective {
fn from(directive: IngressDirective) -> Self {
Self {
budget_cap: directive.budget_cap.unwrap_or_default(),
priority: directive.priority.map_or_else(
|| buffa::EnumValue::from(Priority::PRIORITY_UNSPECIFIED),
buffa::EnumValue::from,
),
required_approver: directive
.required_approver
.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
__buffa_unknown_fields: buffa::UnknownFields::default(),
}
}
}
fn wire_ingress_directive(
directive: IngressDirective,
) -> buffa::MessageField<WireIngressDirective> {
if directive.is_empty() {
return buffa::MessageField::none();
}
buffa::MessageField::some(directive.into())
}
#[must_use]
pub fn encode_messages_for_content_hash(messages: &[Message]) -> Vec<u8> {
use buffa::Message as _;
let mut bytes = Vec::new();
for m in messages {
bytes.extend_from_slice(&m.encode_to_vec());
}
bytes
}
fn issued_unix_ms() -> i64 {
let millis = std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.map_or(0, |d| d.as_millis());
i64::try_from(millis).unwrap_or(i64::MAX)
}
fn build_asserted_attribution(
creds: &EdgeCredentials,
conversation_id: &str,
exec_id: &str,
messages: &[Message],
attribution: &Attribution,
) -> AssertedAttribution {
let content_hash =
polyc_crypto::edge_identity::content_hash_hex(&encode_messages_for_content_hash(messages));
let mut envelope = AssertedAttribution {
edge_id: creds.edge_id().to_owned(),
conversation_id: conversation_id.to_owned(),
nonce: uuid::Uuid::new_v4().to_string(),
issued_unix_ms: issued_unix_ms(),
caller: attribution
.caller
.clone()
.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
participants: attribution.participants.clone(),
signature_hex: String::new(),
exec_id: exec_id.to_owned(),
content_hash,
..Default::default()
};
creds.sign_assertion(&mut envelope);
envelope
}
#[allow(clippy::too_many_arguments)]
fn build_request(
conversation_id: &str,
exec_id: &str,
messages: Vec<Message>,
payment_receipt: Option<PaymentReceipt>,
attribution: &Attribution,
ephemeral_history: bool,
ingress_directive: IngressDirective,
occurrence: &str,
credentials: Option<&EdgeCredentials>,
) -> AgentRequest {
let asserted_attribution = credentials.map(|creds| {
build_asserted_attribution(creds, conversation_id, exec_id, &messages, attribution)
});
AgentRequest {
conversation_id: conversation_id.to_owned(),
exec_id: exec_id.to_owned(),
start: buffa::MessageField::some(AgentStart {
agent_id: String::new(),
agent_config: Vec::new(),
messages,
payment_receipt: payment_receipt
.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
ephemeral_history,
ingress_directive: wire_ingress_directive(ingress_directive),
occurrence: occurrence.to_owned(),
asserted_attribution: asserted_attribution
.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
..Default::default()
}),
..Default::default()
}
}
fn events_from_end(end: AgentEnd) -> Vec<TurnEvent> {
let mut events: Vec<TurnEvent> = end
.pending_approvals
.into_iter()
.map(PendingApprovalPrompt::from)
.map(TurnEvent::from)
.collect();
events.extend(
end.pending_questions
.into_iter()
.map(PendingQuestionPrompt::from)
.map(TurnEvent::from),
);
if let Some(h) = end.handoff.into_option() {
events.push(TurnEvent::HandoffStarted {
child_agent_id: h.child_agent_id,
reason: h.reason,
});
}
for d in end.invite_deliveries {
events.push(TurnEvent::InviteDelivery {
target_user_id: d.target_user_id,
code: d.code,
inviter_display: d.inviter_display,
});
}
if let Some(prompt) = end.wallet_link_prompt.into_option() {
let renewal = prompt.renewal;
let link_url = (!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
events.push(TurnEvent::WalletLinkPrompt { link_url, renewal });
}
if let Some(prompt) = end.persona_credential_prompt.into_option() {
let link_url = (!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
events.push(TurnEvent::PersonaCredentialPrompt { link_url });
}
if let Some(failure) = end.failure.into_option() {
events.push(TurnEvent::TurnFailed {
kind: turn_failure_kind_from_wire(failure.kind.to_i32()),
message: failure.message,
});
}
events.push(TurnEvent::Done);
events
}
const fn turn_failure_kind_from_wire(kind: i32) -> TurnFailureKind {
if kind == WireTurnFailureKind::TURN_FAILURE_KIND_RATE_LIMIT as i32 {
TurnFailureKind::RateLimit
} else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_TIMEOUT as i32 {
TurnFailureKind::Timeout
} else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_UNAVAILABLE as i32 {
TurnFailureKind::Unavailable
} else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_AUTH as i32 {
TurnFailureKind::Auth
} else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_BAD_REQUEST as i32 {
TurnFailureKind::BadRequest
} else {
TurnFailureKind::Other
}
}
fn event_from_compacted(c: ContextCompacted) -> TurnEvent {
let reason = if c.reason.to_i32() == WireCompactionReason::COMPACTION_REASON_SUMMARIZED as i32 {
CompactionReason::Summarized
} else {
CompactionReason::Truncated
};
TurnEvent::ContextCompacted {
reason,
summarized_messages: c.summarized_messages,
summary_preview: c.summary_preview,
}
}
fn message_to_event(msg: Message) -> Option<TurnEvent> {
if msg.internal_only {
return None;
}
let content_block = msg.content.into_option()?;
match content_block.r#type? {
content::Type::ToolCall(tc) => {
let name = match tc.r#type.as_ref() {
Some(tool_call_content::Type::FunctionCall(fc)) if !fc.name.is_empty() => {
fc.name.clone()
}
_ => tc.id.clone(),
};
Some(TurnEvent::ToolStarted { name })
}
content::Type::Text(t) => {
if is_assistant_role(&msg.role) && !t.text.is_empty() {
Some(TurnEvent::TextDelta(t.text))
} else {
None
}
}
_ => None,
}
}
fn is_assistant_role(role: &str) -> bool {
role == "model" || role == "assistant"
}
fn render_content(ty: Option<content::Type>) -> Option<String> {
match ty? {
content::Type::Text(t) => {
if t.text.is_empty() {
None
} else {
Some(t.text)
}
}
content::Type::ToolCall(tc) => {
let name = match tc.r#type.as_ref() {
Some(tool_call_content::Type::FunctionCall(fc)) => fc.name.as_str(),
None => "",
};
if name.is_empty() {
Some(format!("[tool_call:{}]", tc.id))
} else {
Some(format!("[tool_call:{name} {}]", tc.id))
}
}
content::Type::ToolResult(tr) => Some(format!("[tool_result:{}]", tr.call_id)),
_ => None,
}
}
fn aggregate_output_message(
msg: Message,
text_parts: &mut Vec<String>,
scaffolding: &mut Vec<String>,
) {
if msg.internal_only {
return;
}
let is_assistant = matches!(msg.role.as_str(), "model" | "assistant");
let Some(content_block) = msg.content.into_option() else {
return;
};
match content_block.r#type {
Some(content::Type::Text(t)) if is_assistant && !t.text.is_empty() => {
text_parts.push(t.text);
}
Some(content::Type::Text(_)) => {}
other => {
if let Some(rendered) = render_content(other) {
scaffolding.push(rendered);
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct BufferedTurn {
pub reply: String,
pub pending_approvals: Vec<PendingApprovalPrompt>,
pub pending_questions: Vec<PendingQuestionPrompt>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingApprovalPrompt {
pub request_id: String,
pub tool_name: String,
pub title: String,
pub args_json: String,
pub reason: String,
pub resolve_token: String,
pub preview: Option<ApprovalPreview>,
}
impl From<WireAgentPendingApproval> for PendingApprovalPrompt {
fn from(pa: WireAgentPendingApproval) -> Self {
Self {
request_id: pa.request_id,
tool_name: pa.tool_name,
title: pa.title,
args_json: pa.args_json,
reason: pa.reason,
resolve_token: pa.resolve_token,
preview: pa.preview.into_option().map(Into::into),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingQuestionPrompt {
pub call_id: String,
pub index: u32,
pub header: String,
pub question: String,
pub options: Vec<QuestionOptionPrompt>,
pub args_json: String,
pub answer_token: String,
pub already_surfaced: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct QuestionOptionPrompt {
pub label: String,
pub description: String,
pub recommended: bool,
}
impl From<WireAgentQuestionOption> for QuestionOptionPrompt {
fn from(o: WireAgentQuestionOption) -> Self {
Self {
label: o.label,
description: o.description,
recommended: o.recommended,
}
}
}
impl From<QuestionOptionEntry> for QuestionOptionPrompt {
fn from(o: QuestionOptionEntry) -> Self {
Self {
label: o.label,
description: o.description,
recommended: o.recommended,
}
}
}
impl From<WireAgentPendingQuestion> for PendingQuestionPrompt {
fn from(q: WireAgentPendingQuestion) -> Self {
Self {
call_id: q.call_id,
index: q.index,
header: q.header,
question: q.question,
options: q.options.into_iter().map(Into::into).collect(),
args_json: q.args_json,
answer_token: q.answer_token,
already_surfaced: q.already_surfaced,
}
}
}
impl From<PendingQuestionEntry> for PendingQuestionPrompt {
fn from(q: PendingQuestionEntry) -> Self {
Self {
call_id: q.call_id,
index: q.index,
header: q.header,
question: q.question,
options: q.options.into_iter().map(Into::into).collect(),
args_json: q.args_json,
answer_token: q.answer_token,
already_surfaced: q.already_surfaced,
}
}
}
impl From<PendingQuestionPrompt> for TurnEvent {
fn from(p: PendingQuestionPrompt) -> Self {
Self::QuestionPending {
call_id: p.call_id,
index: p.index,
header: p.header,
question: p.question,
options: p.options,
args_json: p.args_json,
answer_token: p.answer_token,
already_surfaced: p.already_surfaced,
}
}
}
impl From<PendingApprovalPrompt> for TurnEvent {
fn from(p: PendingApprovalPrompt) -> Self {
Self::ApprovalPending {
request_id: p.request_id,
tool_name: p.tool_name,
title: p.title,
args_json: p.args_json,
reason: p.reason,
resolve_token: p.resolve_token,
preview: p.preview,
}
}
}
enum WalletLinkPrompt {
None,
Present {
link_url: Option<String>,
renewal: bool,
},
}
fn finalize_buffered_reply(
text_parts: &[String],
scaffolding: &[String],
wallet_link_prompt: WalletLinkPrompt,
persona_credential_prompt: Option<&str>,
) -> String {
if let WalletLinkPrompt::Present { link_url, renewal } = wallet_link_prompt {
return polyc_proto::wallet_link_prompt(link_url.as_deref(), renewal);
}
if let Some(url) = persona_credential_prompt {
return polyc_proto::persona_credential_prompt(Some(url));
}
if text_parts.is_empty() {
scaffolding.join("\n")
} else {
text_parts.join("\n")
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OpsAction {
UpgradeTo {
version: String,
},
EnrollmentNudge {
routine: String,
ceremony_url: String,
},
UpgradeOutcome {
version: String,
outcome: UpgradeOutcomeKind,
},
Unknown,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum UpgradeOutcomeKind {
Success,
Failed,
Unknown,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingNotice {
pub action_id: String,
pub target: String,
pub action: OpsAction,
pub expires_unix: u64,
pub payload_hash: String,
pub delivered: bool,
}
#[derive(Clone)]
pub struct NotificationDialer {
client: Arc<NotificationServiceClient<HttpClient>>,
}
impl NotificationDialer {
pub fn new(addr: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, None, NotificationServiceClient::new)?,
})
}
pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, Some(bearer), NotificationServiceClient::new)?,
})
}
pub async fn poll_pending(&self, provider: &str) -> Result<Vec<PendingNotice>, DialError> {
let request = PollPendingRequest {
provider: provider.to_owned(),
..Default::default()
};
let reply = self
.client
.poll_pending_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.pending.into_iter().map(pending_notice).collect())
}
pub async fn ack(
&self,
provider: &str,
action_id: &str,
target: &str,
) -> Result<bool, DialError> {
let request = AckRequest {
provider: provider.to_owned(),
action_id: action_id.to_owned(),
target: target.to_owned(),
..Default::default()
};
let reply = self
.client
.ack_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.persisted)
}
pub async fn subscribe(
&self,
provider: &str,
) -> Result<impl Stream<Item = Result<PendingNotice, DialError>>, DialError> {
let request = SubscribeRequest {
provider: provider.to_owned(),
..Default::default()
};
let mut stream = self
.client
.subscribe_with_options(request, traced_options())
.await?;
Ok(async_stream::try_stream! {
while let Some(view) = stream.message().await? {
yield pending_notice(view.to_owned_message());
}
})
}
}
fn pending_notice(
p: polyc_proto::proto::polychrome::ops::v1::PendingNotification,
) -> PendingNotice {
let action = p.action.into_option().and_then(|view| view.action).map_or(
OpsAction::Unknown,
|a| match a {
ops_action_view::Action::UpgradeTo(u) => OpsAction::UpgradeTo { version: u.version },
ops_action_view::Action::EnrollmentNudge(n) => OpsAction::EnrollmentNudge {
routine: n.routine,
ceremony_url: n.ceremony_url,
},
ops_action_view::Action::UpgradeOutcome(u) => OpsAction::UpgradeOutcome {
version: u.version,
outcome: match u.kind.as_known() {
Some(upgrade_outcome::Kind::SUCCESS) => UpgradeOutcomeKind::Success,
Some(upgrade_outcome::Kind::FAILED) => UpgradeOutcomeKind::Failed,
Some(
upgrade_outcome::Kind::UNKNOWN | upgrade_outcome::Kind::KIND_UNSPECIFIED,
)
| None => UpgradeOutcomeKind::Unknown,
},
},
},
);
PendingNotice {
action_id: p.action_id,
target: p.target,
action,
expires_unix: p.expires_unix,
payload_hash: p.payload_hash,
delivered: p.delivered,
}
}
#[derive(Clone)]
pub struct RoutineDialer {
client: Arc<RoutineServiceClient<HttpClient>>,
}
impl RoutineDialer {
pub fn new(addr: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, None, RoutineServiceClient::new)?,
})
}
pub async fn fire_routine(
&self,
routine: &str,
occurrence: &str,
actor: ExternalIdentity,
admin_bearer: &str,
) -> Result<FiredRoutine, DialError> {
let request = FireRoutineRequest {
routine: routine.to_owned(),
occurrence: occurrence.to_owned(),
actor: buffa::MessageField::some(actor),
..Default::default()
};
let options = traced_options().with_header(
http::header::AUTHORIZATION,
bearer_header_value(admin_bearer)?,
);
let reply = self
.client
.fire_routine_with_options(request, options)
.await?
.into_owned();
Ok(FiredRoutine {
occurrence: reply.occurrence,
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FiredRoutine {
pub occurrence: String,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CredentialKeyState {
Active,
Retiring,
Revoked,
Unknown,
}
impl From<buffa::EnumValue<WireCredentialKeyState>> for CredentialKeyState {
fn from(state: buffa::EnumValue<WireCredentialKeyState>) -> Self {
match state {
buffa::EnumValue::Known(WireCredentialKeyState::KEY_STATE_ACTIVE) => Self::Active,
buffa::EnumValue::Known(WireCredentialKeyState::KEY_STATE_RETIRING) => Self::Retiring,
buffa::EnumValue::Known(WireCredentialKeyState::KEY_STATE_REVOKED) => Self::Revoked,
buffa::EnumValue::Known(WireCredentialKeyState::KEY_STATE_UNSPECIFIED)
| buffa::EnumValue::Unknown(_) => Self::Unknown,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CredentialKeySummary {
pub kid: String,
pub state: CredentialKeyState,
pub activated_at_ms: u64,
pub not_after_ms: Option<u64>,
pub confirmed_at_ms: Option<u64>,
}
impl From<WireCredentialKeySummary> for CredentialKeySummary {
fn from(wire: WireCredentialKeySummary) -> Self {
Self {
kid: wire.kid,
state: wire.state.into(),
activated_at_ms: wire.activated_at_ms,
not_after_ms: wire.not_after_ms,
confirmed_at_ms: wire.confirmed_at_ms,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CredentialRecordSummary {
pub edge_id: String,
pub principal: String,
pub allowed_namespaces: Vec<String>,
pub keys: Vec<CredentialKeySummary>,
pub revoked_at_ms: Option<u64>,
pub grants_edge: bool,
pub grants_admin: bool,
}
impl From<WireCredentialSummary> for CredentialRecordSummary {
fn from(wire: WireCredentialSummary) -> Self {
Self {
edge_id: wire.edge_id,
principal: wire.principal,
allowed_namespaces: wire.allowed_namespaces,
revoked_at_ms: wire.revoked_at_ms,
grants_edge: wire.grants_edge,
grants_admin: wire.grants_admin,
keys: wire
.keys
.into_iter()
.map(CredentialKeySummary::from)
.collect(),
}
}
}
#[derive(Debug, Clone)]
pub struct CredentialVerifier {
pub kid: String,
pub salt: String,
pub secret_sha256: String,
pub signer_pk_hex: String,
pub not_after_ms: Option<u64>,
}
#[derive(Debug, Clone)]
pub struct CredentialEnrollment {
pub operation_id: String,
pub edge_id: String,
pub principal: String,
pub allowed_namespaces: Vec<String>,
pub key: CredentialVerifier,
pub grants_edge: bool,
pub grants_admin: bool,
}
impl From<CredentialVerifier> for WireCredentialKeyVerifier {
fn from(verifier: CredentialVerifier) -> Self {
Self {
kid: verifier.kid,
salt: verifier.salt,
secret_sha256: verifier.secret_sha256,
signer_pk_hex: verifier.signer_pk_hex,
not_after_ms: verifier.not_after_ms,
__buffa_unknown_fields: buffa::UnknownFields::default(),
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct EnrolledCredential {
pub edge_id: String,
pub kid: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct CredentialSummaryPage {
pub records: Vec<CredentialRecordSummary>,
pub next_after: Option<String>,
pub snapshot_revision: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AddedCredentialKey {
pub edge_id: String,
pub kid: String,
pub retiring_kids: Vec<String>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RetiredCredentialKey {
pub edge_id: String,
pub kid: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RevokedCredential {
pub edge_id: String,
}
#[derive(Clone)]
pub struct CredentialDialer {
client: Arc<CredentialServiceClient<HttpClient>>,
}
impl CredentialDialer {
pub fn new_admin(addr: &str, bearer: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, Some(bearer), CredentialServiceClient::new)?,
})
}
pub async fn list_credentials_page(
&self,
after: Option<&str>,
) -> Result<CredentialSummaryPage, DialError> {
let reply = self
.client
.list_credentials_with_options(
ListCredentialsRequest {
after: after.map(str::to_owned),
limit: 64,
__buffa_unknown_fields: buffa::UnknownFields::default(),
},
traced_options(),
)
.await?
.into_owned();
if reply.records.len() > 64
|| reply.next_after.as_ref().is_some_and(|next| {
reply.records.last().map(|record| &record.edge_id) != Some(next)
|| after.is_some_and(|prior| next.as_str() <= prior)
})
{
return Err(connectrpc::ConnectError::internal(
"credential authority returned an invalid page cursor",
)
.into());
}
Ok(CredentialSummaryPage {
records: reply
.records
.into_iter()
.map(CredentialRecordSummary::from)
.collect(),
next_after: reply.next_after,
snapshot_revision: reply.snapshot_revision,
})
}
pub async fn list_credentials(&self) -> Result<Vec<CredentialRecordSummary>, DialError> {
for _ in 0..3 {
let mut records = Vec::new();
let mut after = None;
let mut revision = None;
let coherent = loop {
let page = self.list_credentials_page(after.as_deref()).await?;
if revision.is_some_and(|expected| expected != page.snapshot_revision) {
break false;
}
revision.get_or_insert(page.snapshot_revision);
records.extend(page.records);
if records.len() > 1_024 {
return Err(connectrpc::ConnectError::internal(
"credential authority exceeded its directory bound",
)
.into());
}
let Some(next) = page.next_after else {
break true;
};
after = Some(next);
};
if coherent {
return Ok(records);
}
}
Err(connectrpc::ConnectError::aborted(
"credential authority changed throughout the bounded listing retry",
)
.into())
}
pub async fn enroll_credential(
&self,
enrollment: CredentialEnrollment,
) -> Result<EnrolledCredential, DialError> {
let request = EnrollCredentialRequest {
operation_id: enrollment.operation_id,
edge_id: enrollment.edge_id,
principal: enrollment.principal,
allowed_namespaces: enrollment.allowed_namespaces,
key: buffa::MessageField::some(enrollment.key.into()),
grants_edge: enrollment.grants_edge,
grants_admin: enrollment.grants_admin,
..Default::default()
};
let reply = self
.client
.enroll_credential_with_options(request, traced_options())
.await?
.into_owned();
Ok(EnrolledCredential {
edge_id: reply.edge_id,
kid: reply.kid,
})
}
pub async fn add_credential_key(
&self,
operation_id: &str,
edge_id: &str,
key: CredentialVerifier,
) -> Result<AddedCredentialKey, DialError> {
let request = AddCredentialKeyRequest {
operation_id: operation_id.to_owned(),
edge_id: edge_id.to_owned(),
key: buffa::MessageField::some(key.into()),
..Default::default()
};
let reply = self
.client
.add_credential_key_with_options(request, traced_options())
.await?
.into_owned();
Ok(AddedCredentialKey {
edge_id: reply.edge_id,
kid: reply.kid,
retiring_kids: reply.retiring_kids,
})
}
pub async fn retire_credential_key(
&self,
operation_id: &str,
edge_id: &str,
kid: &str,
) -> Result<RetiredCredentialKey, DialError> {
let request = RetireCredentialKeyRequest {
operation_id: operation_id.to_owned(),
edge_id: edge_id.to_owned(),
kid: kid.to_owned(),
..Default::default()
};
let reply = self
.client
.retire_credential_key_with_options(request, traced_options())
.await?
.into_owned();
Ok(RetiredCredentialKey {
edge_id: reply.edge_id,
kid: reply.kid,
})
}
pub async fn revoke_credential(
&self,
operation_id: &str,
edge_id: &str,
) -> Result<RevokedCredential, DialError> {
let request = RevokeCredentialRequest {
operation_id: operation_id.to_owned(),
edge_id: edge_id.to_owned(),
..Default::default()
};
let reply = self
.client
.revoke_credential_with_options(request, traced_options())
.await?
.into_owned();
Ok(RevokedCredential {
edge_id: reply.edge_id,
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DecideOutcome {
Applied,
Denied,
Rejected,
Unknown,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DecideResult {
pub outcome: DecideOutcome,
pub detail: String,
}
pub const OPS_PROMPT_HEADING: &str = "Approval needed";
impl OpsAction {
#[must_use]
pub fn summary(&self) -> String {
match self {
Self::UpgradeTo { version } => {
let copy = polyc_runtime::update_copy::update_copy(
&polyc_runtime::compat::Compatibility::Warm,
version,
);
format!("{} — {}", copy.headline, copy.detail)
}
Self::EnrollmentNudge {
routine,
ceremony_url,
} => {
if ceremony_url.is_empty() {
format!(
"The {routine} routine could not run because its approval is missing. \
Ask whoever runs your deployment where to re-approve it."
)
} else {
format!(
"The {routine} routine could not run because its approval is \
missing. Approve it on the enrollment page and the next \
scheduled run will go through: {ceremony_url}"
)
}
}
Self::UpgradeOutcome { version, outcome } => match outcome {
UpgradeOutcomeKind::Success => {
format!("✅ The cluster is now running {version}.")
}
UpgradeOutcomeKind::Failed => format!(
"⚠️ The upgrade to {version} did not confirm healthy — check `polychrome status`."
),
UpgradeOutcomeKind::Unknown => format!(
"❓ Could not confirm whether the upgrade to {version} finished — check `polychrome status`."
),
},
Self::Unknown => "Approve a pending action".to_owned(),
}
}
#[must_use]
pub fn approve_verb(&self) -> Option<&'static str> {
match self {
Self::UpgradeTo { version } => {
polyc_runtime::update_copy::update_copy(
&polyc_runtime::compat::Compatibility::Warm,
version,
)
.action
}
Self::EnrollmentNudge { .. } | Self::UpgradeOutcome { .. } | Self::Unknown => None,
}
}
#[must_use]
pub const fn is_decision(&self) -> bool {
!matches!(
self,
Self::EnrollmentNudge { .. } | Self::UpgradeOutcome { .. }
)
}
}
impl DecideOutcome {
#[must_use]
pub const fn metric_label(&self) -> &'static str {
match self {
Self::Applied => "applied",
Self::Denied => "denied",
Self::Rejected => "rejected",
Self::Unknown => "unknown",
}
}
}
impl DecideResult {
#[must_use]
pub fn decided_line(&self, decider: &str) -> String {
match self.outcome {
DecideOutcome::Applied => format!("✅ Approved by {decider} — applying."),
DecideOutcome::Denied => format!("🚫 Denied by {decider}."),
DecideOutcome::Rejected => {
let why = if self.detail.is_empty() {
"not authorized or no longer valid"
} else {
self.detail.as_str()
};
format!("⛔ Rejected — {why}.")
}
DecideOutcome::Unknown => {
"⚠️ Something went wrong recording that decision — try again.".to_owned()
}
}
}
}
#[derive(Clone)]
pub struct OperatorMailboxDialer {
client: Arc<OperatorMailboxServiceClient<HttpClient>>,
}
impl OperatorMailboxDialer {
pub fn new(addr: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, None, OperatorMailboxServiceClient::new)?,
})
}
pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, Some(bearer), OperatorMailboxServiceClient::new)?,
})
}
pub async fn decide(
&self,
action_id: &str,
approved: bool,
provider: &str,
external_user_id: &str,
reason: &str,
payload_hash: &str,
) -> Result<DecideResult, DialError> {
let request = DecideRequest {
action_id: action_id.to_owned(),
approved,
provider: provider.to_owned(),
external_user_id: external_user_id.to_owned(),
reason: reason.to_owned(),
payload_hash: payload_hash.to_owned(),
..Default::default()
};
let reply = self
.client
.decide_with_options(request, traced_options())
.await?
.into_owned();
let outcome = match reply.outcome.as_known() {
Some(decide_reply::Outcome::APPLIED) => DecideOutcome::Applied,
Some(decide_reply::Outcome::DENIED) => DecideOutcome::Denied,
Some(decide_reply::Outcome::REJECTED) => DecideOutcome::Rejected,
Some(decide_reply::Outcome::OUTCOME_UNSPECIFIED) | None => DecideOutcome::Unknown,
};
Ok(DecideResult {
outcome,
detail: reply.detail,
})
}
}
#[cfg(test)]
mod tests {
#![allow(clippy::pedantic, clippy::nursery, missing_docs)]
use super::*;
use polyc_proto::proto::polychrome::agent::v1::{
Content, FunctionCallContent, TextContent, ThoughtContent, ThoughtSummaryContent,
ToolCallContent, ToolResultContent, thought_summary_content,
};
#[test]
fn expose_nonempty_treats_empty_as_unset() {
assert_eq!(expose_nonempty(None), None);
let empty = Sensitive::new(String::new());
assert_eq!(expose_nonempty(Some(&empty)), None);
let value = Sensitive::new("real-secret".to_owned());
assert_eq!(expose_nonempty(Some(&value)), Some("real-secret"));
}
#[test]
fn ops_copy_is_shared_and_jargon_free() {
let up = OpsAction::UpgradeTo {
version: "1.2.3".to_owned(),
};
let summary = up.summary();
assert!(
summary.contains("1.2.3"),
"summary names the version: {summary}"
);
assert!(
summary.contains("restarts the service"),
"warm upgrade summary is honest about the restart: {summary}"
);
assert_eq!(
up.approve_verb(),
Some(polyc_runtime::update_copy::APPLY_NOW),
"an in-place upgrade carries the shared Apply now verb"
);
assert!(
!summary.to_lowercase().contains("operator"),
"no banned jargon in the upgrade ask: {summary}"
);
assert_eq!(OpsAction::Unknown.approve_verb(), None);
let unknown = OpsAction::Unknown.summary();
assert!(!unknown.to_lowercase().contains("control-plane"));
assert!(!unknown.is_empty());
assert!(!OPS_PROMPT_HEADING.to_lowercase().contains("operator"));
assert_eq!(DecideOutcome::Applied.metric_label(), "applied");
assert_eq!(DecideOutcome::Denied.metric_label(), "denied");
assert_eq!(DecideOutcome::Rejected.metric_label(), "rejected");
assert_eq!(DecideOutcome::Unknown.metric_label(), "unknown");
let applied = DecideResult {
outcome: DecideOutcome::Applied,
detail: String::new(),
}
.decided_line("Chris");
assert!(applied.contains("Approved") && applied.contains("Chris"));
let rejected = DecideResult {
outcome: DecideOutcome::Rejected,
detail: "not authorized to approve".to_owned(),
}
.decided_line("Chris");
assert!(rejected.contains("Rejected") && rejected.contains("not authorized to approve"));
let unknown = DecideResult {
outcome: DecideOutcome::Unknown,
detail: String::new(),
}
.decided_line("Chris");
assert!(!unknown.to_lowercase().contains("please"));
}
#[test]
fn enrollment_nudge_copy_is_a_link_not_a_decision() {
let nudge = OpsAction::EnrollmentNudge {
routine: "standup".to_owned(),
ceremony_url: "https://enroll.polychrome.test/v1/enroll/manage/standup/persona-9"
.to_owned(),
};
assert!(!nudge.is_decision(), "a nudge carries no approve/deny");
assert_eq!(nudge.approve_verb(), None);
let summary = nudge.summary();
assert!(summary.contains("standup"), "names the routine: {summary}");
assert!(
summary.contains("https://enroll.polychrome.test/v1/enroll/manage/standup/persona-9"),
"carries the exact ceremony link: {summary}"
);
let lower = summary.to_lowercase();
for banned in [
"operator",
"sub-agent",
"trifecta",
"context budget",
"state-changing action",
"please",
"sorry",
"unfortunately",
] {
assert!(
!lower.contains(banned),
"banned word {banned:?} in: {summary}"
);
}
assert!(
OpsAction::UpgradeTo {
version: "1.2.3".to_owned()
}
.is_decision()
);
assert!(OpsAction::Unknown.is_decision());
}
#[test]
fn enrollment_nudge_omits_the_link_when_composed_under_a_loopback_base() {
let nudge = OpsAction::EnrollmentNudge {
routine: "standup".to_owned(),
ceremony_url: String::new(),
};
assert!(!nudge.is_decision(), "still a plain DM, not a decision");
let summary = nudge.summary();
assert!(
summary.contains("standup"),
"still names the routine: {summary}"
);
assert!(
!summary.contains("http"),
"no link of any scheme ships when the base was unreachable: {summary}"
);
let lower = summary.to_lowercase();
for banned in [
"operator",
"sub-agent",
"trifecta",
"context budget",
"state-changing action",
"please",
"sorry",
"unfortunately",
] {
assert!(
!lower.contains(banned),
"banned word {banned:?} in: {summary}"
);
}
}
#[test]
fn approval_choice_flags() {
assert!(ApprovalChoice::Approve.approved());
assert!(!ApprovalChoice::Approve.approved_for_session());
assert!(!ApprovalChoice::Approve.is_abort());
assert!(ApprovalChoice::ApproveForSession.approved());
assert!(ApprovalChoice::ApproveForSession.approved_for_session());
assert!(!ApprovalChoice::Deny.approved());
assert!(!ApprovalChoice::Deny.is_abort());
assert!(!ApprovalChoice::Abort.approved());
assert!(ApprovalChoice::Abort.is_abort());
}
#[test]
fn approval_completed_text_reports_the_runtime_outcome() {
let done = approval_completed_text("Remove @vitor's admin role", "Chris", true);
assert!(done.contains("Chris"), "names the decider: {done}");
assert!(
done.contains("Remove @vitor's admin role"),
"names the tool label: {done}"
);
assert!(done.contains("done"), "a success reads as done: {done}");
let failed = approval_completed_text("Remove @vitor's admin role", "Chris", false);
assert_ne!(
done, failed,
"success and failure must not read identically"
);
assert!(
failed.contains("error"),
"a failed run must say so, not claim success: {failed}"
);
for copy in [&done, &failed] {
let lower = copy.to_lowercase();
for banned in ["please", "sorry", "unfortunately", "operator"] {
assert!(
!lower.contains(banned),
"banned word {banned:?} in {copy:?}"
);
}
}
}
#[test]
fn compacted_summarized_maps_with_preview() {
let c = ContextCompacted {
reason: buffa::EnumValue::from(
WireCompactionReason::COMPACTION_REASON_SUMMARIZED as i32,
),
summarized_messages: 12,
summary_preview: "earlier work: fixed bug X".to_owned(),
..Default::default()
};
assert_eq!(
event_from_compacted(c),
TurnEvent::ContextCompacted {
reason: CompactionReason::Summarized,
summarized_messages: 12,
summary_preview: "earlier work: fixed bug X".to_owned(),
}
);
}
#[test]
fn compacted_truncated_maps_without_preview() {
let c = ContextCompacted {
reason: buffa::EnumValue::from(
WireCompactionReason::COMPACTION_REASON_TRUNCATED as i32,
),
..Default::default()
};
assert_eq!(
event_from_compacted(c),
TurnEvent::ContextCompacted {
reason: CompactionReason::Truncated,
summarized_messages: 0,
summary_preview: String::new(),
}
);
}
#[test]
fn compacted_unknown_reason_falls_back_to_truncated() {
let c = ContextCompacted {
reason: buffa::EnumValue::from(
WireCompactionReason::COMPACTION_REASON_UNSPECIFIED as i32,
),
..Default::default()
};
assert!(matches!(
event_from_compacted(c),
TurnEvent::ContextCompacted {
reason: CompactionReason::Truncated,
..
}
));
}
#[test]
fn dial_error_retryable_classification() {
use connectrpc::ErrorCode;
for code in [
ErrorCode::Unavailable,
ErrorCode::DeadlineExceeded,
ErrorCode::ResourceExhausted,
ErrorCode::Aborted,
] {
let err = DialError::Connect(connectrpc::ConnectError::new(code, "x"));
assert_eq!(err.code(), Some(code));
assert!(err.is_retryable(), "{code:?} should be retryable");
}
for code in [
ErrorCode::InvalidArgument,
ErrorCode::Unauthenticated,
ErrorCode::NotFound,
ErrorCode::PermissionDenied,
ErrorCode::Internal,
] {
let err = DialError::Connect(connectrpc::ConnectError::new(code, "x"));
assert_eq!(err.code(), Some(code));
assert!(!err.is_retryable(), "{code:?} should not be retryable");
}
let bad_addr = DialError::InvalidAddress {
addr: "http://a b".to_owned(),
source: "http://a b".parse::<http::Uri>().unwrap_err(),
};
assert_eq!(bad_addr.code(), None);
assert!(!bad_addr.is_retryable());
let tls = DialError::Tls("no provider".to_owned());
assert_eq!(tls.code(), None);
assert!(!tls.is_retryable());
}
#[test]
fn deadline_exceeded_is_the_only_code_that_reads_as_a_routine_recycle() {
let deadline = DialError::Connect(connectrpc::ConnectError::deadline_exceeded("idle"));
assert!(deadline.is_deadline_exceeded());
let unavailable = DialError::Connect(connectrpc::ConnectError::unavailable("gone"));
assert!(
!unavailable.is_deadline_exceeded(),
"a genuinely dropped stream must still be reported as a failure"
);
let local = DialError::Tls("no provider".to_owned());
assert!(
!local.is_deadline_exceeded(),
"a local setup failure has no Connect code and is never a routine recycle"
);
}
fn text(s: &str) -> Option<content::Type> {
Some(content::Type::Text(Box::new(TextContent {
text: s.to_owned(),
..Default::default()
})))
}
fn tool_call(id: &str, name: &str) -> Option<content::Type> {
Some(content::Type::ToolCall(Box::new(ToolCallContent {
id: id.to_owned(),
r#type: Some(tool_call_content::Type::FunctionCall(Box::new(
FunctionCallContent {
name: name.to_owned(),
..Default::default()
},
))),
..Default::default()
})))
}
fn tool_result(call_id: &str) -> Option<content::Type> {
Some(content::Type::ToolResult(Box::new(ToolResultContent {
call_id: call_id.to_owned(),
..Default::default()
})))
}
fn thought(summary: &str) -> Option<content::Type> {
Some(content::Type::Thought(Box::new(ThoughtContent {
summary: vec![ThoughtSummaryContent {
r#type: Some(thought_summary_content::Type::Text(Box::new(TextContent {
text: summary.to_owned(),
..Default::default()
}))),
..Default::default()
}],
..Default::default()
})))
}
#[test]
fn renders_text_verbatim() {
assert_eq!(render_content(text("hi")), Some("hi".to_owned()));
}
#[test]
fn renders_tool_call_with_name_and_id() {
assert_eq!(
render_content(tool_call("call_42", "search")),
Some("[tool_call:search call_42]".to_owned())
);
}
#[test]
fn renders_tool_call_without_function_name() {
assert_eq!(
render_content(Some(content::Type::ToolCall(Box::new(ToolCallContent {
id: "call_bare".to_owned(),
r#type: None,
..Default::default()
})))),
Some("[tool_call:call_bare]".to_owned())
);
}
#[test]
fn renders_tool_result_by_call_id() {
assert_eq!(
render_content(tool_result("call_42")),
Some("[tool_result:call_42]".to_owned())
);
}
#[test]
fn reasoning_is_never_the_reply() {
assert_eq!(render_content(thought("considering options")), None);
}
#[test]
fn thought_only_turn_yields_empty_reply() {
assert_eq!(aggregate(vec![thought("secret reasoning")]), "");
}
#[test]
fn empty_text_skipped() {
assert_eq!(render_content(text("")), None);
}
#[test]
fn unknown_variant_skipped() {
assert_eq!(render_content(None), None);
}
fn aggregate(blocks: Vec<Option<content::Type>>) -> String {
let mut parts: Vec<String> = Vec::new();
for b in blocks {
if let Some(s) = render_content(b) {
parts.push(s);
}
}
parts.join("\n")
}
#[test]
fn aggregate_pure_text_turn() {
assert_eq!(
aggregate(vec![text("hello"), text("world")]),
"hello\nworld"
);
}
#[test]
fn aggregate_tool_call_only_turn() {
assert_eq!(
aggregate(vec![tool_call("call_1", "lookup")]),
"[tool_call:lookup call_1]"
);
}
#[test]
fn aggregate_mixed_text_and_tool_call() {
assert_eq!(
aggregate(vec![text("thinking..."), tool_call("call_1", "search")]),
"thinking...\n[tool_call:search call_1]"
);
}
#[test]
fn buffered_aggregation_skips_internal_only_messages() {
let mut withheld = message("model", text("pending your approval"));
withheld.internal_only = true;
let mut text_parts = Vec::new();
let mut scaffolding = Vec::new();
aggregate_output_message(withheld, &mut text_parts, &mut scaffolding);
assert!(
text_parts.is_empty(),
"withheld text must not become the reply"
);
assert!(
scaffolding.is_empty(),
"withheld text must not fall back to scaffolding either"
);
}
#[test]
fn buffered_aggregation_keeps_visible_assistant_text() {
let mut text_parts = Vec::new();
let mut scaffolding = Vec::new();
aggregate_output_message(
message("model", text("the answer")),
&mut text_parts,
&mut scaffolding,
);
assert_eq!(text_parts, vec!["the answer".to_owned()]);
assert!(scaffolding.is_empty());
}
#[test]
fn finalize_buffered_reply_wallet_link_prompt_wins_with_url() {
let url = "https://polychrome.example/link/abc";
let reply = finalize_buffered_reply(
&[],
&[],
WalletLinkPrompt::Present {
link_url: Some(url.to_owned()),
renewal: false,
},
None,
);
assert_eq!(reply, polyc_proto::wallet_link_prompt(Some(url), false));
assert!(
reply.contains(url),
"the deterministic prompt carries the link: {reply}"
);
}
#[test]
fn finalize_buffered_reply_wallet_link_prompt_wins_without_url() {
let reply = finalize_buffered_reply(
&[],
&[],
WalletLinkPrompt::Present {
link_url: None,
renewal: false,
},
None,
);
assert_eq!(reply, polyc_proto::wallet_link_prompt(None, false));
}
#[test]
fn finalize_buffered_reply_wallet_link_prompt_renewal_wins() {
let url = "https://polychrome.example/link/abc";
let reply = finalize_buffered_reply(
&[],
&[],
WalletLinkPrompt::Present {
link_url: Some(url.to_owned()),
renewal: true,
},
None,
);
assert_eq!(reply, polyc_proto::wallet_link_prompt(Some(url), true));
assert!(reply.to_lowercase().contains("expired"), "{reply}");
assert!(
!reply.to_lowercase().contains("no spending wallet set up"),
"{reply}"
);
}
#[test]
fn finalize_buffered_reply_wallet_link_prompt_replaces_preceding_content() {
let url = "https://polychrome.example/link/abc";
let reply = finalize_buffered_reply(
&["It looks like you'll need to link a wallet first.".to_owned()],
&[],
WalletLinkPrompt::Present {
link_url: Some(url.to_owned()),
renewal: false,
},
None,
);
assert_eq!(
reply,
polyc_proto::wallet_link_prompt(Some(url), false),
"preceding text_parts must not survive — it may be the model's own \
unfiltered wallet-link narration, not genuine unrelated content: {reply}"
);
}
#[test]
fn finalize_buffered_reply_without_wallet_link_prompt_is_unchanged() {
assert_eq!(
finalize_buffered_reply(
&["the answer".to_owned()],
&[],
WalletLinkPrompt::None,
None
),
"the answer"
);
assert_eq!(
finalize_buffered_reply(
&[],
&["[tool_call:foo]".to_owned()],
WalletLinkPrompt::None,
None
),
"[tool_call:foo]"
);
assert_eq!(
finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, None),
""
);
}
#[test]
fn finalize_buffered_reply_persona_credential_prompt_wins() {
let url = "https://wallet.polychrome.example/enroll-passkey?token=abc";
let reply = finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, Some(url));
assert_eq!(reply, polyc_proto::persona_credential_prompt(Some(url)));
assert!(
reply.contains(url),
"the deterministic prompt carries the link: {reply}"
);
}
#[test]
fn finalize_buffered_reply_persona_credential_prompt_replaces_preceding_content() {
let url = "https://wallet.polychrome.example/enroll-passkey?token=abc";
let reply = finalize_buffered_reply(
&["You can approve purchases faster by setting up a passkey.".to_owned()],
&[],
WalletLinkPrompt::None,
Some(url),
);
assert_eq!(
reply,
polyc_proto::persona_credential_prompt(Some(url)),
"preceding text_parts must not survive — it may be the model's own \
unfiltered narration of the same link: {reply}"
);
}
fn message(role: &str, ty: Option<content::Type>) -> Message {
Message {
role: role.to_owned(),
content: buffa::MessageField::some(Content {
r#type: ty,
..Default::default()
}),
..Default::default()
}
}
#[test]
fn assistant_text_becomes_text_delta() {
assert_eq!(
message_to_event(message("assistant", text("hello"))),
Some(TurnEvent::TextDelta("hello".to_owned()))
);
}
#[test]
fn model_role_also_counts_as_assistant() {
assert_eq!(
message_to_event(message("model", text("hi"))),
Some(TurnEvent::TextDelta("hi".to_owned()))
);
}
#[test]
fn tool_role_text_is_skipped() {
assert_eq!(message_to_event(message("tool", text("result blob"))), None);
}
#[test]
fn empty_assistant_text_is_skipped() {
assert_eq!(message_to_event(message("assistant", text(""))), None);
}
#[test]
fn internal_only_assistant_text_is_skipped() {
let mut msg = message("assistant", text("pending your approval"));
msg.internal_only = true;
assert_eq!(message_to_event(msg), None);
}
#[test]
fn tool_call_becomes_tool_started_with_name() {
assert_eq!(
message_to_event(message("model", tool_call("call_7", "search"))),
Some(TurnEvent::ToolStarted {
name: "search".to_owned()
})
);
}
#[test]
fn tool_call_falls_back_to_call_id() {
assert_eq!(
message_to_event(message(
"tool",
Some(content::Type::ToolCall(Box::new(ToolCallContent {
id: "call_bare".to_owned(),
r#type: None,
..Default::default()
})))
)),
Some(TurnEvent::ToolStarted {
name: "call_bare".to_owned()
})
);
}
#[test]
fn tool_result_produces_no_event() {
assert_eq!(
message_to_event(message("tool", tool_result("call_7"))),
None
);
}
fn map_turn(msgs: Vec<Message>) -> Vec<TurnEvent> {
let mut events: Vec<TurnEvent> = msgs.into_iter().filter_map(message_to_event).collect();
events.push(TurnEvent::Done);
events
}
#[test]
fn synthetic_turn_yields_expected_event_sequence() {
let turn = vec![
message("model", text("Let me look that up.")),
message("tool", tool_call("call_1", "search")),
message("model", text("Found it.")),
];
assert_eq!(
map_turn(turn),
vec![
TurnEvent::TextDelta("Let me look that up.".to_owned()),
TurnEvent::ToolStarted {
name: "search".to_owned()
},
TurnEvent::TextDelta("Found it.".to_owned()),
TurnEvent::Done,
]
);
}
use polyc_proto::proto::polychrome::agent::v1::{
Handoff as WireHandoff, PendingApproval as WirePendingApproval,
};
#[test]
fn end_with_nothing_yields_only_done() {
assert_eq!(events_from_end(AgentEnd::default()), vec![TurnEvent::Done]);
}
#[test]
fn end_with_handoff_yields_handoff_then_done() {
let end = AgentEnd {
handoff: buffa::MessageField::some(WireHandoff {
call_id: "call_1".to_owned(),
child_agent_id: "researcher".to_owned(),
reason: "needs deep dive".to_owned(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::HandoffStarted {
child_agent_id: "researcher".to_owned(),
reason: "needs deep dive".to_owned(),
},
TurnEvent::Done,
]
);
}
#[test]
fn end_orders_approvals_before_handoff_before_done() {
let end = AgentEnd {
pending_approvals: vec![WirePendingApproval {
request_id: "r1".to_owned(),
tool_name: "delete_file".to_owned(),
args_json: "{}".to_owned(),
title: "Delete a file".to_owned(),
..Default::default()
}],
handoff: buffa::MessageField::some(WireHandoff {
child_agent_id: "child".to_owned(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::ApprovalPending {
request_id: "r1".to_owned(),
tool_name: "delete_file".to_owned(),
title: "Delete a file".to_owned(),
args_json: "{}".to_owned(),
reason: String::new(),
resolve_token: String::new(),
preview: None,
},
TurnEvent::HandoffStarted {
child_agent_id: "child".to_owned(),
reason: String::new(),
},
TurnEvent::Done,
]
);
}
#[test]
fn end_projects_invite_deliveries_after_handoff_before_done() {
use polyc_proto::proto::polychrome::agent::v1::InviteDelivery as WireInviteDelivery;
let end = AgentEnd {
invite_deliveries: vec![WireInviteDelivery {
target_user_id: "UVITOR".to_owned(),
code: "482913".to_owned(),
inviter_display: "Ada".to_owned(),
..Default::default()
}],
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::InviteDelivery {
target_user_id: "UVITOR".to_owned(),
code: "482913".to_owned(),
inviter_display: "Ada".to_owned(),
},
TurnEvent::Done,
]
);
}
#[test]
fn end_projects_wallet_link_prompt_after_invite_deliveries_before_done() {
use polyc_proto::proto::polychrome::agent::v1::WalletLinkPrompt as WireWalletLinkPrompt;
let end = AgentEnd {
wallet_link_prompt: buffa::MessageField::some(WireWalletLinkPrompt {
link_url: "https://polychrome.example/link/abc".to_owned(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::WalletLinkPrompt {
link_url: Some("https://polychrome.example/link/abc".to_owned()),
renewal: false,
},
TurnEvent::Done,
]
);
}
#[test]
fn end_projects_wallet_link_prompt_without_url_as_none() {
use polyc_proto::proto::polychrome::agent::v1::WalletLinkPrompt as WireWalletLinkPrompt;
let end = AgentEnd {
wallet_link_prompt: buffa::MessageField::some(WireWalletLinkPrompt {
link_url: String::new(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::WalletLinkPrompt {
link_url: None,
renewal: false,
},
TurnEvent::Done,
]
);
}
#[test]
fn end_projects_wallet_link_prompt_renewal_bit() {
use polyc_proto::proto::polychrome::agent::v1::WalletLinkPrompt as WireWalletLinkPrompt;
let end = AgentEnd {
wallet_link_prompt: buffa::MessageField::some(WireWalletLinkPrompt {
link_url: "https://polychrome.example/link/abc".to_owned(),
renewal: true,
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::WalletLinkPrompt {
link_url: Some("https://polychrome.example/link/abc".to_owned()),
renewal: true,
},
TurnEvent::Done,
]
);
}
#[test]
fn end_projects_persona_credential_prompt_before_done() {
use polyc_proto::proto::polychrome::agent::v1::PersonaCredentialPrompt as WirePersonaCredentialPrompt;
let end = AgentEnd {
persona_credential_prompt: buffa::MessageField::some(WirePersonaCredentialPrompt {
link_url: "https://wallet.polychrome.example/enroll-passkey?token=abc".to_owned(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::PersonaCredentialPrompt {
link_url: Some(
"https://wallet.polychrome.example/enroll-passkey?token=abc".to_owned()
),
},
TurnEvent::Done,
]
);
}
#[test]
fn end_without_persona_credential_prompt_yields_only_done() {
let end = AgentEnd::default();
assert_eq!(events_from_end(end), vec![TurnEvent::Done]);
}
#[test]
fn namespaced_id_is_prefix_colon_native() {
assert_eq!(
namespaced_id("mail", "CAF=abc@mail.example"),
"mail:CAF=abc@mail.example"
);
assert_eq!(namespaced_id("web", "abc-123"), "web:abc-123");
}
#[test]
fn hashed_conversation_id_is_deterministic_v5() {
let ns = uuid::Uuid::from_u128(0x1234_5678_9abc_4def_8123_4567_89ab_cdef);
let a = hashed_conversation_id(ns, &["T1", "C1", "169.000"]);
let b = hashed_conversation_id(ns, &["T1", "C1", "169.000"]);
assert_eq!(a, b);
assert_ne!(a, hashed_conversation_id(ns, &["T1", "C2", "169.000"]));
let parsed: uuid::Uuid = a.parse().unwrap();
assert_eq!(parsed.get_version_num(), 5);
}
#[test]
fn framed_conversation_id_resists_separator_collisions() {
let ns = uuid::Uuid::from_u128(0x99);
assert_ne!(
framed_conversation_id(ns, &["a:b", "c"]),
framed_conversation_id(ns, &["a", "b:c"]),
);
assert_eq!(
hashed_conversation_id(ns, &["a:b", "c"]),
hashed_conversation_id(ns, &["a", "b:c"]),
);
let id = framed_conversation_id(ns, &["mail", "<abc@x>"]);
assert_eq!(id, framed_conversation_id(ns, &["mail", "<abc@x>"]));
assert_eq!(id.parse::<uuid::Uuid>().unwrap().get_version_num(), 5);
}
#[test]
fn hashed_conversation_id_matches_slacks_inline_algorithm() {
let ns = uuid::Uuid::from_u128(0xa1b2_c3d4_e5f6_4789_abcd_ef01_2345_6789);
let parts = ["T01234ABCD", "C0000FAKEID", "1700000000.000100"];
let inline = uuid::Uuid::new_v5(&ns, parts.join(":").as_bytes())
.hyphenated()
.to_string();
assert_eq!(hashed_conversation_id(ns, &parts), inline);
}
#[test]
fn link_outcome_messages_honor_presentation_rules() {
assert!(
LinkCeremony::InvalidOrExpired
.user_message()
.contains("invalid or expired")
);
assert!(LinkCeremony::Throttled.user_message().contains("wait"));
assert!(
LinkCeremony::Linked {
persona_id: "p".to_owned()
}
.user_message()
.contains("Linked")
);
}
fn sample_attribution() -> Attribution {
Attribution {
caller: Some(ExternalIdentity {
provider: "slack".to_owned(),
scope: "team-1".to_owned(),
external_id: "U1".to_owned(),
display_name: "Ada".to_owned(),
..Default::default()
}),
participants: vec![ExternalIdentity {
provider: "slack".to_owned(),
scope: "team-1".to_owned(),
external_id: "U2".to_owned(),
display_name: "Bea".to_owned(),
..Default::default()
}],
}
}
#[test]
fn build_request_without_credentials_carries_no_asserted_attribution() {
let request = build_request(
"slack:team-1:general",
"exec-1",
vec![text_message("user", "hi")],
None,
&sample_attribution(),
false,
IngressDirective::default(),
"",
None,
);
let start = request.start.into_option().expect("start set");
assert!(
start.asserted_attribution.into_option().is_none(),
"an unauthenticated dialer's request must carry no asserted_attribution envelope"
);
}
#[test]
fn build_request_with_credentials_signs_a_verifiable_envelope() {
let key_bytes = [7u8; 32];
let signer_pk = polyc_crypto::Signer::from_key_bytes(&key_bytes)
.expect("valid test key material")
.public_key_bytes();
let creds = EdgeCredentials::from_parts(
"slack".to_owned(),
"pc_slack_test-secret".to_owned(),
&polyc_crypto::hex::lower(&key_bytes),
)
.expect("valid test credentials");
let attribution = sample_attribution();
let request = build_request(
"slack:team-1:general",
"exec-1",
vec![text_message("user", "hi")],
None,
&attribution,
false,
IngressDirective::default(),
"",
Some(&creds),
);
let start = request.start.into_option().expect("start set");
let envelope = start
.asserted_attribution
.into_option()
.expect("a credentialed dial signs an asserted_attribution envelope");
assert_eq!(envelope.edge_id, "slack");
assert_eq!(envelope.conversation_id, "slack:team-1:general");
assert!(!envelope.nonce.is_empty(), "nonce must be set per turn");
assert!(envelope.issued_unix_ms > 0);
assert!(
polyc_crypto::edge_identity::verify_edge_assertion(&signer_pk, &envelope),
"the signed envelope must verify under the credential's own public key"
);
assert_eq!(
envelope.caller.into_option().map(|c| c.external_id),
attribution.caller.map(|c| c.external_id),
"the envelope's caller must match the turn's attribution"
);
assert_eq!(
envelope
.participants
.into_iter()
.map(|p| p.external_id)
.collect::<Vec<_>>(),
attribution
.participants
.into_iter()
.map(|p| p.external_id)
.collect::<Vec<_>>(),
"the envelope's participants must match the turn's attribution"
);
assert_eq!(
envelope.exec_id, "exec-1",
"the envelope must bind this turn's exec_id"
);
assert_eq!(
envelope.content_hash,
polyc_crypto::edge_identity::content_hash_hex(&encode_messages_for_content_hash(&[
text_message("user", "hi")
])),
"the envelope's content_hash must match the turn's messages"
);
}
const TEST_ADDR: &str = "http://127.0.0.1:0";
fn approval_credentials() -> (EdgeCredentials, Vec<u8>) {
let key_bytes = [11u8; 32];
let public_key = polyc_crypto::Signer::from_key_bytes(&key_bytes)
.expect("valid test key material")
.public_key_bytes();
let creds = EdgeCredentials::from_parts(
"slack".to_owned(),
"pc_slack_test-secret".to_owned(),
&polyc_crypto::hex::lower(&key_bytes),
)
.expect("valid test credentials");
(creds, public_key)
}
fn sample_responder() -> ExternalIdentity {
ExternalIdentity {
provider: "slack".to_owned(),
scope: "team-1".to_owned(),
external_id: "U1".to_owned(),
display_name: "Ada".to_owned(),
..Default::default()
}
}
fn approve_naming_a_responder(dialer: &ApprovalDialer) -> ApprovalResponseRequest {
dialer.respond_request(
"req-1",
ApprovalChoice::Approve,
"looks right",
"slack:team-1:general",
r#"{"path":"/tmp/a"}"#,
"context",
"resolve-token-1",
Some(sample_responder()),
)
}
#[test]
fn a_credentialed_dialer_signs_a_verifiable_assertion() {
let (creds, public_key) = approval_credentials();
let dialer = ApprovalDialer::with_credentials(TEST_ADDR, Arc::new(creds))
.expect("a credentialed approval dialer builds");
let request = approve_naming_a_responder(&dialer);
let assertion = request
.asserted_approval
.as_option()
.expect("a credentialed edge naming a responder asserts one");
assert_eq!(assertion.edge_id, "slack");
assert_eq!(
assertion
.responder
.as_option()
.map(|r| r.external_id.as_str()),
Some("U1"),
"the assertion must carry the responder the edge named"
);
assert!(
polyc_crypto::approval_assertion::verify_approval_assertion(
[public_key.as_slice()],
&request
),
"the assertion must verify under the edge's own identity key"
);
}
#[test]
fn a_decision_naming_no_responder_asserts_none() {
let (creds, _) = approval_credentials();
let dialer = ApprovalDialer::with_credentials(TEST_ADDR, Arc::new(creds))
.expect("a credentialed approval dialer builds");
let request = dialer.respond_request(
"req-1",
ApprovalChoice::Deny,
"no",
"slack:team-1:general",
"",
"",
"resolve-token-1",
None,
);
assert!(
request.asserted_approval.as_option().is_none(),
"no responder means no assertion, not an assertion of nobody"
);
}
#[test]
fn an_uncredentialed_dialer_asserts_none() {
for (label, dialer) in [
(
"new",
ApprovalDialer::new(TEST_ADDR).expect("an unauthenticated dialer builds"),
),
(
"with_bearer",
ApprovalDialer::with_bearer(TEST_ADDR, "pc_slack_test-secret")
.expect("a bearer-only dialer builds"),
),
] {
let request = approve_naming_a_responder(&dialer);
assert!(
request.asserted_approval.as_option().is_none(),
"{label}: an uncredentialed dialer must not put an unsigned responder on the wire"
);
}
}
#[test]
fn the_assertion_covers_every_field_of_the_final_request() {
let (creds, public_key) = approval_credentials();
let dialer = ApprovalDialer::with_credentials(TEST_ADDR, Arc::new(creds))
.expect("a credentialed approval dialer builds");
let keys = || [public_key.as_slice()];
type Mutation = (&'static str, fn(&mut ApprovalResponseRequest));
let mutations: Vec<Mutation> = vec![
("request_id", |r| r.request_id = "req-2".to_owned()),
("conversation_id", |r| {
r.conversation_id = "slack:team-1:secrets".to_owned();
}),
("resolve_token", |r| {
r.resolve_token = "resolve-token-2".to_owned();
}),
("decision", |r| {
use polyc_proto::proto::polychrome::approval::v1::{
Deny, approval_response_request::Decision,
};
r.decision = Some(Decision::Deny(Box::new(Deny {
reason: "looks right".to_owned(),
abort: false,
__buffa_unknown_fields: buffa::UnknownFields::default(),
})));
}),
("modified_args_json", |r| {
use polyc_proto::proto::polychrome::approval::v1::approval_response_request::Decision;
let Some(Decision::Approve(approve)) = r.decision.as_mut() else {
panic!("the builder produced an Approve");
};
approve.modified_args_json = r#"{"path":"/etc/shadow"}"#.to_owned();
}),
("responder", |r| {
r.asserted_approval
.as_option_mut()
.expect("the request carries an assertion")
.responder
.as_option_mut()
.expect("the assertion carries a responder")
.external_id = "U0EVE".to_owned();
}),
("edge_id", |r| {
r.asserted_approval
.as_option_mut()
.expect("the request carries an assertion")
.edge_id = "messaging".to_owned();
}),
];
for (field, mutate) in mutations {
let mut request = approve_naming_a_responder(&dialer);
assert!(
polyc_crypto::approval_assertion::verify_approval_assertion(keys(), &request),
"{field}: the unmutated request must verify, or this case proves nothing"
);
mutate(&mut request);
assert!(
!polyc_crypto::approval_assertion::verify_approval_assertion(keys(), &request),
"{field}: changing it after signing must invalidate the assertion"
);
}
}
#[test]
fn an_approval_dialer_inherits_the_agent_dialers_edge_identity() {
let (creds, public_key) = approval_credentials();
let agent = AgentDialer::with_credentials(TEST_ADDR, creds)
.expect("a credentialed agent dialer builds");
let approval = agent
.approval_dialer_with_credentials(TEST_ADDR)
.expect("the approval sibling builds");
assert!(
polyc_crypto::approval_assertion::verify_approval_assertion(
[public_key.as_slice()],
&approve_naming_a_responder(&approval)
),
"the approval dialer must sign with the SAME key the agent dialer dispatches with"
);
let plain = AgentDialer::new(TEST_ADDR)
.expect("an unauthenticated agent dialer builds")
.approval_dialer_with_credentials(TEST_ADDR)
.expect("the approval sibling builds");
assert!(
approve_naming_a_responder(&plain)
.asserted_approval
.as_option()
.is_none(),
"an unauthenticated agent dialer has no key to lend, so its approvals assert nobody"
);
}
#[test]
fn with_credentials_rejects_a_bearer_that_is_not_a_valid_header_value() {
let creds = EdgeCredentials::from_parts(
"slack".to_owned(),
"pc_slack_bad\nbearer".to_owned(),
&polyc_crypto::hex::lower(&[9u8; 32]),
)
.expect("valid test credentials");
let err = match AgentDialer::with_credentials("http://127.0.0.1:0", creds) {
Ok(_) => panic!("a newline in the bearer must fail the dial closed"),
Err(err) => err,
};
assert!(matches!(err, DialError::InvalidBearer(_)));
}
#[test]
fn bearer_header_encodes_a_valid_bearer() {
let headers = bearer_header("pc_slack_good").expect("a valid bearer encodes");
let value = headers
.get(http::header::AUTHORIZATION)
.expect("authorization header is set");
assert_eq!(value, "Bearer pc_slack_good");
assert!(
value.is_sensitive(),
"the bearer header must be marked sensitive so the HTTP stack \
never logs it or HPACK-indexes it"
);
}
#[test]
fn bearer_header_rejects_a_bearer_that_is_not_a_valid_header_value() {
let err =
bearer_header("pc_slack_bad\nbearer").expect_err("a newline must fail the dial closed");
assert!(matches!(err, DialError::InvalidBearer(_)));
}
#[test]
fn bearer_header_value_is_sensitive() {
let value = bearer_header_value("pc_admin_good").expect("a valid bearer encodes");
assert_eq!(value, "Bearer pc_admin_good");
assert!(
value.is_sensitive(),
"RoutineDialer::fire_routine's per-call admin bearer must be \
marked sensitive, same as every other dialer's bearer header"
);
}
#[test]
fn every_bearer_dialer_routes_through_bearer_header() {
const ADDR: &str = "http://127.0.0.1:0";
const GOOD: &str = "pc_slack_good";
const BAD: &str = "pc_slack_bad\nbearer";
type Build = fn(&str, &str) -> Result<(), DialError>;
let dialers: &[(&str, Build)] = &[
("ApprovalDialer", |a, b| {
ApprovalDialer::with_bearer(a, b).map(|_| ())
}),
("ApprovalDialer::with_credentials", |a, b| {
let creds = EdgeCredentials::from_parts(
"slack".to_owned(),
b.to_owned(),
&polyc_crypto::hex::lower(&[7u8; 32]),
)
.expect("valid test credentials");
ApprovalDialer::with_credentials(a, Arc::new(creds)).map(|_| ())
}),
("PersonaDialer", |a, b| {
PersonaDialer::with_bearer(a, b).map(|_| ())
}),
("PersonaDialer::new_admin", |a, b| {
PersonaDialer::new_admin(a, b).map(|_| ())
}),
("CredentialDialer::new_admin", |a, b| {
CredentialDialer::new_admin(a, b).map(|_| ())
}),
("NotificationDialer", |a, b| {
NotificationDialer::with_bearer(a, b).map(|_| ())
}),
("OperatorMailboxDialer", |a, b| {
OperatorMailboxDialer::with_bearer(a, b).map(|_| ())
}),
("QuestionDialer", |a, b| {
QuestionDialer::with_bearer(a, b).map(|_| ())
}),
("RoutineDialer::fire_routine", |_a, b| {
bearer_header_value(b).map(|_| ())
}),
];
for (label, build) in dialers {
build(ADDR, GOOD)
.unwrap_or_else(|err| panic!("{label}: a valid bearer must build a dialer: {err}"));
match build(ADDR, BAD) {
Ok(()) => panic!("{label}: a newline in the bearer must fail the dial closed"),
Err(err) => assert!(
matches!(err, DialError::InvalidBearer(_)),
"{label}: expected InvalidBearer, got {err}"
),
}
}
}
#[tokio::test]
async fn fire_routine_rejects_a_malformed_admin_bearer_before_dialing() {
let dialer = RoutineDialer::new("http://127.0.0.1:0").expect("valid address");
let err = dialer
.fire_routine("standup", "", sample_responder(), "pc_admin_bad\nbearer")
.await
.expect_err("a newline in admin_bearer must fail the dial closed");
assert!(
matches!(err, DialError::InvalidBearer(_)),
"expected InvalidBearer, got {err}"
);
}
fn standup_preview() -> ApprovalPreview {
ApprovalPreview {
prompt_text: "Post the daily standup summary.".to_owned(),
next_fires: vec![
fire("2026-07-29T09:00:00-07:00", "2026-07-29T16:00:00Z", ""),
fire("2026-07-30T09:00:00-07:00", "2026-07-30T16:00:00Z", ""),
fire("2026-07-31T09:00:00-07:00", "2026-07-31T16:00:00Z", ""),
],
zone_name: "America/Los_Angeles".to_owned(),
zone_is_fallback: false,
cadence: "every weekday at 9:00 AM PDT (UTC-7)".to_owned(),
}
}
fn fire(local: &str, utc: &str, zone_label: &str) -> ApprovalPreviewFire {
ApprovalPreviewFire {
local_time: local.to_owned(),
utc_time: utc.to_owned(),
zone_label: zone_label.to_owned(),
}
}
fn at(rfc3339: &str) -> DateTime<Utc> {
rfc3339.parse().expect("valid RFC3339")
}
#[test]
fn approval_preview_text_renders_cadence_then_an_aligned_run_table() {
let text = approval_preview_text(&standup_preview(), at("2026-07-28T17:00:00Z"));
assert_eq!(
text,
"This routine will run:\n\
\n\
Post the daily standup summary.\n\
\n\
Runs every weekday at 9:00 AM PDT (UTC-7)\n\
\n\
Next 3 runs\n\
```\n\
\u{20} Wed Jul 29 9:00 AM tomorrow\n\
\u{20} Thu Jul 30 9:00 AM in 2 days\n\
\u{20} Fri Jul 31 9:00 AM in 3 days\n\
```"
);
}
#[test]
fn approval_preview_text_shows_no_machine_timestamps() {
let text = approval_preview_text(&standup_preview(), at("2026-07-28T17:00:00Z"));
assert!(!text.contains("2026-07-29T09:00:00-07:00"), "{text}");
assert!(!text.contains("2026-07-29T16:00:00Z"), "{text}");
assert!(
!text.contains('Z'),
"no UTC instants belong on the card: {text}"
);
}
#[test]
fn approval_preview_text_never_says_fire() {
let mut empty = standup_preview();
empty.next_fires = Vec::new();
for preview in [standup_preview(), empty] {
let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
assert!(
!text.to_lowercase().contains("fire"),
"user-facing copy must say runs, not fires: {text}"
);
}
}
#[test]
fn approval_preview_text_titles_the_list_by_its_real_count() {
let mut preview = standup_preview();
preview.next_fires.truncate(2);
let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
assert!(text.contains("Next 2 runs"), "{text}");
preview.next_fires.truncate(1);
let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
assert!(text.contains("Next run"), "{text}");
assert!(!text.contains("Next 1 runs"), "{text}");
}
#[test]
fn approval_preview_text_omits_an_absent_cadence_line() {
let mut preview = standup_preview();
preview.cadence = String::new();
let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
assert!(!text.contains("Runs "), "{text}");
assert!(text.contains("Next 3 runs"), "{text}");
}
#[test]
fn approval_preview_text_puts_a_runs_own_zone_label_beside_its_time() {
let preview = ApprovalPreview {
prompt_text: "prompt".to_owned(),
next_fires: vec![
fire(
"2026-10-31T09:00:00-07:00",
"2026-10-31T16:00:00Z",
"PDT (UTC-7)",
),
fire(
"2026-11-01T09:00:00-08:00",
"2026-11-01T17:00:00Z",
"PST (UTC-8)",
),
],
zone_name: "America/Los_Angeles".to_owned(),
zone_is_fallback: false,
cadence: "every day at 9:00 AM".to_owned(),
};
let text = approval_preview_text(&preview, at("2026-10-30T17:00:00Z"));
assert!(text.contains("9:00 AM PDT (UTC-7) tomorrow"), "{text}");
assert!(text.contains("9:00 AM PST (UTC-8) in 2 days"), "{text}");
}
#[test]
fn approval_preview_text_anchors_runs_against_render_time() {
let preview = standup_preview();
let text = approval_preview_text(&preview, at("2026-07-30T17:00:00Z"));
assert!(text.contains("Wed Jul 29 9:00 AM passed"), "{text}");
assert!(text.contains("Thu Jul 30 9:00 AM passed"), "{text}");
assert!(text.contains("Fri Jul 31 9:00 AM tomorrow"), "{text}");
}
#[test]
fn approval_preview_text_reckons_days_in_the_runs_own_offset() {
let preview = standup_preview();
let text = approval_preview_text(&preview, at("2026-07-29T15:00:00Z"));
assert!(text.contains("Wed Jul 29 9:00 AM today"), "{text}");
}
#[test]
fn approval_preview_text_adds_the_year_when_it_differs() {
let preview = ApprovalPreview {
prompt_text: "prompt".to_owned(),
next_fires: vec![fire(
"2027-01-01T09:00:00-08:00",
"2027-01-01T17:00:00Z",
"",
)],
zone_name: "America/Los_Angeles".to_owned(),
zone_is_fallback: false,
cadence: "once".to_owned(),
};
let text = approval_preview_text(&preview, at("2026-12-30T17:00:00Z"));
assert!(text.contains("Fri Jan 1, 2027"), "{text}");
}
#[test]
fn approval_preview_text_renders_a_once_schedule_end_to_end() {
let preview = ApprovalPreview {
prompt_text: "Post the launch announcement.".to_owned(),
next_fires: vec![fire(
"2026-07-29T09:00:00-07:00",
"2026-07-29T16:00:00Z",
"PDT (UTC-7)",
)],
zone_name: "America/Los_Angeles".to_owned(),
zone_is_fallback: false,
cadence: "once".to_owned(),
};
let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
assert!(text.contains("Runs once"), "{text}");
assert!(text.contains("Next run"), "{text}");
assert!(!text.contains("Next 1 run"), "{text}");
assert!(text.contains("9:00 AM PDT (UTC-7) tomorrow"), "{text}");
}
#[test]
fn approval_preview_text_labels_the_zone_fallback() {
let mut preview = standup_preview();
preview.zone_is_fallback = true;
let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
assert!(
text.contains("don't know your time zone"),
"a fallback zone must be disclosed, not silently guessed: {text}"
);
assert!(text.contains("America/Los_Angeles"), "{text}");
}
#[test]
fn approval_preview_text_handles_no_upcoming_runs() {
let mut preview = standup_preview();
preview.next_fires = Vec::new();
let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
assert!(
text.contains("no upcoming runs, so approving it won't run anything"),
"{text}"
);
assert!(
!text.contains("Next"),
"an empty schedule has no run list: {text}"
);
}
#[test]
fn approval_preview_text_keeps_a_run_whose_instant_cannot_be_parsed() {
let mut preview = standup_preview();
preview.next_fires[1] = fire("not-an-instant", "", "");
let text = approval_preview_text(&preview, at("2026-07-28T17:00:00Z"));
assert!(text.contains("not-an-instant"), "{text}");
assert!(text.contains("Next 3 runs"), "{text}");
}
#[test]
fn agent_and_list_wire_previews_convert_to_the_same_shape() {
let agent_wire = WireAgentApprovalPreview {
prompt_text: "prompt".to_owned(),
next_fires: vec![WireAgentApprovalPreviewFire {
local_time: "2026-07-21T09:00:00-04:00".to_owned(),
utc_time: "2026-07-21T13:00:00Z".to_owned(),
..Default::default()
}],
zone_name: "America/New_York".to_owned(),
zone_is_fallback: true,
..Default::default()
};
let list_wire = WireListApprovalPreview {
prompt_text: "prompt".to_owned(),
next_fires: vec![WireListApprovalPreviewFire {
local_time: "2026-07-21T09:00:00-04:00".to_owned(),
utc_time: "2026-07-21T13:00:00Z".to_owned(),
..Default::default()
}],
zone_name: "America/New_York".to_owned(),
zone_is_fallback: true,
..Default::default()
};
assert_eq!(
ApprovalPreview::from(agent_wire),
ApprovalPreview::from(list_wire)
);
}
#[test]
fn pending_approval_entry_conversion_carries_the_preview() {
let entry = PendingApprovalEntry {
request_id: "r1".to_owned(),
tool_name: "routine_create".to_owned(),
args_json: "{}".to_owned(),
preview: buffa::MessageField::some(WireListApprovalPreview {
prompt_text: "prompt".to_owned(),
zone_name: "UTC".to_owned(),
..Default::default()
}),
..Default::default()
};
let pending = PendingApproval::from(entry);
assert_eq!(
pending.preview.as_ref().map(|p| p.prompt_text.as_str()),
Some("prompt")
);
}
#[test]
fn pending_question_conversion_carries_every_field() {
let wire = WireAgentPendingQuestion {
call_id: "call-1".to_owned(),
index: 0,
header: "Deploy target".to_owned(),
question: "Which environment?".to_owned(),
options: vec![
WireAgentQuestionOption {
label: "Staging".to_owned(),
description: "Deploys to staging only.".to_owned(),
recommended: false,
..Default::default()
},
WireAgentQuestionOption {
label: "Production".to_owned(),
description: "Deploys straight to production.".to_owned(),
recommended: true,
..Default::default()
},
],
args_json: r#"{"questions":[]}"#.to_owned(),
answer_token: "token-abc".to_owned(),
already_surfaced: true,
..Default::default()
};
let prompt = PendingQuestionPrompt::from(wire);
assert_eq!(prompt.call_id, "call-1");
assert_eq!(prompt.index, 0);
assert_eq!(prompt.header, "Deploy target");
assert_eq!(prompt.question, "Which environment?");
assert_eq!(prompt.options.len(), 2);
assert_eq!(prompt.options[1].label, "Production");
assert!(prompt.options[1].recommended);
assert_eq!(prompt.args_json, r#"{"questions":[]}"#);
assert_eq!(prompt.answer_token, "token-abc");
assert!(prompt.already_surfaced);
}
#[test]
fn events_from_end_projects_one_question_pending_per_entry() {
let end = AgentEnd {
pending_questions: vec![WireAgentPendingQuestion {
call_id: "call-1".to_owned(),
index: 0,
header: "h".to_owned(),
question: "q?".to_owned(),
options: vec![WireAgentQuestionOption {
label: "A".to_owned(),
..Default::default()
}],
args_json: "{}".to_owned(),
answer_token: "tok".to_owned(),
already_surfaced: true,
..Default::default()
}],
..Default::default()
};
let events = events_from_end(end);
let question_events: Vec<&TurnEvent> = events
.iter()
.filter(|e| matches!(e, TurnEvent::QuestionPending { .. }))
.collect();
assert_eq!(question_events.len(), 1);
let TurnEvent::QuestionPending {
call_id,
answer_token,
already_surfaced,
..
} = question_events[0]
else {
unreachable!("filtered above")
};
assert_eq!(call_id, "call-1");
assert_eq!(answer_token, "tok");
assert!(
already_surfaced,
"the streamed event must copy the control plane's already_surfaced bit through, \
not hardcode false (unlike the approval side's streamed path)"
);
assert!(matches!(events.last(), Some(TurnEvent::Done)));
}
}