desktop/src-tauri/src/lib.rs

@@ -45,6 +45,7 @@ pub fn run() {
             log::info!("opening local store: {}", db_path.display());
             let db = local::open(&db_path)?;
             log::info!("local store ready — {}", local::summary(&db));
+            sweep_local_trash(&db);
             app.manage(db);
             // Attachment bytes live beside the database, filed by content hash, so a
             // synced image is readable with no network (M10.7d).
@@ -105,6 +106,26 @@ pub fn run() {
         .expect("error while running the ThoughtSync desktop app");
 }
 
+/// Expire old trash at startup, on an unlinked device only (see `local::retention`).
+///
+/// At startup rather than on a timer: a desktop app isn't a server, and a sweep the
+/// user is present for is one they can see the result of. A failure here is logged and
+/// stepped over — housekeeping must never be the reason the app won't open.
+fn sweep_local_trash(db: &local::Db) {
+    let conn = match db.0.lock() {
+        Ok(conn) => conn,
+        Err(_) => {
+            log::warn!("skipping the trash sweep: store lock poisoned");
+            return;
+        }
+    };
+    match local::retention::sweep_if_unlinked(&conn) {
+        Ok(Some(0)) | Ok(None) => {}
+        Ok(Some(n)) => log::info!("trash retention: purged {n} expired note(s)"),
+        Err(e) => log::warn!("trash sweep failed: {e}"),
+    }
+}
+
 /// Frontend logging bridge: routes boot milestones and errors from the webview into
 /// the same stdout + file log as the Rust side (see frontend/src/desktop/bridge.ts).
 #[tauri::command]

desktop/src-tauri/src/local/commands.rs

@@ -7,20 +7,34 @@ use serde_json::Value;
 use tauri::State;
 
 use crate::local::models::*;
+use crate::local::retention;
 use crate::local::store;
 use crate::local::Db;
+use crate::sync::state;
 
 // A macro would hide the (very regular) locking; kept explicit so each command reads
 // as an obvious lock -> delegate -> stringify.
 
 #[tauri::command]
-pub fn config_get() -> PublicConfig {
+pub fn config_get(db: State<'_, Db>) -> PublicConfig {
+    // What the Trash view counts down against: the linked server's window if we know
+    // it, else this device's own. Reading it here rather than hard-coding the offline
+    // default is what keeps the deadline on screen equal to the one that will actually
+    // be enforced. A store error falls back to the default rather than failing the
+    // call — the app must still boot.
+    let retention_days = db
+        .0
+        .lock()
+        .ok()
+        .and_then(|conn| state::effective_retention_days(&conn, retention::LOCAL_RETENTION_DAYS).ok())
+        .unwrap_or(retention::LOCAL_RETENTION_DAYS);
     // Offline defaults: no signups, no server-side URL unfurling (needs network).
     PublicConfig {
         site_name: "ThoughtSync".to_string(),
         allow_registration: false,
         version: env!("CARGO_PKG_VERSION").to_string(),
         enable_url_unfurl: false,
+        trash_retention_days: retention_days.max(0) as u32,
     }
 }
 

desktop/src-tauri/src/local/mod.rs

@@ -5,6 +5,7 @@
 pub mod commands;
 pub mod derive;
 pub mod models;
+pub mod retention;
 pub mod schema;
 pub mod store;
 

desktop/src-tauri/src/local/models.rs

@@ -19,6 +19,9 @@ pub struct Note {
     pub pinned: bool,
     pub archived: bool,
     pub trashed: bool,
+    /// When it was trashed (null unless trashed). Named for the server's field so the
+    /// shared frontend counts down the retention window identically either way.
+    pub deleted_at: Option<String>,
     pub remind_at: Option<String>,
     pub recurrence: Option<String>,
     pub labels: Vec<NoteLabel>,
@@ -112,6 +115,7 @@ pub struct PublicConfig {
     pub allow_registration: bool,
     pub version: String,
     pub enable_url_unfurl: bool,
+    pub trash_retention_days: u32,
 }
 
 /// The synthetic single user the offline core reports, so the app's auth-gated

desktop/src-tauri/src/local/retention.rs

@@ -0,0 +1,205 @@
+//! Trash retention for a device with no server (M11.3).
+//!
+//! The server owns this policy whenever there IS one: a linked client learns about
+//! every permanent deletion from the delta feed, as a tombstone, and does exactly
+//! what it's told. This module exists for the case the server can't cover — an
+//! offline-only install, where trash would otherwise sit forever and the attachment
+//! bytes with it.
+//!
+//! Which is why the sweep refuses to run while linked. If it didn't, a device could
+//! decide on its own that a note had expired, destroy it, and then push that delete
+//! upstream — overruling a server that was deliberately keeping it (retention off, or
+//! a longer window than this constant). A client's local policy must never outrank
+//! the server's.
+
+use chrono::{DateTime, Duration, Utc};
+use rusqlite::Connection;
+
+use super::store;
+use crate::sync::state;
+
+/// The window an unlinked device uses. Matches the server's default so a device that
+/// later links doesn't see its trash behave differently from one that always was.
+pub const LOCAL_RETENTION_DAYS: i64 = 30;
+
+/// Purge trash older than `retention_days`. Returns how many notes went.
+///
+/// `now` is a parameter so the window arithmetic is testable without waiting a month.
+pub fn sweep_expired_trash(
+    conn: &Connection,
+    retention_days: i64,
+    now: DateTime<Utc>,
+) -> rusqlite::Result<usize> {
+    if retention_days <= 0 {
+        return Ok(0);
+    }
+    let cutoff = now - Duration::days(retention_days);
+    let expired: Vec<String> = {
+        let mut stmt =
+            conn.prepare("SELECT id, trashed_at FROM notes WHERE trashed = 1 AND trashed_at IS NOT NULL")?;
+        let rows = stmt.query_map([], |r| Ok((r.get::<_, String>(0)?, r.get::<_, String>(1)?)))?;
+        rows.filter_map(|row| {
+            let (id, stamped) = row.ok()?;
+            // PARSED, not string-compared. The server writes `+00:00` offsets and this
+            // client writes `Z`, so two timestamps for the same instant don't sort
+            // against each other as text — and the failure would be silent.
+            let trashed_at = DateTime::parse_from_rfc3339(&stamped).ok()?;
+            // An unparseable or missing timestamp means "age unknown", and the only
+            // safe reading of that is to keep the note. Deleting on a guess is the one
+            // outcome nobody can undo.
+            (trashed_at.with_timezone(&Utc) < cutoff).then_some(id)
+        })
+        .collect()
+    };
+    for id in &expired {
+        // Through delete_forever, so a `pending_deletes` tombstone is recorded. That's
+        // right even here: while unlinked this device holds the only copy, so if it
+        // links later the server should learn the note was deleted, not re-send it.
+        store::delete_forever(conn, id)?;
+    }
+    Ok(expired.len())
+}
+
+/// The startup sweep: runs only on an unlinked device (see the module note).
+/// Returns `None` when it didn't run because the device is linked.
+pub fn sweep_if_unlinked(conn: &Connection) -> rusqlite::Result<Option<usize>> {
+    if state::read(conn)?.server_url.is_some() {
+        return Ok(None);
+    }
+    sweep_expired_trash(conn, LOCAL_RETENTION_DAYS, Utc::now()).map(Some)
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::local::schema;
+
+    fn db() -> Connection {
+        let conn = Connection::open_in_memory().expect("in-memory db");
+        schema::migrate(&conn).expect("migrate");
+        conn
+    }
+
+    /// A trashed note stamped `trashed_at` days ago, in the format the CLIENT writes
+    /// (`...Z`, millisecond precision — see `store::now`).
+    fn trashed_note(conn: &Connection, id: &str, days_ago: i64) {
+        let stamped = (Utc::now() - Duration::days(days_ago))
+            .to_rfc3339_opts(chrono::SecondsFormat::Millis, true);
+        conn.execute(
+            "INSERT INTO notes (id, title, body, created_at, updated_at, trashed, trashed_at)
+             VALUES (?1, 'T', 'B', ?2, ?2, 1, ?2)",
+            rusqlite::params![id, stamped],
+        )
+        .expect("insert");
+    }
+
+    fn note_count(conn: &Connection) -> i64 {
+        conn.query_row("SELECT COUNT(*) FROM notes", [], |r| r.get(0))
+            .expect("count")
+    }
+
+    #[test]
+    fn purges_trash_past_the_window_and_keeps_the_rest() {
+        let conn = db();
+        trashed_note(&conn, "old", 40);
+        trashed_note(&conn, "fresh", 3);
+        let purged = sweep_expired_trash(&conn, 30, Utc::now()).expect("sweep");
+        assert_eq!(purged, 1);
+        assert_eq!(note_count(&conn), 1, "only the expired note should go");
+    }
+
+    #[test]
+    fn a_note_exactly_at_the_boundary_survives() {
+        // Strictly older than the cutoff, so the note trashed 30 days ago gets its
+        // full 30 days rather than being cut a moment short.
+        let conn = db();
+        trashed_note(&conn, "boundary", 30);
+        assert_eq!(sweep_expired_trash(&conn, 30, Utc::now()).expect("sweep"), 0);
+    }
+
+    #[test]
+    fn retention_off_purges_nothing() {
+        let conn = db();
+        trashed_note(&conn, "ancient", 4000);
+        assert_eq!(sweep_expired_trash(&conn, 0, Utc::now()).expect("sweep"), 0);
+        assert_eq!(sweep_expired_trash(&conn, -1, Utc::now()).expect("sweep"), 0);
+        assert_eq!(note_count(&conn), 1);
+    }
+
+    #[test]
+    fn an_untrashed_note_is_never_swept() {
+        let conn = db();
+        conn.execute(
+            "INSERT INTO notes (id, title, body, created_at, updated_at, trashed)
+             VALUES ('live', 'T', 'B', '2020-01-01T00:00:00.000Z', '2020-01-01T00:00:00.000Z', 0)",
+            [],
+        )
+        .expect("insert");
+        assert_eq!(sweep_expired_trash(&conn, 30, Utc::now()).expect("sweep"), 0);
+        assert_eq!(note_count(&conn), 1);
+    }
+
+    #[test]
+    fn an_unparseable_timestamp_keeps_the_note() {
+        // "Age unknown" must never resolve to "delete it".
+        let conn = db();
+        conn.execute(
+            "INSERT INTO notes (id, title, body, created_at, updated_at, trashed, trashed_at)
+             VALUES ('weird', 'T', 'B', '2020-01-01T00:00:00.000Z', '2020-01-01T00:00:00.000Z', 1, 'not a date')",
+            [],
+        )
+        .expect("insert");
+        assert_eq!(sweep_expired_trash(&conn, 30, Utc::now()).expect("sweep"), 0);
+        assert_eq!(note_count(&conn), 1);
+    }
+
+    #[test]
+    fn a_server_style_offset_timestamp_is_understood() {
+        // The server serializes with a `+00:00` offset, not `Z`. Comparing those as
+        // strings would quietly never match — this is the case that catches it.
+        let conn = db();
+        let stamped = (Utc::now() - Duration::days(40)).to_rfc3339();
+        conn.execute(
+            "INSERT INTO notes (id, title, body, created_at, updated_at, trashed, trashed_at)
+             VALUES ('server', 'T', 'B', ?1, ?1, 1, ?1)",
+            rusqlite::params![stamped],
+        )
+        .expect("insert");
+        assert_eq!(sweep_expired_trash(&conn, 30, Utc::now()).expect("sweep"), 1);
+    }
+
+    #[test]
+    fn a_purged_note_leaves_a_pending_delete_behind() {
+        // Without the tombstone, linking this device later would let the server
+        // re-send a note the user already destroyed here.
+        let conn = db();
+        trashed_note(&conn, "old", 40);
+        sweep_expired_trash(&conn, 30, Utc::now()).expect("sweep");
+        let pending: i64 = conn
+            .query_row(
+                "SELECT COUNT(*) FROM pending_deletes WHERE entity = 'note' AND id = 'old'",
+                [],
+                |r| r.get(0),
+            )
+            .expect("count");
+        assert_eq!(pending, 1);
+    }
+
+    #[test]
+    fn a_linked_device_does_not_sweep() {
+        // The whole safety rule: with a server present, purging is the server's call.
+        let conn = db();
+        trashed_note(&conn, "old", 400);
+        state::set_link(&conn, "https://notes.example", "token").expect("link");
+        assert_eq!(sweep_if_unlinked(&conn).expect("sweep"), None);
+        assert_eq!(note_count(&conn), 1, "the note must survive on a linked device");
+    }
+
+    #[test]
+    fn an_unlinked_device_sweeps() {
+        let conn = db();
+        trashed_note(&conn, "old", 400);
+        assert_eq!(sweep_if_unlinked(&conn).expect("sweep"), Some(1));
+        assert_eq!(note_count(&conn), 0);
+    }
+}

desktop/src-tauri/src/local/schema.rs

@@ -130,6 +130,23 @@ const SCHEMA_V3: &str = r#"
 ALTER TABLE sync_state ADD COLUMN last_sync_at TEXT;
 "#;
 
+// v4 (M11.3): WHEN a note was trashed.
+//
+// The table only ever recorded THAT a note was trashed, which is enough to draw a
+// Trash view and nothing else. Retention needs an age: without a timestamp there is
+// no way to tell a note trashed this morning from one trashed last spring, so an
+// offline device could never expire its own trash — and the UI couldn't warn anyone
+// before it did.
+// It also records the LINKED server's retention window, captured from /api/config.
+// Once linked, the server's policy is the one that actually applies, so showing this
+// device's offline default would put a countdown on screen that doesn't match what
+// happens — a wrong deadline is worse than none.
+const SCHEMA_V4: &str = r#"
+ALTER TABLE notes ADD COLUMN trashed_at TEXT;
+UPDATE notes SET trashed_at = updated_at WHERE trashed = 1;
+ALTER TABLE sync_state ADD COLUMN server_retention_days INTEGER;
+"#;
+
 /// Bring the database up to the latest schema. Idempotent.
 pub fn migrate(conn: &Connection) -> rusqlite::Result<()> {
     conn.execute_batch("PRAGMA foreign_keys = ON;")?;
@@ -146,5 +163,9 @@ pub fn migrate(conn: &Connection) -> rusqlite::Result<()> {
         conn.execute_batch(SCHEMA_V3)?;
         conn.execute_batch("PRAGMA user_version = 3;")?;
     }
+    if version < 4 {
+        conn.execute_batch(SCHEMA_V4)?;
+        conn.execute_batch("PRAGMA user_version = 4;")?;
+    }
     Ok(())
 }

desktop/src-tauri/src/local/store.rs

@@ -123,7 +123,7 @@ fn load_previews(conn: &Connection, note_id: &str) -> rusqlite::Result<Vec<LinkP
 
 fn load_note(conn: &Connection, id: &str) -> rusqlite::Result<Note> {
     let mut note = conn.query_row(
-        "SELECT id, title, body, color, kind, position, pinned, archived, trashed, remind_at, recurrence, created_at, updated_at
+        "SELECT id, title, body, color, kind, position, pinned, archived, trashed, remind_at, recurrence, created_at, updated_at, trashed_at
          FROM notes WHERE id = ?1",
         [id],
         |r| {
@@ -141,6 +141,7 @@ fn load_note(conn: &Connection, id: &str) -> rusqlite::Result<Note> {
                 pinned: r.get(6)?,
                 archived: r.get(7)?,
                 trashed: r.get(8)?,
+                deleted_at: r.get(13)?,
                 remind_at: r.get(9)?,
                 recurrence: r.get(10)?,
                 labels: Vec::new(),
@@ -621,13 +622,22 @@ pub fn reorder(conn: &Connection, ordered_ids: &[String]) -> rusqlite::Result<()
 }
 
 pub fn trash(conn: &Connection, id: &str) -> rusqlite::Result<Note> {
-    conn.execute("UPDATE notes SET trashed = 1 WHERE id = ?1", [id])?;
+    // COALESCE, so trashing an already-trashed note doesn't restart its retention
+    // clock. The server keeps its `deleted_at` the same way — a note shouldn't earn
+    // another 30 days because something touched it twice.
+    conn.execute(
+        "UPDATE notes SET trashed = 1, trashed_at = COALESCE(trashed_at, ?1) WHERE id = ?2",
+        params![now(), id],
+    )?;
     touch(conn, id)?;
     load_note(conn, id)
 }
 
 pub fn restore(conn: &Connection, id: &str) -> rusqlite::Result<Note> {
-    conn.execute("UPDATE notes SET trashed = 0 WHERE id = ?1", [id])?;
+    conn.execute(
+        "UPDATE notes SET trashed = 0, trashed_at = NULL WHERE id = ?1",
+        [id],
+    )?;
     touch(conn, id)?;
     load_note(conn, id)
 }

desktop/src-tauri/src/sync/commands.rs

@@ -95,6 +95,12 @@ pub async fn sync_link(input: LinkInput, db: State<'_, Db>) -> Result<LinkResult
     let status = {
         let conn = db.0.lock().map_err(|e| e.to_string())?;
         state::set_link(&conn, &base_url, &token).map_err(|e| e.to_string())?;
+        // Adopt the server's trash-retention window immediately, so the Trash view
+        // stops counting down against this device's offline default the moment it's
+        // no longer the policy in force.
+        if let Some(days) = probe.server.trash_retention_days {
+            state::set_server_retention(&conn, days as i64).map_err(|e| e.to_string())?;
+        }
         state::status(&conn).map_err(|e| e.to_string())?
     };
 

desktop/src-tauri/src/sync/compat.rs

@@ -52,6 +52,11 @@ pub struct ServerInfo {
     pub min_client_protocol_version: Option<u32>,
     #[serde(default)]
     pub sync_features: Vec<String>,
+    /// How long the SERVER keeps a trashed note before purging it (0 = forever).
+    /// Once linked this is the window that actually applies, so the desktop's Trash
+    /// countdown has to come from here rather than from its own offline default.
+    #[serde(default)]
+    pub trash_retention_days: Option<u32>,
 }
 
 impl ServerInfo {
@@ -218,6 +223,7 @@ mod tests {
                 .copied()
                 .map(String::from)
                 .collect(),
+            trash_retention_days: Some(30),
         }
     }
 

desktop/src-tauri/src/sync/engine.rs

@@ -51,8 +51,21 @@ pub async fn run_cycle(
         );
     }
 
+    // While we're already talking to this server, re-read what it says about itself.
+    // Today that's the trash-retention window the Trash view counts down against, and
+    // it can change under us whenever an admin edits the setting. Best-effort on
+    // purpose: a config blip must not fail a cycle whose actual work already
+    // succeeded, and the stored value simply stays as it was.
+    let retention = super::client::probe(base_url)
+        .await
+        .ok()
+        .and_then(|p| p.server.trash_retention_days);
+
     let status = {
         let conn = db.0.lock().map_err(|e| e.to_string())?;
+        if let Some(days) = retention {
+            state::set_server_retention(&conn, days as i64).map_err(|e| e.to_string())?;
+        }
         // Stamped only here, after BOTH halves succeeded. A timestamp written after a
         // partial cycle would tell the user they're up to date when they aren't.
         let now = Utc::now().to_rfc3339_opts(SecondsFormat::Millis, true);

desktop/src-tauri/src/sync/pull.rs

@@ -227,13 +227,23 @@ fn upsert_label(conn: &Connection, label: &wire::Label) -> rusqlite::Result<()>
 fn upsert_note(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
     let created = note.created_at.clone().unwrap_or_else(now);
     let updated = note.updated_at.clone().unwrap_or_else(|| created.clone());
+    // The server's `deleted_at` is the authority on trash AGE. Taking it from the feed
+    // rather than stamping "now" locally is what keeps a note trashed three weeks ago
+    // from looking brand-new to a device that only just heard about it — otherwise
+    // every fresh install would silently reset the whole retention clock. Falls back
+    // to the note's updated_at only if an older server omits the field.
+    let trashed_at = if note.trashed {
+        note.deleted_at.clone().or_else(|| Some(updated.clone()))
+    } else {
+        None
+    };
     // `created_at` is deliberately absent from the UPDATE clause: a note's birth time
     // never changes, and the server's copy is the same value anyway.
     conn.execute(
         "INSERT INTO notes (id, title, body, color, kind, position, pinned, archived,
                             trashed, remind_at, recurrence, created_at, updated_at,
-                            sync_revision, dirty)
-              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, 0)
+                            sync_revision, trashed_at, dirty)
+              VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7, ?8, ?9, ?10, ?11, ?12, ?13, ?14, ?15, 0)
          ON CONFLICT(id) DO UPDATE SET
               title         = excluded.title,
               body          = excluded.body,
@@ -247,6 +257,7 @@ fn upsert_note(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
               recurrence    = excluded.recurrence,
               updated_at    = excluded.updated_at,
               sync_revision = excluded.sync_revision,
+              trashed_at    = excluded.trashed_at,
               dirty         = 0",
         params![
             note.id,
@@ -263,6 +274,7 @@ fn upsert_note(conn: &Connection, note: &wire::Note) -> rusqlite::Result<()> {
             created,
             updated,
             note.sync_revision,
+            trashed_at,
         ],
     )?;
 
@@ -494,6 +506,7 @@ mod tests {
             pinned: false,
             archived: false,
             trashed: false,
+            deleted_at: None,
             remind_at: None,
             recurrence: None,
             created_at: Some("2026-07-26T00:00:00.000Z".into()),
@@ -559,6 +572,57 @@ mod tests {
         assert_eq!(count(&conn, "SELECT trashed FROM notes WHERE id = 'n1'"), 1);
     }
 
+    #[test]
+    fn trash_age_comes_from_the_server_not_from_now() {
+        // The retention countdown runs off this timestamp. Stamping it locally would
+        // hand every note a fresh 30 days on any device that syncs it for the first
+        // time — a note trashed last month would never expire anywhere.
+        let conn = db();
+        let mut trashed = note("n1", 1);
+        trashed.trashed = true;
+        trashed.deleted_at = Some("2026-06-01T09:30:00+00:00".into());
+        apply_page(&conn, &page(vec![trashed], vec![], 1)).expect("apply");
+        let stamped: String = conn
+            .query_row("SELECT trashed_at FROM notes WHERE id = 'n1'", [], |r| {
+                r.get(0)
+            })
+            .expect("trashed_at");
+        assert_eq!(stamped, "2026-06-01T09:30:00+00:00");
+    }
+
+    #[test]
+    fn restoring_a_note_server_side_clears_its_trash_stamp() {
+        let conn = db();
+        let mut trashed = note("n1", 1);
+        trashed.trashed = true;
+        trashed.deleted_at = Some("2026-06-01T09:30:00+00:00".into());
+        apply_page(&conn, &page(vec![trashed], vec![], 1)).expect("apply");
+        apply_page(&conn, &page(vec![note("n1", 2)], vec![], 2)).expect("apply");
+        let stamped: Option<String> = conn
+            .query_row("SELECT trashed_at FROM notes WHERE id = 'n1'", [], |r| {
+                r.get(0)
+            })
+            .expect("trashed_at");
+        assert_eq!(stamped, None, "an untrashed note must carry no trash stamp");
+    }
+
+    #[test]
+    fn an_older_server_without_deleted_at_still_ages_the_trash() {
+        // Falls back to updated_at rather than leaving the stamp null, which would
+        // make the note un-expirable and its countdown blank.
+        let conn = db();
+        let mut trashed = note("n1", 1);
+        trashed.trashed = true;
+        trashed.deleted_at = None;
+        apply_page(&conn, &page(vec![trashed], vec![], 1)).expect("apply");
+        let stamped: Option<String> = conn
+            .query_row("SELECT trashed_at FROM notes WHERE id = 'n1'", [], |r| {
+                r.get(0)
+            })
+            .expect("trashed_at");
+        assert_eq!(stamped.as_deref(), Some("2026-07-26T00:00:00.000Z"));
+    }
+
     #[test]
     fn children_are_replaced_not_merged() {
         let conn = db();

desktop/src-tauri/src/sync/state.rs

@@ -19,6 +19,9 @@ pub struct SyncState {
     pub device_token: Option<String>,
     pub last_cursor: i64,
     pub last_sync_at: Option<String>,
+    /// The linked server's trash-retention window, as it last advertised it. `None`
+    /// until a probe or sync has learned it.
+    pub server_retention_days: Option<i64>,
 }
 
 impl SyncState {
@@ -59,12 +62,14 @@ fn present(value: Option<String>) -> Option<String> {
 
 pub fn read(conn: &Connection) -> rusqlite::Result<SyncState> {
     conn.query_row(
-        "SELECT server_url, device_token, last_cursor, last_sync_at FROM sync_state WHERE id = 1",
+        "SELECT server_url, device_token, last_cursor, last_sync_at, server_retention_days
+           FROM sync_state WHERE id = 1",
         [],
         |row| {
             let cursor: Option<String> = row.get(2)?;
             Ok(SyncState {
                 last_sync_at: present(row.get(3)?),
+                server_retention_days: row.get(4)?,
                 server_url: present(row.get(0)?),
                 device_token: present(row.get(1)?),
                 // Stored TEXT (schema) but used as an integer watermark. Absent or
@@ -106,13 +111,39 @@ pub fn clear_link(conn: &Connection) -> rusqlite::Result<()> {
     conn.execute(
         "UPDATE sync_state
             SET server_url = NULL, device_token = NULL, last_cursor = NULL,
-                last_sync_at = NULL
+                last_sync_at = NULL, server_retention_days = NULL
           WHERE id = 1",
         [],
     )?;
     Ok(())
 }
 
+/// Remember the linked server's trash-retention window (0 = it never purges).
+///
+/// Refreshed on every sync rather than only at link time, so changing the setting on
+/// the server reaches the desktop's Trash countdown on the next cycle instead of
+/// waiting for someone to re-link.
+pub fn set_server_retention(conn: &Connection, days: i64) -> rusqlite::Result<()> {
+    conn.execute(
+        "UPDATE sync_state SET server_retention_days = ?1 WHERE id = 1",
+        params![days],
+    )?;
+    Ok(())
+}
+
+/// The retention window in force on THIS device: the linked server's if we know it,
+/// otherwise the caller's offline default. A linked device must never enforce or
+/// advertise its own window over the server's.
+pub fn effective_retention_days(conn: &Connection, offline_default: i64) -> rusqlite::Result<i64> {
+    let state = read(conn)?;
+    if !state.is_linked() {
+        return Ok(offline_default);
+    }
+    // Linked but the server hasn't told us yet (linked by an older build, or no sync
+    // has completed). Fall back to the default rather than claiming "kept forever".
+    Ok(state.server_retention_days.unwrap_or(offline_default))
+}
+
 /// Stamp a completed sync. The cursor can't stand in for this: it's a revision
 /// watermark, and it doesn't move at all when a sync correctly finds nothing new —
 /// so "synced a moment ago, no changes" would be indistinguishable from "never
@@ -172,6 +203,43 @@ mod tests {
         assert_eq!(state.device_token.as_deref(), Some("tok-1"));
     }
 
+    #[test]
+    fn an_unlinked_device_uses_its_own_retention_window() {
+        let conn = db();
+        assert_eq!(effective_retention_days(&conn, 30).expect("read"), 30);
+    }
+
+    #[test]
+    fn a_linked_device_adopts_the_servers_window() {
+        // Including 0 — a server that keeps trash forever must not have this device
+        // showing a 30-day countdown that will never fire.
+        let conn = db();
+        set_link(&conn, "https://notes.example.com", "tok-1").expect("link");
+        set_server_retention(&conn, 0).expect("retention");
+        assert_eq!(effective_retention_days(&conn, 30).expect("read"), 0);
+        set_server_retention(&conn, 90).expect("retention");
+        assert_eq!(effective_retention_days(&conn, 30).expect("read"), 90);
+    }
+
+    #[test]
+    fn a_linked_device_that_hasnt_heard_yet_falls_back() {
+        // Linked by an older build, or no cycle has completed. The default is a
+        // safer guess than "forever", which would promise a note is being kept.
+        let conn = db();
+        set_link(&conn, "https://notes.example.com", "tok-1").expect("link");
+        assert_eq!(effective_retention_days(&conn, 30).expect("read"), 30);
+    }
+
+    #[test]
+    fn unlinking_forgets_the_servers_window() {
+        let conn = db();
+        set_link(&conn, "https://notes.example.com", "tok-1").expect("link");
+        set_server_retention(&conn, 90).expect("retention");
+        clear_link(&conn).expect("unlink");
+        assert_eq!(read(&conn).expect("read").server_retention_days, None);
+        assert_eq!(effective_retention_days(&conn, 30).expect("read"), 30);
+    }
+
     #[test]
     fn relinking_the_same_server_keeps_the_cursor() {
         let conn = db();

desktop/src-tauri/src/sync/wire.rs

@@ -40,6 +40,11 @@ pub struct Note {
     /// The server derives this from `deleted_at` — trash, NOT a tombstone.
     #[serde(default)]
     pub trashed: bool,
+    /// WHEN it was trashed. The trash-retention clock runs from here, so it has to be
+    /// the server's timestamp rather than anything this device invents. Absent from an
+    /// older server, which is why it's optional rather than required.
+    #[serde(default)]
+    pub deleted_at: Option<String>,
     #[serde(default)]
     pub remind_at: Option<String>,
     #[serde(default)]

docs/sync.md

@@ -33,7 +33,8 @@ token, or even has an account:
 { "site_name": "...", "version": "0.1.0",
   "sync_protocol_version": 1,
   "min_client_protocol_version": 1,
-  "sync_features": ["notes", "labels", "attachments", "tombstones", "revisions"] }
+  "sync_features": ["notes", "labels", "attachments", "tombstones", "revisions"],
+  "trash_retention_days": 30 }
 ```
 
 The client identifies itself on every request with
@@ -121,13 +122,31 @@ as `?since=`. `since=0` (or absent) is a **full initial sync**.
 
 Two levels, both propagate:
 
-- **Trash** — `deleted_at` is a normal field. A trashed note still syncs with its
-  content; the client shows it in its Trash. Restoring clears `deleted_at`.
+- **Trash** — `deleted_at` is a normal field, and it's **on the wire**: a trashed
+  note still syncs with its content, the client shows it in its Trash, and the
+  timestamp is what the client counts the retention window against. Restoring
+  clears it.
 - **Purge (permanent delete)** — becomes a **content-less tombstone**: `purged_at`
-  is set, title/body/items/labels/attachments are cleared/removed, and the row is
-  kept. A client seeing `purged_at != null` deletes the row from its local store.
-  Tombstones are retained indefinitely (cheap for a personal store); revisit if
-  they ever grow large.
+  is set, title/body/items/labels/attachments/previews/revisions are cleared or
+  removed, and the row is kept. A client seeing `purged_at != null` deletes the row
+  from its local store. `deleted_at` deliberately SURVIVES a purge, so ordinary
+  server-side queries (`deleted_at IS NULL`) never see a tombstone as a live note.
+  Tombstones themselves are retained indefinitely (cheap for a personal store);
+  revisit if they ever grow large.
+
+### Retention — trash expires
+
+A trashed note is purged automatically once it is older than the server's
+`trash_retention_days` setting (default **30**, `0` = keep forever), advertised on
+`/api/config` so a client can show the countdown. A background sweep on the server
+does the work; clients learn about it as ordinary tombstones and need no special
+handling.
+
+**A linked client must not run its own expiry.** The server owns the policy — one
+clock, one window. A client that purged on its own schedule could destroy a note
+the server was deliberately keeping and then push that delete upstream. An
+*unlinked* client (offline-only, no server to defer to) expires its own trash on
+its own default, which is the only case where nothing else can.
 
 ## Pull — `GET /api/sync/changes`
 
@@ -137,7 +156,8 @@ Response:
 
 ```json
 {
-  "notes":  [ { "...full note...", "sync_revision": 42, "purged_at": null } ],
+  "notes":  [ { "...full note...", "trashed": false, "deleted_at": null,
+               "sync_revision": 42, "purged_at": null } ],
   "labels": [ { "id": "...", "name": "...", "color": "...",
                "sync_revision": 43, "purged_at": null, "created_at": "..." } ],
   "cursor": 43,

frontend/src/components/NoteCard.vue

@@ -14,7 +14,8 @@ import Icon from "./Icon.vue";
 import LinkPreview from "./LinkPreview.vue";
 import MarkdownText from "./MarkdownText.vue";
 import NoteChecklist from "./NoteChecklist.vue";
-import { formatReminder, isOverdue } from "../notes/datetime";
+import { formatReminder, formatTrashCountdown, isOverdue, trashDaysLeft } from "../notes/datetime";
+import { useConfigStore } from "../stores/config";
 
 const props = defineProps<{ note: Note; reorderable?: boolean; active?: boolean }>();
 const emit = defineEmits<{
@@ -24,6 +25,17 @@ const emit = defineEmits<{
   (e: "drop", note: Note): void;
 }>();
 const notes = useNotesStore();
+const config = useConfigStore();
+
+// --- Retention countdown. A note in Trash is on a clock, and the card is the only
+// place someone browsing Trash would ever find that out in time to restore it.
+// Null whenever nothing is going to happen: not trashed, or retention turned off. ---
+const trashDays = computed(() =>
+  props.note.trashed ? trashDaysLeft(props.note.deleted_at, config.trashRetentionDays) : null,
+);
+const trashCountdown = computed(() => formatTrashCountdown(trashDays.value));
+// Same red the overdue reminder uses — the last few days are worth noticing.
+const trashUrgent = computed(() => trashDays.value !== null && trashDays.value <= 3);
 
 // The card previews the first image inline; non-image files show as compact chips.
 const firstImage = computed(() => props.note.attachments.find((a) => a.mime.startsWith("image/")));
@@ -236,6 +248,32 @@ onBeforeUnmount(() => document.removeEventListener("mousedown", onDocMousedown))
       </span>
     </div>
 
+    <div v-if="trashCountdown" class="mt-2">
+      <span
+        class="inline-flex items-center gap-1 rounded-full px-2 py-0.5 text-xs"
+        :class="
+          trashUrgent
+            ? 'bg-red-100 text-red-700 dark:bg-red-950/50 dark:text-red-300'
+            : 'bg-black/5 text-neutral-600 dark:bg-white/10 dark:text-neutral-300'
+        "
+        :title="`Permanently deleted ${config.trashRetentionDays} days after it was trashed`"
+      >
+        <svg
+          class="h-3 w-3"
+          viewBox="0 0 24 24"
+          fill="none"
+          stroke="currentColor"
+          stroke-width="2"
+          stroke-linecap="round"
+          stroke-linejoin="round"
+        >
+          <circle cx="12" cy="12" r="10" />
+          <polyline points="12 6 12 12 16 14" />
+        </svg>
+        {{ trashCountdown }}
+      </span>
+    </div>
+
     <!-- Toolbar overlays the card's top-right on hover/focus as a floating pill
          (window-control style) instead of reserving a permanent row — so at rest
          the card is content-sized with even padding, not text pinned to the top

frontend/src/components/NoteEditor.vue

@@ -65,6 +65,7 @@ const draftNote = computed<Note>(() => ({
   pinned: false,
   archived: false,
   trashed: false,
+  deleted_at: null,
   remind_at: null,
   recurrence: null,
   labels: labelList.value,

frontend/src/notes/datetime.ts

@@ -49,3 +49,36 @@ export function formatLocalDay(d: Date): string {
   const pad = (n: number) => String(n).padStart(2, "0");
   return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}`;
 }
+
+// --- Trash retention. The server permanently deletes a trashed note once it's older
+// than `trash_retention_days` (0 = keep forever). Counting down from the note's own
+// deleted_at is what turns that from a surprise into a policy: a card in Trash can
+// say how long it has left while there's still time to restore it. ---
+
+const MS_PER_DAY = 24 * 60 * 60 * 1000;
+
+// Whole days a trashed note has left, or null when nothing will happen to it
+// (retention off, or the note isn't trashed).
+//
+// Rounds DOWN deliberately. Rounding up would report "1 day left" for a note with
+// ten minutes on the clock — overstating the time remaining is the one error here
+// that actually costs someone a note.
+export function trashDaysLeft(
+  deletedAt: string | null | undefined,
+  retentionDays: number,
+  now: number = Date.now(),
+): number | null {
+  if (!deletedAt || retentionDays <= 0) return null;
+  const trashedAt = new Date(deletedAt).getTime();
+  if (Number.isNaN(trashedAt)) return null;
+  const remaining = trashedAt + retentionDays * MS_PER_DAY - now;
+  return remaining <= 0 ? 0 : Math.floor(remaining / MS_PER_DAY);
+}
+
+// The countdown as the card shows it. "" when there's nothing to say.
+export function formatTrashCountdown(daysLeft: number | null): string {
+  if (daysLeft === null) return "";
+  if (daysLeft <= 0) return "Deletes today";
+  if (daysLeft === 1) return "1 day left";
+  return `${daysLeft} days left`;
+}

frontend/src/stores/config.ts

@@ -7,6 +7,8 @@ export interface PublicConfig {
   allow_registration: boolean;
   version: string;
   enable_url_unfurl: boolean;
+  // How many days a note survives in Trash before the server purges it. 0 = forever.
+  trash_retention_days: number;
 }
 
 // Public, unauthenticated app config (site name, whether signups are open).
@@ -15,6 +17,10 @@ export const useConfigStore = defineStore("config", () => {
   const allowRegistration = ref(true);
   const version = ref("");
   const enableUrlUnfurl = ref(true);
+  // Mirrors the server default (settings.REGISTRY). Only used if /api/config is
+  // unreachable — and 30 is a safer stand-in than 0, since claiming "kept forever"
+  // when the server is actually purging is the wrong way to be wrong.
+  const trashRetentionDays = ref(30);
   const loaded = ref(false);
 
   async function load(): Promise<void> {
@@ -25,6 +31,7 @@ export const useConfigStore = defineStore("config", () => {
       allowRegistration.value = cfg.allow_registration;
       version.value = cfg.version;
       enableUrlUnfurl.value = cfg.enable_url_unfurl ?? true;
+      trashRetentionDays.value = cfg.trash_retention_days ?? 30;
     } catch {
       // Keep defaults if the config endpoint is unreachable.
     } finally {
@@ -37,5 +44,5 @@ export const useConfigStore = defineStore("config", () => {
     await load();
   }
 
-  return { siteName, allowRegistration, version, enableUrlUnfurl, loaded, load, reload };
+  return { siteName, allowRegistration, version, enableUrlUnfurl, trashRetentionDays, loaded, load, reload };
 });

frontend/src/stores/notes.ts

@@ -76,6 +76,9 @@ export interface Note {
   pinned: boolean;
   archived: boolean;
   trashed: boolean;
+  // When it was trashed (null unless trashed). The Trash view counts the retention
+  // window from here to show how long the note has left before it's purged.
+  deleted_at: string | null;
   remind_at: string | null;
   recurrence: string | null;
   labels: NoteLabel[];

frontend/src/views/BoardView.vue

@@ -2,6 +2,7 @@
 import { computed, onBeforeUnmount, onMounted, ref, watch } from "vue";
 import { useRoute, useRouter } from "vue-router";
 import { useNotesStore, type Note, type NoteView } from "../stores/notes";
+import { useConfigStore } from "../stores/config";
 import { useUiStore } from "../stores/ui";
 import { facetCount, facetsFromQuery, facetsToQuery } from "../notes/facets";
 import { useNoteEditor } from "../composables/useNoteEditor";
@@ -12,6 +13,7 @@ import NoteCard from "../components/NoteCard.vue";
 import NoteEditor from "../components/NoteEditor.vue";
 
 const notes = useNotesStore();
+const config = useConfigStore();
 const route = useRoute();
 const ui = useUiStore();
 const router = useRouter();
@@ -144,6 +146,17 @@ watch(
 );
 watch([currentView, currentLabel, facetKey], () => (focusedIndex.value = -1));
 
+// The retention policy, said out loud at the top of Trash. Empty when retention is
+// off — promising a deletion that never comes is its own kind of lie.
+const retentionNotice = computed(() => {
+  if (currentView.value !== "trash") return "";
+  const days = config.trashRetentionDays;
+  if (days <= 0) return "Notes stay in Trash until you delete them.";
+  return days === 1
+    ? "Notes here are permanently deleted 1 day after you trash them. Restore one to keep it."
+    : `Notes here are permanently deleted ${days} days after you trash them. Restore one to keep it.`;
+});
+
 const emptyState = computed(() => {
   if (filtered.value) return { title: "No notes match these filters", subtitle: "Try clearing or loosening a facet." };
   if (currentView.value === "trash") return { title: "Trash is empty", subtitle: "Notes you delete land here first." };
@@ -208,6 +221,13 @@ async function onDrop(target: Note) {
     </button>
     <FilterBar v-if="isMainBoard" />
 
+    <p
+      v-if="retentionNotice"
+      class="mx-auto mb-4 max-w-xl rounded-xl bg-black/5 px-4 py-2.5 text-center text-sm text-neutral-600 dark:bg-white/10 dark:text-neutral-300"
+    >
+      {{ retentionNotice }}
+    </p>
+
     <AsyncState
       :loading="notes.loading"
       :error="loadError || undefined"

src/thoughtsync/app.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
+import asyncio
 import mimetypes
 import os
 import secrets
+from contextlib import suppress
 from datetime import timedelta
 
 from quart import Quart, has_request_context, jsonify, request, send_from_directory
@@ -15,6 +17,7 @@ from .db import session_scope
 from .graph import bp as graph_bp
 from .labels import bp as labels_bp
 from .notes import bp as notes_bp
+from .retention import run_sweeper
 from .saved_filters import bp as saved_filters_bp
 from .settings import get_public_config, get_setting, load_or_create_secret_key
 from .settings_api import bp as settings_bp
@@ -80,6 +83,21 @@ def create_app() -> Quart:
                 app.config["PERMANENT_SESSION_LIFETIME"] = timedelta(days=days)
             except (ValueError, TypeError, KeyError):
                 pass
+        # Expire old trash in the background (retention.py). One task per process is
+        # correct because the image serves with a single hypercorn worker (Dockerfile);
+        # if that ever gains `--workers`, this needs a lock so N workers don't each
+        # sweep. Duplicate sweeps would be harmless but wasteful — a purged row is
+        # skipped by `purged_at IS NULL` — so this is about load, not correctness.
+        app.config["TRASH_SWEEPER"] = asyncio.create_task(run_sweeper())
+
+    @app.after_serving
+    async def _shutdown() -> None:
+        task = app.config.get("TRASH_SWEEPER")
+        if task is not None:
+            task.cancel()
+            # Await the cancellation so shutdown doesn't race a sweep mid-transaction.
+            with suppress(asyncio.CancelledError):
+                await task
 
     @app.get("/api/health")
     async def health():

src/thoughtsync/models/note.py

@@ -83,6 +83,9 @@ class Note(Base):
             "pinned": self.pinned,
             "archived": self.archived,
             "trashed": self.deleted_at is not None,
+            # WHEN it was trashed, not just that it was: clients count the retention
+            # window from here to show how long a note has left before it's purged.
+            "deleted_at": iso(self.deleted_at),
             "remind_at": iso(self.remind_at),
             "recurrence": self.recurrence,
             "created_at": iso(self.created_at),

src/thoughtsync/notes/__init__.py

@@ -36,6 +36,7 @@ from ..models.note_link import NoteLink
 from ..models.note_link_preview import NoteLinkPreview
 from ..models.note_revision import NoteRevision
 from ..responses import json_error, not_found, parse_uuid
+from ..retention import purge_note
 from ..settings import get_setting
 from ..unfurl import UnfurlError, unfurl
 from ._bp import bp
@@ -971,6 +972,10 @@ async def delete_note(note_id: str):
             return not_found()
         if note.deleted_at is None:
             return json_error("note must be trashed before permanent delete", 409)
-        await db.delete(note)
+        # A tombstone, not a dropped row. Deleting the row outright would leave the
+        # server with no record the note ever existed, so a linked device that was
+        # offline at the time would keep its copy forever — and push it back the
+        # next time it was edited. The delete has to be something clients can LEARN.
+        await purge_note(db, note)
         await db.commit()
         return jsonify({"ok": True})

src/thoughtsync/notes/helpers.py

@@ -45,20 +45,33 @@ def parse_list_items(raw: object) -> list[str]:
 
 
 def apply_filter(stmt, filter_name: str):
-    """Narrow a notes query to one board view."""
+    """Narrow a notes query to one board view.
+
+    Every branch excludes purge tombstones — content-less rows kept only so the sync
+    feed can tell offline clients a note is gone (see `retention.purge_note`). The
+    active/archived branches get that for free from `deleted_at IS NULL`, since a
+    tombstone keeps the timestamp; Trash is the one view that has to say so.
+    """
     if filter_name == "archived":
         return stmt.where(Note.deleted_at.is_(None), Note.archived.is_(True))
     if filter_name == "trash":
-        return stmt.where(Note.deleted_at.is_not(None))
+        return stmt.where(Note.deleted_at.is_not(None), Note.purged_at.is_(None))
     return stmt.where(Note.deleted_at.is_(None), Note.archived.is_(False))
 
 
 async def _get_owned(db, note_id: str) -> Note | None:
-    """Fetch a note the current user OWNS (mutations are owner-only in M1/M2)."""
+    """Fetch a note the current user OWNS (mutations are owner-only in M1/M2).
+
+    A purged note reads as absent: the REST API must treat it as gone, so opening,
+    editing or restoring one 404s. The sync push path looks rows up directly rather
+    than through here, which is what still lets a client re-create an id it owns.
+    """
     nid = parse_uuid(note_id)
     if nid is None:
         return None
-    return await db.scalar(select(Note).where(Note.id == nid, Note.owner_id == g.user_id))
+    return await db.scalar(
+        select(Note).where(Note.id == nid, Note.owner_id == g.user_id, Note.purged_at.is_(None))
+    )
 
 
 def _escape_like(s: str) -> str:

src/thoughtsync/retention.py

@@ -0,0 +1,170 @@
+"""Trash retention — what "permanently deleted" means, and when it happens by itself.
+
+Two things live here, deliberately together:
+
+**`purge_note`** — the single definition of destroying a note. Three callers reach
+permanent deletion by different routes (the user's Delete forever in the web UI,
+a client's `op=delete` over sync, and the sweeper below), and if each had its own
+idea of what to tear down they would drift — one would forget the files, another
+the revision history, and "permanently deleted" would quietly mean three different
+things depending on how you got there.
+
+**The sweeper** — trash that nobody empties is not free: a trashed note keeps its
+attachment BYTES on disk for as long as it sits there. So trash expires. The window
+is the `trash_retention_days` setting (default 30, `0` = keep forever), re-read on
+every pass so a change in admin Settings takes effect without a restart.
+
+A purged note is not a deleted ROW — it's a content-less tombstone. That's what lets
+an offline client that reappears next month learn the note is gone instead of
+faithfully resurrecting it on the next push.
+"""
+from __future__ import annotations
+
+import asyncio
+import logging
+from datetime import datetime, timedelta, timezone
+
+from sqlalchemy import delete as sa_delete
+from sqlalchemy import select
+
+from .config import Config
+from .db import session_scope
+from .models.label import NoteLabel
+from .models.note import Note
+from .models.note_attachment import NoteAttachment
+from .models.note_item import NoteItem
+from .models.note_link import NoteLink
+from .models.note_link_preview import NoteLinkPreview
+from .models.note_revision import NoteRevision
+from .settings import get_setting
+
+logger = logging.getLogger(__name__)
+
+# How often the sweeper wakes. Retention is measured in days, so anything under
+# "a few times a day" buys nothing but load — a note trashed at 09:00 expiring at
+# 14:00 rather than 09:00 thirty days later is not a difference anyone can feel.
+SWEEP_INTERVAL_SECONDS = 6 * 60 * 60
+
+# Let the app finish booting (migrations, first requests) before the first sweep.
+SWEEP_STARTUP_DELAY_SECONDS = 60
+
+# Rows purged per transaction. A long-neglected install could have thousands of
+# expired notes on the first sweep; committing in batches keeps that from becoming
+# one enormous transaction holding locks while it deletes files.
+SWEEP_BATCH = 200
+
+
+def expired_before(now: datetime, retention_days: int) -> datetime | None:
+    """The cutoff: trash older than this has expired. `None` = retention is off.
+
+    Kept separate from the query so the window arithmetic — including the two ways
+    to say "never" (0 and negative, the latter reachable by typing a stray minus in
+    Settings) — is testable without a database.
+    """
+    if retention_days <= 0:
+        return None
+    return now - timedelta(days=retention_days)
+
+
+async def purge_note(db, note: Note, edited_at: datetime | None = None) -> None:
+    """Turn a note into a content-less tombstone: delete its children (and the
+    attachment files on disk), clear its content, stamp `purged_at`.
+
+    The row survives on purpose — offline clients read it off the delta feed and
+    learn the note is gone. Everything that carries the note's CONTENT goes, and
+    that includes history: a revision row holds the full body, so leaving revisions
+    behind would mean the text of a "permanently deleted" note is still on the
+    server, recoverable by anyone who can read the table.
+    """
+    atts = (await db.scalars(select(NoteAttachment).where(NoteAttachment.note_id == note.id))).all()
+    for a in atts:
+        try:
+            (Config.media_root() / a.path).unlink(missing_ok=True)
+        except OSError:
+            # A missing or unreadable file must not strand the row: the DB record is
+            # what the user asked us to destroy, and a failed unlink leaving it in
+            # place would make the note reappear whole on the next sweep.
+            logger.warning("couldn't remove attachment file %s during purge", a.path, exc_info=True)
+    await db.execute(sa_delete(NoteAttachment).where(NoteAttachment.note_id == note.id))
+    await db.execute(sa_delete(NoteItem).where(NoteItem.note_id == note.id))
+    await db.execute(sa_delete(NoteLabel).where(NoteLabel.note_id == note.id))
+    await db.execute(sa_delete(NoteLink).where(NoteLink.source_id == note.id))
+    await db.execute(sa_delete(NoteLinkPreview).where(NoteLinkPreview.note_id == note.id))
+    await db.execute(sa_delete(NoteRevision).where(NoteRevision.note_id == note.id))
+    note.title = None
+    note.body = ""
+    note.display_title = ""
+    # `deleted_at` deliberately SURVIVES. It's still true — that is when the note was
+    # deleted — and keeping it means every ordinary query, present and future, that
+    # says "not trashed" (`deleted_at IS NULL`) excludes tombstones for free. Clearing
+    # it would leave a content-less row looking like a perfectly normal active note,
+    # and it would surface on the board as a blank card. Only the Trash view, which
+    # asks for `deleted_at IS NOT NULL`, has to name `purged_at` explicitly.
+    note.remind_at = None
+    note.purged_at = datetime.now(timezone.utc)
+    if edited_at is not None:
+        note.updated_at = edited_at
+
+
+async def sweep_expired_trash(db, retention_days: int, *, now: datetime | None = None) -> int:
+    """Purge every note whose trash has expired. Returns how many were purged.
+
+    Runs across ALL owners — it's a server-wide policy, not a per-user action, and
+    the sweeper has no session to scope it by (rule 47 is about honoring the ACL on
+    user-initiated reads, not about exempting rows from server maintenance).
+    """
+    cutoff = expired_before(now or datetime.now(timezone.utc), retention_days)
+    if cutoff is None:
+        return 0
+    total = 0
+    while True:
+        expired = (
+            await db.scalars(
+                select(Note)
+                .where(
+                    Note.deleted_at.is_not(None),
+                    Note.deleted_at < cutoff,
+                    # Already a tombstone. Without this the purge would re-run on
+                    # every sweep forever, bumping sync_revision each time and
+                    # handing clients an endless stream of "news" about one note.
+                    Note.purged_at.is_(None),
+                )
+                .order_by(Note.deleted_at)
+                .limit(SWEEP_BATCH)
+            )
+        ).all()
+        if not expired:
+            return total
+        for note in expired:
+            await purge_note(db, note)
+        await db.commit()
+        total += len(expired)
+
+
+async def sweep_once() -> int:
+    """One sweep against the live retention setting, in its own session."""
+    async with session_scope() as db:
+        try:
+            days = int(await get_setting(db, "trash_retention_days"))
+        except (KeyError, TypeError, ValueError):
+            return 0
+        return await sweep_expired_trash(db, days)
+
+
+async def run_sweeper() -> None:
+    """The background loop. Started in `before_serving`, cancelled on shutdown.
+
+    A sweep failure (DB blip, unreadable media directory) must never take the loop
+    down with it — the next pass simply finds the same expired rows and tries again.
+    """
+    await asyncio.sleep(SWEEP_STARTUP_DELAY_SECONDS)
+    while True:
+        try:
+            purged = await sweep_once()
+            if purged:
+                logger.info("trash retention: purged %d expired note(s)", purged)
+        except asyncio.CancelledError:
+            raise
+        except Exception:
+            logger.exception("trash retention sweep failed; will retry next pass")
+        await asyncio.sleep(SWEEP_INTERVAL_SECONDS)

src/thoughtsync/settings.py

@@ -43,6 +43,15 @@ REGISTRY: list[SettingDef] = [
         "How long a signed-in session stays valid before another login is required.",
         "Access",
     ),
+    SettingDef(
+        "trash_retention_days",
+        "int",
+        30,
+        "Trash retention (days)",
+        "How long a note stays in Trash before it's permanently deleted, freeing its "
+        "attachments from disk. Set to 0 to keep trashed notes until they're deleted by hand.",
+        "Notes",
+    ),
     SettingDef(
         "max_attachment_mb",
         "int",
@@ -117,11 +126,16 @@ async def get_setting(db, key: str) -> Any:
 
 
 async def get_public_config(db) -> dict:
-    """Non-sensitive settings the unauthenticated login/register screen needs."""
+    """Non-sensitive settings every client reads — the login/register screen before
+    sign-in, and the app itself afterwards. Nothing here is owner-scoped."""
     return {
         "site_name": await get_setting(db, "site_name"),
         "allow_registration": await get_setting(db, "allow_registration"),
         "enable_url_unfurl": await get_setting(db, "enable_url_unfurl"),
+        # Server policy, not user data: clients need it to say how long a note has
+        # left in Trash. A native client also reads it BEFORE linking, which is why
+        # it belongs on the unauthenticated config rather than behind login.
+        "trash_retention_days": await get_setting(db, "trash_retention_days"),
     }
 
 

src/thoughtsync/sync.py

@@ -20,14 +20,11 @@ from sqlalchemy import func, select
 
 from .auth import login_required
 from .common import iso, parse_dt
-from .config import Config
 from .db import session_scope
 from .labeling import reconcile_manual_labels, resolve_owned_label_ids
 from .models.label import Label, NoteLabel
 from .models.note import Note
-from .models.note_attachment import NoteAttachment
 from .models.note_item import NoteItem
-from .models.note_link import NoteLink
 from .models.note_revision import NoteRevision
 from .notes import (
     _reconcile_tags,
@@ -38,6 +35,7 @@ from .notes import (
     normalize_color,
     normalize_recurrence,
 )
+from .retention import purge_note
 from .serialize import serialize_label_sync
 
 bp = Blueprint("sync", __name__, url_prefix="/api/sync")
@@ -245,29 +243,6 @@ async def _apply_note_manual_labels(db, note: Note, ch: dict) -> None:
     await reconcile_manual_labels(db, note, owned)
 
 
-async def _purge_note(db, note: Note, edited_at: datetime | None) -> None:
-    """Turn a note into a content-less tombstone: delete children (+ attachment files),
-    clear content, set purged_at. Kept so offline clients learn it's gone."""
-    atts = (await db.scalars(select(NoteAttachment).where(NoteAttachment.note_id == note.id))).all()
-    for a in atts:
-        try:
-            (Config.media_root() / a.path).unlink(missing_ok=True)
-        except OSError:
-            pass
-    await db.execute(sa_delete(NoteAttachment).where(NoteAttachment.note_id == note.id))
-    await db.execute(sa_delete(NoteItem).where(NoteItem.note_id == note.id))
-    await db.execute(sa_delete(NoteLabel).where(NoteLabel.note_id == note.id))
-    await db.execute(sa_delete(NoteLink).where(NoteLink.source_id == note.id))
-    note.title = None
-    note.body = ""
-    note.display_title = ""
-    note.deleted_at = None
-    note.remind_at = None
-    note.purged_at = datetime.now(timezone.utc)
-    if edited_at is not None:
-        note.updated_at = edited_at
-
-
 async def _apply_note(db, ch: dict) -> dict:
     raw_id = ch.get("id")
     try:
@@ -290,7 +265,7 @@ async def _apply_note(db, ch: dict) -> dict:
             return {"id": str(nid), "entity": "note", "status": "noop"}
         if not client_wins(edited_at, note.updated_at):
             return {"id": str(nid), "entity": "note", "status": "kept", "sync_revision": note.sync_revision}
-        await _purge_note(db, note, edited_at)
+        await purge_note(db, note, edited_at)
         await db.flush()
         await db.refresh(note, ["sync_revision"])
         return {"id": str(nid), "entity": "note", "status": "applied", "sync_revision": note.sync_revision}

tests/test_retention.py

@@ -0,0 +1,83 @@
+from datetime import datetime, timedelta, timezone
+
+import pytest
+
+from thoughtsync.retention import (
+    SWEEP_BATCH,
+    SWEEP_INTERVAL_SECONDS,
+    SWEEP_STARTUP_DELAY_SECONDS,
+    expired_before,
+    sweep_expired_trash,
+)
+from thoughtsync.settings import REGISTRY, get_public_config, validate_updates
+
+NOW = datetime(2026, 7, 26, 12, 0, tzinfo=timezone.utc)
+
+
+def test_expired_before_is_the_window_ago():
+    assert expired_before(NOW, 30) == NOW - timedelta(days=30)
+    assert expired_before(NOW, 1) == NOW - timedelta(days=1)
+
+
+def test_zero_means_keep_forever():
+    # The opt-out. A user who wants Trash to be an indefinite archive gets one, and
+    # `None` is what stops the sweep before it builds a query at all.
+    assert expired_before(NOW, 0) is None
+
+
+def test_a_negative_window_also_means_never():
+    # Reachable by typing a stray minus into the Settings field. The dangerous reading
+    # of -1 would be "expired a day in the FUTURE", which purges the entire trash on
+    # the next sweep; refusing to run is the only safe interpretation.
+    assert expired_before(NOW, -1) is None
+    assert expired_before(NOW, -3650) is None
+
+
+async def test_sweep_is_a_noop_when_retention_is_off():
+    # Passing None as the session proves it: retention off must return before it so
+    # much as touches the database.
+    assert await sweep_expired_trash(None, 0) == 0
+    assert await sweep_expired_trash(None, -1) == 0
+
+
+def test_retention_setting_is_registered_with_a_30_day_default():
+    defn = next((d for d in REGISTRY if d.key == "trash_retention_days"), None)
+    assert defn is not None, "the setting must appear in the admin Settings UI"
+    assert defn.type == "int"
+    assert defn.default == 30
+    # The operator has to be able to tell what it does without reading the code.
+    assert "0" in defn.description, "the keep-forever escape hatch must be documented"
+
+
+def test_retention_setting_accepts_an_int_and_rejects_nonsense():
+    clean, err = validate_updates({"trash_retention_days": "7"})
+    assert err is None
+    assert clean == {"trash_retention_days": 7}
+    _, err = validate_updates({"trash_retention_days": "soon"})
+    assert err is not None
+
+
+async def test_public_config_publishes_the_window():
+    # Clients need it to say how long a note has left in Trash, and a native client
+    # reads it before it holds any credential — so it rides the unauthenticated
+    # config. A stub session stands in for the DB: no row set => registry default.
+    class _NoRows:
+        async def get(self, *_args):
+            return None
+
+    cfg = await get_public_config(_NoRows())
+    assert cfg["trash_retention_days"] == 30
+
+
+@pytest.mark.parametrize(
+    "value", [SWEEP_INTERVAL_SECONDS, SWEEP_STARTUP_DELAY_SECONDS, SWEEP_BATCH]
+)
+def test_sweeper_pacing_constants_are_positive(value):
+    # A zero interval would turn the background loop into a busy spin against the DB.
+    assert value > 0
+
+
+def test_sweep_interval_is_well_under_a_day():
+    # Retention is measured in days, but the sweep still has to run often enough that
+    # "30 days" doesn't quietly become 31.
+    assert SWEEP_INTERVAL_SECONDS <= 12 * 60 * 60