iris/ai-app
1
0 Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

The app hands its log to Dev Updater on the phone, not through ai-server

Browse Source 
Iris's call once the upload route was working: put it in Dev Updater
properly. So the app now exposes its own ring through a ContentProvider
at `<applicationId>.devlog` -- Dev Updater's contract, written down in
that project's README, not something invented here -- and Dev Updater's
phone app reads it on the same device and forwards it to its own build
machine. No tunnel, no token, no second enrolment, and any app that
server delivers can implement the same and get the same Runtime tab.

`DevLogProvider.java` plus `devlog.rs` are the platform glue only: a flat
`String[]` across JNI, a `MatrixCursor` on the Java side, and
`nativeReady` telling Rust the authority the provider actually
registered, so the Diagnostics pane can name somewhere a reader can
query rather than composing a guess. `LogRing::newest_seq()` is the one
addition in `client-core`: an in-memory ring starts again at zero, so it
is what lets a reader notice the process restarted instead of silently
skipping everything since.

Deleted with it, so there is one mechanism: `client_core::log_upload`,
`POST /client-log` on ai-server, the `AI_APP_LOG_*` baking (which left
`build.rs` with nothing to do), and the uploader on both Android
clients. Kept: the ring, `RingLogger`, `install_process_logger`, and the
Diagnostics line -- whose second half is now `devlog provider:
content://<authority>`.

Verified end to end on this checkout's emulator: iris's own
`iris::android::view` startup lines read out of the provider by the
shell, forwarded by Dev Updater's Runtime tab, and served back from
`GET /apps/android-app/components/app/logs?kind=runtime`. A component
whose package has no provider says so in as many words.

Co-Authored-By: Claude Fable 5.1 <noreply@anthropic.com>
This commit is contained in:
irisandClaude Fable 5.1 committed 2026-09-07 20:58:48 -04:00
1 parent e10582a2cd
commit 06b8a1f4b0
15 files changed
+618 -842
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

