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

must_haves:
  truths:
    - "DynamoDbBackend compiles behind the `dynamodb` feature flag"
    - "Crate compiles without `dynamodb` feature (no aws-sdk-dynamodb dependency pulled in)"
    - "DynamoDbBackend implements all 6 StorageBackend methods"
    - "put_if_version uses ConditionExpression for atomic CAS"
    - "cleanup_expired is a no-op returning Ok(0)"
    - "TTL epoch seconds extracted from data and stored as expires_at attribute only when present"
  artifacts:
    - path: "crates/pmcp-tasks/Cargo.toml"
      provides: "dynamodb and dynamodb-tests feature flags with optional aws-sdk-dynamodb and aws-config deps"
      contains: "[features]"
    - path: "crates/pmcp-tasks/src/store/dynamodb.rs"
      provides: "DynamoDbBackend struct implementing StorageBackend"
      min_lines: 150
    - path: "crates/pmcp-tasks/src/store/mod.rs"
      provides: "Conditional dynamodb module declaration"
      contains: "cfg(feature"
    - path: "crates/pmcp-tasks/src/lib.rs"
      provides: "Conditional DynamoDbBackend re-export"
      contains: "DynamoDbBackend"
  key_links:
    - from: "crates/pmcp-tasks/src/store/dynamodb.rs"
      to: "crates/pmcp-tasks/src/store/backend.rs"
      via: "impl StorageBackend for DynamoDbBackend"
      pattern: "impl StorageBackend for DynamoDbBackend"
    - from: "crates/pmcp-tasks/src/store/dynamodb.rs"
      to: "aws-sdk-dynamodb"
      via: "DynamoDB Client API calls"
      pattern: "aws_sdk_dynamodb::Client"
    - from: "crates/pmcp-tasks/src/store/mod.rs"
      to: "crates/pmcp-tasks/src/store/dynamodb.rs"
      via: "conditional module declaration"
      pattern: "cfg.*dynamodb.*mod dynamodb"
---

<objective>
Implement DynamoDbBackend as a StorageBackend behind the `dynamodb` feature flag with all 6 trait methods mapped to DynamoDB API calls.

Purpose: Enable task persistence in DynamoDB for production AWS/Lambda deployments. This is the primary production backend for the storage abstraction layer.
Output: Compilable DynamoDbBackend module with feature flag infrastructure, ready for integration testing in plan 02.
</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/11-dynamodb-backend/11-CONTEXT.md
@.planning/phases/11-dynamodb-backend/11-RESEARCH.md
@crates/pmcp-tasks/Cargo.toml
@crates/pmcp-tasks/src/store/backend.rs
@crates/pmcp-tasks/src/store/mod.rs
@crates/pmcp-tasks/src/store/memory.rs
@crates/pmcp-tasks/src/lib.rs
</context>

<tasks>

<task type="auto">
  <name>Task 1: Feature flag setup and DynamoDbBackend struct with construction</name>
  <files>crates/pmcp-tasks/Cargo.toml, crates/pmcp-tasks/src/store/dynamodb.rs, crates/pmcp-tasks/src/store/mod.rs, crates/pmcp-tasks/src/lib.rs</files>
  <action>
1. **Cargo.toml** -- Add optional dependencies and feature flags:
   - Add `aws-sdk-dynamodb = { version = "1", optional = true }` to `[dependencies]`
   - Add `aws-config = { version = "1", features = ["behavior-version-latest"], optional = true }` to `[dependencies]`
   - Add `[features]` section with:
     - `dynamodb = ["dep:aws-sdk-dynamodb", "dep:aws-config"]`
     - `dynamodb-tests = ["dynamodb"]`

