fix: clippy fixes for arithmetic safety and redundant wrapping

Use saturating_sub, checked_sub_signed, signed_duration_since instead of
deprecated Sub ops in session_service. Flatten nested if-let in observation
tombstone logic. Remove redundant Ok() wrapping in FTS and semantic search.

Ultraworked with [Sisyphus](https://github.com/code-yeongyu/oh-my-opencode)

Co-authored-by: Sisyphus <clio-agent@sisyphuslabs.ai>

Author: Stranmor

crates/service/src/observation_service/mod.rs

@@ -118,30 +118,30 @@ impl ObservationService {
         let infinite_fut = self.store_infinite_memory(&tool_call, None);
         let compress_fut = async {
             let result = self.compress_and_save(id, &tool_call).await?;
-            if let Some((ref obs, false)) = result {
-                if obs.id.0 != id {
-                    let tombstone = Observation::builder(
-                        id.to_owned(),
-                        obs.session_id.clone(),
-                        obs.observation_type,
-                        format!("[merged into {}]", obs.id),
-                    )
-                    .noise_level(opencode_mem_core::NoiseLevel::Negligible)
-                    .noise_reason(format!("Tombstone: merged into {}", obs.id))
-                    .build();
-
-                    let result = self
-                        .storage
-                        .guarded(|| self.storage.save_observation(&tombstone))
-                        .await;
-                    if let Err(e) = self.with_cb(result) {
-                        tracing::warn!(
-                            tombstone_id = %id,
-                            merged_into = %obs.id,
-                            error = %e,
-                            "Failed to save tombstone observation — stale original may remain visible"
-                        );
-                    }
+            if let Some((ref obs, false)) = result
+                && obs.id.0 != id
+            {
+                let tombstone = Observation::builder(
+                    id.to_owned(),
+                    obs.session_id.clone(),
+                    obs.observation_type,
+                    format!("[merged into {}]", obs.id),
+                )
+                .noise_level(opencode_mem_core::NoiseLevel::Negligible)
+                .noise_reason(format!("Tombstone: merged into {}", obs.id))
+                .build();
+
+                let result = self
+                    .storage
+                    .guarded(|| self.storage.save_observation(&tombstone))
+                    .await;
+                if let Err(e) = self.with_cb(result) {
+                    tracing::warn!(
+                        tombstone_id = %id,
+                        merged_into = %obs.id,
+                        error = %e,
+                        "Failed to save tombstone observation — stale original may remain visible"
+                    );
                 }
             }
             Ok::<_, crate::ServiceError>(result)

crates/service/src/session_service.rs

@@ -21,7 +21,10 @@ const STALE_PLACEHOLDER_MINUTES: i64 = 10;
 /// Returns a slice of the most recent observations.
 fn truncate_observations_for_summary(observations: &[Observation]) -> &[Observation] {
     if observations.len() > MAX_OBSERVATIONS_FOR_SUMMARY {
-        &observations[observations.len() - MAX_OBSERVATIONS_FOR_SUMMARY..]
+        let start = observations
+            .len()
+            .saturating_sub(MAX_OBSERVATIONS_FOR_SUMMARY);
+        observations.get(start..).unwrap_or(observations)
     } else {
         observations
     }
@@ -257,12 +260,13 @@ impl SessionService {
                         .learned
                         .as_deref()
                         .is_some_and(|l| l == "processing");
-                    let stale_threshold =
-                        Utc::now() - TimeDelta::minutes(STALE_PLACEHOLDER_MINUTES);
+                    let stale_threshold = Utc::now()
+                        .checked_sub_signed(TimeDelta::minutes(STALE_PLACEHOLDER_MINUTES))
+                        .unwrap_or_else(Utc::now);
                     if is_processing && existing.created_at < stale_threshold {
                         tracing::warn!(
                             session_id = %session.session_id,
-                            age_minutes = (Utc::now() - existing.created_at).num_minutes(),
+                            age_minutes = Utc::now().signed_duration_since(existing.created_at).num_minutes(),
                             "Deleting stale processing placeholder"
                         );
                         let _ = self

crates/storage/src/pg_storage/search/fts.rs

@@ -24,9 +24,7 @@ pub(crate) async fn search(
     .bind(usize_to_i64(limit))
     .fetch_all(&storage.pool)
     .await?;
-    Ok(collect_skipping_corrupt(
-        rows.iter().map(row_to_search_result),
-    )?)
+    collect_skipping_corrupt(rows.iter().map(row_to_search_result))
 }
 
 pub(crate) async fn search_with_filters(
@@ -92,9 +90,7 @@ pub(crate) async fn search_with_filters(
         q = q.bind(&tsquery);
         q = q.bind(usize_to_i64(limit));
         let rows = q.fetch_all(&storage.pool).await?;
-        return Ok(collect_skipping_corrupt(
-            rows.iter().map(row_to_search_result),
-        )?);
+        return collect_skipping_corrupt(rows.iter().map(row_to_search_result));
     }
 
     let where_clause = if conditions.is_empty() {
@@ -115,7 +111,5 @@ pub(crate) async fn search_with_filters(
     }
     q = q.bind(usize_to_i64(limit));
     let rows = q.fetch_all(&storage.pool).await?;
-    Ok(collect_skipping_corrupt(
-        rows.iter().map(row_to_search_result),
-    )?)
+    collect_skipping_corrupt(rows.iter().map(row_to_search_result))
 }

crates/storage/src/pg_storage/search/semantic.rs

@@ -25,7 +25,5 @@ pub(crate) async fn semantic_search(
     .bind(usize_to_i64(limit))
     .fetch_all(&storage.pool)
     .await?;
-    Ok(collect_skipping_corrupt(
-        rows.iter().map(row_to_search_result),
-    )?)
+    collect_skipping_corrupt(rows.iter().map(row_to_search_result))
 }