Add privacy-safe logging foundations - #1133
Conversation
Tasks 1-6 of logging-capabilities plan: Task 1: Define versioned canonical logging contracts (events, envelope, lifecycle, proxy, replay, artifacts) in mesh-llm-events. Task 2: Add operator logging configuration and centralized privacy policy (logging settings, redaction/truncation policy). Task 3: Create mesh-llm-log-store with SQLite migrations and deterministic repositories (summaries, events, artifact pointers). Task 4: Implement secure versioned artifact storage (atomic writes, startup recovery, schema V2 forward-only migrations). Task 5: Build bounded runtime bus, active registry, lifecycle guard, and fail-open writer in mesh-llm-host-runtime/src/logging/. 107/107 tests pass. Task 6: Wire foundation configuration and compatibility gates: - 6a: Fix foundation.rs compile errors + broken test (9/9 pass) - 6b: Wire LoggingFoundation into host init (lib.rs) fail-open + health accessors - 6c: Add AppliedWithRestartRequired classification in config_state.rs for non-dynamic logging setting changes (23 tests pass) - 6d: Verify schema fixtures consistent (9 snapshot tests pass) - 6e: Full verification — fmt/check/test/clippy clean, no proto/ABI diff All 118 logging tests + 23 config_state tests pass. Clippy clean. No mesh/protobuf fields added. No existing telemetry activation changed.
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>
📝 WalkthroughWalkthroughAdded configurable canonical logging across mesh configuration, event contracts, host runtime, and a new SQLite-backed log store. The change includes lifecycle tracking, privacy filtering, replay buffering, artifact storage, pagination, migrations, fail-open behavior, and restart-aware configuration application. ChangesLogging configuration and apply behavior
Canonical event contracts
Host-runtime logging service
SQLite log store and artifact persistence
Estimated code review effort: 5 (Critical) | ~120 minutes Possibly related PRs
Suggested labels: 🚥 Pre-merge checks | ✅ 5✅ Passed checks (5 passed)
✨ Finishing Touches 💡 1🛠️ Fix failing CI checks 💡
📝 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 |
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 5
Note
Due to the large number of review comments, Critical severity comments were prioritized as inline comments.
🟠 Major comments (16)
crates/mesh-llm-config/src/validate.rs-467-505 (1)
467-505: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winClose the traversal gap and take
&Path.Three problems exist in this function:
- The traversal check misses a trailing
..and relative traversal."/var/lib/mesh/.."and"../state"both pass, so the root can still escape the intended directory.- The branch at Line 490-495 is unreachable. The
path_str == *prefixcondition already matches"/", so the filesystem-root message never appears.- The parameter type is
&std::path::PathBuf.clippy::ptr_argis warn-by-default, and the coding guidelines require touched Rust code to be warning-free.Inspect path components instead of matching substrings.
🛡️ Proposed fix
-fn validate_application_state_root( - root: &std::path::PathBuf, - diagnostics: &mut Vec<ConfigDiagnostic>, -) { +fn validate_application_state_root( + root: &std::path::Path, + diagnostics: &mut Vec<ConfigDiagnostic>, +) { if root.as_os_str().is_empty() { diagnostics.push(validation_diagnostic( "logging.application_state_root", "logging.application_state_root must not be empty", )); return; } + if root.components().any(|component| { + matches!(component, std::path::Component::ParentDir) + }) { + diagnostics.push(validation_diagnostic( + "logging.application_state_root", + "logging.application_state_root must not contain directory traversal sequences", + )); + } + + if root == std::path::Path::new("/") { + diagnostics.push(validation_diagnostic( + "logging.application_state_root", + "logging.application_state_root must not be the filesystem root \"/\"", + )); + } + // 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 != "/") { + if *prefix != "/" && path_str.starts_with(prefix) { diagnostics.push(validation_diagnostic( "logging.application_state_root", format!( "logging.application_state_root must not target system directories; rejecting path starting with \"{prefix}\"" ), )); } } - - // 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", - )); - } }As per coding guidelines: "Do not leave compiler or lint warnings in touched Rust code; fix warnings rather than using
#[allow(...)]".#!/bin/bash # Description: Confirm clippy is run with warnings denied and no crate-level allow for ptr_arg. rg -n 'clippy' --glob '*.toml' --glob '.github/**' -C2 rg -n 'ptr_arg|allow\(clippy' crates/mesh-llm-config/src🤖 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 - 505, Update validate_application_state_root to accept &std::path::Path instead of &PathBuf, and inspect root.components() to reject any ParentDir component, including trailing and relative traversal such as "/var/lib/mesh/.." and "../state". Simplify the forbidden-directory validation so the filesystem root "/" is handled explicitly before broader system-prefix checks, ensuring its dedicated diagnostic is reachable and no clippy warnings remain.Source: Coding guidelines
crates/mesh-llm-host-runtime/src/runtime/config_state.rs-140-198 (1)
140-198: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
logging_changes_require_restartreturns on the first changed field, so it can miss a restart.Each comparison uses
return. The result reflects only the first field that differs. If a dynamic field and a static field change in the same apply, the dynamic branch wins and the caller receivesAppliedinstead ofAppliedWithRestartRequired.Concrete case: change
retention_ttl_secs(DynamicApply, checked at Line 160) andexport_limit_bytes(StaticOnLoad, checked at Line 174) together. The function returnsfalse, and the operator never learns that a restart is required.
field_requires_restartalso returnsfalsewhen the schema descriptor is missing. A new field without a descriptor then silently requires no restart. Default totruefor unknown paths.Accumulate the verdict over every changed field, and add a test for the mixed dynamic-plus-static case.
🐛 Proposed fix
fn field_requires_restart(field_name: &str) -> bool { let path = ConfigPath::from_fields(["logging", field_name]); - built_in_config_schema_descriptor(&path) - .is_some_and(|schema| schema.apply_mode != SchemaApplyMode::DynamicApply) + // Unknown paths are treated conservatively as restart-requiring. + built_in_config_schema_descriptor(&path) + .is_none_or(|schema| schema.apply_mode != SchemaApplyMode::DynamicApply) } 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: [(bool, &str); 15] = [ + (old.enabled != new.enabled, "enabled"), + ( + old.application_state_root != new.application_state_root, + "application_state_root", + ), + (old.summary_line_limit != new.summary_line_limit, "summary_line_limit"), + (old.event_buffer_size != new.event_buffer_size, "event_buffer_size"), + (old.retention_ttl_secs != new.retention_ttl_secs, "retention_ttl_secs"), + (old.replay_capacity != new.replay_capacity, "replay_capacity"), + (old.queue_capacity != new.queue_capacity, "queue_capacity"), + ( + old.artifact.capture_mode != new.artifact.capture_mode, + "artifact.capture_mode", + ), + ( + old.artifact.byte_limit_bytes != new.artifact.byte_limit_bytes, + "artifact.byte_limit_bytes", + ), + ( + old.artifact.aggregate_limit_bytes != new.artifact.aggregate_limit_bytes, + "artifact.aggregate_limit_bytes", + ), + (old.export_limit_bytes != new.export_limit_bytes, "export_limit_bytes"), + (old.cleanup_cadence_secs != new.cleanup_cadence_secs, "cleanup_cadence_secs"), + (old.webhook.enabled != new.webhook.enabled, "webhook.enabled"), + (old.webhook.url != new.webhook.url, "webhook.url"), + ( + old.webhook.dead_letter_retention_secs != new.webhook.dead_letter_retention_secs, + "webhook.dead_letter_retention_secs", + ), + ]; + + changed + .iter() + .any(|(differs, field)| *differs && field_requires_restart(field)) + || (old.webhook.max_attempts != new.webhook.max_attempts + && field_requires_restart("webhook.max_attempts")) + || (old.webhook.timeout_secs != new.webhook.timeout_secs + && field_requires_restart("webhook.timeout_secs")) }🤖 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 - 198, Update logging_changes_require_restart to evaluate every changed LoggingConfig field and accumulate whether any field_requires_restart result is true, rather than returning on the first difference; ensure mixed dynamic and static changes require a restart. Change field_requires_restart to default to true when built_in_config_schema_descriptor has no matching schema. Add a test covering simultaneous retention_ttl_secs and export_limit_bytes changes.crates/mesh-llm-log-store/src/repositories.rs-328-334 (1)
328-334: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winDo not put the full payload JSON into
event_type.All three sites set
event_type: payload_json.to_string(). The field name states an event type, but the value is the whole serialized event payload. Two problems follow:
LogStoreErrorimplementsDisplayasduplicate terminal event for summary={} type={}, so the complete payload reaches every log line and error response that formats this error. The payload holds request content and the reserved identity fields (tenant_id,account_id,user_id). That defeats the privacy goal of this PR.- Callers cannot match on the event type, because the value is not a type.
Extract the event type, or pass it in from the caller, which already knows it.
🛡️ Proposed fix
+/// Extract the terminal event type from a payload, without exposing payload content. +fn terminal_event_type(payload_json: &str) -> &'static str { + if payload_json.contains(r#""type":"completed""#) { + "completed" + } else if payload_json.contains(r#""type":"failed""#) { + "failed" + } else if payload_json.contains(r#""type":"rejected""#) { + "rejected" + } else { + "cancelled" + } +}Then replace each
event_type: payload_json.to_string()withevent_type: terminal_event_type(payload_json).to_string().Also applies to: 341-348, 378-383
🤖 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 328 - 334, Replace the full-payload assignment to event_type in all three terminal-duplicate error sites, including the pre-check around check_existing_terminal_raw and the other duplicate handling paths. Use the existing terminal_event_type(payload_json) extraction so LogStoreError::DuplicateTerminalEvent receives only the event type, preserving its Display output without exposing request data.crates/mesh-llm-log-store/src/store.rs-16-21 (1)
16-21: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winUse sub-second timestamp precision.
SystemClock::nowformats to whole seconds. All ordering keys in this crate use these strings:created_at,occurred_at, and the pagination cursors incrates/mesh-llm-log-store/src/repositories.rs. Two consequences follow:
list_events_for_summary(repositories.rs line 520) orders only byoccurred_at ASC. Events emitted in the same second get an arbitrary order, so a lifecycle timeline can showcompletedbeforeadmitted.- Keyset pagination collapses many rows into one timestamp bucket, so it relies fully on the id tiebreaker, which is not time-ordered.
A logging pipeline needs millisecond or finer resolution.
🐛 Proposed fix
impl Clock for SystemClock { fn now(&self) -> String { let dt = chrono::Utc::now(); - format!("{}", dt.format("%Y-%m-%dT%H:%M:%SZ")) + dt.format("%Y-%m-%dT%H:%M:%S%.3fZ").to_string() } }Note that lexicographic comparison stays correct with a fixed-width fractional part.
🤖 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 16 - 21, Update SystemClock::now to format UTC timestamps with a fixed-width sub-second fractional component, preferably milliseconds or finer, while preserving the existing lexicographically sortable UTC layout. Ensure the resulting precision is used consistently by created_at, occurred_at, and pagination cursor values through the existing Clock implementation.crates/mesh-llm-log-store/src/artifacts.rs-63-80 (1)
63-80: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winReject
tmpas arequest_idsegment.
sanitize_segmentallows the valuetmp. The artifact layout places request directories directly underroot, androot/tmpis the staging directory created at lines 138 and 303. A request withrequest_id == "tmp"therefore writes artifacts into the staging directory.cleanup_orphan_tempsdeletes every entry in that directory on the nextopen, so the artifacts disappear while their pointer rows survive.walk_top_level_dirsalso skipstmp, soremove_unreferenced_filesnever inspects those files.Request identifiers arrive from the runtime, so an operator cannot rely on them avoiding this name.
🛡️ Proposed fix
+/// Directory name reserved for staging writes. +const RESERVED_SEGMENT: &str = "tmp"; + /// Reject path segments containing / \ NUL or standalone ".". fn sanitize_segment(segment: &str) -> Result<(), LogStoreError> { - if segment.is_empty() || segment == "." || segment == ".." { + if segment.is_empty() + || segment == "." + || segment == ".." + || segment == RESERVED_SEGMENT + { return Err(LogStoreError::PathUnsafe { segment: segment.to_string(), }); }Consider also rejecting
:so Windows alternate data streams cannot be addressed.🤖 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 63 - 80, Update sanitize_segment to reject the reserved segment name "tmp" and any segment containing ':' in addition to the existing unsafe values and path separators. Preserve the existing LogStoreError::PathUnsafe response and validation behavior for all other segments.crates/mesh-llm-log-store/src/artifacts.rs-864-870 (1)
864-870: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winFix the dead
|| falseand reconsider the Windows privacy gate.Two problems exist here:
std::env::var("MESH_LLM_ALLOW_WEAK_PRIVACY").is_ok() || falsecontains a redundant|| false. Clippy reports this, and the workspace lints deny warnings, so the Windows build fails.- The gate turns a security guarantee into an environment switch. Any process that sets
MESH_LLM_ALLOW_WEAK_PRIVACYto any value, including an empty string, disables the check. Without the variable,ArtifactFileStore::openreturnsPrivacyNotGuaranteedon every Windows host, so artifact logging never starts there. Neither result looks intended.Decide the Windows behavior explicitly. Either restrict the artifact root with a real ACL API, or document that Windows uses the parent directory permissions and remove the unconditional failure.
🐛 Minimal lint fix
- std::env::var("MESH_LLM_ALLOW_WEAK_PRIVACY").is_ok() || false + std::env::var("MESH_LLM_ALLOW_WEAK_PRIVACY").is_ok()Do you want me to open an issue to track the Windows artifact-permission decision?
As per coding guidelines: "Do not leave compiler or lint warnings in touched Rust code".
🤖 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 864 - 870, Update check_windows_privacy to remove the redundant || false and make the Windows privacy policy explicit: either implement actual ACL validation for the artifact root, or rely on documented parent-directory permissions and return the corresponding success result instead of unconditionally failing. Do not use MESH_LLM_ALLOW_WEAK_PRIVACY as an unrestricted security bypass; ensure ArtifactFileStore::open preserves the chosen Windows behavior without compiler or Clippy warnings.Source: Coding guidelines
crates/mesh-llm-log-store/src/store.rs-67-86 (1)
67-86: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winHandle mutex poisoning without a panic and without a wrong error variant.
Two problems exist in the lock handling:
txnmaps aPoisonErrortoLogStoreError::Sqlite(rusqlite::Error::ExecuteReturnedResults). That variant means a statement returned rows. The mapped error misreports the cause and displays assqlite error: Execute returned results - did you mean to call query?.conncalls.expect("connection mutex poisoned"). If one panic occurs while the lock is held, the mutex stays poisoned, so every later store call panics. The PR describes a fail-open logging writer, but this path converts a single panic into a permanent process-wide panic loop.Recover the guard instead, since a poisoned
Connectionremains usable.🛡️ Proposed fix
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)?; @@ 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 67 - 86, Update the mutex handling in txn and conn to recover the guard from a poisoned connection instead of mapping it to a fabricated SQLite error or panicking. Use the recovered Connection guard for both transaction creation and direct access, preserving existing transaction error handling and allowing the logging writer to continue after a prior panic.crates/mesh-llm-log-store/src/cursor.rs-9-14 (1)
9-14: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winUse URL-safe Base64 without padding for cursors.
data_encoding::BASE64uses the standard alphabet, which contains+and/, and it adds=padding. The stack outline states that the next PR adds API and WebSocket surfaces, so these cursors travel in URL query strings. In a query string,+decodes to a space and/and=need percent-encoding. A cursor that a client copies without encoding then failsdecode_cursorwithbase64 decode failed.Change the alphabet now, before any cursor is released to clients.
🐛 Proposed fix
pub fn encode_cursor(occurred_at: &str, id: &str) -> String { let payload = format!("{}|{}", occurred_at, id); // Version byte + base64 of the payload. - let encoded = data_encoding::BASE64.encode(payload.as_bytes()); + let encoded = data_encoding::BASE64URL_NOPAD.encode(payload.as_bytes()); format!("v{CURSOR_VERSION}:{encoded}") }Apply the matching change in
decode_cursorat line 38.Note that
crates/mesh-llm-log-store/src/tests.rsline 394 usesv99:dGVzdA==, which stays a valid unknown-version case.🤖 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/cursor.rs` around lines 9 - 14, Update encode_cursor and the matching decode_cursor implementation to use URL-safe Base64 without padding instead of data_encoding::BASE64. Preserve the existing version prefix and payload format, and keep unknown-version test values such as v99:dGVzdA== valid for version rejection.crates/mesh-llm-log-store/src/migrations.rs-129-147 (1)
129-147: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick winMake migrations atomic with the schema version bump.
Store
user_versiononly after both pending migration batches succeed. A crash afterMIGRATIONS_V2runs but beforePRAGMA user_version = 2leaves the schema in a state where the nextLogStore::openre-runsMIGRATIONS_V2and fails withduplicate column name: media_kind.🤖 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 129 - 147, Update apply_migrations so MIGRATIONS_V1, MIGRATIONS_V2, and the user_version update execute within one transaction, committing only after all pending batches and the schema version bump succeed; roll back on any error so LogStore::open cannot observe partially applied migrations.crates/mesh-llm-log-store/src/lib.rs-8-19 (1)
8-19: 📐 Maintainability & Code Quality | 🟠 Major | ⚡ Quick winExport the row and page types used by public
LogStoremethods.
LogStoreexposes public repository methods whose return types areSummaryRow,LifecycleEventRow,ArtifactPointerRow, andPage<T>. Becauserepositoriesremains private, crates outside this crate cannot type the values returned byget_summary,list_summaries, and the lifecycle/artifact pointer pages. Export these types from the crate root so future consumers can name the API result types.♻️ Proposed change
pub use error::LogStoreError; +pub use repositories::{ArtifactPointerRow, LifecycleEventRow, Page, SummaryRow}; pub use store::{Clock, LogStore, SystemClock as RealClock};🤖 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/lib.rs` around lines 8 - 19, Re-export the public result types used by LogStore methods from the crate root alongside the existing store exports: SummaryRow, LifecycleEventRow, ArtifactPointerRow, and Page. Update the repositories module exports as needed so external crates can name the return types of get_summary, list_summaries, and lifecycle/artifact pointer pagination methods while keeping repositories itself private.crates/mesh-llm-log-store/src/repositories.rs-76-82 (1)
76-82: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick winDistinguish unique-constraint violations from foreign-key violations.
SQLITE_CONSTRAINT is the primary code for both UNIQUE and FOREIGN KEY constraint failures. Matching only
rusqlite::ErrorCode::ConstraintViolationalso maps foreign-key insert failures toAlreadyExistsat these insert sites.Use the unique/primary-keys extended codes (
SQLITE_CONSTRAINT_UNIQUEandSQLITE_CONSTRAINT_PRIMARYKEY) forAlreadyExists; foreign-key failures should returnSqlite. Cover the orphanrequest_idpath with an assertion that it returnsSqlite, notAlreadyExists.🤖 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 76 - 82, Update is_unique_constraint_error to inspect SQLite’s extended error code and return true only for SQLITE_CONSTRAINT_UNIQUE or SQLITE_CONSTRAINT_PRIMARYKEY, while treating foreign-key violations as false so insert callers return Sqlite instead of AlreadyExists. Add an assertion covering the orphan request_id path and verify it produces Sqlite rather than AlreadyExists.crates/mesh-llm-host-runtime/src/logging/service.rs-258-280 (1)
258-280: 🗄️ Data Integrity & Integration | 🟠 Major | 🏗️ Heavy liftNothing forwards bus entries to the persistence worker, so no event is ever persisted.
enqueue_eventpushes the payload ontoself.busat Line 277. The worker task spawned inspawn()only readsWorkerMessagevalues from the mpsc receiver. No code path sendsWorkerMessage::PersistBusEntry, and no code path callsbus.drain()orbus.notified()from the worker. The consequences are:
PersistSinkmethods are never called outside tests.- Entries accumulate on the bus until drop-oldest discards them, so all events are silently lost.
WorkerHandle::sendandWorkerHandle::shutdownare unreachable.If the wiring is intentionally deferred to a later PR in this stack, add an explicit
TODOthat names the follow-up and state in theLoggingServicedoc comment that persistence is not connected yet. Otherwise, drive the worker from the bus.Do you want me to open a tracking issue for the bus-to-worker wiring?
🤖 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 258 - 280, Connect the bus to the persistence worker: update the worker logic created by spawn() to wait for bus.notified(), drain queued entries, and send each one through WorkerMessage::PersistBusEntry so PersistSink is invoked. Preserve nonblocking enqueue_event behavior and ensure shutdown still works through WorkerHandle; if wiring is intentionally deferred instead, add the requested TODO and document the limitation in LoggingService.crates/mesh-llm-host-runtime/src/logging/foundation.rs-149-176 (1)
149-176: 🔒 Security & Privacy | 🟠 Major | ⚡ Quick winHarden the logging state directories: restrict permissions and avoid the shared
/tmpfallback.Two problems exist in the resolution and creation path.
try_create_dirsusescreate_dir_all, which applies the process umask. On most systems this produces0o755. The store directory holds request/response summaries and artifacts, which contain user prompts and headers. Any local user can then read them. Set mode0o700on Unix after creation.- If
dirs::home_dir()returnsNone, the root falls back to the fixed path/tmp/mesh-llm/logging./tmpis world-writable and shared. Another local user can pre-create that path or place a symlink there before the runtime starts. The write test at Line 168 does not detect a pre-existing attacker-owned directory.For the fallback, prefer a per-user location, or treat a missing home directory as an initialization failure and return the unhealthy fail-open instance.
🔒 Proposed permission hardening
fn try_create_dirs(root: &Path, store: &Path, artifacts: &Path) -> bool { for dir in [root, store, artifacts] { match std::fs::create_dir_all(dir) { - Ok(()) => {} + Ok(()) => { + #[cfg(unix)] + { + use std::os::unix::fs::PermissionsExt; + if let Err(e) = + std::fs::set_permissions(dir, std::fs::Permissions::from_mode(0o700)) + { + tracing::debug!( + path = %sanitize_path(dir), + error = %e, + "failed to restrict directory permissions" + ); + return false; + } + } + } Err(e) => { tracing::debug!(path = %sanitize_path(dir), error = %e, "failed to create directory"); return 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-host-runtime/src/logging/foundation.rs` around lines 149 - 176, Harden the logging directory resolution and creation: update the home-directory fallback in the path-resolution function to avoid the shared /tmp/mesh-llm/logging location, preferably returning an initialization failure or using a per-user directory when no home exists. Update try_create_dirs to apply restrictive 0o700 permissions to root, store, and artifacts after creation on Unix, while preserving the existing writability check and failure behavior.crates/mesh-llm-host-runtime/src/lib.rs-139-148 (1)
139-148: 🗄️ Data Integrity & Integration | 🟠 Major | ⚡ Quick win
logging.enabledfrom configuration is ignored, so directories are created even when logging is disabled.
LoggingFoundation::init(true, None)hardcodesenabled = trueandapplication_state_root = None.mesh_llm_config::LoggingConfigalready carriesenabledandapplication_state_root;crates/mesh-llm-host-runtime/src/logging/policy.rsLines 342-343 read both. The documented contract ofLoggingFoundation::disabled()states that no files are created. With this call site, an operator who setslogging.enabled = falsestill gets~/.mesh-llm/logging/storeand~/.mesh-llm/logging/artifactscreated on every start.Read the logging configuration here and pass the real values. Note that
plugin::load_configis currently loaded only under thedynamic-native-runtimefeature, so the load must move out of thatcfgblock.🔧 Sketch of the config-driven initialization
- // Initialize logging foundation (fail-open: serving never blocked by logging failure). - // Config-driven enabled/root are deferred to a later refinement. - let foundation = LoggingFoundation::init(true, None); + // Initialize logging foundation (fail-open: serving never blocked by logging failure). + let logging_config = plugin::load_config(config_path) + .map(|config| config.logging) + .unwrap_or_default(); + let foundation = LoggingFoundation::init( + logging_config.enabled, + logging_config.application_state_root.as_ref(), + );🤖 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 139 - 148, Update the logging initialization around LoggingFoundation::init to load mesh_llm_config::LoggingConfig outside the dynamic-native-runtime cfg block, then pass its enabled and application_state_root values instead of true and None. Preserve fail-open initialization and health warning behavior, and ensure disabled configuration uses the existing LoggingFoundation::disabled() semantics without creating logging directories.crates/mesh-llm-host-runtime/src/logging/policy.rs-126-167 (1)
126-167: 🎯 Functional Correctness | 🟠 Major | ⚡ Quick win
redact_url_queryemits a duplicated#in the fragment.Line 127 already stores the fragment with its leading
#:format!("#{f}"). Lines 157-161 then wrap the same value again:format!("#{fragment}"). For the inputhttps://example.com/p?a=1#fragthe output ishttps://example.com/p?a=1##frag. Every logged URL that has both a query and a fragment is corrupted.The existing test
url_fragment_preservedonly assertscleaned.contains("#"), so it does not detect the defect.🐛 Proposed fix
// Handle fragments. let (query_only, fragment) = match query_part.split_once('#') { Some((q, f)) => (q, format!("#{f}")), None => (query_part, String::new()), }; @@ - let fragment = if fragment.is_empty() { - String::new() - } else { - format!("#{fragment}") - }; if cleaned_params.is_empty() { format!("{base}{fragment}") } else { format!("{}?{}{}", base, cleaned_params.join("&"), fragment) }Add an assertion that pins the exact output, for example
assert_eq!(redact_url_query("https://e.com/p?a=1#frag"), "https://e.com/p?a=1#frag").🤖 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 126 - 167, Fix the fragment handling in redact_url_query so the leading # is added exactly once: preserve the fragment value created by the query_part.split_once('#') branch without wrapping it again before output. Update the url_fragment_preserved test to assert the exact sanitized URL, including a single # between the query and fragment.crates/mesh-llm-host-runtime/src/logging/lifecycle.rs-132-170 (1)
132-170: 🩺 Stability & Availability | 🟠 Major | ⚡ Quick winA race between the CAS and the record write produces a wrong
existingoutcome.The state flag and the record are updated in two steps. Thread A wins the CAS at Line 136 but has not yet stored the outcome at Line 147. Thread B fails the CAS, takes the lock at Line 152, and observes
record.outcome == None. Line 162 then returnsexisting: TerminalOutcome::Dropped(None).The comment at Line 160 states this cannot happen in practice, but the window is a plain time-of-check to time-of-use gap and it is reachable under concurrent terminal attempts. Two consequences follow:
terminate_idempotentcomparesexisting == attempted. With the placeholder it reportsErrfor a genuinely idempotent retry of the same outcome.- Callers log a
droppedterminal state that never occurred.Make the mutex the single source of truth so the flag and the record change together.
🔒 Proposed fix: claim the terminal state under the lock
pub fn terminate(&self, outcome: TerminalOutcome) -> Result<(), DuplicateTerminalError> { - // CAS from ACTIVE to a terminal marker (any non-zero value). We use 1 as "terminal" since we only care about active vs not-active in the flag. - match self - .state_flag - .compare_exchange(STATE_ACTIVE, 1, Ordering::AcqRel, Ordering::Acquire) - { - Ok(_) => { - // Successfully claimed terminal — store the outcome. - 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(_) => { - // Already terminal — return the error with both outcomes. - let record = self.record.lock().expect("terminal record mutex poisoned"); - match &record.outcome { - Some(existing) => Err(DuplicateTerminalError { - existing: existing.clone(), - attempted: outcome, - }), - - None => { - // Edge case: CAS failed but no record yet (shouldn't happen in practice). - // Treat as duplicate with a placeholder. - Err(DuplicateTerminalError { - existing: TerminalOutcome::Dropped(None), - attempted: outcome, - }) - } - } - } - } + // The record mutex is the single source of truth. The flag is a cheap + // read-only hint for `is_active`, published only after the record is set. + 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); + self.state_flag.store(1, Ordering::Release); + Ok(()) }With this shape the placeholder branch disappears, and
is_active()never reports active after a stored outcome.🤖 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 so it acquires the record mutex before checking or changing terminal state, making the mutex-protected record the single source of truth and eliminating the CAS/record-write race. Preserve DuplicateTerminalError with the stored existing outcome for retries, and remove the placeholder None branch; ensure is_active remains consistent with a stored terminal outcome.
🟡 Minor comments (6)
crates/mesh-llm-config/src/validate.rs-526-534 (1)
526-534: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winReject an enabled webhook without a URL.
validate_webhook_confignever comparesenabledwithurl. A configuration withlogging.webhook.enabled = trueand nourlpasses validation, and dispatch can never run. Add a missing-value diagnostic.🐛 Proposed fix
fn validate_webhook_config( config: &crate::LoggingWebhookConfig, diagnostics: &mut Vec<ConfigDiagnostic>, ) { + if config.enabled && config.url.as_ref().is_none_or(|url| url.trim().is_empty()) { + diagnostics.push(validation_diagnostic( + "logging.webhook.url", + "logging.webhook.url must be set when logging.webhook.enabled = true", + )); + } if let Some(ref url) = config.url && let Err(diag) = validate_optional_http_url(Some(url), "logging.webhook.url") { diagnostics.push(diag); }🤖 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 526 - 534, Update validate_webhook_config to emit a missing-value diagnostic when config.enabled is true and config.url is absent, while preserving the existing validate_optional_http_url check for provided URLs. Use the existing diagnostic construction or validation helper conventions in validate.rs.crates/mesh-llm-log-store/src/tests.rs-353-359 (1)
353-359: 🎯 Functional Correctness | 🟡 Minor | ⚡ Quick winThis assertion does not test the query order.
The code copies
all_ids, sorts the copy in descending order, and then asserts onsorted[0]. The sort guarantees that result, so the assertion passes for any order thatlist_summariesreturned. The stated intent, "DESC order means highest ID first", is never checked.Assert on
all_idsdirectly.💚 Proposed fix
- // Verify ordering: DESC on (created_at, request_id), so highest ID first. - let mut sorted = all_ids.clone(); - sorted.sort_unstable_by(|a, b| b.cmp(a)); - assert_eq!( - sorted[0], "same-ts-0004", - "DESC order means highest ID first" - ); + // Verify ordering: DESC on (created_at, request_id), so highest ID first. + let expected: Vec<String> = (0..5u32).rev().map(|i| format!("same-ts-{:04}", i)).collect(); + assert_eq!(all_ids, expected, "pages must be ordered by request_id DESC");🤖 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/tests.rs` around lines 353 - 359, Update the ordering assertion in the test around sorted and all_ids to inspect all_ids directly instead of sorting a copy first. Assert that the first returned ID is "same-ts-0004", preserving the existing DESC-order expectation and message.crates/mesh-llm-log-store/src/artifacts_tests.rs-272-313 (1)
272-313: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winAlign the
write_artifactdocumentation with the tested behavior.The comment at line 294 and the assertions at lines 312-313 state that
byte_limittruncates. The doc comment onwrite_artifactincrates/mesh-llm-log-store/src/artifacts.rsline 208 states that the method "Rejects writes exceeding byte_limit or aggregate_limit before creating any file". Only the aggregate limit rejects. A caller that reads the doc comment will assume it receives an error and will not checkreceipt.truncated, so it can silently store partial content.Update the doc comment on
write_artifact, and rename this test to state the two different behaviors.♻️ Proposed doc change in artifacts.rs
/// Write artifact content to disk with a transactional DB pointer. - /// Rejects writes exceeding byte_limit or aggregate_limit before creating any file. + /// Content over `byte_limit` is truncated and reported through `truncated`. + /// Content over `aggregate_limit` for the request is rejected before any file is created.🤖 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_tests.rs` around lines 272 - 313, Update the write_artifact documentation to state that byte_limit truncates content and reports receipt.truncated, while aggregate_limit rejects the write before creating files. Rename artifact_individual_and_aggregate_limits_rejected_without_partial_files to clearly distinguish truncation from aggregate-limit rejection.crates/mesh-llm-log-store/src/artifacts.rs-587-601 (1)
587-601: 📐 Maintainability & Code Quality | 🟡 Minor | ⚡ Quick winCollapse the identical
ifandelsebranches.Both branches call
fs::remove_file(&path), so thepath.extension()test has no effect. Clippy reportsif_same_then_elsehere, and the workspace lints deny warnings.The loop also ignores subdirectories inside
tmp, so a stale directory keepsfs::remove_dir(&tmp)at line 600 from succeeding.♻️ Proposed change
if let Ok(entries) = fs::read_dir(&tmp) { for entry in entries.flatten() { let path = entry.path(); - if path.extension().is_some_and(|ext| ext == "part") { - let _ = fs::remove_file(&path); - } else { - // Non-.part files in tmp/ are also stale. - let _ = fs::remove_file(&path); - } + // Every entry under tmp/ is stale, whatever its extension. + if path.is_dir() { + let _ = fs::remove_dir_all(&path); + } else { + let _ = fs::remove_file(&path); + } } }As per coding guidelines: "Do not leave compiler or lint warnings in touched Rust code".
🤖 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 587 - 601, In the temporary-directory cleanup loop, remove the redundant path.extension() conditional and delete every entry through the same removal operation. Update the cleanup to handle stale subdirectories as well as files, using the appropriate recursive removal behavior so the subsequent remove_dir(&tmp) can succeed, and leave no lint warnings in this code.Source: Coding guidelines
crates/mesh-llm-log-store/src/migrations.rs-37-42 (1)
37-42: 🗄️ Data Integrity & Integration | 🟡 Minor | ⚡ Quick winStore the lifecycle event type in a dedicated column.
LifecycleEventserializes with#[serde(tag = "type", rename_all = "snake_case")], so the terminal values arecompleted,failed,rejected, andcancelled. The currentLIKEpredicates and duplicate checks work against the default compact JSON shape, but an arbitrary JSON string field can contain"type":"..."anywhere; indexing or querying a typed event column removes this false-positive risk.🤖 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, Update the lifecycle event schema and persistence flow around LifecycleEvent to store its tagged event type in a dedicated column, then change idx_terminal_event_one_per_request and related duplicate checks to compare that column against completed, failed, rejected, and cancelled. Remove the payload_json LIKE predicates and ensure inserts populate the new column from the serialized event type.crates/mesh-llm-host-runtime/src/logging/foundation.rs-192-194 (1)
192-194: 🩺 Stability & Availability | 🟡 Minor | ⚡ Quick win
temp_root()returns one shared path, so parallel tests delete each other's directories.
temp_root()keys only on the process ID.init_enabled_creates_layout,init_idempotent_same_root, andhealth_summary_healthyall call it and all callremove_dir_allon the same path. Cargo runs these tests as threads in one process, so the process ID is identical. One test can remove the directory while another assertsstore_dir_exists_on_disk(). The result is intermittent failures.Give each test a unique subdirectory.
💚 Proposed per-test isolation
- fn temp_root() -> PathBuf { - std::env::temp_dir().join(format!("mesh-llm-log-foundation-{}", std::process::id())) + fn temp_root(case: &str) -> PathBuf { + std::env::temp_dir().join(format!( + "mesh-llm-log-foundation-{}-{case}", + std::process::id() + )) }Then pass a distinct
casename in each test, for exampletemp_root("init_enabled").🤖 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 192 - 194, Update temp_root to accept a per-test case name and append it to the process-specific temporary path, then pass distinct names from init_enabled_creates_layout, init_idempotent_same_root, and health_summary_healthy. Preserve each test’s existing cleanup and assertions while ensuring their directories cannot overlap.
ℹ️ Review info
⚙️ Run configuration
Configuration used: Organization UI
Review profile: CHILL
Plan: Pro Plus
Run ID: cd70b9dd-e8ca-4c3a-a149-7c7c1e9915d8
⛔ Files ignored due to path filters (1)
Cargo.lockis excluded by!**/*.lock
📒 Files selected for processing (45)
Cargo.tomlcrates/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/registry_tests.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.rs
| 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() panics inside a Tokio runtime and cannot run outside one.
Line 217 calls self.worker_handle.blocking_lock(). tokio::sync::Mutex::blocking_lock panics when the current thread is a runtime thread. Line 221 calls tokio::spawn, which panics when no runtime context exists. No single calling context satisfies both conditions:
- Called from an async task or any runtime worker thread:
blocking_lockpanics. - Called from a plain thread outside the runtime:
tokio::spawnpanics.
crates/mesh-llm-host-runtime/src/logging/service_tests.rs Lines 617-625 already work around this by wrapping spawn() in tokio::task::spawn_blocking. That workaround should not be required of production callers.
Make spawn async and use .lock().await, or store the handle in a std::sync::Mutex.
🐛 Proposed fix using a std mutex for the handle
- worker_handle: TokioMutex<Option<WorkerHandle>>,
+ worker_handle: std::sync::Mutex<Option<WorkerHandle>>,- // Store the handle for shutdown.
- let mut wh_guard = self.worker_handle.blocking_lock();
- *wh_guard = Some(handle);
- drop(wh_guard);
+ // Store the handle for shutdown.
+ {
+ let mut wh_guard = self.worker_handle.lock().expect("worker handle mutex poisoned");
+ *wh_guard = Some(handle);
+ }Then update shutdown to take the guard without .await, so no lock is held across an await point.
🤖 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,
Update LoggingService::spawn so it can be called safely from Tokio contexts:
make it async and replace worker_handle.blocking_lock() with lock().await, while
awaiting the task through the appropriate runtime-aware API. Update all spawn
callers and tests to await the new signature, and adjust shutdown to use the
matching lock access without holding its guard across an await.
| 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>, | ||
| } | ||
|
|
||
| impl RecursionGuard { | ||
| pub fn new() -> Self { | ||
| Self { | ||
| in_error_path: std::cell::Cell::new(false), | ||
| depth: Arc::new(AtomicU64::new(0)), | ||
| global_in_error: Arc::new(AtomicBool::new(false)), | ||
| } | ||
| } | ||
|
|
||
| /// Try to enter the error-record path. Returns `true` if entry is allowed, `false` if we are already inside an error record (recursion detected). When returning false, no logging should occur — this prevents self-logging loops. | ||
| pub fn try_enter_error_path(&self) -> bool { | ||
| // Fast-path thread-local check first. | ||
| if self.in_error_path.get() { | ||
| return false; | ||
| } | ||
|
|
||
| // Global atomic guard: prevent any concurrent entry across threads. | ||
| if !self | ||
| .global_in_error | ||
| .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed) | ||
| .is_ok() | ||
| { | ||
| return false; | ||
| } | ||
|
|
||
| self.in_error_path.set(true); | ||
| true | ||
| } | ||
|
|
||
| /// Exit the error-record path. Must be called after every successful `try_enter_error_path()`. | ||
| pub fn exit_error_path(&self) { | ||
| self.in_error_path.set(false); | ||
| self.global_in_error.store(false, Ordering::Release); | ||
| } | ||
|
|
||
| /// Check if currently inside an error path (for observability / tests). | ||
| #[allow(dead_code)] | ||
| pub fn is_in_error_path(&self) -> bool { | ||
| self.in_error_path.get() || self.global_in_error.load(Ordering::Acquire) | ||
| } | ||
|
|
||
| /// Clone the depth counter for external observation. | ||
| #[allow(dead_code)] | ||
| pub fn depth_clone(&self) -> Arc<AtomicU64> { | ||
| self.depth.clone() | ||
| } | ||
| } | ||
|
|
||
| // 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 {} |
There was a problem hiding this comment.
🩺 Stability & Availability | 🔴 Critical | ⚡ Quick win
unsafe impl Sync for RecursionGuard is unsound because the struct holds a Cell<bool>.
Cell<bool> is deliberately !Sync. The manual unsafe impl Sync at Line 70 removes that protection while in_error_path stays reachable from every thread that holds the shared guard:
FailOpenWriterstoresArc<RecursionGuard>(Line 89), andLoggingServicestoresArc<FailOpenWriter>and is shared across Tokio tasks.try_enter_error_pathcallsself.in_error_path.get()at Line 32 andset()at Line 45 on whichever thread invokes it.exit_error_pathcallsset()at Line 51.
Two threads that call these methods concurrently perform unsynchronized reads and writes on the same Cell. That is a data race and undefined behavior, not merely a stale value. The safety comment at Line 68 states the cell is only accessed from its own thread, but the code does not enforce that. The test concurrent_recursion_guard_does_not_allow_cross_thread_duplication at Lines 268-292 shares one guard across two threads and exercises exactly this path.
The global_in_error AtomicBool already provides the mutual exclusion the guard needs. Remove the Cell and the two unsafe impl blocks so the compiler derives Send/Sync on its own. If a per-thread fast path is required, use thread_local! instead of a shared Cell.
Note also that depth is never incremented, so the counter exposed by depth_clone and printed by Debug always reports zero.
🔒 Proposed fix: rely on the atomic guard only
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>,
}
impl RecursionGuard {
pub fn new() -> Self {
Self {
- in_error_path: std::cell::Cell::new(false),
- depth: Arc::new(AtomicU64::new(0)),
global_in_error: Arc::new(AtomicBool::new(false)),
}
}
pub fn try_enter_error_path(&self) -> bool {
- // Fast-path thread-local check first.
- if self.in_error_path.get() {
- return false;
- }
-
- // Global atomic guard: prevent any concurrent entry across threads.
- if !self
- .global_in_error
- .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
- .is_ok()
- {
- return false;
- }
-
- self.in_error_path.set(true);
- true
+ // Global atomic guard: prevent recursive and concurrent entry.
+ self.global_in_error
+ .compare_exchange(false, true, Ordering::Acquire, Ordering::Relaxed)
+ .is_ok()
}
pub fn exit_error_path(&self) {
- self.in_error_path.set(false);
self.global_in_error.store(false, Ordering::Release);
}
#[allow(dead_code)]
pub fn is_in_error_path(&self) -> bool {
- self.in_error_path.get() || self.global_in_error.load(Ordering::Acquire)
+ self.global_in_error.load(Ordering::Acquire)
}
}
-// 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 {}The rewritten try_enter_error_path also removes the !...is_ok() form, which Clippy reports under nonminimal_bool. Update Debug and depth_clone after the depth field is removed.
🤖 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 - 70,
Update RecursionGuard to remove the shared Cell<bool> field, both unsafe
Send/Sync implementations, and the unused depth field with its depth_clone
method; rely solely on global_in_error for entry and exit synchronization, and
simplify try_enter_error_path’s atomic comparison accordingly. Update Debug and
any related references to stop reporting the removed depth counter.
| ApplyResult::AppliedWithRestartRequired { | ||
| revision, | ||
| hash, | ||
| diagnostics, | ||
| } => { | ||
| let _ = self.config_revision_tx.send(revision); | ||
| owner_control_response::apply_response_envelope( | ||
| request_id, | ||
| crate::proto::node::OwnerControlApplyConfigResponse { | ||
| success: true, | ||
| current_revision: revision, | ||
| config_hash: hash.to_vec(), | ||
| error: None, | ||
| apply_mode: ConfigApplyMode::Staged as i32, | ||
| diagnostics: owner_control_response::config_diagnostics_to_proto( | ||
| &diagnostics, | ||
| ), | ||
| }, | ||
| ) | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | ⚡ Quick win
🧩 Analysis chain
🏁 Script executed:
#!/bin/bash
# Description: Compare proto and local ConfigApplyMode discriminants and confirm the helper mapping.
rg -n -A6 'enum ConfigApplyMode' --glob '*.proto'
rg -n -A12 'fn proto_apply_mode' crates/mesh-llm-host-runtime/srcRepository: Mesh-LLM/mesh-llm
Length of output: 155
🏁 Script executed:
#!/bin/bash
set -u
echo "== files =="
git ls-files | rg '(^|/)(owner_control|config_state|owner_control_response|.*proto.*ConfigApplyMode|.*config.*apply.*mode).*' || true
echo "== locate mod.rs snippets =="
for f in $(git ls-files | rg 'crates/mesh-llm-host-runtime/src/.*/owner_control/mod\.rs$|config_state\.rs$|owner_control_response\.rs$|.*\.proto$'); do
echo "-- $f --"
wc -l "$f"
done
echo "== owner_control imports and lines =="
f=$(git ls-files | rg 'crates/mesh-llm-host-runtime/src/mesh/owner_control/mod\.rs$' | head -n1)
if [ -n "$f" ]; then
sed -n '700,875p' "$f" | cat -n -v
fi
echo "== config_state nearby =="
f=$(git ls-files | rg 'config_state\.rs$' | head -n1)
if [ -n "$f" ]; then
rg -n -B5 -A25 'enum ApplyResult|enum ConfigApplyMode|apply_mode|ConfigApplyMode|LoggingChangesRequireRestart|restart' "$f" || true
fi
echo "== proto ApplyMode references =="
rg -n -i -C3 'apply.mode|ApplyMode|config_apply_mode|ConfigApplyMode' --glob '*.rs' --glob '*.proto' crates || trueRepository: Mesh-LLM/mesh-llm
Length of output: 50374
🏁 Script executed:
#!/bin/bash
set -u
echo "== response helper =="
cat -n crates/mesh-llm-host-runtime/src/mesh/owner_control_response.rs | sed -n '1,80p'
echo "== node proto enum =="
cat -n crates/mesh-llm-protocol/proto/node.proto | sed -n '650,710p'
echo "== generated references in repository (if any) =="
rg -n -C2 'enum ConfigApplyMode|ConfigApplyMode::(Unspecified|Staged|Noop)|proto_apply_mode' crates --glob '*.rs' || true
echo "== source proto enum around node config apply response =="
rg -n -C4 -i 'enum.*config.*apply.*mode|ConfigApplyMode|owner_control_apply_config_response|restart_required|restart_required' crates/mesh-llm-protocol/proto/node.proto crates/skippy-protocol/proto/* crates/mesh-llm-plugin/proto/* || true
echo "== rust generated enum discriminants from source proto definitions =="
python3 - <<'PY'
from pathlib import Path
import re
for path, prefix, enum in [
("crates/mesh-llm-protocol/proto/node.proto", "CONFIG_APPLY_MODE_", "ConfigApplyMode"),
("crates/mesh-llm-plugin/proto/plugin.proto", "PLUGIN_CONFIG_APPLY_MODE_", "PluginConfigApplyMode"),
]:
p = Path(path)
text = p.read_text()
m = re.search(rf'(?ms)enum\s+{enum}\s*\{{\s*([^}}]+?)\s*\}}', text)
if m:
print(f"-- {path} {enum} --")
for line in m.group(1).splitlines():
s = line.strip()
if not s or s.startswith("//"): continue
field, num = s.split("=")
print(f"{field.strip()} => {num.strip()}")
else:
print(f"NOT FOUND {path} {enum}")
PY
echo "== local mode discriminants =="
python3 - <<'PY'
text = Path("crates/mesh-llm-host-runtime/src/runtime/config_state.rs").read_text()
m = re.search(r'(?ms)#\[derive\([^)]*Copy[^)]*\)\]\s*pub\(crate\) enum ConfigApplyMode \{\s*([^\}}]+)\s*\}', text)
if m:
variants = []
v = None
for line in m.group(1).splitlines():
s = line.strip()
if not s or s.startswith("//"): continue
if "=" not in s:
v = s.rstrip(",")
variants.append(v)
else:
variants.append(s.split("=",1)[0].strip())
print("-- ConfigApplyMode variants --")
for i, v in enumerate(variants):
print(f"{v} => {i}")
else:
print("NOT FOUND ConfigApplyMode")
PYRepository: Mesh-LLM/mesh-llm
Length of output: 25530
Use the proto apply mode for restart-required responses.
ConfigApplyMode::Staged is the local enum variant 0, but OwnerControlApplyConfigResponse.apply_mode uses the proto enum where Staged is 1. This writes the wrong wire value for successful restart-required applies; use owner_control_response::proto_apply_mode or the proto enum variant instead.
Also, AppliedWithRestartRequired does not expose a restart-required signal in the response, so caller behavior cannot reflect this classification. Add an additive response field for restart-required responses if this is meant to change client 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-host-runtime/src/mesh/owner_control/mod.rs` around lines 813
- 832, The AppliedWithRestartRequired branch in the apply response handler uses
the local ConfigApplyMode value instead of the proto representation; set
apply_mode through owner_control_response::proto_apply_mode or the corresponding
proto enum variant so Staged serializes correctly. Also extend the response
contract with an additive restart-required indicator and populate it only for
AppliedWithRestartRequired responses, preserving existing behavior for other
apply outcomes.
| 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 can reduce non-UTF-8 content to zero bytes.
The loop removes bytes from the end until std::str::from_utf8 succeeds. That strategy only works when the invalid sequence sits at the end of the slice. For binary content, an invalid byte usually appears near the start, so the loop trims the whole slice and returns an empty Vec. The write then stores an empty file with a checksum of empty input and reports truncated: true.
The current tests do not cover this path. Test 4 uses text, and test 5 uses vec![0u8; 1024], and 0x00 is valid UTF-8.
Truncate at a character boundary only when the content is text, and cut at the byte limit otherwise.
🐛 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];
+ // Text content: cut back to the last complete character.
+ // Binary content: keep the full byte window.
+ let end = match std::str::from_utf8(head) {
+ Ok(_) => byte_limit,
+ // `error_len() == None` means the last character is incomplete.
+ Err(e) if e.error_len().is_none() => e.valid_up_to(),
+ // Any other error means the content is not text; do not trim further.
+ Err(_) => byte_limit,
+ };
+
+ (head[..end].to_vec(), true)
}Add a test with binary content that has an invalid UTF-8 byte in the first few bytes.
📝 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]; | |
| // Text content: cut back to the last complete character. | |
| // Binary content: keep the full byte window. | |
| let end = match std::str::from_utf8(head) { | |
| Ok(_) => byte_limit, | |
| // `error_len() == None` means the last character is incomplete. | |
| Err(e) if e.error_len().is_none() => e.valid_up_to(), | |
| // Any other error means the content is not text; do not trim further. | |
| Err(_) => 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 distinguish valid UTF-8 text from binary content: apply
character-boundary truncation only when the input is UTF-8, and otherwise return
the first byte_limit bytes unchanged with truncated set to true. Add a test
using binary data containing an invalid UTF-8 byte near the beginning to verify
it is not reduced to an empty vector.
| // Check for existing pointer before any disk work. | ||
| let exists: bool = self | ||
| .store | ||
| .conn() | ||
| .query_row( | ||
| "SELECT EXISTS(SELECT 1 FROM artifact_pointers WHERE artifact_id = ?)", | ||
| rusqlite::params![artifact_id], | ||
| |r| r.get::<_, i32>(0), | ||
| ) | ||
| .map(|v| v != 0) | ||
| .map_err(LogStoreError::Sqlite)?; | ||
|
|
||
| if exists { | ||
| return Err(LogStoreError::AlreadyExists { | ||
| entity: format!("artifact_pointer {}", artifact_id), | ||
| }); | ||
| } |
There was a problem hiding this comment.
🗄️ Data Integrity & Integration | 🔴 Critical | 🏗️ Heavy lift
Concurrent writes of the same artifact_id can delete the stored file.
The existence check at lines 229-238 releases the connection lock before the write starts. Two concurrent write_artifact calls with the same artifact_id therefore both pass the check. The sequence that follows destroys committed data:
- Both callers build the same
tmp/<artifact_id>.partpath (line 311). Theremove_filepluscreate_new(true)pair at lines 318-324 is itself a race, so one caller can fail withAlreadyExistsfrom the filesystem. - Both callers rename onto the same
final_path(line 344). - The first caller commits the pointer row. The second caller's
INSERThits the primary key and returns through theErrarm at line 387. - The
Errarm runsfs::remove_file(&final_path)at line 389. That path now holds the committed artifact of the first caller. The row survives, and the file disappears.
The result is a pointer row with no file, which read_artifact reports as ArtifactMissing.
Two changes fix this:
- Make the temporary file name unique, for example with a
uuid::Uuid::new_v4()suffix.uuidis already a dependency incrates/mesh-llm-log-store/Cargo.tomlline 21. - Insert the pointer row before the rename, or only remove the file when this call created the row. The insert also needs
is_unique_constraint_errorhandling so a duplicate returnsAlreadyExistsinstead of the currentLogStoreError::Sqlite.
🛡️ Minimal change for the temporary path
- let tmp_path = tmp_dir.join(format!("{}.part", artifact_id));
+ // Unique per call so concurrent writers cannot share a temporary file.
+ let tmp_path = tmp_dir.join(format!("{}.{}.part", artifact_id, uuid::Uuid::new_v4()));Also applies to: 303-345, 358-392
🤖 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 228 - 244, Update
write_artifact to prevent concurrent writes for the same artifact_id from
deleting committed data. Make the temporary artifact path unique using a UUID
suffix, then arrange pointer insertion and file cleanup so a losing concurrent
write cannot remove the winner’s final_path. Handle duplicate pointer insertion
with is_unique_constraint_error and return AlreadyExists rather than
LogStoreError::Sqlite.
|
Superseded by #1136. The replacement reconstructs the foundation layer with deterministic filesystem tests, shipped-binary configuration coverage, and corrected CI crate ownership. The original branch remains preserved as a recovery reference. |
1 participant
Users gain local, privacy-safe logging foundations with bounded configuration, durable storage, secure artifact handling, lifecycle tracking, and runtime initialization.\n\nStack order: this is PR1 of 3 and is the base for the API implementation PR.\n\nValidation: cargo fmt --all --check; mesh-llm-events, mesh-llm-log-store, mesh-llm-config, and host logging tests; package checks and warning-denying Clippy completed.\n\nNo API, WebSocket, frontend, TUI, or documentation surfaces are included.
Summary by CodeRabbit