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        self.qos.admit_queued_request(&policy).map_err(qos_error)
353    }
354
355    fn record_audit(&self, input: AcpAuditInput<'_>) {
356        let event = AgentAuditEvent {
357            sequence: 0,
358            protocol: AgentProtocolKind::Acp,
359            operation: input.operation.to_owned(),
360            request_id: input.request_id.to_owned(),
361            trace_id: format!("trace-acp-{}", input.request_id),
362            runtime_identity: input
363                .session
364                .identity(input.session_id, Some(input.request_id.to_owned())),
365            qos_decision: input.qos_decision,
366            status: input.status,
367            source_scope: input.source_scope.map(str::to_owned),
368            freshness: input.freshness.map(str::to_owned),
369            limit: input.limit,
370            result_count: input.result_count,
371            truncated: input.truncated,
372            elapsed_ms: input.elapsed_ms,
373            error_kind: input.error_kind.map(str::to_owned),
374        };
375        self.audit.record(event.clone());
376        if input.qos_decision == AgentAuditQosDecision::Rejected {
377            self.metrics
378                .record_rejection("acp", input.error_kind.unwrap_or("qos_rejected"));
379            return;
380        }
381        let status_label = match event.status {
382            AgentAuditStatus::Completed => "completed",
383            AgentAuditStatus::Failed => "failed",
384            AgentAuditStatus::Cancelled => "cancelled",
385        };
386        self.metrics.record_request(
387            "acp",
388            input.operation,
389            status_label,
390            input.elapsed_ms,
391            input.truncated,
392        );
393        if event.status == AgentAuditStatus::Cancelled {
394            self.metrics.record_cancelled("acp");
395        }
396    }
397}
398
399struct AcpAuditInput<'a> {
400    operation: &'a str,
401    request_id: &'a str,
402    session_id: &'a str,
403    session: &'a AcpSessionRecord,
404    qos_decision: AgentAuditQosDecision,
405    status: AgentAuditStatus,
406    source_scope: Option<&'a str>,
407    freshness: Option<&'a str>,
408    limit: Option<usize>,
409    result_count: Option<usize>,
410    truncated: bool,
411    elapsed_ms: u64,
412    error_kind: Option<&'a str>,
413}
414
415/// ACP initialize response with relay-knowledge capability metadata.
416#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
417pub struct AcpInitializeResponse {
418    #[serde(rename = "_meta")]
419    pub meta: AcpInitializeMeta,
420}
421
422#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
423pub struct AcpInitializeMeta {
424    #[serde(rename = "relayKnowledge")]
425    pub relay_knowledge: AcpRelayKnowledgeCapability,
426}
427
428#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
429pub struct AcpRelayKnowledgeCapability {
430    #[serde(rename = "graphRetrieval")]
431    pub graph_retrieval: bool,
432    #[serde(rename = "readOnly")]
433    pub read_only: bool,
434    #[serde(rename = "supportsCancellation")]
435    pub supports_cancellation: bool,
436    #[serde(rename = "supportsIndexRefreshPermission")]
437    pub supports_index_refresh_permission: bool,
438}
439
440/// Local ACP session request.
441#[derive(Debug, Clone, Default, PartialEq, Eq, Serialize, Deserialize)]
442pub struct AcpSessionRequest {
443    #[serde(skip_serializing_if = "Option::is_none")]
444    pub client_name: Option<String>,
445    #[serde(skip_serializing_if = "Option::is_none")]
446    pub client_version: Option<String>,
447    #[serde(skip_serializing_if = "Option::is_none")]
448    pub actor_id: Option<String>,
449}
450
451/// Created ACP session metadata.
452#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
453pub struct AcpSession {
454    pub session_id: String,
455    pub runtime_identity: RuntimeIdentity,
456    pub policy_id: String,
457    pub authorized_scope_count: usize,
458}
459
460/// ACP prompt request with structured relay metadata.
461#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
462pub struct AcpPromptRequest {
463    pub prompt: String,
464    #[serde(skip_serializing_if = "Option::is_none")]
465    pub request_id: Option<String>,
466    #[serde(rename = "_meta", skip_serializing_if = "Option::is_none")]
467    pub meta: Option<AcpPromptMeta>,
468}
469
470#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
471pub struct AcpPromptMeta {
472    #[serde(rename = "relayKnowledge", skip_serializing_if = "Option::is_none")]
473    pub relay_knowledge: Option<AcpRelayKnowledgePrompt>,
474}
475
476#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
477pub struct AcpRelayKnowledgePrompt {
478    #[serde(skip_serializing_if = "Option::is_none")]
479    pub query: Option<String>,
480    #[serde(skip_serializing_if = "Option::is_none")]
481    pub source_scope: Option<String>,
482    #[serde(skip_serializing_if = "Option::is_none")]
483    pub limit: Option<usize>,
484    #[serde(skip_serializing_if = "Option::is_none")]
485    pub freshness: Option<String>,
486}
487
488/// ACP prompt response containing bounded progress and an optional context artifact.
489#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
490pub struct AcpPromptResponse {
491    pub session_id: String,
492    pub request_id: String,
493    pub updates: Vec<AcpSessionUpdate>,
494    #[serde(skip_serializing_if = "Option::is_none")]
495    pub context_artifact: Option<AcpContextArtifact>,
496    pub stop_reason: AcpStopReason,
497    #[serde(skip_serializing_if = "Option::is_none")]
498    pub error: Option<AcpErrorPayload>,
499}
500
501#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
502pub struct AcpContextArtifact {
503    pub artifact_id: String,
504    pub result: AgentRetrievalResult,
505}
506
507#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
508#[serde(rename_all = "snake_case")]
509pub enum AcpStopReason {
510    Completed,
511    Failed,
512    Cancelled,
513}
514
515#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
516pub struct AcpErrorPayload {
517    pub error_kind: String,
518    pub message: String,
519}
520
521#[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
522pub struct AcpSessionUpdate {
523    pub request_id: String,
524    pub kind: AcpSessionUpdateKind,
525    pub status: AcpSessionUpdateStatus,
526    pub message: String,
527    #[serde(rename = "_meta", skip_serializing_if = "Option::is_none")]
528    pub meta: Option<Value>,
529}
530
531impl AcpSessionUpdate {
532    fn pending(request_id: &str, message: &str) -> Self {
533        Self::new(
534            request_id,
535            AcpSessionUpdateKind::ToolCallUpdate,
536            AcpSessionUpdateStatus::Pending,
537            message,
538            None,
539        )
540    }
541
542    fn in_progress(request_id: &str, message: &str) -> Self {
543        Self::new(
544            request_id,
545            AcpSessionUpdateKind::ToolCallUpdate,
546            AcpSessionUpdateStatus::InProgress,
547            message,
548            None,
549        )
550    }
551
552    fn meta(request_id: &str, message: &str, meta: Value) -> Self {
553        Self::new(
554            request_id,
555            AcpSessionUpdateKind::SessionUpdate,
556            AcpSessionUpdateStatus::InProgress,
557            message,
558            Some(meta),
559        )
560    }
561
562    fn completed(request_id: &str, message: &str) -> Self {
563        Self::new(
564            request_id,
565            AcpSessionUpdateKind::ToolCallUpdate,
566            AcpSessionUpdateStatus::Completed,
567            message,
568            None,
569        )
570    }
571
572    fn failed(request_id: &str, message: &str, status: AcpSessionUpdateStatus) -> Self {
573        Self::new(
574            request_id,
575            AcpSessionUpdateKind::ToolCallUpdate,
576            status,
577            message,
578            None,
579        )
580    }
581
582    fn new(
583        request_id: &str,
584        kind: AcpSessionUpdateKind,
585        status: AcpSessionUpdateStatus,
586        message: &str,
587        meta: Option<Value>,
588    ) -> Self {
589        Self {
590            request_id: request_id.to_owned(),
591            kind,
592            status,
593            message: message.to_owned(),
594            meta,
595        }
596    }
597}
598
599#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
600#[serde(rename_all = "snake_case")]
601pub enum AcpSessionUpdateKind {
602    SessionUpdate,
603    ToolCallUpdate,
604}
605
606#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
607#[serde(rename_all = "snake_case")]
608pub enum AcpSessionUpdateStatus {
609    Pending,
610    InProgress,
611    Completed,
612    Failed,
613    Cancelled,
614}
615
616#[derive(Clone, Default)]
617struct AcpSessionRegistry {
618    inner: Arc<Mutex<AcpSessionState>>,
619}
620
621#[derive(Default)]
622struct AcpSessionState {
623    sessions: HashMap<String, AcpSessionRecord>,
624    active_requests: HashMap<String, watch::Sender<bool>>,
625}
626
627#[derive(Debug, Clone)]
628struct AcpSessionRecord {
629    client_name: Option<String>,
630    client_version: Option<String>,
631    actor_id: Option<String>,
632}
633
634impl AcpSessionRecord {
635    fn identity(&self, session_id: &str, request_id: Option<String>) -> RuntimeIdentity {
636        RuntimeIdentity::acp(
637            self.client_name.clone(),
638            self.client_version.clone(),
639            self.actor_id.clone(),
640            session_id.to_owned(),
641            request_id,
642        )
643    }
644}
645
646struct ActiveAcpRequest {
647    registry: AcpSessionRegistry,
648    key: String,
649    released: bool,
650}
651
652impl ActiveAcpRequest {
653    fn release(mut self) {
654        self.registry.remove_request(&self.key);
655        self.released = true;
656    }
657}
658
659impl Drop for ActiveAcpRequest {
660    fn drop(&mut self) {
661        if !self.released {
662            self.registry.remove_request(&self.key);
663        }
664    }
665}
666
667impl AcpSessionRegistry {
668    fn insert_session(&self, session_id: String, record: AcpSessionRecord) {
669        self.inner
670            .lock()
671            .unwrap_or_else(|poisoned| poisoned.into_inner())
672            .sessions
673            .insert(session_id, record);
674    }
675
676    fn session(&self, session_id: &str) -> Option<AcpSessionRecord> {
677        self.inner
678            .lock()
679            .unwrap_or_else(|poisoned| poisoned.into_inner())
680            .sessions
681            .get(session_id)
682            .cloned()
683    }
684
685    fn register_request(
686        &self,
687        session_id: &str,
688        request_id: String,
689    ) -> (watch::Receiver<bool>, ActiveAcpRequest) {
690        let (sender, receiver) = watch::channel(false);
691        let key = active_request_key(session_id, &request_id);
692        self.inner
693            .lock()
694            .unwrap_or_else(|poisoned| poisoned.into_inner())
695            .active_requests
696            .insert(key.clone(), sender);
697
698        (
699            receiver,
700            ActiveAcpRequest {
701                registry: self.clone(),
702                key,
703                released: false,
704            },
705        )
706    }
707
708    fn cancel_request(&self, session_id: &str, request_id: &str) -> bool {
709        let key = active_request_key(session_id, request_id);
710        self.inner
711            .lock()
712            .unwrap_or_else(|poisoned| poisoned.into_inner())
713            .active_requests
714            .get(&key)
715            .is_some_and(|sender| sender.send(true).is_ok())
716    }
717
718    fn remove_request(&self, key: &str) {
719        self.inner
720            .lock()
721            .unwrap_or_else(|poisoned| poisoned.into_inner())
722            .active_requests
723            .remove(key);
724    }
725}
726
727#[derive(Debug, Clone, PartialEq, Eq)]
728struct MappedPromptRequest {
729    query: String,
730    source_scope: Option<String>,
731    limit: usize,
732    freshness: FreshnessPolicy,
733}
734
735impl MappedPromptRequest {
736    fn into_retrieval_request(self) -> HybridRetrievalRequest {
737        HybridRetrievalRequest {
738            query: self.query,
739            source_scope: self.source_scope,
740            limit: self.limit,
741            freshness: self.freshness,
742        }
743    }
744}
745
746fn map_prompt_request(
747    agent: &AgentRuntimeConfig,
748    request: AcpPromptRequest,
749) -> Result<MappedPromptRequest, AgentAdapterError> {
750    let relay = request
751        .meta
752        .and_then(|meta| meta.relay_knowledge)
753        .unwrap_or(AcpRelayKnowledgePrompt {
754            query: None,
755            source_scope: None,
756            limit: None,
757            freshness: None,
758        });
759    let query = relay.query.unwrap_or(request.prompt);
760    let source_scope = authorize_scope(relay.source_scope, &agent.access_policy)?;
761    let limit = authorize_limit(relay.limit, &agent.access_policy)?;
762    let freshness = parse_freshness(relay.freshness.as_deref())?;
763
764    if query.trim().is_empty() {
765        return Err(AgentAdapterError::new(
766            AgentAdapterErrorKind::InvalidArgument,
767            "ACP prompt query must not be empty",
768        ));
769    }
770
771    Ok(MappedPromptRequest {
772        query,
773        source_scope,
774        limit,
775        freshness,
776    })
777}
778
779fn parse_freshness(value: Option<&str>) -> Result<FreshnessPolicy, AgentAdapterError> {
780    match value.unwrap_or("allow-stale") {
781        "allow-stale" => Ok(FreshnessPolicy::AllowStale),
782        "wait-until-fresh" => Ok(FreshnessPolicy::WaitUntilFresh),
783        "graph-only" => Ok(FreshnessPolicy::GraphOnly),
784        other => Err(AgentAdapterError::new(
785            AgentAdapterErrorKind::InvalidArgument,
786            format!("invalid freshness '{other}'"),
787        )),
788    }
789}
790
791async fn wait_for_cancellation(cancellation: &mut watch::Receiver<bool>) {
792    while cancellation.changed().await.is_ok() {
793        if *cancellation.borrow() {
794            return;
795        }
796    }
797
798    std::future::pending::<()>().await;
799}
800
801fn failed_prompt(
802    session_id: &str,
803    request_id: String,
804    mut updates: Vec<AcpSessionUpdate>,
805    error: AgentAdapterError,
806    _elapsed_ms: u64,
807) -> AcpPromptResponse {
808    let stop_reason = if error.kind == AgentAdapterErrorKind::Cancelled {
809        AcpStopReason::Cancelled
810    } else {
811        AcpStopReason::Failed
812    };
813    let status = if error.kind == AgentAdapterErrorKind::Cancelled {
814        AcpSessionUpdateStatus::Cancelled
815    } else {
816        AcpSessionUpdateStatus::Failed
817    };
818    updates.push(AcpSessionUpdate::failed(
819        &request_id,
820        &error.message,
821        status,
822    ));
823
824    AcpPromptResponse {
825        session_id: session_id.to_owned(),
826        request_id,
827        updates,
828        context_artifact: None,
829        stop_reason,
830        error: Some(AcpErrorPayload {
831            error_kind: error.kind.as_str().to_owned(),
832            message: error.message,
833        }),
834    }
835}
836
837fn qos_error(reason: RejectReason) -> AgentAdapterError {
838    let message = match reason {
839        RejectReason::ConnectionBudgetExceeded => "connection budget exhausted",
840        RejectReason::RequestBudgetExceeded => "request budget exhausted",
841        RejectReason::QueueBudgetExceeded => "queue budget exhausted",
842    };
843
844    AgentAdapterError::new(AgentAdapterErrorKind::QosRejected, message)
845}
846
847fn api_error_kind(kind: ErrorKind) -> AgentAdapterErrorKind {
848    match kind {
849        ErrorKind::InvalidArgument => AgentAdapterErrorKind::InvalidArgument,
850        ErrorKind::StorageUnavailable => AgentAdapterErrorKind::StorageUnavailable,
851        ErrorKind::Timeout => AgentAdapterErrorKind::Timeout,
852        ErrorKind::Internal => AgentAdapterErrorKind::Internal,
853    }
854}
855
856fn active_request_key(session_id: &str, request_id: &str) -> String {
857    format!("{session_id}|{request_id}")
858}
859
860fn normalized_optional(value: Option<String>) -> Option<String> {
861    value.and_then(|value| {
862        let trimmed = value.trim();
863        (!trimmed.is_empty()).then(|| trimmed.to_owned())
864    })
865}
866
867fn generate_acp_id(prefix: &str) -> Result<String, AgentAdapterError> {
868    let mut entropy = [0_u8; 16];
869    getrandom::getrandom(&mut entropy).map_err(|_| {
870        AgentAdapterError::new(
871            AgentAdapterErrorKind::Internal,
872            "OS session entropy is unavailable",
873        )
874    })?;
875
876    Ok(format!("{prefix}-{}", lowercase_hex(&entropy)))
877}
878
879fn lowercase_hex(bytes: &[u8]) -> String {
880    const HEX: &[u8; 16] = b"0123456789abcdef";
881    let mut output = String::with_capacity(bytes.len() * 2);
882    for byte in bytes {
883        output.push(HEX[usize::from(byte >> 4)] as char);
884        output.push(HEX[usize::from(byte & 0x0f)] as char);
885    }
886
887    output
888}
889
890fn elapsed_millis(started: Instant) -> u64 {
891    u64::try_from(started.elapsed().as_millis()).unwrap_or(u64::MAX)
892}
893
894impl fmt::Debug for LocalAcpSessionAdapter {
895    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
896        formatter
897            .debug_struct("LocalAcpSessionAdapter")
898            .field("agent", &self.agent)
899            .finish_non_exhaustive()
900    }
901}
902
903#[cfg(test)]
904#[path = "acp_tests.rs"]
905mod acp_tests;
relay_knowledge/interfaces/agent/acp.rs

relay_knowledge/interfaces/agent/
acp.rs
