# Topic Inactivity GC Implementation Plan

> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.

**Goal:** Auto-reap topics that have had no publish or consumer activity for a configurable duration, with pattern-based overrides, soft-delete grace period, and operator observability.

**Architecture:** Extend the existing `RetentionTask::sweep()` to evaluate topic inactivity after segment retention. Activity is tracked via two timestamps — a plain `u64` in the engine layer (publish, behind existing Mutex) and a plain `u64` in the namespace manager layer (subscribe/fetch). Topics enter a 24h grace period before hard deletion, with full admin API visibility. Note: the spec calls for `AtomicU64` on both fields, but since `Topic` is already behind `Mutex<Topic>` and `TopicDeliveryState` is behind `Mutex<TopicDeliveryState>`, plain `u64` is sufficient and simpler.

**Tech Stack:** Rust, tokio, arc-swap, axum, metrics crate

**Spec:** `docs/superpowers/specs/2026-03-29-topic-inactivity-gc-design.md`

---

## File Map

| File | Action | Responsibility |
|---|---|---|
| `src/queue/topic.rs` | Modify | Add `last_activity_millis: u64` to `Topic`, update in `publish()`, initialize in `new()` and `from_replay()` |
| `src/queue/engine.rs` | Modify | Add `topic_last_activity()` and `remove_topic()` to `InMemoryQueueEngine`, add trait methods to `QueueEngine` |
| `src/queue.rs` | Modify | Add `topic_last_activity()` and `remove_topic()` to `QueueEngine` trait |
| `src/namespace/state.rs` | Modify | Add `last_subscribe_millis: u64` to `TopicDeliveryState` |
| `src/namespace/manager.rs` | Modify | Update `subscribe()` to stamp `last_subscribe_millis`, add `topic_last_subscribe()`, `has_active_subscribers()`, `has_inflight_messages()`, `teardown_topic()` |
| `src/namespace.rs` | Modify | Add new methods to `NamespaceManager` trait |
| `src/config.rs` | Modify | Add `topic_inactive_secs`, `topic_reap_grace_period_secs`, `topic_inactive_overrides` to `RetentionPolicy`, add `TopicInactiveOverride` struct, update `RuntimeConfig` with pre-compiled `CompiledInactiveOverride` patterns, validate patterns at config load |
| `src/storage.rs` | Modify | Add `close_topic_writer()` and `delete_topic_dir()` to `StorageBackend` trait |
| `src/storage/fs.rs` | Modify | Implement `close_topic_writer()` and `delete_topic_dir()` on `FsStorage` |
| `src/retention.rs` | Modify | Add `grace_pending` state, topic inactivity sweep logic after segment retention, reap execution |
| `src/server/http.rs` | Modify | Register `GET /admin/topics/pending-reap` route |
| `src/server/http/routes.rs` | Modify | Add `admin_pending_reap` handler |
| `config/config.example.toml` | Modify | Add commented examples for topic inactivity GC config |

---

### Task 1: Add `last_activity_millis` to `Topic`

**Files:**
- Modify: `src/queue/topic.rs:42-59` (struct fields), `src/queue/topic.rs:63-78` (`new()`), `src/queue/topic.rs:86-125` (`from_replay()`), `src/queue/topic.rs:134-263` (`publish()`)

- [ ] **Step 1: Write failing tests for `last_activity_millis`**

Add to the `#[cfg(test)]` module at the bottom of `src/queue/topic.rs`:

```rust
#[test]
fn new_topic_has_recent_last_activity() {
    let t = Topic::new(100, 0, BackpressurePolicy::Reject);
    let now = crate::types::now_millis();
    // Should be within 1 second of now
    assert!(t.last_activity_millis() <= now);
    assert!(t.last_activity_millis() >= now - 1000);
}

#[test]
fn publish_updates_last_activity() {
    let mut t = Topic::new(100, 0, BackpressurePolicy::Reject);
    let before = t.last_activity_millis();
    std::thread::sleep(std::time::Duration::from_millis(5));
    t.publish(Arc::new(make_envelope(0, b"data")), &ns(), &topic_name())
        .unwrap();
    assert!(t.last_activity_millis() >= before);
}

#[test]
fn from_replay_sets_last_activity_from_newest_message() {
    use crate::types::{MessageId, PublishedMessage};
    let messages: Vec<PublishedMessage> = (0..3u64)
        .map(|i| PublishedMessage {
            offset: TopicOffset(i),
            envelope: Arc::new(MessageEnvelope {
                id: MessageId::new(i + 1),
                timestamp: 1000 + i * 100, // 1000, 1100, 1200
                payload: b"data".to_vec(),
                headers: vec![],
                ttl_ms: 0,
            }),
        })
        .collect();
    let t = Topic::from_replay(messages, 100, 0, BackpressurePolicy::Reject);
    assert_eq!(t.last_activity_millis(), 1200);
}

#[test]
fn from_replay_empty_uses_now() {
    let t = Topic::from_replay(vec![], 100, 0, BackpressurePolicy::Reject);
    let now = crate::types::now_millis();
    assert!(t.last_activity_millis() <= now);
    assert!(t.last_activity_millis() >= now - 1000);
}

#[test]
fn from_replay_truncated_uses_newest_retained() {
    use crate::types::{MessageId, PublishedMessage};
    // 5 messages but max_depth=2, so only last 2 are retained
    let messages: Vec<PublishedMessage> = (0..5u64)
        .map(|i| PublishedMessage {
            offset: TopicOffset(i),
            envelope: Arc::new(MessageEnvelope {
                id: MessageId::new(i + 1),
                timestamp: 1000 + i * 100, // 1000..1400
                payload: b"data".to_vec(),
                headers: vec![],
                ttl_ms: 0,
            }),
        })
        .collect();
    let t = Topic::from_replay(messages, 2, 0, BackpressurePolicy::Reject);
    // Should be 1400 (newest retained), not 1000 (newest truncated)
    assert_eq!(t.last_activity_millis(), 1400);
}
```

- [ ] **Step 2: Run tests to verify they fail**

Run: `cargo test --lib queue::topic::tests::new_topic_has_recent_last_activity -- --nocapture`
Expected: FAIL — `last_activity_millis` method does not exist

- [ ] **Step 3: Add `last_activity_millis` field and accessor**

In `src/queue/topic.rs`, add to `Topic` struct (after `evictions_since_warn`):

```rust
/// Epoch millis of last publish or consume on this topic. Used by topic inactivity GC.
last_activity_millis: u64,
```

Note: NOT `AtomicU64` — `Topic` is behind `Mutex<Topic>`, so all access is already synchronized. A plain `u64` is sufficient and simpler.

Add public accessor after `bytes_used()`:

```rust
/// Last activity timestamp (epoch millis). Updated on publish and consume.
pub fn last_activity_millis(&self) -> u64 {
    self.last_activity_millis
}
```

Update `new()` to initialize:

```rust
last_activity_millis: crate::types::now_millis(),
```

Update `from_replay()` — in the non-empty path, after constructing the `Self` struct, set `last_activity_millis` to the newest message's timestamp. In the `Self { ... }` block at line 112:

