#![cfg_attr(feature = "fail-on-warnings", deny(warnings))]
#![warn(clippy::all, clippy::pedantic, clippy::nursery, clippy::cargo)]
#![allow(clippy::multiple_crate_versions)]
mod generation;
use bcode_agent_profile::{AgentInfo, PolicyStatusResponse};
use bcode_daemon_lifecycle::{DaemonStartError, EnsureDaemonOptions, ensure_daemon_running};
mod contextual;
use bcode_ipc::{
ClientRuntimeContext, CodecError, EnvelopeKind, ErrorResponse, Event, IpcEndpoint,
LocalIpcStream, PluginServiceResponse, PluginServiceSummary, Request, Response,
ResponsePayload, ServerStopMode, SessionBulkMigrationOperationStatus,
SessionBulkMigrationStartRequest, SessionCompatibilityInventoryRequest,
SessionCompatibilityInventoryResponse, WorktreeCreateOperationStatus, WorktreeCreateRequest,
WorktreeCreateResponse, WorktreeListRequest, WorktreeListResponse, WorktreeRemoveRequest,
WorktreeRemoveResponse, current_working_directory, decode_event, decode_response,
default_endpoint, recv_envelope, request_envelope, send_envelope,
};
use bcode_plugin_sdk::PluginContributions;
use bcode_ralph_models::{
RalphApproveRequest, RalphCancelRequest, RalphCancelResponse, RalphLifecycleRequest,
RalphListIterationsRequest, RalphListIterationsResponse, RalphListRunsRequest,
RalphListRunsResponse, RalphResumeRequest, RalphResumeResponse, RalphRunRequest,
RalphRunResponse, RalphRunStatusRequest, RalphRunStatusResponse, RalphStatusRequest,
RalphStatusResponse,
};
use bcode_session_import::ImportWarning as SessionImportWarning;
use bcode_session_models::{
ClientId, ProjectionWindowRequest, SessionCatalogSourceStatus, SessionCatalogStatus,
SessionDerivationPromptPage, SessionDerivationPromptQuery, SessionDerivationRequest,
SessionDerivationSourceSnapshot, SessionDerivationTerminalOutcome, SessionEvent,
SessionEventKind, SessionHistoryAroundQuery, SessionHistoryPage, SessionHistoryQuery,
SessionHistoryWindow, SessionId, SessionInputHistoryEntry, SessionInspectionPage,
SessionInspectionQuery, SessionSummary, WorkId,
};
use bcode_session_models::{PendingToolExchangeSummary, PermissionSummary};
use bcode_skill_models::{SkillId, SkillList, SkillManifest};
use std::collections::{BTreeMap, VecDeque};
use std::path::PathBuf;
use std::sync::Arc;
use std::time::Duration;
use thiserror::Error;
const DEFAULT_CLIENT_IPC_REQUEST_TIMEOUT: Duration = Duration::from_secs(15);
const DEFAULT_CLIENT_CONNECT_TIMEOUT: Duration = Duration::from_secs(5);
const DEFAULT_CLIENT_DAEMON_START_TIMEOUT: Duration = Duration::from_secs(30);
const LONG_POLL_TRANSPORT_GRACE: Duration = Duration::from_secs(5);
pub use bcode_ipc::SessionProgressSnapshot;
pub use bcode_session_models::SessionArtifactRange;
fn decode_session_progress_response(
payload: ResponsePayload,
session_id: SessionId,
) -> Result<SessionProgressSnapshot, ClientError> {
match payload {
ResponsePayload::SessionProgressSnapshot { snapshot }
if snapshot.version == bcode_workflow_view_models::SESSION_PROGRESS_VERSION
&& snapshot.session_id == session_id.to_string() =>
{
Ok(snapshot)
}
ResponsePayload::SessionProgressSnapshot { .. } => Err(ClientError::Protocol(
"unsupported or mismatched session progress snapshot".to_owned(),
)),
_ => Err(ClientError::UnexpectedResponse),
}
}
fn validate_artifact_range_response(
range: &SessionArtifactRange,
artifact_id: &str,
reference_key: &str,
offset: u64,
length: u32,
) -> Result<(), ClientError> {
if range.artifact_id != artifact_id
|| range.reference_key != reference_key
|| range.offset != offset
|| offset > range.total_bytes
|| range.bytes.len() as u64 > u64::from(length)
|| range.bytes.len() as u64
> u64::from(bcode_session_models::MAX_SESSION_ARTIFACT_RANGE_BYTES)
|| (length > 0 && range.bytes.is_empty() && offset < range.total_bytes)
|| (!range.bytes.is_empty()
&& offset
.checked_add(range.bytes.len() as u64)
.is_none_or(|end| end > range.total_bytes))
{
return Err(ClientError::Protocol(
"artifact range response does not match the requested range".to_owned(),
));
}
Ok(())
}
#[cfg(test)]
mod artifact_range_tests {
use super::SessionArtifactRange;
#[test]
fn range_metadata_supports_eof_and_replacement_detection() {
let range = SessionArtifactRange {
artifact_id: "artifact".to_owned(),
reference_key: "recording".to_owned(),
content_type: Some("application/octet-stream".to_owned()),
offset: 8,
total_bytes: 10,
reference_bytes: Some(10),
reference_revision: 42,
finalized: true,
finalized_event_seq: Some(42),
availability: Some("complete".to_owned()),
complete: Some(true),
checksum_sha256: Some("abc".to_owned()),
bytes: b"89".to_vec(),
};
super::validate_artifact_range_response(&range, "artifact", "recording", 8, 2)
.expect("valid range");
for (id, key, offset, length) in [
("other", "recording", 8, 2),
("artifact", "other", 8, 2),
("artifact", "recording", 7, 2),
("artifact", "recording", 8, 1),
] {
assert!(
super::validate_artifact_range_response(&range, id, key, offset, length).is_err()
);
}
let mut invalid = range.clone();
invalid.total_bytes = 9;
assert!(
super::validate_artifact_range_response(&invalid, "artifact", "recording", 8, 2)
.is_err()
);
invalid.offset = u64::MAX;
invalid.total_bytes = u64::MAX;
assert!(
super::validate_artifact_range_response(&invalid, "artifact", "recording", u64::MAX, 2)
.is_err()
);
invalid.bytes.clear();
invalid.offset = 8;
assert!(
super::validate_artifact_range_response(&invalid, "artifact", "recording", 8, 2)
.is_err()
);
invalid.offset = u64::MAX;
super::validate_artifact_range_response(&invalid, "artifact", "recording", u64::MAX, 2)
.expect("empty EOF range");
assert_eq!(range.next_offset(), 10);
assert!(range.is_eof());
assert_eq!(range.finalized_event_seq, Some(42));
assert_eq!(range.checksum_sha256.as_deref(), Some("abc"));
}
}
pub use bcode_session_models::RuntimeWorkSpan;
const fn runtime_work_span_without_start(work_id: WorkId) -> RuntimeWorkSpan {
RuntimeWorkSpan {
work_id,
parent_work_id: None,
label: String::new(),
status: None,
started_at_ms: None,
finished_at_ms: None,
cancelled: false,
message: None,
}
}
fn runtime_work_spans(events: Vec<SessionEvent>) -> Vec<RuntimeWorkSpan> {
let mut spans = BTreeMap::<WorkId, RuntimeWorkSpan>::new();
for event in events {
let (SessionEventKind::RuntimeWorkStarted { work_id, .. }
| SessionEventKind::RuntimeWorkCancelRequested { work_id, .. }
| SessionEventKind::RuntimeWorkProgress { work_id, .. }
| SessionEventKind::RuntimeWorkFinished { work_id, .. }) = &event.kind
else {
continue;
};
if spans
.get(work_id)
.and_then(|span| span.status)
.is_some_and(bcode_session_models::RuntimeWorkStatus::is_terminal)
{
continue;
}
match event.kind {
SessionEventKind::RuntimeWorkStarted {
work_id,
label,
parent_work_id,
started_at_ms,
..
} => {
spans.insert(
work_id.clone(),
RuntimeWorkSpan {
work_id,
parent_work_id,
label,
status: None,
started_at_ms,
finished_at_ms: None,
cancelled: false,
message: None,
},
);
}
SessionEventKind::RuntimeWorkCancelRequested { work_id, .. } => {
spans
.entry(work_id.clone())
.or_insert_with(|| runtime_work_span_without_start(work_id))
.cancelled = true;
}
SessionEventKind::RuntimeWorkProgress {
work_id, message, ..
} => {
spans
.entry(work_id.clone())
.or_insert_with(|| runtime_work_span_without_start(work_id))
.message = Some(message);
}
SessionEventKind::RuntimeWorkFinished {
work_id,
status,
finished_at_ms,
message,
} => {
let span = spans
.entry(work_id.clone())
.or_insert_with(|| runtime_work_span_without_start(work_id));
span.status = Some(status);
span.finished_at_ms = finished_at_ms;
if message.is_some() {
span.message = message;
}
}
_ => {}
}
}
spans.into_values().collect()
}
#[cfg(test)]
mod runtime_work_history_tests {
use super::*;
use bcode_session_models::RuntimeWorkStatus;
fn event(kind: SessionEventKind) -> SessionEvent {
SessionEvent {
schema_version: bcode_session_models::CURRENT_SESSION_EVENT_SCHEMA_VERSION,
sequence: 1,
timestamp_ms: 1,
session_id: SessionId::new(),
provenance: None,
kind,
}
}
#[test]
fn span_domain_contract_preserves_json_and_unknown_duration() {
let wire = serde_json::json!({
"work_id": "partial", "parent_work_id": null, "label": "",
"status": "cancelled", "started_at_ms": null, "finished_at_ms": 50,
"cancelled": false, "message": "finished λ",
});
let domain: bcode_session_models::RuntimeWorkSpan =
serde_json::from_value(wire.clone()).unwrap();
let compatible: RuntimeWorkSpan = domain;
assert_eq!(serde_json::to_value(&compatible).unwrap(), wire);
assert_eq!(compatible.duration_ms(), None);
let mut reversed = compatible;
reversed.started_at_ms = Some(60);
assert_eq!(reversed.duration_ms(), Some(0));
reversed.finished_at_ms = None;
assert_eq!(reversed.duration_ms(), None);
}
fn started(work_id: &WorkId, label: &str, started_at_ms: u64) -> SessionEventKind {
SessionEventKind::RuntimeWorkStarted {
work_id: work_id.clone(),
kind: bcode_session_models::RuntimeWorkKind::Tool,
label: label.into(),
tool_call_id: None,
plugin_id: None,
service_interface: None,
operation: None,
parent_work_id: Some(WorkId::new("parent")),
started_at_ms: Some(started_at_ms),
cancellable: true,
}
}
#[test]
fn complete_history_preserves_start_metadata_and_finish_message() {
let work_id = WorkId::new("complete");
let spans = runtime_work_spans(vec![
event(started(&work_id, "inspect", 10)),
event(SessionEventKind::RuntimeWorkProgress {
work_id: work_id.clone(),
message: "running".into(),
progress_at_ms: None,
completed_units: None,
total_units: None,
}),
event(SessionEventKind::RuntimeWorkFinished {
work_id: work_id.clone(),
status: RuntimeWorkStatus::Completed,
finished_at_ms: Some(40),
message: Some("done".into()),
}),
]);
assert_eq!(
spans,
vec![RuntimeWorkSpan {
work_id,
parent_work_id: Some(WorkId::new("parent")),
label: "inspect".into(),
status: Some(RuntimeWorkStatus::Completed),
started_at_ms: Some(10),
finished_at_ms: Some(40),
cancelled: false,
message: Some("done".into()),
}]
);
assert_eq!(spans[0].duration_ms(), Some(30));
}
#[test]
fn terminal_history_cannot_be_reopened_or_overwritten() {
let work_id = WorkId::new("terminal");
for status in [
RuntimeWorkStatus::Completed,
RuntimeWorkStatus::Failed,
RuntimeWorkStatus::TimedOut,
RuntimeWorkStatus::Cancelled,
] {
let terminal = event(SessionEventKind::RuntimeWorkFinished {
work_id: work_id.clone(),
status,
finished_at_ms: Some(40),
message: Some("authoritative".into()),
});
let expected = runtime_work_spans(vec![terminal.clone()]);
let events = vec![
terminal.clone(),
terminal,
event(started(&work_id, "stale start", 50)),
event(SessionEventKind::RuntimeWorkProgress {
work_id: work_id.clone(),
message: "stale progress".into(),
progress_at_ms: None,
completed_units: None,
total_units: None,
}),
event(SessionEventKind::RuntimeWorkCancelRequested {
work_id: work_id.clone(),
requested_at_ms: None,
client_id: None,
}),
event(SessionEventKind::RuntimeWorkFinished {
work_id: work_id.clone(),
status: RuntimeWorkStatus::Suspended,
finished_at_ms: Some(90),
message: Some("stale finish".into()),
}),
];
assert_eq!(runtime_work_spans(events), expected);
}
}
#[test]
fn resumed_work_resets_the_previous_partial_attempt() {
let work_id = WorkId::new("resumed");
let mut events = vec![
event(SessionEventKind::RuntimeWorkCancelRequested {
work_id: work_id.clone(),
requested_at_ms: None,
client_id: None,
}),
event(SessionEventKind::RuntimeWorkFinished {
work_id: work_id.clone(),
status: RuntimeWorkStatus::Suspended,
finished_at_ms: Some(20),
message: Some("suspended".into()),
}),
event(started(&work_id, "resumed attempt", 30)),
];
let spans = runtime_work_spans(events.clone());
assert_eq!(spans.len(), 1);
assert_eq!(spans[0].label, "resumed attempt");
assert_eq!(spans[0].started_at_ms, Some(30));
assert_eq!(spans[0].finished_at_ms, None);
assert_eq!(spans[0].status, None);
assert_eq!(spans[0].message, None);
assert!(!spans[0].cancelled);
events.push(event(SessionEventKind::RuntimeWorkFinished {
work_id,
status: RuntimeWorkStatus::Completed,
finished_at_ms: Some(50),
message: None,
}));
let spans = runtime_work_spans(events);
assert_eq!(spans[0].status, Some(RuntimeWorkStatus::Completed));
assert_eq!(spans[0].duration_ms(), Some(20));
assert_eq!(spans[0].message, None);
assert!(!spans[0].cancelled);
}
#[test]
fn bounded_history_retains_each_lifecycle_event_without_a_start() {
let work_id = WorkId::new("partial");
let kinds = [
SessionEventKind::RuntimeWorkCancelRequested {
work_id: work_id.clone(),
requested_at_ms: Some(20),
client_id: None,
},
SessionEventKind::RuntimeWorkProgress {
work_id: work_id.clone(),
message: "progress".into(),
progress_at_ms: Some(30),
completed_units: None,
total_units: None,
},
SessionEventKind::RuntimeWorkFinished {
work_id: work_id.clone(),
status: RuntimeWorkStatus::Failed,
finished_at_ms: Some(40),
message: None,
},
];
for kind in &kinds {
let spans = runtime_work_spans(vec![event(kind.clone())]);
assert_eq!(spans.len(), 1);
assert_eq!(spans[0].work_id, work_id);
assert!(spans[0].label.is_empty());
assert_eq!(spans[0].parent_work_id, None);
assert_eq!(spans[0].started_at_ms, None);
assert_eq!(spans[0].duration_ms(), None);
}
let spans = runtime_work_spans(kinds.into_iter().map(event).collect());
assert_eq!(spans.len(), 1);
assert!(spans[0].cancelled);
assert_eq!(spans[0].message.as_deref(), Some("progress"));
assert_eq!(spans[0].status, Some(RuntimeWorkStatus::Failed));
assert_eq!(spans[0].finished_at_ms, Some(40));
assert_eq!(spans[0].duration_ms(), None);
}
}
#[derive(Debug, Error)]
pub enum ClientError {
#[error("IPC transport error: {0}")]
Transport(#[from] bcode_ipc::IpcTransportError),
#[error("IPC codec error: {0}")]
Codec(#[from] CodecError),
#[error("daemon start error: {0}")]
DaemonStart(#[from] DaemonStartError),
#[error("server returned error {code}: {message}")]
Server { code: String, message: String },
#[error("daemon connection and handshake timed out after {timeout:?}")]
ConnectTimeout { timeout: Duration },
#[error("daemon startup timed out after {timeout:?}")]
DaemonStartupTimeout { timeout: Duration },
#[error("client request timed out after {timeout:?}")]
RequestTimeout { timeout: Duration },
#[error("incompatible daemon: {message}")]
IncompatibleDaemon { message: String },
#[error("client protocol error: {0}")]
Protocol(String),
#[error("worktree creation failed ({code}): {message}")]
WorktreeCreate {
code: String,
message: String,
created_path: Option<std::path::PathBuf>,
},
#[error("unexpected response payload")]
UnexpectedResponse,
#[error("unexpected IPC envelope kind")]
UnexpectedEnvelope,
}
impl bcode_workflow::WorkflowRunApplication for BcodeClient {
async fn start_workflow_template(
&self,
request: bcode_workflow::WorkflowTemplateStartRequest,
) -> Result<bcode_workflow::WorkflowRunStartResponse, Self::Error> {
Self::start_workflow_template(self, request).await
}
async fn repair_workflow_attempt(
&self,
dispatch_identity: String,
resolution: bcode_workflow::RepairResolution,
) -> Result<bcode_workflow::RepairResult, Self::Error> {
Self::repair_workflow_attempt(self, dispatch_identity, resolution).await
}
async fn doctor_workflow_run(
&self,
run_id: String,
limit: usize,
) -> Result<bcode_workflow::WorkflowDoctorReport, Self::Error> {
Self::doctor_workflow_run(self, run_id, limit).await
}
async fn reconcile_orphaned_workflow_runs(
&self,
apply: bool,
limit: usize,
) -> Result<bcode_workflow::OrphanedWorkflowRunReport, Self::Error> {
Self::reconcile_orphaned_workflow_runs(self, apply, limit).await
}
async fn list_all_workflow_mutation_approvals(
&self,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowMutationApprovalInspection>, Self::Error> {
Self::list_all_workflow_mutation_approvals(self, limit).await
}
async fn list_workflow_mutation_approvals(
&self,
run_id: String,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowMutationApprovalInspection>, Self::Error> {
Self::list_workflow_mutation_approvals(self, run_id, limit).await
}
async fn resolve_workflow_mutation_approval(
&self,
approval_id: String,
decision: bcode_workflow::WorkflowMutationApprovalDecision,
) -> Result<bcode_workflow::WorkflowMutationApprovalResolution, Self::Error> {
Self::resolve_workflow_mutation_approval(self, approval_id, decision).await
}
async fn workflow_attempt_history(
&self,
run_id: String,
cursor: Option<bcode_workflow::AttemptCursor>,
limit: usize,
) -> Result<Vec<bcode_workflow::AttemptSummary>, Self::Error> {
Self::workflow_attempt_history(self, run_id, cursor, limit).await
}
async fn workflow_committed_history(
&self,
query: bcode_workflow::WorkflowCommittedHistoryQuery,
) -> Result<bcode_workflow::WorkflowCommittedHistoryPage, Self::Error> {
Self::workflow_committed_history(self, query).await
}
async fn workflow_lifecycle_history(
&self,
query: bcode_workflow::WorkflowLifecycleHistoryQuery,
) -> Result<bcode_workflow::WorkflowLifecycleHistoryPage, Self::Error> {
Self::workflow_lifecycle_history(self, query).await
}
async fn workflow_inventory(
&self,
query: bcode_workflow::WorkflowInventoryQuery,
) -> Result<bcode_workflow::WorkflowInventoryPage, Self::Error> {
Self::workflow_inventory(self, query).await
}
async fn workflow_semantic_history(
&self,
query: bcode_workflow::WorkflowSemanticHistoryQuery,
) -> Result<bcode_workflow::WorkflowSemanticHistoryPage, Self::Error> {
Self::workflow_semantic_history(self, query).await
}
async fn workflow_observations(
&self,
run_id: String,
after_sequence: Option<u64>,
through_sequence: Option<u64>,
limit: usize,
) -> Result<bcode_workflow::WorkflowObservationPage, Self::Error> {
Self::workflow_observations(self, run_id, after_sequence, through_sequence, limit).await
}
async fn workflow_event_history(
&self,
run_id: String,
after_sequence: Option<u64>,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowHistoryEvent>, Self::Error> {
Self::workflow_event_history(self, run_id, after_sequence, limit).await
}
async fn retry_workflow_node(
&self,
run_id: String,
node_id: String,
activation_id: String,
failed_attempt: u32,
) -> Result<bcode_workflow::WorkflowNodeRetryResult, Self::Error> {
Self::retry_workflow_node(self, run_id, node_id, activation_id, failed_attempt).await
}
async fn list_workflow_waits(
&self,
run_id: String,
limit: usize,
) -> Result<Vec<bcode_workflow::WaitingActivation>, Self::Error> {
Self::list_workflow_waits(self, run_id, limit).await
}
async fn provide_workflow_input(
&self,
run_id: String,
node_id: String,
activation_id: String,
value: serde_json::Value,
) -> Result<bcode_workflow::WaitingResolutionResult, Self::Error> {
Self::provide_workflow_input(self, run_id, node_id, activation_id, value).await
}
async fn resolve_workflow_approval(
&self,
run_id: String,
node_id: String,
activation_id: String,
approved: bool,
) -> Result<bcode_workflow::WaitingResolutionResult, Self::Error> {
Self::resolve_workflow_approval(self, run_id, node_id, activation_id, approved).await
}
async fn workflow_continuation_source(
&self,
run_id: String,
) -> Result<bcode_workflow::WorkflowContinuationSource, Self::Error> {
Self::workflow_continuation_source(self, run_id).await
}
async fn continue_workflow(
&self,
request: bcode_workflow::WorkflowContinuationRequest,
) -> Result<bcode_workflow::WorkflowRunStartResponse, Self::Error> {
Self::continue_workflow(self, request).await
}
async fn request_workflow_replacement(
&self,
request: bcode_workflow::WorkflowReplacementRequest,
) -> Result<bcode_workflow::WorkflowReplacementResponse, Self::Error> {
Self::request_workflow_replacement(self, request).await
}
async fn start_workflow_run(
&self,
request: bcode_workflow::WorkflowRunStartRequest,
) -> Result<bcode_workflow::WorkflowRunStartResponse, Self::Error> {
Self::start_workflow_run(self, request).await
}
async fn start_workflow(
&self,
request: bcode_workflow::WorkflowStartRequest,
) -> Result<bcode_workflow::WorkflowRunStartResponse, Self::Error> {
Self::start_workflow(self, request).await
}
async fn workflow_live_event_catch_up(
&self,
after_sequence: u64,
limit: usize,
) -> Result<bcode_workflow_view_models::WorkflowLiveEventPage, Self::Error> {
match self
.send_request(Request::WorkflowLiveEventCatchUp {
after_sequence,
limit,
})
.await?
{
ResponsePayload::WorkflowLiveEventCatchUp { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
async fn associated_workflow_run(
&self,
key: bcode_workflow::WorkflowRunBindingLookup,
) -> Result<Option<bcode_workflow::WorkflowRunSummary>, Self::Error> {
Self::associated_workflow_run(self, key).await
}
async fn inspect_associated_workflow_run(
&self,
key: bcode_workflow::WorkflowRunBindingLookup,
limit: usize,
) -> Result<Option<bcode_workflow::WorkflowRunInspection>, Self::Error> {
Self::inspect_associated_workflow_run(self, key, limit).await
}
async fn control_workflow_run(
&self,
run_id: String,
action: bcode_workflow::WorkflowRunControlAction,
) -> Result<(Option<bcode_workflow::WorkflowRunSummary>, bool), Self::Error> {
Self::control_workflow_run(self, run_id, action).await
}
async fn control_associated_workflow_run(
&self,
key: bcode_workflow::WorkflowRunBindingLookup,
action: bcode_workflow::WorkflowRunControlAction,
) -> Result<(Option<bcode_workflow::WorkflowRunSummary>, bool), Self::Error> {
Self::control_associated_workflow_run(self, key, action).await
}
async fn inspect_workflow_run_graph(
&self,
request: bcode_workflow::WorkflowRunGraphPageRequest,
) -> Result<bcode_workflow::WorkflowRunGraphInspection, Self::Error> {
Self::inspect_workflow_run_graph(self, request).await
}
async fn list_workflow_runs(
&self,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowRunSummary>, Self::Error> {
Self::list_workflow_runs(self, limit).await
}
async fn workflow_run_outputs(
&self,
run_id: String,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowOutputInspection>, Self::Error> {
Self::workflow_run_outputs(self, run_id, limit).await
}
async fn workflow_run_status(
&self,
run_id: String,
) -> Result<Option<bcode_workflow::WorkflowRunSummary>, Self::Error> {
Self::workflow_run_status(self, run_id).await
}
async fn pause_workflow_run(&self, run_id: String) -> Result<bool, Self::Error> {
Self::pause_workflow_run(self, run_id).await
}
async fn resume_workflow_run(&self, run_id: String) -> Result<bool, Self::Error> {
Self::resume_workflow_run(self, run_id).await
}
async fn cancel_workflow_run(&self, run_id: String) -> Result<bool, Self::Error> {
Self::cancel_workflow_run(self, run_id).await
}
async fn inspect_workflow_run(
&self,
run_id: String,
limit: usize,
) -> Result<bcode_workflow::WorkflowRunInspection, Self::Error> {
Self::inspect_workflow_run(self, run_id, limit).await
}
type Error = ClientError;
async fn start_authored_workflow(
&self,
request: bcode_workflow::StartAuthoredWorkflowRequest,
) -> Result<bcode_workflow::AuthoredWorkflowRunStartResponse, Self::Error> {
Self::start_authored_workflow(self, request).await
}
}
impl bcode_workflow::WorkflowAuthoringApplication for BcodeClient {
async fn instantiate_workflow_template(
&self,
request: bcode_workflow::WorkflowTemplateInstantiationRequest,
) -> Result<
(
bcode_workflow::AuthoredWorkflowSnapshot,
bcode_workflow::WorkflowDraftSnapshot,
),
Self::Error,
> {
Self::instantiate_workflow_template(self, request).await
}
async fn inspect_workflow_definition(
&self,
definition_id: String,
version: u32,
) -> Result<Option<bcode_workflow::WorkflowDefinitionInspection>, Self::Error> {
let stored =
Self::describe_workflow_definition(self, definition_id.clone(), version).await?;
if stored
.as_ref()
.is_some_and(|value| value.definition_id != definition_id || value.version != version)
{
let failure = bcode_workflow::WorkflowAuthoringFailure::StateUnavailable;
return Err(ClientError::Server {
code: failure.code().to_string(),
message: failure.to_string(),
});
}
stored
.map(bcode_workflow::WorkflowDefinitionInspection::try_from)
.transpose()
.map_err(|failure| ClientError::Server {
code: failure.code().to_string(),
message: failure.to_string(),
})
}
async fn register_workflow_definition(
&self,
request: bcode_workflow::WorkflowDefinitionRegistrationRequest,
) -> Result<bcode_workflow::StoredWorkflowDefinition, Self::Error> {
Self::register_workflow_definition(self, request).await
}
async fn list_workflow_definitions(
&self,
limit: usize,
) -> Result<Vec<bcode_workflow::StoredWorkflowDefinition>, Self::Error> {
Self::list_workflow_definitions(self, limit).await
}
async fn describe_workflow_definition(
&self,
definition_id: String,
version: u32,
) -> Result<Option<bcode_workflow::StoredWorkflowDefinition>, Self::Error> {
Self::describe_workflow_definition(self, definition_id, version).await
}
async fn publish_workflow_package(
&self,
request: bcode_workflow::PublishWorkflowPackageRequest,
) -> Result<bcode_workflow::WorkflowPackageMutationResult, Self::Error> {
Self::publish_workflow_package(self, request).await
}
async fn apply_workflow_package(
&self,
request: bcode_workflow::ApplyWorkflowPackageRequest,
) -> Result<bcode_workflow::WorkflowPackageMutationResult, Self::Error> {
Self::apply_workflow_package(self, request).await
}
async fn inspect_workflow_templates(
&self,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowTemplateInspection>, Self::Error> {
match self
.send_request(Request::InspectWorkflowTemplates { limit })
.await?
{
ResponsePayload::WorkflowTemplateInspections { templates } => Ok(templates),
_ => Err(ClientError::UnexpectedResponse),
}
}
async fn inspect_workflow_template(
&self,
owner_plugin_id: String,
template_id: String,
template_version: u32,
) -> Result<Option<bcode_workflow::WorkflowTemplateInspection>, Self::Error> {
match self
.send_request(Request::InspectWorkflowTemplate {
owner_plugin_id,
template_id,
template_version,
})
.await?
{
ResponsePayload::WorkflowTemplateInspection { template } => {
Ok(template.map(|template| *template))
}
_ => Err(ClientError::UnexpectedResponse),
}
}
async fn workflow_package_publication(
&self,
package_id: String,
) -> Result<Option<bcode_workflow::WorkflowPackagePublicationReceipt>, Self::Error> {
Self::workflow_package_publication(self, package_id).await
}
async fn cancel_workflow_discovery(&self, token: String) -> Result<bool, Self::Error> {
match self
.send_request(Request::CancelWorkflowDiscovery { token })
.await?
{
ResponsePayload::WorkflowDiscoveryCancelled { released } => Ok(released),
_ => Err(ClientError::UnexpectedResponse),
}
}
async fn workflow_launch_catalog(
&self,
request: bcode_workflow::WorkflowLaunchCatalogRequest,
) -> Result<bcode_workflow::WorkflowLaunchCatalogPage, Self::Error> {
Self::workflow_launch_catalog(self, request).await
}
async fn workflow_launch_detail(
&self,
request: bcode_workflow::WorkflowLaunchDetailRequest,
) -> Result<bcode_workflow::WorkflowLaunchDetail, Self::Error> {
Self::workflow_launch_detail(self, request).await
}
async fn validate_workflow_authoring_with_control(
&self,
document: bcode_workflow::WorkflowAuthoringDocument,
control: bcode_workflow::WorkflowComputationControl,
) -> Result<bcode_workflow::WorkflowValidationReport, Self::Error> {
Self::validate_workflow_authoring_with_control(self, document, control).await
}
async fn preview_workflow_compilation_with_control(
&self,
document: bcode_workflow::WorkflowAuthoringDocument,
configuration: Option<serde_json::Value>,
control: bcode_workflow::WorkflowComputationControl,
) -> Result<bcode_workflow::WorkflowCompilationPreview, Self::Error> {
Self::preview_workflow_compilation_with_control(self, document, configuration, control)
.await
}
async fn validate_workflow_package(
&self,
request: bcode_workflow::WorkflowPackageComputationRequest,
) -> Result<bcode_workflow::WorkflowPackageValidationResult, Self::Error> {
Self::validate_workflow_package(self, request).await
}
async fn preview_workflow_package(
&self,
request: bcode_workflow::WorkflowPackagePreviewRequest,
) -> Result<bcode_workflow::WorkflowPackagePreview, Self::Error> {
Self::preview_workflow_package(self, request).await
}
async fn validate_workflow_source(
&self,
request: bcode_workflow::WorkflowSourceComputationRequest,
) -> Result<bcode_workflow::WorkflowSourceValidationResult, Self::Error> {
Self::validate_workflow_source(self, request).await
}
async fn preview_workflow_source(
&self,
request: bcode_workflow::WorkflowSourcePreviewRequest,
) -> Result<bcode_workflow::WorkflowSourcePreviewResult, Self::Error> {
Self::preview_workflow_source(self, request).await
}
async fn workflow_revision_requirement_inspection(
&self,
workflow_id: String,
revision: u64,
) -> Result<Option<bcode_workflow::WorkflowRevisionRequirementInspection>, Self::Error> {
Self::workflow_revision_requirement_inspection(self, workflow_id, revision).await
}
async fn workflow_authoring_catalog(
&self,
) -> Result<bcode_workflow::WorkflowAuthoringCatalogSnapshot, Self::Error> {
Self::workflow_authoring_catalog(self).await
}
async fn list_workflow_presets(
&self,
workflow_id: String,
cursor: Option<bcode_workflow::WorkflowAuthoringListCursor>,
limit: usize,
) -> Result<
bcode_workflow::WorkflowAuthoringPage<
bcode_workflow::WorkflowPresetSnapshot,
bcode_workflow::WorkflowAuthoringListCursor,
>,
Self::Error,
> {
Self::list_workflow_presets(self, workflow_id, cursor, limit).await
}
async fn workflow_preset(
&self,
workflow_id: String,
preset_id: String,
) -> Result<Option<bcode_workflow::WorkflowPresetSnapshot>, Self::Error> {
Self::workflow_preset(self, workflow_id, preset_id).await
}
async fn list_workflow_revisions(
&self,
workflow_id: String,
cursor: Option<bcode_workflow::WorkflowRevisionListCursor>,
limit: usize,
) -> Result<
bcode_workflow::WorkflowAuthoringPage<
bcode_workflow::WorkflowRevisionSnapshot,
bcode_workflow::WorkflowRevisionListCursor,
>,
Self::Error,
> {
Self::list_workflow_revisions(self, workflow_id, cursor, limit).await
}
async fn workflow_revision(
&self,
workflow_id: String,
revision: u64,
) -> Result<Option<bcode_workflow::WorkflowRevisionSnapshot>, Self::Error> {
Self::workflow_revision(self, workflow_id, revision).await
}
async fn list_workflow_drafts(
&self,
workflow_id: String,
cursor: Option<bcode_workflow::WorkflowAuthoringListCursor>,
limit: usize,
) -> Result<
bcode_workflow::WorkflowAuthoringPage<
bcode_workflow::WorkflowDraftSnapshot,
bcode_workflow::WorkflowAuthoringListCursor,
>,
Self::Error,
> {
Self::list_workflow_drafts(self, workflow_id, cursor, limit).await
}
async fn workflow_draft(
&self,
workflow_id: String,
draft_id: String,
) -> Result<Option<bcode_workflow::WorkflowDraftSnapshot>, Self::Error> {
Self::workflow_draft(self, workflow_id, draft_id).await
}
async fn list_authored_workflows(
&self,
cursor: Option<bcode_workflow::WorkflowAuthoringListCursor>,
limit: usize,
) -> Result<
bcode_workflow::WorkflowAuthoringPage<
bcode_workflow::AuthoredWorkflowSnapshot,
bcode_workflow::WorkflowAuthoringListCursor,
>,
Self::Error,
> {
Self::list_authored_workflows(self, cursor, limit).await
}
async fn authored_workflow(
&self,
workflow_id: String,
) -> Result<Option<bcode_workflow::AuthoredWorkflowSnapshot>, Self::Error> {
Self::authored_workflow(self, workflow_id).await
}
async fn inspect_authored_workflow(
&self,
workflow_id: String,
limit: usize,
) -> Result<Option<bcode_workflow::AuthoredWorkflowInspection>, Self::Error> {
Self::inspect_authored_workflow(self, workflow_id, limit).await
}
async fn import_workflow_draft(
&self,
request: bcode_workflow::ImportWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowDraftImportResult, Self::Error> {
Self::import_workflow_draft(self, request).await
}
async fn import_workflow_revision(
&self,
request: bcode_workflow::ImportWorkflowRevisionRequest,
) -> Result<bcode_workflow::WorkflowRevisionImportResult, Self::Error> {
Self::import_workflow_revision(self, request).await
}
async fn import_workflow(
&self,
request: bcode_workflow::ImportWorkflowRequest,
) -> Result<
(
bcode_workflow::AuthoredWorkflowSnapshot,
bcode_workflow::WorkflowDraftSnapshot,
),
Self::Error,
> {
Self::import_workflow(self, request).await
}
async fn preview_workflow_import(
&self,
request: bcode_workflow::PreviewWorkflowImportRequest,
) -> Result<bcode_workflow::WorkflowImportPreview, Self::Error> {
Self::preview_workflow_import(self, request).await
}
async fn export_workflow_revision(
&self,
request: bcode_workflow::ExportWorkflowRevisionRequest,
) -> Result<bcode_workflow::WorkflowExportBundle, Self::Error> {
Self::export_workflow_revision(self, request).await
}
async fn create_workflow_preset(
&self,
request: bcode_workflow::CreateWorkflowPresetRequest,
) -> Result<bcode_workflow::WorkflowPresetSnapshot, Self::Error> {
Self::create_workflow_preset(self, request).await
}
async fn update_workflow_preset(
&self,
request: bcode_workflow::UpdateWorkflowPresetRequest,
) -> Result<bcode_workflow::WorkflowPresetUpdateResult, Self::Error> {
Self::update_workflow_preset(self, request).await
}
async fn delete_workflow_preset(
&self,
request: bcode_workflow::DeleteWorkflowPresetRequest,
) -> Result<bcode_workflow::WorkflowAuthoringMutationResult, Self::Error> {
Self::delete_workflow_preset(self, request).await
}
async fn fork_workflow_draft(
&self,
request: bcode_workflow::ForkWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowDraftSnapshot, Self::Error> {
Self::fork_workflow_draft(self, request).await
}
async fn apply_workflow_source(
&self,
request: bcode_workflow::ApplyWorkflowSourceRequest,
) -> Result<bcode_workflow::WorkflowSourceApplyResult, Self::Error> {
Self::apply_workflow_source(
self,
request.source_format,
request.source,
request.draft_id,
)
.await
}
async fn set_authored_workflow_archived(
&self,
request: bcode_workflow::SetAuthoredWorkflowArchivedRequest,
) -> Result<bcode_workflow::AuthoredWorkflowSnapshot, Self::Error> {
Self::set_authored_workflow_archived(self, request).await
}
async fn cancel_workflow_computation(&self, operation_id: String) -> Result<bool, Self::Error> {
Self::cancel_workflow_computation(self, operation_id).await
}
async fn create_authored_workflow(
&self,
request: bcode_workflow::CreateAuthoredWorkflowRequest,
) -> Result<
(
bcode_workflow::AuthoredWorkflowSnapshot,
bcode_workflow::WorkflowDraftSnapshot,
),
Self::Error,
> {
Self::create_authored_workflow(self, request).await
}
async fn publish_and_start_workflow(
&self,
request: bcode_workflow::PublishAndStartWorkflowRequest,
) -> Result<bcode_workflow::WorkflowPublishAndStartResult, Self::Error> {
Self::publish_and_start_workflow(self, request).await
}
async fn publish_workflow_draft(
&self,
request: bcode_workflow::PublishWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowPublicationResult, Self::Error> {
Self::publish_workflow_draft(self, request).await
}
type Error = ClientError;
async fn apply_workflow_draft_edits(
&self,
request: bcode_workflow::ApplyWorkflowDraftEditsRequest,
) -> Result<bcode_workflow::WorkflowDraftEditResult, Self::Error> {
Self::apply_workflow_draft_edits(self, request).await
}
async fn update_workflow_draft(
&self,
request: bcode_workflow::UpdateWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowDraftUpdateResult, Self::Error> {
Self::update_workflow_draft(self, request).await
}
async fn activate_workflow_revision(
&self,
request: bcode_workflow::ActivateWorkflowRevisionRequest,
) -> Result<bcode_workflow::WorkflowAuthoringMutationResult, Self::Error> {
Self::activate_workflow_revision(self, request).await
}
async fn discard_workflow_draft(
&self,
request: bcode_workflow::DiscardWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowAuthoringMutationResult, Self::Error> {
Self::discard_workflow_draft(self, request).await
}
}
impl ClientError {
#[allow(clippy::must_use_candidate)]
pub fn auth_pool_error(&self) -> Option<bcode_provider_auth_models::AuthPoolOperationError> {
match self {
Self::Server { code, .. } => {
bcode_provider_auth_models::AuthPoolOperationError::from_code(code)
}
_ => None,
}
}
#[must_use]
pub fn is_optional_domain_unavailable(&self) -> bool {
matches!(
self,
Self::Server { code, .. } if code == "workflow_capability_unavailable"
)
}
#[must_use]
pub fn is_daemon_unavailable(&self) -> bool {
match self {
Self::Transport(bcode_ipc::IpcTransportError::Io(error)) => matches!(
error.kind(),
std::io::ErrorKind::ConnectionRefused
| std::io::ErrorKind::ConnectionReset
| std::io::ErrorKind::NotFound
| std::io::ErrorKind::BrokenPipe
| std::io::ErrorKind::UnexpectedEof
),
Self::Codec(CodecError::Io(error)) => matches!(
error.kind(),
std::io::ErrorKind::ConnectionRefused
| std::io::ErrorKind::ConnectionReset
| std::io::ErrorKind::NotFound
| std::io::ErrorKind::BrokenPipe
| std::io::ErrorKind::UnexpectedEof
),
Self::DaemonStartupTimeout { .. } | Self::DaemonStart(_) => true,
Self::ConnectTimeout { .. }
| Self::RequestTimeout { .. }
| Self::Transport(_)
| Self::Codec(_)
| Self::Server { .. }
| Self::WorktreeCreate { .. }
| Self::IncompatibleDaemon { .. }
| Self::Protocol(_)
| Self::UnexpectedResponse
| Self::UnexpectedEnvelope => false,
}
}
}
pub struct SessionOpenProgressObserver {
pub receiver:
tokio::sync::mpsc::UnboundedReceiver<bcode_session_models::SessionOpenOperationSnapshot>,
pub task: tokio::task::JoinHandle<Result<(), ClientError>>,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct SessionList {
pub sessions: Vec<SessionSummary>,
pub catalog_status: SessionCatalogStatus,
pub catalog_sources: Vec<SessionCatalogSourceStatus>,
pub catalog_revision: u64,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct AttachedSessionHistory {
pub submissions: Vec<bcode_session_models::SubmissionRecord>,
pub session: SessionSummary,
pub history: Vec<SessionEvent>,
pub input_history: Vec<SessionInputHistoryEntry>,
pub usage_summary: bcode_session_models::SessionUsageSummary,
pub import_warnings: Vec<SessionImportWarning>,
pub draft: Option<String>,
pub runtime_selection: bcode_session_models::SessionRuntimeSelection,
pub projection_window: Option<bcode_session_models::ProjectionWindow>,
}
const CLIENT_RUNTIME_ENV_VARS: &[&str] = &[
"BCODE_OPENAI_API_KEY",
"OPENAI_API_KEY",
"BCODE_OPENAI_AUTH_MODE",
"BCODE_OPENAI_AUTH_PROFILE",
"BCODE_OPENAI_AUTH_VAULT",
"BCODE_OPENAI_BASE_URL",
"OPENAI_BASE_URL",
"BCODE_OPENAI_MODEL",
"OPENAI_MODEL",
"BCODE_OPENAI_MODELS",
"OPENAI_MODELS",
"BCODE_OPENAI_DIALECT",
"OPENAI_DIALECT",
"BCODE_OPENAI_CODEX_ACCESS_TOKEN",
"BCODE_OPENAI_CODEX_REFRESH_TOKEN",
"BCODE_OPENAI_CODEX_ID_TOKEN",
"BCODE_OPENAI_CODEX_EXPIRES_AT",
"BCODE_OPENAI_CODEX_ACCOUNT_ID",
"BCODE_XAI_AUTH_MODE",
"BCODE_XAI_AUTH_PROFILE",
"BCODE_XAI_AUTH_VAULT",
"BCODE_XAI_API_KEY",
"XAI_API_KEY",
"BCODE_XAI_BASE_URL",
"XAI_BASE_URL",
"BCODE_XAI_MODEL",
"XAI_MODEL",
"BCODE_XAI_MODELS",
"XAI_MODELS",
"BCODE_BEDROCK_MODEL",
"BEDROCK_MODEL",
"BCODE_BEDROCK_MODELS",
"BEDROCK_MODELS",
"BCODE_BEDROCK_REGION",
"BEDROCK_REGION",
"BCODE_BEDROCK_AWS_PROFILE",
"AWS_PROFILE",
"AWS_REGION",
"AWS_DEFAULT_REGION",
"BCODE_BEDROCK_ENDPOINT_URL",
"BEDROCK_ENDPOINT_URL",
"AWS_ENDPOINT_URL_BEDROCK",
"BCODE_BEDROCK_TRANSPORT",
"BCODE_BEDROCK_MANTLE_BASE_URL",
"BCODE_BEDROCK_MANTLE_AUTH_HEADER",
"BCODE_BEDROCK_FORCE_HTTP1",
"AWS_BEDROCK_FORCE_HTTP1",
"AWS_ACCESS_KEY_ID",
"AWS_SECRET_ACCESS_KEY",
"AWS_SESSION_TOKEN",
"AWS_BEARER_TOKEN_BEDROCK",
];
fn resolve_caller_path(path: Option<std::path::PathBuf>) -> std::path::PathBuf {
resolve_path_from(path, ¤t_working_directory())
}
fn resolve_path_from(
path: Option<std::path::PathBuf>,
caller_cwd: &std::path::Path,
) -> std::path::PathBuf {
let path = path.map_or_else(
|| caller_cwd.to_path_buf(),
|path| {
if path.is_absolute() {
path
} else {
caller_cwd.join(path)
}
},
);
path.canonicalize().unwrap_or(path)
}
fn caller_provider_environment(
selected_environment: Option<&str>,
mut read: impl FnMut(&str) -> Option<String>,
) -> BTreeMap<String, String> {
if selected_environment.is_some() {
return BTreeMap::new();
}
CLIENT_RUNTIME_ENV_VARS
.iter()
.filter_map(|name| {
read(name)
.filter(|value| !value.trim().is_empty())
.map(|value| ((*name).to_owned(), value))
})
.collect()
}
#[cfg(test)]
mod startup_environment_tests {
use super::caller_provider_environment;
#[test]
fn selected_account_never_reads_or_transmits_other_process_accounts() {
let env = caller_provider_environment(Some("OPENROUTER_API_KEY"), |_| {
panic!("selected normalized authentication must not inspect process overrides")
});
assert!(env.is_empty());
}
#[test]
fn explicit_configuration_retains_existing_process_environment_behavior() {
let env = caller_provider_environment(None, |name| match name {
"OPENAI_API_KEY" => Some("fake-test-key".to_owned()),
"BCODE_OPENAI_BASE_URL" => Some("https://example.invalid/v1".to_owned()),
"AWS_SESSION_TOKEN" => Some(" ".to_owned()),
_ => None,
});
assert_eq!(
env.get("OPENAI_API_KEY").map(String::as_str),
Some("fake-test-key")
);
assert_eq!(
env.get("BCODE_OPENAI_BASE_URL").map(String::as_str),
Some("https://example.invalid/v1")
);
assert!(!env.contains_key("AWS_SESSION_TOKEN"));
}
}
fn current_runtime_context() -> Result<ClientRuntimeContext, ClientError> {
let working_directory = current_working_directory();
let (config, launch) = bcode_config::recovery::load_current_config_with_recovery_context()
.map_err(|_| {
ClientError::Protocol(
"Client configuration could not be loaded; daemon defaults were not substituted."
.to_owned(),
)
})?;
let effective_config_toml =
Some(bcode_config::encode_effective_config(&config).map_err(|_| {
ClientError::Protocol(
"Client configuration could not be encoded; daemon defaults were not substituted."
.to_owned(),
)
})?);
let resolved = config.resolved_model_selection();
let selected_environment = bcode_provider_auth::startup::selected_environment_for_selection(
&config,
&resolved,
)
.map_err(|_| ClientError::Protocol(
"The remembered environment connection is inconsistent; reconnect that account. No substitute account was selected.".to_owned(),
))?;
let env = caller_provider_environment(
selected_environment
.as_ref()
.map(|source| source.name.as_str()),
|name| std::env::var(name).ok(),
);
let provider_context = bcode_provider_auth::try_resolve_provider_request_context(
bcode_provider_auth::ProviderRequestContextResolution {
config: &config,
selection: resolved.clone(),
},
).map_err(|_| ClientError::Protocol(
"Authentication selection could not be verified. Inspect the selected account, pool membership, and authentication metadata before connecting. No substitute account was selected.".to_owned(),
))?;
Ok(runtime_context_from_selection(
working_directory,
effective_config_toml,
resolved,
provider_context,
env,
Some(launch),
))
}
fn runtime_context_from_selection(
working_directory: std::path::PathBuf,
effective_config_toml: Option<String>,
resolved: bcode_config::ResolvedModelSelection,
mut provider_context: bcode_model::ProviderRequestContext,
process_env: BTreeMap<String, String>,
launch: Option<bcode_config::ConfigLaunchContext>,
) -> ClientRuntimeContext {
provider_context.env.extend(process_env.clone());
let env_keys = provider_context
.env
.keys()
.cloned()
.map(|key| (key, true))
.collect();
ClientRuntimeContext {
config_launch: launch.map(Box::new),
working_directory: Some(working_directory),
effective_config_toml: effective_config_toml.map(Box::new),
selected_provider_plugin_id: resolved.provider_plugin_id,
selected_model_id: resolved.model_id,
requested_model_id: resolved.selected_model_id,
provider_context,
process_env,
interaction_adapters: Vec::new(),
env_keys,
}
}
#[cfg(test)]
mod runtime_context_auth_tests {
use super::*;
#[test]
fn provider_validation_decode_preserves_negative_result_and_redacts_errors() {
let result = decode_provider_validation(&PluginServiceResponse {
payload: br#"{"valid":false}"#.to_vec(),
error: None,
})
.unwrap();
assert!(!result.valid);
for response in [
PluginServiceResponse {
payload: b"secret-token-invalid-json".to_vec(),
error: None,
},
PluginServiceResponse {
payload: Vec::new(),
error: Some(bcode_ipc::PluginServiceError {
code: "secret-token".to_owned(),
message: "secret-token".to_owned(),
}),
},
] {
let error = decode_provider_validation(&response).unwrap_err();
assert!(!error.to_string().contains("secret-token"));
}
}
#[test]
fn single_account_selection_does_not_infer_a_pool_from_chatgpt_scheme() {
let mut config = bcode_config::BcodeConfig::default();
config.model.auth_profile = Some("custom-account".to_owned());
config.auth.profiles.insert(
"custom-account".to_owned(),
bcode_config::AuthProfileConfig {
backend: "env".to_owned(),
scheme: Some("chatgpt".to_owned()),
settings: BTreeMap::from([("provider".to_owned(), "openai".to_owned())]),
..Default::default()
},
);
let context = bcode_provider_auth::resolve_provider_request_context_with_resolver(
bcode_provider_auth::ProviderRequestContextResolution {
config: &config,
selection: config.resolved_model_selection(),
},
&bcode_config::RuntimeAuthSubscriptions::default(),
|_, _| bcode_provider_auth::ResolvedProviderAuth::default(),
);
assert_eq!(context.auth_profile.as_deref(), Some("custom-account"));
assert!(context.auth_pool.is_none());
assert!(context.auth_candidates.is_empty());
}
#[tokio::test]
async fn invalid_runtime_context_blocks_connection_before_transport() {
let mut client = BcodeClient::new(default_endpoint());
client.runtime_context_error = Some("Configuration unavailable".to_owned());
assert!(
matches!(client.connect("test").await, Err(ClientError::Protocol(message)) if message == "Configuration unavailable")
);
let explicit = client.with_runtime_context(Some(ClientRuntimeContext::default()));
assert!(explicit.runtime_context_error.is_none());
}
#[test]
fn selected_account_and_credentials_survive_client_adaptation_together() {
let auth = bcode_model::ProviderAuthContext {
profile: Some("preferred-account".to_owned()),
credentials: BTreeMap::from([(
"token".to_owned(),
bcode_model::ProviderAuthCredential {
value: "preferred-token".to_owned(),
source: None,
},
)]),
..Default::default()
};
let provider_context = bcode_model::ProviderRequestContext {
auth_profile: Some("preferred-account".to_owned()),
auth: Some(auth.clone()),
env: BTreeMap::from([("SELECTED".to_owned(), "selected-value".to_owned())]),
auth_candidates: vec![bcode_model::ProviderAuthCandidate {
profile: Some("other-account".to_owned()),
auth: bcode_model::ProviderAuthContext::default(),
env: BTreeMap::from([("OTHER_TOKEN".to_owned(), "other-value".to_owned())]),
}],
..Default::default()
};
let process_env = BTreeMap::from([("SHELL_SETTING".to_owned(), "shell-value".to_owned())]);
let context = runtime_context_from_selection(
".".into(),
None,
bcode_config::ResolvedModelSelection {
auth_profile: Some("original-primary".to_owned()),
..Default::default()
},
provider_context,
process_env.clone(),
None,
);
assert_eq!(
context.provider_context.auth_profile.as_deref(),
Some("preferred-account")
);
assert_eq!(context.provider_context.auth, Some(auth));
assert!(!context.provider_context.env.contains_key("OTHER_TOKEN"));
assert!(!context.env_keys.contains_key("OTHER_TOKEN"));
assert_eq!(context.process_env, process_env);
assert!(!context.process_env.contains_key("SELECTED"));
}
#[test]
fn explicit_shell_values_keep_precedence_without_changing_account_identity() {
let context = runtime_context_from_selection(
".".into(),
None,
bcode_config::ResolvedModelSelection::default(),
bcode_model::ProviderRequestContext {
auth_profile: Some("account".to_owned()),
env: BTreeMap::from([("SETTING".to_owned(), "profile-value".to_owned())]),
..Default::default()
},
BTreeMap::from([("SETTING".to_owned(), "shell-value".to_owned())]),
None,
);
assert_eq!(context.provider_context.env["SETTING"], "shell-value");
assert_eq!(
context.provider_context.auth_profile.as_deref(),
Some("account")
);
}
}
fn decode_provider_validation(
response: &PluginServiceResponse,
) -> Result<bcode_model::ValidateConfigResponse, ClientError> {
if response.error.is_some() {
return Err(ClientError::Protocol(
"Provider configuration validation failed.".to_owned(),
));
}
serde_json::from_slice(&response.payload).map_err(|_| {
ClientError::Protocol(
"Provider returned an incompatible configuration validation response.".to_owned(),
)
})
}
impl From<ErrorResponse> for ClientError {
fn from(value: ErrorResponse) -> Self {
Self::Server {
code: value.code,
message: value.message,
}
}
}
pub use bcode_session_models::{
MessageAcceptance, SessionCatalogCursor, SessionCatalogPage, SessionCatalogQuery,
SessionDiscoveryScope,
};
const fn decode_message_acceptance(
response: &ResponsePayload,
) -> Result<MessageAcceptance, ClientError> {
match response {
ResponsePayload::MessageAccepted {
queued,
queue_position,
} => Ok(MessageAcceptance {
queued: *queued,
queue_position: *queue_position,
disposition: bcode_session_models::MessageAcceptanceDisposition::StartedTurn,
}),
ResponsePayload::MessageAcceptedWithDisposition {
queued,
queue_position,
disposition,
} => Ok(MessageAcceptance {
queued: *queued,
queue_position: *queue_position,
disposition: *disposition,
}),
ResponsePayload::MessageSent => Ok(MessageAcceptance::sent()),
_ => Err(ClientError::UnexpectedResponse),
}
}
fn normalize_discovery_scope(
scope: SessionDiscoveryScope,
) -> Result<SessionDiscoveryScope, ClientError> {
match scope {
SessionDiscoveryScope::WorkingDirectory(path) => std::path::absolute(path)
.map(SessionDiscoveryScope::WorkingDirectory)
.map_err(|_| ClientError::Protocol("cannot resolve catalog working directory".into())),
SessionDiscoveryScope::AllWorkingDirectories => Ok(scope),
}
}
fn decode_scoped_session_list(response: ResponsePayload) -> Result<SessionList, ClientError> {
match response {
ResponsePayload::SessionList {
sessions,
catalog_status,
catalog_sources,
catalog_revision,
} => Ok(SessionList {
sessions,
catalog_status,
catalog_sources,
catalog_revision,
}),
_ => Err(ClientError::UnexpectedResponse),
}
}
#[derive(Debug, Clone)]
pub struct BcodeClient {
expected_state_location: Option<bcode_config::StateLocationId>,
endpoint: IpcEndpoint,
runtime_context: Option<ClientRuntimeContext>,
runtime_context_error: Option<String>,
daemon_availability: DaemonAvailability,
connect_timeout: Duration,
startup_timeout: Duration,
startup_gate: Arc<tokio::sync::Mutex<()>>,
contextual_invocations: contextual::Controls,
request_timeout: Duration,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum DaemonAvailability {
RequireRunning,
AutoStart,
}
#[derive(Debug)]
pub struct SessionCatalogWatcher {
connection: ClientConnection,
last_revision: u64,
scope: SessionDiscoveryScope,
}
impl SessionCatalogWatcher {
pub async fn initial_snapshot(&mut self) -> Result<SessionList, ClientError> {
let snapshot = self
.connection
.list_sessions_with_scope(self.scope.clone())
.await?;
self.last_revision = snapshot.catalog_revision;
Ok(snapshot)
}
pub async fn next_revision(&mut self) -> Result<u64, ClientError> {
loop {
if let Event::SessionCatalogUpdated { revision } = self.connection.recv_event().await?
&& revision > self.last_revision
{
self.last_revision = revision;
return Ok(revision);
}
}
}
pub async fn next_snapshot(&mut self) -> Result<SessionList, ClientError> {
loop {
match self.connection.recv_event().await? {
Event::SessionCatalogUpdated { revision } if revision > self.last_revision => {
let snapshot = self
.connection
.list_sessions_with_scope(self.scope.clone())
.await?;
self.last_revision = snapshot.catalog_revision.max(revision);
return Ok(snapshot);
}
Event::SessionCatalogUpdated { .. }
| Event::Session(_)
| Event::SessionLive(_)
| Event::RuntimeWork(_)
| Event::Workflow(_)
| Event::SessionViewResyncRequired { .. } => {}
}
}
}
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum SessionWatchEvent {
Durable(Box<SessionEvent>),
Live(Box<bcode_session_models::SessionLiveEvent>),
ResyncRequired,
}
#[derive(Debug)]
pub struct SessionWatcher {
connection: ClientConnection,
session_id: SessionId,
initial: Option<AttachedSessionHistory>,
}
impl SessionWatcher {
const fn initial_session_id(&self) -> SessionId {
self.session_id
}
#[must_use]
pub const fn take_initial(&mut self) -> Option<AttachedSessionHistory> {
self.initial.take()
}
pub async fn wait_for_turn_outcome(
&mut self,
turn_id: &str,
timeout: Duration,
) -> Result<bcode_session_models::ModelTurnOutcome, ClientError> {
if let Some(initial) = &self.initial {
for event in &initial.history {
if event.session_id == self.session_id
&& let Some(outcome) = matching_turn_outcome(event, turn_id)
{
return Ok(outcome);
}
}
}
tokio::time::timeout(timeout, async {
loop {
match self.next_event().await? {
SessionWatchEvent::Durable(event) => {
if event.session_id == self.session_id
&& let Some(outcome) = matching_turn_outcome(&event, turn_id)
{
return Ok(outcome);
}
}
SessionWatchEvent::Live(_) => {}
SessionWatchEvent::ResyncRequired => {
return Err(ClientError::Protocol(
"turn outcome observation requires resynchronization".to_string(),
));
}
}
}
})
.await
.map_err(|_| ClientError::RequestTimeout { timeout })?
}
pub async fn recover_and_wait_for_turn_outcome(
&mut self,
turn_id: &str,
timeout: Duration,
) -> Result<bcode_session_models::ModelTurnOutcome, ClientError> {
tokio::time::timeout(timeout, async {
let response = self
.connection
.send_request(Request::SessionHistoryPage {
session_id: self.session_id,
query: SessionHistoryQuery {
cursor: None,
limit: bcode_session_models::MAX_SESSION_HISTORY_READ_EVENTS,
direction: bcode_session_models::SessionHistoryDirection::Backward,
},
})
.await?;
let ResponsePayload::SessionHistoryPage { page } = response else {
return Err(ClientError::UnexpectedResponse);
};
if page.session_id != self.session_id {
return Err(ClientError::Protocol(
"turn history session identity mismatch".to_owned(),
));
}
for event in &page.events {
if event.session_id == self.session_id
&& let Some(outcome) = matching_turn_outcome(event, turn_id)
{
return Ok(outcome);
}
}
self.wait_for_turn_outcome(turn_id, timeout).await
})
.await
.map_err(|_| ClientError::RequestTimeout { timeout })?
}
pub async fn next_event(&mut self) -> Result<SessionWatchEvent, ClientError> {
loop {
match self.connection.recv_event().await? {
Event::Session(event) | Event::RuntimeWork(event) => {
return Ok(SessionWatchEvent::Durable(Box::new(event)));
}
Event::SessionLive(event) => {
return Ok(SessionWatchEvent::Live(Box::new(event)));
}
Event::SessionViewResyncRequired {
session_id: required,
} if required == self.initial_session_id() => {
return Ok(SessionWatchEvent::ResyncRequired);
}
Event::SessionCatalogUpdated { .. }
| Event::Workflow(_)
| Event::SessionViewResyncRequired { .. } => {}
}
}
}
}
fn matching_turn_outcome(
event: &SessionEvent,
expected: &str,
) -> Option<bcode_session_models::ModelTurnOutcome> {
match &event.kind {
bcode_session_models::SessionEventKind::ModelTurnFinished {
turn_id, outcome, ..
} if turn_id == expected => Some(*outcome),
_ => None,
}
}
#[derive(Debug)]
pub struct RuntimeWorkWatcher {
connection: ClientConnection,
}
impl RuntimeWorkWatcher {
pub async fn next_event(&mut self) -> Result<SessionEvent, ClientError> {
loop {
match self.connection.recv_event().await? {
Event::RuntimeWork(event) => return Ok(event),
Event::Session(_)
| Event::SessionLive(_)
| Event::Workflow(_)
| Event::SessionViewResyncRequired { .. }
| Event::SessionCatalogUpdated { .. } => {}
}
}
}
}
#[derive(Debug)]
pub struct WorkflowRunWatcher {
connection: ClientConnection,
sequence: bcode_workflow_view_models::WorkflowLiveSequence,
}
pub use bcode_workflow::WorkflowRunWatchEvent;
impl bcode_workflow::WorkflowRunObservationApplication for BcodeClient {
type Error = ClientError;
type Subscription = WorkflowRunWatcher;
async fn watch_workflow_runs(&self) -> Result<Self::Subscription, Self::Error> {
Self::watch_workflow_runs(self).await
}
}
impl bcode_workflow::WorkflowRunSubscription for WorkflowRunWatcher {
type Error = ClientError;
async fn next_event(&mut self) -> Result<WorkflowRunWatchEvent, Self::Error> {
Self::next_event(self).await
}
}
impl WorkflowRunWatcher {
pub async fn next_event(&mut self) -> Result<WorkflowRunWatchEvent, ClientError> {
loop {
let received = match self.connection.recv_event_without_reconnect().await {
Ok(event) => event,
Err(error)
if error.is_daemon_unavailable()
&& self.connection.reconnect_client.is_some() =>
{
self.connection.restore_state.workflow_runs = false;
self.connection.reconnect_and_restore().await?;
let after_sequence = self.connection.subscribe_workflow_runs().await?;
self.sequence =
bcode_workflow_view_models::WorkflowLiveSequence::from_last_observed(
after_sequence,
);
return Ok(WorkflowRunWatchEvent::ResyncRequired);
}
Err(error) => return Err(error),
};
let event = match received {
Event::Workflow(event) => event,
Event::Session(_)
| Event::SessionLive(_)
| Event::RuntimeWork(_)
| Event::SessionViewResyncRequired { .. }
| Event::SessionCatalogUpdated { .. } => continue,
};
match self.sequence.observe(&event) {
bcode_workflow_view_models::WorkflowLiveEventDisposition::Refetch => {
return Ok(WorkflowRunWatchEvent::Changed(event));
}
bcode_workflow_view_models::WorkflowLiveEventDisposition::Duplicate => {}
bcode_workflow_view_models::WorkflowLiveEventDisposition::UnsupportedVersion => {
return Ok(WorkflowRunWatchEvent::UnsupportedVersion {
version: event.version,
});
}
bcode_workflow_view_models::WorkflowLiveEventDisposition::Gap => {
let after_sequence = self.sequence.last_observed().unwrap_or(0);
let page = self
.connection
.workflow_live_event_catch_up(after_sequence, 256)
.await?;
if page.resync_required {
return Ok(WorkflowRunWatchEvent::ResyncRequired);
}
for caught_up in page.events {
match self.sequence.observe(&caught_up) {
bcode_workflow_view_models::WorkflowLiveEventDisposition::Refetch
| bcode_workflow_view_models::WorkflowLiveEventDisposition::Duplicate => {}
bcode_workflow_view_models::WorkflowLiveEventDisposition::Gap => {
return Ok(WorkflowRunWatchEvent::ResyncRequired);
}
bcode_workflow_view_models::WorkflowLiveEventDisposition::UnsupportedVersion => {
return Ok(WorkflowRunWatchEvent::UnsupportedVersion {
version: caught_up.version,
});
}
}
}
return Ok(WorkflowRunWatchEvent::Changed(event));
}
}
}
}
}
fn configured_request_timeout() -> Duration {
bcode_config::load_config().map_or(DEFAULT_CLIENT_IPC_REQUEST_TIMEOUT, |config| {
Duration::from_secs(config.client.request_timeout_secs)
})
}
impl BcodeClient {
#[must_use]
pub fn default_endpoint() -> Self {
let (runtime_context, runtime_context_error) = match current_runtime_context() {
Ok(context) => (Some(context), None),
Err(error) => (None, Some(error.to_string())),
};
Self {
expected_state_location: None,
endpoint: default_endpoint(),
runtime_context,
runtime_context_error,
daemon_availability: DaemonAvailability::AutoStart,
connect_timeout: DEFAULT_CLIENT_CONNECT_TIMEOUT,
startup_timeout: DEFAULT_CLIENT_DAEMON_START_TIMEOUT,
startup_gate: Arc::new(tokio::sync::Mutex::new(())),
contextual_invocations: contextual::Controls::default(),
request_timeout: configured_request_timeout(),
}
}
#[must_use]
pub fn new(endpoint: IpcEndpoint) -> Self {
Self {
expected_state_location: None,
endpoint,
runtime_context: None,
runtime_context_error: None,
daemon_availability: DaemonAvailability::RequireRunning,
connect_timeout: DEFAULT_CLIENT_CONNECT_TIMEOUT,
startup_timeout: DEFAULT_CLIENT_DAEMON_START_TIMEOUT,
startup_gate: Arc::new(tokio::sync::Mutex::new(())),
contextual_invocations: contextual::Controls::default(),
request_timeout: DEFAULT_CLIENT_IPC_REQUEST_TIMEOUT,
}
}
#[must_use]
pub fn for_state_location(
endpoint: IpcEndpoint,
location: &bcode_config::StateLocation,
) -> Self {
let mut client = Self::new(endpoint);
client.expected_state_location = Some(location.id().clone());
client
}
#[must_use]
pub fn with_runtime_context(mut self, runtime_context: Option<ClientRuntimeContext>) -> Self {
self.runtime_context = runtime_context;
self.runtime_context_error = None;
self
}
#[must_use]
pub fn with_interaction_adapters(
mut self,
interaction_adapters: Vec<
bcode_plugin_sdk::interaction::PluginInteractionAdapterCapability,
>,
) -> Self {
let context = self.runtime_context.get_or_insert_default();
context.interaction_adapters = interaction_adapters;
self
}
#[must_use]
pub fn with_interaction_adapter(
mut self,
interaction_adapter: bcode_plugin_sdk::interaction::PluginInteractionAdapterCapability,
) -> Self {
self.runtime_context
.get_or_insert_default()
.interaction_adapters
.push(interaction_adapter);
self
}
#[must_use]
pub const fn with_connect_timeout(mut self, connect_timeout: Duration) -> Self {
self.connect_timeout = connect_timeout;
self
}
#[must_use]
pub const fn with_startup_timeout(mut self, startup_timeout: Duration) -> Self {
self.startup_timeout = startup_timeout;
self
}
#[must_use]
pub const fn with_request_timeout(mut self, request_timeout: Duration) -> Self {
self.request_timeout = request_timeout;
self
}
#[must_use]
pub const fn connect_timeout(&self) -> Duration {
self.connect_timeout
}
#[must_use]
pub const fn startup_timeout(&self) -> Duration {
self.startup_timeout
}
#[must_use]
pub const fn request_timeout(&self) -> Duration {
self.request_timeout
}
#[must_use]
pub const fn with_daemon_availability(
mut self,
daemon_availability: DaemonAvailability,
) -> Self {
self.daemon_availability = daemon_availability;
self
}
pub async fn ensure_daemon_available(&self) -> Result<(), ClientError> {
if self.daemon_availability == DaemonAvailability::RequireRunning
|| self.expected_state_location.is_some()
{
return Ok(());
}
let coordination_started = std::time::Instant::now();
let _startup_guard = self.startup_gate.lock().await;
tracing::debug!(
target: "bcode_client::startup",
elapsed_us = coordination_started.elapsed().as_micros(),
"client startup gate acquired"
);
if self
.connect_with_deadline("bcode-daemon-availability")
.await
.is_ok()
{
return Ok(());
}
tokio::time::timeout(
self.startup_timeout,
ensure_daemon_running(&EnsureDaemonOptions {
endpoint: self.endpoint.clone(),
quiet: true,
log_path: bcode_daemon_lifecycle::default_daemon_log_path(),
}),
)
.await
.map_err(|_| ClientError::DaemonStartupTimeout {
timeout: self.startup_timeout,
})??;
Ok(())
}
pub async fn watch_session_catalog(&self) -> Result<SessionCatalogWatcher, ClientError> {
self.watch_session_catalog_with_scope(SessionDiscoveryScope::WorkingDirectory(
current_working_directory(),
))
.await
}
pub async fn watch_session_catalog_with_scope(
&self,
scope: SessionDiscoveryScope,
) -> Result<SessionCatalogWatcher, ClientError> {
let scope = normalize_discovery_scope(scope)?;
let mut connection = self.connect("bcode-session-catalog").await?;
connection.subscribe_catalog_updates().await?;
Ok(SessionCatalogWatcher {
connection,
last_revision: 0,
scope,
})
}
pub async fn watch_session(
&self,
session_id: SessionId,
history_limit: usize,
) -> Result<SessionWatcher, ClientError> {
let mut connection = self.connect("bcode-session-view").await?;
let initial = connection
.attach_session_recent_with_input_history(session_id, history_limit)
.await?;
Ok(SessionWatcher {
connection,
session_id,
initial: Some(initial),
})
}
pub async fn watch_session_projection_window(
&self,
session_id: SessionId,
request: ProjectionWindowRequest,
) -> Result<SessionWatcher, ClientError> {
let mut connection = self.connect("bcode-session-view").await?;
let initial = connection
.attach_session_projection_window_with_input_history(session_id, request)
.await?;
Ok(SessionWatcher {
connection,
session_id,
initial: Some(initial),
})
}
pub async fn watch_runtime_work(
&self,
session_id: SessionId,
) -> Result<RuntimeWorkWatcher, ClientError> {
let mut connection = self.connect("bcode-runtime-work").await?;
connection.subscribe_runtime_work(session_id).await?;
Ok(RuntimeWorkWatcher { connection })
}
pub async fn watch_workflow_runs(&self) -> Result<WorkflowRunWatcher, ClientError> {
let mut connection = self.connect("bcode-workflow-runs").await?;
let after_sequence = connection.subscribe_workflow_runs().await?;
Ok(WorkflowRunWatcher {
connection,
sequence: bcode_workflow_view_models::WorkflowLiveSequence::from_last_observed(
after_sequence,
),
})
}
pub async fn ping(&self) -> Result<(), ClientError> {
match self.send_request(Request::Ping).await? {
ResponsePayload::Pong => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn ingest_client_metrics(
&self,
batch: bcode_metrics::ClientMetricBatch,
) -> Result<usize, ClientError> {
match self
.send_request(Request::IngestClientMetrics { batch })
.await?
{
ResponsePayload::ClientMetricsIngested { accepted } => Ok(accepted),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn model_catalog_diagnostics(
&self,
) -> Result<bcode_model_catalog_models::ModelCatalogDiagnostics, ClientError> {
match self.send_request(Request::ModelCatalogDiagnostics).await? {
ResponsePayload::ModelCatalogDiagnostics { diagnostics } => Ok(diagnostics),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn server_status(&self) -> Result<bcode_ipc::ServerStatus, ClientError> {
match self
.send_request(Request::ServerStatus {
working_directory: Some(current_working_directory()),
})
.await?
{
ResponsePayload::ServerStatus { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
#[cfg(test)]
fn verify_daemon_identity(status: &bcode_ipc::DaemonStatus) -> Result<(), ClientError> {
Self::verify_daemon_identity_at(status, &bcode_ipc::state_location_id())
}
fn verify_server_identity(&self, status: &bcode_ipc::DaemonStatus) -> Result<(), ClientError> {
let expected = self
.expected_state_location
.as_ref()
.map_or_else(bcode_ipc::state_location_id, |id| id.as_str().to_owned());
Self::verify_daemon_identity_at(status, &expected)
}
fn verify_daemon_identity_at(
status: &bcode_ipc::DaemonStatus,
expected_state_location: &str,
) -> Result<(), ClientError> {
let expected_namespace = bcode_ipc::daemon_namespace();
let expected_protocol = u32::from(bcode_ipc::CURRENT_PROTOCOL_VERSION);
let expected_artifact_id = bcode_ipc::ArtifactId::current();
let expected_writer_epoch = bcode_ipc::CURRENT_SESSION_STORAGE_WRITER_EPOCH;
let expected_event_schema = bcode_session_models::CURRENT_SESSION_EVENT_SCHEMA_VERSION;
let expected_digest = bcode_daemon_lifecycle::current_executable_identity()
.map_err(|error| ClientError::IncompatibleDaemon {
message: error.to_string(),
})?
.1;
if status.executable_digest.as_deref() == Some(expected_digest.as_str())
&& status.namespace == expected_namespace
&& status.protocol_version == expected_protocol
&& status.artifact_id.as_ref() == Some(&expected_artifact_id)
&& status.build_fingerprint == bcode_ipc::BUILD_FINGERPRINT
&& status.storage_writer_epoch == Some(expected_writer_epoch)
&& status.session_event_schema_version == Some(expected_event_schema)
&& status.state_location_id.as_deref() == Some(expected_state_location)
{
return Ok(());
}
Err(ClientError::IncompatibleDaemon {
message: format!(
"client expects namespace={expected_namespace} artifact={expected_artifact_id} protocol={expected_protocol} build={} session_event_schema={expected_event_schema} storage_writer_epoch={expected_writer_epoch} state_location={expected_state_location}; daemon reported namespace={} artifact={} protocol={} build={} executable={} session_event_schema={} storage_writer_epoch={} state_location={}",
bcode_ipc::BUILD_FINGERPRINT,
status.namespace,
status
.artifact_id
.as_ref()
.map_or("<unknown>", bcode_ipc::ArtifactId::as_str),
status.protocol_version,
status.build_fingerprint,
status.executable_digest.as_deref().unwrap_or("<unknown>"),
status
.session_event_schema_version
.map_or_else(|| "<unknown>".to_owned(), |value| value.to_string()),
status
.storage_writer_epoch
.map_or_else(|| "<unknown>".to_owned(), |value| value.to_string()),
status.state_location_id.as_deref().unwrap_or("<unknown>"),
),
})
}
pub async fn verified_server_status(&self) -> Result<bcode_ipc::ServerStatus, ClientError> {
let status = self.server_status().await?;
self.verify_server_identity(&status.daemon)?;
Ok(status)
}
pub async fn server_stop(&self) -> Result<(), ClientError> {
self.server_stop_with_mode(ServerStopMode::Force).await
}
pub async fn server_stop_if_idle(&self) -> Result<(), ClientError> {
self.server_stop_with_mode(ServerStopMode::IfIdle).await
}
pub async fn release_session_ownership(
&self,
session_id: bcode_session_models::SessionId,
) -> Result<bcode_ipc::SessionOwnershipReleaseOutcome, ClientError> {
match self
.send_request(Request::ReleaseSessionOwnership { session_id })
.await?
{
ResponsePayload::SessionOwnershipReleased {
session_id: released_session_id,
outcome,
} if released_session_id == session_id => Ok(outcome),
_ => Err(ClientError::UnexpectedResponse),
}
}
async fn server_stop_with_mode(&self, mode: ServerStopMode) -> Result<(), ClientError> {
match self.send_request(Request::ServerStop { mode }).await? {
ResponsePayload::ServerStopping => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn composer_draft(
&self,
scope: bcode_session_models::ComposerDraftScope,
) -> Result<Option<String>, ClientError> {
match self.send_request(Request::ComposerDraft { scope }).await? {
ResponsePayload::ComposerDraft { draft } => Ok(draft),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn set_composer_draft(
&self,
scope: bcode_session_models::ComposerDraftScope,
text: String,
) -> Result<(), ClientError> {
match self
.send_request(Request::SetComposerDraft { scope, text })
.await?
{
ResponsePayload::ComposerDraftSet => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn create_session(
&self,
name: Option<String>,
) -> Result<SessionSummary, ClientError> {
self.create_session_in_working_directory(name, current_working_directory())
.await
}
pub async fn create_session_in_working_directory(
&self,
name: Option<String>,
working_directory: std::path::PathBuf,
) -> Result<SessionSummary, ClientError> {
let working_directory =
resolve_path_from(Some(working_directory), ¤t_working_directory());
match self
.send_request(Request::CreateSession {
name,
working_directory,
})
.await?
{
ResponsePayload::SessionCreated { session } => Ok(session),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn create_child_session(
&self,
mut request: bcode_session_models::SessionChildCreateRequest,
) -> Result<SessionSummary, ClientError> {
request.working_directory = resolve_path_from(
Some(request.working_directory),
¤t_working_directory(),
);
match self
.send_request(Request::CreateChildSession { request })
.await?
{
ResponsePayload::SessionCreated { session } => Ok(session),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn reconcile_session_children(
&self,
parent_id: SessionId,
apply: bool,
) -> Result<bcode_session_models::SessionChildReconciliation, ClientError> {
match self
.send_request(Request::ReconcileSessionChildren { parent_id, apply })
.await?
{
ResponsePayload::SessionChildrenReconciled { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_summary(
&self,
session_id: SessionId,
) -> Result<SessionSummary, ClientError> {
match self
.send_request(Request::SessionSummary { session_id })
.await?
{
ResponsePayload::SessionSummary { session } => Ok(session),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_activity(
&self,
session_id: SessionId,
query: bcode_session_models::SessionActivityQuery,
) -> Result<bcode_session_models::SessionActivityPage, ClientError> {
match self
.send_request(Request::SessionActivity { session_id, query })
.await?
{
ResponsePayload::SessionActivity { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_progress(
&self,
session_id: SessionId,
) -> Result<SessionProgressSnapshot, ClientError> {
let payload = self
.send_request(Request::SessionProgressSnapshot { session_id })
.await?;
decode_session_progress_response(payload, session_id)
}
pub async fn query_child_sessions(
&self,
query: bcode_session_models::SessionChildQuery,
) -> Result<bcode_session_models::SessionChildPage, ClientError> {
match self
.send_request(Request::SessionChildren { query })
.await?
{
ResponsePayload::SessionChildren { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_sessions(&self) -> Result<Vec<SessionSummary>, ClientError> {
Ok(self.list_sessions_with_status().await?.sessions)
}
pub async fn list_sessions_with_status(&self) -> Result<SessionList, ClientError> {
self.list_sessions_in_working_directory(current_working_directory())
.await
}
pub async fn list_sessions_with_scope(
&self,
scope: SessionDiscoveryScope,
) -> Result<SessionList, ClientError> {
let scope = normalize_discovery_scope(scope)?;
decode_scoped_session_list(
self.send_request(Request::ListSessionsWithScope { scope })
.await?,
)
}
pub async fn list_sessions_page_with_scope(
&self,
scope: SessionDiscoveryScope,
query: SessionCatalogQuery,
cursor: Option<SessionCatalogCursor>,
limit: usize,
) -> Result<SessionCatalogPage, ClientError> {
let scope = normalize_discovery_scope(scope)?;
match self
.send_request(Request::ListSessionsPageWithScope {
scope,
query,
cursor,
limit,
})
.await?
{
ResponsePayload::SessionCatalogPage { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_sessions_in_working_directory(
&self,
working_directory: PathBuf,
) -> Result<SessionList, ClientError> {
let working_directory = std::path::absolute(working_directory).map_err(|_| {
ClientError::Protocol("cannot resolve catalog working directory".into())
})?;
match self
.send_request(Request::ListSessions { working_directory })
.await?
{
ResponsePayload::SessionList {
sessions,
catalog_status,
catalog_sources,
catalog_revision,
} => Ok(SessionList {
sessions,
catalog_status,
catalog_sources,
catalog_revision,
}),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_compatibility_inventory(
&self,
request: SessionCompatibilityInventoryRequest,
) -> Result<SessionCompatibilityInventoryResponse, ClientError> {
match self
.send_request(Request::SessionCompatibilityInventory { request })
.await?
{
ResponsePayload::SessionCompatibilityInventory { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn start_session_bulk_migration(
&self,
request: SessionBulkMigrationStartRequest,
) -> Result<SessionBulkMigrationOperationStatus, ClientError> {
match self
.send_request(Request::SessionBulkMigrationStart { request })
.await?
{
ResponsePayload::SessionBulkMigrationOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_bulk_migration_status(
&self,
operation_id: String,
) -> Result<SessionBulkMigrationOperationStatus, ClientError> {
match self
.send_request(Request::SessionBulkMigrationStatus { operation_id })
.await?
{
ResponsePayload::SessionBulkMigrationOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn wait_session_bulk_migration(
&self,
operation_id: String,
after_revision: u64,
timeout_ms: u64,
) -> Result<SessionBulkMigrationOperationStatus, ClientError> {
validate_session_bulk_migration_wait_timeout(timeout_ms)?;
let server_wait = Duration::from_millis(timeout_ms);
let response_timeout = self
.request_timeout
.max(server_wait.saturating_add(LONG_POLL_TRANSPORT_GRACE));
match self
.send_request_with_timeout(
Request::SessionBulkMigrationWait {
operation_id,
after_revision,
timeout_ms,
},
response_timeout,
)
.await?
{
ResponsePayload::SessionBulkMigrationOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn cancel_session_bulk_migration(
&self,
operation_id: String,
) -> Result<SessionBulkMigrationOperationStatus, ClientError> {
match self
.send_request(Request::SessionBulkMigrationCancel { operation_id })
.await?
{
ResponsePayload::SessionBulkMigrationOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn import_external_session(
&self,
source_id: impl Into<String>,
external_session_id: impl Into<String>,
) -> Result<(SessionSummary, Vec<SessionImportWarning>), ClientError> {
match self
.send_request(Request::ImportExternalSession {
source_id: source_id.into(),
external_session_id: external_session_id.into(),
working_directory: Some(current_working_directory()),
})
.await?
{
ResponsePayload::ExternalSessionImported { session, warnings } => {
Ok((session, warnings))
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn refresh_session_catalog(
&self,
sources: Option<Vec<String>>,
) -> Result<SessionList, ClientError> {
self.refresh_session_catalog_in_working_directory(current_working_directory(), sources)
.await
}
pub async fn refresh_session_catalog_in_working_directory(
&self,
working_directory: PathBuf,
sources: Option<Vec<String>>,
) -> Result<SessionList, ClientError> {
let working_directory = std::path::absolute(working_directory).map_err(|_| {
ClientError::Protocol("cannot resolve catalog working directory".into())
})?;
match self
.send_request(Request::RefreshSessionCatalog {
working_directory: Some(working_directory),
sources,
})
.await?
{
ResponsePayload::SessionCatalogRefreshed {
sessions,
catalog_status,
catalog_sources,
catalog_revision,
} => Ok(SessionList {
sessions,
catalog_status,
catalog_sources,
catalog_revision,
}),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn change_session_working_directory(
&self,
session_id: SessionId,
working_directory: impl Into<std::path::PathBuf>,
) -> Result<SessionSummary, ClientError> {
let working_directory =
resolve_path_from(Some(working_directory.into()), ¤t_working_directory());
match self
.send_request(Request::ChangeSessionWorkingDirectory {
session_id,
working_directory,
})
.await?
{
ResponsePayload::SessionWorkingDirectoryChanged { session, .. } => Ok(session),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_worktrees(
&self,
mut request: WorktreeListRequest,
) -> Result<WorktreeListResponse, ClientError> {
request.cwd = Some(resolve_caller_path(request.cwd));
match self.send_request(Request::ListWorktrees(request)).await? {
ResponsePayload::WorktreeList(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn start_worktree_create(
&self,
operation_id: String,
mut request: WorktreeCreateRequest,
) -> Result<WorktreeCreateOperationStatus, ClientError> {
request.cwd = Some(resolve_caller_path(request.cwd));
match self
.send_request(Request::WorktreeCreateStart {
operation_id,
request,
})
.await?
{
ResponsePayload::WorktreeCreateOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn worktree_create_status(
&self,
operation_id: String,
) -> Result<WorktreeCreateOperationStatus, ClientError> {
match self
.send_request(Request::WorktreeCreateStatus { operation_id })
.await?
{
ResponsePayload::WorktreeCreateOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn wait_worktree_create(
&self,
operation_id: String,
after_revision: u64,
timeout_ms: u64,
) -> Result<WorktreeCreateOperationStatus, ClientError> {
let server_wait = Duration::from_millis(timeout_ms);
let response_timeout = self
.request_timeout
.max(server_wait.saturating_add(LONG_POLL_TRANSPORT_GRACE));
match self
.send_request_with_timeout(
Request::WorktreeCreateWait {
operation_id,
after_revision,
timeout_ms,
},
response_timeout,
)
.await?
{
ResponsePayload::WorktreeCreateOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn create_worktree(
&self,
request: WorktreeCreateRequest,
) -> Result<WorktreeCreateResponse, ClientError> {
let operation_id = format!(
"worktree-{}-{}",
std::process::id(),
std::time::SystemTime::now()
.duration_since(std::time::UNIX_EPOCH)
.unwrap_or_default()
.as_nanos()
);
let mut status = self
.start_worktree_create(operation_id.clone(), request)
.await?;
loop {
if let Some(response) = status.response {
return Ok(response);
}
if let Some(error) = status.error {
return Err(ClientError::WorktreeCreate {
code: error.code,
message: error.message,
created_path: error.created_path,
});
}
status = self
.wait_worktree_create(operation_id.clone(), status.revision, 30_000)
.await?;
}
}
pub async fn remove_worktree(
&self,
mut request: WorktreeRemoveRequest,
) -> Result<WorktreeRemoveResponse, ClientError> {
request.cwd = Some(resolve_caller_path(request.cwd));
match self.send_request(Request::RemoveWorktree(request)).await? {
ResponsePayload::WorktreeRemoved(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn ralph_status(
&self,
request: RalphStatusRequest,
) -> Result<RalphStatusResponse, ClientError> {
match self.send_request(Request::RalphStatus(request)).await? {
ResponsePayload::RalphStatus(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn run_ralph_loop(
&self,
request: RalphRunRequest,
) -> Result<RalphRunResponse, ClientError> {
match self.send_request(Request::RunRalphLoop(request)).await? {
ResponsePayload::RalphRunStarted(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn approve_ralph_run(
&self,
request: RalphApproveRequest,
) -> Result<RalphRunResponse, ClientError> {
match self.send_request(Request::ApproveRalphRun(request)).await? {
ResponsePayload::RalphRunApproved(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn cancel_ralph_loop(
&self,
request: RalphCancelRequest,
) -> Result<RalphCancelResponse, ClientError> {
match self.send_request(Request::CancelRalphLoop(request)).await? {
ResponsePayload::RalphRunCancelled(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_ralph_runs(
&self,
request: RalphListRunsRequest,
) -> Result<RalphListRunsResponse, ClientError> {
match self
.send_request(Request::ListRalphRuns(Box::new(request)))
.await?
{
ResponsePayload::RalphRunsListed(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_ralph_iterations(
&self,
request: RalphListIterationsRequest,
) -> Result<RalphListIterationsResponse, ClientError> {
match self
.send_request(Request::ListRalphIterations(Box::new(request)))
.await?
{
ResponsePayload::RalphIterationsListed(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn resume_ralph_run(
&self,
request: RalphResumeRequest,
) -> Result<RalphResumeResponse, ClientError> {
match self.send_request(Request::ResumeRalphRun(request)).await? {
ResponsePayload::RalphRunResumed(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn ralph_run_status(
&self,
request: RalphRunStatusRequest,
) -> Result<RalphRunStatusResponse, ClientError> {
match self.send_request(Request::RalphRunStatus(request)).await? {
ResponsePayload::RalphRunStatus(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn record_ralph_lifecycle(
&self,
request: RalphLifecycleRequest,
) -> Result<SessionEvent, ClientError> {
match self
.send_request(Request::RecordRalphLifecycle(request))
.await?
{
ResponsePayload::RalphLifecycleRecorded { event } => Ok(event),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn rename_session(
&self,
session_id: SessionId,
name: Option<String>,
) -> Result<SessionSummary, ClientError> {
match self
.send_request(Request::RenameSession { session_id, name })
.await?
{
ResponsePayload::SessionRenamed { session } => Ok(session),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn delete_session(
&self,
session_id: SessionId,
) -> Result<SessionSummary, ClientError> {
match self
.send_request(Request::DeleteSession { session_id })
.await?
{
ResponsePayload::SessionDeleted { session } => Ok(session),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn send_invocation_input(
&self,
session_id: SessionId,
input: bcode_tool::ToolInvocationInput,
) -> Result<(), ClientError> {
match self
.send_request(Request::InvocationInput { session_id, input })
.await?
{
ResponsePayload::InvocationInputAccepted => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_diagnostic_artifacts(
&self,
run_id: String,
dispatch_identity: String,
) -> Result<bcode_workflow::WorkflowDiagnosticArtifacts, ClientError> {
match self
.send_request(Request::ListWorkflowDiagnosticArtifacts {
run_id: run_id.clone(),
dispatch_identity: dispatch_identity.clone(),
})
.await?
{
ResponsePayload::WorkflowDiagnosticArtifacts { artifacts }
if artifacts.version == 1
&& artifacts.run_id == run_id
&& artifacts.dispatch_identity == dispatch_identity
&& artifacts.artifacts.len() + artifacts.unavailable.len() <= 100 =>
{
Ok(artifacts)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_diagnostic_artifact(
&self,
request: bcode_workflow::WorkflowDiagnosticArtifactRequest,
) -> Result<bcode_workflow::WorkflowDiagnosticArtifactRange, ClientError> {
let expected = request.clone();
match self
.send_request(Request::ReadWorkflowDiagnosticArtifact { request })
.await?
{
ResponsePayload::WorkflowDiagnosticArtifactRange { range }
if range.version == 1
&& range.run_id == expected.run_id
&& range.dispatch_identity == expected.dispatch_identity
&& range.artifact_id == expected.artifact_id
&& range.offset == expected.offset
&& range.bytes.len() <= expected.length as usize
&& range
.offset
.checked_add(range.bytes.len() as u64)
.is_some_and(|end| {
end <= range.byte_len && range.eof == (end == range.byte_len)
})
&& range.sha256.len() == 64
&& range.sha256.bytes().all(|byte| byte.is_ascii_hexdigit()) =>
{
Ok(range)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_trace_blob_range(
&self,
request: bcode_session_models::ReadSessionTraceBlobRequest,
) -> Result<bcode_session_models::SessionTraceBlobRange, ClientError> {
let expected = request.clone();
match self
.send_request(Request::ReadSessionTraceBlob { request })
.await?
{
ResponsePayload::SessionTraceBlobRange { range }
if range.version == expected.version
&& range.session_id == expected.session_id
&& range.sequence == expected.sequence
&& range.offset == expected.offset
&& range.bytes.len() <= expected.length as usize
&& range
.offset
.checked_add(range.bytes.len() as u64)
.is_some_and(|end| {
end <= range.byte_len && range.eof == (end == range.byte_len)
}) =>
{
Ok(range)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_artifact_range(
&self,
session_id: SessionId,
artifact_id: String,
reference_key: String,
offset: u64,
length: u32,
) -> Result<SessionArtifactRange, ClientError> {
if length == 0 || length > bcode_session_models::MAX_SESSION_ARTIFACT_RANGE_BYTES {
return Err(ClientError::Protocol(format!(
"artifact range length must be between 1 and {} bytes",
bcode_session_models::MAX_SESSION_ARTIFACT_RANGE_BYTES,
)));
}
match self
.send_request(Request::ReadSessionArtifact {
session_id,
artifact_id: artifact_id.clone(),
reference_key: reference_key.clone(),
offset,
length,
})
.await?
{
ResponsePayload::SessionArtifactRange { range } => {
validate_artifact_range_response(
&range,
&artifact_id,
&reference_key,
offset,
length,
)?;
Ok(range)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn prepare_context_generation(
&self,
request: bcode_session_models::PrepareContextGeneration,
) -> Result<bcode_session_models::PreparedContextGeneration, ClientError> {
match self
.send_request(Request::PrepareContextGeneration { request })
.await?
{
ResponsePayload::PreparedContextGeneration { prepared } => Ok(prepared),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_working_document(
&self,
request: bcode_session_models::SessionWorkingDocumentRequest,
) -> Result<Option<bcode_session_models::SessionWorkingDocument>, ClientError> {
match self
.send_request(Request::SessionWorkingDocument { request })
.await?
{
ResponsePayload::SessionWorkingDocument { document } => Ok(document),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn submit_turn(
&self,
session_id: SessionId,
text: String,
admission: bcode_session_models::TurnAdmissionMetadata,
) -> Result<bcode_session_models::TurnAdmission, ClientError> {
match self
.send_request(Request::SubmitTurn {
session_id,
text,
admission,
})
.await?
{
ResponsePayload::TurnAdmission { admission } => Ok(admission),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_history(
&self,
session_id: SessionId,
) -> Result<Vec<SessionEvent>, ClientError> {
match self
.send_request(Request::SessionHistory { session_id })
.await?
{
ResponsePayload::SessionHistory { history, .. } => Ok(history),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn reprice_session(
&self,
session_id: SessionId,
range: bcode_session_models::SessionCostRange,
catalog: bcode_model_catalog_models::CatalogDocument,
) -> Result<bcode_session_models::SessionRepriceReport, ClientError> {
match self
.send_request(Request::RepriceSession {
session_id,
range,
catalog: Box::new(catalog),
})
.await?
{
ResponsePayload::SessionRepriced { report } => Ok(*report),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn usage_catalog(
&self,
after: Option<SessionId>,
) -> Result<Vec<SessionId>, ClientError> {
match self.send_request(Request::UsageCatalog { after }).await? {
ResponsePayload::UsageCatalog { session_ids } => Ok(session_ids),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn usage_report(
&self,
query: bcode_usage_models::UsageQuery,
) -> Result<bcode_usage_models::UsageReport, ClientError> {
match self.send_request(Request::UsageReport { query }).await? {
ResponsePayload::UsageReport { report } => Ok(report),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_usage(
&self,
session_id: SessionId,
query: bcode_session_models::SessionUsageQuery,
) -> Result<bcode_session_models::SessionUsagePage, ClientError> {
match self
.send_request(Request::SessionUsage { session_id, query })
.await?
{
ResponsePayload::SessionUsage { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn collect_usage(
&self,
session_id: SessionId,
query: bcode_session_models::SessionUsageQuery,
) -> Result<bcode_session_models::SessionUsagePage, ClientError> {
match self
.send_request(Request::UsageCollect { session_id, query })
.await?
{
ResponsePayload::UsageCollected { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_admission(
&self,
session_id: SessionId,
apply: bool,
) -> Result<bcode_session_models::StorageAdmissionReport, ClientError> {
match self
.send_request(Request::SessionAdmission { session_id, apply })
.await?
{
ResponsePayload::SessionAdmission { report } => Ok(report),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn compress_session_page(
&self,
request: bcode_session_models::StorageCompressionRequest,
) -> Result<bcode_session_models::StorageCompressionResult, ClientError> {
match self
.send_request(Request::SessionCompress { request })
.await?
{
ResponsePayload::SessionCompressed { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_storage_usage(
&self,
session_id: SessionId,
entry_budget: u32,
) -> Result<bcode_session_models::SessionStorageUsage, ClientError> {
match self
.send_request(Request::SessionStorageUsage {
session_id,
entry_budget,
})
.await?
{
ResponsePayload::SessionStorageUsage { usage } => Ok(usage),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_history_page(
&self,
session_id: SessionId,
query: SessionHistoryQuery,
) -> Result<SessionHistoryPage, ClientError> {
match self
.send_request(Request::SessionHistoryPage { session_id, query })
.await?
{
ResponsePayload::SessionHistoryPage { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_projection_history_page(
&self,
session_id: SessionId,
request: bcode_session_models::ProjectionWindowRequest,
) -> Result<bcode_session_models::ProjectionHistoryPage, ClientError> {
match self
.send_request(Request::SessionProjectionHistoryPage {
session_id,
request: Box::new(request),
})
.await?
{
ResponsePayload::SessionProjectionHistoryPage { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_history_around(
&self,
session_id: SessionId,
query: SessionHistoryAroundQuery,
) -> Result<SessionHistoryWindow, ClientError> {
match self
.send_request(Request::SessionHistoryAround { session_id, query })
.await?
{
ResponsePayload::SessionHistoryAround { window } => Ok(window),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_inspection(
&self,
session_id: SessionId,
query: SessionInspectionQuery,
) -> Result<SessionInspectionPage, ClientError> {
match self
.send_request(Request::SessionInspection { session_id, query })
.await?
{
ResponsePayload::SessionInspection { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search(
&self,
request: bcode_session_search::SessionSearchRequest,
policy: bcode_session_search::SessionSearchPlanPolicy,
routes: Vec<bcode_session_search::SessionSearchContentRoute>,
hydrate: bool,
) -> Result<
(
bcode_session_search::FederatedSessionSearchResponse,
Vec<bcode_session_search::HydratedSessionSearchHit>,
),
ClientError,
> {
match self
.send_request(Request::SessionSearch {
request,
policy,
routes,
hydrate,
})
.await?
{
ResponsePayload::SessionSearch {
response,
hydrated_hits,
} => Ok((response, hydrated_hits)),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_providers(
&self,
) -> Result<bcode_session_search::ListSessionSearchProvidersResponse, ClientError> {
match self.send_request(Request::SessionSearchProviders).await? {
ResponsePayload::SessionSearchProviders { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_explain(
&self,
request: bcode_session_search::SessionSearchRequest,
policy: bcode_session_search::SessionSearchPlanPolicy,
routes: Vec<bcode_session_search::SessionSearchContentRoute>,
) -> Result<bcode_session_search::SessionSearchPlan, ClientError> {
match self
.send_request(Request::SessionSearchExplain {
request,
policy,
routes,
})
.await?
{
ResponsePayload::SessionSearchPlan { plan } => Ok(plan),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_purge(
&self,
provider_id: String,
confirmation: String,
) -> Result<bcode_session_search::SessionSearchMaintenanceResponse, ClientError> {
match self
.send_request(Request::SessionSearchPurge {
provider_id,
confirmation,
})
.await?
{
ResponsePayload::SessionSearchMaintenance { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_rebuild(
&self,
provider_id: String,
confirmation: String,
) -> Result<bcode_session_search::SessionSearchMaintenanceResponse, ClientError> {
match self
.send_request(Request::SessionSearchRebuild {
provider_id,
confirmation,
})
.await?
{
ResponsePayload::SessionSearchMaintenance { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_complete_backfill_start(
&self,
request: bcode_session_search::CompleteSessionSearchBackfillRequest,
) -> Result<bcode_session_search::StartSessionSearchBackfillResponse, ClientError> {
match self
.send_request(Request::SessionSearchCompleteBackfillStart { request })
.await?
{
ResponsePayload::SessionSearchCompleteBackfillStarted { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_index_all_start(
&self,
request: bcode_session_search::CompleteSessionSearchBackfillRequest,
) -> Result<bcode_session_search::StartSessionSearchBackfillResponse, ClientError> {
self.session_search_complete_backfill_start(request).await
}
pub async fn session_search_backfill_start(
&self,
request: bcode_session_search::BackfillSessionSearchRequest,
) -> Result<bcode_session_search::StartSessionSearchBackfillResponse, ClientError> {
match self
.send_request(Request::SessionSearchBackfillStart { request })
.await?
{
ResponsePayload::SessionSearchBackfillStarted { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_backfill_status(
&self,
operation_id: String,
) -> Result<bcode_session_search::SessionSearchBackfillOperationStatus, ClientError> {
match self
.send_request(Request::SessionSearchBackfillStatus { operation_id })
.await?
{
ResponsePayload::SessionSearchBackfillOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_backfill_wait(
&self,
operation_id: String,
after_revision: u64,
timeout_ms: u64,
) -> Result<bcode_session_search::SessionSearchBackfillOperationStatus, ClientError> {
validate_session_search_backfill_wait_timeout(timeout_ms)?;
let server_wait = Duration::from_millis(timeout_ms);
let response_timeout = self
.request_timeout
.max(server_wait.saturating_add(LONG_POLL_TRANSPORT_GRACE));
match self
.send_request_with_timeout(
Request::SessionSearchBackfillWait {
operation_id,
after_revision,
timeout_ms,
},
response_timeout,
)
.await?
{
ResponsePayload::SessionSearchBackfillOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_backfill_cancel(
&self,
operation_id: String,
) -> Result<bcode_session_search::SessionSearchBackfillOperationStatus, ClientError> {
match self
.send_request(Request::SessionSearchBackfillCancel { operation_id })
.await?
{
ResponsePayload::SessionSearchBackfillOperation { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_search_backfill(
&self,
request: bcode_session_search::BackfillSessionSearchRequest,
) -> Result<bcode_session_search::SessionSearchBackfillResponse, ClientError> {
match self
.send_request(Request::SessionSearchBackfill { request })
.await?
{
ResponsePayload::SessionSearchBackfill { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn pending_submissions(
&self,
session_id: SessionId,
after: Option<u64>,
limit: u32,
) -> Result<bcode_session_models::PendingSubmissionsPage, ClientError> {
match self
.send_request(Request::PendingSubmissions {
session_id,
after,
limit,
})
.await?
{
ResponsePayload::PendingSubmissions { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn queue_status(
&self,
session_id: SessionId,
) -> Result<bcode_session_models::SubmissionQueueStatus, ClientError> {
match self
.send_request(Request::SubmissionQueueStatus { session_id })
.await?
{
ResponsePayload::SubmissionQueueStatus { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn prepare_submission_resume(
&self,
session_id: SessionId,
submission_id: bcode_session_models::SubmissionId,
expected_accepted_sequence: u64,
) -> Result<bcode_session_models::SubmissionResumeConfirmation, ClientError> {
match self
.send_request(Request::PrepareSubmissionResume {
session_id,
submission_id,
expected_accepted_sequence,
})
.await?
{
ResponsePayload::SubmissionResumePrepared { confirmation } => Ok(confirmation),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn resume_submission(
&self,
session_id: SessionId,
confirmation: bcode_session_models::SubmissionResumeConfirmation,
operation_id: String,
) -> Result<bcode_session_models::SubmissionRecoveryResult, ClientError> {
match self
.send_request(Request::ResumeSubmission {
session_id,
confirmation,
operation_id,
})
.await?
{
ResponsePayload::SubmissionRecovered { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn cancel_pending_submission(
&self,
session_id: SessionId,
submission_id: bcode_session_models::SubmissionId,
expected_accepted_sequence: u64,
operation_id: String,
) -> Result<bcode_session_models::SubmissionRecoveryResult, ClientError> {
match self
.send_request(Request::CancelPendingSubmission {
session_id,
submission_id,
expected_accepted_sequence,
operation_id,
})
.await?
{
ResponsePayload::SubmissionRecovered { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn submit_prompt(
&self,
session_id: SessionId,
submission_id: bcode_session_models::SubmissionId,
text: String,
placement: bcode_session_models::SubmissionPlacement,
admission: bcode_session_models::TurnAdmissionMetadata,
) -> Result<bcode_session_models::SubmissionReceipt, ClientError> {
match self
.send_request(Request::SubmitPrompt {
session_id,
submission_id,
text,
placement,
admission,
require_active_turn: false,
})
.await?
{
ResponsePayload::SubmissionAccepted { receipt } => Ok(receipt),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn steer_prompt(
&self,
session_id: SessionId,
submission_id: bcode_session_models::SubmissionId,
text: String,
admission: bcode_session_models::TurnAdmissionMetadata,
) -> Result<bcode_session_models::SubmissionReceipt, ClientError> {
match self
.send_request(Request::SubmitPrompt {
session_id,
submission_id,
text,
placement: bcode_session_models::SubmissionPlacement::Steer,
admission,
require_active_turn: true,
})
.await?
{
ResponsePayload::SubmissionAccepted { receipt } => Ok(receipt),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn submissions_for_history(
&self,
session_id: SessionId,
events: &[SessionEvent],
) -> Result<Vec<bcode_session_models::SubmissionRecord>, ClientError> {
let mut ids = std::collections::BTreeSet::new();
for event in events {
match &event.kind {
SessionEventKind::SubmissionAccepted { submission } => {
ids.insert(submission.id.clone());
}
SessionEventKind::SubmissionCancelled { submission_id }
| SessionEventKind::SubmissionContextObserved { submission_id, .. } => {
ids.insert(submission_id.clone());
}
_ => {}
}
if ids.len() > 256 {
return Err(ClientError::Protocol("submission hydration window exceeds 256 identities; request a smaller history window".to_owned()));
}
}
let mut records = Vec::with_capacity(ids.len());
for id in ids {
let record = self.get_submission(session_id, id).await?.ok_or_else(|| {
ClientError::Protocol(
"canonical submission referenced by history is unavailable".to_owned(),
)
})?;
records.push(record);
}
Ok(records)
}
pub async fn get_submission(
&self,
session_id: SessionId,
submission_id: bcode_session_models::SubmissionId,
) -> Result<Option<bcode_session_models::SubmissionRecord>, ClientError> {
match self
.send_request(Request::GetSubmission {
session_id,
submission_id,
})
.await?
{
ResponsePayload::Submission { record } => Ok(record),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn send_user_message(
&self,
session_id: SessionId,
text: String,
placement: bcode_session_models::PromptPlacement,
) -> Result<MessageAcceptance, ClientError> {
decode_message_acceptance(
&self
.send_request(Request::SendUserMessageWithPlacement {
session_id,
text,
placement,
})
.await?,
)
}
pub async fn send_user_message_with_execution(
&self,
session_id: SessionId,
text: String,
placement: bcode_session_models::PromptPlacement,
execution: bcode_session_models::TurnExecutionOptions,
) -> Result<MessageAcceptance, ClientError> {
decode_message_acceptance(
&self
.send_request(Request::SendUserMessageWithExecution {
session_id,
text,
placement,
execution,
})
.await?,
)
}
pub async fn set_session_model(
&self,
session_id: SessionId,
provider_plugin_id: Option<String>,
model_id: String,
) -> Result<(), ClientError> {
match self
.send_request(Request::SetSessionModel {
session_id,
provider_plugin_id,
model_id,
})
.await?
{
ResponsePayload::SessionModelSet => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn auth_pool_list(
&self,
) -> Result<Vec<bcode_provider_auth_models::AuthPoolSummary>, ClientError> {
match self.send_request(Request::AuthPoolList).await? {
ResponsePayload::AuthPoolList { pools } => Ok(pools),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn set_auth_pool_preference(
&self,
pool: String,
profile: Option<String>,
) -> Result<(), ClientError> {
self.apply_auth_pool_preference(bcode_provider_auth_models::SetAuthPoolPreferenceRequest {
pool,
profile,
})
.await
}
pub async fn apply_auth_pool_preference(
&self,
request: bcode_provider_auth_models::SetAuthPoolPreferenceRequest,
) -> Result<(), ClientError> {
let bcode_provider_auth_models::SetAuthPoolPreferenceRequest { pool, profile } = request;
match self
.send_request(Request::SetAuthPoolPreference { pool, profile })
.await?
{
ResponsePayload::AuthPoolPreferenceSet => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn set_session_reasoning(
&self,
session_id: SessionId,
effort: Option<String>,
summary: Option<String>,
) -> Result<(), ClientError> {
match self
.send_request(Request::SetSessionReasoning {
session_id,
effort,
summary,
})
.await?
{
ResponsePayload::SessionModelSet => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn append_presentation_note(
&self,
session_id: SessionId,
source_id: String,
note_id: String,
text: String,
format: bcode_command::CommandTextFormat,
) -> Result<(), ClientError> {
match self
.send_request(Request::AppendPresentationNote {
session_id,
source_id,
note_id,
text,
format,
})
.await?
{
ResponsePayload::PresentationNoteAppended => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_model_status(
&self,
session_id: SessionId,
) -> Result<bcode_model::SessionModelStatus, ClientError> {
match self
.send_request(Request::SessionModelStatus { session_id })
.await?
{
ResponsePayload::SessionModelStatus { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn default_model_status(
&self,
) -> Result<bcode_model::SessionModelStatus, ClientError> {
match self.send_request(Request::DefaultModelStatus).await? {
ResponsePayload::SessionModelStatus { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_model_list(
&self,
provider_plugin_id: Option<String>,
) -> Result<bcode_model::ModelList, ClientError> {
match self
.send_request(Request::SessionModelList { provider_plugin_id })
.await?
{
ResponsePayload::SessionModelList { models, .. } => Ok(models),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn check_models(
&self,
session_id: Option<SessionId>,
provider_plugin_id: Option<String>,
model_ids: Vec<String>,
timeout_seconds: u64,
allow_provider_requests: bool,
) -> Result<bcode_model::ModelCheckReport, ClientError> {
self.check_models_with_id(
SessionId::new().to_string(),
session_id,
provider_plugin_id,
model_ids,
timeout_seconds,
allow_provider_requests,
)
.await
}
pub async fn check_models_with_id(
&self,
check_id: String,
session_id: Option<SessionId>,
provider_plugin_id: Option<String>,
model_ids: Vec<String>,
timeout_seconds: u64,
allow_provider_requests: bool,
) -> Result<bcode_model::ModelCheckReport, ClientError> {
if check_id.is_empty()
|| check_id.len() > 128
|| !check_id
.bytes()
.all(|byte| byte.is_ascii_alphanumeric() || matches!(byte, b'-' | b'_' | b'.'))
{
return Err(ClientError::UnexpectedResponse);
}
if !allow_provider_requests
|| model_ids.is_empty()
|| model_ids.len() > 16
|| !(1..=60).contains(&timeout_seconds)
|| model_ids
.iter()
.any(|id| id.trim().is_empty() || id.len() > 256)
{
return Err(ClientError::UnexpectedResponse);
}
match self
.send_request(Request::CheckModels {
check_id,
session_id,
provider_plugin_id,
model_ids,
timeout_seconds,
allow_provider_requests,
})
.await?
{
ResponsePayload::CheckModels { result_json } => {
serde_json::from_str(&result_json).map_err(|_| ClientError::UnexpectedResponse)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn cancel_model_check(&self, check_id: String) -> Result<bool, ClientError> {
match self
.send_request(Request::CancelModelCheck { check_id })
.await?
{
ResponsePayload::ModelCheckCancellationRequested {
cancellation_requested,
} => Ok(cancellation_requested),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_model_list_for_session(
&self,
session_id: SessionId,
provider_plugin_id: Option<String>,
) -> Result<bcode_model::ModelList, ClientError> {
match self
.send_request(Request::SessionModelListForSession {
session_id,
provider_plugin_id,
})
.await?
{
ResponsePayload::SessionModelList { models, .. } => Ok(models),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn turn_judgement_observation(
&self,
session_id: SessionId,
turn_id: &str,
) -> Result<Option<bcode_model::judgement::TurnJudgementObservation>, ClientError> {
if turn_id.is_empty() || turn_id.len() > 128 {
return Err(ClientError::UnexpectedResponse);
}
match self
.send_request(Request::TurnJudgementObservation {
session_id,
turn_id: turn_id.into(),
})
.await?
{
ResponsePayload::TurnJudgementObservation {
observation_json: Some(json),
} if json.len() <= bcode_model::judgement::MAX_TURN_OBSERVATION_BYTES => {
let observation: bcode_model::judgement::TurnJudgementObservation =
serde_json::from_str(&json).map_err(|_| ClientError::UnexpectedResponse)?;
observation
.validate()
.map_err(|_| ClientError::UnexpectedResponse)?;
if observation.session_id != session_id.to_string()
|| observation.turn_id != turn_id
{
return Err(ClientError::UnexpectedResponse);
}
Ok(Some(observation))
}
ResponsePayload::TurnJudgementObservation {
observation_json: None,
} => Ok(None),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_categorization(
&self,
session_id: SessionId,
) -> Result<Option<bcode_session_models::SessionCategorization>, ClientError> {
match self
.send_request(Request::SessionCategorization { session_id })
.await?
{
ResponsePayload::SessionCategorization { categorization } => Ok(categorization),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn rank_session_categories(
&self,
request: &bcode_session_models::SessionCategoryRankRequest,
) -> Result<bcode_session_models::SessionCategoryRankResponse, ClientError> {
let payload = serde_json::to_vec(request).map_err(|_| ClientError::UnexpectedResponse)?;
let response = self
.call_plugin_service(
bcode_session_models::CATEGORIZATION_INTERFACE_ID.to_owned(),
bcode_session_models::OP_RANK_CATEGORIZATION.to_owned(),
payload,
)
.await?;
if response.error.is_some() {
return Err(ClientError::UnexpectedResponse);
}
let ranking: bcode_session_models::SessionCategoryRankResponse =
serde_json::from_slice(&response.payload)
.map_err(|_| ClientError::UnexpectedResponse)?;
ranking
.validate_for(request)
.map_err(|_| ClientError::UnexpectedResponse)?;
Ok(ranking)
}
pub async fn optional_judge(
&self,
request: &bcode_model::judgement::OptionalRequest,
) -> Result<bcode_model::judgement::OptionalResponse, ClientError> {
let request_json =
serde_json::to_string(request).map_err(|_| ClientError::UnexpectedResponse)?;
if request_json.len() > bcode_model::judgement::MAX_REQUEST_BYTES {
return Err(ClientError::UnexpectedResponse);
}
match self
.send_request(Request::OptionalJudge { request_json })
.await?
{
ResponsePayload::OptionalJudgement { result_json }
if result_json.len() <= bcode_model::judgement::MAX_RESPONSE_BYTES =>
{
let response: bcode_model::judgement::OptionalResponse =
serde_json::from_str(&result_json)
.map_err(|_| ClientError::UnexpectedResponse)?;
Ok(response)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn select_optional_context(
&self,
request: &bcode_session_models::ContextSelectionRequest,
) -> Result<bcode_session_models::ContextSelectionResponse, ClientError> {
let request_json =
serde_json::to_string(request).map_err(|_| ClientError::UnexpectedResponse)?;
if request_json.len() > bcode_model::judgement::MAX_REQUEST_BYTES {
return Err(ClientError::UnexpectedResponse);
}
match self
.send_request(Request::SelectOptionalContext { request_json })
.await?
{
ResponsePayload::OptionalContextSelection { result_json }
if result_json.len() <= bcode_model::judgement::MAX_RESPONSE_BYTES =>
{
let response: bcode_session_models::ContextSelectionResponse =
serde_json::from_str(&result_json)
.map_err(|_| ClientError::UnexpectedResponse)?;
let original: std::collections::BTreeSet<_> = request
.candidates
.iter()
.map(|candidate| &candidate.id)
.collect();
let ordered: std::collections::BTreeSet<_> = response.ordered_ids.iter().collect();
let matching_order_length = response.ordered_ids.len() == request.candidates.len();
if original != ordered
|| !matching_order_length
|| request
.candidates
.iter()
.zip(&response.ordered_ids)
.any(|(candidate, id)| candidate.protected && candidate.id != *id)
|| response.included_ids.iter().any(|id| !ordered.contains(id))
|| response
.included_ids
.iter()
.collect::<std::collections::BTreeSet<_>>()
.len()
!= response.included_ids.len()
|| request.candidates.iter().any(|candidate| {
candidate.protected && !response.included_ids.contains(&candidate.id)
})
{
return Err(ClientError::UnexpectedResponse);
}
Ok(response)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn rank_optional_context(
&self,
task: &str,
candidates: &[bcode_session_models::JudgementContextCandidate],
) -> Result<Option<Vec<String>>, ClientError> {
let response = self
.select_optional_context(&bcode_session_models::ContextSelectionRequest {
task: task.into(),
candidates: candidates.to_vec(),
budget_bytes: None,
})
.await?;
Ok(response.applied.then_some(response.ordered_ids))
}
pub async fn verify_judgement(
&self,
selection: &bcode_model::judgement::SetupSelection,
) -> Result<bcode_model::judgement::VerificationResult, ClientError> {
let selection_json =
serde_json::to_string(selection).map_err(|_| ClientError::UnexpectedResponse)?;
if selection_json.len() > 1024 {
return Err(ClientError::UnexpectedResponse);
}
match self
.send_request(Request::JudgementVerify { selection_json })
.await?
{
ResponsePayload::JudgementVerified { result_json } if result_json.len() <= 4096 => {
serde_json::from_str(&result_json).map_err(|_| ClientError::UnexpectedResponse)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn judgement_setup(
&self,
) -> Result<bcode_model::judgement::SetupCatalog, ClientError> {
match self.send_request(Request::JudgementSetup).await? {
ResponsePayload::JudgementSetup { catalog_json } if catalog_json.len() <= 262_144 => {
serde_json::from_str(&catalog_json).map_err(|_| ClientError::UnexpectedResponse)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn validate_judgement_setup(
&self,
selection: &bcode_model::judgement::SetupSelection,
) -> Result<bool, ClientError> {
let selection_json =
serde_json::to_string(selection).map_err(|_| ClientError::UnexpectedResponse)?;
if selection_json.len() > 1024 {
return Err(ClientError::UnexpectedResponse);
}
match self
.send_request(Request::JudgementSetupValidate { selection_json })
.await?
{
ResponsePayload::JudgementSetupValidated { available } => Ok(available),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn judge(
&self,
provider_plugin_id: String,
auth_profile: String,
request: bcode_model::judgement::Request,
) -> Result<bcode_model::judgement::Response, ClientError> {
bcode_model::judgement::validate_request(&request)
.map_err(|_| ClientError::UnexpectedResponse)?;
let request_json =
serde_json::to_string(&request).map_err(|_| ClientError::UnexpectedResponse)?;
if request_json.len() > bcode_model::judgement::MAX_REQUEST_BYTES {
return Err(ClientError::UnexpectedResponse);
}
match self
.send_request(Request::Judge {
provider_plugin_id,
auth_profile,
request_json,
})
.await?
{
ResponsePayload::Judgement { result_json } => {
if result_json.len() > bcode_model::judgement::MAX_RESPONSE_BYTES {
return Err(ClientError::UnexpectedResponse);
}
serde_json::from_str(&result_json).map_err(|_| ClientError::UnexpectedResponse)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn model_available_for_setup(
&self,
provider_plugin_id: String,
model_id: &str,
) -> Result<bool, ClientError> {
let models = self.session_model_list(Some(provider_plugin_id)).await?;
Ok(models.models.iter().any(|model| {
model.model_id == model_id
&& matches!(model.visibility, bcode_model::ModelVisibility::Visible)
}))
}
pub async fn cancel_session_turn(&self, session_id: SessionId) -> Result<bool, ClientError> {
self.cancel_session_turn_with_options(session_id, false)
.await
}
pub async fn cancel_session_turn_with_options(
&self,
session_id: SessionId,
clear_queue: bool,
) -> Result<bool, ClientError> {
match self
.send_request(Request::CancelSessionTurn {
session_id,
expected_turn_id: None,
clear_queue,
})
.await?
{
ResponsePayload::TurnCancellationRequested { cancelled } => Ok(cancelled),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn cancel_exact_session_turn(
&self,
session_id: SessionId,
turn_id: String,
) -> Result<bool, ClientError> {
match self
.send_request(Request::CancelSessionTurn {
session_id,
expected_turn_id: Some(turn_id),
clear_queue: false,
})
.await?
{
ResponsePayload::TurnCancellationRequested { cancelled } => Ok(cancelled),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn create_authored_workflow(
&self,
request: bcode_workflow::CreateAuthoredWorkflowRequest,
) -> Result<
(
bcode_workflow::AuthoredWorkflowSnapshot,
bcode_workflow::WorkflowDraftSnapshot,
),
ClientError,
> {
match self
.send_request(Request::CreateAuthoredWorkflow(request))
.await?
{
ResponsePayload::AuthoredWorkflowCreated { workflow, draft } => Ok((workflow, *draft)),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn apply_workflow_source(
&self,
source_format: bcode_workflow::WorkflowSourceFormat,
source: String,
draft_id: String,
) -> Result<bcode_workflow::WorkflowSourceApplyResult, ClientError> {
match self
.send_request(Request::ApplyWorkflowSource(
bcode_workflow::ApplyWorkflowSourceRequest {
source_format,
source,
draft_id,
},
))
.await?
{
ResponsePayload::WorkflowSourceApplied { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn apply_workflow_draft_edits(
&self,
request: bcode_workflow::ApplyWorkflowDraftEditsRequest,
) -> Result<bcode_workflow::WorkflowDraftEditResult, ClientError> {
match self
.send_request(Request::ApplyWorkflowDraftEdits(request))
.await?
{
ResponsePayload::WorkflowDraftEditResult { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn update_workflow_draft(
&self,
request: bcode_workflow::UpdateWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowDraftUpdateResult, ClientError> {
match self
.send_request(Request::UpdateWorkflowDraft(request))
.await?
{
ResponsePayload::WorkflowDraftUpdateResult { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn publish_workflow_draft(
&self,
request: bcode_workflow::PublishWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowPublicationResult, ClientError> {
match self
.send_request(Request::PublishWorkflowDraft(request))
.await?
{
ResponsePayload::WorkflowPublicationResult { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn publish_and_start_workflow(
&self,
request: bcode_workflow::PublishAndStartWorkflowRequest,
) -> Result<bcode_workflow::WorkflowPublishAndStartResult, ClientError> {
match self
.send_request(Request::PublishAndStartWorkflow(Box::new(request)))
.await?
{
ResponsePayload::WorkflowPublishAndStartResult { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn activate_workflow_revision(
&self,
request: bcode_workflow::ActivateWorkflowRevisionRequest,
) -> Result<bcode_workflow::WorkflowAuthoringMutationResult, ClientError> {
match self
.send_request(Request::ActivateWorkflowRevision(request))
.await?
{
ResponsePayload::WorkflowActivationResult { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn set_authored_workflow_archived(
&self,
request: bcode_workflow::SetAuthoredWorkflowArchivedRequest,
) -> Result<bcode_workflow::AuthoredWorkflowSnapshot, ClientError> {
match self
.send_request(Request::SetAuthoredWorkflowArchived(request))
.await?
{
ResponsePayload::AuthoredWorkflowArchived { workflow } => Ok(workflow),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn discard_workflow_draft(
&self,
request: bcode_workflow::DiscardWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowAuthoringMutationResult, ClientError> {
match self
.send_request(Request::DiscardWorkflowDraft(request))
.await?
{
ResponsePayload::WorkflowDraftDiscardResult { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn fork_workflow_draft(
&self,
request: bcode_workflow::ForkWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowDraftSnapshot, ClientError> {
match self
.send_request(Request::ForkWorkflowDraft(request))
.await?
{
ResponsePayload::WorkflowDraftForked { draft } => Ok(*draft),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn create_workflow_preset(
&self,
request: bcode_workflow::CreateWorkflowPresetRequest,
) -> Result<bcode_workflow::WorkflowPresetSnapshot, ClientError> {
match self
.send_request(Request::CreateWorkflowPreset(request))
.await?
{
ResponsePayload::WorkflowPresetCreated { preset } => Ok(preset),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn update_workflow_preset(
&self,
request: bcode_workflow::UpdateWorkflowPresetRequest,
) -> Result<bcode_workflow::WorkflowPresetUpdateResult, ClientError> {
match self
.send_request(Request::UpdateWorkflowPreset(request))
.await?
{
ResponsePayload::WorkflowPresetUpdateResult { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn delete_workflow_preset(
&self,
request: bcode_workflow::DeleteWorkflowPresetRequest,
) -> Result<bcode_workflow::WorkflowAuthoringMutationResult, ClientError> {
match self
.send_request(Request::DeleteWorkflowPreset(request))
.await?
{
ResponsePayload::WorkflowPresetDeleteResult { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn export_workflow_revision(
&self,
request: bcode_workflow::ExportWorkflowRevisionRequest,
) -> Result<bcode_workflow::WorkflowExportBundle, ClientError> {
match self
.send_request(Request::ExportWorkflowRevision(request))
.await?
{
ResponsePayload::WorkflowRevisionExported { bundle } => Ok(*bundle),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn preview_workflow_import(
&self,
request: bcode_workflow::PreviewWorkflowImportRequest,
) -> Result<bcode_workflow::WorkflowImportPreview, ClientError> {
match self
.send_request(Request::PreviewWorkflowImport(request))
.await?
{
ResponsePayload::WorkflowImportPreview { preview } => Ok(*preview),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn import_workflow(
&self,
request: bcode_workflow::ImportWorkflowRequest,
) -> Result<
(
bcode_workflow::AuthoredWorkflowSnapshot,
bcode_workflow::WorkflowDraftSnapshot,
),
ClientError,
> {
match self.send_request(Request::ImportWorkflow(request)).await? {
ResponsePayload::WorkflowImported { workflow, draft } => Ok((workflow, *draft)),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn import_workflow_draft(
&self,
request: bcode_workflow::ImportWorkflowDraftRequest,
) -> Result<bcode_workflow::WorkflowDraftImportResult, ClientError> {
match self
.send_request(Request::ImportWorkflowDraft(request))
.await?
{
ResponsePayload::WorkflowDraftImported { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn import_workflow_revision(
&self,
request: bcode_workflow::ImportWorkflowRevisionRequest,
) -> Result<bcode_workflow::WorkflowRevisionImportResult, ClientError> {
match self
.send_request(Request::ImportWorkflowRevision(request))
.await?
{
ResponsePayload::WorkflowRevisionImported { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn start_authored_workflow(
&self,
request: bcode_workflow::StartAuthoredWorkflowRequest,
) -> Result<bcode_workflow::AuthoredWorkflowRunStartResponse, ClientError> {
match self
.send_request(Request::StartAuthoredWorkflow(request))
.await?
{
ResponsePayload::AuthoredWorkflowRunStarted(started) => Ok(started),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_authored_workflows(
&self,
cursor: Option<bcode_workflow::WorkflowAuthoringListCursor>,
limit: usize,
) -> Result<
bcode_workflow::WorkflowAuthoringPage<
bcode_workflow::AuthoredWorkflowSnapshot,
bcode_workflow::WorkflowAuthoringListCursor,
>,
ClientError,
> {
match self
.send_request(Request::ListAuthoredWorkflows { cursor, limit })
.await?
{
ResponsePayload::AuthoredWorkflowList { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn authored_workflow(
&self,
workflow_id: String,
) -> Result<Option<bcode_workflow::AuthoredWorkflowSnapshot>, ClientError> {
match self
.send_request(Request::GetAuthoredWorkflow { workflow_id })
.await?
{
ResponsePayload::AuthoredWorkflowDescription { workflow } => Ok(workflow),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn inspect_authored_workflow(
&self,
workflow_id: String,
limit: usize,
) -> Result<Option<bcode_workflow::AuthoredWorkflowInspection>, ClientError> {
match self
.send_request(Request::InspectAuthoredWorkflow { workflow_id, limit })
.await?
{
ResponsePayload::AuthoredWorkflowInspection { inspection } => {
Ok(inspection.map(|inspection| *inspection))
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_workflow_drafts(
&self,
workflow_id: String,
cursor: Option<bcode_workflow::WorkflowAuthoringListCursor>,
limit: usize,
) -> Result<
bcode_workflow::WorkflowAuthoringPage<
bcode_workflow::WorkflowDraftSnapshot,
bcode_workflow::WorkflowAuthoringListCursor,
>,
ClientError,
> {
match self
.send_request(Request::ListWorkflowDrafts {
workflow_id,
cursor,
limit,
})
.await?
{
ResponsePayload::WorkflowDraftList { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_draft(
&self,
workflow_id: String,
draft_id: String,
) -> Result<Option<bcode_workflow::WorkflowDraftSnapshot>, ClientError> {
match self
.send_request(Request::GetWorkflowDraft {
workflow_id,
draft_id,
})
.await?
{
ResponsePayload::WorkflowDraftDescription { draft } => Ok(draft.map(|draft| *draft)),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_workflow_revisions(
&self,
workflow_id: String,
cursor: Option<bcode_workflow::WorkflowRevisionListCursor>,
limit: usize,
) -> Result<
bcode_workflow::WorkflowAuthoringPage<
bcode_workflow::WorkflowRevisionSnapshot,
bcode_workflow::WorkflowRevisionListCursor,
>,
ClientError,
> {
match self
.send_request(Request::ListWorkflowRevisions {
workflow_id,
cursor,
limit,
})
.await?
{
ResponsePayload::WorkflowRevisionList { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_revision(
&self,
workflow_id: String,
revision: u64,
) -> Result<Option<bcode_workflow::WorkflowRevisionSnapshot>, ClientError> {
match self
.send_request(Request::GetWorkflowRevision {
workflow_id,
revision,
})
.await?
{
ResponsePayload::WorkflowRevisionDescription { revision } => {
Ok(revision.map(|revision| *revision))
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_revision_requirement_inspection(
&self,
workflow_id: String,
revision: u64,
) -> Result<Option<bcode_workflow::WorkflowRevisionRequirementInspection>, ClientError> {
match self
.send_request(Request::InspectWorkflowRevisionRequirements {
workflow_id,
revision,
})
.await?
{
ResponsePayload::WorkflowRevisionRequirementInspection { inspection } => Ok(inspection),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_workflow_presets(
&self,
workflow_id: String,
cursor: Option<bcode_workflow::WorkflowAuthoringListCursor>,
limit: usize,
) -> Result<
bcode_workflow::WorkflowAuthoringPage<
bcode_workflow::WorkflowPresetSnapshot,
bcode_workflow::WorkflowAuthoringListCursor,
>,
ClientError,
> {
match self
.send_request(Request::ListWorkflowPresets {
workflow_id,
cursor,
limit,
})
.await?
{
ResponsePayload::WorkflowPresetList { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_preset(
&self,
workflow_id: String,
preset_id: String,
) -> Result<Option<bcode_workflow::WorkflowPresetSnapshot>, ClientError> {
match self
.send_request(Request::GetWorkflowPreset {
workflow_id,
preset_id,
})
.await?
{
ResponsePayload::WorkflowPresetDescription { preset } => Ok(preset),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_authoring_catalog(
&self,
) -> Result<bcode_workflow::WorkflowAuthoringCatalogSnapshot, ClientError> {
match self.send_request(Request::WorkflowAuthoringCatalog).await? {
ResponsePayload::WorkflowAuthoringCatalog { catalog } => Ok(catalog),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_launch_catalog(
&self,
request: bcode_workflow::WorkflowLaunchCatalogRequest,
) -> Result<bcode_workflow::WorkflowLaunchCatalogPage, ClientError> {
match self
.send_request(Request::WorkflowLaunchCatalog(request))
.await?
{
ResponsePayload::WorkflowLaunchCatalog { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_launch_catalog_until<F>(
&self,
mut request: bcode_workflow::WorkflowLaunchCatalogRequest,
interrupted: F,
) -> Result<Option<bcode_workflow::WorkflowLaunchCatalogPage>, ClientError>
where
F: std::future::Future<Output = ()> + Send,
{
use bcode_workflow::WorkflowAuthoringApplication;
tokio::pin!(interrupted);
if let Some(token) = request
.cursor
.as_ref()
.and_then(|cursor| cursor.result_token.clone())
{
let started = std::sync::atomic::AtomicBool::new(false);
let page_request = async {
started.store(true, std::sync::atomic::Ordering::Relaxed);
self.workflow_launch_catalog(request).await
};
tokio::pin!(page_request);
let result = tokio::select! {
biased;
() = &mut interrupted => {
if !started.load(std::sync::atomic::Ordering::Relaxed) {
self.cancel_workflow_discovery(token).await?;
return Ok(None);
}
let cancellation = self.cancel_workflow_discovery(token).await;
let page = page_request.await;
if let Ok(page) = &page
&& let Some(successor) = page.detail_token.clone().or_else(|| page.next_cursor.as_ref().and_then(|cursor| cursor.result_token.clone()))
{
self.cancel_workflow_discovery(successor).await?;
}
if !cancellation? {
page?;
}
return Ok(None);
}
result = &mut page_request => result?,
};
return Ok(Some(result));
}
if request.discovery_token.is_none() {
let mut admission = request.clone();
admission.incremental = true;
let admission = self.workflow_launch_catalog(admission);
tokio::pin!(admission);
let (page, cancelled) = tokio::select! {
biased;
() = &mut interrupted => (admission.await?, true),
page = &mut admission => (page?, false),
};
let token = page
.discovery_token
.clone()
.ok_or(ClientError::UnexpectedResponse)?;
if cancelled {
self.cancel_workflow_discovery(token).await?;
return Ok(None);
}
if request.incremental {
return Ok(Some(page));
}
request.discovery_token = Some(token);
}
let token = request
.discovery_token
.clone()
.ok_or(ClientError::UnexpectedResponse)?;
let started = std::sync::atomic::AtomicBool::new(false);
let page_request = async {
started.store(true, std::sync::atomic::Ordering::Relaxed);
self.workflow_launch_catalog(request).await
};
tokio::pin!(page_request);
tokio::select! {
biased;
() = &mut interrupted => {
if !started.load(std::sync::atomic::Ordering::Relaxed) {
self.cancel_workflow_discovery(token).await?;
return Ok(None);
}
let cancellation = self.cancel_workflow_discovery(token).await;
let page = page_request.await;
if let Ok(page) = &page {
let successor = page.discovery_token.clone().or_else(|| page.detail_token.clone()).or_else(|| page.next_cursor.as_ref().and_then(|cursor| cursor.result_token.clone()));
if let Some(successor) = successor {
self.cancel_workflow_discovery(successor).await?;
}
}
if !cancellation? {
page?;
}
Ok(None)
}
result = &mut page_request => result.map(Some),
}
}
pub async fn workflow_launch_detail(
&self,
request: bcode_workflow::WorkflowLaunchDetailRequest,
) -> Result<bcode_workflow::WorkflowLaunchDetail, ClientError> {
match self
.send_request(Request::WorkflowLaunchDetail(request))
.await?
{
ResponsePayload::WorkflowLaunchDetail { detail } => Ok(*detail),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_launch_detail_until<F>(
&self,
mut request: bcode_workflow::WorkflowLaunchDetailRequest,
interrupted: F,
) -> Result<Option<bcode_workflow::WorkflowLaunchDetail>, ClientError>
where
F: std::future::Future<Output = ()> + Send,
{
use bcode_workflow::WorkflowAuthoringApplication as _;
tokio::pin!(interrupted);
if request.catalog_token.is_none()
&& !matches!(
request.source,
bcode_workflow::WorkflowLaunchSourceIdentity::ExplicitSource { .. }
)
{
let mut query = bcode_workflow::WorkflowLaunchCatalogRequest {
version: request.version,
workspace: request.workspace.clone(),
incremental: true,
retain_for_detail: true,
discovery_token: None,
limit: 1,
cursor: None,
search: None,
source_kind: None,
readiness: None,
};
loop {
let Some(page) =
Box::pin(self.workflow_launch_catalog_until(query.clone(), &mut interrupted))
.await?
else {
return Ok(None);
};
if let Some(token) = page.discovery_token {
query.discovery_token = Some(token);
} else {
request.catalog_token =
Some(page.detail_token.ok_or(ClientError::UnexpectedResponse)?);
break;
}
}
}
let token = request.catalog_token.clone();
let detail = self.workflow_launch_detail(request);
tokio::pin!(detail);
tokio::select! {
biased;
() = &mut interrupted => {
if let Some(token) = token {
self.cancel_workflow_discovery(token).await?;
}
Ok(None)
}
result = &mut detail => result.map(Some),
}
}
pub async fn workflow_package_publication(
&self,
package_id: String,
) -> Result<Option<bcode_workflow::WorkflowPackagePublicationReceipt>, ClientError> {
match self
.send_request(Request::GetWorkflowPackagePublication { package_id })
.await?
{
ResponsePayload::WorkflowPackagePublication { receipt } => Ok(receipt),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn apply_workflow_package(
&self,
request: bcode_workflow::ApplyWorkflowPackageRequest,
) -> Result<bcode_workflow::WorkflowPackageMutationResult, ClientError> {
match self
.send_request(Request::ApplyWorkflowPackage(request))
.await?
{
ResponsePayload::WorkflowPackageApplied { result } => Ok(*result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn publish_workflow_package(
&self,
request: bcode_workflow::PublishWorkflowPackageRequest,
) -> Result<bcode_workflow::WorkflowPackageMutationResult, ClientError> {
match self
.send_request(Request::PublishWorkflowPackage(request))
.await?
{
ResponsePayload::WorkflowPackagePublished { result } => Ok(*result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn validate_workflow_package(
&self,
request: bcode_workflow::WorkflowPackageComputationRequest,
) -> Result<bcode_workflow::WorkflowPackageValidationResult, ClientError> {
match self
.send_request(Request::ValidateWorkflowPackage(request))
.await?
{
ResponsePayload::WorkflowPackageValidated { result } => Ok(*result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn preview_workflow_package(
&self,
request: bcode_workflow::WorkflowPackagePreviewRequest,
) -> Result<bcode_workflow::WorkflowPackagePreview, ClientError> {
match self
.send_request(Request::PreviewWorkflowPackage(request))
.await?
{
ResponsePayload::WorkflowPackagePreviewed { preview } => Ok(*preview),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn validate_workflow_source(
&self,
request: bcode_workflow::WorkflowSourceComputationRequest,
) -> Result<bcode_workflow::WorkflowSourceValidationResult, ClientError> {
match self
.send_request(Request::ValidateWorkflowSource(request))
.await?
{
ResponsePayload::WorkflowSourceValidated { result } => Ok(*result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn preview_workflow_source(
&self,
request: bcode_workflow::WorkflowSourcePreviewRequest,
) -> Result<bcode_workflow::WorkflowSourcePreviewResult, ClientError> {
match self
.send_request(Request::PreviewWorkflowSource(request))
.await?
{
ResponsePayload::WorkflowSourcePreviewed { result } => Ok(*result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn validate_workflow_authoring(
&self,
document: bcode_workflow::WorkflowAuthoringDocument,
) -> Result<bcode_workflow::WorkflowValidationReport, ClientError> {
self.validate_workflow_authoring_with_control(
document,
bcode_workflow::WorkflowComputationControl::default(),
)
.await
}
pub async fn validate_workflow_authoring_with_control(
&self,
document: bcode_workflow::WorkflowAuthoringDocument,
control: bcode_workflow::WorkflowComputationControl,
) -> Result<bcode_workflow::WorkflowValidationReport, ClientError> {
match self
.send_request(Request::ValidateWorkflowAuthoring { document, control })
.await?
{
ResponsePayload::WorkflowAuthoringValidated { report } => Ok(report),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn preview_workflow_compilation(
&self,
document: bcode_workflow::WorkflowAuthoringDocument,
configuration: Option<serde_json::Value>,
) -> Result<bcode_workflow::WorkflowCompilationPreview, ClientError> {
self.preview_workflow_compilation_with_control(
document,
configuration,
bcode_workflow::WorkflowComputationControl::default(),
)
.await
}
pub async fn preview_workflow_compilation_with_control(
&self,
document: bcode_workflow::WorkflowAuthoringDocument,
configuration: Option<serde_json::Value>,
control: bcode_workflow::WorkflowComputationControl,
) -> Result<bcode_workflow::WorkflowCompilationPreview, ClientError> {
match self
.send_request(Request::PreviewWorkflowCompilation {
document,
configuration,
control,
})
.await?
{
ResponsePayload::WorkflowCompilationPreview { preview } => Ok(*preview),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn cancel_workflow_computation(
&self,
operation_id: String,
) -> Result<bool, ClientError> {
match self
.send_request(Request::CancelWorkflowComputation { operation_id })
.await?
{
ResponsePayload::WorkflowComputationCancellationRequested { cancelled } => {
Ok(cancelled)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn reset_incompatible_workflow_store(
&self,
confirm: String,
) -> Result<bcode_workflow_store::WorkflowStoreResetReceipt, ClientError> {
match self
.send_request(Request::ResetIncompatibleWorkflowStore { confirm })
.await?
{
ResponsePayload::WorkflowStoreReset { receipt } => Ok(receipt),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_workflow_templates(
&self,
limit: usize,
) -> Result<Vec<bcode_ipc::WorkflowTemplateDescription>, ClientError> {
match self
.send_request(Request::ListWorkflowTemplates { limit })
.await?
{
ResponsePayload::WorkflowTemplateList { templates } => Ok(templates),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn describe_workflow_template(
&self,
owner_plugin_id: String,
template_id: String,
template_version: u32,
) -> Result<Option<bcode_ipc::WorkflowTemplateDescription>, ClientError> {
match self
.send_request(Request::DescribeWorkflowTemplate {
owner_plugin_id,
template_id,
template_version,
})
.await?
{
ResponsePayload::WorkflowTemplateDescription { template } => {
Ok(template.map(|template| *template))
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn start_workflow_template(
&self,
request: bcode_workflow::WorkflowTemplateStartRequest,
) -> Result<bcode_workflow::WorkflowRunStartResponse, ClientError> {
match self
.send_request(Request::StartWorkflowTemplate(request))
.await?
{
ResponsePayload::WorkflowTemplateStarted(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn instantiate_workflow_template(
&self,
request: bcode_workflow::WorkflowTemplateInstantiationRequest,
) -> Result<
(
bcode_workflow::AuthoredWorkflowSnapshot,
bcode_workflow::WorkflowDraftSnapshot,
),
ClientError,
> {
match self
.send_request(Request::InstantiateWorkflowTemplate(request))
.await?
{
ResponsePayload::AuthoredWorkflowCreated { workflow, draft } => Ok((workflow, *draft)),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn register_workflow_definition(
&self,
request: bcode_workflow::WorkflowDefinitionRegistrationRequest,
) -> Result<bcode_workflow::StoredWorkflowDefinition, ClientError> {
match self
.send_request(Request::RegisterWorkflowDefinition(request))
.await?
{
ResponsePayload::WorkflowDefinitionRegistered { definition } => Ok(definition),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn start_workflow(
&self,
request: bcode_workflow::WorkflowStartRequest,
) -> Result<bcode_workflow::WorkflowRunStartResponse, ClientError> {
match self.send_request(Request::StartWorkflow(request)).await? {
ResponsePayload::WorkflowRunStarted(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn start_workflow_package_export(
&self,
request: bcode_workflow::StartWorkflowPackageExportRequest,
) -> Result<bcode_workflow::WorkflowPackageExportRunStartResponse, ClientError> {
match self
.send_request(Request::StartWorkflowPackageExport(request))
.await?
{
ResponsePayload::WorkflowPackageExportRunStarted(response) => Ok(*response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_continuation_source(
&self,
run_id: String,
) -> Result<bcode_workflow::WorkflowContinuationSource, ClientError> {
match self
.send_request(Request::WorkflowContinuationSource { run_id })
.await?
{
ResponsePayload::WorkflowContinuationSource(source) => Ok(*source),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn continue_workflow(
&self,
request: bcode_workflow::WorkflowContinuationRequest,
) -> Result<bcode_workflow::WorkflowRunStartResponse, ClientError> {
match self
.send_request(Request::ContinueWorkflow(request))
.await?
{
ResponsePayload::WorkflowRunStarted(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn request_workflow_replacement(
&self,
request: bcode_workflow::WorkflowReplacementRequest,
) -> Result<bcode_workflow::WorkflowReplacementResponse, ClientError> {
match self
.send_request(Request::RequestWorkflowReplacement(request))
.await?
{
ResponsePayload::WorkflowReplacementRequested(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn start_workflow_run(
&self,
request: bcode_workflow::WorkflowRunStartRequest,
) -> Result<bcode_workflow::WorkflowRunStartResponse, ClientError> {
match self
.send_request(Request::StartWorkflowRun(request))
.await?
{
ResponsePayload::WorkflowRunStarted(response) => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn doctor_workflow_run(
&self,
run_id: String,
limit: usize,
) -> Result<bcode_workflow::WorkflowDoctorReport, ClientError> {
match self
.send_request(Request::DoctorWorkflowRun { run_id, limit })
.await?
{
ResponsePayload::WorkflowDoctorReport { report } => Ok(report),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn repair_workflow_attempt(
&self,
dispatch_identity: String,
resolution: bcode_workflow::RepairResolution,
) -> Result<bcode_workflow::RepairResult, ClientError> {
match self
.send_request(Request::RepairWorkflowAttempt {
dispatch_identity,
resolution,
})
.await?
{
ResponsePayload::WorkflowAttemptRepaired { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_workflow_definitions(
&self,
limit: usize,
) -> Result<Vec<bcode_workflow::StoredWorkflowDefinition>, ClientError> {
match self
.send_request(Request::ListWorkflowDefinitions { limit })
.await?
{
ResponsePayload::WorkflowDefinitionList { definitions } => Ok(definitions),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn describe_workflow_definition(
&self,
definition_id: String,
version: u32,
) -> Result<Option<bcode_workflow::StoredWorkflowDefinition>, ClientError> {
match self
.send_request(Request::DescribeWorkflowDefinition {
definition_id,
version,
})
.await?
{
ResponsePayload::WorkflowDefinitionDescription { definition } => Ok(definition),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn stage_workflow_run_graph_edit(
&self,
request: bcode_workflow::WorkflowRunGraphEditBatch,
) -> Result<bool, ClientError> {
match self
.send_request(Request::StageWorkflowRunGraphEdit { request })
.await?
{
ResponsePayload::WorkflowRunGraphEditStaged { created } => Ok(created),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_delegation_preflight(
&self,
plugin_id: String,
) -> Result<bcode_workflow::WorkflowDelegationPreflight, ClientError> {
match self
.send_request(Request::WorkflowDelegationPreflight {
plugin_id: plugin_id.clone(),
})
.await?
{
ResponsePayload::WorkflowDelegationPreflight { result }
if result.version == 1 && result.plugin_id == plugin_id =>
{
Ok(result)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn accept_workflow_run_graph_publication(
&self,
request: bcode_workflow::WorkflowRunGraphEditBatch,
) -> Result<bcode_workflow::WorkflowRunGraphPublicationStatus, ClientError> {
match self
.send_request(Request::AcceptWorkflowRunGraphPublication { request })
.await?
{
ResponsePayload::WorkflowRunGraphPublicationAccepted { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn publish_workflow_run_graph_edit(
&self,
request: bcode_workflow::WorkflowRunGraphEditBatch,
) -> Result<u64, ClientError> {
match self
.send_request(Request::PublishWorkflowRunGraphEdit { request })
.await?
{
ResponsePayload::WorkflowRunGraphEditPublished { revision } => Ok(revision),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn inspect_workflow_run_graph(
&self,
request: bcode_workflow::WorkflowRunGraphPageRequest,
) -> Result<bcode_workflow::WorkflowRunGraphInspection, ClientError> {
match self
.send_request(Request::InspectWorkflowRunGraph { request })
.await?
{
ResponsePayload::WorkflowRunGraphInspection { graph } => Ok(graph),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn inspect_workflow_run(
&self,
run_id: String,
limit: usize,
) -> Result<bcode_workflow::WorkflowRunInspection, ClientError> {
match self
.send_request(Request::InspectWorkflowRun { run_id, limit })
.await?
{
ResponsePayload::WorkflowRunInspection { inspection } => Ok(*inspection),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_run_status(
&self,
run_id: String,
) -> Result<Option<bcode_workflow_store::WorkflowRunSummary>, ClientError> {
match self
.send_request(Request::WorkflowRunStatus { run_id })
.await?
{
ResponsePayload::WorkflowRunStatus { run } => Ok(run),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn associated_workflow_run(
&self,
key: bcode_workflow::WorkflowRunBindingLookup,
) -> Result<Option<bcode_workflow_store::WorkflowRunSummary>, ClientError> {
match self
.send_request(Request::AssociatedWorkflowRun { key })
.await?
{
ResponsePayload::AssociatedWorkflowRun { run } => Ok(run),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn inspect_associated_workflow_run(
&self,
key: bcode_workflow::WorkflowRunBindingLookup,
limit: usize,
) -> Result<Option<bcode_workflow::WorkflowRunInspection>, ClientError> {
match self
.send_request(Request::InspectAssociatedWorkflowRun { key, limit })
.await?
{
ResponsePayload::AssociatedWorkflowRunInspection { inspection } => {
Ok(inspection.map(|inspection| *inspection))
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn control_workflow_run(
&self,
run_id: String,
action: bcode_workflow::WorkflowRunControlAction,
) -> Result<(Option<bcode_workflow::WorkflowRunSummary>, bool), ClientError> {
match self
.send_request(Request::ControlWorkflowRun { run_id, action })
.await?
{
ResponsePayload::AssociatedWorkflowRunControlled { run, changed } => Ok((run, changed)),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn control_associated_workflow_run(
&self,
key: bcode_workflow::WorkflowRunBindingLookup,
action: bcode_workflow::WorkflowRunControlAction,
) -> Result<(Option<bcode_workflow_store::WorkflowRunSummary>, bool), ClientError> {
match self
.send_request(Request::ControlAssociatedWorkflowRun { key, action })
.await?
{
ResponsePayload::AssociatedWorkflowRunControlled { run, changed } => Ok((run, changed)),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_run_view(
&self,
run_id: String,
limit: usize,
) -> Result<bcode_workflow_view_models::WorkflowRunView, ClientError> {
match self
.send_request(Request::WorkflowRunView { run_id, limit })
.await?
{
ResponsePayload::WorkflowRunView { view } => Ok(*view),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_identity_catalog(
&self,
request: bcode_workflow_view_models::WorkflowIdentityCatalogRequest,
) -> Result<bcode_workflow_view_models::WorkflowIdentityCatalogPage, ClientError> {
match self
.send_request(Request::WorkflowIdentityCatalog { request })
.await?
{
ResponsePayload::WorkflowIdentityCatalog { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_activity_page(
&self,
request: bcode_workflow_view_models::WorkflowActivityPageRequest,
) -> Result<bcode_workflow_view_models::WorkflowActivityPage, ClientError> {
let expected = request.clone();
match self
.send_request(Request::WorkflowActivityPage { request })
.await?
{
ResponsePayload::WorkflowActivityPage { page }
if page.version == bcode_workflow_view_models::WORKFLOW_ACTIVITY_VERSION
&& page.run_id == expected.run_id
&& page.root_run_id == expected.root_run_id
&& page.section == expected.section
&& expected
.scan_id
.as_ref()
.is_none_or(|id| id == &page.scan_id)
&& expected
.expected_revision
.is_none_or(|revision| revision == page.revision) =>
{
Ok(page)
}
ResponsePayload::WorkflowActivityPage { .. } => Err(ClientError::Protocol(
"unsupported or mismatched workflow activity page".into(),
)),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_activity_close(&self, scan_id: String) -> Result<(), ClientError> {
match self
.send_request(Request::WorkflowActivityClose { scan_id })
.await?
{
ResponsePayload::WorkflowActivityClosed => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_scheduling_page(
&self,
request: bcode_workflow::WorkflowSchedulingPageRequest,
) -> Result<
bcode_workflow::WorkflowSchedulingPage<bcode_workflow_view_models::WorkflowSchedulingView>,
ClientError,
> {
match self
.send_request(Request::WorkflowSchedulingPage { request })
.await?
{
ResponsePayload::WorkflowSchedulingPage { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_catalog_view(
&self,
request: bcode_workflow_view_models::WorkflowCatalogRequest,
) -> Result<bcode_workflow_view_models::WorkflowCatalogView, ClientError> {
match self
.send_request(Request::WorkflowCatalogView { request })
.await?
{
ResponsePayload::WorkflowCatalogView { view } => Ok(view),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_workflow_runs(
&self,
limit: usize,
) -> Result<Vec<bcode_workflow_store::WorkflowRunSummary>, ClientError> {
match self
.send_request(Request::ListWorkflowRuns { limit })
.await?
{
ResponsePayload::WorkflowRunList { runs } => Ok(runs),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_run_outputs(
&self,
run_id: String,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowOutputInspection>, ClientError> {
match self
.send_request(Request::WorkflowRunOutputs { run_id, limit })
.await?
{
ResponsePayload::WorkflowRunOutputs { outputs } => Ok(outputs),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn cancel_workflow_run(&self, run_id: String) -> Result<bool, ClientError> {
match self
.send_request(Request::CancelWorkflowRun { run_id })
.await?
{
ResponsePayload::WorkflowRunCancellationRequested { recorded } => Ok(recorded),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn reconcile_orphaned_workflow_runs(
&self,
apply: bool,
limit: usize,
) -> Result<bcode_workflow::OrphanedWorkflowRunReport, ClientError> {
match self
.send_request(Request::ReconcileOrphanedWorkflowRuns { apply, limit })
.await?
{
ResponsePayload::OrphanedWorkflowRunsReconciled { report } => Ok(report),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn pause_workflow_run(&self, run_id: String) -> Result<bool, ClientError> {
match self
.send_request(Request::PauseWorkflowRun { run_id })
.await?
{
ResponsePayload::WorkflowRunPaused { changed } => Ok(changed),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn resume_workflow_run(&self, run_id: String) -> Result<bool, ClientError> {
match self
.send_request(Request::ResumeWorkflowRun { run_id })
.await?
{
ResponsePayload::WorkflowRunResumed { changed } => Ok(changed),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn retry_workflow_node(
&self,
run_id: String,
node_id: String,
activation_id: String,
failed_attempt: u32,
) -> Result<bcode_workflow_store::WorkflowNodeRetryResult, ClientError> {
match self
.send_request(Request::RetryWorkflowNode {
run_id,
node_id,
activation_id,
failed_attempt,
})
.await?
{
ResponsePayload::WorkflowNodeRetried { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_workflow_waits(
&self,
run_id: String,
limit: usize,
) -> Result<Vec<bcode_workflow::WaitingActivation>, ClientError> {
match self
.send_request(Request::ListWorkflowWaits { run_id, limit })
.await?
{
ResponsePayload::WorkflowWaitList { waits } => Ok(waits),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn provide_workflow_input(
&self,
run_id: String,
node_id: String,
activation_id: String,
value: serde_json::Value,
) -> Result<bcode_workflow_store::WaitingResolutionResult, ClientError> {
match self
.send_request(Request::ProvideWorkflowInput {
run_id,
node_id,
activation_id,
value,
})
.await?
{
ResponsePayload::WorkflowWaitResolved { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn resolve_workflow_approval(
&self,
run_id: String,
node_id: String,
activation_id: String,
approved: bool,
) -> Result<bcode_workflow_store::WaitingResolutionResult, ClientError> {
match self
.send_request(Request::ResolveWorkflowApproval {
run_id,
node_id,
activation_id,
approved,
})
.await?
{
ResponsePayload::WorkflowWaitResolved { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_all_workflow_mutation_approvals(
&self,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowMutationApprovalInspection>, ClientError> {
match self
.send_request(Request::ListWorkflowMutationApprovalsAll { limit })
.await?
{
ResponsePayload::WorkflowMutationApprovalList { approvals } => Ok(approvals),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_workflow_mutation_approvals(
&self,
run_id: String,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowMutationApprovalInspection>, ClientError> {
match self
.send_request(Request::ListWorkflowMutationApprovals { run_id, limit })
.await?
{
ResponsePayload::WorkflowMutationApprovalList { approvals } => Ok(approvals),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn resolve_workflow_mutation_approval(
&self,
approval_id: String,
decision: bcode_workflow::WorkflowMutationApprovalDecision,
) -> Result<bcode_workflow::WorkflowMutationApprovalResolution, ClientError> {
match self
.send_request(Request::ResolveWorkflowMutationApproval {
approval_id,
decision,
})
.await?
{
ResponsePayload::WorkflowMutationApprovalResolved { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_attempt_history(
&self,
run_id: String,
cursor: Option<bcode_workflow_store::AttemptCursor>,
limit: usize,
) -> Result<Vec<bcode_workflow::AttemptSummary>, ClientError> {
match self
.send_request(Request::WorkflowAttemptHistory {
run_id,
cursor,
limit,
})
.await?
{
ResponsePayload::WorkflowAttemptHistory { attempts } => Ok(attempts),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_event_history(
&self,
run_id: String,
after_sequence: Option<u64>,
limit: usize,
) -> Result<Vec<bcode_workflow::WorkflowHistoryEvent>, ClientError> {
match self
.send_request(Request::WorkflowEventHistory {
run_id,
after_sequence,
limit,
})
.await?
{
ResponsePayload::WorkflowEventHistory { events } => Ok(events),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_committed_history(
&self,
query: bcode_workflow::WorkflowCommittedHistoryQuery,
) -> Result<bcode_workflow::WorkflowCommittedHistoryPage, ClientError> {
query
.validate()
.map_err(|_| ClientError::UnexpectedResponse)?;
match self
.send_request(Request::WorkflowCommittedHistory {
query: query.clone(),
})
.await?
{
ResponsePayload::WorkflowCommittedHistory { page } => {
query
.validate_page(&page)
.map_err(|_| ClientError::UnexpectedResponse)?;
Ok(page)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_lifecycle_history(
&self,
query: bcode_workflow::WorkflowLifecycleHistoryQuery,
) -> Result<bcode_workflow::WorkflowLifecycleHistoryPage, ClientError> {
query
.validate()
.map_err(|_| ClientError::UnexpectedResponse)?;
match self
.send_request(Request::WorkflowLifecycleHistory {
query: query.clone(),
})
.await?
{
ResponsePayload::WorkflowLifecycleHistory { page } => {
query
.validate_page(&page)
.map_err(|_| ClientError::UnexpectedResponse)?;
Ok(page)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_inventory(
&self,
query: bcode_workflow::WorkflowInventoryQuery,
) -> Result<bcode_workflow::WorkflowInventoryPage, ClientError> {
query
.validate()
.map_err(|_| ClientError::UnexpectedResponse)?;
match self
.send_request(Request::WorkflowInventory {
query: query.clone(),
})
.await?
{
ResponsePayload::WorkflowInventory { page } => {
query
.validate_page(&page)
.map_err(|_| ClientError::UnexpectedResponse)?;
Ok(page)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_semantic_history(
&self,
query: bcode_workflow::WorkflowSemanticHistoryQuery,
) -> Result<bcode_workflow::WorkflowSemanticHistoryPage, ClientError> {
query
.validate()
.map_err(|_| ClientError::UnexpectedResponse)?;
match self
.send_request(Request::WorkflowSemanticHistory {
query: query.clone(),
})
.await?
{
ResponsePayload::WorkflowSemanticHistory { page } => {
query
.validate_page(&page)
.map_err(|_| ClientError::UnexpectedResponse)?;
Ok(page)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_observations(
&self,
run_id: String,
after_sequence: Option<u64>,
through_sequence: Option<u64>,
limit: usize,
) -> Result<bcode_workflow::WorkflowObservationPage, ClientError> {
match self
.send_request(Request::WorkflowObservations {
run_id: run_id.clone(),
after_sequence,
through_sequence,
limit,
})
.await?
{
ResponsePayload::WorkflowObservations { page }
if page.version == 1
&& page.run_id == run_id
&& page.observations.len() <= limit
&& limit <= 100 =>
{
Ok(page)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn cancel_runtime_work(
&self,
session_id: SessionId,
work_id: bcode_session_models::WorkId,
) -> Result<bool, ClientError> {
match self
.send_request(Request::CancelRuntimeWork {
session_id,
work_id,
})
.await?
{
ResponsePayload::RuntimeWorkCancellationRequested { cancelled } => Ok(cancelled),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_runtime_work(
&self,
session_id: SessionId,
) -> Result<Vec<bcode_session_models::RuntimeWorkSnapshot>, ClientError> {
match self
.send_request(Request::ListRuntimeWork { session_id })
.await?
{
ResponsePayload::RuntimeWorkList { work } => Ok(work),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn runtime_work_observations(
&self,
session_id: SessionId,
after: Option<bcode_session_models::WorkId>,
limit: usize,
) -> Result<bcode_session_models::RuntimeWorkObservationPage, ClientError> {
match self
.send_request(Request::RuntimeWorkObservations {
session_id,
after,
limit,
})
.await?
{
ResponsePayload::RuntimeWorkObservations { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn runtime_work_history(
&self,
session_id: SessionId,
limit: usize,
) -> Result<Vec<bcode_session_models::SessionEvent>, ClientError> {
match self
.send_request(Request::RuntimeWorkHistory { session_id, limit })
.await?
{
ResponsePayload::RuntimeWorkHistory { events } => Ok(events),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn runtime_work_spans(
&self,
session_id: SessionId,
limit: usize,
) -> Result<Vec<RuntimeWorkSpan>, ClientError> {
Ok(runtime_work_spans(
self.runtime_work_history(session_id, limit).await?,
))
}
pub async fn compact_session(&self, session_id: SessionId) -> Result<String, ClientError> {
match self
.send_request(Request::CompactSession { session_id })
.await?
{
ResponsePayload::SessionCompacted { message, .. } => Ok(message),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_agents(&self) -> Result<Vec<AgentInfo>, ClientError> {
match self.send_request(Request::ListAgents).await? {
ResponsePayload::AgentList { agents } => Ok(agents),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_skills(&self) -> Result<SkillList, ClientError> {
match self.send_request(Request::ListSkills).await? {
ResponsePayload::SkillList { skills } => Ok(*skills),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn describe_skill(&self, skill_id: SkillId) -> Result<SkillManifest, ClientError> {
match self
.send_request(Request::DescribeSkill { skill_id })
.await?
{
ResponsePayload::SkillManifest { skill } => Ok(*skill),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn invoke_skill(
&self,
session_id: SessionId,
skill_id: SkillId,
arguments: String,
display_text: String,
) -> Result<MessageAcceptance, ClientError> {
self.invoke_skill_request(Request::InvokeSkill {
session_id,
skill_id,
arguments,
display_text,
})
.await
}
pub async fn invoke_skill_with_execution(
&self,
session_id: SessionId,
skill_id: SkillId,
arguments: String,
display_text: String,
execution: bcode_session_models::TurnExecutionOptions,
) -> Result<MessageAcceptance, ClientError> {
self.invoke_skill_request(Request::InvokeSkillWithExecution {
session_id,
skill_id,
arguments,
display_text,
execution,
})
.await
}
async fn invoke_skill_request(
&self,
request: Request,
) -> Result<MessageAcceptance, ClientError> {
decode_message_acceptance(&self.send_request(request).await?)
}
pub async fn activate_skill(
&self,
session_id: SessionId,
skill_id: SkillId,
) -> Result<(), ClientError> {
match self
.send_request(Request::ActivateSkill {
session_id,
skill_id,
})
.await?
{
ResponsePayload::SessionAgentSet => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn deactivate_skill(
&self,
session_id: SessionId,
skill_id: SkillId,
) -> Result<(), ClientError> {
match self
.send_request(Request::DeactivateSkill {
session_id,
skill_id,
})
.await?
{
ResponsePayload::SessionAgentSet => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn active_skills(
&self,
session_id: SessionId,
) -> Result<Vec<bcode_skill_models::SkillContextResponse>, ClientError> {
match self
.send_request(Request::ActiveSkills { session_id })
.await?
{
ResponsePayload::ActiveSkills { skills } => Ok(skills),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn agent_policy_status(&self) -> Result<PolicyStatusResponse, ClientError> {
match self.send_request(Request::AgentPolicyStatus).await? {
ResponsePayload::AgentPolicyStatus { status } => Ok(status),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn set_session_agent(
&self,
session_id: SessionId,
agent_id: String,
) -> Result<(), ClientError> {
match self
.send_request(Request::SetSessionAgent {
session_id,
agent_id,
})
.await?
{
ResponsePayload::SessionAgentSet => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_permissions(&self) -> Result<Vec<PermissionSummary>, ClientError> {
match self.send_request(Request::ListPermissions).await? {
ResponsePayload::PermissionList { permissions } => Ok(permissions),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn resolve_permission(
&self,
permission_id: String,
approved: bool,
) -> Result<bool, ClientError> {
self.resolve_permission_with_remember(permission_id, approved, false)
.await
}
pub async fn resolve_permission_with_remember(
&self,
permission_id: String,
approved: bool,
remember: bool,
) -> Result<bool, ClientError> {
match self
.send_request(Request::ResolvePermission {
permission_id,
approved,
remember,
})
.await?
{
ResponsePayload::PermissionResolved { resolved } => Ok(resolved),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn resolve_permission_batch(
&self,
batch_id: String,
approved: bool,
) -> Result<usize, ClientError> {
match self
.send_request(Request::ResolvePermissionBatch { batch_id, approved })
.await?
{
ResponsePayload::PermissionBatchResolved { resolved } => Ok(resolved),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_pending_tool_exchanges(
&self,
) -> Result<Vec<PendingToolExchangeSummary>, ClientError> {
match self.send_request(Request::ListPendingToolExchanges).await? {
ResponsePayload::PendingToolExchangeList { exchanges } => Ok(exchanges),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn inspect_pending_tool_exchange(
&self,
exchange_id: &str,
) -> Result<Option<PendingToolExchangeSummary>, ClientError> {
select_pending_tool_exchange(self.list_pending_tool_exchanges().await?, exchange_id)
}
pub async fn resolve_tool_exchange(
&self,
exchange_id: String,
resolution: bcode_session_models::ToolExchangeResolution,
) -> Result<bool, ClientError> {
let resolution_json = serde_json::to_value(resolution).map_err(|error| {
ClientError::Protocol(format!("exchange resolution encode failed: {error}"))
})?;
match self
.send_request(Request::ResolveToolExchange {
exchange_id,
resolution_json,
})
.await?
{
ResponsePayload::ToolExchangeResolved { resolved } => Ok(resolved),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn add_permission_rule(
&self,
agent_id: String,
category: String,
pattern: String,
action: String,
) -> Result<String, ClientError> {
match self
.send_request(Request::AddPermissionRule {
agent_id,
category,
pattern,
action,
})
.await?
{
ResponsePayload::PermissionRuleAdded { config_path } => Ok(config_path),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn plugin_services(&self) -> Result<Vec<PluginServiceSummary>, ClientError> {
match self.send_request(Request::ListPluginServices).await? {
ResponsePayload::PluginServices { services } => Ok(services),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn plugin_contributions(&self) -> Result<PluginContributions, ClientError> {
match self.send_request(Request::ListPluginContributions).await? {
ResponsePayload::PluginContributions { contributions } => Ok(contributions),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn generate_session_milestone(
&self,
session_id: SessionId,
) -> Result<bcode_ipc::SessionMilestoneGenerationOutcome, ClientError> {
match self
.send_request(Request::GenerateSessionMilestone { session_id })
.await?
{
ResponsePayload::SessionMilestoneGenerated { result } => Ok(result),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_milestones(
&self,
session_id: SessionId,
query: bcode_session_models::SessionMilestoneQuery,
) -> Result<bcode_session_models::SessionMilestonePage, ClientError> {
match self
.send_request(Request::SessionMilestones { session_id, query })
.await?
{
ResponsePayload::SessionMilestones { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_derivation_snapshot(
&self,
session_id: SessionId,
) -> Result<SessionDerivationSourceSnapshot, ClientError> {
match self
.send_request(Request::SessionDerivationSnapshot { session_id })
.await?
{
ResponsePayload::SessionDerivationSnapshot { snapshot } => Ok(snapshot),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_derivation_prompts(
&self,
session_id: SessionId,
query: SessionDerivationPromptQuery,
) -> Result<SessionDerivationPromptPage, ClientError> {
match self
.send_request(Request::SessionDerivationPrompts { session_id, query })
.await?
{
ResponsePayload::SessionDerivationPrompts { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn derive_session(
&self,
request: SessionDerivationRequest,
) -> Result<SessionDerivationTerminalOutcome, ClientError> {
match self
.send_request(Request::DeriveSession {
request: Box::new(request),
})
.await?
{
ResponsePayload::SessionDerived { outcome } => Ok(outcome),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_derivation_status(
&self,
operation_id: bcode_session_models::SessionDerivationOperationId,
) -> Result<bcode_session_models::SessionDerivationOperationSnapshot, ClientError> {
match self
.send_request(Request::SessionDerivationStatus { operation_id })
.await?
{
ResponsePayload::SessionDerivationStatus { snapshot } => Ok(snapshot),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn cancel_session_derivation(
&self,
operation_id: bcode_session_models::SessionDerivationOperationId,
) -> Result<bool, ClientError> {
match self
.send_request(Request::CancelSessionDerivation { operation_id })
.await?
{
ResponsePayload::SessionDerivationCancellationRequested { accepted } => Ok(accepted),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn invoke_plugin_service_contextual(
&self,
context: bcode_ipc::PluginServiceContext,
plugin_id: String,
interface_id: String,
operation: String,
payload: Vec<u8>,
) -> Result<PluginServiceResponse, ClientError> {
contextual::invoke(self, context, plugin_id, interface_id, operation, payload).await
}
pub async fn cancel_plugin_service_invocation(
&self,
invocation_id: String,
) -> Result<bool, ClientError> {
contextual::cancel(self, &invocation_id).await
}
pub async fn invoke_plugin_service(
&self,
plugin_id: String,
interface_id: String,
operation: String,
payload: Vec<u8>,
) -> Result<PluginServiceResponse, ClientError> {
match self
.send_request(Request::InvokePluginService {
plugin_id,
interface_id,
operation,
payload,
})
.await?
{
ResponsePayload::PluginServiceResult { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn validate_provider_config(
&self,
provider_plugin_id: String,
) -> Result<bcode_model::ValidateConfigResponse, ClientError> {
self.validate_model_config(Some(provider_plugin_id)).await
}
pub async fn validate_model_config(
&self,
provider_plugin_id: Option<String>,
) -> Result<bcode_model::ValidateConfigResponse, ClientError> {
let request = bcode_model::ValidateConfigRequest {
provider_context: self
.runtime_context
.as_ref()
.map_or_else(Default::default, |context| context.provider_context.clone()),
..Default::default()
};
let payload = serde_json::to_vec(&request).map_err(|_| {
ClientError::Protocol("Provider validation request could not be encoded.".to_owned())
})?;
let response = if let Some(provider) = provider_plugin_id {
self.invoke_plugin_service(
provider,
bcode_model::MODEL_PROVIDER_INTERFACE_ID.to_owned(),
bcode_model::OP_VALIDATE_CONFIG.to_owned(),
payload,
)
.await?
} else {
self.call_plugin_service(
bcode_model::MODEL_PROVIDER_INTERFACE_ID.to_owned(),
bcode_model::OP_VALIDATE_CONFIG.to_owned(),
payload,
)
.await?
};
decode_provider_validation(&response)
}
pub async fn call_plugin_service(
&self,
interface_id: String,
operation: String,
payload: Vec<u8>,
) -> Result<PluginServiceResponse, ClientError> {
match self
.send_request(Request::CallPluginService {
interface_id,
operation,
payload,
})
.await?
{
ResponsePayload::PluginServiceResult { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn publish_plugin_event(
&self,
topic: String,
payload: Vec<u8>,
) -> Result<usize, ClientError> {
match self
.send_request(Request::PublishPluginEvent { topic, payload })
.await?
{
ResponsePayload::PluginEventPublished { delivered } => Ok(delivered),
_ => Err(ClientError::UnexpectedResponse),
}
}
async fn send_request(&self, request: Request) -> Result<ResponsePayload, ClientError> {
self.send_request_once(request).await
}
async fn send_request_with_timeout(
&self,
request: Request,
request_timeout: Duration,
) -> Result<ResponsePayload, ClientError> {
let mut connection = self.connect("bcode-cli").await?;
connection.request_timeout = request_timeout;
connection.send_request(request).await
}
async fn send_request_once(&self, request: Request) -> Result<ResponsePayload, ClientError> {
let mut connection = self.connect("bcode-cli").await?;
connection.send_request(request).await
}
pub async fn connect(&self, client_name: &str) -> Result<ClientConnection, ClientError> {
use tracing::Instrument as _;
if let Some(error) = &self.runtime_context_error {
return Err(ClientError::Protocol(error.clone()));
}
let span = tracing::debug_span!(target: "bcode_client::startup", "daemon_connection");
Box::pin(async {
let started = std::time::Instant::now();
let mut acquisition_required = false;
let result = async {
match self.connect_with_deadline(client_name).await {
Ok(connection) => Ok(connection),
Err(error)
if self.daemon_availability == DaemonAvailability::AutoStart
&& self.expected_state_location.is_none()
&& error.is_daemon_unavailable() =>
{
acquisition_required = true;
self.ensure_daemon_available().await?;
self.connect_with_deadline(client_name).await
}
Err(error) => Err(error),
}
}
.await;
tracing::debug!(
target: "bcode_client::startup",
elapsed_us = started.elapsed().as_micros(),
acquisition_required,
success = result.is_ok(),
"daemon connection completed"
);
result
})
.instrument(span)
.await
}
async fn connect_with_deadline(
&self,
client_name: &str,
) -> Result<ClientConnection, ClientError> {
let phase = bcode_metrics::startup::phase("client.verified_connection_attempt");
let started = std::time::Instant::now();
let result =
tokio::time::timeout(self.connect_timeout, self.connect_once(client_name)).await;
if matches!(&result, Ok(Ok(_))) {
phase.finish();
} else {
phase.fail();
}
tracing::debug!(
target: "bcode_client::startup",
elapsed_us = started.elapsed().as_micros(),
timed_out = result.is_err(),
success = matches!(&result, Ok(Ok(_))),
"verified connection attempt completed"
);
result.map_err(|_| ClientError::ConnectTimeout {
timeout: self.connect_timeout,
})?
}
#[must_use]
pub fn observe_session_open(&self, session_id: SessionId) -> SessionOpenProgressObserver {
let client = self.clone();
let (sender, receiver) = tokio::sync::mpsc::unbounded_channel();
let task = tokio::spawn(async move {
let mut connection = client.connect("bcode-session-open-observer").await?;
let close_sender = sender.clone();
let observation = connection.prepare_session_open_while(session_id, |snapshot| {
sender.send(snapshot.clone()).is_ok()
});
tokio::pin!(observation);
tokio::select! {
result = &mut observation => {
result?;
}
() = close_sender.closed() => {}
}
Ok(())
});
SessionOpenProgressObserver { receiver, task }
}
async fn connect_once(&self, client_name: &str) -> Result<ClientConnection, ClientError> {
let phase = bcode_metrics::startup::phase("client.transport_connect");
let transport_started = std::time::Instant::now();
let stream = LocalIpcStream::connect(&self.endpoint).await;
phase.finish_result(&stream);
tracing::debug!(
target: "bcode_client::startup",
elapsed_us = transport_started.elapsed().as_micros(),
success = stream.is_ok(),
"local transport connection completed"
);
let stream = stream?;
let handshake_phase = bcode_metrics::startup::phase("client.hello");
let handshake_started = std::time::Instant::now();
let mut connection = ClientConnection {
stream,
next_request_id: 1,
client_id: None,
pending_events: VecDeque::new(),
request_timeout: self.request_timeout,
reconnect_client: Some(std::sync::Arc::new(self.clone())),
reconnect_name: std::sync::Arc::from(client_name),
restore_state: ConnectionRestoreState {
runtime_context: self.runtime_context.clone(),
..ConnectionRestoreState::default()
},
};
match connection
.send_request(Request::Hello {
client_name: format!("{client_name};cap=message_accepted"),
runtime_context: self.runtime_context.clone(),
daemon_namespace: bcode_ipc::daemon_namespace(),
artifact_id: Some(bcode_ipc::ArtifactId::current()),
build_fingerprint: bcode_ipc::BUILD_FINGERPRINT.to_owned(),
state_location_id: Some(
self.expected_state_location
.as_ref()
.map_or_else(bcode_ipc::state_location_id, |id| id.as_str().to_owned()),
),
})
.await?
{
ResponsePayload::Hello {
client_id, daemon, ..
} => {
self.verify_server_identity(&daemon)?;
tracing::debug!(
target: "bcode_client::startup",
elapsed_us = handshake_started.elapsed().as_micros(),
"daemon identity handshake verified"
);
connection.client_id = Some(client_id);
handshake_phase.finish();
Ok(connection)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
}
#[derive(Debug, Clone)]
enum SessionRestoreAttachment {
Full {
session_id: SessionId,
},
Recent {
session_id: SessionId,
limit: usize,
},
Projection {
session_id: SessionId,
request: ProjectionWindowRequest,
},
}
impl SessionRestoreAttachment {
const fn session_id(&self) -> SessionId {
match self {
Self::Full { session_id }
| Self::Recent { session_id, .. }
| Self::Projection { session_id, .. } => *session_id,
}
}
}
#[derive(Debug, Clone, Default)]
struct ConnectionRestoreState {
runtime_context: Option<ClientRuntimeContext>,
attached_session: Option<SessionRestoreAttachment>,
catalog_updates: bool,
workflow_runs: bool,
runtime_work_sessions: std::collections::BTreeSet<SessionId>,
}
#[derive(Debug)]
pub struct ClientConnection {
stream: LocalIpcStream,
next_request_id: u64,
client_id: Option<ClientId>,
pending_events: VecDeque<Event>,
request_timeout: Duration,
reconnect_client: Option<std::sync::Arc<BcodeClient>>,
reconnect_name: std::sync::Arc<str>,
restore_state: ConnectionRestoreState,
}
impl ClientConnection {
pub async fn submissions_for_history(
&mut self,
session_id: SessionId,
events: &[SessionEvent],
) -> Result<Vec<bcode_session_models::SubmissionRecord>, ClientError> {
let mut ids = std::collections::BTreeSet::new();
for event in events {
match &event.kind {
SessionEventKind::SubmissionAccepted { submission } => {
ids.insert(submission.id.clone());
}
SessionEventKind::SubmissionCancelled { submission_id }
| SessionEventKind::SubmissionContextObserved { submission_id, .. } => {
ids.insert(submission_id.clone());
}
_ => {}
}
if ids.len() > 256 {
return Err(ClientError::Protocol("submission hydration window exceeds 256 identities; request a smaller history window".to_owned()));
}
}
let mut records = Vec::with_capacity(ids.len());
for id in ids {
let record = self.get_submission(session_id, id).await?.ok_or_else(|| {
ClientError::Protocol(
"canonical submission referenced by history is unavailable".to_owned(),
)
})?;
records.push(record);
}
Ok(records)
}
pub async fn get_submission(
&mut self,
session_id: SessionId,
submission_id: bcode_session_models::SubmissionId,
) -> Result<Option<bcode_session_models::SubmissionRecord>, ClientError> {
match self
.send_request(Request::GetSubmission {
session_id,
submission_id,
})
.await?
{
ResponsePayload::Submission { record } => Ok(record),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn send_invocation_input(
&mut self,
session_id: SessionId,
input: bcode_tool::ToolInvocationInput,
) -> Result<(), ClientError> {
match self
.send_request(Request::InvocationInput { session_id, input })
.await?
{
ResponsePayload::InvocationInputAccepted => Ok(()),
_ => Err(ClientError::UnexpectedResponse),
}
}
#[must_use]
pub const fn client_id(&self) -> Option<ClientId> {
self.client_id
}
pub async fn update_runtime_context(
&mut self,
runtime_context: Option<ClientRuntimeContext>,
) -> Result<(), ClientError> {
match self
.send_request(Request::UpdateClientRuntimeContext {
runtime_context: runtime_context.clone(),
})
.await?
{
ResponsePayload::ClientRuntimeContextUpdated => {
self.restore_state.runtime_context = runtime_context;
Ok(())
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn refresh_runtime_context(&mut self) -> Result<(), ClientError> {
self.update_runtime_context(Some(current_runtime_context()?))
.await
}
pub async fn subscribe_catalog_updates(&mut self) -> Result<(), ClientError> {
match self.send_request(Request::SubscribeCatalogUpdates).await? {
ResponsePayload::CatalogUpdatesSubscribed => {
self.restore_state.catalog_updates = true;
Ok(())
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn workflow_live_event_catch_up(
&mut self,
after_sequence: u64,
limit: usize,
) -> Result<bcode_workflow_view_models::WorkflowLiveEventPage, ClientError> {
match self
.send_request(Request::WorkflowLiveEventCatchUp {
after_sequence,
limit,
})
.await?
{
ResponsePayload::WorkflowLiveEventCatchUp { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn subscribe_workflow_runs(&mut self) -> Result<u64, ClientError> {
match self.send_request(Request::SubscribeWorkflowRuns).await? {
ResponsePayload::WorkflowRunsSubscribed { after_sequence } => {
self.restore_state.workflow_runs = true;
Ok(after_sequence)
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn subscribe_runtime_work(
&mut self,
session_id: SessionId,
) -> Result<(), ClientError> {
match self
.send_request(Request::SubscribeRuntimeWork { session_id })
.await?
{
ResponsePayload::RuntimeWorkSubscribed => {
self.restore_state.runtime_work_sessions.insert(session_id);
Ok(())
}
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_sessions_with_status(&mut self) -> Result<SessionList, ClientError> {
self.list_sessions_in_working_directory(current_working_directory())
.await
}
pub async fn list_sessions_with_scope(
&mut self,
scope: SessionDiscoveryScope,
) -> Result<SessionList, ClientError> {
let scope = normalize_discovery_scope(scope)?;
decode_scoped_session_list(
self.send_request(Request::ListSessionsWithScope { scope })
.await?,
)
}
pub async fn list_sessions_page_with_scope(
&mut self,
scope: SessionDiscoveryScope,
query: SessionCatalogQuery,
cursor: Option<SessionCatalogCursor>,
limit: usize,
) -> Result<SessionCatalogPage, ClientError> {
let scope = normalize_discovery_scope(scope)?;
match self
.send_request(Request::ListSessionsPageWithScope {
scope,
query,
cursor,
limit,
})
.await?
{
ResponsePayload::SessionCatalogPage { page } => Ok(page),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn list_sessions_in_working_directory(
&mut self,
working_directory: PathBuf,
) -> Result<SessionList, ClientError> {
let working_directory = std::path::absolute(working_directory).map_err(|_| {
ClientError::Protocol("cannot resolve catalog working directory".into())
})?;
match self
.send_request(Request::ListSessions { working_directory })
.await?
{
ResponsePayload::SessionList {
sessions,
catalog_status,
catalog_sources,
catalog_revision,
} => Ok(SessionList {
sessions,
catalog_status,
catalog_sources,
catalog_revision,
}),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn session_compatibility_inventory(
&mut self,
request: SessionCompatibilityInventoryRequest,
) -> Result<SessionCompatibilityInventoryResponse, ClientError> {
match self
.send_request(Request::SessionCompatibilityInventory { request })
.await?
{
ResponsePayload::SessionCompatibilityInventory { response } => Ok(response),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn attach_session(
&mut self,
session_id: SessionId,
) -> Result<Vec<SessionEvent>, ClientError> {
self.attach_session_with_input_history(session_id)
.await
.map(|attached| attached.history)
}
pub async fn attach_session_with_input_history(
&mut self,
session_id: SessionId,
) -> Result<AttachedSessionHistory, ClientError> {
let mut attached = match self
.send_request(Request::AttachSession { session_id })
.await?
{
ResponsePayload::Attached {
history,
input_history,
usage_summary,
import_warnings,
draft,
runtime_selection,
projection_window,
session,
..
} => AttachedSessionHistory {
submissions: Vec::new(),
session,
history,
input_history,
usage_summary: usage_summary.map_or_else(
bcode_session_models::SessionUsageSummary::default,
|summary| *summary,
),
import_warnings,
draft,
runtime_selection,
projection_window,
},
_ => return Err(ClientError::UnexpectedResponse),
};
self.restore_state.attached_session = Some(SessionRestoreAttachment::Full { session_id });
attached.submissions = self
.submissions_for_history(attached.session.id, &attached.history)
.await?;
Ok(attached)
}
pub async fn prepare_session_open(
&mut self,
session_id: SessionId,
) -> Result<bcode_session_models::SessionOpenOperationSnapshot, ClientError> {
match self
.send_request(Request::PrepareSessionOpen { session_id })
.await?
{
ResponsePayload::SessionOpenPrepared { snapshot } => Ok(snapshot),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn wait_session_open_progress(
&mut self,
session_id: SessionId,
operation_id: bcode_session_models::SessionOpenOperationId,
after_revision: u64,
timeout: Duration,
) -> Result<bcode_session_models::SessionOpenOperationSnapshot, ClientError> {
let timeout_ms = u64::try_from(timeout.as_millis()).unwrap_or(u64::MAX);
match self
.send_request(Request::WaitSessionOpenProgress {
session_id,
operation_id,
after_revision,
timeout_ms,
})
.await?
{
ResponsePayload::SessionOpenPrepared { snapshot } => Ok(snapshot),
_ => Err(ClientError::UnexpectedResponse),
}
}
pub async fn prepare_session_open_until_terminal<F>(
&mut self,
session_id: SessionId,
mut on_progress: F,
) -> Result<bcode_session_models::SessionOpenOperationSnapshot, ClientError>
where
F: FnMut(&bcode_session_models::SessionOpenOperationSnapshot),
{
self.prepare_session_open_while(session_id, |snapshot| {
on_progress(snapshot);
true
})
.await
}
async fn prepare_session_open_while<F>(
&mut self,
session_id: SessionId,
mut on_progress: F,
) -> Result<bcode_session_models::SessionOpenOperationSnapshot, ClientError>
where
F: FnMut(&bcode_session_models::SessionOpenOperationSnapshot) -> bool,
{
let mut snapshot = self.prepare_session_open(session_id).await?;
if !on_progress(&snapshot) {
return Ok(snapshot);
}
let mut reconnect_attempts = 0_u8;
while snapshot.outcome.is_none() {
match self
.wait_session_open_progress(
session_id,
snapshot.operation_id,
snapshot.revision,
Duration::from_secs(5),
)
.await
{
Ok(next) => {
snapshot = next;
if !on_progress(&snapshot) {
return Ok(snapshot);
}
}
Err(error)
if error.is_daemon_unavailable()
&& reconnect_attempts < 3
&& self.reconnect_client.is_some() =>
{
reconnect_attempts = reconnect_attempts.saturating_add(1);
self.reconnect_for_session_open().await?;
snapshot = match self
.wait_session_open_progress(
session_id,
snapshot.operation_id,
snapshot.revision,
Duration::ZERO,
)
.await
{
Ok(recovered) => recovered,
Err(ClientError::Server { code, .. })
if code == "session_open_operation_not_found" =>
{
self.prepare_session_open(session_id).await?
}
Err(error) => return Err(error),
};
if !on_progress(&snapshot) {
return Ok(snapshot);
}
}
Err(error) => return Err(error),
}
}
Ok(snapshot)
}
async fn reconnect_for_session_open(&mut self) -> Result<(), ClientError> {
self.reconnect_and_restore().await
}
async fn reconnect_and_restore(&mut self) -> Result<(), ClientError> {
let client = self
.reconnect_client
.clone()
.ok_or(ClientError::UnexpectedResponse)?;
let restore_state = self.restore_state.clone();
let mut delay = Duration::from_millis(25);
let mut replacement = loop {
match client.connect(&self.reconnect_name).await {
Ok(connection) => break connection,
Err(error) if error.is_daemon_unavailable() => {
tokio::time::sleep(delay).await;
delay = delay.saturating_mul(2).min(Duration::from_secs(2));
}
Err(error) => return Err(error),
}
};
replacement.restore_connection_state(&restore_state).await?;
let mut pending_events = std::mem::take(&mut self.pending_events);
pending_events.append(&mut replacement.pending_events);
if let Some(attachment) = &restore_state.attached_session {
pending_events.push_back(Event::SessionViewResyncRequired {
session_id: attachment.session_id(),
});
}
replacement.pending_events = pending_events;
*self = replacement;
Ok(())
}
async fn restore_connection_state(
&mut self,
restore: &ConnectionRestoreState,
) -> Result<(), ClientError> {
if self.restore_state.runtime_context != restore.runtime_context {
match self
.send_request(Request::UpdateClientRuntimeContext {
runtime_context: restore.runtime_context.clone(),
})
.await?
{
ResponsePayload::ClientRuntimeContextUpdated => {}
_ => return Err(ClientError::UnexpectedResponse),
}
}
if let Some(attachment) = &restore.attached_session {
let request = match attachment {
SessionRestoreAttachment::Full { session_id } => Request::AttachSession {
session_id: *session_id,
},
SessionRestoreAttachment::Recent { session_id, limit } => {
Request::AttachSessionRecent {
session_id: *session_id,
limit: *limit,
}
}
SessionRestoreAttachment::Projection {
session_id,
request,
} => Request::AttachSessionProjectionWindow {
session_id: *session_id,
request: request.clone(),
},
};
if !matches!(
self.send_request(request).await?,
ResponsePayload::Attached { .. }
) {
return Err(ClientError::UnexpectedResponse);
}
}
if restore.catalog_updates
&& !matches!(
self.send_request(Request::SubscribeCatalogUpdates).await?,
ResponsePayload::CatalogUpdatesSubscribed
)
{
return Err(ClientError::UnexpectedResponse);
}
if restore.workflow_runs
&& !matches!(
self.send_request(Request::SubscribeWorkflowRuns).await?,
ResponsePayload::WorkflowRunsSubscribed { .. }
)
{
return Err(ClientError::UnexpectedResponse);
}
for session_id in &restore.runtime_work_sessions {
if !matches!(
self.send_request(Request::SubscribeRuntimeWork {
session_id: *session_id,
})
.await?,
ResponsePayload::RuntimeWorkSubscribed
) {
return Err(ClientError::UnexpectedResponse);
}
}
self.restore_state = restore.clone();
Ok(())
}
pub async fn attach_session_recent(
&mut self,
session_id: SessionId,
limit: usize,
) -> Result<Vec<SessionEvent>, ClientError> {
self.attach_session_recent_with_input_history(session_id, limit)
.await
.map(|attached| attached.history)
}
pub async fn attach_session_recent_with_input_history(
&mut self,
session_id: SessionId,
limit: usize,
) -> Result<AttachedSessionHistory, ClientError> {
let mut attached = match self
.send_request(Request::AttachSessionRecent { session_id, limit })
.await?
{
ResponsePayload::Attached {
history,
input_history,
usage_summary,
import_warnings,
draft,
runtime_selection,
projection_window,
session,
..
} => AttachedSessionHistory {
submissions: Vec::new(),
session,
history,
input_history,
usage_summary: usage_summary.map_or_else(
bcode_session_models::SessionUsageSummary::default,
|summary| *summary,
),
import_warnings,
draft,
runtime_selection,
projection_window,
},
_ => return Err(ClientError::UnexpectedResponse),
};
self.restore_state.attached_session =
Some(SessionRestoreAttachment::Recent { session_id, limit });
attached.submissions = self
.submissions_for_history(attached.session.id, &attached.history)
.await?;
Ok(attached)
}
pub async fn prepare_then_attach_session_projection_window<F>(
&mut self,
session_id: SessionId,
request: bcode_session_models::ProjectionWindowRequest,
on_progress: F,
) -> Result<AttachedSessionHistory, ClientError>
where
F: FnMut(&bcode_session_models::SessionOpenOperationSnapshot),
{
let snapshot = self
.prepare_session_open_until_terminal(session_id, on_progress)
.await?;
session_open_attach_readiness(&snapshot)?;
self.attach_session_projection_window_with_input_history(session_id, request)
.await
}
pub async fn attach_session_projection_window_with_input_history(
&mut self,
session_id: SessionId,
request: ProjectionWindowRequest,
) -> Result<AttachedSessionHistory, ClientError> {
let restore_request = request.clone();
let mut attached = match self
.send_request(Request::AttachSessionProjectionWindow {
session_id,
request,
})
.await?
{
ResponsePayload::Attached {
history,
input_history,
usage_summary,
import_warnings,
draft,
runtime_selection,
projection_window,
session,
..
} => AttachedSessionHistory {
submissions: Vec::new(),
session,
history,
input_history,
usage_summary: usage_summary.map_or_else(
bcode_session_models::SessionUsageSummary::default,
|summary| *summary,
),
import_warnings,
draft,
runtime_selection,
projection_window,
},
_ => return Err(ClientError::UnexpectedResponse),
};
self.restore_state.attached_session = Some(SessionRestoreAttachment::Projection {
session_id,
request: restore_request,
});
attached.submissions = self
.submissions_for_history(attached.session.id, &attached.history)
.await?;
Ok(attached)
}
pub async fn send_user_message(
&mut self,
session_id: SessionId,
text: String,
placement: bcode_session_models::PromptPlacement,
) -> Result<MessageAcceptance, ClientError> {
decode_message_acceptance(
&self
.send_request(Request::SendUserMessageWithPlacement {
session_id,
text,
placement,
})
.await?,
)
}
pub async fn recv_event_without_reconnect(&mut self) -> Result<Event, ClientError> {
loop {
if let Some(event) = self.pending_events.pop_front() {
return Ok(event);
}
let envelope = recv_envelope(&mut self.stream).await?;
if envelope.kind == EnvelopeKind::Event {
return decode_event(&envelope.payload).map_err(ClientError::from);
}
}
}
pub async fn recv_event(&mut self) -> Result<Event, ClientError> {
loop {
if let Some(event) = self.pending_events.pop_front() {
return Ok(event);
}
match recv_envelope(&mut self.stream).await {
Ok(envelope) => {
if envelope.kind != EnvelopeKind::Event {
continue;
}
return decode_event(&envelope.payload).map_err(ClientError::from);
}
Err(error) => {
let error = ClientError::from(error);
if !error.is_daemon_unavailable() || self.reconnect_client.is_none() {
return Err(error);
}
self.reconnect_and_restore().await?;
}
}
}
}
async fn send_request(&mut self, request: Request) -> Result<ResponsePayload, ClientError> {
let request_id = self.next_request_id;
self.next_request_id += 1;
let hello = matches!(&request, Request::Hello { .. });
let encoding = hello.then(|| bcode_metrics::startup::phase("client.hello.encode"));
let envelope = request_envelope(request_id, &request)?;
if let Some(phase) = encoding {
phase.finish();
}
let sending = hello.then(|| bcode_metrics::startup::phase("client.hello.send"));
send_envelope(&mut self.stream, &envelope).await?;
if let Some(phase) = sending {
phase.finish();
}
let waiting = hello.then(|| bcode_metrics::startup::phase("client.hello.response"));
let mut compaction_accepted = false;
loop {
let envelope = if compaction_accepted {
recv_envelope(&mut self.stream).await?
} else {
tokio::time::timeout(self.request_timeout, recv_envelope(&mut self.stream))
.await
.map_err(|_| ClientError::RequestTimeout {
timeout: self.request_timeout,
})??
};
if envelope.kind == EnvelopeKind::Event {
self.pending_events
.push_back(decode_event(&envelope.payload).map_err(ClientError::from)?);
continue;
}
if envelope.kind != EnvelopeKind::Response || envelope.request_id != request_id {
continue;
}
let response: Response = decode_response(&envelope.payload)?;
if matches!(
response,
Response::Ok(ResponsePayload::SessionCompactionAccepted)
) {
if !matches!(request, Request::CompactSession { .. }) || compaction_accepted {
return Err(ClientError::UnexpectedResponse);
}
compaction_accepted = true;
continue;
}
if let Some(phase) = waiting {
phase.finish();
}
return match response {
Response::Ok(payload) => Ok(payload),
Response::Err(error) => Err(error.into()),
};
}
}
}
fn session_open_attach_readiness(
snapshot: &bcode_session_models::SessionOpenOperationSnapshot,
) -> Result<(), ClientError> {
let session_id = snapshot.session_id;
let stage_message = &snapshot.progress.message;
let verified_backup_path = snapshot.backup_path.as_deref();
match &snapshot.outcome {
Some(bcode_session_models::SessionOpenTerminalOutcome::Ready) => Ok(()),
Some(bcode_session_models::SessionOpenTerminalOutcome::DegradedReadOnly {
issue_count,
}) => Err(ClientError::Server {
code: "session_degraded_read_only".to_owned(),
message: format!(
"session contains {issue_count} unsupported persisted event(s); bounded history remains inspectable but writable attach is disabled"
),
}),
Some(bcode_session_models::SessionOpenTerminalOutcome::WriterIncompatible {
actual,
expected,
}) => Err(ClientError::Server {
code: "session_writer_incompatible".to_owned(),
message: terminal_session_open_error_message(
session_id,
stage_message,
&format!(
"session writer epoch {actual:?} is incompatible with expected epoch {expected}"
),
verified_backup_path,
),
}),
Some(bcode_session_models::SessionOpenTerminalOutcome::RepairRequired { reason }) => {
Err(ClientError::Server {
code: "session_repair_required".to_owned(),
message: terminal_session_open_error_message(
session_id,
stage_message,
reason,
verified_backup_path,
),
})
}
Some(bcode_session_models::SessionOpenTerminalOutcome::Failed {
kind,
message,
backup_path,
}) => Err(ClientError::Server {
code: session_open_failure_code(*kind).to_owned(),
message: terminal_session_open_error_message(
session_id,
stage_message,
message,
backup_path.as_deref().or(verified_backup_path),
),
}),
None => Err(ClientError::UnexpectedResponse),
}
}
fn validate_bounded_long_poll_timeout(
operation: &'static str,
timeout_ms: u64,
) -> Result<(), ClientError> {
if timeout_ms == 0 || timeout_ms > 30_000 {
return Err(ClientError::Protocol(format!(
"{operation} wait timeout must be between 1 and 30000 milliseconds"
)));
}
Ok(())
}
fn validate_session_search_backfill_wait_timeout(timeout_ms: u64) -> Result<(), ClientError> {
validate_bounded_long_poll_timeout("backfill", timeout_ms)
}
fn validate_session_bulk_migration_wait_timeout(timeout_ms: u64) -> Result<(), ClientError> {
validate_bounded_long_poll_timeout("bulk migration", timeout_ms)
}
fn terminal_session_open_error_message(
session_id: SessionId,
stage_message: &str,
reason: &str,
backup_path: Option<&std::path::Path>,
) -> String {
let backup = backup_path.map_or_else(String::new, |path| {
format!(" Retained backup: {}.", path.display())
});
format!(
"session preparation failed during {stage_message}: {reason}.{backup} Diagnose with `bcode session diagnose {session_id}`."
)
}
const fn session_open_failure_code(
kind: bcode_session_models::SessionOpenFailureKind,
) -> &'static str {
match kind {
bcode_session_models::SessionOpenFailureKind::OwnedByOtherDaemon => {
"session_active_elsewhere"
}
bcode_session_models::SessionOpenFailureKind::WriterIncompatible => {
"session_writer_incompatible"
}
bcode_session_models::SessionOpenFailureKind::ProjectionStale => "projection_stale",
bcode_session_models::SessionOpenFailureKind::RepairRequired => "session_repair_required",
bcode_session_models::SessionOpenFailureKind::BackupFailed => {
"session_migration_backup_failed"
}
bcode_session_models::SessionOpenFailureKind::MigrationFailed => "session_migration_failed",
bcode_session_models::SessionOpenFailureKind::NotFound => "session_not_found",
}
}
#[cfg(all(test, unix))]
mod turn_observation_tests;
#[cfg(test)]
mod message_acceptance_tests {
use super::*;
use bcode_session_models::MessageAcceptanceDisposition as Disposition;
#[test]
fn outcome_matching_requires_exact_terminal_turn_identity() {
use bcode_session_models::{ModelTurnOutcome, SessionEventKind};
let mut event = SessionEvent {
schema_version: 1,
sequence: 1,
timestamp_ms: 0,
session_id: SessionId::new(),
provenance: None,
kind: SessionEventKind::ModelTurnStarted {
turn_id: "wanted".to_string(),
},
};
assert_eq!(matching_turn_outcome(&event, "wanted"), None);
for outcome in [
ModelTurnOutcome::Completed,
ModelTurnOutcome::Cancelled,
ModelTurnOutcome::Error,
ModelTurnOutcome::IdleTimeout,
ModelTurnOutcome::ToolRoundLimitReached,
ModelTurnOutcome::ProviderUnavailable,
] {
event.kind = SessionEventKind::ModelTurnFinished {
turn_id: "wanted".to_string(),
outcome,
message: None,
};
assert_eq!(matching_turn_outcome(&event, "other"), None);
assert_eq!(matching_turn_outcome(&event, "wanted"), Some(outcome));
}
}
#[test]
fn acceptance_decoder_preserves_all_reported_fields() {
for disposition in [
Disposition::StartedTurn,
Disposition::AppliedSteering,
Disposition::QueuedFollowUp,
Disposition::QueuedTurn,
] {
for queued in [false, true] {
for queue_position in [None, Some(0), Some(u32::MAX)] {
let result = decode_message_acceptance(
&ResponsePayload::MessageAcceptedWithDisposition {
queued,
queue_position,
disposition,
},
)
.expect("acceptance");
assert_eq!(
result,
MessageAcceptance {
queued,
queue_position,
disposition
}
);
}
}
}
}
#[test]
fn acceptance_decoder_preserves_compatibility_and_rejects_unrelated_responses() {
assert_eq!(
decode_message_acceptance(&ResponsePayload::MessageSent).unwrap(),
MessageAcceptance::sent()
);
for queued in [false, true] {
for queue_position in [None, Some(0), Some(u32::MAX)] {
assert_eq!(
decode_message_acceptance(&ResponsePayload::MessageAccepted {
queued,
queue_position
})
.unwrap(),
MessageAcceptance {
queued,
queue_position,
disposition: Disposition::StartedTurn,
}
);
}
}
assert!(matches!(
decode_message_acceptance(&ResponsePayload::ComposerDraftSet),
Err(ClientError::UnexpectedResponse)
));
}
}
fn select_pending_tool_exchange(
exchanges: Vec<PendingToolExchangeSummary>,
exchange_id: &str,
) -> Result<Option<PendingToolExchangeSummary>, ClientError> {
let mut matches = exchanges
.into_iter()
.filter(|exchange| exchange.request.exchange_id == exchange_id);
let exchange = matches.next();
if matches.next().is_some() {
return Err(ClientError::Protocol(
"ambiguous pending interaction identifier".to_owned(),
));
}
Ok(exchange)
}
#[cfg(test)]
mod session_progress_tests {
use super::*;
use bcode_workflow_view_models::{
SESSION_PROGRESS_VERSION, SessionProgressObservation, SessionProgressUnavailableReason,
};
fn snapshot(id: SessionId) -> SessionProgressSnapshot {
SessionProgressSnapshot {
version: SESSION_PROGRESS_VERSION,
session_id: id.to_string(),
observation: SessionProgressObservation::Available {
epoch: "public-epoch".into(),
counter: 7,
last_progress_at_ms: Some(123),
},
}
}
fn wire_decode(value: SessionProgressSnapshot) -> ResponsePayload {
let response = Response::Ok(ResponsePayload::SessionProgressSnapshot { snapshot: value });
let envelope = bcode_ipc::response_envelope(1, &response).unwrap();
match decode_response(&envelope.payload).unwrap() {
Response::Ok(payload) => payload,
Response::Err(_) => panic!("expected successful progress response"),
}
}
#[test]
fn decodes_exact_content_free_public_snapshot_through_the_wire() {
let id = SessionId::new();
let expected = snapshot(id);
let decoded = decode_session_progress_response(wire_decode(expected.clone()), id).unwrap();
assert_eq!(decoded, expected);
assert_eq!(
serde_json::to_value(&decoded).unwrap(),
serde_json::json!({
"version": SESSION_PROGRESS_VERSION,
"session_id": id.to_string(),
"observation": {
"status": "available", "epoch": "public-epoch", "counter": 7,
"last_progress_at_ms": 123
}
})
);
let request =
bcode_ipc::request_envelope(1, &Request::SessionProgressSnapshot { session_id: id })
.unwrap();
assert!(matches!(
bcode_ipc::decode_request(&request.payload).unwrap(),
Request::SessionProgressSnapshot { session_id } if session_id == id
));
}
#[test]
fn preserves_unavailable_coverage_without_inventing_a_counter() {
let id = SessionId::new();
let expected = SessionProgressSnapshot {
observation: SessionProgressObservation::Unavailable {
reason: SessionProgressUnavailableReason::CoverageUnavailable,
},
..snapshot(id)
};
let decoded = decode_session_progress_response(wire_decode(expected.clone()), id).unwrap();
assert_eq!(decoded, expected);
assert_eq!(decoded.advanced_since(&snapshot(id)), None);
}
#[test]
fn workflow_activity_request_roundtrips_retained_scan_coordinates() {
let request = Request::WorkflowActivityPage {
request: bcode_workflow_view_models::WorkflowActivityPageRequest {
version: bcode_workflow_view_models::WORKFLOW_ACTIVITY_VERSION,
scan_id: Some("opaque-scan".into()),
root_run_id: "root".into(),
run_id: "child".into(),
expected_revision: Some(3),
section: bcode_workflow_view_models::WorkflowActivitySection::Children,
cursor: Some("opaque-cursor".into()),
limit: 100,
},
};
let encoded = bcode_ipc::request_envelope(1, &request).unwrap();
let decoded = bcode_ipc::decode_request(&encoded.payload).unwrap();
assert_eq!(
serde_json::to_value(decoded).unwrap(),
serde_json::to_value(request).unwrap()
);
let response = Response::Ok(ResponsePayload::WorkflowActivityClosed);
let encoded = bcode_ipc::response_envelope(1, &response).unwrap();
assert!(matches!(
decode_response(&encoded.payload).unwrap(),
Response::Ok(ResponsePayload::WorkflowActivityClosed)
));
}
#[test]
fn workflow_activity_response_preserves_asof_coverage_and_unavailable_progress() {
let page = bcode_workflow_view_models::WorkflowActivityPage {
version: bcode_workflow_view_models::WORKFLOW_ACTIVITY_VERSION,
scan_id: "scan".into(),
root_run_id: "root".into(),
snapshot_started_at_ms: 123,
snapshot_age_ms: 5,
scan_lifetime_ms: 120_000,
current_idle_eligible: false,
run_id: "root".into(),
revision: 2,
section: bcode_workflow_view_models::WorkflowActivitySection::Attempts,
attempts: vec![bcode_workflow_view_models::WorkflowActivityAttempt {
dispatch_identity: "dispatch".into(),
node_id: "node".into(),
activation_id: "activation".into(),
attempt: 1,
terminal: false,
session_id: Some("session".into()),
observation: bcode_workflow_view_models::AttemptObservation::Unavailable {
reason: bcode_workflow_view_models::WorkflowAttemptUnavailableReason::UnknownDispatchSource,
},
progress: Some(SessionProgressSnapshot {
version: SESSION_PROGRESS_VERSION,
session_id: "session".into(),
observation: SessionProgressObservation::Unavailable {
reason: SessionProgressUnavailableReason::CoverageUnavailable,
},
}),
}],
children: vec![],
next_cursor: None,
inventory_exhausted: true,
subtree_coverage_complete: true,
issues: vec!["live coverage unavailable".into()],
};
let encoded = bcode_ipc::response_envelope(
1,
&Response::Ok(ResponsePayload::WorkflowActivityPage { page: page.clone() }),
)
.unwrap();
match decode_response(&encoded.payload).unwrap() {
Response::Ok(ResponsePayload::WorkflowActivityPage { page: decoded }) => {
assert_eq!(decoded, page);
assert!(decoded.subtree_coverage_complete);
assert!(!decoded.current_idle_eligible);
assert_eq!(
decoded.attempts[0]
.progress
.as_ref()
.unwrap()
.advanced_since(decoded.attempts[0].progress.as_ref().unwrap()),
None
);
}
_ => panic!("expected typed workflow activity page"),
}
}
#[test]
fn rejects_wrong_identity_future_versions_and_unexpected_payloads() {
let id = SessionId::new();
assert!(matches!(
decode_session_progress_response(wire_decode(snapshot(SessionId::new())), id),
Err(ClientError::Protocol(_))
));
let mut legacy = snapshot(id);
legacy.version = 1;
assert!(matches!(
decode_session_progress_response(wire_decode(legacy), id),
Err(ClientError::Protocol(_))
));
let mut future = snapshot(id);
future.version += 1;
assert!(matches!(
decode_session_progress_response(wire_decode(future), id),
Err(ClientError::Protocol(_))
));
assert!(matches!(
decode_session_progress_response(ResponsePayload::PresentationNoteAppended, id),
Err(ClientError::UnexpectedResponse)
));
}
}
#[cfg(test)]
mod client_timeout_tests {
#[test]
fn pending_exchange_selection_preserves_schema_and_rejects_ambiguity() {
let exchange = bcode_session_models::PendingToolExchangeSummary {
interaction_route: None,
session_id: bcode_session_models::SessionId::new(),
request: bcode_session_models::ToolExchangeRequest {
invocation_id: "invocation".to_owned(),
exchange_id: "target".to_owned(),
producer_id: "producer".to_owned(),
schema: "future.schema".to_owned(),
schema_version: u32::MAX,
payload: serde_json::json!({ "opaque": [1, "value"] }),
response_policy: bcode_session_models::ToolExchangeResponsePolicy::Required,
},
};
let mut other = exchange.clone();
other.request.exchange_id = "other".to_owned();
let selected = super::select_pending_tool_exchange(vec![other, exchange.clone()], "target")
.unwrap()
.unwrap();
assert_eq!(
serde_json::to_value(selected).unwrap(),
serde_json::to_value(exchange.clone()).unwrap()
);
for conflicting_session in [false, true] {
let mut duplicate = exchange.clone();
if conflicting_session {
duplicate.session_id = bcode_session_models::SessionId::new();
duplicate.request.payload = serde_json::json!({ "secret": "must-not-leak" });
}
let error =
super::select_pending_tool_exchange(vec![exchange.clone(), duplicate], "target")
.unwrap_err();
assert!(matches!(error, super::ClientError::Protocol(ref message)
if message == "ambiguous pending interaction identifier"));
}
assert!(
super::select_pending_tool_exchange(vec![exchange], "missing")
.unwrap()
.is_none()
);
assert!(
super::select_pending_tool_exchange(vec![], "target")
.unwrap()
.is_none()
);
}
use super::{
BcodeClient, ClientError, resolve_path_from, session_open_attach_readiness,
terminal_session_open_error_message,
};
use bcode_session_models::{
SessionId, SessionMigrationProgress, SessionMigrationStage, SessionOpenOperationId,
SessionOpenOperationSnapshot, SessionOpenTerminalOutcome,
};
use bcode_session_search::{
SessionSearchBackfillOperationState, SessionSearchBackfillOperationStatus,
};
use std::path::Path;
use std::time::Duration;
#[tokio::test]
async fn typed_definition_inspection_rejects_untrusted_responses() {
use bcode_workflow::WorkflowAuthoringApplication as _;
for (identity, version) in [("requested", 1), ("wrong", 1), ("requested", 2)] {
let workflow = bcode_workflow::WorkflowBuilder::new(
"requested",
bcode_workflow::Step::task("node", |value: u32, _context| async move { Ok(value) }),
)
.build()
.expect("valid workflow");
let value = serde_json::to_value(workflow.definition()).expect("definition value");
let mut definition = bcode_workflow::StoredWorkflowDefinition {
definition_id: identity.to_string(),
version,
checksum_sha256: bcode_workflow::workflow_canonical_value_sha256(&value)
.expect("checksum"),
definition_json: serde_json::to_string(&value).expect("definition JSON"),
};
assert!(
definition.definition().is_ok(),
"mismatch fixtures must otherwise validate"
);
if identity == "requested" && version == 1 {
definition.definition_json = "private malformed content".to_string();
}
let directory =
std::path::PathBuf::from(format!("/tmp/bci-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&directory).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(directory.join("inspect.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept");
let hello = bcode_ipc::recv_envelope(&mut stream).await.expect("hello");
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(
bcode_ipc::CURRENT_PROTOCOL_VERSION,
),
client_id: bcode_session_models::ClientId::new(),
daemon: matching_daemon_status(),
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response)
.expect("hello envelope"),
)
.await
.expect("hello reply");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("inspect request");
let response = bcode_ipc::Response::Ok(
bcode_ipc::ResponsePayload::WorkflowDefinitionDescription {
definition: Some(definition),
},
);
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(request.request_id, &response)
.expect("response envelope"),
)
.await
.expect("reply");
});
let error = BcodeClient::new(endpoint)
.inspect_workflow_definition("requested".to_string(), 1)
.await
.expect_err("reject untrusted content");
let failure = bcode_workflow::WorkflowAuthoringFailure::StateUnavailable;
assert!(
matches!(error, ClientError::Server { code, message } if code == failure.code() && message == failure.to_string())
);
server.await.expect("server");
}
}
fn matching_daemon_status() -> bcode_ipc::DaemonStatus {
let (_path, digest) = bcode_daemon_lifecycle::current_executable_identity()
.expect("current executable identity");
bcode_ipc::DaemonStatus {
namespace: bcode_ipc::daemon_namespace(),
protocol_version: u32::from(bcode_ipc::CURRENT_PROTOCOL_VERSION),
artifact_id: Some(bcode_ipc::ArtifactId::current()),
build_fingerprint: bcode_ipc::BUILD_FINGERPRINT.to_owned(),
executable_digest: Some(digest),
storage_writer_epoch: Some(bcode_ipc::CURRENT_SESSION_STORAGE_WRITER_EPOCH),
session_event_schema_version: Some(
bcode_session_models::CURRENT_SESSION_EVENT_SCHEMA_VERSION,
),
state_location_id: Some(bcode_ipc::state_location_id()),
..bcode_ipc::DaemonStatus::default()
}
}
#[cfg(unix)]
#[tokio::test]
async fn provider_validation_transmits_normalized_payload_on_both_dispatch_paths() {
use super::{
ClientRuntimeContext, PluginServiceResponse, Request, Response, ResponsePayload,
};
use std::collections::BTreeMap;
let context = ClientRuntimeContext {
selected_provider_plugin_id: Some("bcode.fake-provider".to_owned()),
provider_context: bcode_model::ProviderRequestContext {
auth_profile: Some("fixture-account".to_owned()),
settings: BTreeMap::from([("fixture".to_owned(), "advanced-chat".to_owned())]),
env: BTreeMap::from([(
"FIXTURE_TOKEN".to_owned(),
"synthetic-test-token".to_owned(),
)]),
..Default::default()
},
..Default::default()
};
for provider in [Some("bcode.fake-provider".to_owned()), None] {
let directory =
std::path::PathBuf::from(format!("/tmp/bcv-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&directory).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(directory.join("validation.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let expected_provider = provider.clone();
let expected_context = context.provider_context.clone();
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.unwrap();
let hello = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let response = Response::Ok(ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(
bcode_ipc::CURRENT_PROTOCOL_VERSION,
),
client_id: bcode_session_models::ClientId::new(),
daemon: matching_daemon_status(),
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response).unwrap(),
)
.await
.unwrap();
let envelope = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let (interface_id, operation, payload) =
match bcode_ipc::decode_request(&envelope.payload).unwrap() {
Request::InvokePluginService {
plugin_id,
interface_id,
operation,
payload,
} => {
assert_eq!(Some(plugin_id), expected_provider);
(interface_id, operation, payload)
}
Request::CallPluginService {
interface_id,
operation,
payload,
} => {
assert!(expected_provider.is_none());
(interface_id, operation, payload)
}
other => panic!("unexpected request: {other:?}"),
};
assert_eq!(interface_id, bcode_model::MODEL_PROVIDER_INTERFACE_ID);
assert_eq!(operation, bcode_model::OP_VALIDATE_CONFIG);
let request: bcode_model::ValidateConfigRequest =
serde_json::from_slice(&payload).unwrap();
assert_eq!(request.provider_context, expected_context);
assert!(request.config.is_empty());
assert!(request.profile.is_none());
let response = Response::Ok(ResponsePayload::PluginServiceResult {
response: PluginServiceResponse {
payload: br#"{"valid":true}"#.to_vec(),
error: None,
},
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(envelope.request_id, &response).unwrap(),
)
.await
.unwrap();
});
let validation = BcodeClient::new(endpoint)
.with_runtime_context(Some(context.clone()))
.validate_model_config(provider)
.await
.unwrap();
assert!(validation.valid);
server.await.unwrap();
std::fs::remove_dir_all(directory).unwrap();
}
}
#[test]
fn daemon_identity_rejects_same_artifact_with_different_executable_digest() {
let matching = matching_daemon_status();
let resigned = bcode_ipc::DaemonStatus {
executable_digest: Some("different-signed-executable-digest".to_owned()),
..matching
};
assert!(BcodeClient::verify_daemon_identity(&resigned).is_err());
}
#[test]
fn daemon_identity_matrix_rejects_every_incompatible_capability() {
let matching = matching_daemon_status();
BcodeClient::verify_daemon_identity(&matching).expect("matching daemon");
let cases = [
bcode_ipc::DaemonStatus {
artifact_id: Some(
bcode_ipc::ArtifactId::parse("other-artifact")
.expect("other artifact identity"),
),
..matching.clone()
},
bcode_ipc::DaemonStatus {
protocol_version: matching.protocol_version.saturating_add(1),
..matching.clone()
},
bcode_ipc::DaemonStatus {
build_fingerprint: "other-build".to_owned(),
..matching.clone()
},
bcode_ipc::DaemonStatus {
storage_writer_epoch: matching.storage_writer_epoch.map(|value| value + 1),
..matching.clone()
},
bcode_ipc::DaemonStatus {
state_location_id: Some("other-state-location".to_owned()),
..matching.clone()
},
bcode_ipc::DaemonStatus {
state_location_id: None,
..matching.clone()
},
bcode_ipc::DaemonStatus {
session_event_schema_version: matching
.session_event_schema_version
.map(|value| value + 1),
..matching.clone()
},
bcode_ipc::DaemonStatus {
storage_writer_epoch: None,
session_event_schema_version: None,
..matching
},
];
for daemon in cases {
let error = BcodeClient::verify_daemon_identity(&daemon)
.expect_err("incompatible capability must fail before requests");
let ClientError::IncompatibleDaemon { message } = error else {
panic!("expected incompatible daemon");
};
assert!(message.contains("artifact="));
assert!(message.contains("session_event_schema="));
assert!(message.contains("storage_writer_epoch="));
assert!(message.contains("protocol="));
assert!(message.contains("build="));
}
}
#[cfg(unix)]
#[tokio::test]
async fn interrupted_dispatched_page_cancels_successor_before_returning() {
for final_page in [false, true] {
interrupted_page_successor(final_page).await;
}
}
#[cfg(unix)]
async fn interrupted_page_successor(final_page: bool) {
let dir = std::path::PathBuf::from(format!("/tmp/bcc-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&dir).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(dir.join("catalog.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let (dispatched, interrupted) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
let mut signal = Some(dispatched);
let mut held_page = None;
for step in 0..3 {
let mut stream = listener.accept().await.unwrap();
let hello = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion::current(),
client_id: bcode_session_models::ClientId::new(),
daemon: matching_daemon_status(),
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response).unwrap(),
)
.await
.unwrap();
let request = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let operation = bcode_ipc::decode_request(&request.payload).unwrap();
if step == 0 {
assert!(matches!(
operation,
bcode_ipc::Request::WorkflowLaunchCatalog(_)
));
held_page = Some((stream, request.request_id));
signal.take().unwrap().send(()).unwrap();
continue;
}
let expected = if step == 1 { "previous" } else { "successor" };
assert!(
matches!(operation, bcode_ipc::Request::CancelWorkflowDiscovery { token } if token == expected)
);
let response = bcode_ipc::Response::Ok(
bcode_ipc::ResponsePayload::WorkflowDiscoveryCancelled {
released: step == 2,
},
);
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(request.request_id, &response).unwrap(),
)
.await
.unwrap();
if step == 1 {
let (mut page_stream, id) = held_page.take().unwrap();
let response = bcode_ipc::Response::Ok(
bcode_ipc::ResponsePayload::WorkflowLaunchCatalog {
page: bcode_workflow::WorkflowLaunchCatalogPage {
detail_token: final_page.then(|| "successor".into()),
version: bcode_workflow::WORKFLOW_LAUNCH_CATALOG_VERSION,
items: vec![],
diagnostics: vec![],
discovery_token: None,
next_cursor: (!final_page).then(|| {
bcode_workflow::WorkflowLaunchCatalogCursor {
title: "next".into(),
source_key: "next".into(),
result_token: Some("successor".into()),
}
}),
},
},
);
bcode_ipc::send_envelope(
&mut page_stream,
&bcode_ipc::response_envelope(id, &response).unwrap(),
)
.await
.unwrap();
}
}
});
let client = BcodeClient::new(endpoint);
let request = serde_json::from_value(serde_json::json!({
"version": bcode_workflow::WORKFLOW_LAUNCH_CATALOG_VERSION, "workspace": ".", "limit": 1,
"cursor": {"title": "previous", "source_key": "previous", "result_token": "previous"}
})).unwrap();
let result = tokio::time::timeout(
std::time::Duration::from_secs(5),
Box::pin(client.workflow_launch_catalog_until(request, async {
interrupted.await.unwrap();
})),
)
.await
.unwrap()
.unwrap();
assert!(result.is_none());
server.await.unwrap();
assert!(!dir.join("catalog.sock").exists());
std::fs::remove_dir(dir).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn interrupted_retained_page_does_not_dispatch_after_cancellation() {
let dir = std::path::PathBuf::from(format!("/tmp/bcc-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&dir).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(dir.join("catalog.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let server = tokio::spawn(async move {
{
let mut stream = listener.accept().await.unwrap();
let hello = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion::current(),
client_id: bcode_session_models::ClientId::new(),
daemon: matching_daemon_status(),
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response).unwrap(),
)
.await
.unwrap();
let request = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let operation = bcode_ipc::decode_request(&request.payload).unwrap();
assert!(
matches!(operation, bcode_ipc::Request::CancelWorkflowDiscovery { token } if token == "previous")
);
let payload =
bcode_ipc::ResponsePayload::WorkflowDiscoveryCancelled { released: true };
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(
request.request_id,
&bcode_ipc::Response::Ok(payload),
)
.unwrap(),
)
.await
.unwrap();
}
});
let client = BcodeClient::new(endpoint);
let request = serde_json::from_value(serde_json::json!({
"version": bcode_workflow::WORKFLOW_LAUNCH_CATALOG_VERSION, "workspace": ".", "limit": 1,
"cursor": {"title": "previous", "source_key": "previous", "result_token": "previous"}
})).unwrap();
let result = tokio::time::timeout(
std::time::Duration::from_secs(5),
Box::pin(client.workflow_launch_catalog_until(request, std::future::ready(()))),
)
.await
.unwrap()
.unwrap();
assert!(result.is_none());
server.await.unwrap();
assert!(!dir.join("catalog.sock").exists());
std::fs::remove_dir(dir).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn interrupted_catalog_admission_releases_token_without_discovery() {
let dir = std::path::PathBuf::from(format!("/tmp/bcc-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&dir).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(dir.join("catalog.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let server = tokio::spawn(async move {
for admission in [true, false] {
let mut stream = listener.accept().await.unwrap();
let hello = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion::current(),
client_id: bcode_session_models::ClientId::new(),
daemon: matching_daemon_status(),
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response).unwrap(),
)
.await
.unwrap();
let request = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let operation = bcode_ipc::decode_request(&request.payload).unwrap();
let payload = if admission {
assert!(
matches!(operation, bcode_ipc::Request::WorkflowLaunchCatalog(ref request)
if request.incremental && request.discovery_token.is_none())
);
bcode_ipc::ResponsePayload::WorkflowLaunchCatalog {
page: bcode_workflow::WorkflowLaunchCatalogPage {
detail_token: None,
version: bcode_workflow::WORKFLOW_LAUNCH_CATALOG_VERSION,
discovery_token: Some("admitted".to_string()),
items: vec![],
diagnostics: vec![],
next_cursor: None,
},
}
} else {
assert!(
matches!(operation, bcode_ipc::Request::CancelWorkflowDiscovery { token } if token == "admitted")
);
bcode_ipc::ResponsePayload::WorkflowDiscoveryCancelled { released: true }
};
let response = bcode_ipc::Response::Ok(payload);
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(request.request_id, &response).unwrap(),
)
.await
.unwrap();
}
});
let client = BcodeClient::new(endpoint);
let request = serde_json::from_value(serde_json::json!({
"version": bcode_workflow::WORKFLOW_LAUNCH_CATALOG_VERSION,
"workspace": ".", "limit": 1,
}))
.unwrap();
let result = tokio::time::timeout(
std::time::Duration::from_secs(5),
Box::pin(client.workflow_launch_catalog_until(request, std::future::ready(()))),
)
.await
.unwrap()
.unwrap();
assert!(result.is_none());
server.await.unwrap();
assert!(!dir.join("catalog.sock").exists());
std::fs::remove_dir(dir).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn custom_endpoint_warm_handshake_uses_one_connection_without_startup() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcw-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("warm.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let expected_client_id = bcode_session_models::ClientId::new();
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let hello = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("receive hello");
assert!(matches!(
bcode_ipc::decode_request(&hello.payload).expect("decode hello"),
bcode_ipc::Request::Hello { artifact_id: Some(artifact_id), .. }
if artifact_id == bcode_ipc::ArtifactId::current()
));
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion::current(),
client_id: expected_client_id,
daemon: matching_daemon_status(),
});
let envelope =
bcode_ipc::response_envelope(hello.request_id, &response).expect("hello response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send hello");
});
let client = BcodeClient::new(endpoint);
let connection = client
.connect("custom-warm-test")
.await
.expect("warm connect");
assert_eq!(connection.client_id(), Some(expected_client_id));
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn missing_custom_endpoint_requires_running_without_auto_start() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcm-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let socket_path = socket_dir.join("missing.sock");
let client = BcodeClient::new(bcode_ipc::IpcEndpoint::unix_socket(socket_path.clone()));
let error = client
.connect("custom-missing-test")
.await
.expect_err("custom endpoint must require a running daemon");
assert!(error.is_daemon_unavailable());
assert!(!socket_path.exists());
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn connection_timeout_is_distinct_and_does_not_trigger_auto_start() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcc-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("connect-timeout.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let _hello = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("receive hello");
tokio::time::sleep(Duration::from_millis(100)).await;
});
let client = BcodeClient::new(endpoint)
.with_daemon_availability(super::DaemonAvailability::AutoStart)
.with_connect_timeout(Duration::from_millis(10))
.with_request_timeout(Duration::from_secs(1));
let error = client
.connect("connect-timeout-test")
.await
.expect_err("unresponsive reachable endpoint must time out");
assert!(matches!(
error,
ClientError::ConnectTimeout { timeout } if timeout == Duration::from_millis(10)
));
assert!(!error.is_daemon_unavailable());
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn mismatched_artifact_hello_is_rejected_explicitly() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bci-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("artifact.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let hello = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("hello request");
let daemon = bcode_ipc::DaemonStatus {
artifact_id: Some(
bcode_ipc::ArtifactId::parse("foreign-artifact")
.expect("foreign artifact identity"),
),
..matching_daemon_status()
};
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(bcode_ipc::CURRENT_PROTOCOL_VERSION),
client_id: bcode_session_models::ClientId::new(),
daemon,
});
let envelope = bcode_ipc::response_envelope(hello.request_id, &response)
.expect("hello response envelope");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send hello response");
});
let client = BcodeClient::new(endpoint)
.with_daemon_availability(super::DaemonAvailability::RequireRunning);
let error = client
.connect("artifact-mismatch-test")
.await
.expect_err("foreign artifact must be rejected");
assert!(matches!(error, ClientError::IncompatibleDaemon { .. }));
assert!(error.to_string().contains("foreign-artifact"));
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn accepted_compaction_outlives_request_deadline() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcc-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("client.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("request");
for payload in [
bcode_ipc::ResponsePayload::SessionCompactionAccepted,
bcode_ipc::ResponsePayload::SessionCompacted {
compacted: true,
message: "compacted".to_owned(),
},
] {
let terminal =
matches!(payload, bcode_ipc::ResponsePayload::SessionCompacted { .. });
if terminal {
tokio::time::sleep(Duration::from_millis(150)).await;
}
let envelope = bcode_ipc::response_envelope(
request.request_id,
&bcode_ipc::Response::Ok(payload),
)
.expect("response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send");
}
});
let stream = bcode_ipc::LocalIpcStream::connect(&endpoint)
.await
.expect("connect");
let mut connection = super::ClientConnection {
stream,
next_request_id: 1,
client_id: None,
pending_events: std::collections::VecDeque::new(),
request_timeout: Duration::from_millis(50),
reconnect_client: None,
reconnect_name: std::sync::Arc::from(""),
restore_state: super::ConnectionRestoreState::default(),
};
assert!(matches!(
connection
.send_request(bcode_ipc::Request::CompactSession {
session_id: SessionId::new(),
})
.await,
Ok(bcode_ipc::ResponsePayload::SessionCompacted {
compacted: true,
..
})
));
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn unrelated_events_remain_buffered_in_fifo_order_during_requests() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bce-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("client.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("request envelope");
for revision in [11, 12] {
let event = bcode_ipc::Event::SessionCatalogUpdated { revision };
let envelope = bcode_ipc::event_envelope(&event).expect("event envelope");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send event");
}
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Pong);
let envelope = bcode_ipc::response_envelope(request.request_id, &response)
.expect("response envelope");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send response");
});
let stream = bcode_ipc::LocalIpcStream::connect(&endpoint)
.await
.expect("connect");
let mut connection = super::ClientConnection {
stream,
next_request_id: 1,
client_id: None,
pending_events: std::collections::VecDeque::new(),
request_timeout: Duration::from_secs(1),
reconnect_client: None,
reconnect_name: std::sync::Arc::from(""),
restore_state: super::ConnectionRestoreState::default(),
};
assert!(matches!(
connection.send_request(bcode_ipc::Request::Ping).await,
Ok(bcode_ipc::ResponsePayload::Pong)
));
for expected in [11, 12] {
assert_eq!(
connection.recv_event().await.expect("buffered event"),
bcode_ipc::Event::SessionCatalogUpdated { revision: expected }
);
}
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("event socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn stateless_catalog_requests_preserve_directory_and_sources() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcs-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("catalog.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let expected = std::env::current_dir().unwrap().join("relative-workspace");
let server = tokio::spawn(async move {
for refresh in [false, true] {
let mut stream = listener.accept().await.unwrap();
let hello = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(
bcode_ipc::CURRENT_PROTOCOL_VERSION,
),
client_id: bcode_session_models::ClientId::new(),
daemon: matching_daemon_status(),
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response).unwrap(),
)
.await
.unwrap();
let request = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let decoded = bcode_ipc::decode_request(&request.payload).unwrap();
let payload = if refresh {
match decoded {
bcode_ipc::Request::RefreshSessionCatalog {
working_directory,
sources,
} => {
assert_eq!(working_directory.as_ref(), Some(&expected));
assert_eq!(sources, Some(vec!["local".to_owned()]));
}
other => panic!("unexpected refresh request: {other:?}"),
}
bcode_ipc::ResponsePayload::SessionCatalogRefreshed {
sessions: vec![],
catalog_status: bcode_session_models::SessionCatalogStatus::Loaded,
catalog_sources: vec![],
catalog_revision: 2,
}
} else {
assert!(
matches!(decoded, bcode_ipc::Request::ListSessions { working_directory } if working_directory == expected)
);
bcode_ipc::ResponsePayload::SessionList {
sessions: vec![],
catalog_status: bcode_session_models::SessionCatalogStatus::Loaded,
catalog_sources: vec![],
catalog_revision: 1,
}
};
let response = bcode_ipc::Response::Ok(payload);
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(request.request_id, &response).unwrap(),
)
.await
.unwrap();
}
});
let client = BcodeClient::new(endpoint).with_request_timeout(Duration::from_secs(5));
let listed = client
.list_sessions_in_working_directory("relative-workspace".into())
.await
.unwrap();
assert_eq!(listed.catalog_revision, 1);
let refreshed = client
.refresh_session_catalog_in_working_directory(
"relative-workspace".into(),
Some(vec!["local".to_owned()]),
)
.await
.unwrap();
assert_eq!(refreshed.catalog_revision, 2);
server.await.unwrap();
std::fs::remove_dir_all(socket_dir).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn exact_cancellation_preserves_target_and_acknowledgement_over_ipc() {
use bcode_ipc::{Request, Response, ResponsePayload};
use bcode_session_models::ClientId;
let socket_dir = std::path::PathBuf::from(format!("/tmp/bcc-{}", SessionId::new()));
std::fs::create_dir_all(&socket_dir).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("cancel.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let session_id = SessionId::new();
let server = tokio::spawn(async move {
for cancelled in [false, true] {
let mut stream = listener.accept().await.unwrap();
let hello = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let response = Response::Ok(ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(
bcode_ipc::CURRENT_PROTOCOL_VERSION,
),
client_id: ClientId::new(),
daemon: matching_daemon_status(),
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response).unwrap(),
)
.await
.unwrap();
let request = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
assert_eq!(
bcode_ipc::decode_request(&request.payload).unwrap(),
Request::CancelSessionTurn {
session_id,
expected_turn_id: Some("turn-具体".to_owned()),
clear_queue: false,
}
);
let response =
Response::Ok(ResponsePayload::TurnCancellationRequested { cancelled });
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(request.request_id, &response).unwrap(),
)
.await
.unwrap();
}
});
let client = BcodeClient::new(endpoint).with_request_timeout(Duration::from_secs(5));
for expected in [false, true] {
assert_eq!(
client
.cancel_exact_session_turn(session_id, "turn-具体".to_owned())
.await
.unwrap(),
expected
);
}
server.await.unwrap();
std::fs::remove_dir_all(socket_dir).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn catalog_relative_directory_is_resolved_before_transport() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcd-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("catalog.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let expected = std::env::current_dir().unwrap().join("relative-workspace");
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.unwrap();
let request = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let decoded = bcode_ipc::decode_request(&request.payload).unwrap();
assert!(
matches!(decoded, bcode_ipc::Request::ListSessions { working_directory } if working_directory == expected)
);
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::SessionList {
sessions: vec![],
catalog_status: bcode_session_models::SessionCatalogStatus::Loaded,
catalog_sources: vec![],
catalog_revision: 1,
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(request.request_id, &response).unwrap(),
)
.await
.unwrap();
});
let stream = bcode_ipc::LocalIpcStream::connect(&endpoint).await.unwrap();
let mut connection = super::ClientConnection {
stream,
next_request_id: 1,
client_id: None,
pending_events: std::collections::VecDeque::new(),
request_timeout: Duration::from_secs(5),
reconnect_client: None,
reconnect_name: std::sync::Arc::from(""),
restore_state: super::ConnectionRestoreState::default(),
};
let result = connection
.list_sessions_in_working_directory("relative-workspace".into())
.await
.unwrap();
assert_eq!(result.catalog_revision, 1);
server.await.unwrap();
std::fs::remove_dir_all(socket_dir).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn catalog_watcher_delivers_queued_completion_and_closes_on_drop() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcw-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("catalog.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.unwrap();
let hello = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(bcode_ipc::CURRENT_PROTOCOL_VERSION),
client_id: bcode_session_models::ClientId::new(),
daemon: matching_daemon_status(),
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response).unwrap(),
)
.await
.unwrap();
let subscription = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
assert!(matches!(
bcode_ipc::decode_request(&subscription.payload).unwrap(),
bcode_ipc::Request::SubscribeCatalogUpdates
));
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(
subscription.request_id,
&bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::CatalogUpdatesSubscribed),
)
.unwrap(),
)
.await
.unwrap();
for revision in [1, 3] {
let request = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
assert!(matches!(
bcode_ipc::decode_request(&request.payload).unwrap(),
bcode_ipc::Request::ListSessionsWithScope {
scope: bcode_session_models::SessionDiscoveryScope::WorkingDirectory(_)
}
));
if revision == 1 {
for queued in [1, 3] {
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::event_envelope(&bcode_ipc::Event::SessionCatalogUpdated {
revision: queued,
})
.unwrap(),
)
.await
.unwrap();
}
}
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::SessionList {
sessions: vec![],
catalog_status: if revision == 1 {
bcode_session_models::SessionCatalogStatus::Loading
} else {
bcode_session_models::SessionCatalogStatus::Loaded
},
catalog_sources: vec![],
catalog_revision: revision,
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(request.request_id, &response).unwrap(),
)
.await
.unwrap();
}
assert!(bcode_ipc::recv_envelope(&mut stream).await.is_err());
});
let client = BcodeClient::new(endpoint).with_request_timeout(Duration::from_secs(5));
let mut watcher = client.watch_session_catalog().await.unwrap();
assert_eq!(
watcher.initial_snapshot().await.unwrap().catalog_status,
bcode_session_models::SessionCatalogStatus::Loading
);
let completed = tokio::time::timeout(Duration::from_secs(5), watcher.next_snapshot())
.await
.unwrap()
.unwrap();
assert_eq!(completed.catalog_revision, 3);
assert_eq!(
completed.catalog_status,
bcode_session_models::SessionCatalogStatus::Loaded
);
drop(watcher);
tokio::time::timeout(Duration::from_secs(5), server)
.await
.unwrap()
.unwrap();
assert!(!socket_dir.join("catalog.sock").exists());
std::fs::remove_dir(socket_dir).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn long_poll_transport_timeout_is_distinct_from_operation_failure() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bct-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("timeout.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let _request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("wait request");
tokio::time::sleep(Duration::from_millis(100)).await;
});
let stream = bcode_ipc::LocalIpcStream::connect(&endpoint)
.await
.expect("connect");
let mut connection = super::ClientConnection {
stream,
next_request_id: 1,
client_id: None,
pending_events: std::collections::VecDeque::new(),
request_timeout: Duration::from_millis(10),
reconnect_client: None,
reconnect_name: std::sync::Arc::from(""),
restore_state: super::ConnectionRestoreState::default(),
};
let session_id = SessionId::new();
assert!(matches!(
connection
.wait_session_open_progress(
session_id,
SessionOpenOperationId::new(),
0,
Duration::from_secs(5),
)
.await,
Err(ClientError::RequestTimeout { timeout })
if timeout == Duration::from_millis(10)
));
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("timeout socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn artifact_client_rejects_invalid_lengths_before_connecting() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bca-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("bounds.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let client = BcodeClient::new(endpoint).with_request_timeout(Duration::from_millis(50));
for length in [
0,
bcode_session_models::MAX_SESSION_ARTIFACT_RANGE_BYTES + 1,
u32::MAX,
] {
let error = client
.session_artifact_range(
SessionId::new(),
"private-artifact".into(),
"private-reference".into(),
0,
length,
)
.await
.unwrap_err();
assert!(matches!(error, ClientError::Protocol(ref message)
if message == &format!("artifact range length must be between 1 and {} bytes", bcode_session_models::MAX_SESSION_ARTIFACT_RANGE_BYTES)));
}
assert!(
tokio::time::timeout(Duration::from_millis(50), listener.accept())
.await
.is_err()
);
drop(listener);
std::fs::remove_dir_all(socket_dir).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn artifact_client_validates_binary_and_eof_ranges_over_ipc() {
for (offset, bytes, valid) in [
(0, vec![], false),
(0, vec![0, 255], true),
(2, vec![], true),
(3, vec![], false),
(u64::MAX, vec![], false),
] {
let expected_bytes = bytes.clone();
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bca-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("range.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let hello = bcode_ipc::recv_envelope(&mut stream).await.expect("hello");
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(
bcode_ipc::CURRENT_PROTOCOL_VERSION,
),
client_id: bcode_session_models::ClientId::new(),
daemon: matching_daemon_status(),
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response)
.expect("hello envelope"),
)
.await
.expect("send hello");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("range request");
let response =
bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::SessionArtifactRange {
range: super::SessionArtifactRange {
artifact_id: "artifact".to_owned(),
reference_key: "reference".to_owned(),
content_type: None,
offset,
total_bytes: 2,
reference_bytes: Some(2),
reference_revision: 1,
finalized: true,
finalized_event_seq: Some(1),
availability: None,
complete: Some(true),
checksum_sha256: None,
bytes,
},
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(request.request_id, &response)
.expect("range envelope"),
)
.await
.expect("send range");
});
let client = BcodeClient::new(endpoint);
let result = client
.session_artifact_range(
SessionId::new(),
"artifact".to_owned(),
"reference".to_owned(),
offset,
2,
)
.await;
if valid {
let range = result.expect("valid range");
assert_eq!(range.bytes, expected_bytes);
assert_eq!(range.offset, offset);
assert_eq!(range.next_offset(), 2);
assert!(range.is_eof());
assert_eq!(range.finalized_event_seq, Some(1));
} else {
assert!(matches!(result, Err(ClientError::Protocol(_))));
}
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
}
#[cfg(unix)]
#[tokio::test]
async fn reconnect_rejects_replacement_executable_before_restoring_session() {
let directory = std::path::PathBuf::from(format!("/tmp/bcp-{}", SessionId::new()));
std::fs::create_dir_all(&directory).unwrap();
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(directory.join("pin.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).unwrap();
let server = tokio::spawn(async move {
for replacement in [false, true] {
let mut stream = listener.accept().await.unwrap();
let hello = bcode_ipc::recv_envelope(&mut stream).await.unwrap();
let mut daemon = matching_daemon_status();
if replacement {
daemon.executable_digest = Some("replacement".into());
}
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion::current(),
client_id: bcode_session_models::ClientId::new(),
daemon,
});
bcode_ipc::send_envelope(
&mut stream,
&bcode_ipc::response_envelope(hello.request_id, &response).unwrap(),
)
.await
.unwrap();
if replacement {
assert!(bcode_ipc::recv_envelope(&mut stream).await.is_err());
}
}
});
let client = BcodeClient::new(endpoint);
let mut connection = client.connect("pin-test").await.unwrap();
assert!(matches!(
connection.reconnect_and_restore().await,
Err(ClientError::IncompatibleDaemon { .. })
));
drop(connection);
server.await.unwrap();
std::fs::remove_dir(directory).unwrap();
}
#[cfg(unix)]
#[tokio::test]
async fn workflow_watcher_requests_resync_after_quiet_reconnect() {
let socket_dir = std::path::PathBuf::from(format!("/tmp/bcw-{}", SessionId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("watch.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let daemon = matching_daemon_status();
let (stop, stopped) = tokio::sync::oneshot::channel();
let server = tokio::spawn(async move {
for sequence in [10, 0] {
let mut stream = listener.accept().await.expect("connection");
let hello = bcode_ipc::recv_envelope(&mut stream).await.expect("hello");
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(
bcode_ipc::CURRENT_PROTOCOL_VERSION,
),
client_id: bcode_session_models::ClientId::new(),
daemon: daemon.clone(),
});
let envelope = bcode_ipc::response_envelope(hello.request_id, &response)
.expect("hello response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send hello");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("subscription");
let response = bcode_ipc::response_envelope(
request.request_id,
&bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::WorkflowRunsSubscribed {
after_sequence: sequence,
}),
)
.expect("response");
bcode_ipc::send_envelope(&mut stream, &response)
.await
.expect("subscribe");
if sequence == 0 {
stopped.await.expect("client observed resync");
break;
}
}
});
let client = BcodeClient::new(endpoint);
let mut watcher = client.watch_workflow_runs().await.expect("watcher");
assert!(matches!(
tokio::time::timeout(Duration::from_secs(5), watcher.next_event())
.await
.expect("deadline")
.expect("reconnect"),
super::WorkflowRunWatchEvent::ResyncRequired
));
assert_eq!(watcher.sequence.last_observed(), Some(0));
stop.send(()).expect("stop");
server.await.expect("server");
std::fs::remove_dir_all(socket_dir).expect("cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn preparation_recovers_retained_operation_after_transport_interruption() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcr-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("reconnect.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let session_id = SessionId::new();
let operation_id = SessionOpenOperationId::new();
let snapshot = |revision, terminal| SessionOpenOperationSnapshot {
operation_id,
revision,
session_id,
source_writer_epoch: Some(3),
target_writer_epoch: 4,
progress: SessionMigrationProgress {
stage: if terminal {
SessionMigrationStage::Complete
} else {
SessionMigrationStage::CopyingBackup
},
completed_units: Some(revision),
total_units: Some(2),
unit: Some(bcode_session_models::SessionMigrationProgressUnit::Files),
message: "migration".to_owned(),
},
outcome: terminal.then_some(SessionOpenTerminalOutcome::Ready),
backup_path: None,
};
let initial = snapshot(1, false);
let terminal = snapshot(2, true);
let server_terminal = terminal.clone();
let daemon = matching_daemon_status();
let server = tokio::spawn(async move {
for (connection_index, prepared) in [initial, server_terminal].into_iter().enumerate() {
let mut stream = listener.accept().await.expect("accept client");
let hello = bcode_ipc::recv_envelope(&mut stream).await.expect("hello");
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(
bcode_ipc::CURRENT_PROTOCOL_VERSION,
),
client_id: bcode_session_models::ClientId::new(),
daemon: daemon.clone(),
});
let envelope = bcode_ipc::response_envelope(hello.request_id, &response)
.expect("hello response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send hello");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("preparation request");
let response =
bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::SessionOpenPrepared {
snapshot: prepared,
});
let envelope = bcode_ipc::response_envelope(request.request_id, &response)
.expect("preparation response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send preparation");
if connection_index == 0 {
let _wait = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("wait request before disconnect");
}
}
});
let client = BcodeClient::new(endpoint).with_request_timeout(Duration::from_secs(1));
let mut connection = client.connect("reconnect-test").await.expect("connect");
let mut revisions = Vec::new();
let recovered = connection
.prepare_session_open_until_terminal(session_id, |snapshot| {
revisions.push(snapshot.revision);
})
.await
.expect("recover preparation");
assert_eq!(recovered, terminal);
assert_eq!(revisions, vec![1, 2]);
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn dropping_progress_receiver_stops_client_observation_cleanly() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcd-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("drop.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let session_id = SessionId::new();
let snapshot = SessionOpenOperationSnapshot {
operation_id: SessionOpenOperationId::new(),
revision: 1,
session_id,
source_writer_epoch: Some(3),
target_writer_epoch: 4,
progress: SessionMigrationProgress {
stage: SessionMigrationStage::CopyingBackup,
completed_units: Some(1),
total_units: Some(2),
unit: Some(bcode_session_models::SessionMigrationProgressUnit::Files),
message: "migration".to_owned(),
},
outcome: None,
backup_path: None,
};
let daemon = matching_daemon_status();
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let hello = bcode_ipc::recv_envelope(&mut stream).await.expect("hello");
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(bcode_ipc::CURRENT_PROTOCOL_VERSION),
client_id: bcode_session_models::ClientId::new(),
daemon,
});
let envelope =
bcode_ipc::response_envelope(hello.request_id, &response).expect("hello response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send hello");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("prepare request");
let response =
bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::SessionOpenPrepared {
snapshot,
});
let envelope = bcode_ipc::response_envelope(request.request_id, &response)
.expect("prepare response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send prepare");
let first = tokio::time::timeout(
Duration::from_millis(250),
bcode_ipc::recv_envelope(&mut stream),
)
.await;
if first.as_ref().is_ok_and(Result::is_ok) {
tokio::time::timeout(
Duration::from_millis(250),
bcode_ipc::recv_envelope(&mut stream),
)
.await
} else {
first
}
});
let client = BcodeClient::new(endpoint).with_request_timeout(Duration::from_secs(1));
let mut observer = client.observe_session_open(session_id);
let first = observer.receiver.recv().await.expect("initial progress");
assert_eq!(first.revision, 1);
drop(observer.receiver);
assert!(observer.task.await.expect("observer task").is_ok());
let next_request = server.await.expect("server task");
assert!(
next_request.is_ok_and(|request| request.is_err()),
"observer sent another wait request after receiver drop"
);
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn application_request_is_not_replayed_after_response_eof() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcd-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("single-send.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let daemon = matching_daemon_status();
let session_id = SessionId::new();
let input = bcode_tool::ToolInvocationInput {
invocation_id: "call".into(),
input_id: "input".into(),
producer_id: "plugin".into(),
schema: "plugin.input".into(),
schema_version: 1,
payload: serde_json::json!({"opaque": "λ"}),
};
let expected_input = input.clone();
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let hello = bcode_ipc::recv_envelope(&mut stream).await.expect("hello");
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion(bcode_ipc::CURRENT_PROTOCOL_VERSION),
client_id: bcode_session_models::ClientId::new(),
daemon,
});
let envelope =
bcode_ipc::response_envelope(hello.request_id, &response).expect("hello response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send hello");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("application request");
assert!(matches!(
bcode_ipc::decode_request(&request.payload).expect("decode request"),
bcode_ipc::Request::InvocationInput { session_id: actual, input }
if actual == session_id && input == expected_input
));
drop(stream);
assert!(
tokio::time::timeout(Duration::from_millis(250), listener.accept())
.await
.is_err(),
"uncertain input must not be replayed on a second connection"
);
});
let client = BcodeClient::new(endpoint)
.with_daemon_availability(super::DaemonAvailability::AutoStart)
.with_request_timeout(Duration::from_secs(1));
let error = client
.send_invocation_input(session_id, input)
.await
.expect_err("response EOF must fail");
assert!(matches!(error, ClientError::Codec(_)));
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[test]
fn only_ready_terminal_outcome_allows_writable_attach() {
let session_id = SessionId::new();
let snapshot = |outcome| SessionOpenOperationSnapshot {
operation_id: SessionOpenOperationId::new(),
revision: 1,
session_id,
source_writer_epoch: Some(3),
target_writer_epoch: 4,
progress: SessionMigrationProgress {
stage: SessionMigrationStage::Failed,
completed_units: None,
total_units: None,
unit: None,
message: "Classifying session".to_owned(),
},
outcome: Some(outcome),
backup_path: Some("/tmp/backup".into()),
};
assert!(
session_open_attach_readiness(&snapshot(SessionOpenTerminalOutcome::Ready)).is_ok()
);
for (outcome, expected_code) in [
(
SessionOpenTerminalOutcome::DegradedReadOnly { issue_count: 1 },
"session_degraded_read_only",
),
(
SessionOpenTerminalOutcome::WriterIncompatible {
actual: Some(5),
expected: 4,
},
"session_writer_incompatible",
),
(
SessionOpenTerminalOutcome::RepairRequired {
reason: "damaged tail".to_owned(),
},
"session_repair_required",
),
(
SessionOpenTerminalOutcome::Failed {
kind: bcode_session_models::SessionOpenFailureKind::BackupFailed,
message: "backup failed".to_owned(),
backup_path: Some("/tmp/failed-backup".into()),
},
"session_migration_backup_failed",
),
] {
assert!(matches!(
session_open_attach_readiness(&snapshot(outcome)),
Err(ClientError::Server { code, .. }) if code == expected_code
));
}
}
#[test]
fn terminal_session_open_error_preserves_recovery_context() {
let session_id = SessionId::new();
let message = terminal_session_open_error_message(
session_id,
"Verifying retained backup",
"hash mismatch",
Some(std::path::Path::new("/tmp/session-backup")),
);
assert!(message.contains("Verifying retained backup"));
assert!(message.contains("hash mismatch"));
assert!(message.contains("/tmp/session-backup"));
assert!(message.contains(&format!("bcode session diagnose {session_id}")));
}
#[test]
fn caller_paths_are_absolute_and_relative_paths_use_the_caller_cwd() {
let caller_cwd = Path::new("/tmp/bcode-client-cwd");
assert_eq!(
resolve_path_from(None, caller_cwd),
caller_cwd.to_path_buf()
);
assert_eq!(
resolve_path_from(Some("nested".into()), caller_cwd),
caller_cwd.join("nested")
);
assert_eq!(
resolve_path_from(Some("/tmp/explicit".into()), caller_cwd),
Path::new("/tmp/explicit")
);
}
#[test]
fn default_endpoint_honors_process_config_override() {
let guard = bcode_config::push_process_config_overrides(
bcode_config::ConfigLoadOverrides::from_env_with_cli(
None,
Some("[client]\nrequest_timeout_secs = 23\n".to_owned()),
),
);
let client = BcodeClient::default_endpoint();
assert_eq!(client.request_timeout(), Duration::from_secs(23));
drop(guard);
}
#[test]
fn bounded_long_poll_timeouts_use_server_bound_plus_transport_grace() {
let client =
BcodeClient::default_endpoint().with_request_timeout(Duration::from_millis(10));
let response_timeout = client
.request_timeout()
.max(Duration::from_secs(30).saturating_add(super::LONG_POLL_TRANSPORT_GRACE));
assert_eq!(response_timeout, Duration::from_secs(35));
assert!(super::validate_session_search_backfill_wait_timeout(1).is_ok());
assert!(super::validate_session_search_backfill_wait_timeout(30_000).is_ok());
assert!(matches!(
super::validate_session_search_backfill_wait_timeout(30_001),
Err(ClientError::Protocol(_))
));
assert!(super::validate_session_bulk_migration_wait_timeout(1).is_ok());
assert!(super::validate_session_bulk_migration_wait_timeout(30_000).is_ok());
assert!(matches!(
super::validate_session_bulk_migration_wait_timeout(30_001),
Err(ClientError::Protocol(_))
));
}
#[cfg(unix)]
#[tokio::test]
async fn backfill_wait_longer_than_default_request_timeout_receives_quiet_response() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcl-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("long-poll.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let operation_id = "quiet-backfill".to_owned();
let expected_operation_id = operation_id.clone();
let daemon = matching_daemon_status();
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let hello = bcode_ipc::recv_envelope(&mut stream).await.expect("hello");
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion::current(),
client_id: bcode_session_models::ClientId::new(),
daemon,
});
let envelope =
bcode_ipc::response_envelope(hello.request_id, &response).expect("hello response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send hello");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("wait request");
assert!(matches!(
bcode_ipc::decode_request(&request.payload).expect("decode wait request"),
bcode_ipc::Request::SessionSearchBackfillWait {
operation_id,
after_revision: 1,
timeout_ms: 80,
} if operation_id == expected_operation_id
));
tokio::time::sleep(Duration::from_millis(40)).await;
let response = bcode_ipc::Response::Ok(
bcode_ipc::ResponsePayload::SessionSearchBackfillOperation {
status: SessionSearchBackfillOperationStatus {
operation_id: expected_operation_id,
provider_id: "bcode.test-search".to_owned(),
revision: 1,
state: SessionSearchBackfillOperationState::Running,
response: None,
complete_progress: None,
complete_response: None,
error: None,
},
},
);
let envelope =
bcode_ipc::response_envelope(request.request_id, &response).expect("wait response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send wait response");
});
let client = BcodeClient::new(endpoint)
.with_daemon_availability(super::DaemonAvailability::RequireRunning)
.with_request_timeout(Duration::from_millis(10));
let status = client
.session_search_backfill_wait(operation_id, 1, 80)
.await
.expect("long poll must outlive generic request timeout");
assert_eq!(status.revision, 1);
assert_eq!(status.state, SessionSearchBackfillOperationState::Running);
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[cfg(unix)]
#[tokio::test]
async fn bulk_migration_wait_longer_than_default_request_timeout_receives_quiet_response() {
let socket_dir =
std::path::PathBuf::from(format!("/tmp/bcmw-{}", SessionOpenOperationId::new()));
std::fs::create_dir_all(&socket_dir).expect("socket directory");
let endpoint = bcode_ipc::IpcEndpoint::unix_socket(socket_dir.join("migration-wait.sock"));
let listener = bcode_ipc::LocalIpcListener::bind(&endpoint).expect("listener");
let operation_id = "quiet-migration".to_owned();
let expected_operation_id = operation_id.clone();
let daemon = matching_daemon_status();
let server = tokio::spawn(async move {
let mut stream = listener.accept().await.expect("accept client");
let hello = bcode_ipc::recv_envelope(&mut stream).await.expect("hello");
let response = bcode_ipc::Response::Ok(bcode_ipc::ResponsePayload::Hello {
protocol_version: bcode_ipc::ProtocolVersion::current(),
client_id: bcode_session_models::ClientId::new(),
daemon,
});
let envelope =
bcode_ipc::response_envelope(hello.request_id, &response).expect("hello response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send hello");
let request = bcode_ipc::recv_envelope(&mut stream)
.await
.expect("wait request");
assert!(matches!(
bcode_ipc::decode_request(&request.payload).expect("decode wait request"),
bcode_ipc::Request::SessionBulkMigrationWait {
operation_id,
after_revision: 2,
timeout_ms: 80,
} if operation_id == expected_operation_id
));
tokio::time::sleep(Duration::from_millis(40)).await;
let response = bcode_ipc::Response::Ok(
bcode_ipc::ResponsePayload::SessionBulkMigrationOperation {
status: bcode_ipc::SessionBulkMigrationOperationStatus {
operation_id: expected_operation_id,
revision: 2,
state: bcode_ipc::SessionBulkMigrationState::Running,
mode: bcode_ipc::SessionBulkMigrationMode::Inventory,
selected: 0,
visited: 0,
migrated: 0,
blocked: 0,
failed: 0,
current_session_id: None,
outcomes: Vec::new(),
},
},
);
let envelope =
bcode_ipc::response_envelope(request.request_id, &response).expect("wait response");
bcode_ipc::send_envelope(&mut stream, &envelope)
.await
.expect("send wait response");
});
let client = BcodeClient::new(endpoint)
.with_daemon_availability(super::DaemonAvailability::RequireRunning)
.with_request_timeout(Duration::from_millis(10));
let status = client
.wait_session_bulk_migration(operation_id, 2, 80)
.await
.expect("long poll must outlive generic request timeout");
assert_eq!(status.revision, 2);
assert_eq!(status.state, bcode_ipc::SessionBulkMigrationState::Running);
server.await.expect("server task");
std::fs::remove_dir_all(socket_dir).expect("socket cleanup");
}
#[test]
fn request_timeout_can_be_overridden() {
let client = BcodeClient::default_endpoint().with_request_timeout(Duration::from_secs(17));
assert_eq!(client.request_timeout(), Duration::from_secs(17));
}
}