openai/codex
1
0
Fork 0
mirror of https://github.com/openai/codex.git synced 2026-11-05 20:10:33 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add cloud thread activation with resolved settings (#59343)

Browse Source 
## What changed

Add `Client::activate` to load a cloud thread through `UpdateConfig`, sending only the thread ID and `exclude_turns=true`. Return serializable resolved settings, thread metadata, and history cursors while leaving turns for separate pagination. Preserve the original thread protobuf, including unknown fields.

Validate required fields and thread identity, reject unsupported security policies, and redact sensitive activation data in debug output. Treat activation as a mutation that is never retried, with uncertain outcomes reported through `MutationError::OutcomeUnknown`.

## Testing

Add coverage for activation request encoding, settings and metadata preservation, serialization, compatibility defaults, unsupported policies, debug redaction, invalid input rejection, and explicit successful gRPC trailers.

GitOrigin-RevId: cd925b7a05f518d782e5b4cf188556567dad9e21
This commit is contained in:
Benjamin Carlsson
2026-11-04 07:26:51 +00:00
committed by copyberry
parent 93b4e5b4c6
commit 3e743d1578
12 changed files with 1157 additions and 71 deletions
Download Patch File Download Diff File Expand all files Collapse all files
codex-rs/Cargo.lock
Generated
+2
View File
@@ -2643,6 +2643,8 @@ dependencies = [
"hyper-util",
"pretty_assertions",
"prost",
"serde",
"serde_json",
"thiserror 2.0.18",
"tokio",
"tokio-rustls",
codex-rs/cloud-client/Cargo.toml
+2
View File
@@ -12,6 +12,7 @@ doctest = false
workspace = true
[dependencies]
serde = { workspace = true }
async-stream = { workspace = true }
bytes = { workspace = true }
codex-http-client = { workspace = true }
@@ -27,6 +28,7 @@ tokio = { workspace = true, features = ["time"] }
url = { workspace = true }
[dev-dependencies]
serde_json = { workspace = true }
anyhow = { workspace = true }
pretty_assertions = { workspace = true }
tokio = { workspace = true, features = ["macros", "rt-multi-thread"] }
codex-rs/cloud-client/src/activation.rs
+516
View File
@@ -0,0 +1,516 @@
//! Authoritative cloud thread settings returned by deliberate native activation.
use crate::Client;
use crate::Error;
use crate::MutationError;
use crate::activation_wire as wire;
use bytes::Bytes;
use prost::Message;
use serde::Serialize;
use std::fmt;
#[cfg(test)]
#[path = "activation_tests.rs"]
mod tests;
/// Resolved settings and metadata returned after loading a cloud thread.
#[derive(Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ThreadActivation {
pub thread: ThreadSnapshot,
pub collaboration_mode: Option<CollaborationMode>,
pub model: String,
pub model_provider: String,
pub service_tier: Option<String>,
pub cwd: String,
pub instruction_sources: Vec<String>,
pub runtime_workspace_roots: Vec<String>,
pub approval_policy: ApprovalPolicy,
pub approvals_reviewer: ApprovalsReviewer,
pub sandbox: SandboxPolicy,
pub active_permission_profile: Option<ActivePermissionProfile>,
pub reasoning_effort: Option<String>,
pub turns_backwards_cursor: Option<String>,
pub items_backwards_cursor: Option<String>,
}
/// Effective collaboration mode and its backend-resolved settings.
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[serde(tag = "mode", content = "settings", rename_all = "lowercase")]
pub enum CollaborationMode {
Plan(CollaborationSettings),
Default(CollaborationSettings),
}
/// The resume protocol uses snake_case for these nested settings.
#[derive(Clone, PartialEq, Eq, Serialize)]
pub struct CollaborationSettings {
pub model: String,
pub reasoning_effort: Option<String>,
pub developer_instructions: Option<String>,
}
/// Metadata needed to reconstruct a turn-free app-server thread snapshot.
#[derive(Clone, PartialEq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ThreadSnapshot {
pub id: String,
pub session_id: String,
pub forked_from_id: Option<String>,
pub parent_thread_id: Option<String>,
pub preview: String,
pub ephemeral: bool,
pub daybreak_enabled: Option<bool>,
pub project_id: Option<String>,
pub history_mode: ThreadHistoryMode,
pub model_provider: String,
pub model: Option<String>,
pub reasoning_effort: Option<String>,
pub created_at: i64,
pub updated_at: i64,
pub recency_at: Option<i64>,
pub status: ThreadStatus,
pub path: Option<String>,
pub cwd: String,
pub cli_version: String,
pub originator: Option<String>,
pub source: SessionSource,
pub can_accept_direct_input: Option<bool>,
pub thread_source: Option<String>,
pub agent_nickname: Option<String>,
pub agent_role: Option<String>,
pub name: Option<String>,
pub environments: Option<Vec<ThreadEnvironment>>,
/// Activation deliberately excludes turns; callers page history separately.
pub turns: [(); 0],
#[serde(skip)]
raw_protobuf: Bytes,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "lowercase")]
pub enum ThreadHistoryMode {
Legacy,
Paginated,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ThreadEnvironment {
pub environment_id: String,
pub cwd: String,
pub runtime_workspace_roots: Vec<String>,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum SessionSource {
Cli,
#[serde(rename = "vscode")]
VsCode,
Exec,
AppServer,
Unknown,
Custom(String),
SubAgent(SubAgentSource),
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "snake_case")]
pub enum SubAgentSource {
Review,
Compact,
MemoryConsolidation,
ThreadSpawn {
parent_thread_id: String,
depth: i32,
agent_path: Option<String>,
agent_nickname: Option<String>,
agent_role: Option<String>,
},
Other(String),
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[serde(
tag = "type",
rename_all = "camelCase",
rename_all_fields = "camelCase"
)]
pub enum ThreadStatus {
NotLoaded,
Idle,
SystemError,
Active { active_flags: Vec<ThreadActiveFlag> },
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum ThreadActiveFlag {
WaitingOnApproval,
WaitingOnUserInput,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "kebab-case")]
pub enum ApprovalPolicy {
#[serde(rename = "untrusted")]
UnlessTrusted,
OnRequest,
Never,
Granular {
sandbox_approval: bool,
rules: bool,
skill_approval: bool,
request_permissions: bool,
mcp_elicitations: bool,
},
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
pub enum ApprovalsReviewer {
#[serde(rename = "user")]
User,
#[serde(rename = "auto_review")]
AutoReview,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub struct ActivePermissionProfile {
pub id: String,
pub extends: Option<String>,
}
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
#[serde(rename_all = "camelCase")]
pub enum NetworkAccess {
Restricted,
Enabled,
}
#[derive(Clone, Debug, PartialEq, Eq, Serialize)]
#[serde(
tag = "type",
rename_all = "camelCase",
rename_all_fields = "camelCase"
)]
pub enum SandboxPolicy {
DangerFullAccess,
ReadOnly {
network_access: bool,
},
ExternalSandbox {
network_access: NetworkAccess,
},
WorkspaceWrite {
writable_roots: Vec<String>,
network_access: bool,
exclude_tmpdir_env_var: bool,
exclude_slash_tmp: bool,
},
}
impl fmt::Debug for CollaborationSettings {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("CollaborationSettings([redacted])")
}
}
impl fmt::Debug for ThreadActivation {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("ThreadActivation([redacted])")
}
}
impl fmt::Debug for ThreadSnapshot {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("ThreadSnapshot([redacted])")
}
}
impl ThreadSnapshot {
/// Complete bounded native Thread payload, including fields this client does not interpret.
pub fn protobuf(&self) -> &Bytes {
&self.raw_protobuf
}
}
impl Client {
/// Loads a thread and returns its resolved settings without hydrating turns.
///
/// This is a mutation-capable RPC and is never retried. A timeout or disconnect
/// can leave the thread loaded even though the caller did not receive settings.
/// Local rejection sends nothing; errors after dispatch leave the outcome unknown.
pub async fn activate(&self, thread_id: &str) -> Result<ThreadActivation, MutationError> {
if thread_id.trim().is_empty() {
return Err(MutationError::InvalidRequest);
}
let bytes = self
.mutate(
wire::UpdateConfigRequest {
exclude_turns: Some(true),
thread_id: Some(thread_id.to_owned()),
},
"/oaiproto.codex_app_server_backend.ThreadService/UpdateConfig",
)
.await?;
let response = wire::UpdateConfigResponse::decode(bytes)
.map_err(|_| MutationError::OutcomeUnknown(Error::InvalidResponse))?;
ThreadActivation::try_from_wire(response, thread_id).map_err(MutationError::OutcomeUnknown)
}
}
impl ThreadActivation {
fn try_from_wire(
response: wire::UpdateConfigResponse,
requested_id: &str,
) -> Result<Self, Error> {
let raw_thread = required(response.thread)?;
let thread = ThreadSnapshot::try_from_wire(
wire::Thread::decode(raw_thread.clone()).map_err(|_| Error::InvalidResponse)?,
raw_thread,
)?;
if thread.id != requested_id {
return Err(Error::InvalidResponse);
}
Ok(Self {
thread,
collaboration_mode: response
.collaboration_mode
.map(parse_collaboration_mode)
.transpose()?,
model: nonempty(response.model)?,
model_provider: nonempty(response.model_provider)?,
service_tier: response.service_tier.and_then(|value| match value.value {
Some(wire::nullable_string::Value::String(value)) => Some(value),
Some(wire::nullable_string::Value::Null(_)) | None => None,
}),
cwd: nonempty(response.cwd)?,
instruction_sources: response
.instruction_sources
.map_or_else(Vec::new, |sources| sources.values),
runtime_workspace_roots: response
.runtime_workspace_roots
.map_or_else(Vec::new, |roots| roots.values),
approval_policy: parse_approval(required(response.approval_policy)?)?,
approvals_reviewer: parse_reviewer(required(response.approvals_reviewer)?)?,
sandbox: parse_sandbox(required(response.sandbox)?)?,
active_permission_profile: response
.active_permission_profile
.map(|profile| -> Result<_, Error> {
Ok(ActivePermissionProfile {
id: nonempty(profile.id)?,
extends: profile.extends,
})
})
.transpose()?,
reasoning_effort: response.reasoning_effort,
turns_backwards_cursor: response.turns_backwards_cursor,
items_backwards_cursor: response.items_backwards_cursor,
})
}
}
impl ThreadSnapshot {
fn try_from_wire(thread: wire::Thread, raw_protobuf: Bytes) -> Result<Self, Error> {
Ok(Self {
id: nonempty(thread.id)?,
session_id: nonempty(thread.session_id)?,
forked_from_id: thread.forked_from_id,
parent_thread_id: thread.parent_thread_id,
preview: required(thread.preview)?,
ephemeral: required(thread.ephemeral)?,
daybreak_enabled: thread.daybreak_enabled,
project_id: thread.project_id,
history_mode: parse_history_mode(thread.history_mode)?,
model_provider: nonempty(thread.model_provider)?,
model: thread.model,
reasoning_effort: thread.reasoning_effort,
created_at: required(thread.created_at)?,
updated_at: required(thread.updated_at)?,
recency_at: thread.recency_at,
status: parse_status(required(thread.status)?)?,
path: thread.path,
cwd: nonempty(thread.cwd)?,
cli_version: required(thread.cli_version)?,
originator: thread.originator,
source: parse_source(required(thread.source)?)?,
can_accept_direct_input: thread.can_accept_direct_input,
thread_source: thread.thread_source,
agent_nickname: thread.agent_nickname,
agent_role: thread.agent_role,
name: thread.name,
environments: thread
.environments
.map(|environments| {
environments
.values
.into_iter()
.map(|environment| -> Result<_, Error> {
Ok(ThreadEnvironment {
environment_id: nonempty(environment.environment_id)?,
cwd: nonempty(environment.cwd)?,
runtime_workspace_roots: required(
environment.runtime_workspace_roots,
)?
.values,
})
})
.collect()
})
.transpose()?,
turns: [],
raw_protobuf,
})
}
}
fn parse_collaboration_mode(value: wire::CollaborationMode) -> Result<CollaborationMode, Error> {
let settings = required(value.settings)?;
let settings = CollaborationSettings {
model: nonempty(settings.model)?,
reasoning_effort: settings.reasoning_effort,
developer_instructions: settings.developer_instructions,
};
match required(value.mode)? {
1 => Ok(CollaborationMode::Plan(settings)),
2 => Ok(CollaborationMode::Default(settings)),
_ => Err(Error::UnsupportedResponse("collaboration mode")),
}
}
fn parse_history_mode(value: Option<i32>) -> Result<ThreadHistoryMode, Error> {
match value {
None | Some(1) => Ok(ThreadHistoryMode::Legacy),
Some(2) => Ok(ThreadHistoryMode::Paginated),
Some(_) => Err(Error::UnsupportedResponse("thread history mode")),
}
}
fn parse_approval(value: wire::ApprovalPolicy) -> Result<ApprovalPolicy, Error> {
match required(value.value)? {
wire::approval_policy::Value::Simple(1) => Ok(ApprovalPolicy::UnlessTrusted),
wire::approval_policy::Value::Simple(2) => Ok(ApprovalPolicy::OnRequest),
wire::approval_policy::Value::Simple(3) => Ok(ApprovalPolicy::Never),
wire::approval_policy::Value::Simple(_) => {
Err(Error::UnsupportedResponse("approval policy"))
}
wire::approval_policy::Value::Granular(wrapper) => {
let granular = required(wrapper.granular)?;
Ok(ApprovalPolicy::Granular {
sandbox_approval: required(granular.sandbox_approval)?,
rules: required(granular.rules)?,
skill_approval: granular.skill_approval.unwrap_or(false),
request_permissions: granular.request_permissions.unwrap_or(false),
mcp_elicitations: required(granular.mcp_elicitations)?,
})
}
}
}
fn parse_reviewer(value: i32) -> Result<ApprovalsReviewer, Error> {
match value {
1 => Ok(ApprovalsReviewer::User),
2 | 3 => Ok(ApprovalsReviewer::AutoReview),
_ => Err(Error::UnsupportedResponse("approvals reviewer")),
}
}
fn parse_sandbox(value: wire::SandboxPolicy) -> Result<SandboxPolicy, Error> {
match required(value.value)? {
wire::sandbox_policy::Value::DangerFullAccess(_) => Ok(SandboxPolicy::DangerFullAccess),
wire::sandbox_policy::Value::ReadOnly(policy) => Ok(SandboxPolicy::ReadOnly {
network_access: policy.network_access.unwrap_or(false),
}),
wire::sandbox_policy::Value::ExternalSandbox(policy) => {
Ok(SandboxPolicy::ExternalSandbox {
network_access: match policy.network_access {
None | Some(1) => NetworkAccess::Restricted,
Some(2) => NetworkAccess::Enabled,
Some(_) => return Err(Error::UnsupportedResponse("external network policy")),
},
})
}
wire::sandbox_policy::Value::WorkspaceWrite(policy) => Ok(SandboxPolicy::WorkspaceWrite {
writable_roots: policy
.writable_roots
.map_or_else(Vec::new, |roots| roots.values),
network_access: policy.network_access.unwrap_or(false),
exclude_tmpdir_env_var: policy.exclude_tmpdir_env_var.unwrap_or(false),
exclude_slash_tmp: policy.exclude_slash_tmp.unwrap_or(false),
}),
}
}
fn parse_source(value: wire::SessionSource) -> Result<SessionSource, Error> {
match required(value.value)? {
wire::session_source::Value::Simple(1) => Ok(SessionSource::Cli),
wire::session_source::Value::Simple(2) => Ok(SessionSource::VsCode),
wire::session_source::Value::Simple(3) => Ok(SessionSource::Exec),
wire::session_source::Value::Simple(4) => Ok(SessionSource::AppServer),
wire::session_source::Value::Simple(0) => Err(Error::UnsupportedResponse("session source")),
wire::session_source::Value::Simple(_) => Ok(SessionSource::Unknown),
wire::session_source::Value::Custom(source) => {
Ok(SessionSource::Custom(nonempty(source.custom)?))
}
wire::session_source::Value::SubAgent(wrapper) => Ok(SessionSource::SubAgent(
parse_subagent_source(required(wrapper.sub_agent)?)?,
)),
}
}
fn parse_subagent_source(value: wire::SubAgentSource) -> Result<SubAgentSource, Error> {
match required(value.value)? {
wire::sub_agent_source::Value::Simple(1) => Ok(SubAgentSource::Review),
wire::sub_agent_source::Value::Simple(2) => Ok(SubAgentSource::Compact),
wire::sub_agent_source::Value::Simple(3) => Ok(SubAgentSource::MemoryConsolidation),
wire::sub_agent_source::Value::Simple(_) => {
Err(Error::UnsupportedResponse("sub-agent source"))
}
wire::sub_agent_source::Value::ThreadSpawn(wrapper) => {
let source = required(wrapper.thread_spawn)?;
Ok(SubAgentSource::ThreadSpawn {
parent_thread_id: nonempty(source.parent_thread_id)?,
depth: i32::try_from(required(source.depth)?)
.map_err(|_| Error::InvalidResponse)?,
agent_path: source.agent_path,
agent_nickname: source.agent_nickname,
agent_role: source.agent_role,
})
}
wire::sub_agent_source::Value::Other(source) => {
Ok(SubAgentSource::Other(nonempty(source.other)?))
}
}
}
fn parse_status(value: wire::ThreadStatus) -> Result<ThreadStatus, Error> {
match required(value.value)? {
wire::thread_status::Value::NotLoaded(_) => Ok(ThreadStatus::NotLoaded),
wire::thread_status::Value::Idle(_) => Ok(ThreadStatus::Idle),
wire::thread_status::Value::SystemError(_) => Ok(ThreadStatus::SystemError),
wire::thread_status::Value::Active(active) => Ok(ThreadStatus::Active {
active_flags: required(active.active_flags)?
.values
.into_iter()
.map(|flag| match flag {
1 => Ok(ThreadActiveFlag::WaitingOnApproval),
2 => Ok(ThreadActiveFlag::WaitingOnUserInput),
_ => Err(Error::UnsupportedResponse("thread active flag")),
})
.collect::<Result<_, _>>()?,
}),
}
}
fn required<T>(value: Option<T>) -> Result<T, Error> {
value.ok_or(Error::InvalidResponse)
}
fn nonempty(value: Option<String>) -> Result<String, Error> {
let value = required(value)?;
if value.is_empty() {
Err(Error::InvalidResponse)
} else {
Ok(value)
}
}
codex-rs/cloud-client/src/activation_tests.rs
+52
View File
@@ -0,0 +1,52 @@
use super::*;
use pretty_assertions::assert_eq;
#[test]
fn wire_debug_redacts_thread_preview_and_raw_bytes() {
let thread = wire::Thread {
preview: Some("synthetic private preview".to_owned()),
..Default::default()
};
let response = wire::UpdateConfigResponse {
thread: Some(Bytes::from(thread.encode_to_vec())),
..Default::default()
};
assert_eq!(format!("{thread:?}"), "Thread([redacted])");
assert_eq!(format!("{response:?}"), "UpdateConfigResponse([redacted])");
}
#[test]
fn missing_history_mode_uses_legacy_compatibility_default() {
assert_eq!(
parse_history_mode(/*value*/ None).unwrap(),
ThreadHistoryMode::Legacy
);
}
#[test]
fn external_sandbox_network_access_defaults_only_when_omitted() {
let decode = |network_access| {
parse_sandbox(wire::SandboxPolicy {
value: Some(wire::sandbox_policy::Value::ExternalSandbox(
wire::ExternalSandbox { network_access },
)),
})
};
for (value, network_access) in [
(None, NetworkAccess::Restricted),
(Some(1), NetworkAccess::Restricted),
(Some(2), NetworkAccess::Enabled),
] {
assert_eq!(
decode(value).unwrap(),
SandboxPolicy::ExternalSandbox { network_access }
);
}
for value in [0, 99] {
assert!(matches!(
decode(Some(value)),
Err(Error::UnsupportedResponse("external network policy"))
));
}
}
codex-rs/cloud-client/src/activation_wire.rs
+354
View File
@@ -0,0 +1,354 @@
//! Minimal protobuf envelopes for authoritative thread activation.
use bytes::Bytes;
use prost::Message;
use std::fmt;
#[derive(Clone, Message)]
pub(super) struct UpdateConfigRequest {
#[prost(bool, optional, tag = "7")]
pub exclude_turns: Option<bool>,
#[prost(string, optional, tag = "18")]
pub thread_id: Option<String>,
}
#[derive(Clone, Message)]
#[prost(skip_debug)]
pub(super) struct UpdateConfigResponse {
#[prost(message, optional, tag = "1")]
pub active_permission_profile: Option<ActivePermissionProfile>,
#[prost(message, optional, tag = "2")]
pub approval_policy: Option<ApprovalPolicy>,
#[prost(int32, optional, tag = "3")]
pub approvals_reviewer: Option<i32>,
#[prost(message, optional, tag = "4")]
pub collaboration_mode: Option<CollaborationMode>,
#[prost(string, optional, tag = "5")]
pub cwd: Option<String>,
#[prost(message, optional, tag = "8")]
pub instruction_sources: Option<StringList>,
#[prost(string, optional, tag = "9")]
pub items_backwards_cursor: Option<String>,
#[prost(string, optional, tag = "10")]
pub model: Option<String>,
#[prost(string, optional, tag = "11")]
pub model_provider: Option<String>,
#[prost(string, optional, tag = "13")]
pub reasoning_effort: Option<String>,
#[prost(message, optional, tag = "14")]
pub runtime_workspace_roots: Option<StringList>,
#[prost(message, optional, tag = "15")]
pub sandbox: Option<SandboxPolicy>,
#[prost(message, optional, tag = "16")]
pub service_tier: Option<NullableString>,
#[prost(bytes = "bytes", optional, tag = "17")]
pub thread: Option<Bytes>,
#[prost(string, optional, tag = "18")]
pub turns_backwards_cursor: Option<String>,
}
impl fmt::Debug for UpdateConfigResponse {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("UpdateConfigResponse([redacted])")
}
}
#[derive(Clone, Message)]
pub(super) struct CollaborationMode {
#[prost(int32, optional, tag = "1")]
pub mode: Option<i32>,
#[prost(message, optional, tag = "2")]
pub settings: Option<Settings>,
}
#[derive(Clone, Message)]
#[prost(skip_debug)]
pub(super) struct Settings {
#[prost(string, optional, tag = "1")]
pub developer_instructions: Option<String>,
#[prost(string, optional, tag = "2")]
pub model: Option<String>,
#[prost(string, optional, tag = "3")]
pub reasoning_effort: Option<String>,
}
impl fmt::Debug for Settings {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("Settings([redacted])")
}
}
#[derive(Clone, Message)]
pub(super) struct ActivePermissionProfile {
#[prost(string, optional, tag = "1")]
pub extends: Option<String>,
#[prost(string, optional, tag = "2")]
pub id: Option<String>,
}
#[derive(Clone, Message)]
pub(super) struct ApprovalPolicy {
#[prost(oneof = "approval_policy::Value", tags = "1, 2")]
pub value: Option<approval_policy::Value>,
}
pub(super) mod approval_policy {
#[derive(Clone, prost::Oneof)]
pub enum Value {
#[prost(int32, tag = "1")]
Simple(i32),
#[prost(message, tag = "2")]
Granular(super::GranularApprovalWrapper),
}
}
#[derive(Clone, Message)]
pub(super) struct GranularApprovalWrapper {
#[prost(message, optional, tag = "1")]
pub granular: Option<GranularApproval>,
}
#[derive(Clone, Message)]
pub(super) struct GranularApproval {
#[prost(bool, optional, tag = "1")]
pub mcp_elicitations: Option<bool>,
#[prost(bool, optional, tag = "2")]
pub request_permissions: Option<bool>,
#[prost(bool, optional, tag = "3")]
pub rules: Option<bool>,
#[prost(bool, optional, tag = "4")]
pub sandbox_approval: Option<bool>,
#[prost(bool, optional, tag = "5")]
pub skill_approval: Option<bool>,
}
#[derive(Clone, Message)]
pub(super) struct SandboxPolicy {
#[prost(oneof = "sandbox_policy::Value", tags = "1, 2, 3, 4")]
pub value: Option<sandbox_policy::Value>,
}
pub(super) mod sandbox_policy {
#[derive(Clone, prost::Oneof)]
pub enum Value {
#[prost(message, tag = "1")]
DangerFullAccess(super::Empty),
#[prost(message, tag = "2")]
ReadOnly(super::ReadOnlySandbox),
#[prost(message, tag = "3")]
ExternalSandbox(super::ExternalSandbox),
#[prost(message, tag = "4")]
WorkspaceWrite(super::WorkspaceWriteSandbox),
}
}
#[derive(Clone, Message)]
pub(super) struct Empty {}
#[derive(Clone, Message)]
pub(super) struct ReadOnlySandbox {
#[prost(bool, optional, tag = "1")]
pub network_access: Option<bool>,
}
#[derive(Clone, Message)]
pub(super) struct ExternalSandbox {
#[prost(int32, optional, tag = "1")]
pub network_access: Option<i32>,
}
#[derive(Clone, Message)]
pub(super) struct WorkspaceWriteSandbox {
#[prost(bool, optional, tag = "1")]
pub exclude_slash_tmp: Option<bool>,
#[prost(bool, optional, tag = "2")]
pub exclude_tmpdir_env_var: Option<bool>,
#[prost(bool, optional, tag = "3")]
pub network_access: Option<bool>,
#[prost(message, optional, tag = "4")]
pub writable_roots: Option<StringList>,
}
#[derive(Clone, Message)]
pub(super) struct StringList {
#[prost(string, repeated, tag = "1")]
pub values: Vec<String>,
}
#[derive(Clone, Message)]
pub(super) struct NullableString {
#[prost(oneof = "nullable_string::Value", tags = "1, 2")]
pub value: Option<nullable_string::Value>,
}
pub(super) mod nullable_string {
#[derive(Clone, prost::Oneof)]
pub enum Value {
#[prost(string, tag = "1")]
String(String),
#[prost(message, tag = "2")]
Null(super::Empty),
}
}
#[derive(Clone, Message)]
#[prost(skip_debug)]
pub(super) struct Thread {
#[prost(string, optional, tag = "1")]
pub agent_nickname: Option<String>,
#[prost(string, optional, tag = "2")]
pub agent_role: Option<String>,
#[prost(bool, optional, tag = "3")]
pub can_accept_direct_input: Option<bool>,
#[prost(string, optional, tag = "4")]
pub cli_version: Option<String>,
#[prost(int64, optional, tag = "5")]
pub created_at: Option<i64>,
#[prost(string, optional, tag = "6")]
pub cwd: Option<String>,
#[prost(bool, optional, tag = "7")]
pub daybreak_enabled: Option<bool>,
#[prost(message, optional, tag = "8")]
pub environments: Option<Environments>,
#[prost(bool, optional, tag = "9")]
pub ephemeral: Option<bool>,
#[prost(string, optional, tag = "11")]
pub forked_from_id: Option<String>,
#[prost(int32, optional, tag = "13")]
pub history_mode: Option<i32>,
#[prost(string, optional, tag = "14")]
pub id: Option<String>,
#[prost(string, optional, tag = "15")]
pub model: Option<String>,
#[prost(string, optional, tag = "16")]
pub model_provider: Option<String>,
#[prost(string, optional, tag = "17")]
pub name: Option<String>,
#[prost(string, optional, tag = "18")]
pub originator: Option<String>,
#[prost(string, optional, tag = "19")]
pub parent_thread_id: Option<String>,
#[prost(string, optional, tag = "20")]
pub path: Option<String>,
#[prost(string, optional, tag = "21")]
pub preview: Option<String>,
#[prost(string, optional, tag = "22")]
pub project_id: Option<String>,
#[prost(string, optional, tag = "23")]
pub reasoning_effort: Option<String>,
#[prost(int64, optional, tag = "24")]
pub recency_at: Option<i64>,
#[prost(string, optional, tag = "27")]
pub session_id: Option<String>,
#[prost(message, optional, tag = "28")]
pub source: Option<SessionSource>,
#[prost(message, optional, tag = "29")]
pub status: Option<ThreadStatus>,
#[prost(string, optional, tag = "30")]
pub thread_source: Option<String>,
#[prost(int64, optional, tag = "32")]
pub updated_at: Option<i64>,
}
impl fmt::Debug for Thread {
fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
f.write_str("Thread([redacted])")
}
}
#[derive(Clone, Message)]
pub(super) struct Environments {
#[prost(message, repeated, tag = "1")]
pub values: Vec<Environment>,
}
#[derive(Clone, Message)]
pub(super) struct Environment {
#[prost(string, optional, tag = "1")]
pub cwd: Option<String>,
#[prost(string, optional, tag = "2")]
pub environment_id: Option<String>,
#[prost(message, optional, tag = "3")]
pub runtime_workspace_roots: Option<StringList>,
}
#[derive(Clone, Message)]
pub(super) struct SessionSource {
#[prost(oneof = "session_source::Value", tags = "1, 2, 3")]
pub value: Option<session_source::Value>,
}
pub(super) mod session_source {
#[derive(Clone, prost::Oneof)]
pub enum Value {
#[prost(int32, tag = "1")]
Simple(i32),
#[prost(message, tag = "2")]
Custom(super::CustomSource),
#[prost(message, tag = "3")]
SubAgent(super::SubAgentSourceWrapper),
}
}
#[derive(Clone, Message)]
pub(super) struct SubAgentSourceWrapper {
#[prost(message, optional, tag = "1")]
pub sub_agent: Option<SubAgentSource>,
}
#[derive(Clone, Message)]
pub(super) struct SubAgentSource {
#[prost(oneof = "sub_agent_source::Value", tags = "1, 2, 3")]
pub value: Option<sub_agent_source::Value>,
}
pub(super) mod sub_agent_source {
#[derive(Clone, prost::Oneof)]
pub enum Value {
#[prost(int32, tag = "1")]
Simple(i32),
#[prost(message, tag = "2")]
ThreadSpawn(super::ThreadSpawnWrapper),
#[prost(message, tag = "3")]
Other(super::OtherSubAgentSource),
}
}
#[derive(Clone, Message)]
pub(super) struct OtherSubAgentSource {
#[prost(string, optional, tag = "1")]
pub other: Option<String>,
}
#[derive(Clone, Message)]
pub(super) struct ThreadSpawnWrapper {
#[prost(message, optional, tag = "1")]
pub thread_spawn: Option<ThreadSpawnSource>,
}
#[derive(Clone, Message)]
pub(super) struct ThreadSpawnSource {
#[prost(string, optional, tag = "1")]
pub agent_nickname: Option<String>,
#[prost(string, optional, tag = "2")]
pub agent_path: Option<String>,
#[prost(string, optional, tag = "3")]
pub agent_role: Option<String>,
#[prost(int64, optional, tag = "4")]
pub depth: Option<i64>,
#[prost(string, optional, tag = "5")]
pub parent_thread_id: Option<String>,
}
#[derive(Clone, Message)]
pub(super) struct CustomSource {
#[prost(string, optional, tag = "1")]
pub custom: Option<String>,
}
#[derive(Clone, Message)]
pub(super) struct ThreadStatus {
#[prost(oneof = "thread_status::Value", tags = "1, 2, 3, 4")]
pub value: Option<thread_status::Value>,
}
pub(super) mod thread_status {
#[derive(Clone, prost::Oneof)]
pub enum Value {
#[prost(message, tag = "1")]
NotLoaded(super::Empty),
#[prost(message, tag = "2")]
Idle(super::Empty),
#[prost(message, tag = "3")]
SystemError(super::Empty),
#[prost(message, tag = "4")]
Active(super::ActiveStatus),
}
}
#[derive(Clone, Message)]
pub(super) struct ActiveStatus {
#[prost(message, optional, tag = "1")]
pub active_flags: Option<IntList>,
}
#[derive(Clone, Message)]
pub(super) struct IntList {
#[prost(int32, repeated, tag = "1")]
pub values: Vec<i32>,
}
codex-rs/cloud-client/src/client_tests.rs
+116
View File
@@ -97,6 +97,122 @@ async fn assert_request(fixture: &mut Fixture, method: &str, expected: &[u8]) {
assert!(!parts.headers.contains_key("grpc-timeout"));
assert_eq!(body, frame(expected));
}
#[tokio::test]
async fn activation_sends_only_identity_and_excludes_turns() {
let mut fixture = fixture(response(vec![frame(&[])], tonic::Status::ok(""))).await;
assert!(matches!(
fixture.client.activate("t").await,
Err(MutationError::OutcomeUnknown(Error::InvalidResponse))
));
// exclude_turns=true (field 7), thread_id="t" (field 18).
assert_request(&mut fixture, "UpdateConfig", &[56, 1, 0x92, 1, 1, b't']).await;
}
#[tokio::test]
async fn activation_preserves_native_settings_and_thread_metadata() {
// Native Thread fields encoded independently of the client's wire types.
let thread_base: &[u8] = &[
0x22, 1, b'v', // cli_version (4)
0x28, 1, // created_at (5)
0x32, 1, b'/', // cwd (6)
0x48, 0, // ephemeral (9)
0x72, 1, b't', // id (14)
0x82, 1, 1, b'p', // model_provider (16)
0xaa, 1, 0, // preview (21)
0xda, 1, 1, b's', // session_id (27)
0xe2, 1, 2, 8, 99, // source (28): future simple value
0xea, 1, 2, 0x12, 0, // status (29): idle
0xfa, 1, 0, // turns (31): empty
0x80, 2, 2, // updated_at (32)
];
let encode_response = |mode: u8, daybreak: Option<bool>, approval: u8| {
let mut thread = thread_base.to_vec();
if let Some(enabled) = daybreak {
thread.extend_from_slice(&[0x38, u8::from(enabled)]); // daybreak_enabled (7)
}
// Settings has developer_instructions (1), model (2), and reasoning_effort (3).
let settings = Bytes::from_static(if mode == 1 {
b"\x0a\x01d\x12\x01m\x1a\x04high"
} else {
b"\x12\x01m"
});
let mut collaboration_mode = vec![8, mode];
prost::encoding::bytes::encode(/* settings */ 2, &settings, &mut collaboration_mode);
let mut response_bytes = vec![0x12, 2, 8, approval, 0x18, 1]; // approval and reviewer
prost::encoding::bytes::encode(
/* collaboration_mode */ 4,
&collaboration_mode,
&mut response_bytes,
);
// instruction_sources (8) wraps repeated string values (1).
response_bytes.extend_from_slice(b"\x42\x0c\x0a\x0a/AGENTS.md");
response_bytes.extend_from_slice(&[
0x2a, 1, b'/', // cwd (5)
0x52, 1, b'm', // model (10)
0x5a, 1, b'p', // model_provider (11)
0x7a, 4, 0x12, 2, 8, 1, // sandbox (15): read-only with network access
]);
prost::encoding::bytes::encode(/* thread */ 17, &thread, &mut response_bytes);
(response_bytes, thread)
};
for (mode, daybreak) in [(1, Some(true)), (2, Some(false)), (2, None)] {
let (response_bytes, thread) = encode_response(mode, daybreak, 2);
let mut server = fixture(response(
vec![frame(&response_bytes)],
tonic::Status::ok(""),
))
.await;
let activation = server.client.activate("t").await.unwrap();
assert_eq!(activation.thread.protobuf().as_ref(), thread.as_slice());
let json = serde_json::to_value(&activation).unwrap();
assert_eq!(json["thread"]["source"], "unknown");
assert_eq!(
json["thread"].get("daybreakEnabled"),
Some(&serde_json::json!(daybreak))
);
assert_eq!(
json["instructionSources"],
serde_json::json!(["/AGENTS.md"])
);
assert_eq!(json["approvalPolicy"], "on-request");
assert_eq!(json["approvalsReviewer"], "user");
assert_eq!(
json["sandbox"],
serde_json::json!({"type": "readOnly", "networkAccess": true})
);
assert_eq!(
json["collaborationMode"],
serde_json::json!({
"mode": if mode == 1 { "plan" } else { "default" },
"settings": {
"model": "m",
"developer_instructions": if mode == 1 { Some("d") } else { None },
"reasoning_effort": if mode == 1 { Some("high") } else { None }
}
})
);
assert_request(&mut server, "UpdateConfig", &[56, 1, 0x92, 1, 1, b't']).await;
assert!(server.requests.try_recv().is_err());
}
// Unknown security policy must remain an error, never a default.
let (response_bytes, _) = encode_response(1, Some(true), 99);
let mut server = fixture(response(
vec![frame(&response_bytes)],
tonic::Status::ok(""),
))
.await;
assert!(matches!(
server.client.activate("t").await,
Err(MutationError::OutcomeUnknown(Error::UnsupportedResponse(
"approval policy"
)))
));
assert_request(&mut server, "UpdateConfig", &[56, 1, 0x92, 1, 1, b't']).await;
assert!(server.requests.try_recv().is_err());
}
#[tokio::test]
async fn resume_sends_protobuf_and_preserves_admission_flags() {
for wait in [false, true] {
codex-rs/cloud-client/src/history.rs
+2 -24
View File
@@ -117,7 +117,7 @@ impl Client {
request: &ListTurnsRequest,
) -> Result<HistoryPage<HistoryTurn>, Error> {
let payload = self
.history_rpc(
.raw_unary(
wire::TurnsRequest {
thread_id: request.thread_id.clone(),
cursor: request.page.cursor.clone(),
@@ -144,7 +144,7 @@ impl Client {
request: &ListItemsRequest,
) -> Result<HistoryPage<HistoryItem>, Error> {
let payload = self
.history_rpc(
.raw_unary(
wire::ItemsRequest {
thread_id: request.thread_id.clone(),
turn_id: request.turn_id.clone(),
@@ -176,28 +176,6 @@ impl Client {
})
.map_err(|_| Error::InvalidResponse)
}
async fn history_rpc<T: Message + Send + 'static>(
&self,
value: T,
path: &'static str,
) -> Result<Bytes, Error> {
let mut grpc = self.grpc.clone();
let mut request = tonic::Request::new(value);
*request.metadata_mut() =
tonic::metadata::MetadataMap::from_headers(self.credentials.headers.clone());
let response = tokio::time::timeout(
crate::REQUEST_TIMEOUT,
grpc.unary(
request,
http::uri::PathAndQuery::from_static(path),
wire::RawResponseCodec::<T>(std::marker::PhantomData),
),
)
.await
.map_err(|_| Error::Timeout)?
.map_err(Error::from)?;
Ok(response.into_inner())
}
}
fn decode_page<T>(
thread_id: &str,
codex-rs/cloud-client/src/history_wire.rs
-29
View File
@@ -1,9 +1,6 @@
//! Native paginated history envelopes. Raw rows and unary responses retain unknown fields.
use bytes::Buf;
use bytes::Bytes;
use prost::Message;
use std::marker::PhantomData;
use tonic::codec::Codec;
#[derive(Clone, Message)]
pub(super) struct TurnsRequest {
@@ -58,29 +55,3 @@ pub(super) struct ItemEntry {
#[prost(int64, optional, tag = "4")]
pub completed_at_ms: Option<i64>,
}
// Tonic still enforces message bounds and status trailers. Only payload decoding
// is delayed so callers can retain the original complete protobuf response.
pub(super) struct RawResponseCodec<T>(pub PhantomData<T>);
impl<T: Message + Send + 'static> Codec for RawResponseCodec<T> {
type Encode = T;
type Decode = Bytes;
type Encoder = tonic_prost::ProstEncoder<T>;
type Decoder = RawDecoder;
fn encoder(&mut self) -> Self::Encoder {
tonic_prost::ProstCodec::<T, Page>::default().encoder()
}
fn decoder(&mut self) -> Self::Decoder {
RawDecoder
}
}
pub(super) struct RawDecoder;
impl tonic::codec::Decoder for RawDecoder {
type Item = Bytes;
type Error = tonic::Status;
fn decode(
&mut self,
buffer: &mut tonic::codec::DecodeBuf<'_>,
) -> Result<Option<Bytes>, tonic::Status> {
Ok(Some(buffer.copy_to_bytes(buffer.remaining())))
}
}
codex-rs/cloud-client/src/lib.rs
+40
View File
@@ -4,6 +4,8 @@
//! a timeout or disconnect can leave Resume admitted. Attach delivers live events only;
//! dropping its stream detaches without interrupting the thread or answering approvals.
mod activation;
mod activation_wire;
mod history;
mod history_wire;
mod item_wire;
@@ -15,6 +17,21 @@ mod turns_wire;
mod types;
mod wire;
pub use activation::ActivePermissionProfile;
pub use activation::ApprovalPolicy;
pub use activation::ApprovalsReviewer;
pub use activation::CollaborationMode;
pub use activation::CollaborationSettings;
pub use activation::NetworkAccess;
pub use activation::SandboxPolicy;
pub use activation::SessionSource;
pub use activation::SubAgentSource;
pub use activation::ThreadActivation;
pub use activation::ThreadActiveFlag;
pub use activation::ThreadEnvironment;
pub use activation::ThreadHistoryMode;
pub use activation::ThreadSnapshot;
pub use activation::ThreadStatus;
pub use history::HistoryAccess;
pub use history::HistoryItem;
pub use history::HistoryOrder;
@@ -79,6 +96,29 @@ pub struct Client {
}
impl Client {
async fn raw_unary<T: prost::Message + Send + 'static>(
&self,
value: T,
path: &'static str,
) -> Result<bytes::Bytes, Error> {
let mut grpc = self.grpc.clone();
let mut request = tonic::Request::new(value);
*request.metadata_mut() =
tonic::metadata::MetadataMap::from_headers(self.credentials.headers.clone());
let response = tokio::time::timeout(
REQUEST_TIMEOUT,
grpc.unary(
request,
http::uri::PathAndQuery::from_static(path),
wire::RawResponseCodec::<T>(std::marker::PhantomData),
),
)
.await
.map_err(|_| Error::Timeout)?
.map_err(Error::from)?;
Ok(response.into_inner())
}
pub fn new(
factory: &HttpClientFactory,
endpoint: &str,
codex-rs/cloud-client/src/turns.rs
+2 -2
View File
@@ -144,7 +144,7 @@ impl Client {
.map_err(|_| MutationError::OutcomeUnknown(Error::InvalidResponse))?;
Ok(())
}
async fn mutate<T: Message + Send + 'static>(
pub(crate) async fn mutate<T: Message + Send + 'static>(
&self,
value: T,
path: &'static str,
@@ -163,7 +163,7 @@ impl Client {
.server_streaming(
request,
http::uri::PathAndQuery::from_static(path),
crate::history_wire::RawResponseCodec::<T>(std::marker::PhantomData),
crate::wire::RawResponseCodec::<T>(std::marker::PhantomData),
)
.await?;
// An initial OK makes tonic skip final-status validation. A unary
codex-rs/cloud-client/src/turns_tests.rs
+39 -16
View File
@@ -220,6 +220,12 @@ async fn invalid_input_never_sends_and_diagnostics_redact() {
fixture.client.interrupt_turn(&input).await,
Err(MutationError::InvalidRequest)
));
for thread_id in ["".to_owned(), " ".to_owned(), "t".repeat(MAX_MESSAGE_BYTES)] {
assert!(matches!(
fixture.client.activate(&thread_id).await,
Err(MutationError::InvalidRequest)
));
}
assert!(fixture.requests.try_recv().is_err());
}
#[tokio::test]
@@ -282,18 +288,27 @@ async fn server_rejection_and_transport_status_do_not_prove_no_admission() {
async fn mutations_require_explicit_successful_grpc_status() {
let mut ok_trailers = http::HeaderMap::new();
ok_trailers.insert("grpc-status", http::HeaderValue::from_static("0"));
for (initial_status, trailers, include_message) in [
(None, None, true),
(None, Some(http::HeaderMap::new()), true),
(Some("0"), None, true),
(Some("0"), None, false),
(Some("0"), Some(ok_trailers), true),
for (initial_status, trailers, include_message, expect_success) in [
(None, None, true, false),
(None, Some(http::HeaderMap::new()), true, false),
(Some("0"), None, true, false),
(Some("0"), None, false, false),
(Some("0"), Some(ok_trailers.clone()), true, false),
(None, Some(ok_trailers), true, true),
] {
for method in ["AddUserMessage", "InterruptTurn"] {
let payload = if method == "AddUserMessage" {
admitted(/*outcome*/ 1)
} else {
vec![]
for method in ["AddUserMessage", "InterruptTurn", "UpdateConfig"] {
let payload = match method {
"AddUserMessage" => admitted(/*outcome*/ 1),
"InterruptTurn" => vec![],
"UpdateConfig" => [
// Approval, reviewer, cwd, model, provider and read-only sandbox.
b"\x12\x02\x08\x02\x18\x01\x2a\x01/\x52\x01m\x5a\x01p\x7a\x02\x12\x00".to_vec(),
message(
/*tag*/ 17,
b"\x22\x01v\x28\x01\x32\x01/\x48\x00\x72\x01t\x82\x01\x01p\xaa\x01\x00\xda\x01\x01s\xe2\x01\x02\x08\x01\xea\x01\x02\x12\x00\x80\x02\x02",
),
].concat(),
_ => unreachable!(),
};
let frames = include_message
.then(|| Frame::data(frame(&payload)))
@@ -303,7 +318,7 @@ async fn mutations_require_explicit_successful_grpc_status() {
if let Some(status) = initial_status {
builder = builder.header("grpc-status", status);
}
let fixture = fixture(
let mut fixture = fixture(
builder
.body(Body::new(StreamBody::new(futures::stream::iter(
frames.map(Ok::<_, Infallible>),
@@ -317,6 +332,8 @@ async fn mutations_require_explicit_successful_grpc_status() {
.add_user_message(&request())
.await
.map(|_| ())
} else if method == "UpdateConfig" {
fixture.client.activate("t").await.map(|_| ())
} else {
fixture
.client
@@ -327,10 +344,16 @@ async fn mutations_require_explicit_successful_grpc_status() {
})
.await
};
assert!(matches!(
result,
Err(MutationError::OutcomeUnknown(Error::InvalidResponse))
));
if expect_success {
result.unwrap();
} else {
assert!(matches!(
result,
Err(MutationError::OutcomeUnknown(Error::InvalidResponse))
));
}
fixture.requests.recv().await.unwrap();
assert!(fixture.requests.try_recv().is_err());
}
}
}
codex-rs/cloud-client/src/wire.rs
+32
View File
@@ -1,6 +1,12 @@
//! Minimal ThreadService protobuf envelopes; field numbers are the wire contract.
//! Length-delimited nested events remain intact, including unknown protobuf fields.
use bytes::Buf;
use bytes::Bytes;
use prost::Message;
use std::marker::PhantomData;
use tonic::codec::Codec;
#[derive(Clone, prost::Message)]
pub(crate) struct ResumeResponse {}
@@ -23,3 +29,29 @@ pub(crate) enum Payload {
#[prost(bytes = "bytes", tag = "2")]
ServerRequest(bytes::Bytes),
}
// Tonic enforces message bounds and status trailers while callers decode the raw payload.
pub(super) struct RawResponseCodec<T>(pub PhantomData<T>);
impl<T: Message + Send + 'static> Codec for RawResponseCodec<T> {
type Encode = T;
type Decode = Bytes;
type Encoder = tonic_prost::ProstEncoder<T>;
type Decoder = RawDecoder;
fn encoder(&mut self) -> Self::Encoder {
tonic_prost::ProstCodec::<T, ResumeResponse>::default().encoder()
}
fn decoder(&mut self) -> Self::Decoder {
RawDecoder
}
}
pub(super) struct RawDecoder;
impl tonic::codec::Decoder for RawDecoder {
type Item = Bytes;
type Error = tonic::Status;
fn decode(
&mut self,
buffer: &mut tonic::codec::DecodeBuf<'_>,
) -> Result<Option<Bytes>, tonic::Status> {
Ok(Some(buffer.copy_to_bytes(buffer.remaining())))
}
}