# Stream K2: the schema (decisions, proposals, attributes, places), from the escalation's spec
(written 2026-09-16 Wed, from the accepted plan W7, W15, W18, W19 and §8, and the escalation-2 specification `.logs/planning/decisions-model-spec-2026-09-16.md`, which is the design; this brief is the work order on top of it. Runs alone, Thu 17 Sep day, after K1's hand-back. The heaviest stream: K3, K4 and K5 all wait on it, so the order of its tasks puts what they need first.)

**Runs:** alone, Thu 17 Sep, after K1's hand-back ("K1 green and ready; K2 may start"), on the
refactor branch in the worktree. Never alongside a collection, never alongside another stream (it
touches `keying.py`, `merges.py`, `cli.py` and the models, which every later stream reads).

## Goal
One migration and one service give every later stream what it builds on: the append-only
`decisions` ledger with its writer, effective read, undo, batch undo and natural-key replay; the
`proposals` store the review sheet reads and the AI pass loads; the attribute registry and the one
accessor every reader uses; `places` and `shop_places` with the comparison unit defined once; uids on
every row a decision can name; `hidden` and `indexed` as columns; the `overrides` table gone and its
readers moved behind a shim. Every existing test green through the shim. Nothing on a page changes.

## The design is the spec; read it whole
`.logs/planning/decisions-model-spec-2026-09-16.md` (436 lines) is the table-by-table, rule-by-rule
design: §0 the five rules, §1 the tables, §2 natural keys, §3 the read path, §4 the write path, §5
pairs and merges as rows, §6 proposals (K4 builds the load, sheet and approval; K2 builds the
tables), §7 undo, §8 replay, §9 the migration, §10 the 22 tests, §11 known limits. Where this
brief and the spec differ, this brief wins (it is later and records decisions the spec left open).

**The spec was written before K1's renames.** K2 runs after K1, so every name in the spec is
read through this table and nothing is written in the old words:

| the spec says | K2 writes |
|---|---|
| `products`, `products.id`, `product_id`, `line_id` | `product_variants`, `variant_id`, `product_line_id` |
| `merge_candidates`, `MergeCandidate`, `decision` values `merged`/`kept_apart` | `suggestions`, `Suggestion`, values `same`/`separate` (K1 backfilled them) |
| `product_merges`, `ProductMerge` | `merges`, `Merge` |
| `variation_aliases`, `VariationAlias`, `attributes.variation` | `attribute_aliases`, `AttributeAlias`, `attributes.<kind>` |
| `canonical_id` | `alias_of_id` |
| `locations`, `location_id`, `location_code` | `shops`, `shop_id`, `shop_code` |
| `lines.py`, `line_key`, `line_for` | `product_lines.py`, `product_line_key`, `product_line_for` |
| `size_ml` | `quantity_ml` |
| spec §9 item 6's backfill of the suggestion values | already done by K1; K2 adds NO CHECK on `suggestions.decision` (a later, unconditional revision does, after the backfill has run on every host: the launch chain runs in one pass, so a CHECK in K2's revision would fail on production) |

## Decisions this brief takes on the spec's open questions (rian may overturn any on the running list)
1. **Newest wins by `(decided_at, id)`**, not by id alone (spec rule 4). An older staging decision
   replayed onto production after a newer production decision on the same field is inserted and
   reported, never materialised. Accepted; it changes nothing for a native write.
2. **Undo restores the previous decision**, releasing to the rules only when none existed (spec §7).
   Plan W15's sentence "value = rule_value" is read as this; the plan is not edited by K2.
