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

relay_knowledge/interfaces/agent/
acp.rs