```rust
last_activity_millis: retained.last().unwrap().envelope.timestamp,
```

The empty case calls `Self::new()` which already sets `now_millis()`.

- [ ] **Step 4: Update `publish()` to stamp activity**

In `publish()`, right before the `Ok((offset, evicted_bytes, evicted_messages))` return at line 263, add:

```rust
self.last_activity_millis = crate::types::now_millis();
```

- [ ] **Step 5: Run all tests to verify they pass**

Run: `cargo test --lib queue::topic`
Expected: All tests PASS including the 5 new ones

- [ ] **Step 6: Commit**

```bash
git add src/queue/topic.rs
git commit -m "feat(gc): add last_activity_millis to Topic, updated on publish"
```

Note: `consume_from()` is NOT stamped. Activity from the consumer side is tracked at the namespace manager layer (`last_subscribe_millis` on subscribe/fetch). The engine-level timestamp only reflects producer activity. Inflight messages block reaping via the `has_inflight_messages()` safety check, so delivery-task reads do not need to stamp activity.

---

### Task 2: Add `last_subscribe_millis` to `TopicDeliveryState`

**Files:**
- Modify: `src/namespace/state.rs:141-170` (struct + constructor)
- Modify: `src/namespace/manager.rs` (subscribe paths)

- [ ] **Step 1: Write failing test**

Add to the `#[cfg(test)]` module in `src/namespace/manager.rs`:

```rust
#[tokio::test]
async fn subscribe_updates_last_subscribe_millis() {
    let engine = make_engine();
    let mgr = make_manager(engine.clone());
    let ns = ns();
    let topic = topic();

    // Publish a message so the topic exists
    let env = make_envelope(0, "ns", "test-topic", b"data");
    mgr.publish(&ns, &topic, env).await.unwrap();

    let ns_state = mgr.namespace_state(&ns).unwrap();
    let topics = ns_state.topics.read().await;
    let topic_state = topics.get(&topic).unwrap().lock().await;
    let before = topic_state.last_subscribe_millis;
    drop(topic_state);
    drop(topics);

    std::thread::sleep(std::time::Duration::from_millis(5));

    // Subscribe
    mgr.subscribe(
        &ns,
        &topic,
        ClientId(1),
        None,
        SubscribeStart::Head,
    )
    .await
    .unwrap();

    let topics = ns_state.topics.read().await;
    let topic_state = topics.get(&topic).unwrap().lock().await;
    assert!(topic_state.last_subscribe_millis >= before);
}
```

- [ ] **Step 2: Run test to verify it fails**

Run: `cargo test --lib namespace::manager::tests::subscribe_updates_last_subscribe_millis`
Expected: FAIL — `last_subscribe_millis` field does not exist

- [ ] **Step 3: Add `last_subscribe_millis` field**

In `src/namespace/state.rs`, add to `TopicDeliveryState` struct:

```rust
/// Epoch millis of last subscribe/fetch on this topic. Used by topic inactivity GC.
pub last_subscribe_millis: u64,
```

Update `TopicDeliveryState::new()` to initialize:

```rust
last_subscribe_millis: 0,
```

- [ ] **Step 4: Update subscribe paths to stamp `last_subscribe_millis`**

In `src/namespace/manager.rs`, find each subscribe method implementation (`subscribe`, `subscribe_at`, `subscribe_from`) and add after the topic delivery state is obtained:

```rust
topic_state.last_subscribe_millis = crate::types::now_millis();
```

Search for `SubscribeStart::Head` and `SubscribeStart::At` and `SubscribeStart::From` to find all paths. The key spots are where `TopicDeliveryState` is accessed with a lock after a successful subscribe operation.

Also update the fetch path in `DefaultNamespaceManager::fetch()`. Fetch is stateless (per CLAUDE.md: "Fetch does not create subscriptions, delivery tasks, or inflight tracking"), so it may not touch `TopicDeliveryState` at all. The approach:

1. In `DefaultNamespaceManager::fetch()`, after the engine fetch call succeeds, check if a `TopicDeliveryState` already exists for this topic (read lock on `ns_state.topics`).
2. If it exists, lock it and stamp `last_subscribe_millis = now_millis()`.
3. If it does NOT exist (no subscriber has ever connected), skip — do not create `TopicDeliveryState` just for fetch. In this case, fetch activity is not tracked, but the topic would only be reaped based on publish activity (engine-level `last_activity_millis`). This is acceptable because a topic with no subscribers AND no recent publishes is indeed inactive.

```rust
// After successful fetch, stamp subscribe activity if delivery state exists
if let Some(ns_state) = self.namespace_state(ns) {
    let topics = ns_state.topics.read().await;
    if let Some(topic_lock) = topics.get(topic) {
        topic_lock.lock().await.last_subscribe_millis = crate::types::now_millis();
    }
}
```

- [ ] **Step 5: Run tests**

Run: `cargo test --lib namespace::manager::tests::subscribe_updates_last_subscribe_millis`
Expected: PASS

- [ ] **Step 6: Commit**

```bash
git add src/namespace/state.rs src/namespace/manager.rs
git commit -m "feat(gc): add last_subscribe_millis to TopicDeliveryState"
```

---

### Task 3: Configuration — `RetentionPolicy` and `TopicInactiveOverride`

**Files:**
- Modify: `src/config.rs:690-707` (`RetentionPolicy`), `src/config.rs:1449-1590` (`RuntimeConfig`)
- Modify: `config/config.example.toml:125-131`

- [ ] **Step 1: Write failing config deserialization test**

Add to `src/config.rs` `#[cfg(test)]` module:

```rust
#[test]
fn retention_policy_with_topic_inactive_secs() {
    let toml_str = r#"
        [storage]
        data_dir = "/tmp/data"

        [namespaces.alpha]
        api_key = "alpha-key-min-24-chars-long"
        permissions = ["Publish"]

        [namespaces.alpha.retention]
        topic_inactive_secs = 604800
        topic_reap_grace_period_secs = 43200

        [[namespaces.alpha.retention.topic_inactive_overrides]]
        pattern = "system.>"
        inactive_secs = 0

        [[namespaces.alpha.retention.topic_inactive_overrides]]
        pattern = "tmp.*"
        inactive_secs = 3600
    "#;
    let config: ServerConfig = toml::from_str(toml_str).unwrap();
    let alpha = &config.namespaces[&NamespaceId("alpha".into())];
    assert_eq!(alpha.retention.topic_inactive_secs, Some(604800));
    assert_eq!(alpha.retention.topic_reap_grace_period_secs, Some(43200));
    assert_eq!(alpha.retention.topic_inactive_overrides.len(), 2);
    assert_eq!(alpha.retention.topic_inactive_overrides[0].pattern, "system.>");
    assert_eq!(alpha.retention.topic_inactive_overrides[0].inactive_secs, 0);
}

#[test]
fn retention_is_active_with_only_topic_inactive_secs() {
    let policy = RetentionPolicy {
        max_age_secs: None,
        max_bytes: None,
        max_segments: None,
        topic_inactive_secs: Some(86400),
        topic_reap_grace_period_secs: None,
        topic_inactive_overrides: vec![],
    };
    assert!(policy.is_active());
}

#[test]
fn retention_policy_default_has_no_gc() {
    let policy = RetentionPolicy::default();
    assert_eq!(policy.topic_inactive_secs, None);
    assert_eq!(policy.topic_reap_grace_period_secs, None);
    assert!(policy.topic_inactive_overrides.is_empty());
    assert!(!policy.is_active());
}
```

