#![forbid(unsafe_code)]
#![warn(missing_docs)]
pub mod edge;
pub use edge::{EdgeAdapter, IngressDirective, Priority, build_attribution};
use std::sync::Arc;
use connectrpc::client::{ClientConfig, HttpClient};
use futures::Stream;
use polyc_agent::text_message;
use polyc_proto::proto::polychrome::agent::v1::{
AgentEnd, AgentRequest, AgentServiceClient, AgentStart, ClassifyRequest,
CompactionReason as WireCompactionReason, ContextCompacted,
IngressDirective as WireIngressDirective, InterruptRequest, Message, ParticipantMessage,
PendingApproval as WireAgentPendingApproval, TurnFailureKind as WireTurnFailureKind, Verdict,
agent_response, content, tool_call_content,
};
use polyc_proto::proto::polychrome::approval::v1::{
ApprovalResponseRequest, ApprovalServiceClient, ListPendingRequest, PendingApprovalEntry,
};
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, AutoLinkOutcome, AutoLinkRequest, CompleteLinkRequest, DescribeRequest,
LinkOutcome, PersonaServiceClient, SetIncognitoRequest, StartDeepLinkRequest, StartLinkRequest,
};
use polyc_proto::proto::polychrome::routine::v1::{
ListLiveEnrollmentsRequest, RoutineServiceClient, StartEnrollmentRequest,
};
#[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),
}
impl DialError {
#[must_use]
pub const fn code(&self) -> Option<connectrpc::ErrorCode> {
match self {
Self::Connect(e) => Some(e.code),
Self::InvalidAddress { .. } | Self::Tls(_) => 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
)
)
}
}
const AGENT_DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_mins(3);
const CONTROL_DIAL_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(30);
const CONNECT_TIMEOUT: std::time::Duration = std::time::Duration::from_secs(5);
fn http_client_for(uri: &http::Uri) -> Result<HttpClient, DialError> {
if uri.scheme_str() == Some("https") {
use rustls_platform_verifier::ConfigVerifierExt;
let tls = rustls::ClientConfig::with_platform_verifier()
.map_err(|e| DialError::Tls(e.to_string()))?;
Ok(HttpClient::builder()
.connect_timeout(CONNECT_TIMEOUT)
.with_tls(std::sync::Arc::new(tls)))
} else {
Ok(HttpClient::builder()
.connect_timeout(CONNECT_TIMEOUT)
.plaintext())
}
}
fn traced_options() -> connectrpc::client::CallOptions {
let mut headers = http::HeaderMap::new();
polyc_runtime::propagation::inject_current_span_into(&mut headers);
connectrpc::client::CallOptions::default()
.with_headers(headers.into_iter().filter_map(|(n, v)| n.map(|n| (n, v))))
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum CompactionReason {
Summarized,
Truncated,
}
impl CompactionReason {
#[must_use]
pub fn notice_headline(self, summarized_messages: u32) -> String {
match self {
Self::Summarized => {
let plural = if summarized_messages == 1 { "" } else { "s" };
format!(
"🧠 Summarized {summarized_messages} earlier message{plural} to keep the \
conversation manageable"
)
}
Self::Truncated => {
"✂️ Trimmed earlier tool output to keep the conversation manageable".to_owned()
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum TurnEvent {
ContextCompacted {
reason: CompactionReason,
summarized_messages: u32,
summary_preview: String,
},
TextDelta(String),
ToolStarted {
name: String,
},
ApprovalPending {
request_id: String,
tool_name: String,
title: String,
args_json: String,
reason: String,
resolve_token: String,
},
HandoffStarted {
child_agent_id: String,
reason: String,
},
InviteDelivery {
target_user_id: String,
code: String,
inviter_display: String,
},
WalletLinkPrompt {
link_url: Option<String>,
},
PersonaCredentialPrompt {
link_url: Option<String>,
},
TurnFailed {
kind: TurnFailureKind,
message: String,
},
Done,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum TurnFailureKind {
RateLimit,
Timeout,
Unavailable,
Auth,
BadRequest,
Other,
}
impl TurnFailureKind {
#[must_use]
pub const fn is_retryable(self) -> bool {
matches!(self, Self::RateLimit | Self::Timeout | Self::Unavailable)
}
}
#[derive(Clone)]
pub struct AgentDialer {
client: Arc<AgentServiceClient<HttpClient>>,
}
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),
})
}
pub fn approval_dialer(addr: &str) -> Result<ApprovalDialer, DialError> {
ApprovalDialer::new(addr)
}
pub async fn run_turn(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
) -> Result<String, DialError> {
self.run_turn_with(conversation_id, exec_id, user_text, Attribution::default())
.await
}
pub async fn run_turn_with(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
attribution: Attribution,
) -> Result<String, DialError> {
Ok(self
.run_turn_with_approvals(
conversation_id,
exec_id,
user_text,
attribution,
IngressDirective::default(),
)
.await?
.reply)
}
pub async fn run_turn_with_approvals(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
attribution: Attribution,
ingress_directive: IngressDirective,
) -> Result<BufferedTurn, DialError> {
let request = build_request(
conversation_id,
exec_id,
vec![text_message("user", user_text)],
None,
attribution,
false,
ingress_directive,
"",
);
self.run_turn_buffered_request(request).await
}
pub async fn run_routine_turn(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
attribution: Attribution,
ingress_directive: IngressDirective,
occurrence: &str,
) -> Result<BufferedTurn, DialError> {
let request = build_request(
conversation_id,
exec_id,
vec![text_message("user", user_text)],
None,
attribution,
true,
ingress_directive,
occurrence,
);
self.run_turn_buffered_request(request).await
}
async fn run_turn_buffered_request(
&self,
request: AgentRequest,
) -> Result<BufferedTurn, DialError> {
let mut stream = self
.client
.connect_with_options(request, traced_options())
.await?;
let mut text_parts: Vec<String> = Vec::new();
let mut scaffolding: Vec<String> = Vec::new();
let mut wallet_link_prompt = WalletLinkPrompt::None;
let mut persona_credential_prompt: Option<String> = None;
let mut pending_approvals: Vec<PendingApprovalPrompt> = Vec::new();
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 link_url =
(!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
wallet_link_prompt = WalletLinkPrompt::Present(link_url);
}
if let Some(prompt) = end.persona_credential_prompt.into_option() {
let link_url =
(!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
persona_credential_prompt = link_url;
}
pending_approvals = end
.pending_approvals
.into_iter()
.map(PendingApprovalPrompt::from)
.collect();
}
Some(agent_response::Type::Compacted(_)) | None => {}
}
}
Ok(BufferedTurn {
reply: finalize_buffered_reply(
&text_parts,
&scaffolding,
wallet_link_prompt,
persona_credential_prompt.as_deref(),
),
pending_approvals,
})
}
pub async fn run_turn_streaming(
&self,
conversation_id: &str,
exec_id: &str,
user_text: &str,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
self.run_turn_streaming_messages(
conversation_id,
exec_id,
vec![text_message("user", user_text)],
)
.await
}
pub async fn run_turn_streaming_messages(
&self,
conversation_id: &str,
exec_id: &str,
messages: Vec<Message>,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
self.run_turn_streaming_messages_with(
conversation_id,
exec_id,
messages,
None,
Attribution::default(),
IngressDirective::default(),
)
.await
}
pub async fn run_turn_streaming_messages_with(
&self,
conversation_id: &str,
exec_id: &str,
messages: Vec<Message>,
payment_receipt: Option<PaymentReceipt>,
attribution: Attribution,
ingress_directive: IngressDirective,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
let request = build_request(
conversation_id,
exec_id,
messages,
payment_receipt,
attribution,
false,
ingress_directive,
"",
);
self.run_turn_streaming_request(request).await
}
async fn run_turn_streaming_request(
&self,
request: AgentRequest,
) -> Result<impl Stream<Item = Result<TurnEvent, DialError>>, DialError> {
let mut stream = self
.client
.connect_with_options(request, traced_options())
.await?;
Ok(async_stream::try_stream! {
let mut ended = false;
while let Some(view) = stream.message().await? {
let response = view.to_owned_message();
match response.r#type {
Some(agent_response::Type::Compacted(c)) => {
yield event_from_compacted(*c);
}
Some(agent_response::Type::Outputs(outputs)) => {
for msg in outputs.messages {
if let Some(event) = message_to_event(msg) {
yield event;
}
}
}
Some(agent_response::Type::End(end)) => {
for event in events_from_end(*end) {
yield event;
}
ended = true;
}
None => {}
}
}
if !ended {
yield TurnEvent::Done;
}
})
}
pub async fn should_respond(
&self,
conversation_id: &str,
bot_name: &str,
transcript: Vec<ParticipantMessage>,
) -> Result<bool, DialError> {
let request = ClassifyRequest {
conversation_id: conversation_id.to_owned(),
bot_name: bot_name.to_owned(),
transcript,
..Default::default()
};
let resp = self
.client
.classify_with_options(request, traced_options())
.await?
.into_owned();
Ok(resp.verdict.to_i32() == Verdict::VERDICT_RESPOND as i32)
}
pub async fn interrupt(&self, conversation_id: &str) -> Result<bool, DialError> {
let request = InterruptRequest {
conversation_id: conversation_id.to_owned(),
..Default::default()
};
let resp = self
.client
.interrupt_with_options(request, traced_options())
.await?
.into_owned();
Ok(resp.interrupted)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ApprovalChoice {
Approve,
ApproveForSession,
Deny,
Abort,
Defer,
}
impl ApprovalChoice {
#[must_use]
pub const fn approved(self) -> bool {
matches!(self, Self::Approve | Self::ApproveForSession)
}
#[must_use]
pub const fn approved_for_session(self) -> bool {
matches!(self, Self::ApproveForSession)
}
#[must_use]
pub const fn is_abort(self) -> bool {
matches!(self, Self::Abort)
}
#[must_use]
pub const fn is_defer(self) -> bool {
matches!(self, Self::Defer)
}
}
#[must_use]
pub fn approval_decided_text(label: &str, choice: ApprovalChoice, decider: &str) -> String {
match choice {
ApprovalChoice::ApproveForSession => format!(
"✅ Approved by {decider} — running \"{label}\"… (won't ask again this session)"
),
ApprovalChoice::Approve => format!("✅ Approved by {decider} — running \"{label}\"…"),
ApprovalChoice::Deny => format!("🚫 Denied by {decider} — \"{label}\" was not run."),
ApprovalChoice::Abort => {
format!("🛑 Aborted by {decider} — \"{label}\" was not run; the turn was stopped.")
}
ApprovalChoice::Defer => {
format!("↩️ Sent back by {decider} — \"{label}\" is still waiting for a decision.")
}
}
}
#[must_use]
pub fn approval_completed_text(label: &str, decider: &str, success: bool) -> String {
if success {
format!("✅ Approved by {decider} — \"{label}\" is done.")
} else {
format!("✅ Approved by {decider} — \"{label}\" ran but hit an error.")
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct ApprovalOutcome {
pub persisted: bool,
pub signature_hex: String,
pub signed_by_hex: String,
}
impl From<polyc_proto::proto::polychrome::approval::v1::ApprovalResponseReply> for ApprovalOutcome {
fn from(reply: polyc_proto::proto::polychrome::approval::v1::ApprovalResponseReply) -> Self {
Self {
persisted: reply.persisted,
signature_hex: reply.signature_hex,
signed_by_hex: reply.signed_by_hex,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingApproval {
pub request_id: String,
pub tool_name: String,
pub args_json: String,
pub reason: String,
pub resolve_token: String,
}
impl From<PendingApprovalEntry> for PendingApproval {
fn from(p: PendingApprovalEntry) -> Self {
Self {
request_id: p.request_id,
tool_name: p.tool_name,
args_json: p.args_json,
reason: p.reason,
resolve_token: p.resolve_token,
}
}
}
#[derive(Clone)]
pub struct ApprovalDialer {
client: Arc<ApprovalServiceClient<HttpClient>>,
}
impl ApprovalDialer {
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 client = ApprovalServiceClient::new(
http,
ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT),
);
Ok(Self {
client: Arc::new(client),
})
}
#[allow(clippy::too_many_arguments)] pub async fn respond(
&self,
request_id: &str,
choice: ApprovalChoice,
reason: &str,
conversation_id: &str,
modified_args_json: &str,
injected_context: &str,
resolve_token: &str,
responder: Option<ExternalIdentity>,
) -> Result<ApprovalOutcome, DialError> {
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 request = ApprovalResponseRequest {
request_id: request_id.to_owned(),
conversation_id: conversation_id.to_owned(),
decision: Some(decision),
resolve_token: resolve_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<PendingApproval>, DialError> {
let mut pending = Vec::new();
let mut page_token = String::new();
loop {
let request = ListPendingRequest {
conversation_id: conversation_id.to_owned(),
page_token: page_token.clone(),
..Default::default()
};
let reply = self
.client
.list_pending_with_options(request, traced_options())
.await?
.into_owned();
pending.extend(reply.pending.into_iter().map(PendingApproval::from));
if reply.next_page_token.is_empty() {
break;
}
page_token = reply.next_page_token;
}
Ok(pending)
}
pub async fn excise_taint(
&self,
conversation_id: &str,
positions: &[u64],
all_quarantined: bool,
source_only: bool,
reason: &str,
actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
) -> Result<ExcisionOutcome, DialError> {
use polyc_proto::proto::polychrome::approval::v1::ExciseTaintRequest;
let request = ExciseTaintRequest {
conversation_id: conversation_id.to_owned(),
positions: positions.to_vec(),
all_quarantined,
source_only,
reason: reason.to_owned(),
actor: buffa::MessageField::some(actor),
..Default::default()
};
let reply = self
.client
.excise_taint_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.into())
}
pub async fn replay_conversation(
&self,
conversation_id: &str,
from: Option<usize>,
to: Option<usize>,
over: Option<ReplayOverrideSpec>,
actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
) -> Result<ReplayReport, DialError> {
use polyc_proto::proto::polychrome::approval::v1::{
ReplayConversationRequest, ReplayOverride,
};
let override_msg = over.map(ReplayOverride::from);
let to_bound = |v: Option<usize>| v.and_then(|n| i64::try_from(n).ok()).unwrap_or(-1);
let request = ReplayConversationRequest {
conversation_id: conversation_id.to_owned(),
from: to_bound(from),
to: to_bound(to),
r#override: override_msg
.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
actor: buffa::MessageField::some(actor),
..Default::default()
};
let reply = self
.client
.replay_conversation_with_options(request, traced_options())
.await?
.into_owned();
let turns = reply
.turns
.into_iter()
.map(ReplayTurnVerdict::from)
.collect();
Ok(ReplayReport {
turns,
all_match: reply.all_match,
})
}
pub async fn verify_conversation(
&self,
conversation_id: &str,
actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
) -> Result<VerificationOutcome, DialError> {
use polyc_proto::proto::polychrome::approval::v1::VerifyConversationRequest;
let request = VerifyConversationRequest {
conversation_id: conversation_id.to_owned(),
actor: buffa::MessageField::some(actor),
..Default::default()
};
let reply = self
.client
.verify_conversation_with_options(request, traced_options())
.await?
.into_owned();
Ok(reply.into())
}
pub async fn repair_conversation(
&self,
conversation_id: &str,
actor: polyc_proto::proto::polychrome::persona::v1::ExternalIdentity,
) -> 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),
..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, 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,
}
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.",
}
}
}
#[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 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(Clone)]
pub struct PersonaDialer {
client: Arc<PersonaServiceClient<HttpClient>>,
}
impl PersonaDialer {
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 client = PersonaServiceClient::new(
http,
ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT),
);
Ok(Self {
client: Arc::new(client),
})
}
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 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::Unspecified) | None => LinkCeremony::Failed,
})
}
pub async fn describe(&self, identity: ExternalIdentity) -> Result<PersonaView, DialError> {
let request = DescribeRequest {
identity: buffa::MessageField::some(identity),
..Default::default()
};
let reply = self
.client
.describe_with_options(request, traced_options())
.await?
.into_owned();
let Some(profile) = reply.profile.into_option() else {
return Ok(PersonaView {
persona_id: reply.persona_id,
..Default::default()
});
};
Ok(PersonaView {
persona_id: reply.persona_id,
status: profile.status,
identities: profile
.identities
.into_iter()
.map(LinkedIdentity::from)
.collect(),
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct LinkedIdentity {
pub provider: String,
pub external_id: String,
pub display_name: String,
}
impl From<ExternalIdentity> for LinkedIdentity {
fn from(id: ExternalIdentity) -> Self {
Self {
provider: id.provider,
external_id: id.external_id,
display_name: id.display_name,
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct PersonaView {
pub persona_id: String,
pub status: String,
pub identities: Vec<LinkedIdentity>,
}
pub use polyc_proto::proto::polychrome::agent::v1::Message as TurnMessage;
pub use polyc_proto::proto::polychrome::agent::v1::ParticipantMessage as GateMessage;
pub use polyc_proto::proto::polychrome::agent::v1::PaymentReceipt;
pub use polyc_proto::proto::polychrome::persona::v1::ExternalIdentity;
pub use connectrpc::ErrorCode;
#[derive(Debug, Clone, Default)]
pub struct Attribution {
pub caller: Option<ExternalIdentity>,
pub participants: Vec<ExternalIdentity>,
}
#[must_use]
pub fn attributed_message(speaker: &str, text: &str) -> TurnMessage {
text_message("user", &format!("{speaker}: {text}"))
}
#[must_use]
pub fn user_message(text: &str) -> TurnMessage {
text_message("user", text)
}
#[must_use]
pub fn namespaced_id(namespace: &str, native_id: &str) -> String {
format!("{namespace}:{native_id}")
}
#[must_use]
pub fn enrollment_conversation_id(routine: &str, persona_id: &str) -> String {
debug_assert!(
!routine.contains(':'),
"routine name `{routine}` must be DNS-1123-shaped (no ':'), enforced by the CRD name \
grammar Kubernetes applies to `Routine.metadata.name`"
);
debug_assert!(
!persona_id.contains(':'),
"persona id `{persona_id}` must be a bare UUID (no ':'), minted by \
`polyc_persona::new_provisional` as `uuid::Uuid::now_v7().to_string()`"
);
format!("cron:{routine}:{persona_id}")
}
#[must_use]
pub fn parse_enrollment_conversation_id(conversation_id: &str) -> Option<(&str, &str)> {
let rest = conversation_id.strip_prefix("cron:")?;
let (routine, persona) = rest.split_once(':')?;
if routine.is_empty() || persona.is_empty() || persona.contains(':') {
return None;
}
Some((routine, persona))
}
#[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())
}
#[allow(clippy::too_many_arguments)]
fn build_request(
conversation_id: &str,
exec_id: &str,
messages: Vec<Message>,
payment_receipt: Option<PaymentReceipt>,
attribution: Attribution,
ephemeral_history: bool,
ingress_directive: IngressDirective,
occurrence: &str,
) -> AgentRequest {
AgentRequest {
conversation_id: conversation_id.to_owned(),
exec_id: exec_id.to_owned(),
start: buffa::MessageField::some(AgentStart {
agent_id: String::new(),
agent_config: Vec::new(),
messages,
payment_receipt: payment_receipt
.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
caller: attribution
.caller
.map_or_else(buffa::MessageField::none, buffa::MessageField::some),
participants: attribution.participants,
ephemeral_history,
ingress_directive: wire_ingress_directive(ingress_directive),
occurrence: occurrence.to_owned(),
..Default::default()
}),
..Default::default()
}
}
fn events_from_end(end: AgentEnd) -> Vec<TurnEvent> {
let mut events: Vec<TurnEvent> = end
.pending_approvals
.into_iter()
.map(PendingApprovalPrompt::from)
.map(TurnEvent::from)
.collect();
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 link_url = (!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
events.push(TurnEvent::WalletLinkPrompt { link_url });
}
if let Some(prompt) = end.persona_credential_prompt.into_option() {
let link_url = (!prompt.link_url.trim().is_empty()).then_some(prompt.link_url);
events.push(TurnEvent::PersonaCredentialPrompt { link_url });
}
if let Some(failure) = end.failure.into_option() {
events.push(TurnEvent::TurnFailed {
kind: turn_failure_kind_from_wire(failure.kind.to_i32()),
message: failure.message,
});
}
events.push(TurnEvent::Done);
events
}
const fn turn_failure_kind_from_wire(kind: i32) -> TurnFailureKind {
if kind == WireTurnFailureKind::TURN_FAILURE_KIND_RATE_LIMIT as i32 {
TurnFailureKind::RateLimit
} else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_TIMEOUT as i32 {
TurnFailureKind::Timeout
} else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_UNAVAILABLE as i32 {
TurnFailureKind::Unavailable
} else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_AUTH as i32 {
TurnFailureKind::Auth
} else if kind == WireTurnFailureKind::TURN_FAILURE_KIND_BAD_REQUEST as i32 {
TurnFailureKind::BadRequest
} else {
TurnFailureKind::Other
}
}
fn event_from_compacted(c: ContextCompacted) -> TurnEvent {
let reason = if c.reason.to_i32() == WireCompactionReason::COMPACTION_REASON_SUMMARIZED as i32 {
CompactionReason::Summarized
} else {
CompactionReason::Truncated
};
TurnEvent::ContextCompacted {
reason,
summarized_messages: c.summarized_messages,
summary_preview: c.summary_preview,
}
}
fn message_to_event(msg: Message) -> Option<TurnEvent> {
if msg.internal_only {
return None;
}
let content_block = msg.content.into_option()?;
match content_block.r#type? {
content::Type::ToolCall(tc) => {
let name = match tc.r#type.as_ref() {
Some(tool_call_content::Type::FunctionCall(fc)) if !fc.name.is_empty() => {
fc.name.clone()
}
_ => tc.id.clone(),
};
Some(TurnEvent::ToolStarted { name })
}
content::Type::Text(t) => {
if is_assistant_role(&msg.role) && !t.text.is_empty() {
Some(TurnEvent::TextDelta(t.text))
} else {
None
}
}
_ => None,
}
}
fn is_assistant_role(role: &str) -> bool {
role == "model" || role == "assistant"
}
fn render_content(ty: Option<content::Type>) -> Option<String> {
match ty? {
content::Type::Text(t) => {
if t.text.is_empty() {
None
} else {
Some(t.text)
}
}
content::Type::ToolCall(tc) => {
let name = match tc.r#type.as_ref() {
Some(tool_call_content::Type::FunctionCall(fc)) => fc.name.as_str(),
None => "",
};
if name.is_empty() {
Some(format!("[tool_call:{}]", tc.id))
} else {
Some(format!("[tool_call:{name} {}]", tc.id))
}
}
content::Type::ToolResult(tr) => Some(format!("[tool_result:{}]", tr.call_id)),
_ => None,
}
}
fn aggregate_output_message(
msg: Message,
text_parts: &mut Vec<String>,
scaffolding: &mut Vec<String>,
) {
if msg.internal_only {
return;
}
let is_assistant = matches!(msg.role.as_str(), "model" | "assistant");
let Some(content_block) = msg.content.into_option() else {
return;
};
match content_block.r#type {
Some(content::Type::Text(t)) if is_assistant && !t.text.is_empty() => {
text_parts.push(t.text);
}
Some(content::Type::Text(_)) => {}
other => {
if let Some(rendered) = render_content(other) {
scaffolding.push(rendered);
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq, Default)]
pub struct BufferedTurn {
pub reply: String,
pub pending_approvals: Vec<PendingApprovalPrompt>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct PendingApprovalPrompt {
pub request_id: String,
pub tool_name: String,
pub title: String,
pub args_json: String,
pub reason: String,
pub resolve_token: String,
}
impl From<WireAgentPendingApproval> for PendingApprovalPrompt {
fn from(pa: WireAgentPendingApproval) -> Self {
Self {
request_id: pa.request_id,
tool_name: pa.tool_name,
title: pa.title,
args_json: pa.args_json,
reason: pa.reason,
resolve_token: pa.resolve_token,
}
}
}
impl From<PendingApprovalPrompt> for TurnEvent {
fn from(p: PendingApprovalPrompt) -> Self {
Self::ApprovalPending {
request_id: p.request_id,
tool_name: p.tool_name,
title: p.title,
args_json: p.args_json,
reason: p.reason,
resolve_token: p.resolve_token,
}
}
}
enum WalletLinkPrompt {
None,
Present(Option<String>),
}
fn finalize_buffered_reply(
text_parts: &[String],
scaffolding: &[String],
wallet_link_prompt: WalletLinkPrompt,
persona_credential_prompt: Option<&str>,
) -> String {
if let WalletLinkPrompt::Present(link_url) = wallet_link_prompt {
return polyc_proto::wallet_link_prompt(link_url.as_deref());
}
if let Some(url) = persona_credential_prompt {
return polyc_proto::persona_credential_prompt(Some(url));
}
if text_parts.is_empty() {
scaffolding.join("\n")
} else {
text_parts.join("\n")
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum OpsAction {
UpgradeTo {
version: String,
},
EnrollmentNudge {
routine: String,
ceremony_url: String,
},
UpgradeOutcome {
version: String,
outcome: UpgradeOutcomeKind,
},
ContentDelivery {
template: String,
destination: std::collections::BTreeMap<String, String>,
body: String,
},
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> {
let uri = addr
.parse::<http::Uri>()
.map_err(|source| DialError::InvalidAddress {
addr: addr.to_owned(),
source,
})?;
let http = http_client_for(&uri)?;
let client = NotificationServiceClient::new(
http,
ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT),
);
Ok(Self {
client: Arc::new(client),
})
}
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());
}
})
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ContentDeliveryOutcome {
Delivered,
PostFailed,
AckFailed,
}
pub async fn deliver_content_notice<F, Fut, E>(
notifications: &NotificationDialer,
provider: &str,
notice: &PendingNotice,
post: F,
) -> ContentDeliveryOutcome
where
F: FnOnce() -> Fut,
Fut: std::future::Future<Output = Result<(), E>>,
E: std::fmt::Display,
{
if let Err(err) = post().await {
tracing::warn!(
error = %err,
%provider,
target = %notice.target,
"ops notifier: content post failed"
);
return ContentDeliveryOutcome::PostFailed;
}
match notifications
.ack(provider, ¬ice.action_id, ¬ice.target)
.await
{
Ok(_) => {
tracing::info!(
action_id = %notice.action_id,
%provider,
target = %notice.target,
"routine content delivered"
);
ContentDeliveryOutcome::Delivered
}
Err(err) => {
tracing::warn!(
error = %err,
action_id = %notice.action_id,
%provider,
"ops notifier: content ack failed after post"
);
ContentDeliveryOutcome::AckFailed
}
}
}
fn pending_notice(
p: polyc_proto::proto::polychrome::ops::v1::PendingNotification,
) -> PendingNotice {
let action = p.action.into_option().and_then(|view| view.action).map_or(
OpsAction::Unknown,
|a| match a {
ops_action_view::Action::UpgradeTo(u) => OpsAction::UpgradeTo { version: u.version },
ops_action_view::Action::EnrollmentNudge(n) => OpsAction::EnrollmentNudge {
routine: n.routine,
ceremony_url: n.ceremony_url,
},
ops_action_view::Action::UpgradeOutcome(u) => OpsAction::UpgradeOutcome {
version: u.version,
outcome: match u.kind.as_known() {
Some(upgrade_outcome::Kind::SUCCESS) => UpgradeOutcomeKind::Success,
Some(upgrade_outcome::Kind::FAILED) => UpgradeOutcomeKind::Failed,
Some(
upgrade_outcome::Kind::UNKNOWN | upgrade_outcome::Kind::KIND_UNSPECIFIED,
)
| None => UpgradeOutcomeKind::Unknown,
},
},
ops_action_view::Action::ContentDelivery(c) => OpsAction::ContentDelivery {
template: c.template,
destination: c.destination.into_iter().collect(),
body: c.body,
},
},
);
PendingNotice {
action_id: p.action_id,
target: p.target,
action,
expires_unix: p.expires_unix,
payload_hash: p.payload_hash,
delivered: p.delivered,
}
}
#[derive(Debug, Clone, PartialEq)]
pub struct LiveEnrollment {
pub persona_id: String,
pub notify: ExternalIdentity,
pub conversation_id: String,
}
impl From<polyc_proto::proto::polychrome::routine::v1::LiveEnrollment> for LiveEnrollment {
fn from(e: polyc_proto::proto::polychrome::routine::v1::LiveEnrollment) -> Self {
Self {
persona_id: e.persona_id,
notify: e.notify_identity.into_option().unwrap_or_default(),
conversation_id: e.conversation_id,
}
}
}
#[derive(Debug, Clone, PartialEq, Default)]
pub struct RoutineEnrollments {
pub known: bool,
pub active: Vec<LiveEnrollment>,
}
#[derive(Clone)]
pub struct LiveEnrollmentsDialer {
client: Arc<RoutineServiceClient<HttpClient>>,
}
impl LiveEnrollmentsDialer {
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 client = RoutineServiceClient::new(
http,
ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT),
);
Ok(Self {
client: Arc::new(client),
})
}
pub async fn list(
&self,
routine: &str,
bearer_token: &str,
originate_nudges: bool,
) -> Result<RoutineEnrollments, DialError> {
let mut active = Vec::new();
let mut page_token = String::new();
let known = loop {
let request = ListLiveEnrollmentsRequest {
routine: routine.to_owned(),
originate_nudges: originate_nudges && page_token.is_empty(),
page_token: page_token.clone(),
bearer_token: bearer_token.to_owned(),
..Default::default()
};
let reply = self
.client
.list_live_enrollments_with_options(request, traced_options())
.await?
.into_owned();
active.extend(reply.enrollments.into_iter().map(LiveEnrollment::from));
if reply.next_page_token.is_empty() {
break reply.known;
}
page_token = reply.next_page_token;
};
Ok(RoutineEnrollments { known, active })
}
pub async fn start_enrollment(
&self,
routine: &str,
notify_identity: ExternalIdentity,
) -> Result<StartedEnrollment, DialError> {
let request = StartEnrollmentRequest {
routine: routine.to_owned(),
notify_identity: buffa::MessageField::some(notify_identity),
..Default::default()
};
let reply = self
.client
.start_enrollment_with_options(request, traced_options())
.await?
.into_owned();
Ok(StartedEnrollment {
ceremony_url: reply.ceremony_url,
enrollment_id: reply.enrollment_id,
persona_id: reply.persona_id,
})
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct StartedEnrollment {
pub ceremony_url: String,
pub enrollment_id: String,
pub persona_id: String,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum DecideOutcome {
Applied,
Denied,
Rejected,
Unknown,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct DecideResult {
pub outcome: DecideOutcome,
pub detail: String,
}
pub const OPS_PROMPT_HEADING: &str = "Approval needed";
impl OpsAction {
#[must_use]
pub fn summary(&self) -> String {
match self {
Self::UpgradeTo { version } => {
let copy = polyc_runtime::update_copy::update_copy(
&polyc_runtime::compat::Compatibility::Warm,
version,
);
format!("{} — {}", copy.headline, copy.detail)
}
Self::EnrollmentNudge {
routine,
ceremony_url,
} => {
if ceremony_url.is_empty() {
format!(
"The {routine} routine could not run because its approval is missing. \
Ask whoever runs your deployment where to re-approve it."
)
} else {
format!(
"The {routine} routine could not run because its approval is \
missing. Approve it on the enrollment page and the next \
scheduled run will go through: {ceremony_url}"
)
}
}
Self::UpgradeOutcome { version, outcome } => match outcome {
UpgradeOutcomeKind::Success => {
format!("✅ The cluster is now running {version}.")
}
UpgradeOutcomeKind::Failed => format!(
"⚠️ The upgrade to {version} did not confirm healthy — check `polychrome status`."
),
UpgradeOutcomeKind::Unknown => format!(
"❓ Could not confirm whether the upgrade to {version} finished — check `polychrome status`."
),
},
Self::ContentDelivery { body, .. } => body.clone(),
Self::Unknown => "Approve a pending action".to_owned(),
}
}
#[must_use]
pub const fn content_delivery(
&self,
) -> Option<(&std::collections::BTreeMap<String, String>, &str)> {
match self {
Self::ContentDelivery {
destination, body, ..
} => Some((destination, body.as_str())),
_ => None,
}
}
#[must_use]
pub fn approve_verb(&self) -> Option<&'static str> {
match self {
Self::UpgradeTo { version } => {
polyc_runtime::update_copy::update_copy(
&polyc_runtime::compat::Compatibility::Warm,
version,
)
.action
}
Self::EnrollmentNudge { .. }
| Self::UpgradeOutcome { .. }
| Self::ContentDelivery { .. }
| Self::Unknown => None,
}
}
#[must_use]
pub const fn is_decision(&self) -> bool {
!matches!(
self,
Self::EnrollmentNudge { .. }
| Self::UpgradeOutcome { .. }
| Self::ContentDelivery { .. }
)
}
}
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> {
let uri = addr
.parse::<http::Uri>()
.map_err(|source| DialError::InvalidAddress {
addr: addr.to_owned(),
source,
})?;
let http = http_client_for(&uri)?;
let client = OperatorMailboxServiceClient::new(
http,
ClientConfig::new(uri).with_default_timeout(CONTROL_DIAL_TIMEOUT),
);
Ok(Self {
client: Arc::new(client),
})
}
pub async fn decide(
&self,
action_id: &str,
approved: bool,
provider: &str,
external_user_id: &str,
reason: &str,
payload_hash: &str,
) -> Result<DecideResult, DialError> {
let request = DecideRequest {
action_id: action_id.to_owned(),
approved,
provider: provider.to_owned(),
external_user_id: external_user_id.to_owned(),
reason: reason.to_owned(),
payload_hash: payload_hash.to_owned(),
..Default::default()
};
let reply = self
.client
.decide_with_options(request, traced_options())
.await?
.into_owned();
let outcome = match reply.outcome.as_known() {
Some(decide_reply::Outcome::APPLIED) => DecideOutcome::Applied,
Some(decide_reply::Outcome::DENIED) => DecideOutcome::Denied,
Some(decide_reply::Outcome::REJECTED) => DecideOutcome::Rejected,
Some(decide_reply::Outcome::OUTCOME_UNSPECIFIED) | None => DecideOutcome::Unknown,
};
Ok(DecideResult {
outcome,
detail: reply.detail,
})
}
}
#[cfg(test)]
mod tests {
#![allow(clippy::pedantic, clippy::nursery, missing_docs)]
use super::*;
use polyc_proto::proto::polychrome::agent::v1::{
Content, FunctionCallContent, TextContent, ThoughtContent, ThoughtSummaryContent,
ToolCallContent, ToolResultContent, thought_summary_content,
};
#[test]
fn 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 an operator".to_owned(),
}
.decided_line("Chris");
assert!(rejected.contains("Rejected") && rejected.contains("not an operator"));
let unknown = DecideResult {
outcome: DecideOutcome::Unknown,
detail: String::new(),
}
.decided_line("Chris");
assert!(!unknown.to_lowercase().contains("please"));
}
#[test]
fn enrollment_nudge_copy_is_a_link_not_a_decision() {
let nudge = OpsAction::EnrollmentNudge {
routine: "standup".to_owned(),
ceremony_url: "https://enroll.polychrome.test/v1/enroll/manage/standup/persona-9"
.to_owned(),
};
assert!(!nudge.is_decision(), "a nudge carries no approve/deny");
assert_eq!(nudge.approve_verb(), None);
let summary = nudge.summary();
assert!(summary.contains("standup"), "names the routine: {summary}");
assert!(
summary.contains("https://enroll.polychrome.test/v1/enroll/manage/standup/persona-9"),
"carries the exact ceremony link: {summary}"
);
let lower = summary.to_lowercase();
for banned in [
"operator",
"sub-agent",
"trifecta",
"context budget",
"state-changing action",
"please",
"sorry",
"unfortunately",
] {
assert!(
!lower.contains(banned),
"banned word {banned:?} in: {summary}"
);
}
assert!(
OpsAction::UpgradeTo {
version: "1.2.3".to_owned()
}
.is_decision()
);
assert!(OpsAction::Unknown.is_decision());
}
#[test]
fn enrollment_nudge_omits_the_link_when_composed_under_a_loopback_base() {
let nudge = OpsAction::EnrollmentNudge {
routine: "standup".to_owned(),
ceremony_url: String::new(),
};
assert!(!nudge.is_decision(), "still a plain DM, not a decision");
let summary = nudge.summary();
assert!(
summary.contains("standup"),
"still names the routine: {summary}"
);
assert!(
!summary.contains("http"),
"no link of any scheme ships when the base was unreachable: {summary}"
);
let lower = summary.to_lowercase();
for banned in [
"operator",
"sub-agent",
"trifecta",
"context budget",
"state-changing action",
"please",
"sorry",
"unfortunately",
] {
assert!(
!lower.contains(banned),
"banned word {banned:?} in: {summary}"
);
}
}
#[test]
fn content_delivery_maps_off_the_wire_and_is_a_channel_post() {
use polyc_proto::proto::polychrome::ops::v1::{
ContentDelivery as WireContentDelivery, OpsActionView, PendingNotification,
ops_action_view,
};
let wire = PendingNotification {
action_id: "content_delivery:tick-42:standup_summary_v1".to_owned(),
target: "C0STANDUP".to_owned(),
action: buffa::MessageField::some(OpsActionView {
action: Some(ops_action_view::Action::ContentDelivery(Box::new(
WireContentDelivery {
template: "standup_summary_v1".to_owned(),
destination: [("channel".to_owned(), "C0STANDUP".to_owned())]
.into_iter()
.collect(),
body: "shipped the release".to_owned(),
..Default::default()
},
))),
..Default::default()
}),
delivered: false,
..Default::default()
};
let notice = super::pending_notice(wire);
assert!(
!notice.action.is_decision(),
"no approve/deny on a delivery"
);
assert_eq!(notice.action.approve_verb(), None);
let (destination, body) = notice
.action
.content_delivery()
.expect("a content delivery exposes its destination + body");
assert_eq!(
destination.get("channel").map(String::as_str),
Some("C0STANDUP")
);
assert_eq!(body, "shipped the release");
assert_eq!(notice.action.summary(), "shipped the release");
assert!(
OpsAction::UpgradeTo {
version: "1.0.0".to_owned()
}
.content_delivery()
.is_none()
);
}
#[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());
}
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(Some(url.to_owned())),
None,
);
assert_eq!(reply, polyc_proto::wallet_link_prompt(Some(url)));
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(None), None);
assert_eq!(reply, polyc_proto::wallet_link_prompt(None));
}
#[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(Some(url.to_owned())),
None,
);
assert_eq!(
reply,
polyc_proto::wallet_link_prompt(Some(url)),
"preceding text_parts must not survive — it may be the model's own \
unfiltered wallet-link narration, not genuine unrelated content: {reply}"
);
}
#[test]
fn finalize_buffered_reply_without_wallet_link_prompt_is_unchanged() {
assert_eq!(
finalize_buffered_reply(
&["the answer".to_owned()],
&[],
WalletLinkPrompt::None,
None
),
"the answer"
);
assert_eq!(
finalize_buffered_reply(
&[],
&["[tool_call:foo]".to_owned()],
WalletLinkPrompt::None,
None
),
"[tool_call:foo]"
);
assert_eq!(
finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, None),
""
);
}
#[test]
fn finalize_buffered_reply_persona_credential_prompt_wins() {
let url = "https://wallet.polychrome.example/enroll-passkey?token=abc";
let reply = finalize_buffered_reply(&[], &[], WalletLinkPrompt::None, Some(url));
assert_eq!(reply, polyc_proto::persona_credential_prompt(Some(url)));
assert!(
reply.contains(url),
"the deterministic prompt carries the link: {reply}"
);
}
#[test]
fn finalize_buffered_reply_persona_credential_prompt_replaces_preceding_content() {
let url = "https://wallet.polychrome.example/enroll-passkey?token=abc";
let reply = finalize_buffered_reply(
&["You can approve purchases faster by setting up a passkey.".to_owned()],
&[],
WalletLinkPrompt::None,
Some(url),
);
assert_eq!(
reply,
polyc_proto::persona_credential_prompt(Some(url)),
"preceding text_parts must not survive — it may be the model's own \
unfiltered narration of the same link: {reply}"
);
}
fn message(role: &str, ty: Option<content::Type>) -> Message {
Message {
role: role.to_owned(),
content: buffa::MessageField::some(Content {
r#type: ty,
..Default::default()
}),
..Default::default()
}
}
#[test]
fn assistant_text_becomes_text_delta() {
assert_eq!(
message_to_event(message("assistant", text("hello"))),
Some(TurnEvent::TextDelta("hello".to_owned()))
);
}
#[test]
fn model_role_also_counts_as_assistant() {
assert_eq!(
message_to_event(message("model", text("hi"))),
Some(TurnEvent::TextDelta("hi".to_owned()))
);
}
#[test]
fn tool_role_text_is_skipped() {
assert_eq!(message_to_event(message("tool", text("result blob"))), None);
}
#[test]
fn empty_assistant_text_is_skipped() {
assert_eq!(message_to_event(message("assistant", text(""))), None);
}
#[test]
fn internal_only_assistant_text_is_skipped() {
let mut msg = message("assistant", text("pending your approval"));
msg.internal_only = true;
assert_eq!(message_to_event(msg), None);
}
#[test]
fn tool_call_becomes_tool_started_with_name() {
assert_eq!(
message_to_event(message("model", tool_call("call_7", "search"))),
Some(TurnEvent::ToolStarted {
name: "search".to_owned()
})
);
}
#[test]
fn tool_call_falls_back_to_call_id() {
assert_eq!(
message_to_event(message(
"tool",
Some(content::Type::ToolCall(Box::new(ToolCallContent {
id: "call_bare".to_owned(),
r#type: None,
..Default::default()
})))
)),
Some(TurnEvent::ToolStarted {
name: "call_bare".to_owned()
})
);
}
#[test]
fn tool_result_produces_no_event() {
assert_eq!(
message_to_event(message("tool", tool_result("call_7"))),
None
);
}
fn map_turn(msgs: Vec<Message>) -> Vec<TurnEvent> {
let mut events: Vec<TurnEvent> = msgs.into_iter().filter_map(message_to_event).collect();
events.push(TurnEvent::Done);
events
}
#[test]
fn synthetic_turn_yields_expected_event_sequence() {
let turn = vec![
message("model", text("Let me look that up.")),
message("tool", tool_call("call_1", "search")),
message("model", text("Found it.")),
];
assert_eq!(
map_turn(turn),
vec![
TurnEvent::TextDelta("Let me look that up.".to_owned()),
TurnEvent::ToolStarted {
name: "search".to_owned()
},
TurnEvent::TextDelta("Found it.".to_owned()),
TurnEvent::Done,
]
);
}
use polyc_proto::proto::polychrome::agent::v1::{
Handoff as WireHandoff, PendingApproval as WirePendingApproval,
};
#[test]
fn end_with_nothing_yields_only_done() {
assert_eq!(events_from_end(AgentEnd::default()), vec![TurnEvent::Done]);
}
#[test]
fn end_with_handoff_yields_handoff_then_done() {
let end = AgentEnd {
handoff: buffa::MessageField::some(WireHandoff {
call_id: "call_1".to_owned(),
child_agent_id: "researcher".to_owned(),
reason: "needs deep dive".to_owned(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::HandoffStarted {
child_agent_id: "researcher".to_owned(),
reason: "needs deep dive".to_owned(),
},
TurnEvent::Done,
]
);
}
#[test]
fn end_orders_approvals_before_handoff_before_done() {
let end = AgentEnd {
pending_approvals: vec![WirePendingApproval {
request_id: "r1".to_owned(),
tool_name: "delete_file".to_owned(),
args_json: "{}".to_owned(),
title: "Delete a file".to_owned(),
..Default::default()
}],
handoff: buffa::MessageField::some(WireHandoff {
child_agent_id: "child".to_owned(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::ApprovalPending {
request_id: "r1".to_owned(),
tool_name: "delete_file".to_owned(),
title: "Delete a file".to_owned(),
args_json: "{}".to_owned(),
reason: String::new(),
resolve_token: String::new(),
},
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()),
},
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 },
TurnEvent::Done,
]
);
}
#[test]
fn end_projects_persona_credential_prompt_before_done() {
use polyc_proto::proto::polychrome::agent::v1::PersonaCredentialPrompt as WirePersonaCredentialPrompt;
let end = AgentEnd {
persona_credential_prompt: buffa::MessageField::some(WirePersonaCredentialPrompt {
link_url: "https://wallet.polychrome.example/enroll-passkey?token=abc".to_owned(),
..Default::default()
}),
..Default::default()
};
assert_eq!(
events_from_end(end),
vec![
TurnEvent::PersonaCredentialPrompt {
link_url: Some(
"https://wallet.polychrome.example/enroll-passkey?token=abc".to_owned()
),
},
TurnEvent::Done,
]
);
}
#[test]
fn end_without_persona_credential_prompt_yields_only_done() {
let end = AgentEnd::default();
assert_eq!(events_from_end(end), vec![TurnEvent::Done]);
}
#[test]
fn namespaced_id_is_prefix_colon_native() {
assert_eq!(
namespaced_id("mail", "CAF=abc@mail.example"),
"mail:CAF=abc@mail.example"
);
assert_eq!(namespaced_id("web", "abc-123"), "web:abc-123");
}
#[test]
fn enrollment_conversation_id_is_cron_routine_persona() {
assert_eq!(
enrollment_conversation_id("standup", "persona-9"),
"cron:standup:persona-9"
);
}
#[test]
fn enrollment_conversation_id_is_deterministic() {
for (routine, persona) in [
("standup", "persona-1"),
("weekly-digest", "abcdef01-2345-6789-abcd-ef0123456789"),
("", ""),
] {
assert_eq!(
enrollment_conversation_id(routine, persona),
enrollment_conversation_id(routine, persona),
);
}
}
#[test]
fn enrollment_conversation_id_is_distinct_per_routine_and_persona() {
let base = enrollment_conversation_id("standup", "persona-1");
assert_ne!(base, enrollment_conversation_id("standup", "persona-2"));
assert_ne!(base, enrollment_conversation_id("digest", "persona-1"));
assert_ne!(base, enrollment_conversation_id("digest", "persona-2"));
assert_ne!(
enrollment_conversation_id("a", "b"),
enrollment_conversation_id("b", "a"),
);
}
#[test]
fn parse_enrollment_conversation_id_round_trips_and_rejects_plain_cron() {
for (routine, persona) in [
("standup", "persona-1"),
("weekly-digest", "abcdef01-2345-6789-abcd-ef0123456789"),
] {
let id = enrollment_conversation_id(routine, persona);
assert_eq!(
parse_enrollment_conversation_id(&id),
Some((routine, persona)),
"round-trips to its (routine, persona)"
);
}
assert_eq!(
parse_enrollment_conversation_id("cron:nightly-report"),
None
);
assert_eq!(parse_enrollment_conversation_id("web:abc-123"), None);
assert_eq!(parse_enrollment_conversation_id("cron::persona"), None);
assert_eq!(parse_enrollment_conversation_id("cron:standup:"), None);
}
#[test]
fn dns_1123_subdomain_names_never_contain_colon() {
for name in [
"standup",
"weekly-digest",
"a",
"a1-b2",
"sub.domain.example",
"routine-123",
"x.y.z",
&"a".repeat(63),
] {
assert!(
is_dns_1123_subdomain(name),
"`{name}` expected to be a valid DNS-1123 subdomain fixture"
);
assert!(
!name.contains(':'),
"DNS-1123 subdomain `{name}` must never contain ':'"
);
}
}
fn is_dns_1123_subdomain(name: &str) -> bool {
!name.is_empty()
&& name.split('.').all(|label| {
!label.is_empty()
&& label.as_bytes()[0].is_ascii_lowercase()
&& label.ends_with(|c: char| c.is_ascii_lowercase() || c.is_ascii_digit())
&& label
.chars()
.all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '-')
})
}
#[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")
);
}
}