3. **Quantity is `attribute:quantity`** in the ledger (W7's one kind with structured storage); the
   shim carries the bare `quantity` field name for K2's release only.
4. **A sheet approval is one transaction per brand**, gated on no collection running (spec §6.4).
5. **Ingest's arrival proposals are K3's** (K3.5); K2 admits the shape (`generator = arrival`, pass
   `arrival:<rules_version>`) and writes no arrival row.
6. **The `decided_member` veto on the automatic fold ships** (a group holding a variant with an
   effective identity decision is a suggestion, never a fold). No person-less carry is built.
7. **`redirects` replaces the spec's `line_slugs`**, because K6 needs the same table for brands and
   places: `redirects (from_slug VARCHAR(240) PK, kind VARCHAR(16) CHECK IN (brand, product_line,
   place), to_slug VARCHAR(240) NOT NULL, since TIMESTAMPTZ NOT NULL DEFAULT now(), decision_id BIGINT
   NULL FK decisions.id)`, index `(kind, to_slug)`; flattened at write (A to B then B to C stores A to
   C); the replay resolver reads it last, as the spec reads `line_slugs`. K6 writes the rows through
   `publish.redirect_write()`; K2 creates the table and the model only.
8. **`place` joins the ledger's entity types** (K6 decides `indexed` and `hidden` on places): the
   `entity_type` CHECK gains `place`, fields `name`, `hidden`, `indexed`; `places` gets `uid`,
   `hidden`, `indexed` like brands and product lines; natural key `place:<slug>`.

## What already exists, so you build on it
- After K1: the renamed tables and models; `overrides` still present (0 rows on every host) with
  `services/overrides.py` (`FIELDS`, `read`, `decide`, `ledger`, `review`, `hidden_ids`, `move`,
  `clear`, the pin and ignore routes) and its readers in `cli.py`, `routers/collectors.py`,
  `keying.py`, `merges.py`, `merge_session.py`, `listings_table.py`, `catalog_queries.py`,
  `award_picker.py`, `product_lines.py`; `test_decided.py` and the kit suites pin today's behaviour.
- `keying.load_maps` and `Maps` (the 300 s cache); `merges.duplicate_groups`, `_conflicts`,
  `merge_duplicates`, `_fold`, `merge_products`, `apply_brand_alias`, `apply_line_alias`;
  `merge_session.confirm`/`reject`; `merge_desk.batch`/`propose`; `suggest.generate`;
  `cli.backfill_lines`, `backfill_prune_lines`, `_collection_running`; `ingest._resolve_product`,
  `_follow_tombstone`, the enrich guard; `merges._ENRICH_FIELDS`.
- `accounts.py`'s `JSONVariant`; `audit_log`; `settings.app_env`; alembic head after K1 (K1's
  handoff names it); `main/scripts/staging-refresh.py` with `--discard-decisions` and its guard query.
