Skip to main content

ignition_core/actions/
logs.rs

1//! Log actions (02-04, HLTH-03/04): the poll-based TAIL loop — serde
2//! models and a sink OUT, no printing (ARCHITECTURE.md layering: the
3//! Phase-6 TUI rides this same layer).
4//!
5//! There is NO server push for gateway logs (02-RESEARCH Don't-Hand-Roll
6//! table): `GET /logs?startTime=<epoch-ms>` IS the tail primitive. The
7//! loop polls through the shared [`crate::poll`] engine (×1.5 adaptive
8//! backoff, Network/GatewayRestarting retried, Auth never) — the same
9//! engine 02-05's `wait` reuses.
10//!
11//! Cursor semantics (plan key_link): start at `since` (or 0 = the
12//! whole buffer); every page advances the cursor to the max timestamp
13//! seen; the next query sends `startTime = cursor + 1` — no overlap,
14//! no gaps. Entries are sorted client-side so the stream order is
15//! timestamp order regardless of the server's page ordering.
16//!
17//! `deadline: None` = run until Ctrl-C (the process default kill —
18//! research: keep Ctrl-C simple, README-documented); `Some(d)` ends
19//! GRACEFULLY: the poll's deadline expiry maps to `Ok` (exit 0 — the
20//! entries already streamed through the sink).
21
22use std::path::Path;
23use std::sync::Mutex;
24use std::time::{Duration, SystemTime, UNIX_EPOCH};
25
26use serde::Serialize;
27
28use crate::client::logs::{LogDownload, LogEntry, LogQuery, LoggerInfo};
29// Public re-export: the list action's return type is the wire-faithful
30// page — callers (and ActionOutput) name it via the action module.
31use crate::client::GatewayApi;
32pub use crate::client::logs::LogPage;
33use crate::client::query::ListEnvelope;
34use crate::error::CoreError;
35use crate::poll::{self, PollConfig, PollState};
36
37/// `ign logs download` output model.
38#[derive(Debug, Serialize)]
39pub struct DownloadResult {
40    /// Path of the file written.
41    pub file: String,
42    /// Bytes written.
43    pub bytes: usize,
44    /// Response content type (`application/x-sqlite3` — verified).
45    #[serde(skip_serializing_if = "Option::is_none")]
46    pub content_type: Option<String>,
47}
48
49/// `ign logs loggers set` output model.
50#[derive(Debug, Serialize)]
51pub struct SetLevelResult {
52    /// The logger that was changed.
53    pub logger: String,
54    /// The level it now carries (uppercase wire form).
55    pub level: String,
56}
57
58/// `ign logs loggers reset` output model.
59#[derive(Debug, Serialize)]
60pub struct ResetResult {
61    /// Always `true` — the reset reset every custom level.
62    pub reset: bool,
63}
64
65/// The logger registry page (`ign logs loggers`).
66pub type LoggersEnvelope = ListEnvelope<LoggerInfo>;
67
68/// `ign logs` (no `--follow`): newest entries first. The query always
69/// sorts `desc(timestamp)` and carries an explicit `limit` — together
70/// they make "the recent log entries" (must-have truth #1) without a
71/// `--since` window guess: `--limit 200` = the NEWEST 200, not the
72/// oldest.
73pub async fn list_logs(
74    api: &dyn GatewayApi,
75    logger: Option<&str>,
76    min_level: Option<&str>,
77    since_ms: Option<i64>,
78    limit: i64,
79) -> Result<LogPage, CoreError> {
80    let query = LogQuery {
81        start_time: since_ms,
82        logger: logger.map(str::to_string),
83        min_level: min_level.map(str::to_string),
84        limit,
85        sort_by: Some("desc(timestamp)".to_string()),
86        ..LogQuery::default()
87    };
88    api.logs(&query).await
89}
90
91/// `ign logs loggers`: the logger registry, explicit `limit` (must-have
92/// truth #5 — even the registry never rides the unlimited default),
93/// optional substring `search`.
94pub async fn loggers(
95    api: &dyn GatewayApi,
96    search: Option<&str>,
97) -> Result<ListEnvelope<LoggerInfo>, CoreError> {
98    let query = crate::client::query::ListQuery {
99        limit: crate::client::logs::DEFAULT_LOG_LIMIT,
100        search: search.map(str::to_string),
101        ..Default::default()
102    };
103    api.loggers(&query).await
104}
105
106/// `ign logs loggers set <name> <LEVEL>`. Confirmation guarding belongs
107/// to the CALLER (the CLI refuses without `--yes` before any API
108/// construction) — the action is the obedient arm.
109pub async fn set_logger_level(
110    api: &dyn GatewayApi,
111    logger: &str,
112    level: &str,
113) -> Result<SetLevelResult, CoreError> {
114    api.set_logger_level(logger, level).await?;
115    Ok(SetLevelResult {
116        logger: logger.to_string(),
117        level: level.to_string(),
118    })
119}
120
121/// `ign logs loggers reset` — same guard contract as set.
122pub async fn reset_logger_levels(api: &dyn GatewayApi) -> Result<ResetResult, CoreError> {
123    api.reset_logger_levels().await?;
124    Ok(ResetResult { reset: true })
125}
126
127/// The download filename: `-o FILE` wins, then the gateway's
128/// `Content-Disposition` name, then `<stem>-logs-<unix_ts>.idb` — NEVER
129/// `.zip` (Pitfall 7: the archive is SQLite).
130fn download_filename(
131    output: Option<&Path>,
132    download: &LogDownload,
133    stem: &str,
134    now_secs: i64,
135) -> String {
136    if let Some(output) = output {
137        return output.display().to_string();
138    }
139    if let Some(filename) = download.filename.as_deref().filter(|name| !name.is_empty()) {
140        return filename.to_string();
141    }
142    format!("{stem}-logs-{now_secs}.idb")
143}
144
145/// `ign logs download` — fetch the `.idb` archive and write it EXACTLY
146/// as received (no transformation, no extraction). `stem` names the
147/// gateway for the fallback filename (the CLI passes the profile name).
148pub async fn download(
149    api: &dyn GatewayApi,
150    output: Option<&Path>,
151    fallback_stem: &str,
152) -> Result<DownloadResult, CoreError> {
153    let fetched = api.logs_download().await?;
154    let now_secs = SystemTime::now()
155        .duration_since(UNIX_EPOCH)
156        .map(|since| since.as_secs() as i64)
157        .unwrap_or_default();
158    let file = download_filename(output, &fetched, fallback_stem, now_secs);
159    std::fs::write(&file, &fetched.bytes)
160        .map_err(|err| CoreError::Internal(format!("cannot write log archive {file}: {err}")))?;
161    Ok(DownloadResult {
162        bytes: fetched.bytes.len(),
163        content_type: fetched.content_type,
164        file,
165    })
166}
167
168/// Parse `--since`: an absolute `EPOCH_MS` value or a relative span
169/// `Nms` / `Ns` / `Nmin` / `Nh` (resolved against `now_ms`). Returned
170/// as a plain `String` error so clap can surface it as a usage-class
171/// parse failure (exit 2) — validation happens at arg-parse time, not
172/// deep in dispatch.
173pub fn parse_since(spec: &str, now_ms: i64) -> Result<i64, String> {
174    let spec = spec.trim();
175    // Order matters: "ms" before bare "s"; "min" between them.
176    for (suffix, unit_ms) in [
177        ("ms", 1_i64),
178        ("min", 60_000),
179        ("h", 3_600_000),
180        ("s", 1_000),
181    ] {
182        if let Some(digits) = spec.strip_suffix(suffix)
183            && let Ok(count) = digits.parse::<i64>()
184            && count >= 0
185        {
186            return Ok(now_ms - count * unit_ms);
187        }
188    }
189    // Absolute epoch-ms (also covers "0" = the whole buffer).
190    match spec.parse::<i64>() {
191        Ok(epoch_ms) if epoch_ms >= 0 => Ok(epoch_ms),
192        _ => Err(format!(
193            "invalid --since {spec:?}: expected EPOCH-MS or a relative span like 500ms, 30s, 5min, 2h"
194        )),
195    }
196}
197
198/// `ign logs --follow` result — how much streamed before the tail
199/// ended. (Ctrl-C never reaches here: the process default kill emits
200/// no envelope at all, README-documented.)
201#[derive(Debug, Default, Serialize)]
202pub struct TailResult {
203    /// Entries delivered to the sink.
204    pub streamed: usize,
205}
206
207/// The tail loop's probe scratch: the cursor (epoch ms) and the sink.
208/// Owned by [`poll`], lent fresh to every probe call.
209struct TailState<'a> {
210    /// Max timestamp delivered so far (-1 before the first page).
211    cursor: i64,
212    /// Receives each entry as it arrives (the action stays
213    /// printer-free — the dispatch owns stdout). `+ Send` so the tail
214    /// future can cross `tokio::spawn` on the multi-thread runtime
215    /// (06-01: the TUI's logs worker; the rig.rs sinks set this
216    /// convention first).
217    sink: &'a mut (dyn FnMut(&LogEntry) + Send),
218}
219
220/// Stream new log entries to `sink` as they arrive. The action is
221/// printer-free — the dispatch owns stdout (human lines or NDJSON).
222///
223/// Every query carries an explicit limit ([`LogQuery::default`] —
224/// Pitfall 9); a page larger than the limit still advances the cursor
225/// correctly (cursor = max timestamp seen, so the next poll resumes
226/// exactly past it).
227pub async fn tail(
228    api: &dyn GatewayApi,
229    logger: Option<&str>,
230    min_level: Option<&str>,
231    since_ms: Option<i64>,
232    interval: Duration,
233    deadline: Option<Duration>,
234    sink: &mut (dyn FnMut(&LogEntry) + Send),
235) -> Result<TailResult, CoreError> {
236    // -1 so the FIRST query's start_time = cursor + 1 = since exactly
237    // (or 0 when no --since — the whole buffer).
238    let state = TailState {
239        cursor: since_ms.unwrap_or(0) - 1,
240        sink,
241    };
242    // The stream count lives OUTSIDE the poll call (the probe bumps it
243    // through a shared borrow; poll consumes the state). A `Mutex` (not
244    // `Cell`) so the probe future is Send — the 06-02 TUI spawns tails.
245    let streamed = Mutex::new(0usize);
246
247    let cfg = PollConfig {
248        subject: "log tail (GET /data/api/v1/logs)".to_string(),
249        interval,
250        deadline: deadline.unwrap_or(Duration::MAX),
251        ..PollConfig::default()
252    };
253
254    let outcome = poll::poll(cfg, state, |state| {
255        Box::pin(async {
256            let query = LogQuery {
257                start_time: Some(state.cursor + 1),
258                logger: logger.map(str::to_string),
259                min_level: min_level.map(str::to_string),
260                ..LogQuery::default()
261            };
262            let page = api.logs(&query).await?;
263            let mut entries = page.items;
264            // Timestamp order regardless of server page ordering.
265            entries.sort_by_key(|entry| entry.timestamp);
266            let observation = entries
267                .last()
268                .map(|last| format!("{} entries, latest at {}", entries.len(), last.timestamp));
269            for entry in &entries {
270                (state.sink)(entry);
271            }
272            if let Some(last) = entries.last() {
273                state.cursor = last.timestamp;
274            }
275            *streamed.lock().expect("streamed count") += entries.len();
276            Ok(PollState::<()>::Pending(observation))
277        })
278    })
279    .await;
280
281    match outcome {
282        // The probe never reports Done (T = ()) — the only Ok is
283        // unreachable; kept for match totality.
284        Ok(()) => Ok(TailResult {
285            streamed: *streamed.lock().expect("streamed count"),
286        }),
287        // Deadline expiry = GRACEFUL end (exit 0): poll retries genuine
288        // Network errors until the deadline, so a None-source Network
289        // error IS the timeout. The entries already streamed.
290        Err(CoreError::Network { source: None, .. }) => Ok(TailResult {
291            streamed: *streamed.lock().expect("streamed count"),
292        }),
293        Err(err) => Err(err),
294    }
295}
296
297#[cfg(test)]
298mod tests {
299    use std::sync::Mutex;
300    use std::time::Duration;
301
302    use super::{download_filename, parse_since, tail};
303    use crate::client::GatewayApi;
304    use crate::client::logs::{LogDownload, LogEntry, LogQuery};
305    use crate::client::query::{ListEnvelope, ListMetadata};
306    use crate::error::CoreError;
307
308    /// `--since` accepts EPOCH-MS and every relative suffix, parsed
309    /// against a fixed now; junk is a usage-class String error.
310    #[test]
311    fn parse_since_accepts_epoch_and_relative_spans() {
312        const NOW: i64 = 1_787_346_747_022;
313        assert_eq!(parse_since("1787346747022", NOW), Ok(1787346747022));
314        assert_eq!(parse_since("0", NOW), Ok(0));
315        assert_eq!(parse_since("500ms", NOW), Ok(NOW - 500));
316        assert_eq!(parse_since("30s", NOW), Ok(NOW - 30_000));
317        assert_eq!(parse_since("5min", NOW), Ok(NOW - 300_000));
318        assert_eq!(parse_since("2h", NOW), Ok(NOW - 7_200_000));
319        // suffix order matters: "ms" before bare "s"
320        assert_eq!(parse_since("1s", NOW), Ok(NOW - 1_000));
321        assert!(parse_since("banana", NOW).is_err());
322        assert!(
323            parse_since("-5s", NOW).is_err(),
324            "negative spans are invalid"
325        );
326        assert!(parse_since("", NOW).is_err());
327    }
328
329    /// Filename precedence: `-o FILE` > Content-Disposition > the
330    /// `<stem>-logs-<ts>.idb` fallback — and NEVER a `.zip` (Pitfall 7).
331    #[test]
332    fn download_filename_precedence() {
333        let fetched = LogDownload {
334            bytes: Vec::new(),
335            filename: Some("GW_Ignition_logs_20260822-0307.idb".into()),
336            content_type: Some("application/x-sqlite3".into()),
337        };
338        assert_eq!(
339            download_filename(
340                Some(std::path::Path::new("/tmp/out.idb")),
341                &fetched,
342                "dev",
343                1000
344            ),
345            "/tmp/out.idb",
346            "-o wins"
347        );
348        assert_eq!(
349            download_filename(None, &fetched, "dev", 1000),
350            "GW_Ignition_logs_20260822-0307.idb",
351            "Content-Disposition name second"
352        );
353        let anonymous = LogDownload {
354            bytes: Vec::new(),
355            filename: None,
356            content_type: None,
357        };
358        assert_eq!(
359            download_filename(None, &anonymous, "dev", 1_787_346_747),
360            "dev-logs-1787346747.idb",
361            "fallback = <stem>-logs-<unix_ts>.idb — never .zip"
362        );
363    }
364
365    /// A scripted double: serves `pages` in order (then empty pages
366    /// forever) and records every query it saw.
367    #[derive(Default)]
368    struct TailRig {
369        pages: Mutex<std::collections::VecDeque<Vec<LogEntry>>>,
370        queries: Mutex<Vec<LogQuery>>,
371    }
372
373    fn entry(timestamp: i64, message: &str) -> LogEntry {
374        LogEntry {
375            timestamp,
376            logger_name: "GatewayManager".into(),
377            level: "INFO".into(),
378            message: message.into(),
379            stack: Vec::new(),
380            mdc: Default::default(),
381            extra: Default::default(),
382        }
383    }
384
385    #[async_trait::async_trait]
386    impl GatewayApi for TailRig {
387        async fn bundle_generate(
388            &self,
389        ) -> Result<crate::client::diagnostics::BundleStatusWire, CoreError> {
390            unreachable!("not part of this action")
391        }
392        async fn bundle_status(
393            &self,
394        ) -> Result<crate::client::diagnostics::BundleStatusWire, CoreError> {
395            unreachable!("not part of this action")
396        }
397        async fn bundle_download(
398            &self,
399            _out: &std::path::Path,
400        ) -> Result<crate::client::projects::ExportMeta, CoreError> {
401            unreachable!("not part of this action")
402        }
403        async fn tag_provider_list(
404            &self,
405            _query: &crate::client::query::ListQuery,
406        ) -> Result<
407            crate::client::query::ListEnvelope<crate::client::tags::TagProviderRecord>,
408            CoreError,
409        > {
410            unreachable!("not part of this action")
411        }
412        async fn tag_provider_find(
413            &self,
414            _name: &str,
415        ) -> Result<crate::client::tags::TagProviderRecord, CoreError> {
416            unreachable!("not part of this action")
417        }
418        async fn tag_provider_create(
419            &self,
420            _body: &[crate::client::tags::TagProviderCreate],
421        ) -> Result<(), CoreError> {
422            unreachable!("not part of this action")
423        }
424        async fn tag_provider_delete(
425            &self,
426            _name: &str,
427            _signature: &str,
428        ) -> Result<(), CoreError> {
429            unreachable!("not part of this action")
430        }
431        async fn trial_status_wire(&self) -> Result<crate::client::trial::TrialWire, CoreError> {
432            unreachable!("not part of this action")
433        }
434        async fn banners(&self) -> Result<crate::client::trial::BannerSet, CoreError> {
435            unreachable!("not part of this action")
436        }
437        async fn trial_reset_wire(&self) -> Result<crate::client::trial::TrialWire, CoreError> {
438            unreachable!("not part of this action")
439        }
440        async fn backup_download(
441            &self,
442            _out: &std::path::Path,
443            _backup_type: crate::client::backup::BackupType,
444        ) -> Result<crate::client::projects::ExportMeta, CoreError> {
445            unreachable!("not part of this action")
446        }
447        async fn backup_restore(&self, _gwbk: &std::path::Path) -> Result<(), CoreError> {
448            unreachable!("not part of this action")
449        }
450        async fn eam_task_history(
451            &self,
452            _limit: Option<u32>,
453            _search: Option<&str>,
454        ) -> Result<crate::client::query::ListEnvelope<crate::client::eam::EamHistoryItem>, CoreError>
455        {
456            unreachable!("not part of this action")
457        }
458        async fn eam_task_definitions(
459            &self,
460        ) -> Result<crate::client::query::ListEnvelope<crate::client::eam::EamTaskRecord>, CoreError>
461        {
462            unreachable!("not part of this action")
463        }
464        async fn eam_task_find(
465            &self,
466            _name: &str,
467        ) -> Result<crate::client::eam::EamTaskRecord, CoreError> {
468            unreachable!("not part of this action")
469        }
470        async fn eam_task_create(&self, _definition: &serde_json::Value) -> Result<(), CoreError> {
471            unreachable!("not part of this action")
472        }
473        async fn eam_task_force(&self, _owner: &str, _name: &str) -> Result<(), CoreError> {
474            unreachable!("not part of this action")
475        }
476        async fn eam_task_suspend(&self, _name: &str) -> Result<(), CoreError> {
477            unreachable!("not part of this action")
478        }
479        async fn eam_task_resume(&self, _name: &str) -> Result<(), CoreError> {
480            unreachable!("not part of this action")
481        }
482        async fn eam_task_cancel(&self, _name: &str) -> Result<(), CoreError> {
483            unreachable!("not part of this action")
484        }
485        async fn eam_tasks_scheduled(
486            &self,
487            _running: bool,
488        ) -> Result<Vec<crate::client::eam::EamScheduledTask>, CoreError> {
489            unreachable!("not part of this action")
490        }
491        async fn eam_task_modify(
492            &self,
493            _definition: &serde_json::Value,
494        ) -> Result<Option<crate::client::eam::ModifyOutcome>, CoreError> {
495            unreachable!("not part of this action")
496        }
497        async fn eam_task_delete(
498            &self,
499            _name: &str,
500            _signature: &str,
501            _confirm: bool,
502        ) -> Result<crate::client::eam::DeleteOutcome, CoreError> {
503            unreachable!("not part of this action")
504        }
505        async fn api_call(
506            &self,
507            _call: &crate::client::apicall::ApiCallRequest,
508        ) -> Result<crate::client::apicall::ApiCallData, CoreError> {
509            unreachable!("not part of this action")
510        }
511        async fn license_status(
512            &self,
513        ) -> Result<crate::client::license::LicenseStatusWire, CoreError> {
514            unreachable!("not part of this action")
515        }
516        async fn redundancy_status(
517            &self,
518        ) -> Result<crate::client::redundancy::RedundancyStatusWire, CoreError> {
519            unreachable!("not part of this action")
520        }
521        async fn gan_status(&self) -> Result<crate::client::gan::GanStatusWire, CoreError> {
522            unreachable!("not part of this action")
523        }
524        async fn logs(&self, filter: &LogQuery) -> Result<ListEnvelope<LogEntry>, CoreError> {
525            self.queries.lock().unwrap().push(filter.clone());
526            let items = self.pages.lock().unwrap().pop_front().unwrap_or_default();
527            Ok(ListEnvelope {
528                metadata: ListMetadata {
529                    total: items.len() as i64,
530                    matching: items.len() as i64,
531                    limit: 200,
532                    offset: 0,
533                },
534                items,
535            })
536        }
537        async fn gateway_info(&self) -> Result<crate::client::version::GatewayInfo, CoreError> {
538            unreachable!("not part of this action")
539        }
540        async fn overview(&self) -> Result<crate::client::status::Overview, CoreError> {
541            unreachable!("not part of this action")
542        }
543        async fn status_ping(&self) -> Result<crate::client::status::StatusPing, CoreError> {
544            unreachable!("not part of this action")
545        }
546        async fn modules(
547            &self,
548            _quarantined: bool,
549            _query: &crate::client::query::ListQuery,
550        ) -> Result<ListEnvelope<crate::client::status::ModuleInfo>, CoreError> {
551            unreachable!("not part of this action")
552        }
553        async fn metrics_current(
554            &self,
555        ) -> Result<crate::client::metrics::CurrentGauges, CoreError> {
556            unreachable!("not part of this action")
557        }
558        async fn metrics_historic(
559            &self,
560        ) -> Result<crate::client::metrics::PerformanceCharts, CoreError> {
561            unreachable!("not part of this action")
562        }
563        async fn metrics_threads(&self) -> Result<crate::client::metrics::ThreadCounts, CoreError> {
564            unreachable!("not part of this action")
565        }
566        async fn designers(
567            &self,
568            _query: &crate::client::query::ListQuery,
569        ) -> Result<ListEnvelope<crate::client::sessions::DesignerInfo>, CoreError> {
570            unreachable!("not part of this action")
571        }
572        async fn perspective_sessions(
573            &self,
574            _query: &crate::client::query::ListQuery,
575        ) -> Result<ListEnvelope<crate::client::sessions::PerspectiveSession>, CoreError> {
576            unreachable!("not part of this action")
577        }
578        async fn vision_clients(
579            &self,
580            _query: &crate::client::query::ListQuery,
581        ) -> Result<ListEnvelope<crate::client::sessions::VisionClient>, CoreError> {
582            unreachable!("not part of this action")
583        }
584        async fn terminate_perspective_session(
585            &self,
586            _id: &str,
587            _message: Option<&str>,
588        ) -> Result<(), CoreError> {
589            unreachable!("not part of this action")
590        }
591        async fn terminate_vision_client(&self, _id: &str) -> Result<(), CoreError> {
592            unreachable!("not part of this action")
593        }
594        async fn prune_designer(&self, _id: &str) -> Result<(), CoreError> {
595            unreachable!("not part of this action")
596        }
597        async fn database_connections(
598            &self,
599        ) -> Result<ListEnvelope<crate::client::connections::GatewayConnection>, CoreError>
600        {
601            unreachable!("not part of this action")
602        }
603        async fn opc_connections(
604            &self,
605        ) -> Result<ListEnvelope<crate::client::connections::GatewayConnection>, CoreError>
606        {
607            unreachable!("not part of this action")
608        }
609        async fn logs_download(&self) -> Result<crate::client::logs::LogDownload, CoreError> {
610            unreachable!("not part of this action")
611        }
612        async fn loggers(
613            &self,
614            _query: &crate::client::query::ListQuery,
615        ) -> Result<ListEnvelope<crate::client::logs::LoggerInfo>, CoreError> {
616            unreachable!("not part of this action")
617        }
618        async fn set_logger_level(&self, _logger: &str, _level: &str) -> Result<(), CoreError> {
619            unreachable!("not part of this action")
620        }
621        async fn reset_logger_levels(&self) -> Result<(), CoreError> {
622            unreachable!("not part of this action")
623        }
624        async fn restart(&self) -> Result<(), CoreError> {
625            unreachable!("not part of this action")
626        }
627        async fn scan_projects(&self) -> Result<(), CoreError> {
628            unreachable!("not part of this action")
629        }
630        async fn security_properties(
631            &self,
632        ) -> Result<crate::client::restart::SecurityProperties, CoreError> {
633            unreachable!("not part of this action")
634        }
635        async fn webdev_route_status(&self, _route: &str) -> Result<u16, CoreError> {
636            unreachable!("not part of this action")
637        }
638        async fn webdev_route_call(
639            &self,
640            _project: &str,
641            _route: &str,
642            _body: &serde_json::Value,
643            _extra_headers: &[(&str, &str)],
644        ) -> Result<serde_json::Value, CoreError> {
645            unreachable!("not part of this action")
646        }
647        async fn webdev_route_probe(
648            &self,
649            _project: &str,
650            _route: &str,
651            _extra_headers: &[(&str, &str)],
652        ) -> Result<crate::client::webdev::RouteProbe, CoreError> {
653            unreachable!("not part of this action")
654        }
655        async fn projects(
656            &self,
657            _query: &crate::client::query::ListQuery,
658        ) -> Result<
659            crate::client::query::ListEnvelope<crate::client::projects::ProjectRecord>,
660            CoreError,
661        > {
662            unreachable!("not part of this action")
663        }
664        async fn project_find(
665            &self,
666            _name: &str,
667        ) -> Result<crate::client::projects::ProjectRecord, CoreError> {
668            unreachable!("not part of this action")
669        }
670        async fn project_create(
671            &self,
672            _body: &crate::client::projects::ProjectCreate,
673        ) -> Result<(), CoreError> {
674            unreachable!("not part of this action")
675        }
676        async fn project_copy(&self, _from: &str, _to: &str) -> Result<(), CoreError> {
677            unreachable!("not part of this action")
678        }
679        async fn project_rename(&self, _name: &str, _new_name: &str) -> Result<(), CoreError> {
680            unreachable!("not part of this action")
681        }
682        async fn project_modify(
683            &self,
684            _name: &str,
685            _body: &crate::client::projects::ProjectModify,
686        ) -> Result<(), CoreError> {
687            unreachable!("not part of this action")
688        }
689        async fn project_delete(&self, _name: &str) -> Result<(), CoreError> {
690            unreachable!("not part of this action")
691        }
692        async fn project_export_to_file(
693            &self,
694            _name: &str,
695            _out: &std::path::Path,
696        ) -> Result<crate::client::projects::ExportMeta, CoreError> {
697            unreachable!("not part of this action")
698        }
699        async fn project_import(
700            &self,
701            _name: &str,
702            _zip: Vec<u8>,
703            _overwrite: bool,
704        ) -> Result<crate::client::projects::ImportOutcome, CoreError> {
705            unreachable!("not part of this action")
706        }
707    }
708
709    /// Two pages then silence under a short deadline: entries arrive in
710    /// TIMESTAMP order through the sink, the cursor advances past each
711    /// page's max (next query's startTime = max + 1), and the deadline
712    /// expiry ends the tail CLEANLY (Ok, exit 0 semantics).
713    #[tokio::test]
714    async fn tail_streams_pages_in_order_and_ends_cleanly_on_deadline() {
715        let rig = TailRig {
716            pages: Mutex::new(
717                vec![
718                    // Deliberately out of order WITHIN the page: client-side
719                    // sort must fix the stream order.
720                    vec![entry(1010, "second"), entry(1005, "first")],
721                    vec![entry(1022, "third"), entry(1018, "wait, also")],
722                ]
723                .into(),
724            ),
725            queries: Mutex::new(Vec::new()),
726        };
727
728        let mut received: Vec<(i64, String)> = Vec::new();
729        let sink: &mut (dyn FnMut(&LogEntry) + Send) = &mut |entry: &LogEntry| {
730            received.push((entry.timestamp, entry.message.clone()));
731        };
732
733        let result = tail(
734            &rig,
735            None,
736            None,
737            Some(1000), // since → first query startTime = 1000
738            Duration::from_millis(5),
739            // Deadline budget generous on purpose: 40ms starved the
740            // second page under a full parallel workspace run (the
741            // rig serves pages from memory — only scheduler latency
742            // competes). 400ms gives ~10x headroom while keeping the
743            // isolated test under half a second.
744            Some(Duration::from_millis(400)),
745            sink,
746        )
747        .await
748        .expect("deadline expiry ends the tail cleanly");
749
750        // Entries delivered in timestamp order across BOTH pages.
751        assert_eq!(
752            received,
753            vec![
754                (1005, "first".into()),
755                (1010, "second".into()),
756                (1018, "wait, also".into()),
757                (1022, "third".into()),
758            ],
759            "stream order is timestamp order (client-side sort)"
760        );
761        assert_eq!(result.streamed, 4);
762
763        // Cursor discipline: first query starts at `since` exactly;
764        // after page 1 (max 1010) the next startTime = 1011; after
765        // page 2 (max 1022) the next startTime = 1023 (then silence).
766        let queries = rig.queries.lock().unwrap();
767        assert_eq!(queries[0].start_time, Some(1000), "first = since");
768        assert!(
769            queries.len() >= 3,
770            "polled again after each page: {}",
771            queries.len()
772        );
773        assert_eq!(queries[1].start_time, Some(1011), "cursor = max + 1");
774        assert!(
775            queries[2].start_time == Some(1023),
776            "cursor advanced past page 2: {:?}",
777            queries[2].start_time
778        );
779        // Every query carries the explicit limit (Pitfall 9).
780        assert!(queries.iter().all(|query| query.limit == 200));
781    }
782
783    /// Auth failures surface immediately — the tail never retries a
784    /// rejected token (the poll engine's never-retry rule, proven at
785    /// the action seam).
786    #[tokio::test]
787    async fn tail_fails_fast_on_auth() {
788        struct AuthRig;
789        #[async_trait::async_trait]
790        impl GatewayApi for AuthRig {
791            async fn bundle_generate(
792                &self,
793            ) -> Result<crate::client::diagnostics::BundleStatusWire, CoreError> {
794                unreachable!("not part of this action")
795            }
796            async fn bundle_status(
797                &self,
798            ) -> Result<crate::client::diagnostics::BundleStatusWire, CoreError> {
799                unreachable!("not part of this action")
800            }
801            async fn bundle_download(
802                &self,
803                _out: &std::path::Path,
804            ) -> Result<crate::client::projects::ExportMeta, CoreError> {
805                unreachable!("not part of this action")
806            }
807            async fn tag_provider_list(
808                &self,
809                _query: &crate::client::query::ListQuery,
810            ) -> Result<
811                crate::client::query::ListEnvelope<crate::client::tags::TagProviderRecord>,
812                CoreError,
813            > {
814                unreachable!("not part of this action")
815            }
816            async fn tag_provider_find(
817                &self,
818                _name: &str,
819            ) -> Result<crate::client::tags::TagProviderRecord, CoreError> {
820                unreachable!("not part of this action")
821            }
822            async fn tag_provider_create(
823                &self,
824                _body: &[crate::client::tags::TagProviderCreate],
825            ) -> Result<(), CoreError> {
826                unreachable!("not part of this action")
827            }
828            async fn tag_provider_delete(
829                &self,
830                _name: &str,
831                _signature: &str,
832            ) -> Result<(), CoreError> {
833                unreachable!("not part of this action")
834            }
835            async fn trial_status_wire(
836                &self,
837            ) -> Result<crate::client::trial::TrialWire, CoreError> {
838                unreachable!("not part of this action")
839            }
840            async fn banners(&self) -> Result<crate::client::trial::BannerSet, CoreError> {
841                unreachable!("not part of this action")
842            }
843            async fn trial_reset_wire(&self) -> Result<crate::client::trial::TrialWire, CoreError> {
844                unreachable!("not part of this action")
845            }
846            async fn backup_download(
847                &self,
848                _out: &std::path::Path,
849                _backup_type: crate::client::backup::BackupType,
850            ) -> Result<crate::client::projects::ExportMeta, CoreError> {
851                unreachable!("not part of this action")
852            }
853            async fn backup_restore(&self, _gwbk: &std::path::Path) -> Result<(), CoreError> {
854                unreachable!("not part of this action")
855            }
856            async fn eam_task_history(
857                &self,
858                _limit: Option<u32>,
859                _search: Option<&str>,
860            ) -> Result<
861                crate::client::query::ListEnvelope<crate::client::eam::EamHistoryItem>,
862                CoreError,
863            > {
864                unreachable!("not part of this action")
865            }
866            async fn eam_task_definitions(
867                &self,
868            ) -> Result<
869                crate::client::query::ListEnvelope<crate::client::eam::EamTaskRecord>,
870                CoreError,
871            > {
872                unreachable!("not part of this action")
873            }
874            async fn eam_task_find(
875                &self,
876                _name: &str,
877            ) -> Result<crate::client::eam::EamTaskRecord, CoreError> {
878                unreachable!("not part of this action")
879            }
880            async fn eam_task_create(
881                &self,
882                _definition: &serde_json::Value,
883            ) -> Result<(), CoreError> {
884                unreachable!("not part of this action")
885            }
886            async fn eam_task_force(&self, _owner: &str, _name: &str) -> Result<(), CoreError> {
887                unreachable!("not part of this action")
888            }
889            async fn eam_task_suspend(&self, _name: &str) -> Result<(), CoreError> {
890                unreachable!("not part of this action")
891            }
892            async fn eam_task_resume(&self, _name: &str) -> Result<(), CoreError> {
893                unreachable!("not part of this action")
894            }
895            async fn eam_task_cancel(&self, _name: &str) -> Result<(), CoreError> {
896                unreachable!("not part of this action")
897            }
898            async fn eam_tasks_scheduled(
899                &self,
900                _running: bool,
901            ) -> Result<Vec<crate::client::eam::EamScheduledTask>, CoreError> {
902                unreachable!("not part of this action")
903            }
904            async fn eam_task_modify(
905                &self,
906                _definition: &serde_json::Value,
907            ) -> Result<Option<crate::client::eam::ModifyOutcome>, CoreError> {
908                unreachable!("not part of this action")
909            }
910            async fn eam_task_delete(
911                &self,
912                _name: &str,
913                _signature: &str,
914                _confirm: bool,
915            ) -> Result<crate::client::eam::DeleteOutcome, CoreError> {
916                unreachable!("not part of this action")
917            }
918            async fn api_call(
919                &self,
920                _call: &crate::client::apicall::ApiCallRequest,
921            ) -> Result<crate::client::apicall::ApiCallData, CoreError> {
922                unreachable!("not part of this action")
923            }
924            async fn license_status(
925                &self,
926            ) -> Result<crate::client::license::LicenseStatusWire, CoreError> {
927                unreachable!("not part of this action")
928            }
929            async fn redundancy_status(
930                &self,
931            ) -> Result<crate::client::redundancy::RedundancyStatusWire, CoreError> {
932                unreachable!("not part of this action")
933            }
934            async fn gan_status(&self) -> Result<crate::client::gan::GanStatusWire, CoreError> {
935                unreachable!("not part of this action")
936            }
937            async fn logs(&self, _filter: &LogQuery) -> Result<ListEnvelope<LogEntry>, CoreError> {
938                Err(CoreError::Auth {
939                    status: 401,
940                    endpoint: Some("http://gw/data/api/v1/logs".into()),
941                })
942            }
943            async fn gateway_info(&self) -> Result<crate::client::version::GatewayInfo, CoreError> {
944                unreachable!("not part of this action")
945            }
946            async fn overview(&self) -> Result<crate::client::status::Overview, CoreError> {
947                unreachable!("not part of this action")
948            }
949            async fn status_ping(&self) -> Result<crate::client::status::StatusPing, CoreError> {
950                unreachable!("not part of this action")
951            }
952            async fn modules(
953                &self,
954                _quarantined: bool,
955                _query: &crate::client::query::ListQuery,
956            ) -> Result<ListEnvelope<crate::client::status::ModuleInfo>, CoreError> {
957                unreachable!("not part of this action")
958            }
959            async fn metrics_current(
960                &self,
961            ) -> Result<crate::client::metrics::CurrentGauges, CoreError> {
962                unreachable!("not part of this action")
963            }
964            async fn metrics_historic(
965                &self,
966            ) -> Result<crate::client::metrics::PerformanceCharts, CoreError> {
967                unreachable!("not part of this action")
968            }
969            async fn metrics_threads(
970                &self,
971            ) -> Result<crate::client::metrics::ThreadCounts, CoreError> {
972                unreachable!("not part of this action")
973            }
974            async fn designers(
975                &self,
976                _query: &crate::client::query::ListQuery,
977            ) -> Result<ListEnvelope<crate::client::sessions::DesignerInfo>, CoreError>
978            {
979                unreachable!("not part of this action")
980            }
981            async fn perspective_sessions(
982                &self,
983                _query: &crate::client::query::ListQuery,
984            ) -> Result<ListEnvelope<crate::client::sessions::PerspectiveSession>, CoreError>
985            {
986                unreachable!("not part of this action")
987            }
988            async fn vision_clients(
989                &self,
990                _query: &crate::client::query::ListQuery,
991            ) -> Result<ListEnvelope<crate::client::sessions::VisionClient>, CoreError>
992            {
993                unreachable!("not part of this action")
994            }
995            async fn terminate_perspective_session(
996                &self,
997                _id: &str,
998                _message: Option<&str>,
999            ) -> Result<(), CoreError> {
1000                unreachable!("not part of this action")
1001            }
1002            async fn terminate_vision_client(&self, _id: &str) -> Result<(), CoreError> {
1003                unreachable!("not part of this action")
1004            }
1005            async fn prune_designer(&self, _id: &str) -> Result<(), CoreError> {
1006                unreachable!("not part of this action")
1007            }
1008            async fn database_connections(
1009                &self,
1010            ) -> Result<ListEnvelope<crate::client::connections::GatewayConnection>, CoreError>
1011            {
1012                unreachable!("not part of this action")
1013            }
1014            async fn opc_connections(
1015                &self,
1016            ) -> Result<ListEnvelope<crate::client::connections::GatewayConnection>, CoreError>
1017            {
1018                unreachable!("not part of this action")
1019            }
1020            async fn logs_download(&self) -> Result<crate::client::logs::LogDownload, CoreError> {
1021                unreachable!("not part of this action")
1022            }
1023            async fn loggers(
1024                &self,
1025                _query: &crate::client::query::ListQuery,
1026            ) -> Result<ListEnvelope<crate::client::logs::LoggerInfo>, CoreError> {
1027                unreachable!("not part of this action")
1028            }
1029            async fn set_logger_level(&self, _logger: &str, _level: &str) -> Result<(), CoreError> {
1030                unreachable!("not part of this action")
1031            }
1032            async fn reset_logger_levels(&self) -> Result<(), CoreError> {
1033                unreachable!("not part of this action")
1034            }
1035            async fn restart(&self) -> Result<(), CoreError> {
1036                unreachable!("not part of this action")
1037            }
1038            async fn scan_projects(&self) -> Result<(), CoreError> {
1039                unreachable!("not part of this action")
1040            }
1041            async fn security_properties(
1042                &self,
1043            ) -> Result<crate::client::restart::SecurityProperties, CoreError> {
1044                unreachable!("not part of this action")
1045            }
1046            async fn webdev_route_status(&self, _route: &str) -> Result<u16, CoreError> {
1047                unreachable!("not part of this action")
1048            }
1049            async fn webdev_route_call(
1050                &self,
1051                _project: &str,
1052                _route: &str,
1053                _body: &serde_json::Value,
1054                _extra_headers: &[(&str, &str)],
1055            ) -> Result<serde_json::Value, CoreError> {
1056                unreachable!("not part of this action")
1057            }
1058            async fn webdev_route_probe(
1059                &self,
1060                _project: &str,
1061                _route: &str,
1062                _extra_headers: &[(&str, &str)],
1063            ) -> Result<crate::client::webdev::RouteProbe, CoreError> {
1064                unreachable!("not part of this action")
1065            }
1066            async fn projects(
1067                &self,
1068                _query: &crate::client::query::ListQuery,
1069            ) -> Result<
1070                crate::client::query::ListEnvelope<crate::client::projects::ProjectRecord>,
1071                CoreError,
1072            > {
1073                unreachable!("not part of this action")
1074            }
1075            async fn project_find(
1076                &self,
1077                _name: &str,
1078            ) -> Result<crate::client::projects::ProjectRecord, CoreError> {
1079                unreachable!("not part of this action")
1080            }
1081            async fn project_create(
1082                &self,
1083                _body: &crate::client::projects::ProjectCreate,
1084            ) -> Result<(), CoreError> {
1085                unreachable!("not part of this action")
1086            }
1087            async fn project_copy(&self, _from: &str, _to: &str) -> Result<(), CoreError> {
1088                unreachable!("not part of this action")
1089            }
1090            async fn project_rename(&self, _name: &str, _new_name: &str) -> Result<(), CoreError> {
1091                unreachable!("not part of this action")
1092            }
1093            async fn project_modify(
1094                &self,
1095                _name: &str,
1096                _body: &crate::client::projects::ProjectModify,
1097            ) -> Result<(), CoreError> {
1098                unreachable!("not part of this action")
1099            }
1100            async fn project_delete(&self, _name: &str) -> Result<(), CoreError> {
1101                unreachable!("not part of this action")
1102            }
1103            async fn project_export_to_file(
1104                &self,
1105                _name: &str,
1106                _out: &std::path::Path,
1107            ) -> Result<crate::client::projects::ExportMeta, CoreError> {
1108                unreachable!("not part of this action")
1109            }
1110            async fn project_import(
1111                &self,
1112                _name: &str,
1113                _zip: Vec<u8>,
1114                _overwrite: bool,
1115            ) -> Result<crate::client::projects::ImportOutcome, CoreError> {
1116                unreachable!("not part of this action")
1117            }
1118        }
1119
1120        let sink: &mut (dyn FnMut(&LogEntry) + Send) = &mut |_| {};
1121        let err = tail(
1122            &AuthRig,
1123            None,
1124            None,
1125            None,
1126            Duration::from_millis(5),
1127            Some(Duration::from_secs(5)),
1128            sink,
1129        )
1130        .await
1131        .expect_err("auth must fail fast");
1132        assert!(matches!(err, CoreError::Auth { status: 401, .. }));
1133        assert_eq!(err.exit_code(), 5);
1134    }
1135}