relay_knowledge/application/
code_repository_set_service.rs

1use crate::{
2    api::{
3        ApiError, ApiMetadata, CodeRepositorySetAddResponse, CodeRepositorySetCreateResponse,
4        CodeRepositorySetQueryResponse, CodeRepositorySetRefreshResponse,
5        CodeRepositorySetStatusResponse, RequestContext,
6    },
7    code::{CodeIndexError, resolve_repository_snapshot},
8    domain::{
9        CodeRepositorySelector, CodeRepositorySetAddMemberRequest, CodeRepositorySetCreateRequest,
10        CodeRepositorySetMemberStatus, CodeRepositorySetQueryHit, CodeRepositorySetQueryRequest,
11        CodeRepositorySetStatus, CodeRepositoryStatus, CodeRetrievalRequest, FreshnessPolicy,
12    },
13    storage::{
14        CodeRepositorySetMemberSeed, CodeRepositorySetRefreshTaskClaimRequest,
15        CodeRepositorySetRefreshTaskCompletion, CodeRepositorySetRefreshTaskFailure,
16        CodeRepositorySetRefreshTaskSeed, CodeRepositorySetSeed, StorageError,
17    },
18};
19use std::path::PathBuf;
20
21use super::{
22    RelayKnowledgeService,
23    code_repository_set_plan::{
24        dependency_symbol_plan_needs_hybrid_fallback, repository_set_member_query_plan,
25    },
26    code_repository_set_query::{
27        OverlayEvidenceIndex, apply_bridge_support_bonus, dedupe_sort_truncate,
28        per_member_candidate_limit, prune_returned_overlay_evidence, repository_set_score,
29    },
30    code_service::apply_code_grep_fallback,
31};
32
33const REPOSITORY_SET_REFRESH_TASK_LEASE_MS: u64 = 10 * 60 * 1000;
34const REPOSITORY_SET_REFRESH_TASK_MAX_ATTEMPTS: u32 = 3;
35const REPOSITORY_SET_REFRESH_TASK_RETRY_BACKOFF_MS: u64 = 60_000;
36
37impl RelayKnowledgeService {
38    /// Creates or updates a thin repository set.
39    pub async fn create_code_repository_set(
40        &self,
41        request: CodeRepositorySetCreateRequest,
42        context: RequestContext,
43    ) -> Result<CodeRepositorySetCreateResponse, ApiError> {
44        let store = self.store().await.map_err(storage_api_error)?;
45        let repository_set = store
46            .create_code_repository_set(CodeRepositorySetSeed {
47                alias: request.alias.clone(),
48                description: request.description.clone(),
49                default_ref_policy_json: request.default_ref_policy_json.clone(),
50                now_ms: now_millis(),
51            })
52            .await
53            .map_err(storage_api_error)?;
54        let graph_version = store
55            .current_graph_version()
56            .await
57            .map_err(storage_api_error)?;
58
59        Ok(CodeRepositorySetCreateResponse {
60            metadata: ApiMetadata::graph_only(&context, graph_version),
61            request,
62            repository_set,
63        })
64    }
65
66    /// Adds one already-indexed repository snapshot to a repository set.
67    pub async fn add_code_repository_set_member(
68        &self,
69        request: CodeRepositorySetAddMemberRequest,
70        context: RequestContext,
71    ) -> Result<CodeRepositorySetAddResponse, ApiError> {
72        let store = self.store().await.map_err(storage_api_error)?;
73        let repository = store
74            .code_repository_status(request.repository_alias.clone())
75            .await
76            .map_err(storage_api_error)?
77            .ok_or_else(|| {
78                ApiError::invalid_argument(format!(
79                    "code repository '{}' is not registered",
80                    request.repository_alias
81                ))
82            })?;
83        let root_path = repository.root_path.clone();
84        let ref_selector = request.ref_selector.clone();
85        let (resolved_commit_sha, _tree_hash) =
86            run_blocking_code(move || resolve_repository_snapshot(root_path, &ref_selector))
87                .await?;
88        let path_filters = merged_filters(&repository.path_filters, &request.path_filters);
89        let language_filters =
90            merged_filters(&repository.language_filters, &request.language_filters);
91        let scope = store
92            .code_repository_scope_status(
93                request.repository_alias.clone(),
94                resolved_commit_sha.clone(),
95                path_filters.clone(),
96                language_filters.clone(),
97            )
98            .await
99            .map_err(storage_api_error)?
100            .ok_or_else(|| {
101                ApiError::invalid_argument(format!(
102                    "code repository '{}' has no indexed scope for ref {} and requested filters",
103                    request.repository_alias, request.ref_selector
104                ))
105            })?;
106        let source_scope = scope.last_indexed_scope_id.clone().ok_or_else(|| {
107            ApiError::invalid_argument(format!(
108                "code repository '{}' matching scope has no source scope",
109                request.repository_alias
110            ))
111        })?;
112        let member = store
113            .add_code_repository_set_member(CodeRepositorySetMemberSeed {
114                set_alias: request.set_alias.clone(),
115                repository_id: repository.repository_id,
116                repository_alias: request.repository_alias.clone(),
117                ref_selector: request.ref_selector.clone(),
118                resolved_commit_sha,
119                source_scope,
120                path_filters,
121                language_filters,
122                priority: request.priority,
123            })
124            .await
125            .map_err(storage_api_error)?;
126        let status = required_set_status(&store, &request.set_alias).await?;
127        let graph_version = store
128            .current_graph_version()
129            .await
130            .map_err(storage_api_error)?;
131
132        Ok(CodeRepositorySetAddResponse {
133            metadata: ApiMetadata::graph_only(&context, graph_version),
134            request,
135            member,
136            status,
137        })
138    }
139
140    /// Queries every member scope and merges ranked candidates without changing single-repo search.
141    pub async fn query_code_repository_set(
142        &self,
143        request: CodeRepositorySetQueryRequest,
144        context: RequestContext,
145    ) -> Result<CodeRepositorySetQueryResponse, ApiError> {
146        let store = self.store().await.map_err(storage_api_error)?;
147        let status = required_set_status(&store, &request.set_alias).await?;
148        let graph_version = store
149            .current_graph_version()
150            .await
151            .map_err(storage_api_error)?;
152        if request.freshness_policy == FreshnessPolicy::GraphOnly {
153            return Ok(CodeRepositorySetQueryResponse {
154                metadata: ApiMetadata::graph_only(&context, graph_version),
155                request,
156                status,
157                results: Vec::new(),
158                truncated: false,
159                degraded_reason: Some("graph_only freshness policy selected".to_owned()),
160            });
161        }
162        if let Some(error) = unfresh_set_error_for_wait_policy(&request, &status) {
163            return Err(error);
164        }
165        let edges = store
166            .code_repository_set_cross_edges(status.repository_set.set_id.clone())
167            .await
168            .map_err(storage_api_error)?;
169        let edge_index = OverlayEvidenceIndex::new(&edges);
170        let mut results = Vec::new();
171        let mut fallback_degraded_reason = None;
172        let candidate_limit = per_member_candidate_limit(request.limit, status.members.len());
173        let highest_priority = status
174            .members
175            .iter()
176            .map(|member| member.member.priority)
177            .max()
178            .unwrap_or(0);
179        for member_status in &status.members {
180            let member = &member_status.member;
181            let selector = CodeRepositorySelector::new(
182                member.repository_alias.clone(),
183                member.resolved_commit_sha.clone(),
184                request.path_filters.clone(),
185                request.language_filters.clone(),
186            )
187            .map_err(|error| ApiError::invalid_argument(error.to_string()))?;
188            let member_query_plan =
189                repository_set_member_query_plan(&request, member_status, highest_priority);
190            let search_request = CodeRetrievalRequest::new(
191                member_query_plan.query,
192                selector.clone(),
193                member_query_plan.kind,
194                candidate_limit,
195                FreshnessPolicy::AllowStale,
196            )
197            .map_err(|error| ApiError::invalid_argument(error.to_string()))?;
198            let mut active_request = search_request.clone();
199            let mut hits = store
200                .search_code_scope(member.source_scope.clone(), search_request)
201                .await
202                .map_err(storage_api_error)?;
203            if dependency_symbol_plan_needs_hybrid_fallback(&request, member_query_plan.kind, &hits)
204            {
205                let fallback_request = CodeRetrievalRequest::new(
206                    request.query.clone(),
207                    selector,
208                    request.code_query_kind,
209                    candidate_limit,
210                    FreshnessPolicy::AllowStale,
211                )
212                .map_err(|error| ApiError::invalid_argument(error.to_string()))?;
213                active_request = fallback_request.clone();
214                hits = store
215                    .search_code_scope(member.source_scope.clone(), fallback_request)
216                    .await
217                    .map_err(storage_api_error)?;
218            }
219            let base_status = required_member_repository(&store, &member.repository_id).await?;
220            let scoped_member_status =
221                code_status_for_repository_set_member(&base_status, member_status);
222            fallback_degraded_reason = fallback_degraded_reason.or(apply_code_grep_fallback(
223                &store,
224                &base_status,
225                &scoped_member_status,
226                &active_request,
227                &mut hits,
228            )
229            .await?);
230            for hit in hits {
231                let overlay_evidence = edge_index.evidence_for_hit(&hit);
232                let score = repository_set_score(&hit, member_status, &overlay_evidence);
233                results.push(CodeRepositorySetQueryHit {
234                    member: member.clone(),
235                    hit,
236                    overlay_evidence,
237                    score,
238                });
239            }
240        }
241        apply_bridge_support_bonus(&mut results);
242        let truncated = dedupe_sort_truncate(&mut results, request.limit, &request.query);
243        prune_returned_overlay_evidence(&mut results);
244        let degraded_reason = join_degraded_reasons([
245            status.degraded_reason.clone(),
246            status
247                .overlay
248                .stale
249                .then(|| "repository set overlay is stale".to_owned()),
250            fallback_degraded_reason,
251        ]);
252
253        Ok(CodeRepositorySetQueryResponse {
254            metadata: ApiMetadata::graph_only(&context, graph_version),
255            request,
256            status,
257            results,
258            truncated,
259            degraded_reason,
260        })
261    }
262
263    /// Returns repository-set freshness and member diagnostics.
264    pub async fn code_repository_set_status(
265        &self,
266        set_alias: String,
267        context: RequestContext,
268    ) -> Result<CodeRepositorySetStatusResponse, ApiError> {
269        let store = self.store().await.map_err(storage_api_error)?;
270        let status = required_set_status(&store, &set_alias).await?;
271        let graph_version = store
272            .current_graph_version()
273            .await
274            .map_err(storage_api_error)?;
275
276        Ok(CodeRepositorySetStatusResponse {
277            metadata: ApiMetadata::graph_only(&context, graph_version),
278            status,
279        })
280    }
281
282    /// Rebuilds cross-repository import/module overlay edges.
283    pub async fn refresh_code_repository_set(
284        &self,
285        set_alias: String,
286        context: RequestContext,
287    ) -> Result<CodeRepositorySetRefreshResponse, ApiError> {
288        let store = self.store().await.map_err(storage_api_error)?;
289        let summary = store
290            .refresh_code_repository_set_overlay(set_alias.clone(), now_millis())
291            .await
292            .map_err(storage_api_error)?;
293        let status = required_set_status(&store, &set_alias).await?;
294        let graph_version = store
295            .current_graph_version()
296            .await
297            .map_err(storage_api_error)?;
298
299        Ok(CodeRepositorySetRefreshResponse {
300            metadata: ApiMetadata::graph_only(&context, graph_version),
301            status,
302            summary: Some(summary),
303            task: None,
304        })
305    }
306
307    /// Queues a repository-set overlay refresh task.
308    pub async fn start_code_repository_set_refresh(
309        &self,
310        set_alias: String,
311        context: RequestContext,
312    ) -> Result<CodeRepositorySetRefreshResponse, ApiError> {
313        let store = self.store().await.map_err(storage_api_error)?;
314        let status = required_set_status(&store, &set_alias).await?;
315        let fingerprint = repository_set_refresh_fingerprint(&status);
316        let task = store
317            .queue_code_repository_set_refresh_task(CodeRepositorySetRefreshTaskSeed {
318                set_id: status.repository_set.set_id.clone(),
319                set_alias: status.repository_set.alias.clone(),
320                input_fingerprint: fingerprint,
321                now_ms: now_millis(),
322            })
323            .await
324            .map_err(storage_api_error)?;
325        let graph_version = store
326            .current_graph_version()
327            .await
328            .map_err(storage_api_error)?;
329
330        Ok(CodeRepositorySetRefreshResponse {
331            metadata: ApiMetadata::graph_only(&context, graph_version),
332            status,
333            summary: None,
334            task: Some(task),
335        })
336    }
337
338    /// Runs one queued repository-set overlay refresh task under a lease.
339    pub async fn run_code_repository_set_refresh_task_once(
340        &self,
341        task_id: Option<String>,
342        context: RequestContext,
343    ) -> Result<Option<crate::domain::CodeRepositorySetRefreshTaskRecord>, ApiError> {
344        let store = self.store().await.map_err(storage_api_error)?;
345        let lease_owner = format!("code-repository-set-refresh-worker-{}", std::process::id());
346        let Some(task) = store
347            .claim_code_repository_set_refresh_task(CodeRepositorySetRefreshTaskClaimRequest {
348                task_id,
349                lease_owner: lease_owner.clone(),
350                lease_duration_ms: REPOSITORY_SET_REFRESH_TASK_LEASE_MS,
351                max_attempts: REPOSITORY_SET_REFRESH_TASK_MAX_ATTEMPTS,
352                now_ms: now_millis(),
353            })
354            .await
355            .map_err(storage_api_error)?
356        else {
357            return Ok(None);
358        };
359        let result = self
360            .refresh_code_repository_set(task.set_alias.clone(), context)
361            .await;
362        match result {
363            Ok(_) => store
364                .complete_code_repository_set_refresh_task(CodeRepositorySetRefreshTaskCompletion {
365                    task_id: task.task_id,
366                    lease_owner,
367                    attempt_count: task.attempt_count,
368                    now_ms: now_millis(),
369                })
370                .await
371                .map(Some)
372                .map_err(storage_api_error),
373            Err(error) => {
374                let _ = store
375                    .fail_code_repository_set_refresh_task(CodeRepositorySetRefreshTaskFailure {
376                        task_id: task.task_id,
377                        lease_owner,
378                        attempt_count: task.attempt_count,
379                        error_kind: "repository_set_overlay".to_owned(),
380                        error_message: error.message.clone(),
381                        retry_backoff_ms: REPOSITORY_SET_REFRESH_TASK_RETRY_BACKOFF_MS,
382                        max_attempts: REPOSITORY_SET_REFRESH_TASK_MAX_ATTEMPTS,
383                        now_ms: now_millis(),
384                    })
385                    .await;
386                Err(error)
387            }
388        }
389    }
390
391    pub(crate) async fn code_repository_set_member_scopes(
392        &self,
393        set_alias: String,
394    ) -> Result<Option<Vec<(String, String)>>, ApiError> {
395        let store = self.store().await.map_err(storage_api_error)?;
396        store
397            .code_repository_set_status(set_alias)
398            .await
399            .map(|status| {
400                status.map(|status| {
401                    status
402                        .members
403                        .into_iter()
404                        .map(|member| (member.member.repository_alias, member.member.source_scope))
405                        .collect()
406                })
407            })
408            .map_err(storage_api_error)
409    }
410}
411
412pub(super) async fn required_set_status(
413    store: &std::sync::Arc<dyn crate::storage::KnowledgeStore>,
414    set_alias: &str,
415) -> Result<CodeRepositorySetStatus, ApiError> {
416    let mut status = store
417        .code_repository_set_status(set_alias.to_owned())
418        .await
419        .map_err(storage_api_error)?
420        .ok_or_else(|| {
421            ApiError::invalid_argument(format!(
422                "code repository set '{set_alias}' is not registered"
423            ))
424        })?;
425    refresh_moving_member_freshness(store, &mut status).await?;
426    refresh_repository_set_freshness(&mut status);
427
428    Ok(status)
429}
430
431fn join_degraded_reasons(reasons: impl IntoIterator<Item = Option<String>>) -> Option<String> {
432    let mut joined = Vec::new();
433    for reason in reasons.into_iter().flatten() {
434        if !joined.contains(&reason) {
435            joined.push(reason);
436        }
437    }
438
439    (!joined.is_empty()).then(|| joined.join("; "))
440}
441
442async fn refresh_moving_member_freshness(
443    store: &std::sync::Arc<dyn crate::storage::KnowledgeStore>,
444    status: &mut CodeRepositorySetStatus,
445) -> Result<(), ApiError> {
446    for index in 0..status.members.len() {
447        let member = status.members[index].member.clone();
448        let Some(reason) = moving_member_stale_reason(store, &member).await? else {
449            continue;
450        };
451        status.members[index].stale = true;
452        status.members[index].freshness_state = "stale".to_owned();
453        status.members[index].degraded_reason = Some(reason);
454    }
455
456    Ok(())
457}
458
459async fn moving_member_stale_reason(
460    store: &std::sync::Arc<dyn crate::storage::KnowledgeStore>,
461    member: &crate::domain::CodeRepositorySetMember,
462) -> Result<Option<String>, ApiError> {
463    if !member_ref_tracks_repository(&member.ref_selector, &member.resolved_commit_sha) {
464        return Ok(None);
465    }
466    let repository = store
467        .code_repository_status(member.repository_id.clone())
468        .await
469        .map_err(storage_api_error)?
470        .ok_or_else(|| {
471            ApiError::invalid_argument(format!(
472                "code repository '{}' is not registered",
473                member.repository_alias
474            ))
475        })?;
476    let root_path = PathBuf::from(repository.root_path);
477    let ref_selector = member.ref_selector.clone();
478    let resolved =
479        tokio::task::spawn_blocking(move || resolve_repository_snapshot(root_path, &ref_selector))
480            .await
481            .map_err(|error| ApiError::storage_unavailable(error.to_string()))?;
482
483    match resolved {
484        Ok((current_commit, _)) if current_commit == member.resolved_commit_sha => Ok(None),
485        Ok((current_commit, _)) => Ok(Some(format!(
486            "repository set member '{}' ref '{}' now resolves to {}, not stored snapshot {}",
487            member.repository_alias,
488            member.ref_selector,
489            current_commit,
490            member.resolved_commit_sha
491        ))),
492        Err(error) => Ok(Some(format!(
493            "repository set member '{}' ref '{}' could not be resolved: {error}",
494            member.repository_alias, member.ref_selector
495        ))),
496    }
497}
498
499fn member_ref_tracks_repository(ref_selector: &str, resolved_commit_sha: &str) -> bool {
500    let ref_selector = ref_selector.trim();
501    !(ref_selector == resolved_commit_sha
502        || (is_git_oid_prefix(ref_selector) && resolved_commit_sha.starts_with(ref_selector)))
503}
504
505fn is_git_oid_prefix(value: &str) -> bool {
506    (7..=64).contains(&value.len()) && value.bytes().all(|byte| byte.is_ascii_hexdigit())
507}
508
509fn refresh_repository_set_freshness(status: &mut CodeRepositorySetStatus) {
510    let member_stale = status.members.iter().any(|member| member.stale);
511    if member_stale && !status.overlay.stale {
512        status.overlay.stale = true;
513        status.overlay.state = "overlay_stale".to_owned();
514    }
515    status.freshness_state = if status.members.is_empty() {
516        "incomplete"
517    } else if member_stale {
518        "stale"
519    } else if status.overlay.stale {
520        "overlay_stale"
521    } else {
522        "fresh"
523    }
524    .to_owned();
525    status.degraded_reason = status
526        .members
527        .iter()
528        .find_map(|member| member.degraded_reason.clone())
529        .or_else(|| status.overlay.degraded_reason.clone());
530}
531
532fn unfresh_set_error_for_wait_policy(
533    request: &CodeRepositorySetQueryRequest,
534    status: &CodeRepositorySetStatus,
535) -> Option<ApiError> {
536    if request.freshness_policy != FreshnessPolicy::WaitUntilFresh {
537        return None;
538    }
539    if status.members.is_empty() {
540        return Some(ApiError::invalid_argument(format!(
541            "code repository set '{}' has no members",
542            status.repository_set.alias
543        )));
544    }
545    if let Some(member) = status.members.iter().find(|member| member.stale) {
546        return Some(ApiError::invalid_argument(format!(
547            "code repository set '{}' member '{}' scope '{}' is stale",
548            status.repository_set.alias, member.member.repository_alias, member.member.source_scope
549        )));
550    }
551    if status.overlay.stale {
552        return Some(ApiError::invalid_argument(format!(
553            "code repository set '{}' overlay is stale; run repo-set refresh before querying with wait_until_fresh",
554            status.repository_set.alias
555        )));
556    }
557
558    None
559}
560
561fn repository_set_refresh_fingerprint(status: &CodeRepositorySetStatus) -> String {
562    let mut parts = vec![status.repository_set.set_id.clone()];
563    parts.extend(status.members.iter().map(|member| {
564        format!(
565            "{}:{}:{}:{}:{}",
566            member.member.repository_id,
567            member.member.source_scope,
568            member.member.resolved_commit_sha,
569            member.tree_hash,
570            member.stale
571        )
572    }));
573    parts.join("|")
574}
575
576fn merged_filters(left: &[String], right: &[String]) -> Vec<String> {
577    let mut merged = Vec::new();
578    for value in left.iter().chain(right.iter()) {
579        if !merged.contains(value) {
580            merged.push(value.clone());
581        }
582    }
583
584    merged
585}
586
587async fn required_member_repository(
588    store: &std::sync::Arc<dyn crate::storage::KnowledgeStore>,
589    repository_id: &str,
590) -> Result<CodeRepositoryStatus, ApiError> {
591    store
592        .code_repository_status(repository_id.to_owned())
593        .await
594        .map_err(storage_api_error)?
595        .ok_or_else(|| {
596            ApiError::invalid_argument(format!(
597                "code repository set member repository '{repository_id}' is not registered"
598            ))
599        })
600}
601
602fn code_status_for_repository_set_member(
603    base_status: &CodeRepositoryStatus,
604    member_status: &CodeRepositorySetMemberStatus,
605) -> CodeRepositoryStatus {
606    let member = &member_status.member;
607    CodeRepositoryStatus {
608        repository_id: member.repository_id.clone(),
609        alias: member.repository_alias.clone(),
610        root_path: base_status.root_path.clone(),
611        path_filters: member.path_filters.clone(),
612        language_filters: member.language_filters.clone(),
613        last_indexed_scope_id: Some(member.source_scope.clone()),
614        last_indexed_commit: Some(member.resolved_commit_sha.clone()),
615        tree_hash: Some(member_status.tree_hash.clone()),
616        state: member_status.freshness_state.clone(),
617        indexed_file_count: member_status.indexed_file_count,
618        symbol_count: member_status.symbol_count,
619        reference_count: member_status.reference_count,
620        chunk_count: member_status.chunk_count,
621        stale: member_status.stale,
622        degraded_reason: member_status.degraded_reason.clone(),
623    }
624}
625
626fn now_millis() -> u64 {
627    std::time::SystemTime::now()
628        .duration_since(std::time::UNIX_EPOCH)
629        .map_or(0, |duration| {
630            u64::try_from(duration.as_millis()).unwrap_or(u64::MAX)
631        })
632}
633
634async fn run_blocking_code<T, F>(operation: F) -> Result<T, ApiError>
635where
636    T: Send + 'static,
637    F: FnOnce() -> Result<T, CodeIndexError> + Send + 'static,
638{
639    tokio::task::spawn_blocking(operation)
640        .await
641        .map_err(|error| ApiError::storage_unavailable(error.to_string()))?
642        .map_err(code_api_error)
643}
644
645fn code_api_error(error: CodeIndexError) -> ApiError {
646    match error {
647        CodeIndexError::InvalidInput(message) => ApiError::invalid_argument(message),
648        CodeIndexError::Git { .. } | CodeIndexError::Io(_) | CodeIndexError::TreeSitter(_) => {
649            ApiError::storage_unavailable(error.to_string())
650        }
651    }
652}
653
654pub(super) fn storage_api_error(error: StorageError) -> ApiError {
655    match error {
656        StorageError::InvalidInput(message) => ApiError::invalid_argument(message),
657        other => ApiError::storage_unavailable(other.to_string()),
658    }
659}
660
661#[cfg(test)]
662mod tests {
663    use super::*;
664    use crate::{
665        api::ErrorKind,
666        domain::CodeRepositorySetMemberStatus,
667        domain::{CodeRepositorySet, CodeRepositorySetMember, CodeRepositorySetOverlayStatus},
668        storage::SqliteGraphStore,
669    };
670    use std::sync::Arc;
671
672    #[test]
673    fn helper_policy_reports_wait_until_fresh_blockers() {
674        let request = CodeRepositorySetQueryRequest::new(
675            "workspace",
676            "serve",
677            crate::domain::CodeQueryKind::Definition,
678            5,
679            FreshnessPolicy::WaitUntilFresh,
680            Vec::new(),
681            Vec::new(),
682        )
683        .expect("request should validate");
684        let empty = status_with_members(Vec::new(), overlay(true));
685        assert!(
686            unfresh_set_error_for_wait_policy(&request, &empty)
687                .expect("empty set should block")
688                .message
689                .contains("has no members")
690        );
691
692        let mut stale_member = member_status("app", "scope-app", 0);
693        stale_member.stale = true;
694        let stale_status = status_with_members(vec![stale_member], overlay(false));
695        assert!(
696            unfresh_set_error_for_wait_policy(&request, &stale_status)
697                .expect("stale member should block")
698                .message
699                .contains("member 'app'")
700        );
701
702        let overlay_status =
703            status_with_members(vec![member_status("app", "scope-app", 0)], overlay(true));
704        assert!(
705            unfresh_set_error_for_wait_policy(&request, &overlay_status)
706                .expect("stale overlay should block")
707                .message
708                .contains("overlay is stale")
709        );
710
711        let allow_stale = CodeRepositorySetQueryRequest::new(
712            "workspace",
713            "serve",
714            crate::domain::CodeQueryKind::Definition,
715            5,
716            FreshnessPolicy::AllowStale,
717            Vec::new(),
718            Vec::new(),
719        )
720        .expect("request should validate");
721        assert!(unfresh_set_error_for_wait_policy(&allow_stale, &overlay_status).is_none());
722    }
723
724    #[test]
725    fn helper_fingerprint_and_error_mapping_are_stable() {
726        let status = status_with_members(
727            vec![
728                member_status("app", "scope-app", 1),
729                member_status("svc", "scope-svc", 0),
730            ],
731            overlay(false),
732        );
733        let fingerprint = repository_set_refresh_fingerprint(&status);
734        assert!(fingerprint.contains("set-workspace"));
735        assert!(fingerprint.contains("repo-app:scope-app:commit-scope-app:tree-scope-app:false"));
736        assert_eq!(
737            merged_filters(&["src".to_owned()], &["src".to_owned(), "tests".to_owned()]),
738            ["src".to_owned(), "tests".to_owned()]
739        );
740        assert_eq!(
741            code_api_error(CodeIndexError::InvalidInput("bad ref".to_owned())).error_kind,
742            ErrorKind::InvalidArgument
743        );
744        assert_eq!(
745            code_api_error(CodeIndexError::Io(std::io::Error::other("disk"))).error_kind,
746            ErrorKind::StorageUnavailable
747        );
748        assert_eq!(
749            storage_api_error(StorageError::InvalidInput("bad storage".to_owned())).error_kind,
750            ErrorKind::InvalidArgument
751        );
752    }
753
754    #[tokio::test]
755    async fn helper_required_status_reports_missing_sets() {
756        let store: Arc<dyn crate::storage::KnowledgeStore> =
757            Arc::new(SqliteGraphStore::open_in_memory().expect("store should open"));
758        let error = required_set_status(&store, "missing")
759            .await
760            .expect_err("missing set should fail");
761
762        assert_eq!(error.error_kind, ErrorKind::InvalidArgument);
763        assert!(error.message.contains("is not registered"));
764    }
765
766    fn status_with_members(
767        members: Vec<CodeRepositorySetMemberStatus>,
768        overlay: CodeRepositorySetOverlayStatus,
769    ) -> CodeRepositorySetStatus {
770        CodeRepositorySetStatus {
771            repository_set: CodeRepositorySet {
772                set_id: "set-workspace".to_owned(),
773                alias: "workspace".to_owned(),
774                description: None,
775                default_ref_policy_json: "{\"default_ref\":\"HEAD\"}".to_owned(),
776                created_at_ms: 1,
777                updated_at_ms: 1,
778            },
779            members,
780            overlay,
781            freshness_state: "fresh".to_owned(),
782            degraded_reason: None,
783        }
784    }
785
786    fn member_status(
787        repository_alias: &str,
788        source_scope: &str,
789        priority: i32,
790    ) -> CodeRepositorySetMemberStatus {
791        CodeRepositorySetMemberStatus {
792            member: CodeRepositorySetMember {
793                set_id: "set-workspace".to_owned(),
794                repository_id: format!("repo-{repository_alias}"),
795                repository_alias: repository_alias.to_owned(),
796                ref_selector: "HEAD".to_owned(),
797                resolved_commit_sha: format!("commit-{source_scope}"),
798                source_scope: source_scope.to_owned(),
799                path_filters: vec!["src".to_owned()],
800                language_filters: vec!["rust".to_owned()],
801                priority,
802            },
803            tree_hash: format!("tree-{source_scope}"),
804            freshness_state: "fresh".to_owned(),
805            stale: false,
806            indexed_file_count: 1,
807            symbol_count: 1,
808            reference_count: 0,
809            chunk_count: 1,
810            degraded_reason: None,
811        }
812    }
813
814    fn overlay(stale: bool) -> CodeRepositorySetOverlayStatus {
815        CodeRepositorySetOverlayStatus {
816            state: if stale { "overlay_stale" } else { "fresh" }.to_owned(),
817            stale,
818            edge_count: usize::from(!stale),
819            refreshed_at_ms: (!stale).then_some(10),
820            degraded_reason: None,
821        }
822    }
823}
relay_knowledge/application/code_repository_set_service.rs

relay_knowledge/application/
code_repository_set_service.rs