- `shops.iata` (K1's name for `locations.iata`); the 23 shops; `catalog_queries` counts of shops.
- `main/docs/REVIEW-PROCESS.md` (K0): §3 says what a proposal carries; §4 the spot-check rule; K2's
  `proposals` columns hold exactly those.
- The kit: `main/tests/` runs against in-memory SQLite in seconds; `dfp-devdb` (127.0.0.1:5433)
  for the Postgres rehearsal (`DATABASE_URL=postgresql+psycopg://dfp:dfp@127.0.0.1:5433/dfp_k2`).

## The signatures K3, K4, K5 and K6 call (fixed here; a later change is an issue on the running list)
```
# services/decisions/effective.py
decisions.effective(db, entity_type, entity_ids=None, fields=None) -> dict[(entity_id, field), Decision]
    # newest wins by (decided_at, id); an effective `release` row is absent from the result
decisions.separated(db) -> set[frozenset[int]]        # variant pairs kept separate, survivor-chased
Maps.decided[variant_id][field] -> Decision            # filled by keying.load_maps in one query
Maps.separated -> set[frozenset[int]]

# services/decisions/writer.py
with decisions.batch(db, kind, mode, by, scope=None, pass_id=None, note=None) as b: ...
decisions.record(b, entity_type, entity, field, value, *, origin, proposal=None, reason=None,
                 decided_at=None, effect="set", cause=None) -> Decision      # raises decisions.Refused(code)
decisions.undo(b, decision_id, reason) -> Decision
decisions.undo_batch(db, batch_uid, by, reason) -> DecisionBatch             # its own batch, kind=undo
decisions.Refused codes: DECISION_HELD, ENTITY_ALIASED, PRODUCT_MERGED, PAIR_SIDE_MERGED, NOTE_REQUIRED,
    VALUE_INVALID, FIELD_UNKNOWN, DECISION_SUPERSEDED, ENTITY_MERGED, UNMERGE_CHAIN, COLLECTION_RUNNING,
    SLUG_TAKEN, LINE_MISMATCH, VARIANT_SPLIT, LISTING_MISSING, REVERSAL_TARGET_MISSING

# services/decisions/natural_keys.py
natural_keys.build(entity) -> (key: str, detail: dict)
natural_keys.parse(key) -> (entity_type, parts)
natural_keys.resolve(db, entity_type, key, detail, *, mint=False) -> row | Parked(code, candidates)

# services/decisions/replay.py  (CLI: app.cli decisions export | replay | verify | list | undo-batch)
replay.export(db, *, since=None, batch_uid=None, pass_name=None, all_=False, out) -> counts
replay.replay(db, file, *, check=False, label=None) -> counts   # parked rows to replay-remainder.jsonl

# services/proposals_store.py  (K2 owns the table and this thin store; K4 owns load/sheet/approve)
proposals.write(db, pass_name, rows: list[ProposalRow], *, kind, generator, rules_version,
                process_version, scope_brand_slug=None) -> ProposalPass   # get-or-create the pass by name;
                # upsert rows on (pass_id, entity_type, natural_key, field); resolve entity_id; savepoint per row
ProposalRow(entity_type, natural_key, natural_key_detail, field, value, reason, evidence, confidence,
            brand_slug, sheet_line_ref=None, position=0)

# services/attributes.py
attributes.KINDS: dict[kind, Kind(name, label, value_type, unit, verticals, categories, identity, display, storage)]
attributes.of(variant) -> list[Attribute(kind, value, unit, label, identity, display)]   # every kind with a value
attributes.get(variant, kind) -> value | None
attributes.set(variant, kind, value) -> None            # validates by the registry; writes the storage it names
attributes.identity_part(variant) -> str                # "kind=value|kind=value", identity kinds sorted by name
attributes.quantity_slot(variant) -> str                # "100ml" | "50g" | "200pcs" | "unknown" | "n/a"
attributes.kinds_for(vertical, category) -> list[Kind]
attributes.GUARDED_COLUMNS: dict[field, column]         # the enrich guard reads this, never a hand list
attributes.register_option(shop_name) -> Kind           # `option:<name>`: certain, identity, text, display=picked

# services/places.py
places.KINDS: dict[kind, PlaceKind(name, comparable, pages, address_prefix, moves)]
places.primary_of(db) -> dict[shop_id, place_id]        # one query; cached with Maps' lifetime
places.count_units(shop_ids, primary) -> int            # "shops counted once per primary place"
places.unit_join(query) -> query                        # the SQL form: joins shop_places(role=primary)

# services/publish.py is K6's; K2 ships the STUBS with these signatures so K3 and K5 compile:
publish.slug_at_confirm(row, chosen_name) -> str | None   # no-op stub returns None
publish.is_indexed(page) -> bool                          # reads the `indexed` column
publish.is_hidden(page) -> bool                           # reads the `hidden` column
publish.redirect_for(slug, kind) -> str | None            # reads `redirects`
```

## Tasks, in order (ids on /plan; commit prefix `K2:`; what K3, K4 and K5 need lands first)
1. **K2.1 The migration.** One Alembic revision chained on K1's head, schema-only (spec §9 with the
   name table above and decisions 7 and 8): `uid` on `brands`, `product_lines`, `product_variants`,
   `suggestions`, `places`; `proposal_passes`, `decision_batches`, `proposals`, `decisions` (the
   `fk_decisions_proposal` cycle with `use_alter`), `redirects`, `places`, `shop_places`; the two
   Postgres triggers (`decisions_append_only`, `decision_batches_guard`) by `op.execute` guarded on
   the dialect; `hidden`/`indexed` on brands, product lines, places with partial indexes;
   `suggestions.closed_reason`, `suggestions.decision_id`, the `separate` partial index;
   `merges.decision_id/batch_id/reversed_by_id/reversed_at`; `DROP TABLE overrides` after asserting
   0 rows. Downgrade exact, FKs dropped first. Rehearse `upgrade head`, `downgrade -1`, `upgrade
   head` on `dfp_k2`, a fresh copy of the newest dump with K1's migration and backfill applied;
   `alembic heads` one head. Commit before anything else, so a K1 follow-up can chain on it.
2. **K2.2 Models, natural keys, the effective read.** `models/decisions.py` (`ProposalPass`,
   `DecisionBatch`, `Proposal`, `Decision`, `Redirect`, `LEDGER_TABLES`), `models/places.py`
   (`Place`, `ShopPlace`); `services/decisions/natural_keys.py` (spec §2, round-trip tested on real
   slugs, SKUs with `:`, shop codes; `line:new:<slug>`; `pair:<level>:<a>||<b>` sorted); `effective.py`
   (spec §3.3; the one ordered query, dialect-neutral, `release` applied); `keying.load_maps` filling
   `Maps.decided` from it in one query with the legacy field aliases (`quantity`, `line_id`,
   `variation`) for one release, and `Maps.separated` with the `{id: merged_into_id}` chase.