- [ ] **Step 2: Run tests to verify they fail**

Run: `cargo test --lib config::tests::retention_policy_with_topic_inactive_secs`
Expected: FAIL — fields don't exist

- [ ] **Step 3: Add `TopicInactiveOverride` struct**

In `src/config.rs`, after the `RetentionPolicy` impl block (after line 708):

```rust
/// Per-topic override for inactivity GC thresholds. Evaluated in order; first match wins.
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
pub struct TopicInactiveOverride {
    /// NATS-style wildcard pattern (e.g., "system.>", "tmp.*").
    pub pattern: String,
    /// Inactivity threshold in seconds. 0 = never reap matching topics.
    pub inactive_secs: u64,
}
```

- [ ] **Step 4: Add new fields to `RetentionPolicy`**

```rust
#[derive(Debug, Clone, Default, PartialEq, Serialize, Deserialize)]
#[serde(default, deny_unknown_fields)]
pub struct RetentionPolicy {
    pub max_age_secs: Option<u64>,
    pub max_bytes: Option<u64>,
    pub max_segments: Option<usize>,
    /// Reap entire topic after this many seconds of inactivity. None = no reaping.
    pub topic_inactive_secs: Option<u64>,
    /// Grace period before hard delete. None = 86400 (24 hours).
    pub topic_reap_grace_period_secs: Option<u64>,
    /// Per-topic overrides. First matching pattern wins. Empty = use namespace default.
    #[serde(default)]
    pub topic_inactive_overrides: Vec<TopicInactiveOverride>,
}
```

Update `is_active()`:

```rust
pub fn is_active(&self) -> bool {
    self.max_age_secs.is_some()
        || self.max_bytes.is_some()
        || self.max_segments.is_some()
        || self.topic_inactive_secs.is_some()
}
```

- [ ] **Step 5: Update `RuntimeConfig` to carry the new fields**

The `namespace_retention` field already carries `RetentionPolicy` per namespace, so `topic_inactive_secs`, `topic_reap_grace_period_secs`, and `topic_inactive_overrides` flow through automatically via the existing `RetentionPolicy` clone in `RuntimeConfig::from_server_config`.

Verify: check that `namespace_retention.insert(ns_id.clone(), ns_cfg.retention.clone())` still compiles with the new fields. It should, since `Clone` is derived.

Update the `RuntimeConfigSnapshot` (used for `GET /admin/config` display) if it includes retention details — check if the new fields show up in the JSON output.

- [ ] **Step 6: Add pre-compiled override struct and config validation**

