---
phase: 12-redis-backend
plan: 01
type: execute
wave: 1
depends_on: []
files_modified:
  - crates/pmcp-tasks/Cargo.toml
  - crates/pmcp-tasks/src/store/mod.rs
  - crates/pmcp-tasks/src/store/redis.rs
  - crates/pmcp-tasks/src/lib.rs
autonomous: true
requirements: [RDIS-01, RDIS-02, RDIS-03, RDIS-04, RDIS-05]

must_haves:
  truths:
    - "RedisBackend compiles behind the redis feature flag without affecting default builds"
    - "Tasks stored in Redis use hash-based storage with version, data, and expires_at as separate hash fields"
    - "All write operations (put, put_if_version, delete) are atomic via Lua scripts that update hash + sorted set + TTL in a single round-trip"
    - "Expired tasks are filtered out during get and list_by_prefix via application-level expires_at check"
    - "Owner-scoped listing uses per-owner sorted set indexes"
    - "cleanup_expired is a no-op returning Ok(0)"
  artifacts:
    - path: "crates/pmcp-tasks/src/store/redis.rs"
      provides: "RedisBackend struct implementing StorageBackend with all 6 methods"
      min_lines: 200
    - path: "crates/pmcp-tasks/Cargo.toml"
      provides: "redis and redis-tests feature flags with optional redis dependency"
      contains: "redis"
    - path: "crates/pmcp-tasks/src/store/mod.rs"
      provides: "Conditional redis module declaration"
      contains: "pub mod redis"
    - path: "crates/pmcp-tasks/src/lib.rs"
      provides: "Conditional RedisBackend re-export"
      contains: "RedisBackend"
  key_links:
    - from: "crates/pmcp-tasks/src/store/redis.rs"
      to: "crates/pmcp-tasks/src/store/backend.rs"
      via: "impl StorageBackend for RedisBackend"
      pattern: "impl StorageBackend for RedisBackend"
    - from: "crates/pmcp-tasks/src/store/mod.rs"
      to: "crates/pmcp-tasks/src/store/redis.rs"
      via: "feature-gated module declaration"
      pattern: "cfg.*feature.*redis.*pub mod redis"
    - from: "crates/pmcp-tasks/src/lib.rs"
      to: "crates/pmcp-tasks/src/store/redis.rs"
      via: "conditional re-export"
      pattern: "cfg.*feature.*redis.*RedisBackend"
---

<objective>
Implement RedisBackend struct with all 6 StorageBackend trait methods behind a `redis` feature flag, using the `redis` crate with async/tokio support, Lua scripts for atomic writes, hash-based storage, and sorted set indexing.

Purpose: Prove the StorageBackend trait generalizes beyond DynamoDB to a fundamentally different storage system (Redis). Developers can persist tasks in Redis for long-running server deployments by enabling the `redis` feature flag.

Output: A complete, compilable `RedisBackend` in `crates/pmcp-tasks/src/store/redis.rs` with feature flag infrastructure in Cargo.toml, conditional module declaration in store/mod.rs, and conditional re-export in lib.rs.
</objective>

<execution_context>
@./.claude/get-shit-done/workflows/execute-plan.md
@./.claude/get-shit-done/templates/summary.md
</execution_context>

<context>
@.planning/PROJECT.md
@.planning/ROADMAP.md
@.planning/STATE.md
@.planning/phases/12-redis-backend/12-CONTEXT.md
@.planning/phases/12-redis-backend/12-RESEARCH.md
@.planning/phases/11-dynamodb-backend/11-01-SUMMARY.md

# Reference implementation (follow same structural patterns):
@crates/pmcp-tasks/src/store/dynamodb.rs
@crates/pmcp-tasks/src/store/backend.rs
@crates/pmcp-tasks/src/store/mod.rs
@crates/pmcp-tasks/src/lib.rs
@crates/pmcp-tasks/Cargo.toml
</context>

<tasks>

<task type="auto">
  <name>Task 1: Feature flag setup and RedisBackend struct with all 6 StorageBackend methods</name>
  <files>
    crates/pmcp-tasks/Cargo.toml
    crates/pmcp-tasks/src/store/redis.rs
    crates/pmcp-tasks/src/store/mod.rs
    crates/pmcp-tasks/src/lib.rs
  </files>
  <action>
**Cargo.toml changes:**

Add the redis dependency and feature flags. In `[dependencies]`:
```toml
redis = { version = "1.0", features = ["tokio-comp", "script"], optional = true }
```

