feat(logging): Add privacy-safe logging foundations - #1136
Conversation
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-openagent) Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>
📝 WalkthroughWalkthroughThe PR adds a privacy-safe logging system across configuration, event contracts, host runtime, SQLite persistence, artifact storage, configuration apply handling, and CI workspace routing. ChangesLogging foundation
Estimated code review effort: 5 (Critical) | ~120 minutes Sequence Diagram(s)sequenceDiagram
participant HostRuntime
participant LoggingService
participant ReplayBus
participant LogStore
participant ArtifactFileStore
HostRuntime->>LoggingService: register and enqueue lifecycle event
LoggingService->>ReplayBus: buffer canonical envelope
LoggingService->>LogStore: persist summaries and events
LoggingService->>ArtifactFileStore: write artifact and metadata pointer
ArtifactFileStore->>LogStore: commit artifact pointer
Possibly related PRs
🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches📝 Generate docstrings
🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
|
Canonical API child replacement is now #1137 (base: |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 4
Note
Due to the large number of review comments, Critical severity comments were prioritized as inline comments.
🟠 Major comments (24)
crates/mesh-llm-log-store/src/migrations.rs-37-42 (1)
37-42: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftThe terminal-event unique index matches on raw JSON substrings.
idx_terminal_event_one_per_requestusespayload_json LIKE '%"type":"completed"%'to identify terminal events. Any payload that embeds that substring in a different position marks the event as terminal. An error message such as{"type":"failed_upstream","error":"...\"type\":\"completed\"..."}matches, and the row then blocks all later terminal events for that request.The same substring matching is duplicated in
repositories.rsinis_terminal_payload,check_existing_terminal_raw, andwrite_terminal_event. Store the event type in a dedicatedevent_type TEXTcolumn, then index and compare on that column.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/migrations.rs` around lines 37 - 42, The terminal-event detection relies on unsafe raw JSON substring matching. Add and populate a dedicated event_type TEXT column for lifecycle events, then update idx_terminal_event_one_per_request and the is_terminal_payload, check_existing_terminal_raw, and write_terminal_event paths to compare the parsed event type against terminal values through event_type, removing the LIKE-based checks.crates/mesh-llm-log-store/src/repositories.rs-182-297 (1)
182-297: 📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy liftExtract one shared keyset-pagination helper.
list_summaries,list_lifecycle_events, andlist_artifact_pointerseach contain a cursor branch and a no-cursor branch. The six blocks differ only in the table name, the column list, and the two key columns. Each block repeats the same steps: build the SQL, collect rows, return early when empty, run anEXISTSprobe against the last row, and encode the next cursor.The duplication already shows drift. The probe SQL is indented differently between the two branches of the same function, and the comments diverge ("Probe: is there at least one more row" versus "Probe for more rows"). Any fix to the pagination boundary must be applied in six places.
Extract a helper that takes the table name, the column list, the two key column names, the limit, and the cursor, and that returns
Page<T>through a row-mapping closure. The no-cursor branch becomes the same call with a sentinel-freeWHEREclause.This also addresses the file length.
repositories.rsis 1088 lines. The coding guidelines require extracting separable responsibilities and keeping the resulting files under 1,000 lines. The pagination helper, plus moving the artifact-pointer methods into their own module, brings the file under that limit.As per coding guidelines: "When modifying a Rust source file over 1,000 lines, extract any separable responsibility into a semantically named module, keep the new file under 1,000 lines, and move relevant tests with the extracted behavior."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/repositories.rs` around lines 182 - 297, Extract a shared generic keyset-pagination helper used by list_summaries, list_lifecycle_events, and list_artifact_pointers, accepting table/column/key metadata, limit, cursor, and a row-mapping closure while centralizing query, empty-page, probe, and next-cursor handling. Replace each method’s cursor and no-cursor branches with this helper, using a sentinel-free condition for the initial page. Move artifact-pointer methods and their related tests into a semantically named module so the original source and new module remain under 1,000 lines.Source: Coding guidelines
crates/mesh-llm-log-store/src/repositories.rs-329-334 (1)
329-334: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winDo not put the full payload JSON into the error.
DuplicateTerminalEvent.event_typereceivespayload_json.to_string(), which is the complete event payload.LogStoreError::Displayinerror.rsprints that field asduplicate terminal event for summary={} type={}. Any caller that logs the error writes the whole payload, including upstream error text and request metadata, into the host log.The PR states that redaction is centralized before persistence. This path bypasses that guarantee, because the payload reaches the log through an error message rather than through the store.
Pass only the event type discriminator. The same problem occurs at Line 346 and at Line 381 in
write_terminal_event.🛡️ Proposed fix
+/// Extract the terminal event type discriminator from a payload, without the payload body. +fn terminal_event_type(payload_json: &str) -> &'static str { + for candidate in ["completed", "failed", "rejected", "cancelled"] { + if payload_json.contains(&format!(r#""type":"{candidate}""#)) { + return candidate; + } + } + "unknown" +} ... if is_terminal_payload(payload_json) && check_existing_terminal_raw(&conn, request_id)? { return Err(LogStoreError::DuplicateTerminalEvent { summary_id: request_id.to_string(), - event_type: payload_json.to_string(), + event_type: terminal_event_type(payload_json).to_string(), }); }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/repositories.rs` around lines 329 - 334, Replace payload_json.to_string() with the event type discriminator when constructing DuplicateTerminalEvent in the branches near is_terminal_payload, including the corresponding cases around lines 346 and 381 in write_terminal_event. Ensure LogStoreError::event_type receives only the non-sensitive event type value and never the complete payload JSON, preserving centralized redaction before persistence.crates/mesh-llm-log-store/src/store.rs-71-86 (1)
71-86: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winDo not panic on a poisoned connection mutex, and report poisoning accurately.
conn()calls.expect("connection mutex poisoned").conn()is the entry point for almost every repository method. If any thread panics while it holds the guard, every later logging call panics. The PR describes logging as fail-open, so this turns a logging fault into a host-runtime crash.
txn()handles the same condition differently. It maps the poison error toLogStoreError::Sqlite(rusqlite::Error::ExecuteReturnedResults). That error text is unrelated to mutex poisoning and will mislead operators reading the log.A poisoned guard still holds a usable
Connection. An uncommittedTransactionrolls back when it drops. Recover the guard in both places.🐛 Proposed fix: recover from poisoning
pub fn txn<T>( &self, f: impl FnOnce(&Transaction) -> Result<T, LogStoreError>, ) -> Result<T, LogStoreError> { - let mut conn = self - .conn - .lock() - .map_err(|_| LogStoreError::Sqlite(rusqlite::Error::ExecuteReturnedResults))?; + let mut conn = self.conn.lock().unwrap_or_else(|e| e.into_inner()); let tx = conn.transaction().map_err(LogStoreError::Sqlite)?; let result = f(&tx); if result.is_ok() { tx.commit().map_err(LogStoreError::Sqlite)?; } result } pub fn conn(&self) -> std::sync::MutexGuard<'_, Connection> { - self.conn.lock().expect("connection mutex poisoned") + self.conn.lock().unwrap_or_else(|e| e.into_inner()) }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/store.rs` around lines 71 - 86, Update the connection-lock handling in both txn() and conn() to recover the poisoned MutexGuard instead of panicking or converting poisoning to rusqlite::Error::ExecuteReturnedResults. Reuse the recovered guard’s usable Connection, while preserving transaction rollback on drop for uncommitted transactions and the existing transaction/result flow.crates/mesh-llm-log-store/src/artifacts.rs-863-870 (1)
863-870: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winAn environment variable disables the artifact privacy guarantee.
check_windows_privacyreturns true whenMESH_LLM_ALLOW_WEAK_PRIVACYis present.std::env::var(..).is_ok()accepts any value, including an empty string. Nothing restricts the variable to test runs. A production Windows host that inherits this variable from its parent process stores artifact content, which contains prompt and response data, without any privacy guarantee, andArtifactFileStore::openreports success.The trailing
|| falsehas no effect and should be removed.Restrict the override to test builds, and require an explicit value.
🛡️ Proposed fix
/// Best-effort Windows privacy check. Returns true if we can proceed (or it's not a real risk). #[cfg(windows)] fn check_windows_privacy(_root: &Path) -> bool { - // Pure std has no ACL APIs for checking Everyone/Users permissions. - // Return false to trigger PrivacyNotGuaranteed on Windows in production, - // but allow it during CI/testing (env var override). - std::env::var("MESH_LLM_ALLOW_WEAK_PRIVACY").is_ok() || false + // Pure std has no ACL APIs for checking Everyone/Users permissions, so the + // check fails closed on Windows. Test builds may opt out explicitly. + if cfg!(test) { + return matches!( + std::env::var("MESH_LLM_ALLOW_WEAK_PRIVACY").as_deref(), + Ok("1") | Ok("true") + ); + } + false }🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/artifacts.rs` around lines 863 - 870, Update check_windows_privacy to allow the environment-variable override only in test builds and only when its value explicitly enables the override, rather than merely being present. Remove the redundant trailing “|| false” and keep production Windows builds returning false without the test-only override.crates/mesh-llm-log-store/src/repositories.rs-973-985 (1)
973-985: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftDeleting orphaned summaries cascades to artifact rows whose files are never collected.
cascade_cleanup_artifact_rows_innercollects only theartifact_idvalues whoseoccurred_atis before the cutoff. TheDELETE FROM summariesat Line 976 then removes summaries that have no remaining lifecycle events andcreated_at < cutoff.artifact_pointers.request_iddeclaresON DELETE CASCADE, so SQLite also deletes every remaining artifact pointer row for those summaries, including rows whoseoccurred_atis at or after the cutoff.Those artifact IDs are absent from the returned
Vec<String>. The caller,ArtifactFileStore::delete_artifact_files, therefore never deletes the corresponding files. The files stay on disk with no database row.remove_unreferenced_filesonly removes them at the next startup, so artifact content that the retention policy deleted persists until the host restarts.Collect the cascaded artifact IDs before the summary delete.
🐛 Proposed fix
// Delete orphaned summaries: no remaining lifecycle_events AND created_at < cutoff. + // Collect artifact IDs that ON DELETE CASCADE will remove with those summaries. + let mut cascaded: Vec<String> = { + let mut stmt = tx + .prepare( + "SELECT artifact_id FROM artifact_pointers WHERE request_id IN (\ + SELECT request_id FROM summaries \ + WHERE request_id NOT IN (SELECT DISTINCT request_id FROM lifecycle_events) \ + AND created_at < ?)", + ) + .map_err(LogStoreError::Sqlite)?; + stmt.query_map(rusqlite::params![cutoff_occurred_at], |row| { + row.get::<_, String>(0) + }) + .map_err(LogStoreError::Sqlite)? + .collect::<Result<Vec<_>, _>>() + .map_err(|e| LogStoreError::QueryFailed(e.to_string()))? + }; + let orphans: usize = tx .execute( "DELETE FROM summaries \ WHERE request_id NOT IN (SELECT DISTINCT request_id FROM lifecycle_events) \ AND created_at < ?", rusqlite::params![cutoff_occurred_at], ) .map_err(LogStoreError::Sqlite)?; total += orphans as i64; + total += cascaded.len() as i64; - Ok((total, artifact_ids)) as Result<(i64, Vec<String>), LogStoreError> + let mut artifact_ids = artifact_ids; + artifact_ids.append(&mut cascaded); + Ok((total, artifact_ids)) as Result<(i64, Vec<String>), LogStoreError>🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/repositories.rs` around lines 973 - 985, Update cascade_cleanup_artifact_rows_inner to collect all artifact IDs belonging to orphaned summaries before executing the DELETE FROM summaries statement. Include artifact pointers regardless of their occurred_at value, merge them with the existing artifact_ids result without duplicates, and preserve the current summary deletion and returned tuple behavior.crates/mesh-llm-log-store/src/migrations.rs-116-147 (1)
116-147: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winMake migration step version bumps atomic before applying statements.
apply_migrationsruns each migration batch in autocommit and updatesPRAGMA user_versiononly after the final batch succeeds. If the process crashes or a statement fails duringMIGRATIONS_V2, someartifact_pointerscolumns persist whileuser_versionremains old. The next open repeatsMIGRATIONS_V2, and the first already-appliedALTER TABLE artifact_pointers ADD COLUMN media_kind TEXTfails withduplicate column name, leavingLogStore::openreturningMigrationFailed. UseBEGIN; ... PRAGMA user_version = N; COMMIT;for each schema migration step so the version bump and schema change commit together.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/migrations.rs` around lines 116 - 147, Update apply_migrations to execute each pending migration step transactionally: begin a transaction before MIGRATIONS_V1 or MIGRATIONS_V2, apply that step’s statements, set PRAGMA user_version to the step’s version, and commit before proceeding. Ensure failures roll back both schema changes and the version update, and remove the single final version bump so each migration’s schema changes and version advance commit atomically.crates/mesh-llm-log-store/src/repositories.rs-211-224 (1)
211-224: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winBind
limitas a parameter and clamp it.
limitis interpolated into the SQL text withformat!. Two consequences follow.First, every distinct
limitvalue produces a distinct SQL string, so rusqlite's prepared-statement cache never reuses the plan.Second,
limitis unbounded. A caller that passes a large value, orusize::MAX, makes SQLite return every matching row, andcollect::<Vec<_>>holds them all in memory. The summaries table grows with every request, so this can exhaust memory on a busy host.Bind the value and clamp it to a maximum page size. The same pattern occurs in
list_lifecycle_events(Lines 425-429, 468-472) andlist_artifact_pointers(Lines 745-749, 788-792).🐛 Proposed fix
+const MAX_PAGE_SIZE: usize = 1000; + ... - let sql = format!( - "SELECT request_id, state, created_at, terminal_at, route, model, provider, engine, \ - status_code, error_msg, tenant_id, account_id, user_id \ - FROM summaries WHERE (created_at, request_id) < (?, ?) ORDER BY created_at DESC, request_id DESC LIMIT {}", - limit - ); - let mut stmt = conn.prepare(&sql).map_err(LogStoreError::Sqlite)?; + let sql = "SELECT request_id, state, created_at, terminal_at, route, model, provider, engine, \ + status_code, error_msg, tenant_id, account_id, user_id \ + FROM summaries WHERE (created_at, request_id) < (?, ?) ORDER BY created_at DESC, request_id DESC LIMIT ?"; + let mut stmt = conn.prepare(sql).map_err(LogStoreError::Sqlite)?; + let capped = limit.min(MAX_PAGE_SIZE) as i64; // Collect exactly `limit` rows. let items: Vec<SummaryRow> = stmt - .query_map(rusqlite::params![ts, id], &row_fn) + .query_map(rusqlite::params![ts, id, capped], &row_fn)🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/repositories.rs` around lines 211 - 224, Update the pagination queries in the summary listing block and the corresponding list_lifecycle_events and list_artifact_pointers paths to clamp the requested limit to the established maximum page size, bind the clamped value as a SQLite parameter, and remove its interpolation from the SQL text. Preserve the existing ordering, cursor conditions, and row collection behavior while applying the same bounded-parameter pattern to every affected query.crates/mesh-llm-log-store/src/artifacts.rs-271-279 (1)
271-279: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winThe aggregate limit check runs outside the transaction that records the bytes.
sum_artifact_bytes_for_requestreads the current total, and theINSERTplusUPDATEthat recordbyteshappen later in a separate transaction at Lines 359-385. Nothing holds a lock across the two steps.If two threads write artifacts for the same
request_idat the same time, both read the sameexisting_bytes, both pass the check, and both commit. The stored total then exceedsaggregate_limit. The limit exists to bound how much request content the host retains, so exceeding it weakens a privacy and disk-usage guarantee.Move the check into the same transaction as the insert.
🐛 Proposed fix
- // Check aggregate limit for this request (existing bytes + new bytes). - let existing_bytes = self.store.sum_artifact_bytes_for_request(request_id)?; - if existing_bytes + stored.len() as i64 > aggregate_limit as i64 { - return Err(LogStoreError::ArtifactLimitExceeded { - artifact_id: artifact_id.to_string(), - limit_bytes: aggregate_limit, - kind: "aggregate".to_string(), - }); - }Then, at the start of the
self.store.txn(|tx| { .. })closure at Line 359:let existing_bytes: i64 = tx .query_row( "SELECT COALESCE(SUM(bytes), 0) FROM artifact_pointers WHERE request_id = ?", rusqlite::params![request_id], |row| row.get(0), ) .map_err(LogStoreError::Sqlite)?; if existing_bytes + stored.len() as i64 > aggregate_limit as i64 { return Err(LogStoreError::ArtifactLimitExceeded { artifact_id: artifact_id.to_string(), limit_bytes: aggregate_limit, kind: "aggregate".to_string(), }); }The existing failure path at Lines 387-391 already removes the written file when the transaction returns an error, so the rejection still leaves no partial file.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/artifacts.rs` around lines 271 - 279, Move the aggregate byte-limit validation from the pre-transaction path into the `self.store.txn(|tx| { ... })` closure, querying the current total through `tx` before the insert/update operations. Preserve the existing `LogStoreError::ArtifactLimitExceeded` response and rely on the existing transaction-error cleanup path for rejected writes.crates/mesh-llm-log-store/src/artifacts.rs-52-58 (1)
52-58: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winKeep
LogStoreshared behindArc.
ArtifactFileStorecurrently ownsLogStoreby value, so callers cannot continue using the same connection after opening artifact storage.ArtifactFileStoreusesconn()/txn(), while host code needsLogStoremethods such asinsert_summary/txn; opening a secondLogStoreon the same path creates a separateMutex<Connection>. Store both asArc<LogStore>and expose a public accessor for consumers/tests.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-log-store/src/artifacts.rs` around lines 52 - 58, Change ArtifactFileStore.store to Arc<LogStore> and update its construction and conn()/txn() usage to clone or dereference the shared Arc as needed. Add a public accessor on ArtifactFileStore that returns the shared Arc<LogStore>, allowing callers to reuse the same connection for methods such as insert_summary and txn instead of opening another LogStore.crates/mesh-llm-host-runtime/src/logging/policy.rs-284-301 (1)
284-301: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick win
sanitize_pathis duplicated across two logging modules.
crates/mesh-llm-host-runtime/src/logging/foundation.rsdefines its ownsanitize_path(lines 179-186) with different behavior: it usesdirs::home_dir()andstrip_prefix, while this version uses theHOMEenvironment variable and a substringreplace. The two functions produce different output for the same path, and both are used for privacy-sensitive diagnostics. Keep one implementation inpolicyand call it fromfoundation.As per coding guidelines: "Duplicate code (copy/paste, similar logic, abstractions)."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-host-runtime/src/logging/policy.rs` around lines 284 - 301, Remove the duplicate sanitize_path implementation from logging/foundation.rs and reuse the existing logging::policy::sanitize_path function instead. Update foundation’s call sites and visibility/imports as needed while preserving the policy implementation as the single source of truth for path sanitization.Source: Coding guidelines
crates/mesh-llm-host-runtime/src/logging/service.rs-213-238 (1)
213-238: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftNothing ever sends a message to the persistence worker.
The worker loop consumes
WorkerMessage::PersistBusEntry, but no code constructs that variant.enqueue_eventpushes the payload intoReplayBusand returns. The worker never drains the bus and never awaitsReplayBus::notified(). As a result, the spawned task idles until the channel closes, and no event reachesPersistSink. Under sustained load the bus evicts entries with drop-oldest and every event is lost.Either drive the worker from the bus, or send each entry through the channel on enqueue.
🔧 Proposed direction (bus-driven worker)
tokio::spawn(async move { - while let Some(msg) = rx.recv().await { - match msg { - WorkerMessage::PersistBusEntry(entry) => { + loop { + tokio::select! { + msg = rx.recv() => match msg { + Some(WorkerMessage::Shutdown) | None => break, + Some(WorkerMessage::PersistBusEntry(entry)) => { + if let Some(sink) = &sink_opt { + let _ = Self::process_bus_entry(&bus, sink.as_ref(), &entry).await; + } + } + }, + _ = bus.notified() => { + for entry in bus.drain() { if let Some(sink) = &sink_opt { let _ = Self::process_bus_entry(&bus, sink.as_ref(), &entry).await; } + } } - WorkerMessage::Shutdown => { - break; - } } } });The test suite does not catch this, because
pump_syncdrains the bus and only counts entries; it never asserts onTestSinkrecords. Add a test that spawns the worker and asserts that the sink received the events.Also applies to: 257-280
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-host-runtime/src/logging/service.rs` around lines 213 - 238, Connect the persistence worker to the event flow: update the worker setup around WorkerHandle and its spawned loop, plus enqueue_event, so each queued event produces a WorkerMessage::PersistBusEntry or the worker actively drains ReplayBus and awaits its notifications. Preserve shutdown handling and ensure events reach PersistSink without being silently evicted; add a test that starts the worker and asserts TestSink records the enqueued events.crates/mesh-llm-host-runtime/src/logging/registry.rs-172-183 (1)
172-183: 🚀 Performance & Scalability | 🟠 Major | 🏗️ Heavy liftEviction scans the whole map on every insert at capacity.
evict_oldestrunsmin_by_keyover all entries, so each insert at capacity costs O(n) while the mutex is held. With the defaults (max_recent= 8192), every terminal transition under sustained load scans 8192 entries on the request path. Keep insertion order in aVecDeque<String>next to the map, and pop the front key for O(1) eviction. The FIFO order also removes the dependency on lexicographiccreated_atcomparison, which is only correct when all timestamps use the same format and timezone.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-host-runtime/src/logging/registry.rs` around lines 172 - 183, Replace the full-map min_by_key logic in evict_oldest with FIFO tracking via a VecDeque<String> maintained alongside the entries HashMap. Push each newly inserted key to the deque and pop the front key during eviction, removing that key from the map in O(1); update all insertion and removal paths to keep both structures consistent and eliminate reliance on created_at ordering.crates/mesh-llm-host-runtime/src/lib.rs-44-59 (1)
44-59: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winRemove the public logging accessors or expose the logging module/type.
mod logging;is private, so publiclogging_foundation()andlogging_health_summary()incrates/mesh-llm-host-runtime/src/lib.rsreturn/useLoggingFoundationthrough a private path. Makelogging::foundation::LoggingFoundationpublicly reexportable or make the accessors internal; external crate use will not compile otherwise.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-host-runtime/src/lib.rs` around lines 44 - 59, Resolve the visibility mismatch around the public accessors logging_foundation() and logging_health_summary(): either expose the logging module and LoggingFoundation type publicly for external callers, or make both accessors non-public if they are internal-only. Preserve the existing OnceLock-backed behavior and health-summary results.Source: Coding guidelines
crates/mesh-llm-host-runtime/src/logging/foundation.rs-149-152 (1)
149-152: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winAvoid the shared
/tmp/mesh-llm/loggingfallback.If
dirs::home_dir()returnsNone, the code writes logging state to a fixed path under/tmp. On a multi-user host,/tmpis world-writable and the path is predictable. Another local user can pre-create the directory, or replace a component with a symlink, and then read or redirect the log store and artifacts. Logging state includes redacted request records, so this weakens the privacy guarantee of the feature.Prefer failing open (disabled) when no home directory is available, or use a per-user path such as
std::env::temp_dir()combined with the effective user id and restrictive permissions.🔒 Proposed direction
dirs::home_dir() .map(|home| home.join(".mesh-llm").join("logging")) - .unwrap_or_else(|| PathBuf::from("/tmp/mesh-llm/logging")) + .unwrap_or_else(|| { + // No home directory: use a per-user temp path instead of a shared, predictable one. + std::env::temp_dir() + .join(format!("mesh-llm-{}", std::process::id())) + .join("logging") + })Note that a fully disabled foundation may be the safer choice here. Confirm the intended behavior for hosts without a home directory.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-host-runtime/src/logging/foundation.rs` around lines 149 - 152, Remove the fixed `/tmp/mesh-llm/logging` fallback from the logging path helper around `dirs::home_dir()`. When no home directory is available, disable the logging foundation by propagating an unavailable result through its callers rather than constructing a shared path; preserve the existing per-home `.mesh-llm/logging` path when available.crates/mesh-llm-host-runtime/src/logging/lifecycle.rs-132-170 (1)
132-170: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winThe CAS and the record write are not atomic together.
The winning thread sets the state flag at line 136 and takes the mutex at line 140. A second thread that fails the CAS in that window reads
record.outcome == Noneand returns the placeholderDropped(None)asexisting(lines 159-166). Two effects follow. The duplicate error reports an outcome that never happened.terminate_idempotentthen returnsErrfor a repeat of the same outcome, which breaks the documented idempotency.Take the mutex first and perform both the check and the store under it. The atomic flag can stay as a fast path for
is_active.🔒 Proposed fix
pub fn terminate(&self, outcome: TerminalOutcome) -> Result<(), DuplicateTerminalError> { - match self - .state_flag - .compare_exchange(STATE_ACTIVE, 1, Ordering::AcqRel, Ordering::Acquire) - { - Ok(_) => { - let mut record = self.record.lock().expect("terminal record mutex poisoned"); - if let Some(existing) = &record.outcome { - return Err(DuplicateTerminalError { - existing: existing.clone(), - attempted: outcome, - }); - } - record.outcome = Some(outcome); - Ok(()) - } - Err(_) => { /* ... */ } - } + let mut record = self.record.lock().expect("terminal record mutex poisoned"); + if let Some(existing) = &record.outcome { + return Err(DuplicateTerminalError { + existing: existing.clone(), + attempted: outcome, + }); + } + record.outcome = Some(outcome); + // Publish the terminal marker only after the outcome is visible. + self.state_flag.store(1, Ordering::Release); + Ok(()) }The current test
concurrent_terminate_only_one_succeedsstill passes with the race present, because it only counts successes. Add an assertion onexistingfrom the losing threads.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-host-runtime/src/logging/lifecycle.rs` around lines 132 - 170, Update terminate to acquire the record mutex before checking terminal state, then perform duplicate detection and outcome storage while holding that mutex; do not use the state_flag CAS as the ownership decision, while preserving its use for is_active’s fast path. Ensure concurrent losers report the winner’s actual outcome, and extend concurrent_terminate_only_one_succeeds to assert each DuplicateTerminalError.existing matches the stored winning outcome so terminate_idempotent remains idempotent.crates/mesh-llm-host-runtime/src/lib.rs-138-149 (1)
138-149: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winUse the loaded logging config when initializing the foundation.
LoggingFoundation::init(true, None)ignores the already-loaded config, sologging.enabled = falsestill resolves the logging root and createsstore/artifactsdirectories on startup. Passconfig.logging.enabledandconfig.logging.application_state_root.as_ref()intoLoggingFoundation::init; for builds withoutdynamic-native-runtime, keep the fallback behavior explicit.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-host-runtime/src/lib.rs` around lines 138 - 149, Update the LoggingFoundation::init call to use config.logging.enabled and config.logging.application_state_root.as_ref() instead of hardcoded values, preventing disabled logging from resolving its root or creating directories. For builds without dynamic-native-runtime, preserve the fallback behavior explicitly while keeping the existing health warning and LOGGING_FOUNDATION initialization flow unchanged.crates/mesh-llm-events/src/logging/summaries.rs-14-77 (1)
14-77: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winRemove the unnecessary
#[allow(dead_code)]attributes from the public logging contract APIs. These fields and methods arepuband reachable throughpub mod logging;, so thedead_codelint does not apply to them. Keep only attributes that suppress an actual warning.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-events/src/logging/summaries.rs` around lines 14 - 77, Remove unnecessary #[allow(dead_code)] attributes from public logging contract fields and methods in crates/mesh-llm-events/src/logging/summaries.rs (lines 14-77), artifacts.rs (30-97), envelope.rs (41-87), identifiers.rs (28-28), lifecycle.rs (26-80), and replay.rs (26-26). Preserve any attributes that suppress warnings other than dead_code, and make no other changes.Source: Coding guidelines
crates/mesh-llm-events/src/logging/lifecycle.rs-26-26 (1)
26-26: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winRemove unnecessary
#[allow(dead_code)]on public lifecycle methods.
crates/mesh-llm-events/src/logging/mod.rsexposespub mod lifecycle, and the crate exposespub mod logging, soLifecycleState::as_str,LifecycleGuard::active, andLifecycleGuard::stateare public exports. These public items are not subject todead_code; instead, remove the attributes to surface dead methods.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-events/src/logging/lifecycle.rs` at line 26, Remove the unnecessary #[allow(dead_code)] attributes from the public lifecycle methods LifecycleState::as_str, LifecycleGuard::active, and LifecycleGuard::state in the lifecycle module, allowing dead-code diagnostics to surface for these exports.Source: Coding guidelines
crates/mesh-llm-config/src/model.rs-1371-1559 (1)
1371-1559: 📐 Maintainability & Code Quality | 🟠 Major | 🏗️ Heavy liftMove the logging configuration types into a dedicated module.
model.rsalready exceeds 1,000 lines, and this change adds a further self-contained responsibility (~190 lines) to it. The crate already has amodel/directory (model/runtime.rs,model/built_in_schema.rs). ExtractCaptureMode,LoggingConfig,LoggingArtifactConfig,LoggingWebhookConfig, and their default helpers intomodel/logging.rs, and re-export them frommodel.rs.Note one coupling:
crates/mesh-llm-config/src/lib.rscomputes the schema field inventory withinclude_str!("model.rs")andinclude_str!("model/runtime.rs"). If you extract the types, addinclude_str!("model/logging.rs")to thatsourcesarray, otherwiseextract_struct_fieldswill panic with "struct LoggingConfig not found in config model sources".As per coding guidelines: "When modifying a Rust source file over 1,000 lines, extract any separable responsibility into a semantically named module, keep the new file under 1,000 lines, and move relevant tests with the extracted behavior."
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-config/src/model.rs` around lines 1371 - 1559, Extract CaptureMode, LoggingConfig, LoggingArtifactConfig, LoggingWebhookConfig, their default helpers, and Default implementations from model.rs into a new model/logging.rs module, then declare and re-export them from model.rs. Update the schema source list in lib.rs to include model/logging.rs so LoggingConfig remains discoverable by extract_struct_fields. Move any tests associated with these logging types alongside the extracted module.Source: Coding guidelines
crates/mesh-llm-config/src/validate.rs-467-470 (1)
467-470: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winTake
&Pathinstead of&PathBuf.
clippy::ptr_argis enabled by default and will warn on the&std::path::PathBufparameter. The function only reads the path, so&Pathis sufficient and the call site at Line 360 needs no change beyond a deref.As per coding guidelines: "Do not leave compiler or lint warnings in touched Rust code; fix warnings rather than using
#[allow(...)]".♻️ Proposed signature change
fn validate_application_state_root( - root: &std::path::PathBuf, + root: &std::path::Path, diagnostics: &mut Vec<ConfigDiagnostic>, ) {Call site:
if let Some(root) = &config.application_state_root { validate_application_state_root(root, &mut diagnostics); }
&PathBufderefs to&Path, so the call site compiles unchanged.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-config/src/validate.rs` around lines 467 - 470, Update validate_application_state_root to accept &std::path::Path instead of &std::path::PathBuf, leaving its read-only behavior unchanged. Keep the existing call site unchanged since &PathBuf coerces to &Path, and do not suppress the clippy::ptr_arg warning.Source: Coding guidelines
crates/mesh-llm-host-runtime/src/mesh/owner_control/mod.rs-813-832 (1)
813-832: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winUse the proto apply-mode mapping for restart-required applies.
ApplyResult::AppliedWithRestartRequiredsendsConfigApplyMode::Staged as i32, but that local enum has only{ Staged = 0, Noop = 1 }, while the proto enum has{ Unspecified = 0, Staged = 1, Live = 2, Noop = 3 }. Clients seeUnspecified. Usecrate::proto::node::ConfigApplyMode::Staged as i32, or route this throughowner_control_response::proto_apply_modewith a protoRestartRequiredvalue.Also add an API-visible restart signal, such as an additive proto
CONFIG_APPLY_MODE_RESTART_REQUIRED, so the caller does not treat a restart-required apply as a normal staged apply.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-host-runtime/src/mesh/owner_control/mod.rs` around lines 813 - 832, Update the ApplyResult::AppliedWithRestartRequired response branch to use the proto node ConfigApplyMode mapping instead of the local ConfigApplyMode enum, and expose the restart requirement through an additive proto apply-mode value such as CONFIG_APPLY_MODE_RESTART_REQUIRED. Ensure the response’s apply_mode communicates restart-required status while preserving the existing revision, hash, diagnostics, and success fields.Source: Coding guidelines
crates/mesh-llm-host-runtime/src/runtime/config_state.rs-140-144 (1)
140-144: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winResolve nested
.fieldnames before lookup, and fail safe for missing descriptors.
ConfigPath::from_fields(["logging", "artifact.capture_mode"])treats"artifact.capture_mode"as one segment, while the schema path is parsed withConfigPath::parse_rendered(...), so nested logging fields render as eitherlogging.artifact.capture_modeorlogging.artifact.capture_mode.capture_mode. Those lookups are not in the built-in schema, sofield_requires_restartreturnsfalseand misses required restarts forartifact.*andwebhook.*changes. Pass the nested segments individually, or parse the dot-separated name before lookup. Also make the default require restart when no descriptor exists, instead of treating an unknown field as safely dynamic.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-host-runtime/src/runtime/config_state.rs` around lines 140 - 144, Update field_requires_restart to split dotted field_name values into separate ConfigPath segments before built_in_config_schema_descriptor lookup, matching the schema’s rendered paths for nested logging fields such as artifact.* and webhook.*. Change the missing-descriptor fallback to require a restart, while preserving SchemaApplyMode::DynamicApply as the only non-restart case for known descriptors.crates/mesh-llm-config/src/validate.rs-479-505 (1)
479-505: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winFix the system-path and traversal checks; the current string matching misreports and misses cases.
Three concrete problems in this block:
- Line 490 emits the wrong diagnostic. The
else ifarm is reachable only whenpath_str != "/"and thestarts_witharm did not match. Combined withpath_str.len() <= 1, it fires for any single-character relative path such as"a"or".", and reports "must not be the filesystem root "/"". A valid relative root is therefore rejected with a false reason.- When
rootreally is/, the first arm matches instead, so the message reads "rejecting path starting with "/"". The dedicated filesystem-root message is unreachable.- Bare system directories bypass the check.
"/etc","/proc","/sys", and"/dev"do not equal any prefix and do not start with the trailing-slash prefixes, so they validate successfully. Only"/etc/..."and similar are rejected.- The traversal check at Line 499 misses
"..","a/..", and any path ending in/.., because it requires a leading/..or an embedded/../.Match on
Pathcomponents instead of raw string prefixes. That removes all four cases and the separator-portability concern.🐛 Proposed fix using path components
- // Reject absolute system paths that should never contain application state. - let forbidden_prefixes = ["/", "/etc/", "/dev/", "/proc/", "/sys/"]; - if let Some(path_str) = root.to_str() { - for prefix in &forbidden_prefixes { - if path_str == *prefix || (path_str.starts_with(prefix) && *prefix != "/") { - diagnostics.push(validation_diagnostic( - "logging.application_state_root", - format!( - "logging.application_state_root must not target system directories; rejecting path starting with \"{prefix}\"" - ), - )); - } else if *prefix == "/" && path_str.len() <= 1 { - diagnostics.push(validation_diagnostic( - "logging.application_state_root", - "logging.application_state_root must not be the filesystem root \"/\"", - )); - } - } - - // Reject paths that escape via symlink-like patterns. - if path_str.contains("..") && (path_str.starts_with("/..") || path_str.contains("/../")) { - diagnostics.push(validation_diagnostic( - "logging.application_state_root", - "logging.application_state_root must not contain directory traversal sequences", - )); - } - } + use std::path::Component; + + // Reject any directory traversal component, wherever it appears. + if root + .components() + .any(|component| component == Component::ParentDir) + { + diagnostics.push(validation_diagnostic( + "logging.application_state_root", + "logging.application_state_root must not contain directory traversal sequences", + )); + } + + if root.parent().is_none() { + diagnostics.push(validation_diagnostic( + "logging.application_state_root", + "logging.application_state_root must not be the filesystem root \"/\"", + )); + } else { + // Reject system directories and anything nested under them. + for forbidden in ["/etc", "/dev", "/proc", "/sys"] { + if root.starts_with(forbidden) { + diagnostics.push(validation_diagnostic( + "logging.application_state_root", + format!( + "logging.application_state_root must not target system directories; rejecting path under \"{forbidden}\"" + ), + )); + } + } + }
Path::starts_withcompares whole components, so"/etc"and"/etc/state"both match while"/etcetera"does not.Add test coverage for
"/etc","a","/", and"state/.."alongside the existing defaults test.🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the rest with a brief reason, keep changes minimal, and validate. In `@crates/mesh-llm-config/src/validate.rs` around lines 479 - 505, Replace the raw string checks in the application-state root validation block with Path component-based checks using Path::starts_with and component-aware traversal detection. Ensure "/" receives the dedicated filesystem-root diagnostic, bare system directories and their descendants are rejected, valid relative paths such as "a" are accepted, and traversal components including ".." and "state/.." are rejected without matching names like "etcetera". Extend the existing defaults validation tests to cover "/etc", "a", "/", and "state/..".
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: b88505a1-dfb8-42a2-a635-24eae5b910c7
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (49)
.agents/skills/manage-ci/references/current-inventory.mdCargo.tomlci/ci.mdcrates/mesh-llm-config/src/lib.rscrates/mesh-llm-config/src/model.rscrates/mesh-llm-config/src/model/built_in_schema.rscrates/mesh-llm-config/src/validate.rscrates/mesh-llm-events/Cargo.tomlcrates/mesh-llm-events/src/lib.rscrates/mesh-llm-events/src/logging/artifacts.rscrates/mesh-llm-events/src/logging/envelope.rscrates/mesh-llm-events/src/logging/events.rscrates/mesh-llm-events/src/logging/identifiers.rscrates/mesh-llm-events/src/logging/lifecycle.rscrates/mesh-llm-events/src/logging/mod.rscrates/mesh-llm-events/src/logging/proxy.rscrates/mesh-llm-events/src/logging/replay.rscrates/mesh-llm-events/src/logging/summaries.rscrates/mesh-llm-events/src/logging/tests.rscrates/mesh-llm-host-runtime/src/lib.rscrates/mesh-llm-host-runtime/src/logging/bus.rscrates/mesh-llm-host-runtime/src/logging/foundation.rscrates/mesh-llm-host-runtime/src/logging/lifecycle.rscrates/mesh-llm-host-runtime/src/logging/mod.rscrates/mesh-llm-host-runtime/src/logging/policy.rscrates/mesh-llm-host-runtime/src/logging/registry.rscrates/mesh-llm-host-runtime/src/logging/sequences.rscrates/mesh-llm-host-runtime/src/logging/service.rscrates/mesh-llm-host-runtime/src/logging/service_tests.rscrates/mesh-llm-host-runtime/src/logging/writer.rscrates/mesh-llm-host-runtime/src/mesh/owner_control/mod.rscrates/mesh-llm-host-runtime/src/mesh/plugin_config.rscrates/mesh-llm-host-runtime/src/protocol/config_tests.rscrates/mesh-llm-host-runtime/src/protocol/convert.rscrates/mesh-llm-host-runtime/src/protocol/tests/config.rscrates/mesh-llm-host-runtime/src/runtime/config_state.rscrates/mesh-llm-log-store/Cargo.tomlcrates/mesh-llm-log-store/src/artifacts.rscrates/mesh-llm-log-store/src/artifacts_tests.rscrates/mesh-llm-log-store/src/cursor.rscrates/mesh-llm-log-store/src/error.rscrates/mesh-llm-log-store/src/lib.rscrates/mesh-llm-log-store/src/migrations.rscrates/mesh-llm-log-store/src/repositories.rscrates/mesh-llm-log-store/src/store.rscrates/mesh-llm-log-store/src/tests.rscrates/mesh-llm/src/commands/runtime.rsscripts/affected-crates.shscripts/plan-clippy-batches.sh
| pub fn spawn(&self) -> bool { | ||
| // Prevent double-spawn. | ||
| if self.spawned.swap(true, Ordering::AcqRel) { | ||
| return false; | ||
| } | ||
|
|
||
| let bus = Arc::clone(&self.bus); | ||
| let sink_opt = self.sink.clone(); | ||
|
|
||
| let (tx, mut rx) = mpsc::channel::<WorkerMessage>(64); | ||
| let handle = WorkerHandle { tx }; | ||
|
|
||
| // Store the handle for shutdown. | ||
| let mut wh_guard = self.worker_handle.blocking_lock(); | ||
| *wh_guard = Some(handle); | ||
| drop(wh_guard); | ||
|
|
||
| tokio::spawn(async move { | ||
| while let Some(msg) = rx.recv().await { | ||
| match msg { | ||
| WorkerMessage::PersistBusEntry(entry) => { | ||
| // Parse the bus entry and persist via sink. | ||
| if let Some(sink) = &sink_opt { | ||
| // Best-effort: failures are absorbed by fail-open writer. | ||
| let _ = Self::process_bus_entry(&bus, sink.as_ref(), &entry).await; | ||
| } | ||
| } | ||
| WorkerMessage::Shutdown => { | ||
| break; | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // On drop or shutdown, the channel closes and we exit cleanly. No leaked tasks. | ||
| }); | ||
|
|
||
| true | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
spawn() blocks on a Tokio mutex, which forces a panic risk in production and a lock workaround in tests. LoggingService::spawn calls TokioMutex::blocking_lock while it also requires a Tokio runtime for tokio::spawn; blocking_lock panics inside a runtime, so no caller can satisfy both requirements.
crates/mesh-llm-host-runtime/src/logging/service.rs#L204-L241: storeworker_handlein astd::sync::Mutex, or makespawnasync and use.lock().await; updateshutdown()to match.crates/mesh-llm-host-runtime/src/logging/service_tests.rs#L603-L656: after the fix, drop theArc<std::sync::Mutex<LoggingService>>wrapper, thespawn_blockingcalls, and the#[allow(clippy::await_holding_lock)]attributes, and callsvc.spawn()directly inside#[tokio::test].
📍 Affects 2 files
crates/mesh-llm-host-runtime/src/logging/service.rs#L204-L241(this comment)crates/mesh-llm-host-runtime/src/logging/service_tests.rs#L603-L656
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/mesh-llm-host-runtime/src/logging/service.rs` around lines 204 - 241,
Replace the Tokio mutex used for LoggingService.worker_handle with a
std::sync::Mutex, and update shutdown() to use the matching synchronous lock so
spawn() can safely call worker_handle.lock() before tokio::spawn. In
crates/mesh-llm-host-runtime/src/logging/service_tests.rs lines 603-656, remove
the Arc<std::sync::Mutex<LoggingService>> wrapper, spawn_blocking calls, and
#[allow(clippy::await_holding_lock)] attributes, then invoke svc.spawn()
directly within the Tokio tests.
| pub struct RecursionGuard { | ||
| /// Per-thread flag checked before entering the error-record path. When `true`, we are already inside an error record and must not re-enter. | ||
| in_error_path: std::cell::Cell<bool>, | ||
|
|
||
| /// Atomic global guard for cross-thread recursion detection (belt-and-suspenders). | ||
| depth: Arc<AtomicU64>, | ||
|
|
||
| /// Global atomic flag preventing any thread from entering when another is already in the error path. | ||
| global_in_error: Arc<AtomicBool>, | ||
| } |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
The manual unsafe impl Send/Sync introduces a data race.
RecursionGuard holds std::cell::Cell<bool>, which is not Sync. The comment states that the Cell is accessed only from its owning thread, but the guard is shared across threads by design: FailOpenWriter stores Arc<RecursionGuard>, LoggingService stores Arc<FailOpenWriter> and is used from the worker task and request paths, and the test at lines 268-292 clones the Arc into a second thread and calls try_enter_error_path() there. That call reads and writes the same Cell<bool> from two threads without synchronization, which is undefined behavior in Rust and is not detectable by the borrow checker once unsafe impl suppresses it.
Replace the Cell<bool> with an AtomicBool, or move the per-thread flag to a thread_local!. Then delete both unsafe impl lines and let the compiler derive the auto traits.
🔒 Proposed fix
pub struct RecursionGuard {
- /// Per-thread flag checked before entering the error-record path. When `true`, we are already inside an error record and must not re-enter.
- in_error_path: std::cell::Cell<bool>,
+ /// Flag checked before entering the error-record path. When `true`, an error record is already in progress.
+ in_error_path: AtomicBool,
@@
-// RecursionGuard is Send + Sync: Cell<bool> is only accessed from its own thread, AtomicBool handles cross-thread.
-unsafe impl Send for RecursionGuard {}
-unsafe impl Sync for RecursionGuard {}Update the accessors accordingly:
pub fn try_enter_error_path(&self) -> bool {
self.global_in_error
.compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
.is_ok()
}
pub fn exit_error_path(&self) {
self.global_in_error.store(false, Ordering::Release);
}With a single AtomicBool, the in_error_path field becomes redundant, because global_in_error already enforces the invariant.
Also applies to: 68-70
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/mesh-llm-host-runtime/src/logging/writer.rs` around lines 9 - 18,
Replace RecursionGuard’s Cell<bool> and redundant per-thread tracking with the
existing global_in_error AtomicBool, update try_enter_error_path and
exit_error_path to use the specified atomic operations, and remove both unsafe
Send/Sync implementations so auto traits apply safely.
| fn logging_changes_require_restart(old: &LoggingConfig, new: &LoggingConfig) -> bool { | ||
| if old.enabled != new.enabled { | ||
| return field_requires_restart("enabled"); | ||
| } | ||
| if old.application_state_root != new.application_state_root { | ||
| return field_requires_restart("application_state_root"); | ||
| } | ||
| if old.summary_line_limit != new.summary_line_limit { | ||
| return field_requires_restart("summary_line_limit"); | ||
| } | ||
| if old.event_buffer_size != new.event_buffer_size { | ||
| return field_requires_restart("event_buffer_size"); | ||
| } | ||
| // Dynamic fields: retention/replay limits apply atomically at runtime. | ||
| if old.retention_ttl_secs != new.retention_ttl_secs { | ||
| return field_requires_restart("retention_ttl_secs"); | ||
| } | ||
| if old.replay_capacity != new.replay_capacity { | ||
| return field_requires_restart("replay_capacity"); | ||
| } | ||
| if old.queue_capacity != new.queue_capacity { | ||
| return field_requires_restart("queue_capacity"); | ||
| } | ||
| if old.artifact != new.artifact { | ||
| return field_requires_restart("artifact.capture_mode") | ||
| || field_requires_restart("artifact.byte_limit_bytes") | ||
| || field_requires_restart("artifact.aggregate_limit_bytes"); | ||
| } | ||
| if old.export_limit_bytes != new.export_limit_bytes { | ||
| return field_requires_restart("export_limit_bytes"); | ||
| } | ||
| if old.cleanup_cadence_secs != new.cleanup_cadence_secs { | ||
| return field_requires_restart("cleanup_cadence_secs"); | ||
| } | ||
| // Webhook settings. | ||
| if old.webhook.enabled != new.webhook.enabled { | ||
| return field_requires_restart("webhook.enabled"); | ||
| } | ||
| if old.webhook.url != new.webhook.url { | ||
| return field_requires_restart("webhook.url"); | ||
| } | ||
| if old.webhook.max_attempts != new.webhook.max_attempts { | ||
| return field_requires_restart("webhook.max_attempts"); | ||
| } | ||
| if old.webhook.timeout_secs != new.webhook.timeout_secs { | ||
| return field_requires_restart("webhook.timeout_secs"); | ||
| } | ||
| if old.webhook.dead_letter_retention_secs != new.webhook.dead_letter_retention_secs { | ||
| return field_requires_restart("webhook.dead_letter_retention_secs"); | ||
| } | ||
|
|
||
| false | ||
| } |
There was a problem hiding this comment.
🎯 Functional Correctness | 🔴 Critical | ⚡ Quick win
The early return masks a restart-required change when several logging fields change together.
Each comparison returns immediately. If a dynamic field and a static field both change in one apply, the first differing field decides the result.
Concrete case: an operator changes retention_ttl_secs (DynamicApply) and enabled (StaticOnLoad) in the same edit. The check at Line 160 matches first and returns field_requires_restart("retention_ttl_secs"), which is false. apply then returns ApplyResult::Applied, and the operator is never told that a restart is required. The logging subsystem keeps running with the old enabled value.
Evaluate every changed field and combine the results.
🐛 Proposed fix: accumulate instead of returning early
fn logging_changes_require_restart(old: &LoggingConfig, new: &LoggingConfig) -> bool {
- if old.enabled != new.enabled {
- return field_requires_restart("enabled");
- }
- if old.application_state_root != new.application_state_root {
- return field_requires_restart("application_state_root");
- }
- if old.summary_line_limit != new.summary_line_limit {
- return field_requires_restart("summary_line_limit");
- }
- if old.event_buffer_size != new.event_buffer_size {
- return field_requires_restart("event_buffer_size");
- }
- // Dynamic fields: retention/replay limits apply atomically at runtime.
- if old.retention_ttl_secs != new.retention_ttl_secs {
- return field_requires_restart("retention_ttl_secs");
- }
- if old.replay_capacity != new.replay_capacity {
- return field_requires_restart("replay_capacity");
- }
- if old.queue_capacity != new.queue_capacity {
- return field_requires_restart("queue_capacity");
- }
- if old.artifact != new.artifact {
- return field_requires_restart("artifact.capture_mode")
- || field_requires_restart("artifact.byte_limit_bytes")
- || field_requires_restart("artifact.aggregate_limit_bytes");
- }
- if old.export_limit_bytes != new.export_limit_bytes {
- return field_requires_restart("export_limit_bytes");
- }
- if old.cleanup_cadence_secs != new.cleanup_cadence_secs {
- return field_requires_restart("cleanup_cadence_secs");
- }
- // Webhook settings.
- if old.webhook.enabled != new.webhook.enabled {
- return field_requires_restart("webhook.enabled");
- }
- if old.webhook.url != new.webhook.url {
- return field_requires_restart("webhook.url");
- }
- if old.webhook.max_attempts != new.webhook.max_attempts {
- return field_requires_restart("webhook.max_attempts");
- }
- if old.webhook.timeout_secs != new.webhook.timeout_secs {
- return field_requires_restart("webhook.timeout_secs");
- }
- if old.webhook.dead_letter_retention_secs != new.webhook.dead_letter_retention_secs {
- return field_requires_restart("webhook.dead_letter_retention_secs");
- }
-
- false
+ let changed: [(&str, bool); 15] = [
+ ("enabled", old.enabled != new.enabled),
+ (
+ "application_state_root",
+ old.application_state_root != new.application_state_root,
+ ),
+ (
+ "summary_line_limit",
+ old.summary_line_limit != new.summary_line_limit,
+ ),
+ (
+ "event_buffer_size",
+ old.event_buffer_size != new.event_buffer_size,
+ ),
+ (
+ "retention_ttl_secs",
+ old.retention_ttl_secs != new.retention_ttl_secs,
+ ),
+ ("replay_capacity", old.replay_capacity != new.replay_capacity),
+ ("queue_capacity", old.queue_capacity != new.queue_capacity),
+ (
+ "artifact.capture_mode",
+ old.artifact.capture_mode != new.artifact.capture_mode,
+ ),
+ (
+ "artifact.byte_limit_bytes",
+ old.artifact.byte_limit_bytes != new.artifact.byte_limit_bytes,
+ ),
+ (
+ "artifact.aggregate_limit_bytes",
+ old.artifact.aggregate_limit_bytes != new.artifact.aggregate_limit_bytes,
+ ),
+ (
+ "export_limit_bytes",
+ old.export_limit_bytes != new.export_limit_bytes,
+ ),
+ (
+ "cleanup_cadence_secs",
+ old.cleanup_cadence_secs != new.cleanup_cadence_secs,
+ ),
+ (
+ "webhook.enabled",
+ old.webhook.enabled != new.webhook.enabled,
+ ),
+ ("webhook.url", old.webhook.url != new.webhook.url),
+ (
+ "webhook.dead_letter_retention_secs",
+ old.webhook.dead_letter_retention_secs != new.webhook.dead_letter_retention_secs,
+ ),
+ ];
+
+ changed
+ .into_iter()
+ .any(|(field, differs)| differs && field_requires_restart(field))
}The proposed form also removes the imprecise old.artifact != new.artifact branch, which currently ORs the restart scope of all three artifact fields regardless of which one changed. Add the remaining webhook.max_attempts and webhook.timeout_secs entries to the array.
Add a test that changes a dynamic field and a static field in the same apply and asserts AppliedWithRestartRequired.
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/mesh-llm-host-runtime/src/runtime/config_state.rs` around lines 146 -
198, Update logging_changes_require_restart to evaluate every changed field and
accumulate whether any changed field requires restart instead of returning on
the first difference. Replace the broad old.artifact != new.artifact check with
per-field comparisons for the artifact settings, and include
webhook.max_attempts and webhook.timeout_secs in the accumulated checks. Add a
test combining a dynamic field such as retention_ttl_secs with the static
enabled field and assert ApplyResult::AppliedWithRestartRequired.
| fn truncate_content(content: &[u8], byte_limit: usize) -> (Vec<u8>, bool) { | ||
| if content.len() <= byte_limit { | ||
| return (content.to_vec(), false); | ||
| } | ||
|
|
||
| // UTF-8 safe truncation: find last valid char boundary within limit. | ||
| let mut truncated = &content[..byte_limit]; | ||
| while !truncated.is_empty() { | ||
| match std::str::from_utf8(truncated) { | ||
| Ok(_) => break, | ||
| Err(e) => { | ||
| truncated = &truncated[..truncated.len().saturating_sub(e.error_len().unwrap_or(1))] | ||
| } | ||
| } | ||
| } | ||
|
|
||
| (truncated.to_vec(), true) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
truncate_content discards all content when the input is not valid UTF-8.
The loop calls std::str::from_utf8(truncated). On error it removes e.error_len() bytes from the end of the slice. error_len() describes the invalid sequence at e.valid_up_to(), which is near the start for binary content. Removing bytes from the end never fixes an invalid byte at the start, so the loop keeps shrinking the slice until truncated.is_empty() ends it.
For a binary artifact, the function therefore returns an empty Vec and reports was_truncated = true. write_artifact then stores a zero-byte file, records a SHA-256 of the empty input, and returns a receipt with bytes: 0. The artifact content is lost silently.
The loop also runs byte_limit iterations, and each iteration scans the slice, so the cost is O(byte_limit²).
The existing test does not catch this. artifacts_tests.rs Line 295 uses vec![0u8; 1024], and a run of NUL bytes is valid UTF-8, so from_utf8 succeeds on the first iteration.
Cut once at the last valid boundary, and only when the input is UTF-8 text.
🐛 Proposed fix
fn truncate_content(content: &[u8], byte_limit: usize) -> (Vec<u8>, bool) {
if content.len() <= byte_limit {
return (content.to_vec(), false);
}
- // UTF-8 safe truncation: find last valid char boundary within limit.
- let mut truncated = &content[..byte_limit];
- while !truncated.is_empty() {
- match std::str::from_utf8(truncated) {
- Ok(_) => break,
- Err(e) => {
- truncated = &truncated[..truncated.len().saturating_sub(e.error_len().unwrap_or(1))]
- }
- }
- }
-
- (truncated.to_vec(), true)
+ let head = &content[..byte_limit];
+ let end = match std::str::from_utf8(head) {
+ // Already ends on a character boundary.
+ Ok(_) => byte_limit,
+ Err(e) => match e.error_len() {
+ // A truncated multi-byte character at the end: cut before it.
+ None => e.valid_up_to(),
+ // Genuinely invalid UTF-8, so the content is binary. Cut at the limit.
+ Some(_) => byte_limit,
+ },
+ };
+
+ (head[..end].to_vec(), true)
}Add a test that truncates binary content and asserts the stored length.
#[test]
fn truncate_binary_content_keeps_bytes() {
let content = vec![0xFFu8; 1024];
let (out, truncated) = truncate_content(&content, 64);
assert!(truncated);
assert_eq!(out.len(), 64, "binary content must not be discarded");
}📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
| fn truncate_content(content: &[u8], byte_limit: usize) -> (Vec<u8>, bool) { | |
| if content.len() <= byte_limit { | |
| return (content.to_vec(), false); | |
| } | |
| // UTF-8 safe truncation: find last valid char boundary within limit. | |
| let mut truncated = &content[..byte_limit]; | |
| while !truncated.is_empty() { | |
| match std::str::from_utf8(truncated) { | |
| Ok(_) => break, | |
| Err(e) => { | |
| truncated = &truncated[..truncated.len().saturating_sub(e.error_len().unwrap_or(1))] | |
| } | |
| } | |
| } | |
| (truncated.to_vec(), true) | |
| } | |
| fn truncate_content(content: &[u8], byte_limit: usize) -> (Vec<u8>, bool) { | |
| if content.len() <= byte_limit { | |
| return (content.to_vec(), false); | |
| } | |
| let head = &content[..byte_limit]; | |
| let end = match std::str::from_utf8(head) { | |
| // Already ends on a character boundary. | |
| Ok(_) => byte_limit, | |
| Err(e) => match e.error_len() { | |
| // A truncated multi-byte character at the end: cut before it. | |
| None => e.valid_up_to(), | |
| // Genuinely invalid UTF-8, so the content is binary. Cut at the limit. | |
| Some(_) => byte_limit, | |
| }, | |
| }; | |
| (head[..end].to_vec(), true) | |
| } |
🤖 Prompt for AI Agents
Verify each finding against current code. Fix only still-valid issues, skip the
rest with a brief reason, keep changes minimal, and validate.
In `@crates/mesh-llm-log-store/src/artifacts.rs` around lines 102 - 119, Update
truncate_content to avoid UTF-8 boundary processing for binary or invalid UTF-8
input: attempt UTF-8 parsing once, and return the first byte_limit bytes
unchanged when parsing fails. For valid UTF-8 text, truncate once at the last
valid character boundary at or before byte_limit, preserving the truncated flag
and avoiding the current shrinking loop. Add a regression test covering invalid
binary bytes and asserting the output length remains byte_limit.
ndizazzo
changed the title
|
Can you get a README.md with mermaid diagrams for the new crates? Having an agent code review now. |
i386
left a comment
i386
left a comment
There was a problem hiding this comment.
Found two blocking issues:
-
initialize_host_runtime_with_configalways callsLoggingFoundation::init(true, None)(host-runtimesrc/lib.rs:139-148). This ignores the validated[logging]configuration: an operator who setslogging.enabled = falsestill gets a logging directory created, andlogging.application_state_rootis ignored. Please load the logging config regardless of the dynamic-native-runtime feature and passconfig.logging.enabledplusconfig.logging.application_state_root.as_ref()into initialization. -
LoggingService::spawncreates anmpscworker, butenqueue_eventonly pushes the serialized event intoReplayBus; no path sendsWorkerMessage::PersistBusEntryto the worker (host-runtimesrc/logging/service.rs:203-279). Consequently, a spawned service with a sink never calls the sink for any enqueued event. Please connect the producer/worker path (and add a test that an enqueued event is persisted afterspawn).
|
Closing - I have a revised PR stack coming with a stronger setup |
2 participants
Stack
This is PR 1 of planned 3. It targets
main; canonical API child: #1140, targetinglogging-foundation-canonical. PR3 is deferred until Todo 17 completes and Todo 18 creates a non-empty frontend commit.What Changed
mesh-llm-log-storein affected-crate and Clippy fail-open routing.Migration and Rollback
The
[logging]configuration section is additive. Existing config files retain bounded metadata-only defaults. Disable logging or revert this PR to stop durable initialization; the feature adds no mesh wire fields or migrations outside the local log-store database.Protocol and Privacy
No mesh protocol, ALPN, protobuf, or Skippy ABI changes. Logging remains local-only; metadata-only capture is default and centralized redaction applies before persistence.
Validation
Local validation passed:
cargo fmt --all --check; events, log-store, config, and host logging tests; host/log-store/shipped-binary check and warning-denying Clippy;cargo run --locked -p xtask -- repo-consistency ci-crate-lists; shellcheck for changed CI scripts; full host-runtime library suite (1967 passed, 8 ignored); andgit diff --check. GitHub checks have not yet completed.Rollback Reference
PR #1133 remains preserved as the superseded recovery reference until this replacement is reviewed.