relay_knowledge/interfaces/agent/
acp.rs

1use std::{
2    collections::HashMap,
3    fmt,
4    sync::{Arc, Mutex},
5    time::{Duration, Instant},
6};
7
8use serde::{Deserialize, Serialize};
9use serde_json::{Value, json};
10use tokio::sync::watch;
11
12use crate::{
13    api::{
14        AgentProtocolKind, AgentRetrievalResult, ErrorKind, HybridRetrievalRequest, InterfaceKind,
15        RequestContext, RuntimeIdentity,
16    },
17    application::{AgentRuntimeConfig, RelayKnowledgeService},
18    domain::FreshnessPolicy,
19    net::{
20        NetworkRuntime,
21        qos::{QosPermit, QosRuntime, RejectReason},
22    },
23    observability::AgentProtocolMetrics,
24};
25
26use super::{
27    AgentAdapterError, AgentAdapterErrorKind, AgentAuditEvent, AgentAuditLog,
28    AgentAuditQosDecision, AgentAuditSink, AgentAuditStatus, authorize_limit, authorize_scope,
29};
30
31/// Local ACP session adapter for resident relay-knowledge processes.
32#[derive(Clone)]
33pub struct LocalAcpSessionAdapter {
34    service: RelayKnowledgeService,
35    network: NetworkRuntime,
36    agent: AgentRuntimeConfig,
37    qos: QosRuntime,
38    audit: AgentAuditLog,
39    metrics: AgentProtocolMetrics,
40    sessions: AcpSessionRegistry,
41}
42
43impl LocalAcpSessionAdapter {
44    /// Creates an ACP local session adapter without opening sockets.
45    pub fn new(
46        service: RelayKnowledgeService,
47        network: NetworkRuntime,
48        agent: AgentRuntimeConfig,
49    ) -> Self {
50        let metrics = service.observability().agent_metrics();
51        let audit = if agent.audit_sink_enabled {
52            AgentAuditSink::jsonl(service.agent_audit_log_path(), agent.audit_queue_depth)
53                .map(AgentAuditLog::with_sink)
54                .unwrap_or_default()
55        } else {
56            AgentAuditLog::default()
57        };
58
59        Self {
60            service,
61            network,
62            agent,
63            qos: QosRuntime::default(),
64            audit,
65            metrics,
66            sessions: AcpSessionRegistry::default(),
67        }
68    }
69
70    /// Returns the ACP initialize capability payload.
71    pub fn initialize(&self) -> AcpInitializeResponse {
72        AcpInitializeResponse {
73            meta: AcpInitializeMeta {
74                relay_knowledge: AcpRelayKnowledgeCapability {
75                    graph_retrieval: true,
76                    read_only: true,
77                    supports_cancellation: true,
78                    supports_index_refresh_permission: true,
79                },
80            },
81        }
82    }
83
84    /// Creates a bounded local ACP session and captures untrusted client identity.
85    pub fn new_session(&self, request: AcpSessionRequest) -> Result<AcpSession, AgentAdapterError> {
86        let permit = self.admit_request()?;
87        let session_id = generate_acp_id("acp-session")?;
88        let record = AcpSessionRecord {
89            client_name: normalized_optional(request.client_name),
90            client_version: normalized_optional(request.client_version),
91            actor_id: normalized_optional(request.actor_id),
92        };
93        self.sessions
94            .insert_session(session_id.clone(), record.clone());
95        drop(permit);
96
97        Ok(AcpSession {
98            session_id: session_id.clone(),
99            runtime_identity: record.identity(&session_id, None),
100            policy_id: "local-acp-policy".to_owned(),
101            authorized_scope_count: self.agent.access_policy.allowed_scopes.len(),
102        })
103    }
104
105    /// Runs an ACP prompt turn, returning progress updates and a context artifact.
106    pub async fn prompt(
107        &self,
108        session_id: &str,
109        mut request: AcpPromptRequest,
110    ) -> AcpPromptResponse {
111        let started = Instant::now();
112        let request_id = request.request_id.take().unwrap_or_else(|| {
113            generate_acp_id("acp-request").unwrap_or_else(|_| "acp-request-unavailable".to_owned())
114        });
115        let mut updates = vec![AcpSessionUpdate::pending(&request_id, "accepted")];
116        let Some(session) = self.sessions.session(session_id) else {
117            return failed_prompt(
118                session_id,
119                request_id,
120                updates,
121                AgentAdapterError::new(
122                    AgentAdapterErrorKind::InvalidArgument,
123                    "unknown ACP session",
124                ),
125                elapsed_millis(started),
126            );
127        };
128        let permit = match self.admit_request() {
129            Ok(permit) => permit,
130            Err(error) => {
131                self.record_audit(AcpAuditInput {
132                    operation: "session/prompt",
133                    request_id: &request_id,
134                    session_id,
135                    session: &session,
136                    qos_decision: AgentAuditQosDecision::Rejected,
137                    status: AgentAuditStatus::Failed,
138                    source_scope: None,
139                    freshness: None,
140                    limit: None,
141                    result_count: None,
142                    truncated: false,
143                    elapsed_ms: elapsed_millis(started),
144                    error_kind: Some(error.kind.as_str()),
145                });
146                return failed_prompt(
147                    session_id,
148                    request_id,
149                    updates,
150                    error,
151                    elapsed_millis(started),
152                );
153            }
154        };
155        updates.push(AcpSessionUpdate::in_progress(
156            &request_id,
157            "retrieval request mapped",
158        ));
159
160        let mapped = match map_prompt_request(&self.agent, request) {
161            Ok(mapped) => mapped,
162            Err(error) => {
163                drop(permit);
164                self.record_audit(AcpAuditInput {
165                    operation: "session/prompt",
166                    request_id: &request_id,
167                    session_id,
168                    session: &session,
169                    qos_decision: AgentAuditQosDecision::Admitted,
170                    status: AgentAuditStatus::Failed,
171                    source_scope: None,
172                    freshness: None,
173                    limit: None,
174                    result_count: None,
175                    truncated: false,
176                    elapsed_ms: elapsed_millis(started),
177                    error_kind: Some(error.kind.as_str()),
178                });
179                return failed_prompt(
180                    session_id,
181                    request_id,
182                    updates,
183                    error,
184                    elapsed_millis(started),
185                );
186            }
187        };
188        updates.push(AcpSessionUpdate::meta(
189            &request_id,
190            "freshness checked",
191            json!({
192                "relayKnowledge": {
193                    "freshness": crate::api::freshness_label(mapped.freshness),
194                    "source_scope": mapped.source_scope
195                }
196            }),
197        ));
198
199        let (mut cancellation, registration) = self
200            .sessions
201            .register_request(session_id, request_id.clone());
202        let identity = session.identity(session_id, Some(request_id.clone()));
203        let context = RequestContext::with_ids(
204            InterfaceKind::Acp,
205            request_id.clone(),
206            format!("trace-acp-{request_id}"),
207        );
208        let service = self.service.clone();
209        let request_timeout = Duration::from_millis(self.agent.access_policy.max_runtime_ms);
210        let source_scope = mapped.source_scope.clone();
211        let freshness = mapped.freshness;
212        let limit = mapped.limit;
213        let max_context_bytes = self.agent.access_policy.max_context_bytes;
214        let retrieval = service.retrieve_context(mapped.into_retrieval_request(), context);
215
216        let response = tokio::select! {
217            result = tokio::time::timeout(request_timeout, retrieval) => {
218                match result {
219                    Ok(Ok(response)) => {
220                        let result = AgentRetrievalResult::from_retrieval(
221                            response,
222                            identity,
223                            max_context_bytes,
224                            elapsed_millis(started),
225                        );
226                        let artifact_id = format!("relay-context:{session_id}:{request_id}");
227                        updates.push(AcpSessionUpdate::meta(
228                            &request_id,
229                            "context ready",
230                            json!({"relayKnowledge": {"artifact_id": artifact_id}}),
231                        ));
232                        updates.push(AcpSessionUpdate::completed(&request_id, "completed"));
233                        self.record_audit(AcpAuditInput {
234                            operation: "session/prompt",
235                            request_id: &request_id,
236                            session_id,
237                            session: &session,
238                            qos_decision: AgentAuditQosDecision::Admitted,
239                            status: AgentAuditStatus::Completed,
240                            source_scope: source_scope.as_deref(),
241                            freshness: Some(crate::api::freshness_label(freshness)),
242                            limit: Some(limit),
243                            result_count: Some(result.results.len()),
244                            truncated: result.truncated,
245                            elapsed_ms: elapsed_millis(started),
246                            error_kind: None,
247                        });
248                        AcpPromptResponse {
249                            session_id: session_id.to_owned(),
250                            request_id: request_id.clone(),
251                            updates,
252                            context_artifact: Some(AcpContextArtifact {
253                                artifact_id,
254                                result,
255                            }),
256                            stop_reason: AcpStopReason::Completed,
257                            error: None,
258                        }
259                    }
260                    Ok(Err(error)) => {
261                        let adapter_error = AgentAdapterError::new(
262                            api_error_kind(error.error_kind),
263                            error.message,
264                        );
265                        self.record_audit(AcpAuditInput {
266                            operation: "session/prompt",
267                            request_id: &request_id,
268                            session_id,
269                            session: &session,
270                            qos_decision: AgentAuditQosDecision::Admitted,
271                            status: AgentAuditStatus::Failed,
272                            source_scope: source_scope.as_deref(),
273                            freshness: Some(crate::api::freshness_label(freshness)),
274                            limit: Some(limit),
275                            result_count: None,
276                            truncated: false,
277                            elapsed_ms: elapsed_millis(started),
278                            error_kind: Some(adapter_error.kind.as_str()),
279                        });
280                        failed_prompt(session_id, request_id.clone(), updates, adapter_error, elapsed_millis(started))
281                    }
282                    Err(_) => {
283                        let adapter_error = AgentAdapterError::new(
284                            AgentAdapterErrorKind::Timeout,
285                            "ACP prompt exceeded max_runtime_ms",
286                        );
287                        self.record_audit(AcpAuditInput {
288                            operation: "session/prompt",
289                            request_id: &request_id,
290                            session_id,
291                            session: &session,
292                            qos_decision: AgentAuditQosDecision::Admitted,
293                            status: AgentAuditStatus::Failed,
294                            source_scope: source_scope.as_deref(),
295                            freshness: Some(crate::api::freshness_label(freshness)),
296                            limit: Some(limit),
297                            result_count: None,
298                            truncated: false,
299                            elapsed_ms: elapsed_millis(started),
300                            error_kind: Some(adapter_error.kind.as_str()),
301                        });
302                        failed_prompt(session_id, request_id.clone(), updates, adapter_error, elapsed_millis(started))
303                    }
304                }
305            }
306            _ = wait_for_cancellation(&mut cancellation) => {
307                let adapter_error = AgentAdapterError::new(
308                    AgentAdapterErrorKind::Cancelled,
309                    "ACP prompt was cancelled",
310                );
311                self.record_audit(AcpAuditInput {
312                    operation: "session/prompt",
313                    request_id: &request_id,
314                    session_id,
315                    session: &session,
316                    qos_decision: AgentAuditQosDecision::Admitted,
317                    status: AgentAuditStatus::Cancelled,
318                    source_scope: source_scope.as_deref(),
319                    freshness: Some(crate::api::freshness_label(freshness)),
320                    limit: Some(limit),
321                    result_count: None,
322                    truncated: false,
323                    elapsed_ms: elapsed_millis(started),
324                    error_kind: Some(adapter_error.kind.as_str()),
325                });
326                failed_prompt(session_id, request_id.clone(), updates, adapter_error, elapsed_millis(started))
327            }
328        };
329
330        registration.release();
331        drop(permit);
332        response
333    }
334
335    /// Cancels an active prompt request if the session still owns it.
336    pub fn cancel(&self, session_id: &str, request_id: &str) -> bool {
337        self.sessions.cancel_request(session_id, request_id)
338    }
339
340    /// Returns agent audit events retained by the bounded in-process log.
341    pub fn audit_snapshot(&self) -> Vec<AgentAuditEvent> {
342        self.audit.snapshot()
343    }
344
345    #[cfg(test)]
346    pub fn qos_snapshot(&self) -> crate::net::qos::QosSnapshot {
347        self.qos.snapshot()
348    }
349
350    fn admit_request(&self) -> Result<QosPermit, AgentAdapterError> {
351        let policy = self.network.current().qos;
352        let queued = self.qos.reserve_queue(&policy).map_err(qos_error)?;
353        let permit = self.qos.admit_request(&policy).map_err(qos_error);
354        drop(queued);
355        permit
356    }
357
358    fn record_audit(&self, input: AcpAuditInput<'_>) {
359        let event = AgentAuditEvent {
360            sequence: 0,
361            protocol: AgentProtocolKind::Acp,
362            operation: input.operation.to_owned(),
363            request_id: input.request_id.to_owned(),
364            trace_id: format!("trace-acp-{}", input.request_id),
365            runtime_identity: input
366                .session
367                .identity(input.session_id, Some(input.request_id.to_owned())),
368            qos_decision: input.qos_decision,
369            status: input.status,
370            source_scope: input.source_scope.map(str::to_owned),
371            freshness: input.freshness.map(str::to_owned),
372            limit: input.limit,
373            result_count: input.result_count,
374            truncated: input.truncated,
375            elapsed_ms: input.elapsed_ms,
376            error_kind: input.error_kind.map(str::to_owned),
377        };
378        self.audit.record(event.clone());
379        if input.qos_decision == AgentAuditQosDecision::Rejected {
380            self.metrics
381                .record_rejection("acp", input.error_kind.unwrap_or("qos_rejected"));
382            return;
383        }
384        let status_label = match event.status {
385            AgentAuditStatus::Completed => "completed",
386            AgentAuditStatus::Failed => "failed",
387            AgentAuditStatus::Cancelled => "cancelled",
388        };
389        self.metrics.record_request(
390            "acp",
391            input.operation,
392            status_label,
393            input.elapsed_ms,
394            input.truncated,
395        );
396        if event.status == AgentAuditStatus::Cancelled {
397            self.metrics.record_cancelled("acp");
398        }
399    }
400}
401
402struct AcpAuditInput<'a> {
403    operation: &'a str,
404    request_id: &'a str,
405    session_id: &'a str,
406    session: &'a AcpSessionRecord,
407    qos_decision: AgentAuditQosDecision,
408    status: AgentAuditStatus,
409    source_scope: Option<&'a str>,
410    freshness: Option<&'a str>,
411    limit: Option<usize>,
412    result_count: Option<usize>,
413    truncated: bool,
414    elapsed_ms: u64,
415    error_kind: Option<&'a str>,
416}
417
418/// ACP initialize response with relay-knowledge capability metadata.
419#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
420pub struct AcpInitializeResponse {
421    #[serde(rename = "_meta")]
422    pub meta: AcpInitializeMeta,
423}
424
425#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
426pub struct AcpInitializeMeta {
427    #[serde(rename = "relayKnowledge")]
428    pub relay_knowledge: AcpRelayKnowledgeCapability,
429}
430
431#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
432pub struct AcpRelayKnowledgeCapability {
433    #[serde(rename = "graphRetrieval")]
434    pub graph_retrieval: bool,
435    #[serde(rename = "readOnly")]
436    pub read_only: bool,
437    #[serde(rename = "supportsCancellation")]
438    pub supports_cancellation: bool,
439    #[serde(rename = "supportsIndexRefreshPermission")]
440    pub supports_index_refresh_permission: bool,
441}
442
443/// Local ACP session request.
444#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
445pub struct AcpSessionRequest {
446    #[serde(skip_serializing_if = "Option::is_none")]
447    pub client_name: Option<String>,
448    #[serde(skip_serializing_if = "Option::is_none")]
449    pub client_version: Option<String>,
450    #[serde(skip_serializing_if = "Option::is_none")]
451    pub actor_id: Option<String>,
452}
453
454/// Created ACP session metadata.
455#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
456pub struct AcpSession {
457    pub session_id: String,
458    pub runtime_identity: RuntimeIdentity,
459    pub policy_id: String,
460    pub authorized_scope_count: usize,
461}
462
463/// ACP prompt request with structured relay metadata.
464#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
465pub struct AcpPromptRequest {
466    pub prompt: String,
467    #[serde(skip_serializing_if = "Option::is_none")]
468    pub request_id: Option<String>,
469    #[serde(rename = "_meta", skip_serializing_if = "Option::is_none")]
470    pub meta: Option<AcpPromptMeta>,
471}
472
473#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
474pub struct AcpPromptMeta {
475    #[serde(rename = "relayKnowledge", skip_serializing_if = "Option::is_none")]
476    pub relay_knowledge: Option<AcpRelayKnowledgePrompt>,
477}
478
479#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
480pub struct AcpRelayKnowledgePrompt {
481    #[serde(skip_serializing_if = "Option::is_none")]
482    pub query: Option<String>,
483    #[serde(skip_serializing_if = "Option::is_none")]
484    pub source_scope: Option<String>,
485    #[serde(skip_serializing_if = "Option::is_none")]
486    pub limit: Option<usize>,
487    #[serde(skip_serializing_if = "Option::is_none")]
488    pub freshness: Option<String>,
489}
490
491/// ACP prompt response containing bounded progress and an optional context artifact.
492#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
493pub struct AcpPromptResponse {
494    pub session_id: String,
495    pub request_id: String,
496    pub updates: Vec<AcpSessionUpdate>,
497    #[serde(skip_serializing_if = "Option::is_none")]
498    pub context_artifact: Option<AcpContextArtifact>,
499    pub stop_reason: AcpStopReason,
500    #[serde(skip_serializing_if = "Option::is_none")]
501    pub error: Option<AcpErrorPayload>,
502}
503
504#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
505pub struct AcpContextArtifact {
506    pub artifact_id: String,
507    pub result: AgentRetrievalResult,
508}
509
510#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
511#[serde(rename_all = "snake_case")]
512pub enum AcpStopReason {
513    Completed,
514    Failed,
515    Cancelled,
516}
517
518#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
519pub struct AcpErrorPayload {
520    pub error_kind: String,
521    pub message: String,
522}
523
524#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
525pub struct AcpSessionUpdate {
526    pub request_id: String,
527    pub kind: AcpSessionUpdateKind,
528    pub status: AcpSessionUpdateStatus,
529    pub message: String,
530    #[serde(rename = "_meta", skip_serializing_if = "Option::is_none")]
531    pub meta: Option<Value>,
532}
533
534impl AcpSessionUpdate {
535    fn pending(request_id: &str, message: &str) -> Self {
536        Self::new(
537            request_id,
538            AcpSessionUpdateKind::ToolCallUpdate,
539            AcpSessionUpdateStatus::Pending,
540            message,
541            None,
542        )
543    }
544
545    fn in_progress(request_id: &str, message: &str) -> Self {
546        Self::new(
547            request_id,
548            AcpSessionUpdateKind::ToolCallUpdate,
549            AcpSessionUpdateStatus::InProgress,
550            message,
551            None,
552        )
553    }
554
555    fn meta(request_id: &str, message: &str, meta: Value) -> Self {
556        Self::new(
557            request_id,
558            AcpSessionUpdateKind::SessionUpdate,
559            AcpSessionUpdateStatus::InProgress,
560            message,
561            Some(meta),
562        )
563    }
564
565    fn completed(request_id: &str, message: &str) -> Self {
566        Self::new(
567            request_id,
568            AcpSessionUpdateKind::ToolCallUpdate,
569            AcpSessionUpdateStatus::Completed,
570            message,
571            None,
572        )
573    }
574
575    fn failed(request_id: &str, message: &str, status: AcpSessionUpdateStatus) -> Self {
576        Self::new(
577            request_id,
578            AcpSessionUpdateKind::ToolCallUpdate,
579            status,
580            message,
581            None,
582        )
583    }
584
585    fn new(
586        request_id: &str,
587        kind: AcpSessionUpdateKind,
588        status: AcpSessionUpdateStatus,
589        message: &str,
590        meta: Option<Value>,
591    ) -> Self {
592        Self {
593            request_id: request_id.to_owned(),
594            kind,
595            status,
596            message: message.to_owned(),
597            meta,
598        }
599    }
600}
601
602#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
603#[serde(rename_all = "snake_case")]
604pub enum AcpSessionUpdateKind {
605    SessionUpdate,
606    ToolCallUpdate,
607}
608
609#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
610#[serde(rename_all = "snake_case")]
611pub enum AcpSessionUpdateStatus {
612    Pending,
613    InProgress,
614    Completed,
615    Failed,
616    Cancelled,
617}
618
619#[derive(Clone, Default)]
620struct AcpSessionRegistry {
621    inner: Arc<Mutex<AcpSessionState>>,
622}
623
624#[derive(Default)]
625struct AcpSessionState {
626    sessions: HashMap<String, AcpSessionRecord>,
627    active_requests: HashMap<String, watch::Sender<bool>>,
628}
629
630#[derive(Debug, Clone)]
631struct AcpSessionRecord {
632    client_name: Option<String>,
633    client_version: Option<String>,
634    actor_id: Option<String>,
635}
636
637impl AcpSessionRecord {
638    fn identity(&self, session_id: &str, request_id: Option<String>) -> RuntimeIdentity {
639        RuntimeIdentity::acp(
640            self.client_name.clone(),
641            self.client_version.clone(),
642            self.actor_id.clone(),
643            session_id.to_owned(),
644            request_id,
645        )
646    }
647}
648
649struct ActiveAcpRequest {
650    registry: AcpSessionRegistry,
651    key: String,
652    released: bool,
653}
654
655impl ActiveAcpRequest {
656    fn release(mut self) {
657        self.registry.remove_request(&self.key);
658        self.released = true;
659    }
660}
661
662impl Drop for ActiveAcpRequest {
663    fn drop(&mut self) {
664        if !self.released {
665            self.registry.remove_request(&self.key);
666        }
667    }
668}
669
670impl AcpSessionRegistry {
671    fn insert_session(&self, session_id: String, record: AcpSessionRecord) {
672        self.inner
673            .lock()
674            .unwrap_or_else(|poisoned| poisoned.into_inner())
675            .sessions
676            .insert(session_id, record);
677    }
678
679    fn session(&self, session_id: &str) -> Option<AcpSessionRecord> {
680        self.inner
681            .lock()
682            .unwrap_or_else(|poisoned| poisoned.into_inner())
683            .sessions
684            .get(session_id)
685            .cloned()
686    }
687
688    fn register_request(
689        &self,
690        session_id: &str,
691        request_id: String,
692    ) -> (watch::Receiver<bool>, ActiveAcpRequest) {
693        let (sender, receiver) = watch::channel(false);
694        let key = active_request_key(session_id, &request_id);
695        self.inner
696            .lock()
697            .unwrap_or_else(|poisoned| poisoned.into_inner())
698            .active_requests
699            .insert(key.clone(), sender);
700
701        (
702            receiver,
703            ActiveAcpRequest {
704                registry: self.clone(),
705                key,
706                released: false,
707            },
708        )
709    }
710
711    fn cancel_request(&self, session_id: &str, request_id: &str) -> bool {
712        let key = active_request_key(session_id, request_id);
713        self.inner
714            .lock()
715            .unwrap_or_else(|poisoned| poisoned.into_inner())
716            .active_requests
717            .get(&key)
718            .is_some_and(|sender| sender.send(true).is_ok())
719    }
720
721    fn remove_request(&self, key: &str) {
722        self.inner
723            .lock()
724            .unwrap_or_else(|poisoned| poisoned.into_inner())
725            .active_requests
726            .remove(key);
727    }
728}
729
730#[derive(Debug, Clone, PartialEq, Eq)]
731struct MappedPromptRequest {
732    query: String,
733    source_scope: Option<String>,
734    limit: usize,
735    freshness: FreshnessPolicy,
736}
737
738impl MappedPromptRequest {
739    fn into_retrieval_request(self) -> HybridRetrievalRequest {
740        HybridRetrievalRequest {
741            query: self.query,
742            source_scope: self.source_scope,
743            limit: self.limit,
744            freshness: self.freshness,
745        }
746    }
747}
748
749fn map_prompt_request(
750    agent: &AgentRuntimeConfig,
751    request: AcpPromptRequest,
752) -> Result<MappedPromptRequest, AgentAdapterError> {
753    let relay = request
754        .meta
755        .and_then(|meta| meta.relay_knowledge)
756        .unwrap_or(AcpRelayKnowledgePrompt {
757            query: None,
758            source_scope: None,
759            limit: None,
760            freshness: None,
761        });
762    let query = relay.query.unwrap_or(request.prompt);
763    let source_scope = authorize_scope(relay.source_scope, &agent.access_policy)?;
764    let limit = authorize_limit(relay.limit, &agent.access_policy)?;
765    let freshness = parse_freshness(relay.freshness.as_deref())?;
766
767    if query.trim().is_empty() {
768        return Err(AgentAdapterError::new(
769            AgentAdapterErrorKind::InvalidArgument,
770            "ACP prompt query must not be empty",
771        ));
772    }
773
774    Ok(MappedPromptRequest {
775        query,
776        source_scope,
777        limit,
778        freshness,
779    })
780}
781
782fn parse_freshness(value: Option<&str>) -> Result<FreshnessPolicy, AgentAdapterError> {
783    match value.unwrap_or("allow-stale") {
784        "allow-stale" => Ok(FreshnessPolicy::AllowStale),
785        "wait-until-fresh" => Ok(FreshnessPolicy::WaitUntilFresh),
786        "graph-only" => Ok(FreshnessPolicy::GraphOnly),
787        other => Err(AgentAdapterError::new(
788            AgentAdapterErrorKind::InvalidArgument,
789            format!("invalid freshness '{other}'"),
790        )),
791    }
792}
793
794async fn wait_for_cancellation(cancellation: &mut watch::Receiver<bool>) {
795    while cancellation.changed().await.is_ok() {
796        if *cancellation.borrow() {
797            return;
798        }
799    }
800
801    std::future::pending::<()>().await;
802}
803
804fn failed_prompt(
805    session_id: &str,
806    request_id: String,
807    mut updates: Vec<AcpSessionUpdate>,
808    error: AgentAdapterError,
809    _elapsed_ms: u64,
810) -> AcpPromptResponse {
811    let stop_reason = if error.kind == AgentAdapterErrorKind::Cancelled {
812        AcpStopReason::Cancelled
813    } else {
814        AcpStopReason::Failed
815    };
816    let status = if error.kind == AgentAdapterErrorKind::Cancelled {
817        AcpSessionUpdateStatus::Cancelled
818    } else {
819        AcpSessionUpdateStatus::Failed
820    };
821    updates.push(AcpSessionUpdate::failed(
822        &request_id,
823        &error.message,
824        status,
825    ));
826
827    AcpPromptResponse {
828        session_id: session_id.to_owned(),
829        request_id,
830        updates,
831        context_artifact: None,
832        stop_reason,
833        error: Some(AcpErrorPayload {
834            error_kind: error.kind.as_str().to_owned(),
835            message: error.message,
836        }),
837    }
838}
839
840fn qos_error(reason: RejectReason) -> AgentAdapterError {
841    let message = match reason {
842        RejectReason::ConnectionBudgetExceeded => "connection budget exhausted",
843        RejectReason::RequestBudgetExceeded => "request budget exhausted",
844        RejectReason::QueueBudgetExceeded => "queue budget exhausted",
845    };
846
847    AgentAdapterError::new(AgentAdapterErrorKind::QosRejected, message)
848}
849
850fn api_error_kind(kind: ErrorKind) -> AgentAdapterErrorKind {
851    match kind {
852        ErrorKind::InvalidArgument => AgentAdapterErrorKind::InvalidArgument,
853        ErrorKind::StorageUnavailable => AgentAdapterErrorKind::StorageUnavailable,
854        ErrorKind::Timeout => AgentAdapterErrorKind::Timeout,
855        ErrorKind::Internal => AgentAdapterErrorKind::Internal,
856    }
857}
858
859fn active_request_key(session_id: &str, request_id: &str) -> String {
860    format!("{session_id}|{request_id}")
861}
862
863fn normalized_optional(value: Option<String>) -> Option<String> {
864    value.and_then(|value| {
865        let trimmed = value.trim();
866        (!trimmed.is_empty()).then(|| trimmed.to_owned())
867    })
868}
869
870fn generate_acp_id(prefix: &str) -> Result<String, AgentAdapterError> {
871    let mut entropy = [0_u8; 16];
872    getrandom::getrandom(&mut entropy).map_err(|_| {
873        AgentAdapterError::new(
874            AgentAdapterErrorKind::Internal,
875            "OS session entropy is unavailable",
876        )
877    })?;
878
879    Ok(format!("{prefix}-{}", lowercase_hex(&entropy)))
880}
881
882fn lowercase_hex(bytes: &[u8]) -> String {
883    const HEX: &[u8; 16] = b"0123456789abcdef";
884    let mut output = String::with_capacity(bytes.len() * 2);
885    for byte in bytes {
886        output.push(HEX[usize::from(byte >> 4)] as char);
887        output.push(HEX[usize::from(byte & 0x0f)] as char);
888    }
889
890    output
891}
892
893fn elapsed_millis(started: Instant) -> u64 {
894    u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
895}
896
897impl fmt::Debug for LocalAcpSessionAdapter {
898    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
899        formatter
900            .debug_struct("LocalAcpSessionAdapter")
901            .field("agent", &self.agent)
902            .finish_non_exhaustive()
903    }
904}
905
906#[cfg(test)]
907#[path = "acp_tests.rs"]
908mod acp_tests;
relay_knowledge/interfaces/agent/acp.rs

relay_knowledge/interfaces/agent/
acp.rs
