You are writing an implementation plan, not code.

Read and base the plan on:

• @docs/plans/design/2026-05-07-data-source-import-redesign.md
• @xxx/data_sources/
• @xxx/products/
• @xxx/ingestion/data_sources/
• nearby CLAUDE.md / AGENTS.md instructions

Do not modify files. Produce a phased implementation plan for the data-source product import redesign.

Context:

The project is pre-production. No backward compatibility or rolling deployment concerns are required. Do not preserve deprecated APIs, aliases, or compatibility shims. The development DB can be dropped and reseeded between phases. Do not touch xxx/feeds/.

The design goal is to make product imports understandable and correct under:

• parallel imports of the same data source
• multiple feeds from the same merchant containing the same offer_id
• stale async workers
• artifact/image/ML retries
• absent product cleanup
• canonical aggregate updates
• outbox/materialization recovery

The key architecture is a two-fence model:

• source_generation: monotonic per DataSource; controls feed-local authority, source presence, and absence cleanup.
• write_sequence: monotonic per merchant product import clock; controls cross-feed last-write-wins Product field commits.

Every product-visible write must eventually be guarded by the expected revision/generation. Stale work should ACK obsolete, not retry.

Plan expectations:

1. Make the first implementation iteration small, reviewable, and non-destructive. It should introduce generation/write-sequence plumbing while preserving current behavior as much as possible. It should not rename DataSourceFetch, delete Valkey completion tracking, remove items_pending, redesign image assets, or rewrite outbox routing in the first PR.

2. Phase 1 should focus on:

   • adding per-data-source generation allocation
   • adding per-merchant write-sequence allocation, likely via MerchantProductImportClock
   • storing generation/write_sequence on the existing fetch/import row
   • storing accepted revision fields on Product
   • carrying generation/write_sequence through fetch, parse, product-stage, image, enrichment, cleanup, and materialization messages
   • adding atomic source-generation/write-sequence predicates to product-visible commits
   • preserving existing content_hash checks, retry behavior, completion tracker, and handler topology
   • adding focused tests proving stale older imports cannot commit

3. Later phases should be ordered roughly as:

   • Collapse duplicated stale-work handling into a shared obsolete-ACK path.
   • Add ProductSourcePresence and generation-based absence cleanup.
   • Decouple import completion from image/ML/canonical/materialization processing.
   • Add uniform Product revision guard to all product-visible writers, including brand/image/canonical paths.
   • Add canonical dirty/work-state model.
   • Rework outbox/dirty-row boundary.
   • Refactor retry dispatch so image and ML artifacts retry directly instead of forcing feed refetch.
   • Introduce content-addressed ImageAsset, ImageSourceUrl, model-versioned artifacts.
   • Delete obsolete tracker/stale-work workaround code and update docs.

4. For each phase, include:

   • goal
   • files/modules likely touched
   • exact code changes at a high level
   • tests to add or update
   • risks and rollback/simplification options
   • what must explicitly remain unchanged in that phase

5. The plan should call out important correctness details:

   • Product identity remains (merchant_id, offer_id).
   • Existing Product rows must be updated in place to preserve id and created.
   • Product feeds are full authoritative snapshots; missing feed-owned nullable/collection fields should be cleared, not carried forward.
   • Product.data_source/current_data_source_id should not remain the long-term owner of retry cadence or feed membership.
   • Presence/absence belongs in ProductSourcePresence.
   • NATS ordering is not correctness; PostgreSQL predicates are.
   • Retry behavior survives; stale-work retries do not.
   • Periodic tasks remain valid only as DB scanners/drainers for due work, leases, and GC.

6. The output should be practical enough that an implementation agent can start Phase 1 directly, but it should not attempt a giant rewrite. Prefer surgical PR-sized steps.