2. **store/dynamodb.rs** -- Create the module with DynamoDbBackend struct and constructors:
   - `DynamoDbBackend { client: aws_sdk_dynamodb::Client, table_name: String }`
   - Derive `Debug, Clone`
   - `new(client: Client, table_name: impl Into<String>) -> Self` -- accepts pre-built client
   - `async fn from_env() -> Self` -- uses `aws_config::load_defaults(BehaviorVersion::latest()).await`, default table name `"pmcp_tasks"`
   - `async fn from_env_with_table(table_name: impl Into<String>) -> Self` -- from env with custom table name
   - Add private helper functions:
     - `fn make_pk(owner_id: &str) -> String` -- returns `format!("OWNER#{owner_id}")`
     - `fn make_sk(task_id: &str) -> String` -- returns `format!("TASK#{task_id}")`
     - `fn parse_pk(pk: &str) -> Option<&str>` -- strips `"OWNER#"` prefix
     - `fn parse_sk(sk: &str) -> Option<&str>` -- strips `"TASK#"` prefix
     - `fn split_key(key: &str) -> Result<(String, String), StorageError>` -- splits composite `{owner_id}:{task_id}` key into `(make_pk(owner_id), make_sk(task_id))`
     - `fn split_prefix(prefix: &str) -> Result<String, StorageError>` -- strips trailing `:` from `{owner_id}:`, returns `make_pk(owner_id)`
     - `fn extract_ttl_epoch(data: &[u8]) -> Option<i64>` -- parses serialized JSON, extracts `expiresAt` field, converts RFC 3339 datetime to epoch seconds via `chrono::DateTime::parse_from_rfc3339`. Returns None if no expiry (attribute omitted per locked decision -- no dummy values).
     - `fn map_sdk_error(err: impl std::fmt::Display + std::error::Error + Send + Sync + 'static, key: &str) -> StorageError` -- maps any SDK error to `StorageError::Backend { message, source }`
   - Add comprehensive module-level rustdoc explaining: single-table design, PK/SK schema, relationship to GenericTaskStore, no domain logic in backend
   - Add rustdoc to all public items (struct, constructors) with `no_run` code examples

3. **store/mod.rs** -- Add conditional module:
   - Add `#[cfg(feature = "dynamodb")] pub mod dynamodb;` after `pub mod memory;`
   - Update module-level docs to list DynamoDbBackend under Backends section

4. **lib.rs** -- Add conditional re-export:
   - Add `#[cfg(feature = "dynamodb")] pub use store::dynamodb::DynamoDbBackend;` near the other re-exports

Verify the crate compiles both with and without the feature flag.
  </action>
  <verify>
Run `cargo check -p pmcp-tasks` (no feature -- must compile without dynamodb).
Run `cargo check -p pmcp-tasks --features dynamodb` (with feature -- must compile with DynamoDbBackend struct available).
Run `cargo clippy -p pmcp-tasks --features dynamodb -- -D warnings` (zero warnings).
Run `cargo fmt -p pmcp-tasks --check` (clean formatting).
  </verify>
  <done>DynamoDbBackend struct with constructors and helpers compiles behind dynamodb feature flag. Crate compiles cleanly without the feature flag. All clippy/fmt checks pass.</done>
</task>

<task type="auto">
  <name>Task 2: Implement all 6 StorageBackend methods for DynamoDbBackend</name>
  <files>crates/pmcp-tasks/src/store/dynamodb.rs</files>
  <action>
Implement `#[async_trait] impl StorageBackend for DynamoDbBackend` with all 6 methods. Use the patterns from 11-RESEARCH.md exactly. The backend is a thin adapter -- no domain logic.

1. **`get(key) -> VersionedRecord`**:
   - `split_key(key)` to get `(pk, sk)`
   - `self.client.get_item().table_name(&self.table_name).key("PK", S(pk)).key("SK", S(sk)).send().await`
   - Map missing item to `StorageError::NotFound { key }`
   - Extract `version` from `"version"` attribute (N type, parse as u64)
   - Extract `data` from `"data"` attribute (S type, convert to bytes)
   - Return `VersionedRecord { data, version }`
   - Map SDK errors via `map_sdk_error`

2. **`put(key, data) -> u64`** (unconditional create/overwrite):
   - `split_key(key)` to get `(pk, sk)`
   - Convert data to UTF-8 string (fail with Backend error if not valid UTF-8)
   - Read current version: call `self.get(key).await` -- if NotFound, current_version=0; if other error, propagate
   - `new_version = current_version + 1`
   - Build PutItem: set PK, SK, version (N), data (S)
   - Conditionally add `expires_at` (N, epoch seconds) via `extract_ttl_epoch(data)` -- only if Some
   - Send PutItem, map errors
   - Return `new_version`