3. **K2.3 The attribute registry and accessor.** `services/attributes.py` per W7 and the signatures
   above: the kinds with their storage (`quantity` typed and structured; `abv`, `country_of_origin`,
   `is_exclusive` typed; every other kind under `attributes.<kind>`), `identity` and `display` per
   vertical and category, `not_applicable` quantity per category (`n/a` equals itself; `unknown` never
   does), `option:<shop name>` kinds registered on sight, a GIN index on `attributes` if the copy
   shows the containment query needs it (measure, do not assume). `GUARDED_COLUMNS` derived from the
   registry; `ingest`'s enrich guard and `merges._ENRICH_FIELDS` read it. K1's `ATTRIBUTE_KINDS` and
   `ATTRIBUTE_RULES` fold into the registry (one definition). `attribute_aliases` rows stay the
   wording table (`meaning`, `display`, `kind`).
4. **K2.4 Places.** `services/places.py` per W19: the kind registry (airport, mall, port, ship with
   `moves`, border crossing, in-flight with `pages: false`, online with no primary place),
   `app.cli backfill places` (idempotent: one airport place per IATA code from `shops.iata`, mirrored
   into `identifiers` as `(iata, code)`; every shop's `shop_places` row with role `primary`;
   second run zero), `primary_of`, `count_units`, `unit_join`; every comparison-unit count in
   `catalog_queries.py` (`_brand_counts`, the savings and airport counts, the "two or more shops"
   rule) routed through it; **a test that fails on any `COUNT(DISTINCT shop_id)` or count of shop
   rows in `catalog_queries.py`** (grep-based, as the survivor test in K1). `/airports/` pages
   unchanged; the counts they print equal yesterday's on the copy (one airport per shop today, so
   the numbers are identical: prove it, do not assume it).
5. **K2.5 The writer and the shim.** `services/decisions/writer.py` (spec §4: `batch`, `record`
   steps a to g, advisory locks on Postgres, human-first check inside the writer, `rule_value` by
   spec §1.7, `value_ref_id`, uuid5 consequences), `appliers.py` (`APPLIERS[(entity_type, field)]`
   split into `materialise` and `consequences`; today's appliers moved behind it: `apply_brand_alias`,
   `apply_line_alias` with `detail.products_moved`, `apply_variation_alias`, the pin and ignore
   writes, the suggestion decision, `merge_products` for `merged_into` with the carry of spec §5).
   `services/overrides.py` becomes the shim (same public names, the field renames in one table,
   `clear` an undo, `move` the carry, `review(state)` the `hidden`/`indexed` booleans with `checked`
   refused `VALUE_INVALID`); every reader routed through it; `overrides.decide_product` a batch of
   one (`kind = route`); `merge_session.confirm`/`reject` and `merge_desk.batch`/`propose` batches
   (`kind = desk`; `reject` writes Keep separate); `cli.backfill_lines` carrying by consequence rows;
   `Override = Decision` alias with `collected_value`, `set_by`, `set_at` read-only properties.
   `test_decided.py` changes only its TABLES list and the `collected_value` reads.
6. **K2.6 The machine paths honour the ledger.** `merges._conflicts(members, maps)` gains
   `kept_separate` and `decided_member` (spec §3.1); `_fold` loads the separated set fresh and stamps
   `merges.batch_id` at a batch's tail; `suggest.generate` writes `closed_reason` only, in a savepoint
   per row, tolerating a pre-minted `decided` pair; the desk, queue and `progress()` read open pairs as
   `decision IS NULL AND closed_reason IS NULL`; `backfill_prune_lines` keeps any line named by any
   decision row (value, `value_ref_id`, `rule_value`, entity) or a `redirects` row; `rederive` and
   `backfill_lines` never re-point a variant holding a `product_line` decision nor re-key a line
   holding a `name` or `alias_of` decision (K3 changes the key; K2 keeps the guard).