client-core/src/lib.rs
-1
View File
@@ -9,7 +9,6 @@ pub mod durations;
pub mod event_stream;
pub mod highlight;
pub mod log_ring;
pub mod log_upload;
pub mod markdown_blocks;
pub mod notifications;
pub mod sse;
client-core/src/log_ring.rs
+39 -4
View File
@@ -11,10 +11,12 @@
//!
//! Two consumers, both reading the same ring rather than each keeping
//! their own: the bench app's `Copy report`/`Diagnostics` (which reads
//! [`LogRing::to_text`] and [`LogRing::summary`]) and the uploader in
//! [`crate::log_upload`] (which reads [`LogRing::since`]). That is why
//! reading does not consume: a line the uploader has sent must still be in
//! the report, and a report taken twice must say the same thing.
//! [`LogRing::to_text`] and [`LogRing::summary`]) and whatever hands the
//! log out of the process -- on Android, the `DevLogProvider` Dev Updater
//! queries, which reads [`LogRing::since`] and [`LogRing::newest_seq`].
//! That is why reading does not consume: a line already handed over must
//! still be in the report, and a report taken twice must say the same
//! thing.
use std::collections::VecDeque;
use std::sync::{Arc, Mutex, OnceLock};
@@ -207,6 +209,21 @@ impl LogRing {
self.with(|inner| inner.dropped)
}
/// The sequence number of the newest line held, or `None` for a ring
/// nothing has been written to.
///
/// What a reader needs to notice that this process **restarted**: the
/// ring is in memory, so a new process starts again at zero, and a
/// reader holding a cursor from the previous one would otherwise ask
/// for lines after a number nothing will reach for hours and see
/// nothing at all -- silently, which is worse than seeing the log
/// begin again. Answering `None` rather than 0 for an empty ring is
/// the same distinction [`Self::summary`] draws: "nothing has been
/// logged" is not a sequence number.
pub fn newest_seq(&self) -> Option<u64> {
self.with(|inner| inner.lines.back().map(|line| line.seq))
}
/// When the newest line was written, in unix milliseconds, or `None`
/// for a ring nothing has been written to.
pub fn last_at_ms(&self) -> Option<u64> {
@@ -426,6 +443,24 @@ mod tests {
assert_eq!(cursor, 4);
}
/// The restart signal: a reader that saw sequence 4 and is now told
/// the newest is 0 knows the process is not the one it was reading.
#[test]
fn the_newest_sequence_says_where_the_ring_is_and_nothing_for_an_empty_one() {
let ring = LogRing::new(100, 1 << 20);
assert_eq!(ring.newest_seq(), None, "an empty ring has no newest line");
fill(&ring, 5);
assert_eq!(ring.newest_seq(), Some(4));
let restarted = LogRing::new(100, 1 << 20);
fill(&restarted, 1);
assert_eq!(
restarted.newest_seq(),
Some(0),
"a fresh ring starts again, which is exactly what a reader has to notice"
);
}
#[test]
fn reading_does_not_consume() {
let ring = LogRing::new(100, 1 << 20);
client-core/src/log_upload.rs
-427
View File
@@ -1,427 +0,0 @@
//! Sending [`crate::log_ring`]'s lines to `ai-server`, so a phone with no
//! `logcat` still has a way for a `log::info!` to reach a person.
//!
//! **Where they end up**: `POST /client-log` re-emits each line into
//! `ai-server`'s own `tracing` output, which Dev Updater already shows as
//! that component's *runtime log* (it runs `ai-server` as a `Managed`
//! service, and a managed service's stdout is redirected to a file its
//! service script reports). So this needs no new route, storage or viewer
//! in Dev Updater at all -- see `docs/DECISIONS.md`, 2026-09-07.
//!
//! **Nothing here calls `log!`.** Every line this module logged would land
//! in the ring it is draining and be uploaded, so a server that is down
//! would produce a growing conversation with itself. Failures are recorded
//! in [`UploadStatus`] instead and shown in the app's diagnostics pane,
//! which is where somebody looking for "why is nothing arriving" is
//! already looking (UI_RULES.md: a failure is reported where it happened).
use std::sync::{Arc, Condvar, Mutex};
use std::time::Duration;
use crate::api::{ApiError, Body, Transport};
use crate::log_ring::LogRing;
/// The most lines one request carries. A phone that has been offline for
/// an hour has thousands waiting, and one request holding all of them is a
/// body the server has to buffer whole; the rest go in the next batch,
/// which the loop takes immediately rather than after the next interval.
pub const MAX_LINES_PER_BATCH: usize = 500;
/// How much of one message is sent. Long enough for a stack trace line,
/// short enough that one pathological message cannot dominate a batch.
/// Truncation is marked, because a silently shortened line reads as a line
/// that ended there.
pub const MAX_MESSAGE_BYTES: usize = 4096;
/// The route this posts to, on `server/src/routes.rs`'s surface.
pub const CLIENT_LOG_PATH: &str = "/client-log";
/// What the last upload attempt did, for a diagnostics pane. `None` for
/// "nothing has been tried yet", which is deliberately distinct from a
/// success that sent nothing.
#[derive(Debug, Clone, Default)]
pub struct UploadStatus {
pub sent: u64,
pub last_error: Option<String>,
pub attempted: bool,
}
impl UploadStatus {
/// One line for the diagnostics pane, in the same voice as
/// [`LogRing::summary`].
pub fn summary(&self) -> String {
match (&self.last_error, self.attempted) {
(Some(err), _) => format!("log upload: failing -- {err} ({} sent so far)", self.sent),
(None, false) => "log upload: not tried yet".to_string(),
(None, true) => format!("log upload: {} lines sent", self.sent),
}
}
}
/// Drains a [`LogRing`] into `POST /client-log`, remembering how far it
/// got so a line is sent once and stays in the ring for the report.
pub struct LogUploader {
ring: LogRing,
transport: Arc<dyn Transport>,
source: String,
cursor: u64,
status: Arc<Mutex<UploadStatus>>,
}
impl LogUploader {
/// `source` names the build these lines came from -- it is what
/// distinguishes them in `ai-server`'s log from the server's own
/// lines and from another device's.
pub fn new(ring: LogRing, transport: Arc<dyn Transport>, source: impl Into<String>) -> Self {
Self {
ring,
transport,
source: source.into(),
cursor: 0,
status: Arc::new(Mutex::new(UploadStatus::default())),
}
}
/// A handle on what the last attempt did, shareable with the UI.
pub fn status(&self) -> Arc<Mutex<UploadStatus>> {
Arc::clone(&self.status)
}
/// Sends up to [`MAX_LINES_PER_BATCH`] waiting lines. Answers how many
/// went, and whether more are waiting -- the loop uses the second to
/// decide whether to go round again at once.
pub fn flush_once(&mut self) -> Result<(usize, bool), ApiError> {
let (mut lines, mut next) = self.ring.since(self.cursor);
let more = lines.len() > MAX_LINES_PER_BATCH;
if more {
lines.truncate(MAX_LINES_PER_BATCH);
next = lines.last().map(|line| line.seq + 1).unwrap_or(next);
}
if lines.is_empty() {
return Ok((0, false));
}
let body = serde_json::json!({
"source": self.source,
"lines": lines.iter().map(|line| serde_json::json!({
"seq": line.seq,
"at": line.at_ms,
"level": line.level.as_str(),
"target": line.target,
"message": truncate(&line.message),
})).collect::<Vec<_>>(),
});
let result = self
.transport
.request("POST", CLIENT_LOG_PATH, Some(Body::Json(body)));
let mut status = self.status.lock().unwrap_or_else(|e| e.into_inner());
status.attempted = true;
match result {
Ok(response) if (200..300).contains(&response.status) => {
// Only on success: a failed batch is retried from the same
// cursor next time, which is what makes a dropped tunnel
// cost nothing but a delay.
self.cursor = next;
status.sent += lines.len() as u64;
status.last_error = None;
Ok((lines.len(), more))
}
Ok(response) => {
let message = format!("{} from {CLIENT_LOG_PATH}", response.status);
status.last_error = Some(message.clone());
Err(ApiError {
message,
status: Some(response.status),
})
}
Err(err) => {
status.last_error = Some(err.message.clone());
Err(err)
}
}
}
}
/// Cuts a message to [`MAX_MESSAGE_BYTES`] on a character boundary, saying
/// so, rather than letting one line dominate a batch.
fn truncate(message: &str) -> String {
if message.len() <= MAX_MESSAGE_BYTES {
return message.to_string();
}
let mut end = MAX_MESSAGE_BYTES;
while end > 0 && !message.is_char_boundary(end) {
end -= 1;
}
format!("{}… [{} bytes cut]", &message[..end], message.len() - end)
}
/// The background half: a thread that flushes on a timer and on demand.
///
/// Its path out is [`Drop`] -- dropping the handle stops the thread and
/// waits for it, so an app that tears the uploader down does not leave one
/// posting behind it.
pub struct LogUpload {
signal: Arc<(Mutex<Signal>, Condvar)>,
thread: Option<std::thread::JoinHandle<()>>,
status: Arc<Mutex<UploadStatus>>,
}
/// Only "stop": a nudge from [`LogUpload::flush_now`] needs no flag,
/// because the thread's reaction to waking is to flush, and flushing an
/// empty ring costs nothing -- so a spurious wakeup is already correct.
#[derive(Default)]
struct Signal {
stop: bool,
}
impl LogUpload {
/// Starts the loop. `every` is how long it waits between flushes when
/// nobody nudges it -- a compromise between a line arriving promptly
/// and a radio the app woke for one line.
pub fn spawn(
ring: LogRing,
transport: Arc<dyn Transport>,
source: impl Into<String>,
every: Duration,
) -> Self {
let mut uploader = LogUploader::new(ring, transport, source);
let status = uploader.status();
let signal = Arc::new((Mutex::new(Signal::default()), Condvar::new()));
let thread = {
let signal = Arc::clone(&signal);
std::thread::Builder::new()
.name("client-log-upload".into())
.spawn(move || {
loop {
// Keep going while a batch was capped, so a
// backlog drains at once rather than one batch per
// interval.
while let Ok((_, more)) = uploader.flush_once() {
if !more {
break;
}
}
let (lock, condvar) = &*signal;
let state = lock.lock().unwrap_or_else(|e| e.into_inner());
if state.stop {
return;
}
let (state, _) = condvar
.wait_timeout(state, every)
.unwrap_or_else(|e| e.into_inner());
if state.stop {
return;
}
}
})
.expect("spawning the log upload thread")
};
Self {
signal,
thread: Some(thread),
status,
}
}
/// Sends what is waiting now -- what `Copy report` calls, so the lines
/// a person is about to describe are already on the server.
pub fn flush_now(&self) {
let (lock, condvar) = &*self.signal;
let _state = lock.lock().unwrap_or_else(|e| e.into_inner());
condvar.notify_all();
}
pub fn status(&self) -> UploadStatus {
self.status
.lock()
.unwrap_or_else(|e| e.into_inner())
.clone()
}
}
impl Drop for LogUpload {
fn drop(&mut self) {
{
let (lock, condvar) = &*self.signal;
let mut state = lock.lock().unwrap_or_else(|e| e.into_inner());
state.stop = true;
condvar.notify_all();
}
if let Some(thread) = self.thread.take() {
let _ = thread.join();
}
}
}
#[cfg(test)]
mod tests {
use super::*;
use crate::api::RawResponse;
use log::Level;
/// Records every body posted, and answers whatever status the test set.
struct Fake {
posted: Mutex<Vec<serde_json::Value>>,
status: Mutex<u16>,
}
impl Fake {
fn new() -> Arc<Self> {
Arc::new(Self {
posted: Mutex::new(Vec::new()),
status: Mutex::new(200),
})
}
fn bodies(&self) -> Vec<serde_json::Value> {
self.posted.lock().unwrap().clone()
}
}
impl Transport for Fake {
fn request(
&self,
method: &str,
path: &str,
body: Option<Body>,
) -> Result<RawResponse, ApiError> {
assert_eq!(method, "POST");
assert_eq!(path, CLIENT_LOG_PATH);
if let Some(Body::Json(value)) = body {
self.posted.lock().unwrap().push(value);
} else {
panic!("the client log is posted as JSON");
}
Ok(RawResponse {
status: *self.status.lock().unwrap(),
body: Vec::new(),
})
}
fn stream(&self, _path: &str) -> Result<Box<dyn std::io::Read + Send>, ApiError> {
unreachable!("the log uploader never streams")
}
}
fn ring_with(count: usize) -> LogRing {
let ring = LogRing::with_defaults();
for n in 0..count {
ring.push(Level::Info, "t", format!("line {n}"));
}
ring
}
#[test]
fn an_empty_ring_posts_nothing() {
let fake = Fake::new();
let mut uploader = LogUploader::new(LogRing::with_defaults(), fake.clone(), "test");
assert_eq!(uploader.flush_once().unwrap(), (0, false));
assert!(
fake.bodies().is_empty(),
"no request at all, not an empty one"
);
}
#[test]
fn a_line_is_sent_once() {
let fake = Fake::new();
let ring = ring_with(3);
let mut uploader = LogUploader::new(ring.clone(), fake.clone(), "test");
assert_eq!(uploader.flush_once().unwrap().0, 3);
assert_eq!(
uploader.flush_once().unwrap(),
(0, false),
"nothing repeats"
);
ring.push(Level::Warn, "t", "later".into());
assert_eq!(uploader.flush_once().unwrap().0, 1);
assert_eq!(fake.bodies().len(), 2);
assert_eq!(ring.len(), 4, "and the report still holds all of them");
}
/// The half the change had no reason to touch: a server that refuses
/// must not lose the lines.
#[test]
fn a_failed_batch_is_retried_from_the_same_place() {
let fake = Fake::new();
*fake.status.lock().unwrap() = 503;
let mut uploader = LogUploader::new(ring_with(2), fake.clone(), "test");
assert!(uploader.flush_once().is_err());
assert!(
uploader.status().lock().unwrap().last_error.is_some(),
"and it says why, where somebody can see it"
);
*fake.status.lock().unwrap() = 200;
assert_eq!(uploader.flush_once().unwrap().0, 2, "the same two lines");
assert!(uploader.status().lock().unwrap().last_error.is_none());
}
#[test]
fn a_backlog_is_capped_per_batch_and_says_there_is_more() {
let fake = Fake::new();
let ring = LogRing::new(MAX_LINES_PER_BATCH * 3, 1 << 30);
for n in 0..(MAX_LINES_PER_BATCH + 7) {
ring.push(Level::Info, "t", format!("{n}"));
}
let mut uploader = LogUploader::new(ring, fake.clone(), "test");
assert_eq!(uploader.flush_once().unwrap(), (MAX_LINES_PER_BATCH, true));
assert_eq!(uploader.flush_once().unwrap(), (7, false));
}
#[test]
fn the_body_carries_the_source_and_each_line_whole() {
let fake = Fake::new();
let ring = LogRing::with_defaults();
ring.push(Level::Error, "iris::android", "surface lost".into());
LogUploader::new(ring, fake.clone(), "iris-bench 1.2")
.flush_once()
.unwrap();
let body = &fake.bodies()[0];
assert_eq!(body["source"], "iris-bench 1.2");
let line = &body["lines"][0];
assert_eq!(line["level"], "ERROR");
assert_eq!(line["target"], "iris::android");
assert_eq!(line["message"], "surface lost");
assert!(line["at"].as_u64().is_some(), "the app's own clock");
}
#[test]
fn an_enormous_message_is_cut_and_says_so() {
let cut = truncate(&"x".repeat(MAX_MESSAGE_BYTES + 100));
assert!(cut.starts_with("xxxx"));
assert!(cut.contains("bytes cut"), "{cut}");
assert!(cut.len() < MAX_MESSAGE_BYTES + 64);
let short = truncate("fine");
assert_eq!(short, "fine", "a short message is untouched");
}
#[test]
fn the_status_line_distinguishes_untried_from_sent_nothing() {
let untried = UploadStatus::default();
assert_eq!(untried.summary(), "log upload: not tried yet");
let sent_none = UploadStatus {
attempted: true,
..Default::default()
};
assert_eq!(sent_none.summary(), "log upload: 0 lines sent");
}
/// The path out: dropping the handle must stop the thread, not leave
/// it posting.
#[test]
fn dropping_the_handle_stops_the_thread() {
let fake = Fake::new();
let upload = LogUpload::spawn(
ring_with(1),
fake.clone(),
"test",
Duration::from_millis(10),
);
upload.flush_now();
drop(upload);
let after = fake.bodies().len();
std::thread::sleep(Duration::from_millis(60));
assert_eq!(fake.bodies().len(), after, "nothing posted after the drop");
}
}