3. **`put_if_version(key, data, expected_version) -> u64`** (CAS):
   - `split_key(key)` to get `(pk, sk)`
   - Convert data to UTF-8 string
   - `new_version = expected_version + 1`
   - Build PutItem with all attributes (PK, SK, version=new_version, data)
   - Conditionally add `expires_at` via `extract_ttl_epoch(data)`
   - Set `condition_expression("#v = :expected")` with expression_attribute_names `#v -> version` and expression_attribute_values `:expected -> N(expected_version)`
   - On success: return `new_version`
   - On error: check `sdk_err.as_service_error()` then `.is_conditional_check_failed_exception()` -- if true, return `StorageError::VersionConflict { key, expected: expected_version, actual: expected_version }` (we don't know the actual version without an extra read; report expected per discretion decision in research)
   - All other errors: `map_sdk_error`

4. **`delete(key) -> bool`**:
   - `split_key(key)` to get `(pk, sk)`
   - Use `self.client.delete_item()` with `return_values(ReturnValue::AllOld)`
   - If `attributes()` returns Some with items -> true (item existed)
   - If `attributes()` returns None or empty -> false (item didn't exist)
   - Map SDK errors via `map_sdk_error`

5. **`list_by_prefix(prefix) -> Vec<(String, VersionedRecord)>`**:
   - `split_prefix(prefix)` to get PK
   - Use `self.client.query()` with `key_condition_expression("PK = :pk")` and `:pk -> S(pk_value)`
   - Handle pagination: loop with `exclusive_start_key` until `last_evaluated_key` is None/empty
   - For each item: extract PK, SK, version, data; reconstruct composite key as `format!("{owner}:{task_id}")` using `parse_pk`/`parse_sk`
   - Skip items with missing/malformed attributes (defensive)
   - Return accumulated results

6. **`cleanup_expired() -> usize`**:
   - No-op: `Ok(0)` -- DynamoDB native TTL handles cleanup automatically
   - Add doc comment explaining why this is a no-op

**Important implementation notes:**
- Use `aws_sdk_dynamodb::types::AttributeValue` (S, N variants) for all attribute construction
- Use `use aws_sdk_dynamodb::types::ReturnValue` for delete
- Number attributes MUST use `.to_string()` (DynamoDB N type is a string representation)
- Never use `Scan` -- only `Query` for list_by_prefix
- Data stored as `AttributeValue::S` (String), NOT `AttributeValue::B` (Binary) per locked decision for human-readability
- Version stored as separate `version` attribute (NOT inside the data blob, where it's `#[serde(skip)]`)
- `expires_at` attribute ONLY present when task has explicit expiry (no dummy far-future values per locked decision)
- Size cap (DYNA-05) is already enforced by GenericTaskStore via StoreConfig::max_variable_size_bytes -- the backend never sees oversized items. Add a doc comment on the struct noting this.
  </action>
  <verify>
Run `cargo check -p pmcp-tasks --features dynamodb` (compiles).
Run `cargo clippy -p pmcp-tasks --features dynamodb -- -D warnings` (zero warnings).
Run `cargo fmt -p pmcp-tasks --check` (clean formatting).
Run `cargo test -p pmcp-tasks` (all existing 561+ tests still pass -- no regressions).
Run `cargo doc -p pmcp-tasks --features dynamodb --no-deps` (docs generate without warnings).
  </verify>
  <done>DynamoDbBackend fully implements StorageBackend with all 6 methods. CAS uses ConditionExpression (DYNA-03). TTL uses expires_at epoch seconds (DYNA-04). Single-table design with PK/SK composite keys (DYNA-02). cleanup_expired is no-op. All existing tests pass. Zero clippy warnings. Clean docs.</done>
</task>

</tasks>

<verification>
1. `cargo check -p pmcp-tasks` passes (no dynamodb feature -- clean compilation)
2. `cargo check -p pmcp-tasks --features dynamodb` passes (with feature)
3. `cargo clippy -p pmcp-tasks --features dynamodb -- -D warnings` passes (zero warnings)
4. `cargo test -p pmcp-tasks` passes (all existing tests, no regression)
5. `cargo doc -p pmcp-tasks --features dynamodb --no-deps` succeeds
6. The `dynamodb.rs` file exists with `impl StorageBackend for DynamoDbBackend`
7. Cargo.toml has `dynamodb` and `dynamodb-tests` feature flags
</verification>

<success_criteria>
- DynamoDbBackend compiles and implements all 6 StorageBackend methods behind `dynamodb` feature flag
- Crate compiles without `dynamodb` feature (no AWS SDK pulled in)
- PK = OWNER#owner_id, SK = TASK#task_id composite key schema implemented
- put_if_version uses ConditionExpression for atomic CAS
- expires_at attribute set only when task has TTL (no dummy values)
- cleanup_expired returns Ok(0) (no-op for DynamoDB native TTL)
- All existing 561+ tests pass unchanged
- Zero clippy warnings, clean formatting, docs generated
</success_criteria>

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