---
phase: 05-partial-execution-engine
plan: 02
type: execute
wave: 2
depends_on: ["05-01"]
files_modified:
  - src/server/workflow/task_prompt_handler.rs
autonomous: true
requirements: [EXEC-01, EXEC-02, EXEC-03, EXEC-04]

must_haves:
  truths:
    - "When a task-aware workflow is invoked, a task is created and each resolvable step's result is accumulated in memory and batch-written to task variables at the end"
    - "Execution pauses at the first unresolvable step without failing the task — task stays Working, completed steps have results, remaining steps stay Pending"
    - "When a tool execution fails, the step is marked Failed, a PauseReason::ToolError is produced with retryable flag, and the task stays Working"
    - "When a step can't resolve params because a prior step failed, PauseReason::UnresolvedDependency is produced with blocked_step, missing_output, producing_step, suggested_tool"
    - "When all steps succeed, the task is auto-completed (Completed status) with no client action needed"
    - "If batch write to task store fails, the prompt result is returned anyway with _meta from in-memory state (graceful degradation)"
  artifacts:
    - path: "src/server/workflow/task_prompt_handler.rs"
      provides: "Active execution engine with PauseReason classification, batch write, auto-complete"
      contains: "PauseReason"
      min_lines: 200
  key_links:
    - from: "src/server/workflow/task_prompt_handler.rs"
      to: "src/server/workflow/prompt_handler.rs"
      via: "self.inner.resolve_tool_parameters(), self.inner.execute_tool_step(), etc."
      pattern: "self\\.inner\\."
    - from: "src/server/workflow/task_prompt_handler.rs"
      to: "crates/pmcp-tasks/src/types/workflow.rs"
      via: "PauseReason enum, WorkflowProgress, StepStatus, key constants"
      pattern: "(PauseReason|WorkflowProgress|StepStatus|WORKFLOW_PROGRESS_KEY|WORKFLOW_PAUSE_REASON_KEY|workflow_result_key)"
    - from: "src/server/workflow/task_prompt_handler.rs"
      to: "src/server/tasks.rs"
      via: "TaskRouter::create_workflow_task, set_task_variables, complete_workflow_task"
      pattern: "self\\.task_router\\.(create_workflow_task|set_task_variables|complete_workflow_task)"
---

<objective>
Rewrite TaskWorkflowPromptHandler.handle() from passive delegation to an active execution engine with durable progress tracking.

Purpose: This is the core of Phase 5. The handler currently delegates to the inner WorkflowPromptHandler and infers step statuses from the message trace after the fact. This plan replaces that with an active step loop that: (1) executes steps sequentially using the inner handler's pub(crate) helpers, (2) accumulates results in memory, (3) classifies failures into typed PauseReason variants, (4) batch-writes all state to the task store at the end, and (5) auto-completes the task when all steps succeed.

Output: Fully functional execution engine in task_prompt_handler.rs with comprehensive tests.
</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/05-partial-execution-engine/05-CONTEXT.md
@.planning/phases/05-partial-execution-engine/05-RESEARCH.md
@.planning/phases/05-partial-execution-engine/05-01-SUMMARY.md
@src/server/workflow/task_prompt_handler.rs
@src/server/workflow/prompt_handler.rs
@crates/pmcp-tasks/src/types/workflow.rs
@src/server/tasks.rs
@src/server/workflow/workflow_step.rs
@src/server/workflow/data_source.rs
@src/server/workflow/newtypes.rs
</context>

<tasks>

<task type="auto">
  <name>Task 1: Rewrite TaskWorkflowPromptHandler.handle() with active execution loop</name>
  <files>src/server/workflow/task_prompt_handler.rs</files>
  <action>
Replace the current `handle()` method that delegates to `self.inner.handle()` with an active execution engine. The new handle() must follow this exact flow per CONTEXT.md locked decisions:

**1. Setup phase (reuse existing code):**
- Resolve owner via `self.resolve_owner(&extra)`
- Build initial progress using the typed `WorkflowProgress` struct (replace the raw JSON `build_initial_progress` with a new `build_initial_progress_typed` that returns `WorkflowProgress` with all steps set to `StepStatus::Pending`)
- Create task via `self.task_router.create_workflow_task()` (same graceful degradation pattern)

**2. Active execution loop (NEW — replaces `self.inner.handle()` delegation):**
- Initialize: `let mut messages: Vec<PromptMessage> = Vec::new()`, `let mut execution_context = ExecutionContext::new()`, `let mut step_results: Vec<(String, Value)> = Vec::new()`, `let mut step_statuses: Vec<StepStatus> = vec![StepStatus::Pending; step_count]`, `let mut pause_reason: Option<PauseReason> = None`
- Add header messages: `messages.push(self.inner.create_user_intent(&args))`, `messages.push(self.inner.create_assistant_plan()?)`
- Import `ExecutionContext` from `super::prompt_handler::ExecutionContext` (now pub(crate) from Plan 01)

