Local execution planning

pipewhy.execution.prepare_enron_execution(configuration, root, ...) verifies pinned shard schemas and row counts, then atomically persists an immutable plan and its source manifest. This prepares work; it does not dispatch workers or publish a completed observation. The existing CLI scan remains sequential.

Plans bind source/version/scope, C3 context, exact adjacent ranges and execution policy. Their IDs exclude cache locations, and conflicting or corrupt persisted documents fail explicitly. load_plan(directory) and load_source(directory) reopen verified planning metadata without a Parquet dependency.

Planning supports explicit reference or cpu worker profiles. Defaults record 4 workers, 2 attempts, a 600,000 ms operational timeout and 128 Arrow rows per batch. Limits are 61 workers, 10 attempts, a 24-hour timeout and 65,536 Arrow rows; plans/source documents are bounded to 4 MiB. These are implementation policies, not performance guarantees.

execute_partition(directory, cache, run_id, index, attempt=1) executes one prepared Enron shard without network or journal/snapshot writes. It returns a versioned WorkerResult; call result.validate(plan, run_id) before consuming its state. Results bind run/plan/partition, revision, source/context identity and backend, and verify exact C3 count/global range. Failed results carry safe reason codes and progress with no comparable state. The shard API explicitly represents partition scope; it never claims a complete dataset on its own. The same C2 converter serves full and shard reads.

RunJournal.create(root, plan) creates a new journal with its own run ID. journal.accept(result) admits a complete matching result once; an identical retransmission returns False, and conflicting deliveries fail. Failed envelopes have a separate immutable history. Journal payloads have checksums, and reopening validates identities before reconstructing state.

journal.reconcile() returns explicit complete/failed/missing partition IDs and accepted row progress. Its state remains None until every planned partition, including zero-row shards, has a valid complete receipt. Merging follows manifest order. A source failure or missing partition cannot become a complete fingerprint.

Coordinated dispatch, durable attempt budgets, retries/resume, OS ownership and replay-safe C5 publication are implemented. Use execute_run, resume_execution and publish_execution as described in Parallel scans. The CLI scan stays sequential. See blueprint/EXECUTION_CONTRACT.txt for qualification evidence.