In `[features]`:
```toml
redis = ["dep:redis"]
redis-tests = ["redis"]
```

**store/mod.rs changes:**

Add conditional module declaration (next to dynamodb):
```rust
#[cfg(feature = "redis")]
pub mod redis;
```

Update module docs to mention Redis backend alongside DynamoDB.

**lib.rs changes:**

Add conditional re-export (next to DynamoDbBackend):
```rust
#[cfg(feature = "redis")]
pub use store::redis::RedisBackend;
```

**store/redis.rs -- complete implementation:**

Follow the DynamoDB backend's structural patterns. The module should contain:

1. **Module-level docs** with key schema table, usage example (no_run), and relationship to GenericTaskStore explanation.

2. **RedisBackend struct** holding `MultiplexedConnection` (Clone) and `key_prefix: String`:
   - `pub async fn new(url: &str) -> Result<Self, StorageError>` -- connect via `redis::Client::open(url)` then `get_multiplexed_async_connection()`. Fail-fast on connection error, returning StorageError::Backend. Default prefix "pmcp".
   - `pub fn with_connection(conn: MultiplexedConnection) -> Self` -- for pre-built connections.
   - `pub fn with_prefix(mut self, prefix: impl Into<String>) -> Self` -- builder for test isolation.

3. **Private helpers:**
   - `fn task_key(&self, owner_id: &str, task_id: &str) -> String` -- returns `{prefix}:tasks:{owner_id}:{task_id}`
   - `fn index_key(&self, owner_id: &str) -> String` -- returns `{prefix}:idx:{owner_id}`
   - `fn split_key(key: &str) -> Result<(&str, &str), StorageError>` -- splits composite `{owner_id}:{task_id}` on first colon
   - `fn split_prefix(prefix: &str) -> Result<&str, StorageError>` -- strips trailing colon from `{owner_id}:`
   - `fn extract_ttl_epoch(data: &[u8]) -> Option<i64>` -- parses expiresAt from JSON, converts to epoch seconds (reuse DynamoDB's logic)
   - `fn extract_created_at_ms(data: &[u8]) -> Option<i64>` -- parses task.createdAt from JSON, converts to epoch milliseconds for sorted set score
   - `fn map_redis_error(err: ::redis::RedisError, key: &str) -> StorageError` -- maps redis errors to StorageError::Backend
   - `fn is_expired(fields: &std::collections::HashMap<String, String>) -> bool` -- checks expires_at field against current time

4. **Three Lua script constants** (const &str):

   **LUA_PUT** -- Unconditional put (hash + index + TTL):
   - KEYS[1] = task hash key, KEYS[2] = owner index sorted set key
   - ARGV[1] = data JSON, ARGV[2] = expires_at epoch (or "" if no TTL), ARGV[3] = task_id, ARGV[4] = creation timestamp score
   - Read current version from HGET, increment (or start at 1)
   - HSET version + data fields
   - If expires_at: HSET expires_at + EXPIREAT; else: HDEL expires_at + PERSIST
   - ZADD NX to sorted set index (preserves original creation timestamp)
   - Return new_version

   **LUA_PUT_IF_VERSION** -- CAS put:
   - KEYS[1] = task hash key, KEYS[2] = owner index sorted set key
   - ARGV[1] = data, ARGV[2] = expected_version, ARGV[3] = expires_at, ARGV[4] = task_id, ARGV[5] = creation timestamp
   - Check HGET version: if nil return {-1, 0}; if mismatch return {0, actual}
   - On match: increment, HSET, TTL handling, ZADD NX, return {1, new_version}

   **LUA_DELETE** -- Delete hash + remove from index:
   - KEYS[1] = task hash key, KEYS[2] = owner index sorted set key
   - ARGV[1] = task_id
   - EXISTS check, DEL hash, ZREM from index, return 1 if existed else 0

5. **StorageBackend impl for RedisBackend** -- all 6 methods:

   **get:** HGETALL on task_key. If empty HashMap -> NotFound. Application-level expiry check (is_expired). Parse version (u64) and data fields. Return VersionedRecord.

   **put:** Split key, compute task_key + index_key. UTF-8 validate data. Extract TTL epoch and created_at_ms from JSON. Invoke LUA_PUT via `redis::Script::new(LUA_PUT)` with .key().key().arg().arg().arg().arg().invoke_async(). Return new_version.

   **put_if_version:** Split key, compute keys. UTF-8 validate. Extract TTL + created_at_ms. Invoke LUA_PUT_IF_VERSION. Parse return tuple (i64, i64). Match: 1 -> Ok(new_version), 0 -> VersionConflict, -1 -> VersionConflict (per DynamoDB pattern: missing key with CAS returns VersionConflict).

   **delete:** Split key, compute keys. Invoke LUA_DELETE. Return `result == 1`.

   **list_by_prefix:** Split prefix to get owner_id. ZRANGE on index_key (0 to -1) to get all task_ids. For each task_id: HGETALL on task_key. Skip orphans (empty hash -- lazily ZREM from index). Skip expired (application-level filter). Collect (composite_key, VersionedRecord). After loop, batch ZREM orphaned entries. Return results.

   **cleanup_expired:** No-op, return Ok(0). Redis EXPIRE/EXPIREAT handles cleanup automatically.

**IMPORTANT implementation notes:**
- Use `::redis::` (absolute path) for redis crate imports to avoid module name collision (the module is named `redis` and the crate is also `redis`).
- Clone `self.conn` for each operation -- MultiplexedConnection is designed for cheap cloning, sharing the same TCP connection.
- The `redis::Script` type handles EVALSHA/EVAL fallback automatically.
- For put_if_version on missing key: return VersionConflict (not NotFound), consistent with DynamoDB backend behavior per Phase 11 decisions.
- For list_by_prefix version parsing: use `.unwrap_or(0)` for robustness (should not happen but defensive).
- For list_by_prefix data: use `.cloned().unwrap_or_default()` for robustness.
- EXPIREAT uses absolute epoch seconds (not relative EXPIRE), since expires_at is already an absolute timestamp.
- Comprehensive rustdoc on all public items with `no_run` examples.
  </action>
  <verify>
Run these commands to verify:
1. `cargo check -p pmcp-tasks --features redis` -- compiles with redis feature
2. `cargo check -p pmcp-tasks` -- compiles without redis feature (no contamination)
3. `cargo check -p pmcp-tasks --features dynamodb,redis` -- compiles with both backends
4. `cargo clippy -p pmcp-tasks --features redis -- -D warnings` -- zero clippy warnings
5. `cargo fmt -p pmcp-tasks --check` -- formatting is clean
6. `cargo test -p pmcp-tasks` -- all existing 561+ tests still pass
  </verify>
  <done>
RedisBackend struct exists in `crates/pmcp-tasks/src/store/redis.rs` implementing all 6 StorageBackend methods with Lua scripts. Feature flags `redis` and `redis-tests` are defined. Conditional module declaration in store/mod.rs and re-export in lib.rs. Compiles cleanly with and without the redis feature. Zero clippy warnings. All existing tests pass unchanged.
  </done>
</task>

</tasks>

<verification>
1. `cargo check -p pmcp-tasks --features redis` passes (redis backend compiles)
2. `cargo check -p pmcp-tasks` passes (default build unaffected)
3. `cargo check -p pmcp-tasks --features dynamodb,redis` passes (both backends compile together)
4. `cargo clippy -p pmcp-tasks --features redis -- -D warnings` passes (zero warnings)
5. `cargo fmt -p pmcp-tasks --check` passes (clean formatting)
6. `cargo test -p pmcp-tasks` passes (no regressions in existing 561+ tests)
7. `grep -r "impl StorageBackend for RedisBackend" crates/pmcp-tasks/` finds the trait implementation
8. `grep -r "LUA_PUT\|LUA_PUT_IF_VERSION\|LUA_DELETE" crates/pmcp-tasks/src/store/redis.rs` finds all 3 Lua script constants
</verification>

<success_criteria>
- RedisBackend implements all 6 StorageBackend methods behind the `redis` feature flag
- Hash-based storage with version, data, expires_at as separate hash fields
- All write operations use Lua scripts for atomicity (hash + sorted set + TTL in single round-trip)
- Per-owner sorted set indexes for list_by_prefix
- Application-level expiry filtering in get and list_by_prefix
- cleanup_expired is a no-op returning Ok(0)
- Compiles with `--features redis`, without redis, and with `--features dynamodb,redis`
- Zero clippy warnings, clean formatting
- All existing tests pass without modification
</success_criteria>

<output>
After completion, create `.planning/phases/12-redis-backend/12-01-SUMMARY.md`
</output>