Add a struct for pre-compiled overrides (does NOT need serde — it's runtime-only):

```rust
/// Pre-compiled version of TopicInactiveOverride for use at sweep time.
#[derive(Debug, Clone)]
pub struct CompiledInactiveOverride {
    pub pattern: crate::wildcard::TopicPattern,
    pub inactive_secs: u64,
}
```

Add a new field to `RuntimeConfig`:

```rust
pub compiled_inactive_overrides: HashMap<NamespaceId, Vec<CompiledInactiveOverride>>,
```

In `RuntimeConfig::from_server_config()`, after inserting into `namespace_retention`, compile the overrides:

```rust
let mut compiled_inactive_overrides = HashMap::new();
for (ns_id, ns_cfg) in &config.namespaces {
    let mut compiled = Vec::new();
    for ov in &ns_cfg.retention.topic_inactive_overrides {
        let pattern = crate::wildcard::TopicPattern::parse(&ov.pattern)
            .unwrap_or_else(|e| panic!(
                "invalid topic_inactive_overrides pattern '{}' in namespace {}: {e}",
                ov.pattern, ns_id.0
            ));
        compiled.push(CompiledInactiveOverride {
            pattern,
            inactive_secs: ov.inactive_secs,
        });
    }
    if !compiled.is_empty() {
        compiled_inactive_overrides.insert(ns_id.clone(), compiled);
    }
}
```

Note: Using `panic!` at config load is consistent with other config validation in the project. Invalid patterns are a hard error — the server should not start with a config it cannot evaluate.

- [ ] **Step 7: Write test for invalid pattern rejection**

```rust
#[test]
#[should_panic(expected = "invalid topic_inactive_overrides pattern")]
fn invalid_override_pattern_panics_at_config_load() {
    let mut config = ServerConfig::default();
    let ns_id = NamespaceId("test".into());
    config.namespaces.insert(ns_id, NamespaceConfig {
        // ... minimal valid fields ...
        retention: RetentionPolicy {
            topic_inactive_secs: Some(3600),
            topic_inactive_overrides: vec![TopicInactiveOverride {
                pattern: "invalid.[".into(), // malformed pattern
                inactive_secs: 0,
            }],
            ..Default::default()
        },
        // ... other required fields ...
    });
    RuntimeConfig::from_server_config(&config); // should panic
}
```

- [ ] **Step 8: Update `config.example.toml`**

After line 131 (after `# max_segments = 64`), add:

```toml
# topic_inactive_secs = 604800           # reap entire topic after 7 days of inactivity (optional, default: no reaping)
# topic_reap_grace_period_secs = 86400   # grace period before hard delete (optional, default: 24 hours)
#
# Per-topic overrides — first matching pattern wins. 0 = never reap.
# [[namespaces.default.retention.topic_inactive_overrides]]
# pattern = "system.>"
# inactive_secs = 0
```

- [ ] **Step 9: Run all config tests**

Run: `cargo test --lib config::tests`
Expected: All tests PASS including the new ones. Existing tests still pass.

- [ ] **Step 10: Commit**

```bash
git add src/config.rs config/config.example.toml
git commit -m "feat(gc): add topic_inactive_secs, grace period, and overrides to RetentionPolicy"
```

---

### Task 4: Add `QueueEngine` trait methods and implementation

**Files:**
- Modify: `src/queue.rs:28-140` (`QueueEngine` trait)
- Modify: `src/queue/engine.rs:391+` (`InMemoryQueueEngine` impl)
- Modify: `src/storage.rs` (mock storage if it implements `QueueEngine`)

- [ ] **Step 1: Write failing test for `topic_last_activity`**

Add to `src/queue/engine.rs` `#[cfg(test)]` module:

```rust
#[tokio::test]
async fn topic_last_activity_returns_none_for_missing_topic() {
    let engine = make_engine();
    assert!(engine.topic_last_activity(&ns(), &topic()).await.is_none());
}

#[tokio::test]
async fn topic_last_activity_returns_timestamp_after_publish() {
    let engine = make_engine();
    let before = crate::types::now_millis();
    engine
        .publish(&ns(), &topic(), make_envelope(1, b"data"))
        .await
        .unwrap();
    let activity = engine.topic_last_activity(&ns(), &topic()).await.unwrap();
    assert!(activity >= before);
}

#[tokio::test]
async fn remove_topic_removes_from_map() {
    let engine = make_engine();
    engine
        .publish(&ns(), &topic(), make_envelope(1, b"data"))
        .await
        .unwrap();
    assert!(engine.topic_last_activity(&ns(), &topic()).await.is_some());
    engine.remove_topic(&ns(), &topic()).await.unwrap();
    assert!(engine.topic_last_activity(&ns(), &topic()).await.is_none());
}

#[tokio::test]
async fn remove_topic_returns_freed_stats() {
    let engine = make_engine();
    engine
        .publish(&ns(), &topic(), make_envelope(1, b"hello"))
        .await
        .unwrap();
    let bytes_before = engine.total_bytes();
    assert!(bytes_before > 0);
    let (bytes_freed, messages_freed) = engine.remove_topic(&ns(), &topic()).await.unwrap();
    assert!(bytes_freed > 0);
    assert_eq!(messages_freed, 1);
    assert_eq!(engine.total_bytes(), 0);
}

#[tokio::test]
async fn remove_topic_nonexistent_returns_zero() {
    let engine = make_engine();
    let (bytes, msgs) = engine.remove_topic(&ns(), &topic()).await.unwrap();
    assert_eq!(bytes, 0);
    assert_eq!(msgs, 0);
}
```

- [ ] **Step 2: Run tests to verify they fail**

Run: `cargo test --lib queue::engine::tests::topic_last_activity_returns_none`
Expected: FAIL — method does not exist

- [ ] **Step 3: Add trait methods to `QueueEngine`**

In `src/queue.rs`, add to the `QueueEngine` trait (before `shutdown`):

```rust
/// Return the last activity timestamp (epoch millis) for a topic.
/// Returns `None` if the topic does not exist.
fn topic_last_activity(
    &self,
    ns: &NamespaceId,
    topic: &TopicName,
) -> impl Future<Output = Option<u64>> + Send;

/// Remove a topic from the in-memory map. No-op if the topic does not exist.
/// Updates global memory accounting. Does NOT touch storage — caller handles that.
/// Returns (bytes_freed, messages_freed).
fn remove_topic(
    &self,
    ns: &NamespaceId,
    topic: &TopicName,
) -> impl Future<Output = Result<(u64, usize)>> + Send;
```

- [ ] **Step 4: Implement on `InMemoryQueueEngine`**

In `src/queue/engine.rs`, add to the `impl QueueEngine for InMemoryQueueEngine<S>` block:

```rust
async fn topic_last_activity(&self, ns: &NamespaceId, topic: &TopicName) -> Option<u64> {
    let map = self.topics.read().await;
    let topic_lock = map.get(&(ns.clone(), topic.clone()))?;
    let t = topic_lock.lock().await;
    Some(t.last_activity_millis())
}

async fn remove_topic(&self, ns: &NamespaceId, topic: &TopicName) -> Result<(u64, usize)> {
    let mut map = self.topics.write().await;
    let key = (ns.clone(), topic.clone());
    if let Some(topic_lock) = map.remove(&key) {
        let t = topic_lock.lock().await;
        let bytes_freed = t.bytes_used() as u64;
        let messages_freed = t.depth();
        drop(t);
        self.total_bytes.fetch_sub(bytes_freed, Relaxed);
        tracing::debug!(namespace = %ns, topic = %topic, bytes_freed, messages_freed, "topic removed from engine");
        Ok((bytes_freed, messages_freed))
    } else {
        Ok((0, 0))
    }
}
```

- [ ] **Step 5: Update mock storage / test implementations**

If `MockStorage` or any test helper implements `QueueEngine`, add default/stub implementations of the new trait methods. Search for `impl QueueEngine` across the codebase to find all implementors.

- [ ] **Step 6: Run tests**

Run: `cargo test --lib queue::engine`
Expected: All tests PASS

- [ ] **Step 7: Commit**

```bash
git add src/queue.rs src/queue/engine.rs
git commit -m "feat(gc): add topic_last_activity and remove_topic to QueueEngine"
```

---

### Task 5: Add `NamespaceManager` trait methods and implementation

**Files:**
- Modify: `src/namespace.rs:37+` (`NamespaceManager` trait)
- Modify: `src/namespace/manager.rs` (`DefaultNamespaceManager`)

- [ ] **Step 1: Write failing tests**

Add to `src/namespace/manager.rs` `#[cfg(test)]` module:

```rust
#[tokio::test]
async fn topic_last_subscribe_returns_zero_without_subscribe() {
    let engine = make_engine();
    let mgr = make_manager(engine.clone());
    let ns = ns();
    let topic = topic();

    // Publish creates delivery state but no subscribe
    let env = make_envelope(0, "ns", "test-topic", b"data");
    mgr.publish(&ns, &topic, env).await.unwrap();

    assert_eq!(mgr.topic_last_subscribe(&ns, &topic).await, Some(0));
}

#[tokio::test]
async fn has_active_subscribers_false_when_none() {
    let engine = make_engine();
    let mgr = make_manager(engine.clone());
    let ns = ns();
    let topic = topic();

    let env = make_envelope(0, "ns", "test-topic", b"data");
    mgr.publish(&ns, &topic, env).await.unwrap();

    assert!(!mgr.has_active_subscribers(&ns, &topic).await);
}

#[tokio::test]
async fn has_active_subscribers_true_after_subscribe() {
    let engine = make_engine();
    let mgr = make_manager(engine.clone());
    let ns = ns();
    let topic = topic();

    let env = make_envelope(0, "ns", "test-topic", b"data");
    mgr.publish(&ns, &topic, env).await.unwrap();

    mgr.connect(&ns, ClientId(1)).await.unwrap();
    mgr.subscribe(&ns, &topic, ClientId(1), None, SubscribeStart::Head)
        .await
        .unwrap();

    assert!(mgr.has_active_subscribers(&ns, &topic).await);
}

#[tokio::test]
async fn teardown_topic_removes_delivery_state() {
    let engine = make_engine();
    let mgr = make_manager(engine.clone());
    let ns = ns();
    let topic = topic();

    let env = make_envelope(0, "ns", "test-topic", b"data");
    mgr.publish(&ns, &topic, env).await.unwrap();

    mgr.teardown_topic(&ns, &topic).await.unwrap();

    // Delivery state should be gone
    let ns_state = mgr.namespace_state(&ns).unwrap();
    let topics = ns_state.topics.read().await;
    assert!(!topics.contains_key(&topic));
}
```

- [ ] **Step 2: Run tests to verify they fail**

Run: `cargo test --lib namespace::manager::tests::topic_last_subscribe_returns_zero`
Expected: FAIL — methods don't exist

- [ ] **Step 3: Add trait methods to `NamespaceManager`**

In `src/namespace.rs`, add to the `NamespaceManager` trait:

```rust
/// Return the last subscribe timestamp (epoch millis) for a topic.
/// Returns `None` if no delivery state exists, `Some(0)` if state exists but no subscribe yet.
fn topic_last_subscribe(
    &self,
    ns: &NamespaceId,
    topic: &TopicName,
) -> impl Future<Output = Option<u64>> + Send;

/// Returns true if any individual subscriber or consumer group member is connected.
fn has_active_subscribers(
    &self,
    ns: &NamespaceId,
    topic: &TopicName,
) -> impl Future<Output = bool> + Send;

/// Returns true if any inflight messages are pending ack.
fn has_inflight_messages(
    &self,
    ns: &NamespaceId,
    topic: &TopicName,
) -> impl Future<Output = bool> + Send;

/// Tear down all delivery state for a topic: abort delivery tasks, remove TopicDeliveryState.
/// Wildcard subscription entries in NamespaceState are NOT removed.
fn teardown_topic(
    &self,
    ns: &NamespaceId,
    topic: &TopicName,
) -> impl Future<Output = Result<()>> + Send;
```

- [ ] **Step 4: Implement on `DefaultNamespaceManager`**

In `src/namespace/manager.rs`, add implementations:

```rust
async fn topic_last_subscribe(&self, ns: &NamespaceId, topic: &TopicName) -> Option<u64> {
    let ns_state = self.namespace_state(ns)?;
    let topics = ns_state.topics.read().await;
    let topic_state = topics.get(topic)?;
    let state = topic_state.lock().await;
    Some(state.last_subscribe_millis)
}

async fn has_active_subscribers(&self, ns: &NamespaceId, topic: &TopicName) -> bool {
    let Some(ns_state) = self.namespace_state(ns) else {
        return false;
    };
    let topics = ns_state.topics.read().await;
    let Some(topic_lock) = topics.get(topic) else {
        return false;
    };
    let state = topic_lock.lock().await;
    !state.individual.is_empty()
        || state.groups.values().any(|g| !g.members.is_empty())
}

async fn has_inflight_messages(&self, ns: &NamespaceId, topic: &TopicName) -> bool {
    let Some(ns_state) = self.namespace_state(ns) else {
        return false;
    };
    let topics = ns_state.topics.read().await;
    let Some(topic_lock) = topics.get(topic) else {
        return false;
    };
    let state = topic_lock.lock().await;
    state.individual.values().any(|s| !s.inflight.is_empty())
        || state.groups.values().any(|g| !g.inflight.is_empty())
}

async fn teardown_topic(&self, ns: &NamespaceId, topic: &TopicName) -> Result<()> {
    let ns_state = self
        .namespace_state(ns)
        .ok_or_else(|| Error::NamespaceNotFound(ns.0.clone()))?;

    let mut topics = ns_state.topics.write().await;
    if let Some(topic_lock) = topics.remove(topic) {
        let state = topic_lock.lock().await;
        // Abort all delivery task handles
        for (_, handle) in &state.individual_handles {
            handle.abort();
        }
        for (_, handle) in &state.group_handles {
            handle.abort();
        }
        drop(state);
        // Decrement topic count only when we actually removed a topic
        ns_state
            .topic_count
            .fetch_sub(1, std::sync::atomic::Ordering::Relaxed);
    }
    Ok(())
}
```

- [ ] **Step 5: Run tests**

Run: `cargo test --lib namespace::manager::tests`
Expected: All tests PASS including the 4 new ones

- [ ] **Step 6: Commit**

```bash
git add src/namespace.rs src/namespace/manager.rs
git commit -m "feat(gc): add topic_last_subscribe, has_active_subscribers, has_inflight_messages, teardown_topic"
```

---

### Task 6: Storage layer — `close_topic_writer` and `delete_topic_dir`

**Files:**
- Modify: `src/storage.rs:22-132` (`StorageBackend` trait)
- Modify: `src/storage/fs.rs` (`FsStorage` implementation)

- [ ] **Step 1: Write failing tests**

Add to `src/storage/fs.rs` `#[cfg(test)]` module:

```rust
#[tokio::test]
async fn delete_topic_dir_removes_directory() {
    use crate::types::{MessageEnvelope, MessageId, now_millis};

    let tmp = TempDir::new().unwrap();
    let storage = FsStorage::new(tmp.path().join("data"));
    storage.init().await.unwrap();

    let ns = NamespaceId("ns1".into());
    let topic = TopicName("topic-a".into());

    // Write a message to create the topic directory
    let msg = PublishedMessage {
        offset: TopicOffset(0),
        envelope: Arc::new(MessageEnvelope {
            id: MessageId::new(1),
            timestamp: now_millis(),
            payload: b"data".to_vec(),
            headers: vec![],
            ttl_ms: 0,
        }),
    };
    storage.append(&ns, &topic, &msg).await.unwrap();
    storage.flush(&ns, &topic).await.unwrap();

    // Directory exists
    let seg_dir = storage.segments_dir(&ns, &topic);
    assert!(seg_dir.exists());

    // Close writer then delete
    storage.close_topic_writer(&ns, &topic).await.unwrap();
    storage.delete_topic_dir(&ns, &topic).await.unwrap();

    // Directory gone
    assert!(!seg_dir.exists());
}

#[tokio::test]
async fn close_topic_writer_allows_subsequent_write() {
    use crate::types::{MessageEnvelope, MessageId, now_millis};

    let tmp = TempDir::new().unwrap();
    let storage = FsStorage::new(tmp.path().join("data"));
    storage.init().await.unwrap();

    let ns = NamespaceId("ns1".into());
    let topic = TopicName("topic-a".into());

    let msg = PublishedMessage {
        offset: TopicOffset(0),
        envelope: Arc::new(MessageEnvelope {
            id: MessageId::new(1),
            timestamp: now_millis(),
            payload: b"data".to_vec(),
            headers: vec![],
            ttl_ms: 0,
        }),
    };
    storage.append(&ns, &topic, &msg).await.unwrap();
    storage.close_topic_writer(&ns, &topic).await.unwrap();

    // Should be able to write again (get_or_create_topic reopens)
    let msg2 = PublishedMessage {
        offset: TopicOffset(1),
        envelope: Arc::new(MessageEnvelope {
            id: MessageId::new(2),
            timestamp: now_millis(),
            payload: b"data2".to_vec(),
            headers: vec![],
            ttl_ms: 0,
        }),
    };
    storage.append(&ns, &topic, &msg2).await.unwrap();
}

#[tokio::test]
async fn delete_topic_dir_nonexistent_is_ok() {
    let tmp = TempDir::new().unwrap();
    let storage = FsStorage::new(tmp.path().join("data"));
    storage.init().await.unwrap();

    let ns = NamespaceId("ns1".into());
    let topic = TopicName("nonexistent".into());

    // Should not error
    storage.delete_topic_dir(&ns, &topic).await.unwrap();
}
```

- [ ] **Step 2: Run tests to verify they fail**

Run: `cargo test --lib storage::fs::tests::delete_topic_dir_removes_directory`
Expected: FAIL — methods don't exist

- [ ] **Step 3: Add trait methods to `StorageBackend`**

In `src/storage.rs`, add to the `StorageBackend` trait (before `shutdown`):

```rust
/// Flush and close the active writer for a topic, evicting it from the fd cache.
/// No-op if no active writer exists. Must be called before `delete_topic_dir`.
fn close_topic_writer(
    &self,
    namespace: &NamespaceId,
    topic: &TopicName,
) -> impl Future<Output = Result<()>> + Send;

/// Delete the entire segment directory for a topic.
/// No-op if the directory does not exist.
fn delete_topic_dir(
    &self,
    namespace: &NamespaceId,
    topic: &TopicName,
) -> impl Future<Output = Result<()>> + Send;
```

- [ ] **Step 4: Implement on `FsStorage`**

In `src/storage/fs.rs`, implement both methods. The writer cache is the LRU cache inside `FsStorage`. Look at how `shutdown()` flushes all writers — `close_topic_writer` does the same for a single topic.

`close_topic_writer`: acquire the writer cache lock, look up the key `(namespace, topic)`, flush the segment file + index file if present, remove the entry from the cache.

`delete_topic_dir`: call `self.segments_dir(namespace, topic)` to get the path, then `tokio::fs::remove_dir_all(path).await`. Return `Ok(())` if the directory doesn't exist.

```rust
async fn close_topic_writer(&self, namespace: &NamespaceId, topic: &TopicName) -> Result<()> {
    let key = (namespace.0.clone(), topic.0.clone());
    let mut cache = self.writers.lock().await;
    if let Some(writer) = cache.pop(&key) {
        writer.file.into_inner().shutdown().await?;
        if let Some(idx) = writer.index_file {
            idx.into_inner().shutdown().await?;
        }
    }
    Ok(())
}

async fn delete_topic_dir(&self, namespace: &NamespaceId, topic: &TopicName) -> Result<()> {
    let dir = self.segments_dir(namespace, topic);
    match tokio::fs::remove_dir_all(&dir).await {
        Ok(()) => Ok(()),
        Err(e) if e.kind() == std::io::ErrorKind::NotFound => Ok(()),
        Err(e) => Err(e.into()),
    }
}
```

Note: The exact `close_topic_writer` implementation depends on how the writer cache stores entries. Read the `FsStorage` struct and its writer cache type to get the exact field names and flush API. The above is directional — adapt to match the actual struct fields (`self.writers` or similar).

- [ ] **Step 5: Update `MockStorage`**

In `src/storage.rs` mock module, add stub implementations:

```rust
async fn close_topic_writer(&self, _ns: &NamespaceId, _topic: &TopicName) -> Result<()> {
    Ok(())
}

async fn delete_topic_dir(&self, _ns: &NamespaceId, _topic: &TopicName) -> Result<()> {
    Ok(())
}
```

- [ ] **Step 6: Run tests**

Run: `cargo test --lib storage::fs::tests`
Expected: All tests PASS

- [ ] **Step 7: Commit**

```bash
git add src/storage.rs src/storage/fs.rs
git commit -m "feat(gc): add close_topic_writer and delete_topic_dir to StorageBackend"
```

---

### Task 7: Topic inactivity sweep in `RetentionTask`

**Files:**
- Modify: `src/retention.rs`

- [ ] **Step 1: Add `grace_pending` state to `RetentionTask`**

Use `Arc<tokio::sync::RwLock<>>` from the start so the admin API (Task 8) can read it without refactoring.

```rust
use std::collections::HashMap as StdHashMap;

type GracePendingMap = StdHashMap<(NamespaceId, TopicName), std::time::Instant>;

pub struct RetentionTask<S: StorageBackend, M: NamespaceManager, Q: QueueEngine> {
    storage: std::sync::Arc<S>,
    manager: std::sync::Arc<M>,
    engine: std::sync::Arc<Q>,
    shared_config: SharedConfig,
    grace_pending: Arc<tokio::sync::RwLock<GracePendingMap>>,
}
```

Update `new()` to initialize `grace_pending: Arc::new(tokio::sync::RwLock::new(HashMap::new()))`.

Add a public accessor for the admin API:

```rust
pub fn grace_pending(&self) -> Arc<tokio::sync::RwLock<GracePendingMap>> {
    Arc::clone(&self.grace_pending)
}
```

`run()` keeps its current `self` (by value) signature — `Arc<RwLock<>>` is cloneable, so the admin API clones the handle before `run()` consumes `self`.

- [ ] **Step 2: Add the topic inactivity sweep method**

Add a new method `sweep_inactive_topics()` to `RetentionTask`, called at the end of `sweep()`:

```rust
async fn sweep_inactive_topics(
    &self,
    all_topics: &[(NamespaceId, TopicName)],
) {
    let cfg = self.shared_config.load();
    let namespace_retention = &cfg.namespace_retention;
    let compiled_overrides = &cfg.compiled_inactive_overrides;

    let now_millis = crate::types::now_millis();
    let now_instant = std::time::Instant::now();
    let default_grace_secs: u64 = 86400;

    for (ns, topic) in all_topics {
        // 1. Load config
        let Some(policy) = namespace_retention.get(ns) else { continue };
        let Some(ns_inactive_secs) = policy.topic_inactive_secs else { continue };

        // 2. Check overrides (pre-compiled patterns from RuntimeConfig)
        let overrides = compiled_overrides.get(ns).map(|v| v.as_slice()).unwrap_or(&[]);
        let threshold_secs = Self::resolve_inactive_threshold(
            topic,
            ns_inactive_secs,
            overrides,
        );
        if threshold_secs == 0 {
            // Exempt
            continue;
        }

        // 3. Compute effective last activity
        let engine_activity = self.engine.topic_last_activity(ns, topic).await.unwrap_or(0);
        let manager_activity = self.manager.topic_last_subscribe(ns, topic).await.unwrap_or(0);
        let effective = engine_activity.max(manager_activity);

        let inactive_millis = now_millis.saturating_sub(effective);
        let threshold_millis = threshold_secs * 1000;

        // 4. Active?
        let key = (ns.clone(), topic.clone());
        if inactive_millis < threshold_millis {
            let mut grace = self.grace_pending.write().await;
            if grace.remove(&key).is_some() {
                tracing::info!(namespace = %ns, topic = %topic, "topic inactivity GC: activity resumed, removed from grace");
                metrics::gauge!("fafb_topic_reap_grace_pending", "namespace" => ns.0.clone())
                    .decrement(1.0);
            }
            continue;
        }

        // Track as exceeded threshold
        metrics::counter!("fafb_topic_inactive_threshold_exceeded", "namespace" => ns.0.clone()).increment(1);

        // 5. Safety checks
        if self.manager.has_active_subscribers(ns, topic).await {
            metrics::counter!("fafb_topic_reap_skipped_active", "namespace" => ns.0.clone()).increment(1);
            continue;
        }
        if self.manager.has_inflight_messages(ns, topic).await {
            metrics::counter!("fafb_topic_reap_skipped_active", "namespace" => ns.0.clone()).increment(1);
            continue;
        }

        let grace_period_secs = policy.topic_reap_grace_period_secs.unwrap_or(default_grace_secs);

        // 6. Enter grace?
        let mut grace = self.grace_pending.write().await;
        if !grace.contains_key(&key) {
            grace.insert(key, now_instant);
            metrics::gauge!("fafb_topic_reap_grace_pending", "namespace" => ns.0.clone())
                .increment(1.0);
            tracing::warn!(
                namespace = %ns, topic = %topic,
                inactive_secs = inactive_millis / 1000,
                threshold_secs,
                grace_period_secs,
                "topic entered reap grace period"
            );
            continue;
        }

        // 7. Grace elapsed?
        let grace_entered = grace[&key];
        if now_instant.duration_since(grace_entered).as_secs() < grace_period_secs {
            continue;
        }
        grace.remove(&key);
        drop(grace); // release lock before reap

        // REAP
        self.reap_topic(ns, topic, inactive_millis / 1000).await;
        metrics::gauge!("fafb_topic_reap_grace_pending", "namespace" => ns.0.clone())
            .decrement(1.0);
    }
}
```

- [ ] **Step 3: Add `resolve_inactive_threshold` helper**

```rust
fn resolve_inactive_threshold(
    topic: &TopicName,
    default_secs: u64,
    overrides: &[crate::config::CompiledInactiveOverride],
) -> u64 {
    for ov in overrides {
        if ov.pattern.matches(&topic.0) {
            return ov.inactive_secs;
        }
    }
    default_secs
}
```

Patterns are pre-compiled at config load time (Task 3, Step 6) and carried in `RuntimeConfig::compiled_inactive_overrides`. No parsing at sweep time.

- [ ] **Step 4: Add `reap_topic` method**

```rust
async fn reap_topic(&self, ns: &NamespaceId, topic: &TopicName, inactive_secs: u64) {
    // 1-2. Teardown delivery state (aborts tasks, removes TopicDeliveryState)
    if let Err(e) = self.manager.teardown_topic(ns, topic).await {
        tracing::warn!(namespace = %ns, topic = %topic, error = %e, "reap: teardown_topic failed");
        return;
    }

    // 3. Close storage writer
    if let Err(e) = self.storage.close_topic_writer(ns, topic).await {
        tracing::warn!(namespace = %ns, topic = %topic, error = %e, "reap: close_topic_writer failed");
    }

    // 4. Remove from engine (returns bytes/messages freed)
    let (bytes_freed, messages_freed) = match self.engine.remove_topic(ns, topic).await {
        Ok(result) => result,
        Err(e) => {
            tracing::warn!(namespace = %ns, topic = %topic, error = %e, "reap: remove_topic failed");
            (0, 0)
        }
    };

    // 5. Delete segment directory
    if let Err(e) = self.storage.delete_topic_dir(ns, topic).await {
        tracing::warn!(namespace = %ns, topic = %topic, error = %e, "reap: delete_topic_dir failed");
        return;
    }

    tracing::warn!(
        namespace = %ns, topic = %topic,
        inactive_days = inactive_secs / 86400,
        bytes_freed, messages_freed,
        "topic reaped"
    );
    metrics::counter!("fafb_topic_reaped_total",
        "namespace" => ns.0.clone(), "topic" => topic.0.clone()
    ).increment(1);
}
```

- [ ] **Step 5: Wire into `sweep()`**

At the end of the existing `sweep()` method, after the segment retention loop and elapsed time recording, add:

```rust
// --- Topic inactivity GC ---
self.sweep_inactive_topics(&all_topics).await;
```

`sweep()` keeps `&self` — `grace_pending` is `Arc<RwLock<>>` so interior mutability is handled by the lock.

- [ ] **Step 6: Run existing retention tests**

Run: `cargo test --lib retention`
Expected: All existing tests still PASS

- [ ] **Step 7: Commit**

```bash
git add src/retention.rs src/main.rs
git commit -m "feat(gc): implement topic inactivity sweep in RetentionTask with grace period"
```

---

### Task 8: Admin API — `GET /admin/topics/pending-reap`

**Files:**
- Modify: `src/server/http.rs:228-237` (route registration)
- Modify: `src/server/http/routes.rs` (handler function)
- Modify: `src/retention.rs` (expose grace_pending for reading)

- [ ] **Step 1: Wire `grace_pending` into `AppState`**

`grace_pending` is already `Arc<RwLock<GracePendingMap>>` from Task 7. The `RetentionTask::grace_pending()` accessor returns a clone of the handle.

- [ ] **Step 2: Pass `grace_pending` to `AppState`**

In `src/server/http.rs`, add the `grace_pending` handle to `AppState`:

```rust
pub grace_pending: Arc<tokio::sync::RwLock<GracePendingMap>>,
```

Wire it from `main.rs` where the `RetentionTask` is created, passing `retention_task.grace_pending()` into `AppState`.

- [ ] **Step 3: Write the handler**

In `src/server/http/routes.rs`, add:

```rust
#[derive(serde::Serialize)]
struct PendingReapEntry {
    namespace: String,
    topic: String,
    grace_entered_secs_ago: u64,
    grace_remaining_secs: i64,  // negative if overdue (waiting for sweep)
}

pub async fn admin_pending_reap<M: NamespaceManager>(
    State(state): State<Arc<AppState<M>>>,
) -> axum::Json<Vec<PendingReapEntry>> {
    let map = state.grace_pending.read().await;
    let now = std::time::Instant::now();

    // Load config for grace period values
    let cfg = state.shared_config.load();

    let entries: Vec<PendingReapEntry> = map
        .iter()
        .map(|((ns, topic), entered)| {
            let elapsed_secs = now.duration_since(*entered).as_secs();
            let grace_period = cfg.namespace_retention
                .get(ns)
                .and_then(|p| p.topic_reap_grace_period_secs)
                .unwrap_or(86400);
            PendingReapEntry {
                namespace: ns.0.clone(),
                topic: topic.0.clone(),
                grace_entered_secs_ago: elapsed_secs,
                grace_remaining_secs: grace_period as i64 - elapsed_secs as i64,
            }
        })
        .collect();
    axum::Json(entries)
}
```

Note: `Instant` is a monotonic clock, so we cannot produce RFC 3339 wall-clock timestamps for `grace_entered`. The `grace_entered_secs_ago` and `grace_remaining_secs` fields give operators the information they need to act. The spec's `last_activity` timestamp is not included here to avoid additional state in the grace map — operators can check topic info for last activity details.

- [ ] **Step 4: Register the route**

In `src/server/http.rs`, add to `admin_routes`:

```rust
.route("/topics/pending-reap", get(admin_pending_reap::<M>))
```

Update the `use routes::{ ... }` import to include `admin_pending_reap`.

- [ ] **Step 5: Run the build**

Run: `cargo build`
Expected: Compiles successfully

- [ ] **Step 6: Commit**

```bash
git add src/retention.rs src/server/http.rs src/server/http/routes.rs src/main.rs
git commit -m "feat(gc): add GET /admin/topics/pending-reap endpoint"
```

---

### Task 9: Integration tests

**Files:**
- Create: `tests/topic_gc_integration.rs`

- [ ] **Step 1: Write basic reap lifecycle test**

```rust
//! Integration tests for topic inactivity GC.

mod common;

use common::{TestServer, TestClient};
use fafb_q::types::{TopicName, TopicOffset};

/// Publish to a topic, wait past the inactivity threshold + grace period,
/// verify the topic is fully removed from memory and disk.
#[tokio::test]
async fn topic_reaped_after_inactivity_and_grace() {
    // Configure with very short thresholds for testing:
    // topic_inactive_secs = 2, topic_reap_grace_period_secs = 1
    // retention_check_interval_secs = 1
    let server = TestServer::builder()
        .with_retention(|r| {
            r.topic_inactive_secs = Some(2);
            r.topic_reap_grace_period_secs = Some(1);
        })
        .retention_check_interval_secs(1)
        .build()
        .await;

    let mut client = TestClient::connect_and_auth(
        server.addr(),
        "default",
        server.api_key(),
    ).await;

    // Publish a message
    client.publish(TopicName("ephemeral".into()), b"data", vec![], 0).await.unwrap();

    // Topic should exist
    let info = client.topic_info(TopicName("ephemeral".into())).await;
    assert!(info.is_ok());

    // Wait for inactivity threshold (2s) + grace (1s) + sweep interval margin (2s)
    tokio::time::sleep(std::time::Duration::from_secs(6)).await;

    // Topic should be gone — info should return not found
    let info = client.topic_info(TopicName("ephemeral".into())).await;
    assert!(info.is_err());
}
```

Note: The exact `TestServer` builder API may differ. Adapt to match the existing `tests/common/mod.rs` patterns. Check how other integration tests configure retention and intervals.

- [ ] **Step 2: Write publish-resets-clock test**

```rust
#[tokio::test]
async fn publish_resets_inactivity_clock() {
    let server = TestServer::builder()
        .with_retention(|r| {
            r.topic_inactive_secs = Some(3);
            r.topic_reap_grace_period_secs = Some(1);
        })
        .retention_check_interval_secs(1)
        .build()
        .await;

    let mut client = TestClient::connect_and_auth(
        server.addr(),
        "default",
        server.api_key(),
    ).await;

    client.publish(TopicName("kept-alive".into()), b"msg1", vec![], 0).await.unwrap();

    // Wait 2s (close to threshold but not past)
    tokio::time::sleep(std::time::Duration::from_secs(2)).await;

    // Publish again — resets clock
    client.publish(TopicName("kept-alive".into()), b"msg2", vec![], 0).await.unwrap();

    // Wait another 2s + margin — would have been reaped without the second publish
    tokio::time::sleep(std::time::Duration::from_secs(3)).await;

    // Topic should still exist
    let info = client.topic_info(TopicName("kept-alive".into())).await;
    assert!(info.is_ok());
}
```

- [ ] **Step 3: Write active-subscriber-blocks-reap test**

```rust
#[tokio::test]
async fn active_subscriber_blocks_reap() {
    let server = TestServer::builder()
        .with_retention(|r| {
            r.topic_inactive_secs = Some(2);
            r.topic_reap_grace_period_secs = Some(1);
        })
        .retention_check_interval_secs(1)
        .build()
        .await;

    let mut pub_client = TestClient::connect_and_auth(
        server.addr(),
        "default",
        server.api_key(),
    ).await;

    let mut sub_client = TestClient::connect_and_auth(
        server.addr(),
        "default",
        server.api_key(),
    ).await;

    pub_client.publish(TopicName("subscribed".into()), b"data", vec![], 0).await.unwrap();

    // Subscribe — keeps topic alive
    sub_client.subscribe_at_and_consume(TopicName("subscribed".into()), None, TopicOffset(0)).await.unwrap();

    // Wait well past threshold + grace
    tokio::time::sleep(std::time::Duration::from_secs(6)).await;

    // Topic should still exist because subscriber is connected
    let info = pub_client.topic_info(TopicName("subscribed".into())).await;
    assert!(info.is_ok());
}
```

- [ ] **Step 4: Write override-exemption test**

```rust
#[tokio::test]
async fn override_exempts_topic_from_reaping() {
    let server = TestServer::builder()
        .with_retention(|r| {
            r.topic_inactive_secs = Some(2);
            r.topic_reap_grace_period_secs = Some(1);
            r.topic_inactive_overrides = vec![
                fafb_q::config::TopicInactiveOverride {
                    pattern: "system.>".into(),
                    inactive_secs: 0,
                },
            ];
        })
        .retention_check_interval_secs(1)
        .build()
        .await;

    let mut client = TestClient::connect_and_auth(
        server.addr(),
        "default",
        server.api_key(),
    ).await;

    client.publish(TopicName("system.logs".into()), b"data", vec![], 0).await.unwrap();

    // Wait well past threshold + grace
    tokio::time::sleep(std::time::Duration::from_secs(6)).await;

    // Topic should still exist — exempted by override
    let info = client.topic_info(TopicName("system.logs".into())).await;
    assert!(info.is_ok());
}
```

- [ ] **Step 5: Run all integration tests**

Run: `cargo test --test topic_gc_integration -- --nocapture`
Expected: All tests PASS

- [ ] **Step 6: Commit**

```bash
git add tests/topic_gc_integration.rs
git commit -m "test: add integration tests for topic inactivity GC"
```

---

### Task 10: Documentation updates

**Files:**
- Modify: `README.md` (if it documents retention config)
- Modify: `config/config.example.toml` (already done in Task 3)
- Modify: `CLAUDE.md` (add to Gotchas/Invariants and Key Design Decisions)

- [ ] **Step 1: Update CLAUDE.md Gotchas**

Add to the Gotchas and Invariants section:

```markdown
- **Topic inactivity GC has a grace period** — topics are not deleted immediately when they exceed `topic_inactive_secs`. They enter a grace period (default 24h) first. Grace period state is in-memory only — lost on restart, requiring full re-qualification.
- **Active subscribers block topic reaping** — a topic with any connected subscriber or inflight messages is never reaped, regardless of activity timestamps.
- **`teardown_topic()` does NOT clean up wildcard subscriptions** — only `TopicDeliveryState` is removed. Wildcard entries in `subscription_trie` / `wildcard_subs` survive so auto-subscribe fires if the topic is re-created.
```

- [ ] **Step 2: Update CLAUDE.md Key Design Decisions**

Add a section:

```markdown
**Topic inactivity GC** — topics auto-created on first publish can be automatically reaped after a configurable inactivity window. Activity = last publish OR last subscribe/fetch (whichever is newest). Info queries do not count. Config is per-namespace with NATS-style wildcard pattern overrides (`topic_inactive_overrides`). Opt-in only (`topic_inactive_secs: None` by default). Soft-delete with 24h grace period before hard delete. Grace period state is in-memory only — on restart, topics re-qualify through the full inactivity window.
```

- [ ] **Step 3: Update README.md if needed**

Check if README.md has a retention configuration section. If so, add a brief mention of `topic_inactive_secs`.

- [ ] **Step 4: Commit**

```bash
git add CLAUDE.md README.md
git commit -m "docs: document topic inactivity GC in CLAUDE.md and README"
```
