relay_knowledge/interfaces/
cli.rs

1//! CLI adapter for the shared application service.
2
3#[path = "cli_grammar.rs"]
4mod cli_grammar;
5#[path = "cli_render.rs"]
6mod cli_render;
7#[path = "cli_spec.rs"]
8mod cli_spec;
9#[path = "files_cli.rs"]
10mod files_cli;
11#[path = "knowledge_cli.rs"]
12mod knowledge_cli;
13#[path = "ops_cli.rs"]
14mod ops_cli;
15#[path = "repo_cli.rs"]
16mod repo_cli;
17#[path = "setup_cli.rs"]
18mod setup_cli;
19#[path = "version_cli.rs"]
20mod version_cli;
21
22use std::{error::Error, fmt};
23
24use crate::{
25    api::{
26        GraphInspectionRequest, HybridRetrievalRequest, IndexRefreshRequest, IngestEvidence,
27        IngestRequest, InterfaceKind, RequestContext,
28    },
29    application::{RelayKnowledgeService, RuntimeConfiguration},
30    domain::{FreshnessPolicy, IndexKind, ProposalState, ServiceManagerAction, WorkerKind},
31    interfaces::{agent::mcp::McpServer, web},
32    net::qos::QosRuntime,
33};
34
35use cli_render::{render_project_status, render_response, serialize_line};
36
37/// Supported CLI output formats.
38#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
39pub enum OutputFormat {
40    #[default]
41    Text,
42    Json,
43    Markdown,
44    StreamingJson,
45}
46
47impl OutputFormat {
48    fn as_str(self) -> &'static str {
49        match self {
50            Self::Text => "text",
51            Self::Json => "json",
52            Self::Markdown => "markdown",
53            Self::StreamingJson => "streaming-json",
54        }
55    }
56
57    /// Parses a CLI output format value.
58    pub fn parse(value: &str) -> Result<Self, CliError> {
59        match value {
60            "text" => Ok(Self::Text),
61            "json" => Ok(Self::Json),
62            "markdown" => Ok(Self::Markdown),
63            "streaming-json" => Ok(Self::StreamingJson),
64            other => Err(CliError::invalid_format(other)),
65        }
66    }
67}
68
69/// Parsed CLI command.
70#[derive(Debug, Clone, PartialEq, Eq)]
71pub struct CliCommand {
72    pub action: CliAction,
73    pub format: OutputFormat,
74    pub help: bool,
75}
76
77impl CliCommand {
78    /// Parses the CLI arguments after the binary name.
79    pub fn parse<I, S>(args: I) -> Result<Self, CliError>
80    where
81        I: IntoIterator<Item = S>,
82        S: Into<String>,
83    {
84        let tokens = args.into_iter().map(Into::into).collect::<Vec<_>>();
85        let mut action_tokens = Vec::new();
86        let mut format = OutputFormat::default();
87        let mut help = false;
88        let mut version = false;
89        let mut command_seen = false;
90        let mut delimiter_value = false;
91        let mut index = 0;
92
93        while index < tokens.len() {
94            let arg = &tokens[index];
95            if delimiter_value {
96                action_tokens.push(arg.clone());
97                delimiter_value = false;
98                index += 1;
99            } else if arg == "--format" {
100                let value = tokens
101                    .get(index + 1)
102                    .ok_or(CliError::MissingFormatValue)?
103                    .clone();
104                format = OutputFormat::parse(&value)?;
105                index += 2;
106            } else if let Some(value) = arg.strip_prefix("--format=") {
107                format = OutputFormat::parse(value)?;
108                index += 1;
109            } else if arg == "--help" || arg == "-h" {
110                help = true;
111                index += 1;
112            } else if arg == "--version" && !command_seen {
113                version = true;
114                index += 1;
115            } else if arg == "--" {
116                action_tokens.push(arg.clone());
117                delimiter_value = true;
118                index += 1;
119            } else if option_consumes_value(arg) {
120                action_tokens.push(arg.clone());
121                if let Some(value) = tokens.get(index + 1) {
122                    action_tokens.push(value.clone());
123                    index += 2;
124                } else {
125                    index += 1;
126                }
127            } else {
128                command_seen |= is_command_word(arg);
129                action_tokens.push(arg.clone());
130                index += 1;
131            }
132        }
133
134        let action = if help {
135            CliAction::Help {
136                path: help_path(action_tokens),
137            }
138        } else if version {
139            if let Some(token) = action_tokens.first() {
140                let error = CliError::UnexpectedArgument(token.clone());
141                return Err(cli_grammar::diagnose(&action_tokens, error, format));
142            }
143            CliAction::Version
144        } else {
145            match parse_action(action_tokens.clone()) {
146                Ok(action) => action,
147                Err(error) => return Err(cli_grammar::diagnose(&action_tokens, error, format)),
148            }
149        };
150
151        Ok(Self {
152            action,
153            format,
154            help,
155        })
156    }
157}
158
159fn option_consumes_value(option: &str) -> bool {
160    matches!(
161        option,
162        "--source"
163            | "--content"
164            | "--entity"
165            | "--limit"
166            | "--freshness"
167            | "--kind"
168            | "--alias"
169            | "--path"
170            | "--language"
171            | "--ref"
172            | "--base"
173            | "--head"
174            | "--query"
175            | "--mcp"
176            | "--state"
177            | "--by"
178            | "--reason"
179            | "--operation"
180            | "--input"
181            | "--root"
182    )
183}
184
185fn is_command_word(token: &str) -> bool {
186    matches!(
187        token,
188        "status"
189            | "ingest"
190            | "query"
191            | "repo"
192            | "files"
193            | "graph"
194            | "index"
195            | "worker"
196            | "proposal"
197            | "audit"
198            | "provider"
199            | "health"
200            | "service"
201            | "setup"
202            | "version"
203            | "help"
204    )
205}
206
207/// CLI action after global options are removed.
208#[derive(Debug, Clone, PartialEq, Eq)]
209pub enum CliAction {
210    Status,
211    Ingest {
212        source_scope: String,
213        content: String,
214        entity_labels: Vec<String>,
215    },
216    Query {
217        query: String,
218        source_scope: Option<String>,
219        limit: usize,
220        freshness: FreshnessPolicy,
221    },
222    FilesIndex {
223        source_scope: Option<String>,
224        roots: Vec<String>,
225    },
226    FilesQuery {
227        query: String,
228        source_scope: Option<String>,
229        root_id: Option<String>,
230        limit: usize,
231    },
232    GraphInspect,
233    IndexRefresh {
234        kinds: Vec<IndexKind>,
235    },
236    WorkerStatus {
237        kind: Option<WorkerKind>,
238    },
239    WorkerRunOnce {
240        kind: Option<WorkerKind>,
241    },
242    ProposalList {
243        state: Option<ProposalState>,
244        limit: usize,
245    },
246    ProposalShow {
247        proposal_id: String,
248    },
249    ProposalAccept {
250        proposal_id: String,
251        actor: String,
252        reason: Option<String>,
253    },
254    ProposalReject {
255        proposal_id: String,
256        actor: String,
257        reason: Option<String>,
258    },
259    ProposalSupersede {
260        proposal_id: String,
261        actor: String,
262        reason: Option<String>,
263    },
264    AuditQuery {
265        operation: Option<String>,
266        limit: usize,
267    },
268    ProviderProbe,
269    Repo(repo_cli::RepoCommand),
270    Health,
271    ServiceStatus,
272    ServicePlan {
273        action: ServiceManagerAction,
274    },
275    ServiceDefinitionWrite,
276    ServiceOperatorStatus,
277    ServiceOperatorPause,
278    ServiceOperatorResume,
279    ServiceRun {
280        mcp: ServiceMcpTransport,
281        web: bool,
282    },
283    SetupDoctor,
284    SetupProfile {
285        profile: setup_cli::SetupProfile,
286    },
287    Version,
288    VersionCheck,
289    Help {
290        path: Vec<String>,
291    },
292}
293
294/// MCP transport option for foreground service mode.
295#[derive(Debug, Clone, Copy, PartialEq, Eq)]
296pub enum ServiceMcpTransport {
297    Configured,
298    StreamableHttp,
299}
300
301/// CLI adapter error.
302#[derive(Debug, Clone, PartialEq, Eq)]
303pub enum CliError {
304    Diagnostic(Box<CliDiagnostic>),
305    InvalidFormat(String),
306    InvalidCodeQueryKind(String),
307    InvalidFreshness(String),
308    InvalidIndexKind(String),
309    InvalidWorkerKind(String),
310    InvalidProposalState(String),
311    InvalidServiceAction(String),
312    InvalidLimit(String),
313    MissingFormatValue,
314    MissingValue(&'static str),
315    UnsupportedVersionFormat(OutputFormat),
316    UnknownHelpTopic(String),
317    UnexpectedArgument(String),
318    RuntimeConfigFailed(String),
319    ApiFailed(String),
320    ServiceRunFailed(String),
321    RenderFailed(String),
322}
323
324impl CliError {
325    fn invalid_format(format: &str) -> Self {
326        Self::InvalidFormat(format.to_owned())
327    }
328
329    /// Returns the process exit code for the error.
330    pub fn exit_code(&self) -> i32 {
331        match self {
332            Self::Diagnostic(_)
333            | Self::InvalidFormat(_)
334            | Self::InvalidCodeQueryKind(_)
335            | Self::InvalidFreshness(_)
336            | Self::InvalidIndexKind(_)
337            | Self::InvalidWorkerKind(_)
338            | Self::InvalidProposalState(_)
339            | Self::InvalidServiceAction(_)
340            | Self::InvalidLimit(_)
341            | Self::MissingFormatValue
342            | Self::MissingValue(_)
343            | Self::UnsupportedVersionFormat(_)
344            | Self::UnknownHelpTopic(_)
345            | Self::UnexpectedArgument(_) => 2,
346            Self::RuntimeConfigFailed(_)
347            | Self::ApiFailed(_)
348            | Self::ServiceRunFailed(_)
349            | Self::RenderFailed(_) => 1,
350        }
351    }
352
353    /// Renders the process stderr payload for this error.
354    pub fn render_stderr(&self) -> String {
355        match self {
356            Self::Diagnostic(diagnostic) => diagnostic.render_stderr(),
357            _ => self.to_string(),
358        }
359    }
360}
361
362impl fmt::Display for CliError {
363    fn fmt(&self, formatter: &mut fmt::Formatter<'_>) -> fmt::Result {
364        match self {
365            Self::Diagnostic(diagnostic) => write!(formatter, "{}", diagnostic.render_text()),
366            Self::InvalidFormat(format) => write!(
367                formatter,
368                "invalid --format value '{format}', expected text, json, markdown, or streaming-json"
369            ),
370            Self::InvalidCodeQueryKind(value) => write!(
371                formatter,
372                "invalid --kind value '{value}', expected hybrid, symbol, definition, references, callers, callees, or imports"
373            ),
374            Self::InvalidFreshness(value) => write!(
375                formatter,
376                "invalid --freshness value '{value}', expected allow-stale, wait-until-fresh, or graph-only"
377            ),
378            Self::InvalidIndexKind(value) => write!(
379                formatter,
380                "invalid --kind value '{value}', expected bm25, semantic, or vector"
381            ),
382            Self::InvalidWorkerKind(value) => write!(
383                formatter,
384                "invalid worker kind '{value}', expected embedding, ocr, vision, or extractor"
385            ),
386            Self::InvalidProposalState(value) => write!(
387                formatter,
388                "invalid proposal state '{value}', expected proposed, accepted, rejected, or superseded"
389            ),
390            Self::InvalidServiceAction(value) => write!(
391                formatter,
392                "invalid service action '{value}', expected install or uninstall"
393            ),
394            Self::InvalidLimit(value) => write!(formatter, "invalid --limit value '{value}'"),
395            Self::MissingFormatValue => write!(formatter, "missing value for --format"),
396            Self::MissingValue(flag) => write!(formatter, "missing value for {flag}"),
397            Self::UnsupportedVersionFormat(format) => {
398                write!(
399                    formatter,
400                    "version does not support --format {}",
401                    format.as_str()
402                )
403            }
404            Self::UnknownHelpTopic(topic) => write!(formatter, "unknown help topic '{topic}'"),
405            Self::UnexpectedArgument(argument) => {
406                write!(formatter, "unexpected argument '{argument}'")
407            }
408            Self::RuntimeConfigFailed(message) => {
409                write!(formatter, "failed to load runtime configuration: {message}")
410            }
411            Self::ApiFailed(message) => write!(formatter, "{message}"),
412            Self::ServiceRunFailed(message) => write!(formatter, "{message}"),
413            Self::RenderFailed(message) => write!(formatter, "failed to render output: {message}"),
414        }
415    }
416}
417
418impl Error for CliError {}
419
420/// Structured parse diagnostic produced from the CLI grammar.
421#[derive(Debug, Clone, PartialEq, Eq)]
422pub struct CliDiagnostic {
423    message: String,
424    usage: Option<String>,
425    suggestion: Option<String>,
426    matched_path: Vec<String>,
427    unexpected_token: Option<String>,
428    expected: Vec<String>,
429    format: OutputFormat,
430}
431
432impl CliDiagnostic {
433    fn new(
434        message: String,
435        usage: Option<String>,
436        suggestion: Option<String>,
437        matched_path: Vec<String>,
438        unexpected_token: Option<String>,
439        expected: Vec<String>,
440        format: OutputFormat,
441    ) -> Self {
442        Self {
443            message,
444            usage,
445            suggestion,
446            matched_path,
447            unexpected_token,
448            expected,
449            format,
450        }
451    }
452
453    fn render_text(&self) -> String {
454        let mut output = self.message.clone();
455        if let Some(suggestion) = &self.suggestion {
456            output.push_str("\nTry: ");
457            output.push_str(suggestion);
458        }
459        if let Some(usage) = &self.usage {
460            output.push_str("\nUsage: ");
461            output.push_str(usage);
462        }
463
464        output
465    }
466
467    fn render_stderr(&self) -> String {
468        if self.format == OutputFormat::Json {
469            return serde_json::json!({
470                "error": self.message,
471                "usage": self.usage,
472                "suggestion": self.suggestion,
473                "matched_path": self.matched_path,
474                "unexpected_token": self.unexpected_token,
475                "expected": self.expected,
476            })
477            .to_string();
478        }
479
480        self.render_text()
481    }
482}
483
484/// Runs the CLI command and renders its response.
485pub async fn run<I, S>(args: I) -> Result<String, CliError>
486where
487    I: IntoIterator<Item = S>,
488    S: Into<String>,
489{
490    let command = CliCommand::parse(args)?;
491    run_command(command).await
492}
493
494/// Rendered stdout/stderr for the process entry point.
495#[derive(Debug, Clone, PartialEq, Eq)]
496pub struct CliProcessOutput {
497    pub stdout: String,
498    pub stderr: String,
499}
500
501/// Runs the CLI command and renders only the command result.
502pub async fn run_process<I, S>(
503    args: I,
504    _interactive_text_output: bool,
505) -> Result<CliProcessOutput, CliError>
506where
507    I: IntoIterator<Item = S>,
508    S: Into<String>,
509{
510    let command = CliCommand::parse(args)?;
511    let stdout = run_command(command).await?;
512
513    Ok(CliProcessOutput {
514        stdout,
515        stderr: String::new(),
516    })
517}
518
519/// Renders best-effort process-only notices after primary command output is emitted.
520pub async fn process_update_notice<I, S>(args: I, interactive_text_output: bool) -> Option<String>
521where
522    I: IntoIterator<Item = S>,
523    S: Into<String>,
524{
525    let command = CliCommand::parse(args).ok()?;
526    version_cli::update_notice_for_process(&command, interactive_text_output).await
527}
528
529async fn run_command(command: CliCommand) -> Result<String, CliError> {
530    if let CliAction::Help { path } = &command.action {
531        return cli_spec::render_help(path, command.format);
532    }
533    if command.action == CliAction::Version {
534        return version_cli::render_version(command.format);
535    }
536    if let CliAction::ServiceRun { mcp, web } = command.action.clone() {
537        return run_service(mcp, web).await;
538    }
539
540    let service = RelayKnowledgeService::from_process_environment()
541        .await
542        .map_err(|error| CliError::RuntimeConfigFailed(error.to_string()))?;
543    let context = RequestContext::for_interface(InterfaceKind::Cli);
544
545    run_with_service(&service, command, context).await
546}
547
548/// Runs a parsed CLI command with an already composed service.
549pub async fn run_with_service(
550    service: &RelayKnowledgeService,
551    command: CliCommand,
552    context: RequestContext,
553) -> Result<String, CliError> {
554    let format = command.format;
555    if let Some(output) =
556        ops_cli::run_operational_action(service, &command.action, context.clone(), format).await?
557    {
558        return Ok(output);
559    }
560    if let Some(output) =
561        setup_cli::run_setup_action(service, &command.action, context.clone(), format)?
562    {
563        return Ok(output);
564    }
565    if let Some(output) =
566        files_cli::run_files(service, &command.action, context.clone(), format).await?
567    {
568        return Ok(output);
569    }
570    match command.action {
571        CliAction::Status => {
572            let response = service
573                .project_status(context)
574                .await
575                .map_err(|error| CliError::ApiFailed(error.message))?;
576
577            render_project_status(&response, format)
578        }
579        CliAction::Ingest {
580            source_scope,
581            content,
582            entity_labels,
583        } => {
584            let response = service
585                .ingest(
586                    IngestRequest {
587                        source_scope,
588                        evidence: vec![IngestEvidence {
589                            id: None,
590                            source_path: None,
591                            span: None,
592                            confidence: None,
593                            status: None,
594                            content,
595                            entity_labels,
596                            extraction: None,
597                        }],
598                        relations: Vec::new(),
599                        claims: Vec::new(),
600                        events: Vec::new(),
601                    },
602                    context,
603                )
604                .await
605                .map_err(|error| CliError::ApiFailed(error.message))?;
606
607            render_response(
608                "knowledge.ingest",
609                response.metadata.clone(),
610                &response,
611                format,
612            )
613        }
614        CliAction::Query {
615            query,
616            source_scope,
617            limit,
618            freshness,
619        } => {
620            let response = service
621                .retrieve_context(
622                    HybridRetrievalRequest {
623                        query,
624                        source_scope,
625                        limit,
626                        freshness,
627                    },
628                    context,
629                )
630                .await
631                .map_err(|error| CliError::ApiFailed(error.message))?;
632
633            render_response(
634                "knowledge.retrieve_context",
635                response.metadata.clone(),
636                &response,
637                format,
638            )
639        }
640        CliAction::GraphInspect => {
641            let response = service
642                .inspect_graph(GraphInspectionRequest { source_scope: None }, context)
643                .await
644                .map_err(|error| CliError::ApiFailed(error.message))?;
645
646            render_response(
647                "graph.inspect",
648                response.metadata.clone(),
649                &response,
650                format,
651            )
652        }
653        CliAction::IndexRefresh { kinds } => {
654            let response = service
655                .refresh_indexes(IndexRefreshRequest { kinds }, context)
656                .await
657                .map_err(|error| CliError::ApiFailed(error.message))?;
658
659            render_response(
660                "index.refresh",
661                response.metadata.clone(),
662                &response,
663                format,
664            )
665        }
666        CliAction::Repo(command) => repo_cli::run_repo(service, command, context, format).await,
667        CliAction::Health => {
668            let response = service
669                .health(context)
670                .await
671                .map_err(|error| CliError::ApiFailed(error.message))?;
672
673            render_response(
674                "service.health",
675                response.metadata.clone(),
676                &response,
677                format,
678            )
679        }
680        CliAction::ProviderProbe => {
681            let response = service
682                .probe_embedding_provider(context)
683                .await
684                .map_err(|error| CliError::ApiFailed(error.message))?;
685
686            render_response(
687                "provider.embedding.probe",
688                response.metadata.clone(),
689                &response,
690                format,
691            )
692        }
693        CliAction::VersionCheck => version_cli::run_version_check(service, format).await,
694        CliAction::ServiceRun { .. } => Err(CliError::ServiceRunFailed(
695            "service run requires process runtime".to_owned(),
696        )),
697        CliAction::Help { path } => cli_spec::render_help(&path, format),
698        CliAction::WorkerStatus { .. }
699        | CliAction::FilesIndex { .. }
700        | CliAction::FilesQuery { .. }
701        | CliAction::WorkerRunOnce { .. }
702        | CliAction::ProposalList { .. }
703        | CliAction::ProposalShow { .. }
704        | CliAction::ProposalAccept { .. }
705        | CliAction::ProposalReject { .. }
706        | CliAction::ProposalSupersede { .. }
707        | CliAction::AuditQuery { .. }
708        | CliAction::ServiceStatus
709        | CliAction::ServicePlan { .. }
710        | CliAction::ServiceDefinitionWrite
711        | CliAction::ServiceOperatorStatus
712        | CliAction::ServiceOperatorPause
713        | CliAction::ServiceOperatorResume
714        | CliAction::SetupDoctor
715        | CliAction::SetupProfile { .. } => Err(CliError::ApiFailed(
716            "operational command was not handled by the service adapter".to_owned(),
717        )),
718        CliAction::Version => version_cli::render_version(command.format),
719    }
720}
721
722fn parse_action(tokens: Vec<String>) -> Result<CliAction, CliError> {
723    if tokens.is_empty() || tokens == ["status"] {
724        return Ok(CliAction::Status);
725    }
726
727    match tokens[0].as_str() {
728        "status" => Err(CliError::UnexpectedArgument(
729            tokens
730                .get(1)
731                .cloned()
732                .unwrap_or_else(|| "status".to_owned()),
733        )),
734        "ingest" => knowledge_cli::parse_ingest(&tokens[1..]),
735        "query" => knowledge_cli::parse_query(&tokens[1..]),
736        "files" => files_cli::parse_files(&tokens[1..]),
737        "repo" => repo_cli::parse_repo(&tokens[1..]).map(CliAction::Repo),
738        "graph" => knowledge_cli::parse_graph(&tokens[1..]),
739        "index" => knowledge_cli::parse_index(&tokens[1..]),
740        "worker" => ops_cli::parse_worker(&tokens[1..]),
741        "proposal" => ops_cli::parse_proposal(&tokens[1..]),
742        "audit" => ops_cli::parse_audit(&tokens[1..]),
743        "provider" => parse_provider(&tokens[1..]),
744        "health" if tokens.len() == 1 => Ok(CliAction::Health),
745        "service" => ops_cli::parse_service(&tokens[1..]),
746        "setup" => setup_cli::parse_setup(&tokens[1..]),
747        "version" if tokens.len() == 1 => Ok(CliAction::Version),
748        "version" if tokens == ["version", "check"] => Ok(CliAction::VersionCheck),
749        "help" => Ok(CliAction::Help {
750            path: help_path(tokens[1..].to_vec()),
751        }),
752        other => Err(CliError::UnexpectedArgument(other.to_owned())),
753    }
754}
755
756fn help_path(tokens: Vec<String>) -> Vec<String> {
757    tokens
758        .into_iter()
759        .filter(|token| token != "--")
760        .filter(|token| !token.starts_with('-'))
761        .collect()
762}
763
764fn parse_provider(tokens: &[String]) -> Result<CliAction, CliError> {
765    if tokens == ["probe"] {
766        return Ok(CliAction::ProviderProbe);
767    }
768
769    Err(CliError::UnexpectedArgument(
770        tokens
771            .first()
772            .cloned()
773            .unwrap_or_else(|| "provider".to_owned()),
774    ))
775}
776
777pub(super) fn value_after(
778    tokens: &[String],
779    index: usize,
780    flag: &'static str,
781) -> Result<String, CliError> {
782    tokens
783        .get(index + 1)
784        .cloned()
785        .ok_or(CliError::MissingValue(flag))
786}
787
788pub(super) fn parse_freshness(value: &str) -> Result<FreshnessPolicy, CliError> {
789    match value {
790        "allow-stale" => Ok(FreshnessPolicy::AllowStale),
791        "wait-until-fresh" => Ok(FreshnessPolicy::WaitUntilFresh),
792        "graph-only" => Ok(FreshnessPolicy::GraphOnly),
793        other => Err(CliError::InvalidFreshness(other.to_owned())),
794    }
795}
796
797async fn run_service(mcp: ServiceMcpTransport, web_enabled: bool) -> Result<String, CliError> {
798    let mut runtime = RuntimeConfiguration::from_process_environment()
799        .await
800        .map_err(|error| CliError::RuntimeConfigFailed(error.to_string()))?;
801    if mcp == ServiceMcpTransport::StreamableHttp {
802        runtime.agent = runtime.agent.clone().with_streamable_http_enabled();
803    }
804    runtime.observability.initialize();
805
806    let service = RelayKnowledgeService::new(runtime.clone());
807    service
808        .reconcile_startup_indexes(RequestContext::for_interface(InterfaceKind::Cli))
809        .await
810        .map_err(|error| CliError::ServiceRunFailed(error.message))?;
811    let (file_index_shutdown, file_index_shutdown_receiver) = tokio::sync::watch::channel(false);
812    let file_index_task = if runtime.file_index.enabled {
813        Some(tokio::spawn(files_cli::run_file_index_loop(
814            service.clone(),
815            runtime.file_index.scan_interval,
816            file_index_shutdown_receiver,
817        )))
818    } else {
819        None
820    };
821    let (code_index_shutdown, code_index_shutdown_receiver) = tokio::sync::watch::channel(false);
822    let code_index_task = tokio::spawn(run_code_index_loop(
823        service.clone(),
824        std::time::Duration::from_secs(5),
825        code_index_shutdown_receiver,
826    ));
827    if web_enabled {
828        let network_config = runtime.network.current();
829        ensure_web_remote_bind_allowed(
830            &network_config.http,
831            runtime.agent.access_policy.allow_remote_clients,
832        )?;
833        let mut router = web::router(service.clone(), network_config.http.max_request_body_bytes);
834        if runtime.agent.mcp_streamable_http_enabled {
835            let mcp_router = McpServer::new(
836                service.clone(),
837                runtime.network.clone(),
838                runtime.agent.clone(),
839            )
840            .checked_router()
841            .map_err(|error| CliError::ServiceRunFailed(error.to_string()))?;
842            router = router.merge(mcp_router);
843        }
844        crate::net::http::serve_router_with_qos(
845            router,
846            network_config.http,
847            QosRuntime::default(),
848            network_config.qos,
849            service_shutdown_signal(),
850        )
851        .await
852        .map_err(|error| CliError::ServiceRunFailed(error.to_string()))?;
853    } else if runtime.agent.mcp_streamable_http_enabled {
854        let server = McpServer::new(service, runtime.network.clone(), runtime.agent.clone());
855        server
856            .serve_until_shutdown(service_shutdown_signal())
857            .await
858            .map_err(|error| CliError::ServiceRunFailed(error.to_string()))?;
859    } else {
860        service_shutdown_signal().await;
861    }
862    if let Some(task) = file_index_task {
863        let _ = file_index_shutdown.send(true);
864        let _ = task.await;
865    }
866    let _ = code_index_shutdown.send(true);
867    let _ = code_index_task.await;
868    runtime.observability.shutdown();
869
870    Ok(String::new())
871}
872
873async fn run_code_index_loop(
874    service: RelayKnowledgeService,
875    interval: std::time::Duration,
876    mut shutdown: tokio::sync::watch::Receiver<bool>,
877) {
878    loop {
879        if *shutdown.borrow() {
880            break;
881        }
882        let context = RequestContext::for_interface(InterfaceKind::Cli);
883        if let Ok(Some(_)) = service.run_code_index_task_once(None, context).await {
884            continue;
885        }
886        tokio::select! {
887            _ = shutdown.changed() => {
888                if *shutdown.borrow() {
889                    break;
890                }
891            }
892            _ = tokio::time::sleep(interval) => {}
893        }
894    }
895}
896
897fn ensure_web_remote_bind_allowed(
898    config: &crate::net::http::HttpConfig,
899    allow_remote_clients: bool,
900) -> Result<(), CliError> {
901    if crate::net::http::remote_clients_allowed(config, allow_remote_clients) {
902        Ok(())
903    } else {
904        Err(CliError::ServiceRunFailed(
905            "Web remote bind requires allow_remote_clients=true".to_owned(),
906        ))
907    }
908}
909
910async fn service_shutdown_signal() {
911    #[cfg(unix)]
912    {
913        use tokio::signal::unix::{SignalKind, signal};
914
915        match signal(SignalKind::terminate()) {
916            Ok(mut terminate) => {
917                tokio::select! {
918                    _ = tokio::signal::ctrl_c() => {}
919                    _ = terminate.recv() => {}
920                }
921            }
922            Err(_) => {
923                let _ = tokio::signal::ctrl_c().await;
924            }
925        }
926    }
927
928    #[cfg(not(unix))]
929    {
930        let _ = tokio::signal::ctrl_c().await;
931    }
932}
933
934#[cfg(test)]
935#[path = "cli_naming_tests.rs"]
936mod cli_naming_tests;
937
938#[cfg(test)]
939#[path = "cli_tests.rs"]
940mod cli_tests;
941
942#[cfg(test)]
943#[path = "cli_version_tests.rs"]
944mod cli_version_tests;
relay_knowledge/interfaces/cli.rs

relay_knowledge/interfaces/
cli.rs
