relay_knowledge/api/
stream.rs

1use serde::{Deserialize, Serialize};
2
3use super::{ApiMetadata, ErrorKind, ProjectStatusResponse, RuntimeStatus};
4
5/// Stream event categories for newline-delimited JSON output.
6#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
7#[serde(rename_all = "kebab-case")]
8pub enum StreamEventKind {
9    Started,
10    Progress,
11    Item,
12    Completed,
13    Failed,
14}
15
16/// A single streaming API event. Each serialized event is one NDJSON line.
17#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
18pub struct ApiStreamEvent {
19    pub event: StreamEventKind,
20    pub operation: String,
21    #[serde(skip_serializing_if = "Option::is_none")]
22    pub message: Option<String>,
23    #[serde(skip_serializing_if = "Option::is_none")]
24    pub project_name: Option<String>,
25    #[serde(skip_serializing_if = "Option::is_none")]
26    pub runtime: Option<RuntimeStatus>,
27    #[serde(skip_serializing_if = "Option::is_none")]
28    pub payload: Option<serde_json::Value>,
29    #[serde(skip_serializing_if = "Option::is_none")]
30    pub error_kind: Option<ErrorKind>,
31    #[serde(skip_serializing_if = "Option::is_none")]
32    pub metadata: Option<ApiMetadata>,
33}
34
35impl ApiStreamEvent {
36    /// Creates a stream event for the project status operation.
37    pub fn project_status(
38        event: StreamEventKind,
39        response: &ProjectStatusResponse,
40        message: Option<&str>,
41    ) -> Self {
42        Self {
43            event,
44            operation: "project.status".to_owned(),
45            message: message.map(str::to_owned),
46            project_name: (event == StreamEventKind::Item).then(|| response.project_name.clone()),
47            runtime: (event == StreamEventKind::Item).then(|| response.runtime.clone()),
48            payload: None,
49            error_kind: None,
50            metadata: Some(response.metadata.clone()),
51        }
52    }
53
54    /// Creates a generic streaming event for non-status operations.
55    pub fn operation(
56        event: StreamEventKind,
57        operation: impl Into<String>,
58        metadata: ApiMetadata,
59        message: Option<&str>,
60        payload: Option<serde_json::Value>,
61    ) -> Self {
62        Self {
63            event,
64            operation: operation.into(),
65            message: message.map(str::to_owned),
66            project_name: None,
67            runtime: None,
68            payload,
69            error_kind: None,
70            metadata: Some(metadata),
71        }
72    }
73}
74
75#[cfg(test)]
76mod tests {
77    use serde_json::json;
78
79    use super::*;
80    use crate::{
81        api::{InterfaceKind, RequestContext},
82        domain::GraphVersion,
83    };
84
85    #[test]
86    fn project_status_event_only_attaches_payload_to_item() {
87        let context = RequestContext::with_ids(InterfaceKind::Cli, "req", "trace");
88        let response = ProjectStatusResponse {
89            project_name: "relay-knowledge".to_owned(),
90            metadata: ApiMetadata::graph_only(&context, GraphVersion::ZERO),
91            runtime: RuntimeStatus {
92                config_dir: "/config".to_owned(),
93                data_dir: "/data".to_owned(),
94                state_dir: "/state".to_owned(),
95                cache_dir: "/cache".to_owned(),
96                log_dir: "/logs".to_owned(),
97                temp_dir: "/tmp".to_owned(),
98                runtime_dir: "/run".to_owned(),
99                service_dir: "/service".to_owned(),
100                http_bind: "127.0.0.1:8791".to_owned(),
101                http_request_timeout_ms: 30000,
102                http_graceful_shutdown_timeout_ms: 10000,
103                http_max_request_body_bytes: 1024,
104                http_proxy_configured: false,
105                http_no_proxy_rules: 0,
106                http_ssl_verify: true,
107                qos_max_connections: 1,
108                qos_max_in_flight_requests: 1,
109                qos_max_queue_depth: 1,
110                worker_embedding_endpoint_configured: false,
111                worker_ocr_endpoint_configured: false,
112                worker_vision_endpoint_configured: false,
113                worker_extractor_endpoint_configured: false,
114                worker_max_in_flight: 2,
115                silent_updates_enabled: false,
116                file_index_enabled: false,
117                file_index_root_count: 0,
118                file_index_max_depth: 32,
119                file_index_max_file_bytes: 512 * 1024 * 1024,
120                file_index_scan_interval_ms: 900_000,
121                file_index_scan_timeout_ms: 300_000,
122                file_index_max_files_per_root: 50_000,
123                file_query_timeout_ms: 750,
124                semantic_backend_mode: "local".to_owned(),
125                vector_backend_mode: "local".to_owned(),
126                rerank_backend_mode: "local".to_owned(),
127                rerank_model: Some("relay-local-deterministic-rerank-v1".to_owned()),
128                rerank_candidate_multiplier: 4,
129                rerank_max_candidates: 64,
130                rerank_timeout_ms: 100,
131                embedding_provider: None,
132                embedding_base_url: None,
133                embedding_api_key_configured: false,
134                text_embedding_model: "relay-local-hash-ann-v1".to_owned(),
135                image_embedding_model: "relay-local-image-hash-v1".to_owned(),
136                embedding_dimension: 16,
137                embedding_batch_size: None,
138                embedding_timeout_ms: None,
139                embedding_max_concurrency: None,
140                model_profiles: crate::model_provider::ModelProfileRuntimeSummary {
141                    loaded: true,
142                    profile_count: 0,
143                    default_profile: None,
144                    error: None,
145                },
146                telemetry: crate::observability::ObservabilityRuntime::new(
147                    crate::observability::TelemetryConfig::from_environment(
148                        &crate::env::TelemetryEnvOverrides::default(),
149                    ),
150                )
151                .status(),
152            },
153        };
154
155        let started =
156            ApiStreamEvent::project_status(StreamEventKind::Started, &response, Some("starting"));
157        let item = ApiStreamEvent::project_status(StreamEventKind::Item, &response, None);
158
159        assert_eq!(started.project_name, None);
160        assert_eq!(started.message, Some("starting".to_owned()));
161        assert_eq!(item.project_name, Some("relay-knowledge".to_owned()));
162        assert!(item.runtime.is_some());
163    }
164
165    #[test]
166    fn operation_event_carries_generic_payload() {
167        let context = RequestContext::with_ids(InterfaceKind::Api, "req", "trace");
168        let metadata = ApiMetadata::graph_only(&context, GraphVersion::ZERO);
169
170        let event = ApiStreamEvent::operation(
171            StreamEventKind::Item,
172            "health",
173            metadata,
174            None,
175            Some(json!({"healthy": true})),
176        );
177
178        assert_eq!(event.operation, "health");
179        assert_eq!(event.payload, Some(json!({"healthy": true})));
180    }
181}
relay_knowledge/api/stream.rs

relay_knowledge/api/
stream.rs