- **For each step** (sequential, stop at first failure/unresolvable):
  - Check cancellation via `extra.is_cancelled()` — return error if cancelled (same pattern as inner handler)
  - Report progress via `extra.report_count()`
  - Add guidance message if step has guidance (same as inner handler)
  - Fetch pre-tool resources if applicable (same pattern as inner handler)

  - **Resource-only steps**: Add assistant message, fetch post-tool resources if needed, mark step `Completed`, store result as `Value::Null` in step_results, continue

  - **Tool steps**:
    - Call `self.inner.create_tool_call_announcement(step, &args, &execution_context)`:
      - On `Err(_)`: Classify as PauseReason using `classify_resolution_failure()` (see below), break
      - On `Ok(announcement)`:
        - Call `self.inner.resolve_tool_parameters(step, &args, &execution_context)`:
          - On `Err(_)`: Should not happen if announcement succeeded, but break as safety
        - Call `self.inner.params_satisfy_tool_schema(step, &params)`:
          - On `Err(_)`: break
          - On `Ok(false)`: Set `pause_reason = Some(PauseReason::SchemaMismatch { blocked_step, missing_fields: vec!["unknown".to_string()], suggested_tool })`, break
          - On `Ok(true)`: Proceed to execution
        - Push announcement message
        - Call `self.inner.execute_tool_step(step, &args, &execution_context, &extra).await`:
          - On `Ok(result)`:
            - Push user message with tool result (same format as inner handler)
            - `step_results.push((step.name().to_string(), result.clone()))`
            - `step_statuses[idx] = StepStatus::Completed`
            - Store binding: `execution_context.store_binding(binding, result)` if step has binding
            - Fetch post-tool resources if applicable
          - On `Err(e)`:
            - Push user error message
            - Store error as JSON value: `step_results.push((step_name, json!({"error": e.to_string()})))`
            - `step_statuses[idx] = StepStatus::Failed`
            - `pause_reason = Some(PauseReason::ToolError { failed_step, error: e.to_string(), retryable: step.is_retryable(), suggested_tool })`
            - break

**3. `classify_resolution_failure()` helper function** (private, NOT on self):
Implements EXEC-04 (runtime dependency check). When parameter resolution fails, inspect the step's arguments to determine if the failure is due to a dependency on a failed/skipped step:
- For each argument in `step.arguments()`, if the DataSource is `StepOutput { step: binding_name, .. }`:
  - Find the producing step by checking which earlier step has that binding name (iterate workflow steps, check `step.binding()`)
  - If the producing step's status is `Failed` or `Skipped`, return `PauseReason::UnresolvedDependency { blocked_step, missing_output: binding_name, producing_step, suggested_tool }`
- Default: return `PauseReason::UnresolvableParams { blocked_step, missing_param: first unresolvable arg name, suggested_tool }`
- The `suggested_tool` is the blocked step's own tool name (or empty string for resource-only steps)

**4. Batch write (after loop exits):**
Per CONTEXT.md: accumulate in memory, write all at end.
- Build updated `WorkflowProgress` with actual `step_statuses`
- Build batch `HashMap<String, Value>`:
  - `WORKFLOW_PROGRESS_KEY` -> serialized WorkflowProgress
  - For each `(step_name, result)` in step_results: `workflow_result_key(step_name)` -> result
  - If `pause_reason.is_some()`: `WORKFLOW_PAUSE_REASON_KEY` -> serialized PauseReason
- Call `self.task_router.set_task_variables(&task_id, &owner_id, batch_value).await`
- On error: `tracing::warn!()` — graceful degradation, continue to return result

**5. Auto-complete (if all steps succeeded):**
Per CONTEXT.md: auto-complete when all steps succeed, no client action needed.
- If `pause_reason.is_none()` AND all step_statuses are `Completed`:
  - Build completion result value (e.g., json with "completed": true, step count)
  - Call `self.task_router.complete_workflow_task(&task_id, &owner_id, result).await`
  - On success: set task_status = "completed" in _meta
  - On error: `tracing::warn!()` — graceful degradation, task stays Working

**6. Build _meta and return:**
Update `build_meta_map` to accept an `Option<&PauseReason>` and task_status string ("working" or "completed"):
- Include `task_id`, `task_status`, `steps` array with name+status
- If pause_reason is present, include `pause_reason` in meta
- Set `result._meta = Some(meta)`

**7. Report final workflow completion** (same pattern as inner handler).

**Important implementation notes:**
- Remove the `infer_step_statuses()` method — it's replaced by direct status tracking
- Replace `build_initial_progress()` with `build_initial_progress_typed()` returning `WorkflowProgress`
- Update `build_meta_map` signature to include pause_reason and task_status
- The step loop MUST handle resource-only steps (no tool, no binding) — mark Completed, continue
- All `use` imports needed: `PauseReason`, `WorkflowProgress`, `WorkflowStepProgress`, `StepStatus`, `WORKFLOW_PROGRESS_KEY`, `WORKFLOW_PAUSE_REASON_KEY`, `workflow_result_key` from `pmcp_tasks::types::workflow`; `ExecutionContext` from `super::prompt_handler`; `DataSource` from `super::data_source`
- The module-level doc comment should be updated to reflect the new active execution role
  </action>
  <verify>
