---
phase: 02-in-memory-backend-and-owner-security
plan: 01
type: execute
wave: 1
depends_on: []
files_modified:
  - crates/pmcp-tasks/src/store/memory.rs
  - crates/pmcp-tasks/src/store/mod.rs
  - crates/pmcp-tasks/src/security.rs
  - crates/pmcp-tasks/src/lib.rs
  - crates/pmcp-tasks/Cargo.toml
autonomous: true
requirements:
  - STOR-05
  - STOR-06
  - STOR-07
  - HNDL-02
  - HNDL-03
  - SEC-01
  - SEC-02
  - SEC-03
  - SEC-04
  - SEC-05
  - SEC-06
  - SEC-07
  - SEC-08

must_haves:
  truths:
    - "InMemoryTaskStore implements all 11 TaskStore trait methods and compiles"
    - "Creating a task with owner A and retrieving with owner B returns NotFound (never OwnerMismatch)"
    - "Creating more tasks than max_tasks_per_owner returns ResourceExhausted error"
    - "Requesting a TTL above max_ttl_ms returns an error (not silently clamped)"
    - "Anonymous access (empty/local owner) is rejected when allow_anonymous is false"
    - "State transitions are validated atomically within the DashMap entry"
    - "Variable merge applies null-deletion semantics and checks size after merge"
    - "resolve_owner_id returns OAuth subject > client_id > session_id > 'local' in priority order"
  artifacts:
    - path: "crates/pmcp-tasks/src/store/memory.rs"
      provides: "InMemoryTaskStore with DashMap storage and full TaskStore implementation"
      contains: "impl TaskStore for InMemoryTaskStore"
    - path: "crates/pmcp-tasks/src/security.rs"
      provides: "TaskSecurityConfig struct and resolve_owner_id function"
      contains: "TaskSecurityConfig"
    - path: "crates/pmcp-tasks/src/store/mod.rs"
      provides: "Module organization: trait in mod.rs, memory backend as submodule"
      contains: "pub mod memory"
  key_links:
    - from: "crates/pmcp-tasks/src/store/memory.rs"
      to: "crates/pmcp-tasks/src/store/mod.rs"
      via: "implements TaskStore trait"
      pattern: "impl TaskStore for InMemoryTaskStore"
    - from: "crates/pmcp-tasks/src/store/memory.rs"
      to: "crates/pmcp-tasks/src/security.rs"
      via: "enforces TaskSecurityConfig limits in create()"
      pattern: "self\\.security"
    - from: "crates/pmcp-tasks/src/store/memory.rs"
      to: "crates/pmcp-tasks/src/error.rs"
      via: "returns TaskError variants including NotFound for owner mismatch"
      pattern: "TaskError::NotFound"
---

<objective>
Implement InMemoryTaskStore, TaskSecurityConfig, and owner resolution -- the core storage backend with security enforcement.

Purpose: This is the foundational deliverable of Phase 2. Every subsequent plan (TaskContext, tests) depends on a working in-memory store with owner isolation and security limits enforced at the store level.

Output: Three new source files (store/memory.rs, security.rs, store/mod.rs restructured) implementing the full TaskStore trait with DashMap-backed concurrent storage, structural owner isolation, configurable security limits, and an owner resolution function for Phase 3 integration.
</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/02-in-memory-backend-and-owner-security/02-CONTEXT.md
@.planning/phases/02-in-memory-backend-and-owner-security/02-RESEARCH.md

Source files to read before implementing:
@crates/pmcp-tasks/src/store.rs (current trait + supporting types -- will become store/mod.rs)
@crates/pmcp-tasks/src/error.rs (TaskError variants to return)
@crates/pmcp-tasks/src/domain/record.rs (TaskRecord, is_expired(), to_wire_task_with_variables())
@crates/pmcp-tasks/src/types/task.rs (TaskStatus, validate_transition())
@crates/pmcp-tasks/src/lib.rs (current re-exports to extend)
@crates/pmcp-tasks/Cargo.toml (dependencies -- dashmap needs adding)
@src/server/auth/traits.rs (AuthContext struct fields for resolve_owner_id)
</context>

<tasks>

<task type="auto">
  <name>Task 1: Create TaskSecurityConfig and owner resolution in security.rs</name>
  <files>crates/pmcp-tasks/src/security.rs, crates/pmcp-tasks/src/lib.rs, crates/pmcp-tasks/Cargo.toml</files>
  <action>
Create `crates/pmcp-tasks/src/security.rs` with:

1. **TaskSecurityConfig struct** (derive Debug, Clone):
   - `max_tasks_per_owner: usize` (default: 100) per SEC-04
   - `allow_anonymous: bool` (default: false) per SEC-07
   - Implement `Default` with these values
   - Builder methods: `with_max_tasks_per_owner(usize)`, `with_allow_anonymous(bool)`
   - Include comprehensive rustdoc with examples

   Note: TTL limits (default_ttl_ms, max_ttl_ms) already exist in `StoreConfig` from Phase 1. Do NOT duplicate them. TaskSecurityConfig handles owner-specific security. StoreConfig handles storage-level limits including TTL (per SEC-05, SEC-06 which are enforced via StoreConfig).

2. **DEFAULT_LOCAL_OWNER constant**: `pub const DEFAULT_LOCAL_OWNER: &str = "local";`
   - Used for anonymous/local single-user servers per CONTEXT.md locked decision

3. **resolve_owner_id function** per SEC-01:
   ```rust
   pub fn resolve_owner_id(
       auth_subject: Option<&str>,
       client_id: Option<&str>,
       session_id: Option<&str>,
   ) -> String
   ```
   - Priority: auth_subject (non-empty) > client_id (non-empty) > session_id (non-empty) > DEFAULT_LOCAL_OWNER
   - Takes `&str` slices instead of `AuthContext` directly to avoid coupling pmcp-tasks to the pmcp crate's auth types. Phase 3 middleware will call `resolve_owner_id(Some(&auth.subject), auth.client_id.as_deref(), session_id)`.
   - Empty strings are treated as absent (skip to next priority)
   - Include unit tests in `#[cfg(test)] mod tests` block covering all priority fallback paths

4. **Update `crates/pmcp-tasks/src/lib.rs`**: Add `pub mod security;` and re-export `TaskSecurityConfig`, `DEFAULT_LOCAL_OWNER`, `resolve_owner_id` from the crate root.

5. **Update `crates/pmcp-tasks/Cargo.toml`**: Add `dashmap = "6.1"` to `[dependencies]` (needed by Task 2).

Verify the module compiles: `cargo check --package pmcp-tasks`
  </action>
  <verify>
`cargo check --package pmcp-tasks` succeeds with zero errors.
`cargo test --package pmcp-tasks -- security` runs the resolve_owner_id unit tests and they pass.
  </verify>
  <done>
TaskSecurityConfig has max_tasks_per_owner (default 100) and allow_anonymous (default false) with builder methods. resolve_owner_id correctly falls through the priority chain. DEFAULT_LOCAL_OWNER is "local". All re-exported from crate root. dashmap added to Cargo.toml.
  </done>
</task>

<task type="auto">
  <name>Task 2: Implement InMemoryTaskStore with full TaskStore trait and security enforcement</name>
  <files>crates/pmcp-tasks/src/store/memory.rs, crates/pmcp-tasks/src/store/mod.rs, crates/pmcp-tasks/src/lib.rs</files>
  <action>
**Restructure store module:** Convert `crates/pmcp-tasks/src/store.rs` into a directory module:
1. Rename `store.rs` to `store/mod.rs` (keep ALL existing content -- StoreConfig, ListTasksOptions, TaskPage, TaskStore trait, tests)
2. Add `pub mod memory;` to the top of `store/mod.rs`
3. Update `lib.rs` re-exports to also export `InMemoryTaskStore` from `store::memory`

**Create `crates/pmcp-tasks/src/store/memory.rs`** implementing `InMemoryTaskStore`:

Struct fields:
```rust
pub struct InMemoryTaskStore {
    tasks: DashMap<String, TaskRecord>,
    config: StoreConfig,
    security: TaskSecurityConfig,
    default_poll_interval: u64,
}
```

Constructor and builder:
- `pub fn new() -> Self` with defaults (StoreConfig::default(), TaskSecurityConfig::default(), poll_interval 5000)
- `pub fn with_config(mut self, config: StoreConfig) -> Self`
- `pub fn with_security(mut self, security: TaskSecurityConfig) -> Self`
- `pub fn with_poll_interval(mut self, ms: u64) -> Self`
- Implement `Default` delegating to `new()`

**Implement all 11 TaskStore methods** (use `#[async_trait]`):

For **every method that takes owner_id** (get, update_status, set_variables, set_result, get_result, cancel, complete_with_result):
- After DashMap lookup, check `record.owner_id != owner_id` and return `TaskError::NotFound` (NEVER `OwnerMismatch`) per locked decision. Use `tracing::warn!` for internal logging of the actual mismatch.