7. **K2.7 Undo and batch undo.** `services/decisions/undo.py` (spec §7: consequences first,
   `reverses_id`/`restores_id`, the merge undo from the `merges` record with `reversed_by_id`, the
   alias undo by `detail.products_moved`, `UNMERGE_CHAIN`, `DECISION_SUPERSEDED`), `undo_batch`
   (one transaction, tail folds first, descending id, superseded rows skipped and named); CLI
   `app.cli decisions list [--entity <type> <id> | --batch <uid> | --pass <name>] [--since]`,
   `decisions undo-batch <uid> --reason`, `decisions verify` (read-only drift report; joins
   `check.sh`'s after-deploy chain in the RUNBOOK). The route `POST /review/batches/<uid>/undo` is
   K4's; K2 exposes the service.
8. **K2.8 Export and replay by natural key.** `services/decisions/replay.py` and the CLI (spec §8:
   export in `(decided_at, id)` order with no local id in the payload, consequences omitted; replay
   per source batch with the same uid, per row in a savepoint, the eight steps, `SUPERSEDED_ON_TARGET`
   inserted not materialised, the remainder file, `--check` in one rolled-back transaction, non-zero
   exit on parked rows; `COLLECTION_RUNNING` gate). `main/scripts/staging-refresh.py`: the guard
   counts `decisions` and `proposals`, runs with `-v ON_ERROR_STOP=1`, fails hard on a non-zero
   psql exit; the preserved set carries the ledger by `decisions export --all` before the drop and
   `decisions replay` after `alembic upgrade head`; `--discard-decisions` keeps its meaning.
9. **K2.9 Tests, the two-copy rehearsal, docs, hand-back.** The spec's 22 tests (§10), each
   docstring naming the failure it pins; the spec §8 acceptance on `dfp_k2` and `dfp_k2b` (about
   fifty decisions across every entity type including a desk Confirm same, a brand alias, a Keep
   separate and one undo, exported, replayed onto the second copy with shifted ids, an extra derived
   line and a `-2` slug: 0 parked, every column equal alias-followed, a second replay applies 0,
   `undo-batch` on B of a batch made on A reverses it); `rederive --check` on the copy printing zero
   changed keys (K2 changes no identity); `backfill places` twice. Docs: `DATA-MODEL.md` (the five
   tables, the registry, places, the natural keys), `RUNBOOK.md`'s after-deploy block (the migration
   id, `backfill places`, `decisions verify`), `VOCABULARY.md`'s "code says" column for decision,
   proposal, attribute, place, `CHANGELOG.md` Unreleased; `docmap.py --write`; `main/check.sh` green.

## Owns
`main/alembic/versions/` (one revision), `app/models/decisions.py`, `models/places.py`,
`models/catalog.py` (the uid, hidden, indexed, closed_reason, decision_id, batch_id columns),
`app/services/decisions/` (new package), `services/attributes.py` (new), `services/places.py`
(new), `services/proposals_store.py` (new; the thin `write`), `services/publish.py` (the stubs
only; K6 replaces the bodies), `services/overrides.py` (the shim), `keying.py` (`load_maps`,
`Maps.decided`, `Maps.separated` only), `merges.py` (`_conflicts`, `_fold`, the appliers moved
behind the registry, `merge_products`'s decision and batch stamps), `merge_session.py` and
`merge_desk.py` (batches only), `suggest.py` (`closed_reason`, the savepoint), `ingest.py` (the
enrich guard reading `GUARDED_COLUMNS` only), `cli.py` (`decisions *`, `backfill places`,
`backfill decisions-from-columns`, the carry in `backfill_lines`, `backfill_prune_lines`),
`catalog_queries.py` (the counting routes through `places` only), `main/scripts/staging-refresh.py`,
their tests, `DATA-MODEL.md`, `RUNBOOK.md`'s after-deploy block, `VOCABULARY.md`'s code column.

## Must not touch
Routers and the SPA (K4 the review, K5 the page; nothing on a page changes in K2), identity's
rules (`product_lines.py`, `normalize.py`, the key functions, the collectors: K3), `seo.py`,
`indexnow.py`, `robots_policy.py`, `urls.py` (K6), `deploy/` (K7), `.app.env`, `.env`, the live
databases, `import/` beyond `progress.json` through the script. K2 changes no identity: `rederive
--check` prints zero.

## Rehearsal and acceptance
- `dfp_k2` from the newest nightly dump with K1's migration and backfill applied: `upgrade head`,
  `downgrade -1`, `upgrade head`; the `overrides` assertion; the trigger refusing an UPDATE and a
  DELETE (prove it with a psql statement that fails); `backfill places` twice; `rederive --check`
  zero; `decisions verify` zero drift; the counts the airport pages print unchanged.
- The two-copy replay acceptance (K2.9) with its numbers in the handoff.
- `../.venv-dev/bin/pytest tests -q` green through the shim; `main/check.sh` green;
  `docs-check.sh --strict` no FAIL.

## Rollback
`downgrade -1` (the ledger tables hold 0 native rows on every host until the review pass; the
revision recreates `overrides` exactly); the previous commit for the code. The uid columns are
additive.

## Hand-back
Handoff (≤25 lines) with the migration id, the after-deploy commands in order (`alembic upgrade
head`; `backfill places`; `decisions verify`), the two-copy numbers, the signatures file's location
(this brief's block is the contract; say if any signature changed and why), and the line "K2 green
and ready; K3, K4 and K5 may start". Anything only rian can decide (the eight decisions above if
one proved wrong in the building) on the running list with the assumption proceeded under.