Run: `cargo build` — compiles without errors.
Run: `cargo clippy -- -D warnings` — zero warnings.
  </verify>
  <done>
TaskWorkflowPromptHandler.handle() runs its own step loop using inner handler's pub(crate) helpers, accumulates results in memory, classifies failures into PauseReason variants, batch-writes to task store, auto-completes on full success, and includes PauseReason in _meta.
  </done>
</task>

<task type="auto">
  <name>Task 2: Add comprehensive tests for the execution engine</name>
  <files>src/server/workflow/task_prompt_handler.rs</files>
  <action>
Add tests to the existing `mod tests` block in `task_prompt_handler.rs`. These tests verify the execution engine logic, PauseReason classification, batch write shape, and _meta construction.

**Unit tests** (in `mod tests`):

1. `build_initial_progress_typed_all_pending` — verify `build_initial_progress_typed()` returns WorkflowProgress with all steps Pending, correct goal, schema_version 1
2. `build_meta_map_working_with_pause_reason` — verify _meta includes `pause_reason` when PauseReason is provided, task_status is "working"
3. `build_meta_map_completed_no_pause_reason` — verify _meta has task_status "completed" and no pause_reason field
4. `build_meta_map_with_step_statuses` — verify _meta steps array correctly maps StepStatus enum values to strings
5. `classify_resolution_failure_unresolved_dependency` — set up a scenario where step B depends on step A's binding and step A is Failed, verify `PauseReason::UnresolvedDependency` is returned with correct fields
6. `classify_resolution_failure_generic_unresolvable` — set up a scenario where a step can't resolve a PromptArg (not a step dependency), verify `PauseReason::UnresolvableParams` is returned
7. `classify_resolution_failure_skipped_producer` — same as dependency test but producer is Skipped instead of Failed, verify same UnresolvedDependency result

**Note on integration tests:** Full execution loop tests (creating a real WorkflowPromptHandler with mock tools, executing through handle()) will be in Phase 7 integration tests. Phase 5 unit tests focus on the individual helper functions (classify_resolution_failure, build_initial_progress_typed, build_meta_map) since those can be tested without a full server setup.

Update the existing tests:
- `build_meta_map_basic` — update to match new signature (add task_status and pause_reason params)
- `build_meta_map_empty_steps` — update to match new signature
- Remove `infer_step_statuses_*` tests (method was removed) — these are replaced by the new direct status tracking

Ensure all tests use the actual types from pmcp-tasks (`StepStatus`, `PauseReason`, `WorkflowProgress`) not string comparisons where typed comparisons are possible.
  </action>
  <verify>
Run: `cargo test --lib server::workflow::task_prompt_handler` — all new and updated tests pass.
Run: `cargo test --lib server::workflow` — all workflow tests pass (existing WorkflowPromptHandler tests unaffected).
Run: `cargo clippy -- -D warnings` — zero warnings.
  </verify>
  <done>
7+ unit tests verify: initial progress construction, _meta shapes for working/completed/with-pause-reason, classify_resolution_failure for unresolved dependency vs generic unresolvable params. All existing WorkflowPromptHandler tests still pass unchanged.
  </done>
</task>

</tasks>

<verification>
1. `cargo test --workspace` — all tests pass across entire workspace
2. `cargo clippy --workspace -- -D warnings` — zero warnings
3. `cargo build --workspace` — compiles cleanly
4. Verify _meta shapes manually:
   - Working with pause: `{"task_id": "...", "task_status": "working", "steps": [...], "pause_reason": {"type": "toolError", ...}}`
   - Completed: `{"task_id": "...", "task_status": "completed", "steps": [...]}`
5. Verify batch write includes all three key types: `_workflow.progress`, `_workflow.result.<step>`, `_workflow.pause_reason`
6. Verify existing non-task workflow tests pass unchanged (backward compatibility)
</verification>

<success_criteria>
- TaskWorkflowPromptHandler.handle() runs its own active execution loop, not delegating to inner.handle()
- Step results accumulated in memory, batch-written to task store at end (EXEC-01)
- Execution pauses at first unresolvable step, task stays Working (EXEC-02)
- Tool errors produce PauseReason::ToolError with retryable flag, task stays Working (EXEC-03)
- Unresolved dependencies produce PauseReason::UnresolvedDependency distinct from UnresolvableParams (EXEC-04)
- Auto-complete transitions task to Completed when all steps succeed
- Graceful degradation: batch write failure returns prompt result with in-memory _meta
- All existing tests pass unchanged
- Zero clippy warnings
</success_criteria>

<output>
After completion, create `.planning/phases/05-partial-execution-engine/05-02-SUMMARY.md`
</output>
