a3s_code_core/
agent_protocol_host.rs

1//! Code-owned adapter from the versioned headless protocol to `AgentSession`.
2//!
3//! This adapter deliberately stores no parallel run state or event journal.
4//! Exact command replay is resolved by the session's authoritative run store,
5//! and event pages are projected directly from that same store.
6
7use crate::agent_api::{AgentRunSpawn, AgentSession, ExactRecoveryError, ExactRecoveryPreparation};
8use crate::agent_protocol::{
9    validate_lower_sha256, AgentProtocolChangeSetRequestV1, AgentProtocolChangeSetV1,
10    AgentProtocolCommandReceiptV1, AgentProtocolCommandV1, AgentProtocolError,
11    AgentProtocolEventPageRequestV1, AgentProtocolEventPageV1, AgentProtocolRunIdentityV1,
12    AgentProtocolRunRecoverExactV1, AGENT_PROTOCOL_CHANGE_SET_ENCODING_V1,
13    AGENT_PROTOCOL_CHANGE_SET_FORMAT_V1, AGENT_PROTOCOL_MAX_CHANGE_SET_BYTES,
14    AGENT_PROTOCOL_MAX_EVENTS_PER_PAGE,
15};
16use crate::error::CodeError;
17use crate::release::{AgentReleaseManifest, AGENT_PROTOCOL_V1};
18use crate::run::{RunSnapshot, RunWorkspaceChangeSet};
19use crate::session_checkpoint::SessionCheckpointError;
20use base64::Engine as _;
21use sha2::{Digest, Sha256};
22use std::collections::HashMap;
23use std::sync::Arc;
24use thiserror::Error;
25use tokio::sync::{Mutex, RwLock};
26
27/// Stable failures returned by the Code-owned headless protocol adapter.
28#[derive(Debug, Error)]
29pub enum AgentProtocolHostError {
30    #[error(transparent)]
31    Protocol(#[from] AgentProtocolError),
32    #[error("A3S Code Agent command targets another release")]
33    ReleaseMismatch,
34    #[error("A3S Code Agent release declares another protocol")]
35    ReleaseProtocolMismatch,
36    #[error("A3S Code Agent command targets another session")]
37    SessionMismatch,
38    #[error("A3S Code Agent run was not found")]
39    RunNotFound,
40    #[error("A3S Code Agent run is not active; recover it from a durable checkpoint")]
41    RunUnavailable,
42    #[error("A3S Code Agent sequence cannot be represented on this host")]
43    SequenceOverflow,
44    #[error("A3S Code Agent change set is still being captured")]
45    ChangeSetPending,
46    #[error("A3S Code Agent run has no Git-compatible change set")]
47    ChangeSetUnavailable,
48    #[error(transparent)]
49    Code(#[from] CodeError),
50}
51
52impl AgentProtocolHostError {
53    pub const fn code(&self) -> &'static str {
54        match self {
55            Self::Protocol(error) => error.code(),
56            Self::ReleaseMismatch => "a3s.code.agent_protocol.release_mismatch",
57            Self::ReleaseProtocolMismatch => "a3s.code.agent_protocol.release_protocol_mismatch",
58            Self::SessionMismatch => "a3s.code.agent_protocol.session_mismatch",
59            Self::RunNotFound => "a3s.code.agent_protocol.run_not_found",
60            Self::RunUnavailable => "a3s.code.agent_protocol.run_unavailable",
61            Self::SequenceOverflow => "a3s.code.agent_protocol.sequence_overflow",
62            Self::ChangeSetPending => "a3s.code.agent_protocol.change_set_pending",
63            Self::ChangeSetUnavailable => "a3s.code.agent_protocol.change_set_unavailable",
64            Self::Code(error) => error.code(),
65        }
66    }
67}
68
69/// Failures specific to the additive evidence-bound recovery entry point.
70///
71/// Keeping checkpoint drift in this adjacent error preserves the variant set
72/// of [`AgentProtocolHostError`] for existing v1 command callers.
73#[derive(Debug, Error)]
74pub enum AgentProtocolExactRecoveryError {
75    #[error(transparent)]
76    Host(#[from] AgentProtocolHostError),
77    #[error(transparent)]
78    Checkpoint(#[from] SessionCheckpointError),
79}
80
81impl AgentProtocolExactRecoveryError {
82    pub const fn code(&self) -> &'static str {
83        match self {
84            Self::Host(error) => error.code(),
85            Self::Checkpoint(error) => error.code(),
86        }
87    }
88}
89
90impl From<AgentProtocolError> for AgentProtocolExactRecoveryError {
91    fn from(error: AgentProtocolError) -> Self {
92        Self::Host(AgentProtocolHostError::Protocol(error))
93    }
94}
95
96impl From<ExactRecoveryError> for AgentProtocolExactRecoveryError {
97    fn from(error: ExactRecoveryError) -> Self {
98        match error {
99            ExactRecoveryError::Checkpoint(error) => Self::Checkpoint(error),
100            ExactRecoveryError::Code(error) => Self::Host(AgentProtocolHostError::Code(error)),
101        }
102    }
103}
104
105/// One release- and session-bound A3S Code headless protocol host.
106///
107/// Cloud, Fleet, and other callers may transport commands and receipts, but
108/// this adapter is the sole mapping into Code's run lifecycle and event store.
109#[derive(Clone)]
110pub struct AgentProtocolHost {
111    agent_release_identity: String,
112    session: Arc<AgentSession>,
113    change_set_admission: Arc<Mutex<()>>,
114    change_set_states: Arc<RwLock<HashMap<String, ChangeSetCaptureState>>>,
115}
116
117#[derive(Debug, Clone, Copy, PartialEq, Eq)]
118enum ChangeSetCaptureState {
119    Capturing,
120}
121
122impl AgentProtocolHost {
123    pub fn new(
124        agent_release_identity: impl Into<String>,
125        session: Arc<AgentSession>,
126    ) -> Result<Self, AgentProtocolHostError> {
127        let agent_release_identity = agent_release_identity.into();
128        validate_lower_sha256("agent_release_identity", &agent_release_identity)?;
129        Ok(Self {
130            agent_release_identity,
131            session,
132            change_set_admission: Arc::new(Mutex::new(())),
133            change_set_states: Arc::new(RwLock::new(HashMap::new())),
134        })
135    }
136
137    /// Bind an admitted v1 release manifest to its Code session.
138    ///
139    /// Capability compatibility remains an activation concern for the process
140    /// host, but a manifest for another protocol can never enter this v1 host.
141    pub fn from_manifest(
142        manifest: &AgentReleaseManifest,
143        session: Arc<AgentSession>,
144    ) -> Result<Self, AgentProtocolHostError> {
145        if manifest.protocol() != AGENT_PROTOCOL_V1 {
146            return Err(AgentProtocolHostError::ReleaseProtocolMismatch);
147        }
148        Ok(Self {
149            agent_release_identity: manifest.artifact().digest().to_string(),
150            session,
151            change_set_admission: Arc::new(Mutex::new(())),
152            change_set_states: Arc::new(RwLock::new(HashMap::new())),
153        })
154    }
155
156    /// Construct a host after the Harness (or equivalent) has already verified
157    /// protocol compatibility for this manifest.
158    pub fn from_verified_manifest(
159        manifest: &AgentReleaseManifest,
160        session: Arc<AgentSession>,
161    ) -> Self {
162        debug_assert_eq!(manifest.protocol(), AGENT_PROTOCOL_V1);
163        Self {
164            agent_release_identity: manifest.artifact().digest().to_string(),
165            session,
166            change_set_admission: Arc::new(Mutex::new(())),
167            change_set_states: Arc::new(RwLock::new(HashMap::new())),
168        }
169    }
170
171    pub fn agent_release_identity(&self) -> &str {
172        &self.agent_release_identity
173    }
174
175    pub fn session(&self) -> &Arc<AgentSession> {
176        &self.session
177    }
178
179    /// Execute, cancel, or recover one exact run and return a digest-bound
180    /// receipt. Start and recovery return after Code has admitted the detached
181    /// worker; progress is observed through [`Self::event_page`].
182    pub async fn execute(
183        &self,
184        command: &AgentProtocolCommandV1,
185    ) -> Result<AgentProtocolCommandReceiptV1, AgentProtocolHostError> {
186        command.validate()?;
187        self.validate_identity(command.identity())?;
188
189        let _change_set_admission = self.change_set_admission.lock().await;
190        let replayed = match command {
191            AgentProtocolCommandV1::Start { request } => {
192                let baseline = self.prepare_change_set(&request.identity).await;
193                let spawned = match self
194                    .session
195                    .spawn_run_with_id(&request.identity.run_id, &request.prompt)
196                    .await
197                {
198                    Ok(spawned) => spawned,
199                    Err(error) => {
200                        self.mark_change_set_unavailable(&request.identity).await;
201                        return Err(error.into());
202                    }
203                };
204                self.detach_with_change_set_capture(&request.identity, spawned, baseline)
205                    .await
206            }
207            AgentProtocolCommandV1::Recover { request } => {
208                let baseline = self.prepare_change_set(&request.identity).await;
209                let spawned = match self
210                    .session
211                    .spawn_recovery_with_run_id(
212                        &request.checkpoint_run_id,
213                        &request.identity.run_id,
214                    )
215                    .await
216                {
217                    Ok(spawned) => spawned,
218                    Err(error) => {
219                        self.mark_change_set_unavailable(&request.identity).await;
220                        return Err(error.into());
221                    }
222                };
223                self.detach_with_change_set_capture(&request.identity, spawned, baseline)
224                    .await
225            }
226            AgentProtocolCommandV1::Cancel { request } => {
227                let snapshot = self.snapshot(&request.identity).await?;
228                if snapshot.status.is_terminal() {
229                    true
230                } else if self.session.cancel_run(&request.identity.run_id).await {
231                    false
232                } else if self.snapshot(&request.identity).await?.status.is_terminal() {
233                    true
234                } else {
235                    return Err(AgentProtocolHostError::RunUnavailable);
236                }
237            }
238        };
239
240        let snapshot = self.snapshot(command.identity()).await?;
241        let receipt = AgentProtocolCommandReceiptV1 {
242            schema: AgentProtocolCommandReceiptV1::SCHEMA.into(),
243            action: command.action(),
244            request_id: command.request_id().into(),
245            identity: command.identity().clone(),
246            command_digest: command.digest()?,
247            state: snapshot.status.into(),
248            latest_event_sequence_exclusive: u64::try_from(snapshot.event_count)
249                .map_err(|_| AgentProtocolHostError::SequenceOverflow)?,
250            observed_at_ms: now_ms().max(snapshot.updated_at_ms),
251            replayed,
252        };
253        receipt.validate_for(command)?;
254        Ok(receipt)
255    }
256
257    /// Recover only when the locally loadable loop boundary matches the
258    /// logical component of the complete portable-checkpoint descriptor.
259    ///
260    /// This additive entry point leaves [`AgentProtocolCommandV1`] and its v1
261    /// transport shape unchanged. Validation pins the immutable checkpoint in
262    /// memory before workspace baseline capture and before the target Run is
263    /// admitted, so a concurrently overwritten store entry cannot change what
264    /// the worker resumes.
265    pub async fn execute_exact_recovery(
266        &self,
267        request: &AgentProtocolRunRecoverExactV1,
268    ) -> Result<AgentProtocolCommandReceiptV1, AgentProtocolExactRecoveryError> {
269        request.validate()?;
270        self.validate_identity(&request.identity)?;
271        let logical_resume = request.logical_resume()?;
272
273        let _change_set_admission = self.change_set_admission.lock().await;
274        let prepared = self
275            .session
276            .prepare_recovery_with_evidence(
277                logical_resume,
278                &request.checkpoint.descriptor_digest,
279                &request.identity.run_id,
280            )
281            .await?;
282        self.finish_exact_recovery(request, prepared).await
283    }
284
285    /// Recover from the logical value decoded from the same validated portable
286    /// checkpoint as `request`, without first publishing split store writes.
287    pub async fn execute_exact_recovery_from_checkpoint(
288        &self,
289        request: &AgentProtocolRunRecoverExactV1,
290        checkpoint: crate::loop_checkpoint::LoopCheckpoint,
291    ) -> Result<AgentProtocolCommandReceiptV1, AgentProtocolExactRecoveryError> {
292        request.validate()?;
293        self.validate_identity(&request.identity)?;
294        let logical_resume = request.logical_resume()?;
295
296        let _change_set_admission = self.change_set_admission.lock().await;
297        let prepared = self
298            .session
299            .prepare_recovery_from_checkpoint(
300                logical_resume,
301                &request.checkpoint.descriptor_digest,
302                &request.identity.run_id,
303                checkpoint,
304            )
305            .await?;
306        self.finish_exact_recovery(request, prepared).await
307    }
308
309    async fn finish_exact_recovery(
310        &self,
311        request: &AgentProtocolRunRecoverExactV1,
312        prepared: ExactRecoveryPreparation,
313    ) -> Result<AgentProtocolCommandReceiptV1, AgentProtocolExactRecoveryError> {
314        let replayed = match prepared {
315            ExactRecoveryPreparation::Replayed(spawned) => spawned.replayed(),
316            ExactRecoveryPreparation::Ready(prepared) => {
317                let baseline = self.prepare_change_set(&request.identity).await;
318                let spawned = match self.session.spawn_prepared_recovery(prepared).await {
319                    Ok(spawned) => spawned,
320                    Err(error) => {
321                        self.mark_change_set_unavailable(&request.identity).await;
322                        return Err(error.into());
323                    }
324                };
325                self.detach_with_change_set_capture(&request.identity, spawned, baseline)
326                    .await
327            }
328        };
329
330        let snapshot = self.snapshot(&request.identity).await?;
331        let receipt = AgentProtocolCommandReceiptV1 {
332            schema: AgentProtocolCommandReceiptV1::SCHEMA.into(),
333            action: crate::agent_protocol::AgentProtocolCommandActionV1::Recover,
334            request_id: request.request_id.clone(),
335            identity: request.identity.clone(),
336            command_digest: request.digest()?,
337            state: snapshot.status.into(),
338            latest_event_sequence_exclusive: u64::try_from(snapshot.event_count)
339                .map_err(|_| AgentProtocolHostError::SequenceOverflow)?,
340            observed_at_ms: now_ms().max(snapshot.updated_at_ms),
341            replayed,
342        };
343        receipt.validate_for_exact_recovery(request)?;
344        Ok(receipt)
345    }
346
347    /// Project a bounded cursor page directly from Code's authoritative run
348    /// store without introducing a second provider event model.
349    pub async fn event_page(
350        &self,
351        identity: &AgentProtocolRunIdentityV1,
352        after_event_sequence: Option<u64>,
353        limit: usize,
354    ) -> Result<AgentProtocolEventPageV1, AgentProtocolHostError> {
355        identity.validate()?;
356        self.validate_identity(identity)?;
357        if limit == 0 || limit > AGENT_PROTOCOL_MAX_EVENTS_PER_PAGE {
358            return Err(AgentProtocolError::InvalidField("limit").into());
359        }
360        let after_sequence = after_event_sequence
361            .map(|sequence| {
362                usize::try_from(sequence).map_err(|_| AgentProtocolHostError::SequenceOverflow)
363            })
364            .transpose()?;
365        let observation = self
366            .session
367            .run_event_observation(&identity.run_id, after_sequence, limit)
368            .await
369            .ok_or(AgentProtocolHostError::RunNotFound)?;
370        if observation.snapshot.session_id != identity.session_id {
371            return Err(AgentProtocolHostError::SessionMismatch);
372        }
373        AgentProtocolEventPageV1::from_run_page(
374            identity.clone(),
375            observation.snapshot.status,
376            now_ms().max(observation.snapshot.updated_at_ms),
377            after_sequence,
378            &observation.page,
379        )
380        .map_err(Into::into)
381    }
382
383    /// Execute the canonical transport-facing event page query.
384    pub async fn event_page_for(
385        &self,
386        request: &AgentProtocolEventPageRequestV1,
387    ) -> Result<AgentProtocolEventPageV1, AgentProtocolHostError> {
388        request.validate()?;
389        self.event_page(
390            &request.identity,
391            request.after_event_sequence,
392            usize::from(request.limit),
393        )
394        .await
395    }
396
397    /// Read the immutable Git-compatible patch captured for one terminal run.
398    pub async fn change_set_for(
399        &self,
400        request: &AgentProtocolChangeSetRequestV1,
401    ) -> Result<AgentProtocolChangeSetV1, AgentProtocolHostError> {
402        request.validate()?;
403        self.validate_identity(&request.identity)?;
404        let snapshot = self.snapshot(&request.identity).await?;
405        if !snapshot.status.is_terminal() {
406            return Err(AgentProtocolHostError::ChangeSetPending);
407        }
408        if let Some(change_set) = snapshot.workspace_change_set {
409            let response = AgentProtocolChangeSetV1 {
410                schema: AgentProtocolChangeSetV1::SCHEMA.into(),
411                identity: request.identity.clone(),
412                state: snapshot.status.into(),
413                format: AGENT_PROTOCOL_CHANGE_SET_FORMAT_V1.into(),
414                encoding: AGENT_PROTOCOL_CHANGE_SET_ENCODING_V1.into(),
415                base_tree: change_set.base_tree,
416                result_tree: change_set.result_tree,
417                patch_digest: change_set.patch_digest,
418                patch_bytes: change_set.patch_bytes,
419                patch_base64: change_set.patch_base64,
420                observed_at_ms: change_set.observed_at_ms,
421            };
422            response.validate()?;
423            return Ok(response);
424        }
425        match self
426            .change_set_states
427            .read()
428            .await
429            .get(&request.identity.run_id)
430            .copied()
431        {
432            Some(ChangeSetCaptureState::Capturing) => Err(AgentProtocolHostError::ChangeSetPending),
433            None => Err(AgentProtocolHostError::ChangeSetUnavailable),
434        }
435    }
436
437    async fn prepare_change_set(
438        &self,
439        identity: &AgentProtocolRunIdentityV1,
440    ) -> Option<crate::git::WorkspaceTreeSnapshot> {
441        if self.session.run_snapshot(&identity.run_id).await.is_some() {
442            return None;
443        }
444        let workspace = self.session.workspace().to_path_buf();
445        let baseline =
446            tokio::task::spawn_blocking(move || crate::git::snapshot_workspace_tree(&workspace))
447                .await
448                .ok()
449                .and_then(Result::ok);
450        if baseline.is_some() {
451            self.change_set_states
452                .write()
453                .await
454                .insert(identity.run_id.clone(), ChangeSetCaptureState::Capturing);
455        }
456        baseline
457    }
458
459    async fn mark_change_set_unavailable(&self, identity: &AgentProtocolRunIdentityV1) {
460        self.change_set_states
461            .write()
462            .await
463            .remove(&identity.run_id);
464    }
465
466    async fn detach_with_change_set_capture(
467        &self,
468        identity: &AgentProtocolRunIdentityV1,
469        spawned: AgentRunSpawn,
470        baseline: Option<crate::git::WorkspaceTreeSnapshot>,
471    ) -> bool {
472        match spawned {
473            AgentRunSpawn::Started { worker, .. } => {
474                let Some(baseline) = baseline else {
475                    drop(worker);
476                    return false;
477                };
478                let workspace = self.session.workspace().to_path_buf();
479                let session = Arc::clone(&self.session);
480                let run_id = identity.run_id.clone();
481                let pin_identity = format!(
482                    "{}:{}:{}",
483                    self.agent_release_identity, identity.session_id, identity.run_id
484                );
485                let states = Arc::clone(&self.change_set_states);
486                tokio::spawn(async move {
487                    let _ = worker.await;
488                    let run_id_for_state = run_id.clone();
489                    let failed = session
490                        .run_snapshot(&run_id_for_state)
491                        .await
492                        .is_some_and(|snapshot| snapshot.status == crate::run::RunStatus::Failed);
493                    let evidence = tokio::task::spawn_blocking(move || {
494                        let result = crate::git::snapshot_workspace_tree(&workspace)?;
495                        let patch = crate::git::diff_workspace_trees(
496                            &workspace,
497                            &baseline,
498                            &result,
499                            AGENT_PROTOCOL_MAX_CHANGE_SET_BYTES,
500                        )?;
501                        crate::git::pin_workspace_tree(&workspace, &pin_identity, &result)?;
502                        if failed {
503                            let _ =
504                                crate::git::restore_unverified_worktree(&workspace, &baseline.tree);
505                        }
506                        let patch_bytes = u64::try_from(patch.len())
507                            .map_err(|_| anyhow::anyhow!("change-set byte count overflowed"))?;
508                        Ok::<_, anyhow::Error>(RunWorkspaceChangeSet {
509                            base_tree: format!("git-tree:{}", baseline.tree),
510                            result_tree: format!("git-tree:{}", result.tree),
511                            patch_digest: format!("sha256:{:x}", Sha256::digest(&patch)),
512                            patch_bytes,
513                            patch_base64: base64::engine::general_purpose::STANDARD.encode(patch),
514                            observed_at_ms: now_ms(),
515                        })
516                    })
517                    .await
518                    .ok()
519                    .and_then(Result::ok);
520                    if let Some(evidence) = evidence {
521                        let _ = session
522                            .record_workspace_change_set(&run_id_for_state, evidence)
523                            .await;
524                    }
525                    // A completed or failed capture is represented by the
526                    // authoritative Run snapshot. Retain only in-flight
527                    // entries so this map cannot grow with session age.
528                    states.write().await.remove(&run_id_for_state);
529                });
530                false
531            }
532            AgentRunSpawn::Replayed { .. } => true,
533        }
534    }
535
536    fn validate_identity(
537        &self,
538        identity: &AgentProtocolRunIdentityV1,
539    ) -> Result<(), AgentProtocolHostError> {
540        if identity.agent_release_identity != self.agent_release_identity {
541            return Err(AgentProtocolHostError::ReleaseMismatch);
542        }
543        if identity.session_id != self.session.session_id() {
544            return Err(AgentProtocolHostError::SessionMismatch);
545        }
546        Ok(())
547    }
548
549    async fn snapshot(
550        &self,
551        identity: &AgentProtocolRunIdentityV1,
552    ) -> Result<RunSnapshot, AgentProtocolHostError> {
553        let snapshot = self
554            .session
555            .run_snapshot(&identity.run_id)
556            .await
557            .ok_or(AgentProtocolHostError::RunNotFound)?;
558        if snapshot.session_id != identity.session_id {
559            return Err(AgentProtocolHostError::SessionMismatch);
560        }
561        Ok(snapshot)
562    }
563}
564
565fn now_ms() -> u64 {
566    std::time::SystemTime::now()
567        .duration_since(std::time::UNIX_EPOCH)
568        .map(|duration| duration.as_millis() as u64)
569        .unwrap_or_default()
570}
571
572#[cfg(test)]
573mod tests {
574    use super::*;
575    use crate::agent_protocol::AgentProtocolError;
576    use crate::error::CodeError;
577    use crate::session_checkpoint::SessionCheckpointError;
578
579    #[test]
580    fn host_error_codes_are_stable() {
581        assert_eq!(
582            AgentProtocolHostError::Protocol(AgentProtocolError::Encoding).code(),
583            AgentProtocolError::Encoding.code()
584        );
585        assert_eq!(
586            AgentProtocolHostError::ReleaseMismatch.code(),
587            "a3s.code.agent_protocol.release_mismatch"
588        );
589        assert_eq!(
590            AgentProtocolHostError::ReleaseProtocolMismatch.code(),
591            "a3s.code.agent_protocol.release_protocol_mismatch"
592        );
593        assert_eq!(
594            AgentProtocolHostError::SessionMismatch.code(),
595            "a3s.code.agent_protocol.session_mismatch"
596        );
597        assert_eq!(
598            AgentProtocolHostError::RunNotFound.code(),
599            "a3s.code.agent_protocol.run_not_found"
600        );
601        assert_eq!(
602            AgentProtocolHostError::RunUnavailable.code(),
603            "a3s.code.agent_protocol.run_unavailable"
604        );
605        assert_eq!(
606            AgentProtocolHostError::SequenceOverflow.code(),
607            "a3s.code.agent_protocol.sequence_overflow"
608        );
609        assert_eq!(
610            AgentProtocolHostError::ChangeSetPending.code(),
611            "a3s.code.agent_protocol.change_set_pending"
612        );
613        assert_eq!(
614            AgentProtocolHostError::ChangeSetUnavailable.code(),
615            "a3s.code.agent_protocol.change_set_unavailable"
616        );
617        assert_eq!(
618            AgentProtocolHostError::Code(CodeError::TaskSchedulerClosed).code(),
619            CodeError::TaskSchedulerClosed.code()
620        );
621    }
622
623    #[test]
624    fn exact_recovery_error_codes_and_from_impls_are_stable() {
625        let protocol: AgentProtocolExactRecoveryError = AgentProtocolError::IdentityMismatch.into();
626        assert_eq!(protocol.code(), AgentProtocolError::IdentityMismatch.code());
627        assert_eq!(
628            AgentProtocolExactRecoveryError::Checkpoint(SessionCheckpointError::ContentDrift(
629                "drift".into()
630            ))
631            .code(),
632            SessionCheckpointError::ContentDrift("drift".into()).code()
633        );
634        assert_eq!(
635            AgentProtocolExactRecoveryError::Host(AgentProtocolHostError::RunNotFound).code(),
636            AgentProtocolHostError::RunNotFound.code()
637        );
638        let from_exact = AgentProtocolExactRecoveryError::from(ExactRecoveryError::Checkpoint(
639            SessionCheckpointError::InvalidPayload("x".into()),
640        ));
641        assert!(matches!(
642            from_exact,
643            AgentProtocolExactRecoveryError::Checkpoint(_)
644        ));
645        let from_code = AgentProtocolExactRecoveryError::from(ExactRecoveryError::Code(
646            CodeError::TaskSchedulerClosed,
647        ));
648        assert!(matches!(
649            from_code,
650            AgentProtocolExactRecoveryError::Host(AgentProtocolHostError::Code(_))
651        ));
652    }
653}
a3s_code_core/agent_protocol_host.rs

a3s_code_core/
agent_protocol_host.rs
