#![forbid(unsafe_code)]
#![warn(missing_docs)]
pub mod edge;
pub use edge::{
ClaimedNamespace, ClaimedNamespaceError, EdgeAdapter, IngressDirective, IngressIdentity,
IngressIdentityError, MAX_CLAIMED_NAMESPACE_BYTES, Priority, build_attribution,
};
pub mod edge_credentials;
#[cfg(feature = "test-util")]
pub mod test_util {
use super::{
AssertedAttribution, Attribution, ClaimedNamespace, EdgeCredentials, IngressIdentity,
Message, ParticipantMessage, build_asserted_attribution, build_classify_attribution,
};
use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;
#[must_use]
pub fn signed_classify_envelope(
creds: &EdgeCredentials,
conversation_id: &str,
source_identity: &IngressIdentity,
namespace: &ClaimedNamespace,
transcript: &[ParticipantMessage],
bot_name: &str,
surface: &str,
) -> AssertedAttribution {
build_classify_attribution(
creds,
conversation_id,
source_identity,
namespace,
transcript,
bot_name,
surface,
)
}
#[must_use]
#[allow(
clippy::too_many_arguments,
reason = "mirrors `build_asserted_attribution`'s parameter list on \
purpose: a fixture that could omit one would stop testing \
the envelope production dials really send"
)]
pub fn signed_envelope(
creds: &EdgeCredentials,
conversation_id: &str,
exec_id: &str,
source_identity: &IngressIdentity,
namespace: &ClaimedNamespace,
messages: &[Message],
caller: Option<ExternalIdentity>,
participants: Vec<ExternalIdentity>,
) -> AssertedAttribution {
build_asserted_attribution(
creds,
conversation_id,
exec_id,
source_identity,
namespace,
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, AttachIngressRequest,
ClassifyRequest, CompactionReason as WireCompactionReason, ContextCompacted,
IngressDirective as WireIngressDirective, IngressReceipt as WireIngressReceipt,
IngressSourceIdentity as WireIngressSourceIdentity, 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, ListApprovalRecoveryRequest,
ListPendingRequest, PendingApprovalEntry, ReadPreview as WireListReadPreview,
RecordedApprovalDecision as WireRecordedApprovalDecision, RecoverableApprovalEntry,
RepromptRequired as WireRepromptRequired, SearchHitPreview as WireSearchHitPreview,
SearchPreview as WireSearchPreview, read_preview as wire_list_read_preview,
recoverable_approval_entry,
};
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};
use polyc_proto::proto::polychrome::task::v1::{
AgentTask as WireAgentTask, AgentTaskOwnership as WireAgentTaskOwnership,
AgentTaskServiceClient, AgentTaskState as WireAgentTaskState, CancelAgentTaskRequest,
ClaimAgentTaskRequest, CreateAgentTaskRequest, GetAgentTaskRequest, ListAgentTasksRequest,
RenewAgentTaskClaimRequest, TransitionAgentTaskRequest,
};
#[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),
#[error("turn ingress requires edge credentials so its source identity is signed")]
MissingIngressCredentials,
#[error("invalid durable ingress receipt: {0}")]
InvalidIngressReceipt(&'static str),
#[error("invalid approval recovery entry: {0}")]
InvalidApprovalRecovery(&'static str),
}
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(_)
| Self::MissingIngressCredentials
| Self::InvalidIngressReceipt(_)
| Self::InvalidApprovalRecovery(_) => 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))
}
#[must_use]
pub const fn is_stale_capability(&self) -> bool {
matches!(
self.code(),
Some(
connectrpc::ErrorCode::Unauthenticated | connectrpc::ErrorCode::FailedPrecondition
)
)
}
}
pub const AGENT_DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(4);
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 {
turn_id: String,
request_id: String,
tool_name: String,
title: String,
args_json: String,
reason: String,
resolve_token: String,
preview: Option<ApprovalPreview>,
},
QuestionPending {
turn_id: String,
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,
requested: bool,
},
WalletUpdatePrompt {
update_url: String,
new_limit: String,
},
WalletRevokePrompt {
revoke_url: 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(Debug, Clone)]
pub struct TurnIngress {
conversation_id: String,
exec_id: String,
source_identity: IngressIdentity,
namespace: ClaimedNamespace,
messages: Vec<Message>,
payment_receipt: Option<PaymentReceipt>,
attribution: Attribution,
ephemeral_history: bool,
ingress_directive: IngressDirective,
occurrence: String,
}
impl TurnIngress {
#[must_use]
pub fn new(
conversation_id: impl Into<String>,
exec_id: impl Into<String>,
source_identity: IngressIdentity,
namespace: ClaimedNamespace,
messages: Vec<Message>,
) -> Self {
Self {
conversation_id: conversation_id.into(),
exec_id: exec_id.into(),
source_identity,
namespace,
messages,
payment_receipt: None,
attribution: Attribution::default(),
ephemeral_history: false,
ingress_directive: IngressDirective::default(),
occurrence: String::new(),
}
}
#[must_use]
pub fn with_payment_receipt(mut self, receipt: PaymentReceipt) -> Self {
self.payment_receipt = Some(receipt);
self
}
#[must_use]
pub fn with_attribution(mut self, attribution: Attribution) -> Self {
self.attribution = attribution;
self
}
#[must_use]
pub fn with_ingress_directive(mut self, directive: IngressDirective) -> Self {
self.ingress_directive = directive;
self
}
#[must_use]
pub const fn with_ephemeral_history(mut self) -> Self {
self.ephemeral_history = true;
self
}
#[must_use]
pub fn with_occurrence(mut self, occurrence: impl Into<String>) -> Self {
self.occurrence = occurrence.into();
self
}
}
#[derive(Debug, Clone)]
pub struct DurablyReceivedTurn {
receipt: WireIngressReceipt,
attach_attempt: String,
}
impl DurablyReceivedTurn {
#[must_use]
pub fn dispatch_id(&self) -> &str {
&self.receipt.dispatch_id
}
#[must_use]
pub fn conversation_id(&self) -> &str {
&self.receipt.conversation_id
}
#[must_use]
pub fn receipt_bytes(&self) -> &[u8] {
&self.receipt.receipt
}
}
#[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_with_credentials(
&self,
addr: &str,
) -> Result<QuestionDialer, DialError> {
self.credentials.as_ref().map_or_else(
|| QuestionDialer::new(addr),
|creds| QuestionDialer::with_bearer(addr, creds.bearer()),
)
}
pub async fn receive_ingress(
&self,
ingress: TurnIngress,
) -> Result<DurablyReceivedTurn, DialError> {
let expected_source = ingress.source_identity.to_wire();
let expected_conversation = ingress.conversation_id.clone();
let request = build_request(ingress, self.credentials.as_deref())?;
let receipt = self
.client
.receive_ingress_with_options(request, traced_options())
.await?
.into_owned();
validate_ingress_receipt(&receipt, &expected_source, &expected_conversation)?;
Ok(DurablyReceivedTurn {
receipt,
attach_attempt: uuid::Uuid::now_v7().to_string(),
})
}
pub async fn attach_ingress(
&self,
received: &DurablyReceivedTurn,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>> + use<>, DialError> {
self.run_turn_streaming_received(received).await
}
pub async fn attach_ingress_buffered(
&self,
received: &DurablyReceivedTurn,
) -> Result<BufferedTurn, DialError> {
self.run_turn_buffered_received(received).await
}
pub async fn run_turn(
&self,
conversation_id: &str,
exec_id: &str,
source_identity: IngressIdentity,
namespace: ClaimedNamespace,
user_text: &str,
) -> Result<String, DialError> {
self.run_turn_with(
conversation_id,
exec_id,
source_identity,
namespace,
user_text,
Attribution::default(),
)
.await
}
pub async fn run_turn_with(
&self,
conversation_id: &str,
exec_id: &str,
source_identity: IngressIdentity,
namespace: ClaimedNamespace,
user_text: &str,
attribution: Attribution,
) -> Result<String, DialError> {
Ok(self
.run_turn_with_approvals(
conversation_id,
exec_id,
source_identity,
namespace,
user_text,
attribution,
IngressDirective::default(),
)
.await?
.reply)
}
#[allow(
clippy::too_many_arguments,
reason = "each argument is an independent turn premise the signed \
envelope binds; grouping them would hide which fields the \
edge must supply, and a missing one must not compile"
)]
pub async fn run_turn_with_approvals(
&self,
conversation_id: &str,
exec_id: &str,
source_identity: IngressIdentity,
namespace: ClaimedNamespace,
user_text: &str,
attribution: Attribution,
ingress_directive: IngressDirective,
) -> Result<BufferedTurn, DialError> {
let ingress = TurnIngress::new(
conversation_id,
exec_id,
source_identity,
namespace,
vec![text_message("user", user_text)],
)
.with_attribution(attribution)
.with_ingress_directive(ingress_directive);
let received = self.receive_ingress(ingress).await?;
self.attach_ingress_buffered(&received).await
}
pub async fn run_routine_turn(&self, ingress: TurnIngress) -> Result<BufferedTurn, DialError> {
let received = self.receive_ingress(ingress).await?;
self.attach_ingress_buffered(&received).await
}
async fn run_turn_buffered_received(
&self,
received: &DurablyReceivedTurn,
) -> Result<BufferedTurn, DialError> {
let mut stream = self
.client
.attach_ingress_with_options(
AttachIngressRequest {
receipt: buffa::MessageField::some(received.receipt.clone()),
attach_attempt: received.attach_attempt.clone(),
..Default::default()
},
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 wallet_update_prompt: Option<(String, String)> = None;
let mut wallet_revoke_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 requested = prompt.requested;
let link_url =
(!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
wallet_link_prompt = WalletLinkPrompt::Present {
link_url,
renewal,
requested,
};
}
if let Some(prompt) = end.wallet_update_prompt.into_option() {
wallet_update_prompt = Some((prompt.update_url, prompt.new_limit));
}
if let Some(prompt) = end.wallet_revoke_prompt.into_option() {
wallet_revoke_prompt = Some(prompt.revoke_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,
wallet_update_prompt
.as_ref()
.map(|(url, limit)| (url.as_str(), limit.as_str())),
wallet_revoke_prompt.as_deref(),
),
pending_approvals,
pending_questions,
})
}
pub async fn run_turn_streaming(
&self,
conversation_id: &str,
exec_id: &str,
source_identity: IngressIdentity,
namespace: ClaimedNamespace,
user_text: &str,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
self.run_turn_streaming_messages(
conversation_id,
exec_id,
source_identity,
namespace,
vec![text_message("user", user_text)],
)
.await
}
pub async fn run_turn_streaming_messages(
&self,
conversation_id: &str,
exec_id: &str,
source_identity: IngressIdentity,
namespace: ClaimedNamespace,
messages: Vec<Message>,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
self.run_turn_streaming_ingress(TurnIngress::new(
conversation_id,
exec_id,
source_identity,
namespace,
messages,
))
.await
}
#[allow(clippy::too_many_arguments)] pub async fn run_turn_streaming_messages_with(
&self,
conversation_id: &str,
exec_id: &str,
source_identity: IngressIdentity,
namespace: ClaimedNamespace,
messages: Vec<Message>,
payment_receipt: Option<PaymentReceipt>,
attribution: Attribution,
ingress_directive: IngressDirective,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
let mut ingress = TurnIngress::new(
conversation_id,
exec_id,
source_identity,
namespace,
messages,
)
.with_attribution(attribution)
.with_ingress_directive(ingress_directive);
if let Some(receipt) = payment_receipt {
ingress = ingress.with_payment_receipt(receipt);
}
self.run_turn_streaming_ingress(ingress).await
}
pub async fn run_turn_streaming_ingress(
&self,
ingress: TurnIngress,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
let received = self.receive_ingress(ingress).await?;
self.run_turn_streaming_received(&received).await
}
async fn run_turn_streaming_received(
&self,
received: &DurablyReceivedTurn,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>> + use<>, DialError> {
let mut stream = self
.client
.attach_ingress_with_options(
AttachIngressRequest {
receipt: buffa::MessageField::some(received.receipt.clone()),
attach_attempt: received.attach_attempt.clone(),
..Default::default()
},
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>,
source_identity: &IngressIdentity,
namespace: &ClaimedNamespace,
) -> Result<bool, DialError> {
let credentials = self
.credentials
.as_ref()
.ok_or(DialError::MissingIngressCredentials)?;
let asserted_attribution = build_classify_attribution(
credentials,
conversation_id,
source_identity,
namespace,
&transcript,
bot_name,
surface,
);
let request = ClassifyRequest {
conversation_id: conversation_id.to_owned(),
bot_name: bot_name.to_owned(),
surface: surface.to_owned(),
transcript,
asserted_attribution: buffa::MessageField::some(asserted_attribution),
..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 turn_id: String,
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>,
pub read_preview: Option<ReadPreview>,
pub already_surfaced: bool,
}
impl TryFrom<PendingApprovalEntry> for PendingApproval {
type Error = DialError;
fn try_from(p: PendingApprovalEntry) -> Result<Self, Self::Error> {
let PendingApprovalEntry {
request_id,
tool_name,
args_json,
reason,
resolve_token,
preview,
read_preview,
already_surfaced,
turn_id,
title,
__buffa_unknown_fields: _,
} = p;
Ok(Self {
turn_id,
request_id,
tool_name,
title,
args_json,
reason,
resolve_token,
preview: preview.into_option().map(Into::into),
read_preview: read_preview
.into_option()
.map(ReadPreview::try_from)
.transpose()?,
already_surfaced,
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RecordedApprovalRecovery {
pub approved: bool,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ApprovalRecoveryState {
DecisionRecorded(RecordedApprovalRecovery),
RepromptRequired,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct RecoverableApproval {
pub turn_id: String,
pub request_id: String,
pub tool_name: String,
pub title: String,
pub args_json: String,
pub reason: String,
pub preview: Option<ApprovalPreview>,
pub read_preview: Option<ReadPreview>,
pub state: ApprovalRecoveryState,
}
impl TryFrom<RecoverableApprovalEntry> for RecoverableApproval {
type Error = DialError;
fn try_from(entry: RecoverableApprovalEntry) -> Result<Self, Self::Error> {
let RecoverableApprovalEntry {
turn_id,
request_id,
tool_name,
title,
args_json,
reason,
preview,
read_preview,
state,
__buffa_unknown_fields: _,
} = entry;
let state = match state {
Some(recoverable_approval_entry::State::DecisionRecorded(decision)) => {
let WireRecordedApprovalDecision {
approved,
__buffa_unknown_fields: _,
} = *decision;
ApprovalRecoveryState::DecisionRecorded(RecordedApprovalRecovery { approved })
}
Some(recoverable_approval_entry::State::RepromptRequired(marker)) => {
let WireRepromptRequired {
__buffa_unknown_fields: _,
} = *marker;
ApprovalRecoveryState::RepromptRequired
}
None => {
return Err(DialError::InvalidApprovalRecovery(
"approval recovery entry omitted its state",
));
}
};
Ok(Self {
turn_id,
request_id,
tool_name,
title,
args_json,
reason,
preview: preview.into_option().map(Into::into),
read_preview: read_preview
.into_option()
.map(ReadPreview::try_from)
.transpose()?,
state,
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum ReadPreview {
Search {
canonical_query: String,
scope_count: u32,
},
Hit {
origin_query: String,
conversation_label: String,
max_bytes: u32,
},
}
impl TryFrom<WireListReadPreview> for ReadPreview {
type Error = DialError;
fn try_from(preview: WireListReadPreview) -> Result<Self, Self::Error> {
let WireListReadPreview {
detail,
__buffa_unknown_fields: _,
} = preview;
match detail {
Some(wire_list_read_preview::Detail::Search(search)) => {
let WireSearchPreview {
canonical_query,
scope_count,
__buffa_unknown_fields: _,
} = *search;
Ok(Self::Search {
canonical_query,
scope_count,
})
}
Some(wire_list_read_preview::Detail::Hit(hit)) => {
let WireSearchHitPreview {
origin_query,
conversation_label,
max_bytes,
__buffa_unknown_fields: _,
} = *hit;
Ok(Self::Hit {
origin_query,
conversation_label,
max_bytes,
})
}
None => Err(DialError::InvalidApprovalRecovery(
"approval read preview omitted its detail",
)),
}
}
}
#[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,
pub routine_name: 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 {
let WireAgentApprovalPreview {
prompt_text,
next_fires,
zone_name,
zone_is_fallback,
cadence,
routine_name,
__buffa_unknown_fields: _,
} = p;
Self {
prompt_text,
next_fires: next_fires.into_iter().map(Into::into).collect(),
zone_name,
zone_is_fallback,
cadence,
routine_name,
}
}
}
impl From<WireListApprovalPreview> for ApprovalPreview {
fn from(p: WireListApprovalPreview) -> Self {
let WireListApprovalPreview {
prompt_text,
next_fires,
zone_name,
zone_is_fallback,
cadence,
routine_name,
__buffa_unknown_fields: _,
} = p;
Self {
prompt_text,
next_fires: next_fires.into_iter().map(Into::into).collect(),
zone_name,
zone_is_fallback,
cadence,
routine_name,
}
}
}
#[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")
}
#[must_use]
pub fn approval_preview_plain_text(preview: &ApprovalPreview, now: DateTime<Utc>) -> String {
approval_preview_text(preview, now)
.lines()
.filter(|line| *line != FENCE)
.collect::<Vec<_>>()
.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,
turn_id: &str,
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(
turn_id,
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,
turn_id: &str,
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 {
turn_id: turn_id.to_owned(),
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::try_from)
.collect::<Result<Vec<_>, _>>()?,
);
if reply.next_page_token.is_empty() {
break;
}
page_token = reply.next_page_token;
}
Ok(pending)
}
pub async fn list_recovery(
&self,
conversation_id: &str,
) -> Result<Vec<RecoverableApproval>, DialError> {
let mut recovery = Vec::new();
let mut page_token = String::new();
let mut seen_page_tokens = std::collections::HashSet::new();
loop {
let request = ListApprovalRecoveryRequest {
conversation_id: conversation_id.to_owned(),
page_size: 0,
page_token: page_token.clone(),
__buffa_unknown_fields: buffa::UnknownFields::default(),
};
let reply = self
.client
.list_recovery_with_options(request, traced_options())
.await?
.into_owned();
recovery.extend(
reply
.recovery
.into_iter()
.map(RecoverableApproval::try_from)
.collect::<Result<Vec<_>, _>>()?,
);
if reply.next_page_token.is_empty() {
break;
}
if !seen_page_tokens.insert(reply.next_page_token.clone()) {
return Err(DialError::InvalidApprovalRecovery(
"recovery listing repeated a continuation token",
));
}
page_token = reply.next_page_token;
}
Ok(recovery)
}
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)?,
})
}
#[allow(clippy::too_many_arguments)] pub async fn respond(
&self,
turn_id: &str,
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 {
turn_id: turn_id.to_owned(),
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;
pub use connectrpc::ConnectError;
#[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
}
#[must_use]
pub fn encode_transcript_for_content_hash(
transcript: &[ParticipantMessage],
bot_name: &str,
surface: &str,
) -> Vec<u8> {
use buffa::Message as _;
let mut bytes = Vec::new();
for field in [bot_name.as_bytes(), surface.as_bytes()] {
bytes.extend_from_slice(&(field.len() as u64).to_be_bytes());
bytes.extend_from_slice(field);
}
for message in transcript {
let encoded = message.encode_to_vec();
bytes.extend_from_slice(&(encoded.len() as u64).to_be_bytes());
bytes.extend_from_slice(&encoded);
}
bytes
}
fn build_classify_attribution(
creds: &EdgeCredentials,
conversation_id: &str,
source_identity: &IngressIdentity,
namespace: &ClaimedNamespace,
transcript: &[ParticipantMessage],
bot_name: &str,
surface: &str,
) -> AssertedAttribution {
let content_hash = polyc_crypto::edge_identity::content_hash_hex(
&encode_transcript_for_content_hash(transcript, bot_name, surface),
);
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: buffa::MessageField::none(),
participants: Vec::new(),
signature_hex: String::new(),
exec_id: String::new(),
content_hash,
source_identity: buffa::MessageField::some(source_identity.to_wire()),
namespace: namespace.as_str().to_owned(),
..Default::default()
};
creds.sign_assertion(&mut envelope);
envelope
}
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,
source_identity: &IngressIdentity,
namespace: &ClaimedNamespace,
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,
source_identity: buffa::MessageField::some(source_identity.to_wire()),
namespace: namespace.as_str().to_owned(),
..Default::default()
};
creds.sign_assertion(&mut envelope);
envelope
}
fn build_request(
ingress: TurnIngress,
credentials: Option<&EdgeCredentials>,
) -> Result<AgentRequest, DialError> {
let credentials = credentials.ok_or(DialError::MissingIngressCredentials)?;
let asserted_attribution = build_asserted_attribution(
credentials,
&ingress.conversation_id,
&ingress.exec_id,
&ingress.source_identity,
&ingress.namespace,
&ingress.messages,
&ingress.attribution,
);
Ok(AgentRequest {
conversation_id: ingress.conversation_id,
exec_id: ingress.exec_id,
start: buffa::MessageField::some(AgentStart {
agent_id: String::new(),
agent_config: Vec::new(),
messages: ingress.messages,
payment_receipt: ingress
.payment_receipt
.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
ephemeral_history: ingress.ephemeral_history,
ingress_directive: wire_ingress_directive(ingress.ingress_directive),
occurrence: ingress.occurrence,
asserted_attribution: buffa::MessageField::some(asserted_attribution),
..Default::default()
}),
..Default::default()
})
}
fn validate_ingress_receipt(
receipt: &WireIngressReceipt,
expected_source: &WireIngressSourceIdentity,
expected_conversation: &str,
) -> Result<(), DialError> {
let Some(source) = receipt.source_identity.as_option() else {
return Err(DialError::InvalidIngressReceipt(
"source identity is absent",
));
};
if source != expected_source {
return Err(DialError::InvalidIngressReceipt(
"source identity does not match the signed request",
));
}
if receipt.receipt.is_empty() {
return Err(DialError::InvalidIngressReceipt("State receipt is empty"));
}
if receipt.dispatch_id.is_empty() {
return Err(DialError::InvalidIngressReceipt("dispatch id is empty"));
}
if receipt.conversation_id != expected_conversation {
return Err(DialError::InvalidIngressReceipt(
"conversation does not match the request",
));
}
Ok(())
}
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 requested = prompt.requested;
let link_url = (!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
events.push(TurnEvent::WalletLinkPrompt {
link_url,
renewal,
requested,
});
}
if let Some(prompt) = end.wallet_update_prompt.into_option() {
events.push(TurnEvent::WalletUpdatePrompt {
update_url: prompt.update_url,
new_limit: prompt.new_limit,
});
}
if let Some(prompt) = end.wallet_revoke_prompt.into_option() {
events.push(TurnEvent::WalletRevokePrompt {
revoke_url: prompt.revoke_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 turn_id: String,
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 {
turn_id: pa.turn_id,
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 turn_id: String,
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 {
turn_id: q.turn_id,
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 {
turn_id: q.turn_id,
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 {
turn_id: p.turn_id,
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 {
turn_id: p.turn_id,
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,
requested: bool,
},
}
#[allow(clippy::too_many_arguments)] fn finalize_buffered_reply(
text_parts: &[String],
scaffolding: &[String],
wallet_link_prompt: WalletLinkPrompt,
wallet_update_prompt: Option<(&str, &str)>,
wallet_revoke_prompt: Option<&str>,
) -> String {
if let WalletLinkPrompt::Present {
link_url,
renewal,
requested,
} = wallet_link_prompt
{
return if requested {
polyc_proto::wallet_link_requested_prompt(link_url.as_deref().unwrap_or_default())
} else {
polyc_proto::wallet_link_prompt(link_url.as_deref(), renewal)
};
}
if let Some((url, new_limit)) = wallet_update_prompt {
return polyc_proto::wallet_update_prompt(url, new_limit);
}
if let Some(url) = wallet_revoke_prompt {
return polyc_proto::wallet_revoke_prompt(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,
},
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::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,
pub grants_attest_email: 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,
grants_attest_email: wire.grants_attest_email,
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,
pub grants_attest_email: 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,
grants_attest_email: enrollment.grants_attest_email,
..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::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::UpgradeOutcome { .. } | Self::Unknown => None,
}
}
#[must_use]
pub const fn is_decision(&self) -> bool {
!matches!(self, 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,
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
pub enum AgentTaskState {
Submitted,
Working,
InputRequired,
AuthRequired,
Completed,
Failed,
Canceled,
Rejected,
}
impl AgentTaskState {
#[must_use]
pub const fn is_terminal(self) -> bool {
matches!(
self,
Self::Completed | Self::Failed | Self::Canceled | Self::Rejected
)
}
}
impl From<AgentTaskState> for WireAgentTaskState {
fn from(value: AgentTaskState) -> Self {
match value {
AgentTaskState::Submitted => Self::AGENT_TASK_STATE_SUBMITTED,
AgentTaskState::Working => Self::AGENT_TASK_STATE_WORKING,
AgentTaskState::InputRequired => Self::AGENT_TASK_STATE_INPUT_REQUIRED,
AgentTaskState::AuthRequired => Self::AGENT_TASK_STATE_AUTH_REQUIRED,
AgentTaskState::Completed => Self::AGENT_TASK_STATE_COMPLETED,
AgentTaskState::Failed => Self::AGENT_TASK_STATE_FAILED,
AgentTaskState::Canceled => Self::AGENT_TASK_STATE_CANCELED,
AgentTaskState::Rejected => Self::AGENT_TASK_STATE_REJECTED,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AgentTaskRecord {
pub task_id: String,
pub context_id: String,
pub state: AgentTaskState,
pub status_detail: Vec<u8>,
pub artifacts: Vec<Vec<u8>>,
pub history: Vec<Vec<u8>>,
pub metadata: Vec<u8>,
pub created_at_ms: u64,
pub updated_at_ms: u64,
}
impl TryFrom<WireAgentTask> for AgentTaskRecord {
type Error = DialError;
fn try_from(value: WireAgentTask) -> Result<Self, Self::Error> {
let WireAgentTask {
task_id,
context_id,
state,
status_detail,
artifacts,
history,
metadata,
created_at_ms,
updated_at_ms,
__buffa_unknown_fields: _,
} = value;
let state = match state.as_known() {
Some(WireAgentTaskState::AGENT_TASK_STATE_SUBMITTED) => AgentTaskState::Submitted,
Some(WireAgentTaskState::AGENT_TASK_STATE_WORKING) => AgentTaskState::Working,
Some(WireAgentTaskState::AGENT_TASK_STATE_INPUT_REQUIRED) => {
AgentTaskState::InputRequired
}
Some(WireAgentTaskState::AGENT_TASK_STATE_AUTH_REQUIRED) => {
AgentTaskState::AuthRequired
}
Some(WireAgentTaskState::AGENT_TASK_STATE_COMPLETED) => AgentTaskState::Completed,
Some(WireAgentTaskState::AGENT_TASK_STATE_FAILED) => AgentTaskState::Failed,
Some(WireAgentTaskState::AGENT_TASK_STATE_CANCELED) => AgentTaskState::Canceled,
Some(WireAgentTaskState::AGENT_TASK_STATE_REJECTED) => AgentTaskState::Rejected,
Some(WireAgentTaskState::AGENT_TASK_STATE_UNSPECIFIED) | None => {
return Err(DialError::Connect(connectrpc::ConnectError::internal(
"the task record names no lifecycle state",
)));
}
};
Ok(Self {
task_id,
context_id,
state,
status_detail,
artifacts,
history,
metadata,
created_at_ms,
updated_at_ms,
})
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AgentTaskTransition {
pub status_detail: Vec<u8>,
pub artifacts: Vec<Vec<u8>>,
pub metadata: Vec<u8>,
pub appended_history: Vec<Vec<u8>>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AgentTaskOwnership {
pub dispatch_id: String,
pub worker_instance: String,
pub attempt_id: String,
pub fence: u64,
pub claim_digest: Vec<u8>,
}
impl From<&AgentTaskOwnership> for WireAgentTaskOwnership {
fn from(value: &AgentTaskOwnership) -> Self {
Self {
dispatch_id: value.dispatch_id.clone(),
worker_instance: value.worker_instance.clone(),
attempt_id: value.attempt_id.clone(),
fence: value.fence,
claim_digest: value.claim_digest.clone(),
__buffa_unknown_fields: buffa::UnknownFields::default(),
}
}
}
impl From<WireAgentTaskOwnership> for AgentTaskOwnership {
fn from(value: WireAgentTaskOwnership) -> Self {
Self {
dispatch_id: value.dispatch_id,
worker_instance: value.worker_instance,
attempt_id: value.attempt_id,
fence: value.fence,
claim_digest: value.claim_digest,
}
}
}
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct AgentTaskPage {
pub tasks: Vec<AgentTaskRecord>,
pub next_after: Option<String>,
}
#[derive(Clone)]
pub struct TaskDialer {
client: Arc<AgentTaskServiceClient<HttpClient>>,
}
const fn is_ambiguous_task_error(error: &connectrpc::ConnectError) -> bool {
matches!(
error.code,
connectrpc::ErrorCode::Unavailable
| connectrpc::ErrorCode::DeadlineExceeded
| connectrpc::ErrorCode::ResourceExhausted
| connectrpc::ErrorCode::Aborted
)
}
impl TaskDialer {
pub fn new(addr: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, None, AgentTaskServiceClient::new)?,
})
}
pub fn with_bearer(addr: &str, bearer: &str) -> Result<Self, DialError> {
Ok(Self {
client: build_control_client(addr, Some(bearer), AgentTaskServiceClient::new)?,
})
}
pub async fn create_task(
&self,
task_id: &str,
context_id: &str,
history: Vec<Vec<u8>>,
) -> Result<AgentTaskRecord, DialError> {
let request = CreateAgentTaskRequest {
task_id: task_id.to_owned(),
context_id: context_id.to_owned(),
history,
..Default::default()
};
let reply = self
.client
.create_task_with_options(request, traced_options())
.await?
.into_owned();
Self::require_task(reply.task.into_option())
}
pub async fn transition_task(
&self,
task_id: &str,
state: AgentTaskState,
successor: AgentTaskTransition,
ownership: &AgentTaskOwnership,
) -> Result<AgentTaskRecord, DialError> {
let AgentTaskTransition {
status_detail,
artifacts,
metadata,
appended_history,
} = successor;
let request = TransitionAgentTaskRequest {
task_id: task_id.to_owned(),
state: WireAgentTaskState::from(state).into(),
status_detail,
artifacts,
metadata,
appended_history,
ownership: buffa::MessageField::some(WireAgentTaskOwnership::from(ownership)),
operation_id: uuid::Uuid::now_v7().to_string(),
..Default::default()
};
let first = self
.client
.transition_task_with_options(request.clone(), traced_options())
.await;
let reply = match first {
Ok(reply) => reply,
Err(error) if is_ambiguous_task_error(&error) => {
self.client
.transition_task_with_options(request, traced_options())
.await?
}
Err(error) => return Err(DialError::Connect(error)),
}
.into_owned();
Self::require_task(reply.task.into_option())
}
pub async fn claim_task(
&self,
task_id: &str,
dispatch_id: &str,
worker_instance: &str,
attempt_id: &str,
appended_history: Vec<Vec<u8>>,
) -> Result<(AgentTaskRecord, AgentTaskOwnership), DialError> {
let request = ClaimAgentTaskRequest {
task_id: task_id.to_owned(),
dispatch_id: dispatch_id.to_owned(),
worker_instance: worker_instance.to_owned(),
attempt_id: attempt_id.to_owned(),
appended_history,
..Default::default()
};
let first = self
.client
.claim_task_with_options(request.clone(), traced_options())
.await;
let reply = match first {
Ok(reply) => reply,
Err(error) if is_ambiguous_task_error(&error) => {
self.client
.claim_task_with_options(request, traced_options())
.await?
}
Err(error) => return Err(DialError::Connect(error)),
}
.into_owned();
let task = Self::require_task(reply.task.into_option())?;
let ownership = reply.ownership.into_option().ok_or_else(|| {
DialError::Connect(connectrpc::ConnectError::internal(
"a successful task claim returned no ownership capability",
))
})?;
Ok((task, AgentTaskOwnership::from(ownership)))
}
pub async fn renew_task_claim(
&self,
task_id: &str,
ownership: &AgentTaskOwnership,
ordinal: u64,
) -> Result<(), DialError> {
let request = RenewAgentTaskClaimRequest {
task_id: task_id.to_owned(),
ownership: buffa::MessageField::some(WireAgentTaskOwnership::from(ownership)),
ordinal,
..Default::default()
};
let first = self
.client
.renew_task_claim_with_options(request.clone(), traced_options())
.await;
match first {
Ok(_) => Ok(()),
Err(error) if is_ambiguous_task_error(&error) => self
.client
.renew_task_claim_with_options(request, traced_options())
.await
.map(|_| ())
.map_err(DialError::Connect),
Err(error) => Err(DialError::Connect(error)),
}
}
pub async fn cancel_task(&self, task_id: &str) -> Result<AgentTaskRecord, DialError> {
let request = CancelAgentTaskRequest {
task_id: task_id.to_owned(),
..Default::default()
};
let reply = self
.client
.cancel_task_with_options(request, traced_options())
.await?
.into_owned();
Self::require_task(reply.task.into_option())
}
pub async fn get_task(&self, task_id: &str) -> Result<Option<AgentTaskRecord>, DialError> {
let request = GetAgentTaskRequest {
task_id: task_id.to_owned(),
..Default::default()
};
let reply = self
.client
.get_task_with_options(request, traced_options())
.await?
.into_owned();
reply
.task
.into_option()
.map(AgentTaskRecord::try_from)
.transpose()
}
pub async fn list_tasks(
&self,
context_id: &str,
page_size: u32,
after: Option<&str>,
) -> Result<AgentTaskPage, DialError> {
let request = ListAgentTasksRequest {
context_id: context_id.to_owned(),
page_size,
after: after.unwrap_or_default().to_owned(),
..Default::default()
};
let reply = self
.client
.list_tasks_with_options(request, traced_options())
.await?
.into_owned();
let tasks = reply
.tasks
.into_iter()
.map(AgentTaskRecord::try_from)
.collect::<Result<Vec<_>, _>>()?;
Ok(AgentTaskPage {
tasks,
next_after: (!reply.next_after.is_empty()).then_some(reply.next_after),
})
}
fn require_task(task: Option<WireAgentTask>) -> Result<AgentTaskRecord, DialError> {
task.ok_or_else(|| {
DialError::Connect(connectrpc::ConnectError::internal(
"the task operation returned no task record",
))
})
.and_then(AgentTaskRecord::try_from)
}
}
#[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 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,
requested: false,
},
None,
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,
requested: false,
},
None,
None,
);
assert_eq!(reply, polyc_proto::wallet_link_prompt(None, false));
}
#[test]
fn finalize_buffered_reply_wallet_link_prompt_requested_wins() {
let url = "https://polychrome.example/link/abc";
let reply = finalize_buffered_reply(
&[],
&[],
WalletLinkPrompt::Present {
link_url: Some(url.to_owned()),
renewal: false,
requested: true,
},
None,
None,
);
assert_eq!(reply, polyc_proto::wallet_link_requested_prompt(url));
assert!(
reply.contains(url),
"the requested-link prompt carries the link: {reply}"
);
assert_ne!(
reply,
polyc_proto::wallet_link_prompt(Some(url), false),
"the requested branch must read differently from the payment-interrupt one"
);
}
#[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,
requested: false,
},
None,
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,
requested: false,
},
None,
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,
None,
),
"the answer"
);
assert_eq!(
finalize_buffered_reply(
&[],
&["[tool_call:foo]".to_owned()],
WalletLinkPrompt::None,
None,
None,
),
"[tool_call:foo]"
);
assert_eq!(
finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, None, None),
""
);
}
#[test]
fn finalize_buffered_reply_wallet_update_prompt_wins() {
let url = "https://polychrome.example/update/xyz";
let reply =
finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, Some((url, "20")), None);
assert_eq!(reply, polyc_proto::wallet_update_prompt(url, "20"));
assert!(reply.contains(url), "{reply}");
assert!(reply.contains("20"), "{reply}");
}
#[test]
fn finalize_buffered_reply_wallet_revoke_prompt_wins() {
let url = "https://polychrome.example/revoke/xyz";
let reply = finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, None, Some(url));
assert_eq!(reply, polyc_proto::wallet_revoke_prompt(url));
assert!(reply.contains(url), "{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 {
turn_id: "00000000-0000-0000-0000-000000000001".to_owned(),
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 {
turn_id: "00000000-0000-0000-0000-000000000001".to_owned(),
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,
requested: 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,
requested: 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,
requested: false,
},
TurnEvent::Done,
]
);
}
#[test]
fn end_projects_wallet_link_prompt_requested_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(),
requested: 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: false,
requested: true,
},
TurnEvent::Done,
]
);
}
#[test]
fn end_projects_wallet_update_prompt_before_done() {
use polyc_proto::proto::polychrome::agent::v1::WalletUpdatePrompt as WireWalletUpdatePrompt;
let end = AgentEnd {
wallet_update_prompt: buffa::MessageField::some(WireWalletUpdatePrompt {
update_url: "https://polychrome.example/update/xyz".to_owned(),
new_limit: "20".to_owned(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::WalletUpdatePrompt {
update_url: "https://polychrome.example/update/xyz".to_owned(),
new_limit: "20".to_owned(),
},
TurnEvent::Done,
]
);
}
#[test]
fn end_projects_wallet_revoke_prompt_before_done() {
use polyc_proto::proto::polychrome::agent::v1::WalletRevokePrompt as WireWalletRevokePrompt;
let end = AgentEnd {
wallet_revoke_prompt: buffa::MessageField::some(WireWalletRevokePrompt {
revoke_url: "https://polychrome.example/revoke/xyz".to_owned(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::WalletRevokePrompt {
revoke_url: "https://polychrome.example/revoke/xyz".to_owned(),
},
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_fails_before_transport() {
let ingress = TurnIngress::new(
"slack:team-1:general",
"exec-1",
IngressIdentity::reported("team-1", "message-1").expect("valid source identity"),
ClaimedNamespace::new("slack").expect("a real namespace is a valid claim"),
vec![text_message("user", "hi")],
);
assert!(
matches!(
build_request(ingress, None),
Err(DialError::MissingIngressCredentials)
),
"source identity cannot ride outside a signed 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 source_identity =
IngressIdentity::reported("team-1", "message-1").expect("valid source identity");
let ingress = TurnIngress::new(
"slack:team-1:general",
"exec-1",
source_identity.clone(),
ClaimedNamespace::new("slack").expect("a real namespace is a valid claim"),
vec![text_message("user", "hi")],
)
.with_attribution(attribution.clone());
let request = build_request(ingress, Some(&creds)).expect("credentialed request builds");
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"
);
let mut tampered_source = envelope.clone();
tampered_source.source_identity = buffa::MessageField::some(
IngressIdentity::reported("team-1", "message-2")
.expect("valid source identity")
.to_wire(),
);
assert!(
!polyc_crypto::edge_identity::verify_edge_assertion(&signer_pk, &tampered_source),
"source identity must be covered by the edge signature"
);
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"
);
assert_eq!(
envelope.source_identity.into_option(),
Some(source_identity.to_wire()),
"the stable source identity must ride inside the signed envelope"
);
}
#[test]
fn source_identity_and_content_digest_do_not_follow_exec_id() {
let creds = EdgeCredentials::from_parts(
"edge".to_owned(),
"bearer".to_owned(),
&polyc_crypto::hex::lower(&[8; 32]),
)
.expect("valid credentials");
let source =
IngressIdentity::reported("source", "event-42").expect("valid source identity");
let messages = vec![text_message("user", "same authenticated content")];
let envelope = |exec_id: &str| {
build_request(
TurnIngress::new(
"conv",
exec_id,
source.clone(),
ClaimedNamespace::new("slack").expect("a real namespace is a valid claim"),
messages.clone(),
),
Some(&creds),
)
.expect("request builds")
.start
.into_option()
.expect("start set")
.asserted_attribution
.into_option()
.expect("assertion set")
};
let first = envelope("attempt-1");
let retry = envelope("attempt-2");
assert_ne!(first.exec_id, retry.exec_id);
assert_eq!(first.source_identity, retry.source_identity);
assert_eq!(first.content_hash, retry.content_hash);
}
#[test]
fn transport_success_without_complete_receipt_is_not_acknowledgement_authority() {
let source = IngressIdentity::reported("source", "event-42")
.expect("valid source identity")
.to_wire();
let valid = WireIngressReceipt {
source_identity: buffa::MessageField::some(source.clone()),
receipt: b"receipt".to_vec(),
dispatch_id: "dispatch".to_owned(),
conversation_id: "conv".to_owned(),
..Default::default()
};
assert!(validate_ingress_receipt(&valid, &source, "conv").is_ok());
let malformed = [
WireIngressReceipt {
source_identity: buffa::MessageField::none(),
..valid.clone()
},
WireIngressReceipt {
receipt: Vec::new(),
..valid.clone()
},
WireIngressReceipt {
dispatch_id: String::new(),
..valid.clone()
},
WireIngressReceipt {
conversation_id: "other".to_owned(),
..valid
},
];
for receipt in malformed {
assert!(matches!(
validate_ingress_receipt(&receipt, &source, "conv"),
Err(DialError::InvalidIngressReceipt(_))
));
}
}
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(
"00000000-0000-0000-0000-000000000001",
"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(
"00000000-0000-0000-0000-000000000001",
"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}"
);
}
#[test]
fn recovery_conversion_preserves_every_display_projection() {
use polyc_proto::proto::polychrome::approval::v1::{
RecordedApprovalDecision, SearchPreview,
};
let wire = RecoverableApprovalEntry {
turn_id: "turn-1".to_owned(),
request_id: "call-1".to_owned(),
tool_name: "conversation_search".to_owned(),
title: "Search past conversations".to_owned(),
args_json: r#"{"query":"release"}"#.to_owned(),
reason: "Review the search scope.".to_owned(),
preview: buffa::MessageField::none(),
read_preview: buffa::MessageField::some(WireListReadPreview {
detail: Some(wire_list_read_preview::Detail::Search(Box::new(
SearchPreview {
canonical_query: "release decision".to_owned(),
scope_count: 7,
__buffa_unknown_fields: buffa::UnknownFields::default(),
},
))),
__buffa_unknown_fields: buffa::UnknownFields::default(),
}),
state: Some(recoverable_approval_entry::State::DecisionRecorded(
Box::new(RecordedApprovalDecision {
approved: true,
__buffa_unknown_fields: buffa::UnknownFields::default(),
}),
)),
__buffa_unknown_fields: buffa::UnknownFields::default(),
};
let recovered = RecoverableApproval::try_from(wire).expect("valid recovery projection");
assert_eq!(recovered.turn_id, "turn-1");
assert_eq!(recovered.request_id, "call-1");
assert_eq!(recovered.tool_name, "conversation_search");
assert_eq!(recovered.title, "Search past conversations");
assert_eq!(recovered.args_json, r#"{"query":"release"}"#);
assert_eq!(recovered.reason, "Review the search scope.");
assert_eq!(
recovered.read_preview,
Some(ReadPreview::Search {
canonical_query: "release decision".to_owned(),
scope_count: 7,
})
);
assert_eq!(
recovered.state,
ApprovalRecoveryState::DecisionRecorded(RecordedApprovalRecovery { approved: true })
);
}
fn standup_preview() -> ApprovalPreview {
ApprovalPreview {
routine_name: "standup".to_owned(),
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_plain_text_keeps_copy_and_removes_markup() {
let preview = standup_preview();
let now = at("2026-07-28T17:00:00Z");
let chat = approval_preview_text(&preview, now);
let plain = approval_preview_plain_text(&preview, now);
assert!(chat.contains("```"));
assert!(!plain.contains("```"));
assert!(plain.contains("This routine will run:"));
assert!(plain.contains("Wed Jul 29 9:00 AM tomorrow"));
}
#[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 {
routine_name: "standup".to_owned(),
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 {
routine_name: "standup".to_owned(),
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 {
routine_name: "standup".to_owned(),
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 {
turn_id: "018f47f0-5f70-7cc5-98df-123456789abc".to_owned(),
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::try_from(entry).expect("valid pending approval projection");
assert_eq!(pending.turn_id, "018f47f0-5f70-7cc5-98df-123456789abc");
assert_eq!(
pending.preview.as_ref().map(|p| p.prompt_text.as_str()),
Some("prompt")
);
}
#[test]
fn pending_approval_rejects_a_present_read_preview_without_a_detail() {
let entry = PendingApprovalEntry {
read_preview: buffa::MessageField::some(WireListReadPreview {
detail: None,
__buffa_unknown_fields: buffa::UnknownFields::default(),
}),
..Default::default()
};
assert!(matches!(
PendingApproval::try_from(entry),
Err(DialError::InvalidApprovalRecovery(
"approval read preview omitted its detail"
))
));
}
#[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)));
}
}