**create():**
- Check `security.allow_anonymous`: if false and `owner_id == DEFAULT_LOCAL_OWNER` or `owner_id.is_empty()`, return error
- Count tasks for this owner via `self.tasks.iter().filter(|e| e.value().owner_id == owner_id).count()` and check against `security.max_tasks_per_owner`. If exceeded, return `TaskError::ResourceExhausted { suggested_action: Some("Cancel or wait for existing tasks to expire") }`
- Check TTL: if `ttl` is provided and `config.max_ttl_ms` is set and `ttl > max_ttl_ms`, return `TaskError::StoreError("TTL exceeds maximum allowed".to_string())` -- hard reject, no clamping per locked decision
- Apply default TTL: if `ttl` is None and `config.default_ttl_ms` is Some, use default per SEC-06
- Create `TaskRecord::new(owner_id, request_method, effective_ttl)`
- Set `record.task.poll_interval = Some(self.default_poll_interval)`
- Insert into DashMap and return the record

**get():**
- Lookup by task_id, check owner, return record even if expired per locked decision (expired tasks readable, `is_expired()` flag available to caller). Only mutation methods reject expired tasks.

**update_status():**
- Use `self.tasks.get_mut(task_id)` for atomic mutable access
- Check owner, check `is_expired()` (reject with `TaskError::Expired`), validate transition via `record.task.status.validate_transition(task_id, &new_status)?`
- Apply: set status, status_message, update last_updated_at with `chrono::Utc::now().to_rfc3339_opts(chrono::SecondsFormat::Millis, true)`
- Return `record.clone()`

**set_variables():**
- Use `get_mut`, check owner, check expired
- Clone variables first, perform merge on clone (null deletes, non-null upserts), check size via `serde_json::to_vec(&merged)?.len()` against `config.max_variable_size_bytes`, then commit merged result to record. This avoids needing rollback per Research pitfall 6.
- Update last_updated_at, return `record.clone()`

**set_result():**
- get_mut, check owner, check expired, set `record.result = Some(result)`, update last_updated_at

**get_result():**
- get (read-only), check owner, check if terminal status (if not, return `TaskError::NotReady`), return `record.result.clone().ok_or(TaskError::NotReady)`

**complete_with_result():**
- get_mut, check owner, check expired, validate transition, then atomically: set status + status_message + result + last_updated_at. Clone and return. Atomicity guaranteed because DashMap get_mut holds the entry lock.

**list():**
- Iterate DashMap, filter by owner_id, clone into Vec, sort by created_at descending
- Apply cursor-based pagination: find cursor position, skip to it, take limit (default 50)
- Compute next_cursor from last item if more results exist

**cancel():**
- Delegate to `self.update_status(task_id, owner_id, TaskStatus::Cancelled, None).await`

**cleanup_expired():**
- Use `self.tasks.retain(|_, record| !record.is_expired())` -- DashMap retain removes entries atomically
- Return count of removed entries (compute by comparing len before and after)

**config():**
- Return `&self.config`

Important implementation notes:
- Do NOT hold DashMap guards (Ref, RefMut) across `.await` points. Since all operations are synchronous (DashMap + CPU-only), this should not be an issue, but be aware.
- Use `use dashmap::DashMap;` and import TaskSecurityConfig from `crate::security`.
- Include comprehensive rustdoc on the struct and each method.
  </action>
  <verify>
`cargo check --package pmcp-tasks` succeeds.
`cargo test --package pmcp-tasks` passes all existing Phase 1 tests (no regressions).
`cargo clippy --package pmcp-tasks -- -D warnings` reports zero warnings.
  </verify>
  <done>
InMemoryTaskStore implements all 11 TaskStore trait methods with DashMap storage. Owner isolation returns NotFound on mismatch. Max tasks per owner enforced with hard reject. TTL enforcement rejects above max (no clamping). Anonymous access controlled by allow_anonymous. Expired tasks readable on get() but rejected on mutations. Variable merge uses clone-check-commit pattern. State transitions validated atomically within DashMap entry lock.
  </done>
</task>

</tasks>

<verification>
```bash
# Full compilation check
cargo check --package pmcp-tasks

# No regressions on Phase 1 tests
cargo test --package pmcp-tasks

# Zero clippy warnings
cargo clippy --package pmcp-tasks -- -D warnings

# Verify new types are exported
cargo doc --package pmcp-tasks --no-deps 2>&1 | grep -E "(InMemoryTaskStore|TaskSecurityConfig|resolve_owner_id)"
```
</verification>

<success_criteria>
1. InMemoryTaskStore compiles and implements the full TaskStore trait
2. TaskSecurityConfig provides configurable max_tasks_per_owner and allow_anonymous
3. resolve_owner_id follows the correct priority chain
4. All Phase 1 tests still pass (no regressions)
5. Zero clippy warnings
</success_criteria>

<output>
After completion, create `.planning/phases/02-in-memory-backend-and-owner-security/02-01-SUMMARY.md`
</output>
