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

relay_knowledge/interfaces/
cli.rs
