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

relay_knowledge/interfaces/
cli.rs
