kranz_server/
multi.rs

1//! Operator-owned repository catalog and the routing plane around per-repo
2//! [`MissionHost`](crate::MissionHost) instances.
3
4use crate::MissionHost;
5use anyhow::{anyhow, Context, Result};
6use kranz_engine::event_log::EventLog;
7use kranz_engine::git_ops::GitRepo;
8use kranz_engine::merged::merged_bit;
9use kranz_engine::paths::MissionPaths;
10use kranz_engine::reducer;
11use kranz_engine::ticket::{Ticket, TicketState};
12use kranz_engine::types::MissionStatus;
13use serde::{Deserialize, Serialize};
14use std::collections::{BTreeMap, HashSet};
15use std::path::{Component, Path, PathBuf};
16use std::sync::{Arc, Mutex};
17use std::time::Duration;
18use tokio::sync::Semaphore;
19
20fn default_max_concurrent_repos() -> usize {
21    1
22}
23
24/// The `host` object in the operator's global `~/.kranz/config.json`.
25#[derive(Clone, Debug, Deserialize, Serialize)]
26#[serde(rename_all = "camelCase", default)]
27pub struct HostConfig {
28    pub default_repo: Option<String>,
29    pub max_concurrent_repos: usize,
30    pub repos: Vec<RepoConfig>,
31}
32
33impl Default for HostConfig {
34    fn default() -> Self {
35        Self {
36            default_repo: None,
37            max_concurrent_repos: default_max_concurrent_repos(),
38            repos: Vec::new(),
39        }
40    }
41}
42
43/// One operator-owned repository catalog row.
44#[derive(Clone, Debug, Deserialize, Serialize)]
45#[serde(rename_all = "camelCase")]
46pub struct RepoConfig {
47    pub id: String,
48    pub root: PathBuf,
49    #[serde(default)]
50    pub display_name: Option<String>,
51    #[serde(default)]
52    pub group: Option<String>,
53    #[serde(default)]
54    pub pinned: bool,
55    #[serde(default)]
56    pub slack: RepoSlackConfig,
57}
58
59/// Per-repository Slack routing and authorization owned by global config.
60#[derive(Clone, Debug, Default, Deserialize, Serialize)]
61#[serde(rename_all = "camelCase", default)]
62pub struct RepoSlackConfig {
63    pub channels: Vec<SlackChannelRoute>,
64    pub allow_users: Vec<String>,
65}
66
67/// Exact Slack workspace/channel route.
68#[derive(Clone, Debug, Deserialize, Serialize, PartialEq, Eq, Hash)]
69pub struct SlackChannelRoute {
70    pub team: String,
71    pub channel: String,
72}
73
74#[derive(Default, Deserialize)]
75#[serde(default)]
76struct GlobalConfig {
77    host: HostConfig,
78}
79
80/// Load only the operator-owned `host` block from a global config file.
81/// Missing files are equivalent to an empty catalog; malformed files fail
82/// startup instead of silently selecting another repository.
83pub fn load_host_config(path: &Path) -> Result<HostConfig> {
84    let text = match std::fs::read_to_string(path) {
85        Ok(text) => text,
86        Err(error) if error.kind() == std::io::ErrorKind::NotFound => {
87            return Ok(HostConfig::default())
88        }
89        Err(error) => return Err(error).with_context(|| format!("cannot read {}", path.display())),
90    };
91    let global: GlobalConfig = serde_json::from_str(&text)
92        .with_context(|| format!("invalid JSON in {}", path.display()))?;
93    Ok(global.host)
94}
95
96/// A resolved catalog entry. The root is fixed at startup and requests only
97/// ever resolve this entry by its validated id.
98#[derive(Clone)]
99pub struct RepoContext {
100    config: RepoConfig,
101    host: Option<Arc<MissionHost>>,
102    unavailable_reason: Option<String>,
103}
104
105impl RepoContext {
106    pub fn id(&self) -> &str {
107        &self.config.id
108    }
109
110    pub fn root(&self) -> &Path {
111        &self.config.root
112    }
113
114    pub fn config(&self) -> &RepoConfig {
115        &self.config
116    }
117
118    pub fn host(&self) -> Option<&Arc<MissionHost>> {
119        self.host.as_ref()
120    }
121
122    pub fn is_healthy(&self) -> bool {
123        self.host.is_some()
124    }
125
126    pub fn unavailable_reason(&self) -> Option<&str> {
127        self.unavailable_reason.as_deref()
128    }
129
130    fn summary(&self, is_default: bool) -> RepoSummary {
131        RepoSummary {
132            id: self.config.id.clone(),
133            root: self.config.root.to_string_lossy().into_owned(),
134            display_name: self
135                .config
136                .display_name
137                .clone()
138                .unwrap_or_else(|| self.config.id.clone()),
139            group: self.config.group.clone(),
140            pinned: self.config.pinned,
141            is_default,
142            status: if self.is_healthy() {
143                "healthy".to_string()
144            } else {
145                "unavailable".to_string()
146            },
147            error: self.unavailable_reason.clone(),
148            activity: if self.is_healthy() {
149                repo_activity(&self.config.root)
150            } else {
151                RepoActivity::default()
152            },
153        }
154    }
155}
156
157/// Public `GET /api/repos` row.
158#[derive(Clone, Debug, Deserialize, Serialize)]
159#[serde(rename_all = "camelCase")]
160pub struct RepoSummary {
161    pub id: String,
162    pub root: String,
163    pub display_name: String,
164    #[serde(default, skip_serializing_if = "Option::is_none")]
165    pub group: Option<String>,
166    pub pinned: bool,
167    pub is_default: bool,
168    pub status: String,
169    #[serde(default, skip_serializing_if = "Option::is_none")]
170    pub error: Option<String>,
171    #[serde(default)]
172    pub activity: RepoActivity,
173}
174
175/// At-a-glance pipeline counts for the repository picker. Counts follow the
176/// dashboard's work-item projection: ticket-backed missions count once under
177/// the ticket state; only ticketless missions are folded independently.
178#[derive(Clone, Debug, Default, Deserialize, Serialize, PartialEq, Eq)]
179#[serde(rename_all = "camelCase")]
180pub struct RepoActivity {
181    pub queued: usize,
182    pub running: usize,
183    pub needs_input: usize,
184    pub complete_unmerged: usize,
185    pub failed: usize,
186}
187
188fn repo_activity(repo_root: &Path) -> RepoActivity {
189    let mut activity = RepoActivity::default();
190    let tickets = Ticket::list(repo_root);
191    let mut linked = HashSet::new();
192    for ticket in tickets {
193        let mission_id = Ticket::mission_for(repo_root, &ticket.slug);
194        if let Some(id) = mission_id.as_ref() {
195            linked.insert(id.clone());
196        }
197        match Ticket::read_state(repo_root, &ticket.slug) {
198            TicketState::Queued => activity.queued += 1,
199            TicketState::Running => activity.running += 1,
200            // NeedsContext and the wrong-plan escalation are both parked
201            // awaiting the operator — one "needs input" count.
202            TicketState::NeedsContext | TicketState::WrongPlan => activity.needs_input += 1,
203            TicketState::Failed => activity.failed += 1,
204            TicketState::Done
205                if mission_id.is_some()
206                    && kranz_engine::merged::ticket_merged(repo_root, &ticket.slug)
207                        != Some(true) =>
208            {
209                activity.complete_unmerged += 1;
210            }
211            _ => {}
212        }
213    }
214
215    let repo = GitRepo::open(repo_root).ok();
216    for id in MissionPaths::list_missions(repo_root) {
217        if linked.contains(&id) {
218            continue;
219        }
220        let paths = MissionPaths::new(repo_root, &id);
221        let state = EventLog::read_events(&paths.events_file())
222            .ok()
223            .and_then(|events| reducer::fold(&events).ok());
224        let Some(state) = state else {
225            activity.failed += 1;
226            continue;
227        };
228        match state.mission.status {
229            MissionStatus::Approved => activity.queued += 1,
230            MissionStatus::Running | MissionStatus::Paused | MissionStatus::Validating => {
231                activity.running += 1
232            }
233            MissionStatus::Blocked => activity.needs_input += 1,
234            MissionStatus::Complete
235                if repo
236                    .as_ref()
237                    .and_then(|repo| merged_bit(repo, &state.mission))
238                    != Some(true) =>
239            {
240                activity.complete_unmerged += 1;
241            }
242            MissionStatus::Failed => activity.failed += 1,
243            _ => {}
244        }
245    }
246    activity
247}
248
249/// Static process-lifetime catalog plus one existing single-repo host per
250/// healthy root.
251pub struct MultiRepoHost {
252    repos: BTreeMap<String, Arc<RepoContext>>,
253    default_repo: Option<String>,
254    max_concurrent_repos: usize,
255    operator_catalog: bool,
256    global_run_permits: Arc<Semaphore>,
257    auto_work_cursor: Mutex<usize>,
258    auto_work: Mutex<Option<tokio::task::JoinHandle<()>>>,
259}
260
261impl MultiRepoHost {
262    /// Wrap an already-constructed host (including test hosts with injected
263    /// backends) in a one-repository catalog.
264    pub fn with_host(host: Arc<MissionHost>) -> Self {
265        let root = host.repo_root().clone();
266        let id = fallback_repo_id(&root);
267        let context = Arc::new(RepoContext {
268            config: RepoConfig {
269                id: id.clone(),
270                root,
271                display_name: None,
272                group: None,
273                pinned: false,
274                slack: RepoSlackConfig::default(),
275            },
276            host: Some(host),
277            unavailable_reason: None,
278        });
279        Self {
280            repos: BTreeMap::from([(id.clone(), context)]),
281            default_repo: Some(id),
282            max_concurrent_repos: 1,
283            operator_catalog: false,
284            global_run_permits: Arc::new(Semaphore::new(1)),
285            auto_work_cursor: Mutex::new(0),
286            auto_work: Mutex::new(None),
287        }
288    }
289
290    /// Preserve the historical one-repository serve path when no global host
291    /// catalog is configured.
292    pub fn single(repo_root: PathBuf) -> Result<Self> {
293        let canonical = canonical_or_lexical(&repo_root);
294        let id = fallback_repo_id(&canonical);
295        Self::from_config_with_mode(
296            HostConfig {
297                default_repo: Some(id.clone()),
298                max_concurrent_repos: 1,
299                repos: vec![RepoConfig {
300                    id,
301                    root: canonical,
302                    display_name: None,
303                    group: None,
304                    pinned: false,
305                    slack: RepoSlackConfig::default(),
306                }],
307            },
308            false,
309        )
310    }
311
312    /// Load the global catalog, falling back to the historical current repo
313    /// only when the `host.repos` list is absent or empty.
314    pub fn from_global_config(global_path: Option<&Path>, fallback_root: PathBuf) -> Result<Self> {
315        let config = match global_path {
316            Some(path) => load_host_config(path)?,
317            None => HostConfig::default(),
318        };
319        if config.repos.is_empty() {
320            Self::single(fallback_root)
321        } else {
322            Self::from_config(config)
323        }
324    }
325
326    pub fn from_config(config: HostConfig) -> Result<Self> {
327        Self::from_config_with_mode(config, true)
328    }
329
330    fn from_config_with_mode(mut config: HostConfig, operator_catalog: bool) -> Result<Self> {
331        if config.repos.is_empty() {
332            return Err(anyhow!("host.repos must contain at least one repository"));
333        }
334        if config.max_concurrent_repos == 0 {
335            return Err(anyhow!("host.maxConcurrentRepos must be at least 1"));
336        }
337
338        let mut ids = HashSet::new();
339        let mut roots = HashSet::new();
340        let mut repos = BTreeMap::new();
341        let global_run_permits = Arc::new(Semaphore::new(config.max_concurrent_repos));
342
343        for repo in &mut config.repos {
344            validate_repo_id(&repo.id)?;
345            if !ids.insert(repo.id.clone()) {
346                return Err(anyhow!("duplicate host repository id '{}'", repo.id));
347            }
348            if !repo.root.is_absolute() {
349                return Err(anyhow!(
350                    "host repository '{}' root must be absolute: {}",
351                    repo.id,
352                    repo.root.display()
353                ));
354            }
355            repo.root = canonical_or_lexical(&repo.root);
356            if !roots.insert(repo.root.clone()) {
357                return Err(anyhow!(
358                    "duplicate host repository root: {}",
359                    repo.root.display()
360                ));
361            }
362
363            let unavailable_reason = repository_unavailable_reason(&repo.root);
364            let host = unavailable_reason.is_none().then(|| {
365                Arc::new(MissionHost::new_with_global_run_permits(
366                    repo.root.clone(),
367                    Arc::clone(&global_run_permits),
368                ))
369            });
370            repos.insert(
371                repo.id.clone(),
372                Arc::new(RepoContext {
373                    config: repo.clone(),
374                    host,
375                    unavailable_reason,
376                }),
377            );
378        }
379
380        if let Some(default) = config.default_repo.as_deref() {
381            validate_repo_id(default)?;
382            if !repos.contains_key(default) {
383                return Err(anyhow!(
384                    "host.defaultRepo '{}' does not name a configured repository",
385                    default
386                ));
387            }
388        }
389
390        Ok(Self {
391            repos,
392            default_repo: config.default_repo,
393            max_concurrent_repos: config.max_concurrent_repos,
394            operator_catalog,
395            global_run_permits,
396            auto_work_cursor: Mutex::new(0),
397            auto_work: Mutex::new(None),
398        })
399    }
400
401    pub fn contexts(&self) -> impl Iterator<Item = Arc<RepoContext>> + '_ {
402        self.repos.values().cloned()
403    }
404
405    pub fn healthy_contexts(&self) -> impl Iterator<Item = Arc<RepoContext>> + '_ {
406        self.contexts().filter(|context| context.is_healthy())
407    }
408
409    pub fn resolve(&self, id: &str) -> Option<Arc<RepoContext>> {
410        self.repos.get(id).cloned()
411    }
412
413    /// One fair auto-work scheduling pass (test + serve watcher entrypoint).
414    pub async fn auto_work_tick(&self) -> usize {
415        self.auto_work_tick_inner().await
416    }
417
418    /// Existing unscoped routes are available for an explicit default, or
419    /// when exactly one healthy root makes the choice unambiguous.
420    pub fn compatibility_context(&self) -> Option<Arc<RepoContext>> {
421        if let Some(default) = self.default_repo.as_deref() {
422            return self.resolve(default);
423        }
424        let mut healthy = self.healthy_contexts();
425        let only = healthy.next()?;
426        healthy.next().is_none().then_some(only)
427    }
428
429    pub fn summaries(&self) -> Vec<RepoSummary> {
430        self.repos
431            .values()
432            .map(|context| context.summary(self.default_repo.as_deref() == Some(context.id())))
433            .collect()
434    }
435
436    pub fn max_concurrent_repos(&self) -> usize {
437        self.max_concurrent_repos
438    }
439
440    pub fn default_repo(&self) -> Option<&str> {
441        self.default_repo.as_deref()
442    }
443
444    pub fn uses_operator_catalog(&self) -> bool {
445        self.operator_catalog
446    }
447
448    /// One fair auto-work scheduling pass. At most one pass over the static
449    /// catalog is made; each successful start advances the next pass beyond
450    /// that repository, while the shared semaphore enforces the configured
451    /// concurrency bound for the complete drain lifetime.
452    async fn auto_work_tick_inner(&self) -> usize {
453        let contexts: Vec<_> = self.healthy_contexts().collect();
454        if contexts.is_empty() {
455            return 0;
456        }
457        let start = *self.auto_work_cursor.lock().expect("auto-work cursor lock") % contexts.len();
458        let mut started = 0;
459        let mut last_started = None;
460        // Do not pre-check `available_permits()` — that races HTTP start/drain
461        // and can skip a fair rotation pass. Each host's try_acquire inside
462        // start/drain is the authoritative saturation gate (returns false).
463        for offset in 0..contexts.len() {
464            let index = (start + offset) % contexts.len();
465            if let Some(host) = contexts[index].host() {
466                if host.auto_work_tick().await {
467                    started += 1;
468                    last_started = Some(index);
469                }
470            }
471        }
472        if let Some(index) = last_started {
473            *self.auto_work_cursor.lock().expect("auto-work cursor lock") =
474                (index + 1) % contexts.len();
475        }
476        started
477    }
478
479    /// Spawn the single process-wide fair auto-work watcher at most once.
480    pub fn ensure_auto_work_started(self: &Arc<Self>) {
481        let mut guard = self.auto_work.lock().expect("multi auto-work lock");
482        if guard.is_some() {
483            return;
484        }
485        let host = Arc::clone(self);
486        *guard = Some(tokio::spawn(async move {
487            const AUTO_WORK_INTERVAL: Duration = Duration::from_secs(10);
488            loop {
489                tokio::time::sleep(AUTO_WORK_INTERVAL).await;
490                let _ = host.auto_work_tick().await;
491            }
492        }));
493    }
494
495    /// Remaining process-wide run slots under `host.maxConcurrentRepos`.
496    pub fn available_global_run_permits(&self) -> usize {
497        self.global_run_permits.available_permits()
498    }
499
500    #[cfg(test)]
501    fn auto_work_cursor(&self) -> usize {
502        *self.auto_work_cursor.lock().expect("auto-work cursor lock")
503    }
504
505    /// Test-only catalog that reuses already-constructed hosts (injected
506    /// backends + a shared semaphore) without rediscovering Claude.
507    #[cfg(test)]
508    fn from_injected_hosts(
509        entries: Vec<(String, Arc<MissionHost>)>,
510        max_concurrent_repos: usize,
511        global_run_permits: Arc<Semaphore>,
512    ) -> Self {
513        let mut repos = BTreeMap::new();
514        for (id, host) in entries {
515            let root = host.repo_root().clone();
516            repos.insert(
517                id.clone(),
518                Arc::new(RepoContext {
519                    config: RepoConfig {
520                        id: id.clone(),
521                        root,
522                        display_name: None,
523                        group: None,
524                        pinned: false,
525                        slack: RepoSlackConfig::default(),
526                    },
527                    host: Some(host),
528                    unavailable_reason: None,
529                }),
530            );
531        }
532        Self {
533            repos,
534            default_repo: None,
535            max_concurrent_repos,
536            operator_catalog: true,
537            global_run_permits,
538            auto_work_cursor: Mutex::new(0),
539            auto_work: Mutex::new(None),
540        }
541    }
542}
543
544fn validate_repo_id(id: &str) -> Result<()> {
545    let mut chars = id.chars();
546    let valid_first = chars.next().is_some_and(|ch| ch.is_ascii_alphanumeric());
547    let valid_rest = chars.all(|ch| ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_'));
548    if valid_first && valid_rest {
549        Ok(())
550    } else {
551        Err(anyhow!(
552            "invalid host repository id '{id}': use ASCII letters, digits, '-' or '_', starting with a letter or digit"
553        ))
554    }
555}
556
557fn canonical_or_lexical(path: &Path) -> PathBuf {
558    std::fs::canonicalize(path).unwrap_or_else(|_| lexical_normalize(path))
559}
560
561fn lexical_normalize(path: &Path) -> PathBuf {
562    let mut normalized = PathBuf::new();
563    for component in path.components() {
564        match component {
565            Component::CurDir => {}
566            Component::ParentDir => {
567                normalized.pop();
568            }
569            other => normalized.push(other.as_os_str()),
570        }
571    }
572    normalized
573}
574
575fn repository_unavailable_reason(root: &Path) -> Option<String> {
576    if !root.exists() {
577        return Some(format!(
578            "repository root does not exist: {}",
579            root.display()
580        ));
581    }
582    if !root.is_dir() {
583        return Some(format!(
584            "repository root is not a directory: {}",
585            root.display()
586        ));
587    }
588    if !root.join(".git").exists() {
589        return Some(format!(
590            "repository root is not a Git worktree: {}",
591            root.display()
592        ));
593    }
594    GitRepo::open(root).err().map(|error| {
595        format!(
596            "repository root is not a usable Git worktree: {} ({error})",
597            root.display()
598        )
599    })
600}
601
602fn fallback_repo_id(root: &Path) -> String {
603    let raw = root
604        .file_name()
605        .and_then(|name| name.to_str())
606        .unwrap_or("repo");
607    let mut id = String::new();
608    for ch in raw.chars() {
609        if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_') {
610            id.push(ch);
611        } else if !id.ends_with('-') {
612            id.push('-');
613        }
614    }
615    let id = id.trim_matches(|ch| matches!(ch, '-' | '_'));
616    if id.is_empty() {
617        "repo".to_string()
618    } else {
619        id.to_string()
620    }
621}
622
623#[cfg(test)]
624mod tests {
625    use super::*;
626
627    fn init_git(root: &Path) {
628        std::fs::create_dir_all(root).unwrap();
629        let status = std::process::Command::new("git")
630            .args(["init", "-q"])
631            .arg(root)
632            .status()
633            .unwrap();
634        assert!(status.success());
635    }
636
637    fn repo_config(id: &str, root: PathBuf) -> RepoConfig {
638        RepoConfig {
639            id: id.to_string(),
640            root,
641            display_name: None,
642            group: None,
643            pinned: false,
644            slack: RepoSlackConfig::default(),
645        }
646    }
647
648    #[test]
649    fn single_repo_fallback_normalizes_leading_underscores() {
650        let temp = tempfile::tempdir().unwrap();
651        let root = temp.path().join("_project");
652        init_git(&root);
653
654        let catalog = MultiRepoHost::single(root).unwrap();
655
656        assert_eq!(catalog.compatibility_context().unwrap().id(), "project");
657    }
658
659    #[test]
660    fn catalog_rejects_duplicate_ids_and_canonical_roots() {
661        let temp = tempfile::tempdir().unwrap();
662        init_git(temp.path());
663        let root = std::fs::canonicalize(temp.path()).unwrap();
664
665        let duplicate_id = HostConfig {
666            repos: vec![
667                repo_config("one", root.clone()),
668                repo_config("one", root.join("other")),
669            ],
670            ..HostConfig::default()
671        };
672        assert!(MultiRepoHost::from_config(duplicate_id)
673            .err()
674            .unwrap()
675            .to_string()
676            .contains("duplicate host repository id"));
677
678        let duplicate_root = HostConfig {
679            repos: vec![repo_config("one", root.clone()), repo_config("two", root)],
680            ..HostConfig::default()
681        };
682        assert!(MultiRepoHost::from_config(duplicate_root)
683            .err()
684            .unwrap()
685            .to_string()
686            .contains("duplicate host repository root"));
687    }
688
689    #[test]
690    fn missing_root_stays_visible_but_unavailable() {
691        let temp = tempfile::tempdir().unwrap();
692        let missing = temp.path().join("missing");
693        let catalog = MultiRepoHost::from_config(HostConfig {
694            repos: vec![repo_config("missing", missing)],
695            ..HostConfig::default()
696        })
697        .unwrap();
698        let summary = &catalog.summaries()[0];
699        assert_eq!(summary.status, "unavailable");
700        assert!(summary.error.as_deref().unwrap().contains("does not exist"));
701    }
702
703    #[test]
704    fn fake_dot_git_directory_is_unavailable() {
705        let temp = tempfile::tempdir().unwrap();
706        let root = temp.path().join("fake");
707        std::fs::create_dir_all(root.join(".git")).unwrap();
708        let catalog = MultiRepoHost::from_config(HostConfig {
709            repos: vec![repo_config("fake", root)],
710            ..HostConfig::default()
711        })
712        .unwrap();
713        let summary = &catalog.summaries()[0];
714        assert_eq!(summary.status, "unavailable");
715        assert!(summary.error.as_deref().unwrap().contains("usable Git"));
716    }
717
718    #[test]
719    fn picker_activity_counts_each_ticket_once_and_complete_unmerged_separately() {
720        use kranz_engine::event_log::LockForce;
721        use kranz_engine::events::EventKind;
722        use kranz_engine::types::MissionConfig;
723
724        let temp = tempfile::tempdir().unwrap();
725        init_git(temp.path());
726        for (slug, state) in [
727            ("queued", TicketState::Queued),
728            ("running", TicketState::Running),
729            ("needs-input", TicketState::NeedsContext),
730            ("failed", TicketState::Failed),
731        ] {
732            Ticket::scaffold(temp.path(), slug, slug, None, None).unwrap();
733            Ticket::write_state(temp.path(), slug, state, None).unwrap();
734        }
735        let paths = MissionPaths::new(temp.path(), "m-unmerged");
736        let mut log =
737            EventLog::acquire(&paths, "m-unmerged", Duration::ZERO, LockForce::No).unwrap();
738        log.append(EventKind::MissionCreated {
739            goal: "complete but not merged".into(),
740            base_branch: "main".into(),
741            mission_branch: "kranz/mission-m-unmerged".into(),
742            config: MissionConfig::default(),
743        })
744        .unwrap();
745        log.append(EventKind::MissionCompleted {}).unwrap();
746        drop(log);
747
748        assert_eq!(
749            repo_activity(temp.path()),
750            RepoActivity {
751                queued: 1,
752                running: 1,
753                needs_input: 1,
754                complete_unmerged: 1,
755                failed: 1,
756            }
757        );
758    }
759
760    #[test]
761    fn unscoped_alias_requires_default_or_one_healthy_repo() {
762        let temp = tempfile::tempdir().unwrap();
763        let a = temp.path().join("a");
764        let b = temp.path().join("b");
765        for root in [&a, &b] {
766            init_git(root);
767        }
768        let catalog = MultiRepoHost::from_config(HostConfig {
769            repos: vec![repo_config("a", a), repo_config("b", b)],
770            ..HostConfig::default()
771        })
772        .unwrap();
773        assert!(catalog.compatibility_context().is_none());
774    }
775
776    #[test]
777    fn configured_concurrency_limit_is_shared_across_repository_hosts() {
778        let temp = tempfile::tempdir().unwrap();
779        let a = temp.path().join("a");
780        let b = temp.path().join("b");
781        for root in [&a, &b] {
782            init_git(root);
783        }
784        let catalog = MultiRepoHost::from_config(HostConfig {
785            max_concurrent_repos: 1,
786            repos: vec![repo_config("a", a), repo_config("b", b)],
787            ..HostConfig::default()
788        })
789        .unwrap();
790        let a = catalog.resolve("a").unwrap();
791        let b = catalog.resolve("b").unwrap();
792        let permit = a.host().unwrap().try_global_run_permit().unwrap().unwrap();
793        assert_eq!(catalog.available_global_run_permits(), 0);
794        let error = b.host().unwrap().try_global_run_permit().unwrap_err();
795        assert_eq!(error.status, axum::http::StatusCode::CONFLICT);
796        drop(permit);
797        assert_eq!(catalog.available_global_run_permits(), 1);
798        assert!(b.host().unwrap().try_global_run_permit().unwrap().is_some());
799    }
800
801    fn write_auto_work_config(root: &Path, enabled: bool) {
802        let dir = root.join(".kranz");
803        std::fs::create_dir_all(&dir).unwrap();
804        std::fs::write(
805            dir.join("config.json"),
806            serde_json::json!({ "autoWork": enabled }).to_string(),
807        )
808        .unwrap();
809    }
810
811    fn enqueue_placeholder(root: &Path, mission_id: &str) {
812        kranz_engine::queue::enqueue(
813            root,
814            kranz_engine::queue::QueueEntry {
815                mission_id: mission_id.to_string(),
816                ticket_slug: None,
817                priority: 2,
818                seq: 0,
819            },
820        )
821        .unwrap();
822    }
823
824    async fn wait_for_idle_drain(host: &MissionHost) {
825        let deadline = tokio::time::Instant::now() + Duration::from_secs(10);
826        loop {
827            let state = host.queue_state();
828            if state["drain"]["live"] == false {
829                return;
830            }
831            assert!(
832                tokio::time::Instant::now() < deadline,
833                "drain never settled idle: {state}"
834            );
835            tokio::time::sleep(Duration::from_millis(20)).await;
836        }
837    }
838
839    #[tokio::test]
840    async fn auto_work_tick_is_noop_when_global_permits_saturated() {
841        let temp = tempfile::tempdir().unwrap();
842        let a = temp.path().join("a");
843        let b = temp.path().join("b");
844        for root in [&a, &b] {
845            init_git(root);
846            write_auto_work_config(root, true);
847            enqueue_placeholder(root, "m-queued");
848        }
849        let permits = Arc::new(Semaphore::new(1));
850        let backend: Arc<dyn kranz_engine::backend::AgentBackend> =
851            Arc::new(kranz_engine::backend_mock::MockBackend::new());
852        let host_a = Arc::new(MissionHost::with_backend_and_global_run_permits(
853            a,
854            Arc::clone(&backend),
855            Arc::clone(&permits),
856        ));
857        let host_b = Arc::new(MissionHost::with_backend_and_global_run_permits(
858            b,
859            backend,
860            Arc::clone(&permits),
861        ));
862        let catalog = MultiRepoHost::from_injected_hosts(
863            vec![("a".into(), host_a), ("b".into(), host_b)],
864            1,
865            Arc::clone(&permits),
866        );
867        let _hold = Arc::clone(&permits).try_acquire_owned().unwrap();
868
869        assert_eq!(catalog.auto_work_tick().await, 0);
870        assert_eq!(catalog.auto_work_cursor(), 0);
871    }
872
873    #[tokio::test]
874    async fn auto_work_tick_skips_busy_repo_and_starts_next_ready_repo() {
875        let temp = tempfile::tempdir().unwrap();
876        let a = temp.path().join("a");
877        let b = temp.path().join("b");
878        for root in [&a, &b] {
879            init_git(root);
880            write_auto_work_config(root, true);
881            enqueue_placeholder(root, "m-queued");
882        }
883        let busy = kranz_engine::queue::acquire_repo_busy(&a, "m-external").unwrap();
884        let permits = Arc::new(Semaphore::new(1));
885        let backend: Arc<dyn kranz_engine::backend::AgentBackend> =
886            Arc::new(kranz_engine::backend_mock::MockBackend::new());
887        let host_a = Arc::new(MissionHost::with_backend_and_global_run_permits(
888            a,
889            Arc::clone(&backend),
890            Arc::clone(&permits),
891        ));
892        let host_b = Arc::new(MissionHost::with_backend_and_global_run_permits(
893            b,
894            backend,
895            Arc::clone(&permits),
896        ));
897        let catalog = MultiRepoHost::from_injected_hosts(
898            vec![
899                ("a".into(), Arc::clone(&host_a)),
900                ("b".into(), Arc::clone(&host_b)),
901            ],
902            1,
903            permits,
904        );
905
906        assert_eq!(catalog.auto_work_tick().await, 1);
907        assert!(!host_a.drain_is_live());
908        assert!(host_b.drain_is_live());
909        assert_eq!(catalog.auto_work_cursor(), 0);
910
911        wait_for_idle_drain(&host_b).await;
912        drop(busy);
913    }
914
915    #[tokio::test]
916    async fn auto_work_tick_rotates_fairly_under_max_concurrent_one() {
917        let temp = tempfile::tempdir().unwrap();
918        let a = temp.path().join("a");
919        let b = temp.path().join("b");
920        for root in [&a, &b] {
921            init_git(root);
922            write_auto_work_config(root, true);
923            enqueue_placeholder(root, "m-queued");
924        }
925        let permits = Arc::new(Semaphore::new(1));
926        let backend: Arc<dyn kranz_engine::backend::AgentBackend> =
927            Arc::new(kranz_engine::backend_mock::MockBackend::new());
928        let host_a = Arc::new(MissionHost::with_backend_and_global_run_permits(
929            a,
930            Arc::clone(&backend),
931            Arc::clone(&permits),
932        ));
933        let host_b = Arc::new(MissionHost::with_backend_and_global_run_permits(
934            b,
935            backend,
936            Arc::clone(&permits),
937        ));
938        let catalog = MultiRepoHost::from_injected_hosts(
939            vec![
940                ("a".into(), Arc::clone(&host_a)),
941                ("b".into(), Arc::clone(&host_b)),
942            ],
943            1,
944            permits,
945        );
946
947        assert_eq!(catalog.auto_work_tick().await, 1);
948        assert!(host_a.drain_is_live() || host_b.drain_is_live());
949        // BTreeMap order is a then b; cursor starts at 0 so a starts first.
950        assert!(host_a.drain_is_live());
951        assert!(!host_b.drain_is_live());
952        assert_eq!(catalog.auto_work_cursor(), 1);
953
954        wait_for_idle_drain(&host_a).await;
955        assert_eq!(catalog.auto_work_tick().await, 1);
956        assert!(host_b.drain_is_live());
957        assert_eq!(catalog.auto_work_cursor(), 0);
958        wait_for_idle_drain(&host_b).await;
959    }
960}
kranz_server/multi.rs

kranz_server/
multi.rs
