---
title: 'Harden tag status PATCH contract (GAP-041 follow-up)'
type: 'refactor'
created: '2026-04-23'
status: 'done'
baseline_commit: 'd7580e87b1d748ceb3c2fda151f4f5c97584bbe5'
context:
  - '_planning/docs/planning-artifacts/2-planning/system-gaps-and-changes.md#gap-041'
  - 'src/api/schemas/tags.py'
---

<frozen-after-approval reason="human-owned intent — do not modify unless human renegotiates">

## Intent

**Problem:** The tag PATCH contract in `airflow/plugins/api/tags.py` diverges from the canonical tag schema at `src/api/schemas/tags.py`: (1) the status `Literal` omits `"pending"` while `_determine_tag_status` in `semantic_tagger.py` routinely writes `pending` tags; (2) emitted history `action` strings (`"status_changed"`, `"renamed"`) ignore the canonical `TagHistoryAction` enum (`suggested | confirmed | rejected | edited | migrated`); (3) no state-transition validation — any status overwrites any status, including no-op `verified → verified`, still firing a Milvus upsert.

**Approach:** Widen the PATCH `Literal` to the three live statuses (`pending | verified | rejected`); map emitted history actions onto the canonical enum; add a module-level transition matrix that rejects same-status and any unknown current-state with 400 before any write.

## Boundaries & Constraints

**Always:**
- Forward-only. Existing DB rows with `"status_changed"` / `"renamed"` history actions stay as-is — no backfill, no migration.
- Preserve the module's `Do NOT import from src.*` rule (it runs in the Airflow api-server process). Re-declare minimal string constants locally; do NOT import `TagHistoryAction` / `TagStatus`.
- All three PATCH status outcomes (`verified | rejected | pending`) MUST fire the same `upsert_chunk_field` Milvus partial update and append a history entry.
- Invalid-transition responses use the existing `error_response(code, msg, status_code=400)` helper with a new code `INVALID_TRANSITION`.
- Bulk endpoint stays all-or-nothing: pre-validate every transition before any write; on any offender, return 400 listing all offending `tag_id`s and write nothing.

**Never:**
- Do not change the DELETE endpoint, the rename endpoint's duplicate-name check, or the `old_name` extra field on rename history entries.
- Do not import from `src.*` inside `airflow/plugins/api/tags.py`.
- Do not add strict Pydantic validation of history `action` against an enum inside this module — existing rows carry legacy strings.

## I/O & Edge-Case Matrix

| Scenario | Input / State | Expected Output / Behavior | Error Handling |
|----------|--------------|---------------------------|----------------|
| Confirm a pending tag | PATCH status; tag.status=pending; body.status=verified | 200; tag.status=verified; history += `{action:"confirmed", by, ts}`; Milvus upsert | — |
| Reject a pending tag | PATCH status; tag.status=pending; body.status=rejected | 200; tag.status=rejected; history += `{action:"rejected", by, ts}`; Milvus upsert | — |
| Re-open a verified tag | PATCH status; tag.status=verified; body.status=pending | 200; tag.status=pending; history += `{action:"edited", by, ts}`; Milvus upsert | — |
| Same-status no-op | tag.status=verified; body.status=verified | 400 `INVALID_TRANSITION` "Tag already in status 'verified'" | no write, no Milvus call |
| Bulk: one invalid transition | 3 updates, 1 same-status | 400 `INVALID_TRANSITION` with offending `tag_id`s listed; atomic — no row writes | rls_connection rollback via `_EarlyReturn` |
| Rename emits enum action | PATCH name | 200; history += `{action:"edited", old_name, by, ts}` (was `action:"renamed"`) | — |

</frozen-after-approval>

## Code Map

- `airflow/plugins/api/tags.py` — PATCH endpoints; widen Literal, add transition matrix, remap action strings (lines 48–50, 111–115, 145–148, 194–197, 248–250)
- `src/api/schemas/tags.py` — canonical `TagStatus` + `TagHistoryAction` enums (reference only; do not import)
- `tests/airflow/test_plugins/test_tags.py` — add cases for `pending` PATCH, invalid transitions, same-status rejection, bulk atomicity, enum action strings
- `airflow/dags/tasks/semantic_tagger.py` — already emits `action="suggested"` at creation (no change; read for parity only)

## Tasks & Acceptance

**Execution:**
- [x] `airflow/plugins/api/tags.py` -- widen `TagStatusUpdate.status` to `Literal["pending","verified","rejected"]` -- align PATCH contract with stored domain
- [x] `airflow/plugins/api/tags.py` -- add module-level `VALID_TRANSITIONS` dict and `_status_to_history_action(to_status) -> str` helper -- state machine + enum-aligned action emission
- [x] `airflow/plugins/api/tags.py` -- update `update_tag_status` to call the transition guard and map action via the helper BEFORE any write or Milvus call -- single-tag path
- [x] `airflow/plugins/api/tags.py` -- update `bulk_update_tag_status` to pre-validate every transition and return 400 with all offending `tag_id`s before applying any update -- all-or-nothing atomicity
- [x] `airflow/plugins/api/tags.py` -- change `rename_tag`'s history emission from `action="renamed"` to `action="edited"` (preserve `old_name` extra field) -- enum alignment
- [x] `tests/airflow/test_plugins/test_tags.py` -- add cases: (a) PATCH `status=pending` on a verified tag, (b) 400 on same-status, (c) bulk rejects all when any transition is invalid and no row is written, (d) emitted history `action` equals `"confirmed"|"rejected"|"edited"` per transition, (e) rename emits `"edited"` -- coverage

**Acceptance Criteria:**
- Given a tag with `status=pending`, when PATCH `body.status=verified`, then response is 200 and the new history entry has `action="confirmed"`.
- Given a tag with `status=verified`, when PATCH `body.status=verified`, then response is 400 with code `INVALID_TRANSITION` and no Milvus `upsert_chunk_field` call occurs.
- Given a bulk PATCH with 2 valid transitions plus 1 same-status, when the endpoint is called, then response is 400 listing the offending `tag_id`, and the `data_hierarchical_nodes` row is byte-identical to before the call.
- Given a PATCH rename, when it succeeds, then the appended history entry has `action="edited"` (not `"renamed"`) and preserves `old_name`.

## Design Notes

State machine as a single module-level constant keeps the rules one-glance visible and avoids per-request allocation:

```python
VALID_TRANSITIONS = {
    "pending":  {"verified", "rejected"},
    "verified": {"pending", "rejected"},
    "rejected": {"pending", "verified"},
}

def _status_to_history_action(to_status: str) -> str:
    # Enum values from src.api.schemas.tags.TagHistoryAction (re-declared locally per module rule)
    return {"verified": "confirmed", "rejected": "rejected", "pending": "edited"}[to_status]
```

Transition guard returns a tuple `(ok: bool, reason: str | None)` so the bulk path can collect offenders without raising. The state machine trusts `from_status ∈ {pending, verified, rejected}` — no defensive branch for malformed/legacy statuses (delete is hard-delete at line 284, so `"deleted"` is unreachable; `_determine_tag_status` in `semantic_tagger.py` only emits the three live values). If a corrupt value somehow exists, the guard's `KeyError` falls through to the existing `except Exception` at line 158/210 and returns 500 — acceptable for an impossible state. History entries stay untyped strings on the wire; legacy entries with `"status_changed"` / `"renamed"` remain readable.

## Verification

**Commands:**
- `uv run pytest tests/airflow/test_plugins/test_tags.py -v` — expected: all new cases pass, all existing cases still pass
- `uv run ruff check airflow/plugins/api/tags.py` — expected: clean
- `grep -nE '"status_changed"|"renamed"' airflow/plugins/api/tags.py` — expected: no matches (legacy emission strings gone)

**Manual checks:**
- Inspect `airflow/plugins/api/tags.py` for one module-level `VALID_TRANSITIONS` and `_status_to_history_action`; confirm both PATCH endpoints call the guard BEFORE `_write_tags` / `upsert_chunk_field`.

## Suggested Review Order

Read in this order to see the contract tighten from declaration → state machine → call sites → tests.

### 1. Contract surface (start here)
- [`airflow/plugins/api/tags.py:48`](../../airflow/plugins/api/tags.py#L48) — `TagStatusUpdate.status` Literal widened to `pending | verified | rejected` (was missing `pending` despite stored tags routinely carrying that status).

### 2. State machine (the heart of the change)
- [`airflow/plugins/api/tags.py:120`](../../airflow/plugins/api/tags.py#L120) — `VALID_TRANSITIONS` module-level constant; one-glance rules, no per-request allocation.
- [`airflow/plugins/api/tags.py:132`](../../airflow/plugins/api/tags.py#L132) — `_status_to_history_action` remaps PATCH target status onto the canonical `TagHistoryAction` enum values (`confirmed | rejected | edited`) without importing from `src.*`.
- [`airflow/plugins/api/tags.py:136`](../../airflow/plugins/api/tags.py#L136) — `_check_transition` returns `(ok, reason)` so the bulk path can collect offenders without raising.

### 3. Single-tag PATCH
- [`airflow/plugins/api/tags.py:151`](../../airflow/plugins/api/tags.py#L151) — `update_tag_status`; guard + enum-aligned history emission fire BEFORE `_write_tags` / `upsert_chunk_field`, so same-status no-ops short-circuit with `400 INVALID_TRANSITION` and never touch Milvus.

### 4. Bulk PATCH (all-or-nothing)
- [`airflow/plugins/api/tags.py:196`](../../airflow/plugins/api/tags.py#L196) — `bulk_update_tag_status`:
  - Missing-tag check (existing behavior, preserved).
  - New `DUPLICATE_TAG_ID` rejection closes a guard-bypass: split validate/apply loops would otherwise see already-mutated state on a repeated `tag_id`.
  - Pre-validation loop collects every offender as `tag_id (reason)` before any `_write_tags` — rollback is via `_EarlyReturn` → `rls_connection` context exit.

### 5. Rename (enum alignment only)
- [`airflow/plugins/api/tags.py:282`](../../airflow/plugins/api/tags.py#L282) — `rename_tag`; history `action` flipped from `"renamed"` → `"edited"` while preserving the `old_name` extra field. Duplicate-name check untouched.

### 6. Tests (behavioral proof)
- [`tests/airflow/test_plugins/test_tags.py:178`](../../tests/airflow/test_plugins/test_tags.py#L178) — `test_reopen_verified_to_pending`: verified→pending round-trip emits `action="edited"`.
- [`tests/airflow/test_plugins/test_tags.py:197`](../../tests/airflow/test_plugins/test_tags.py#L197) — `test_same_status_returns_invalid_transition`: 400 + zero Milvus call.
- [`tests/airflow/test_plugins/test_tags.py:225`](../../tests/airflow/test_plugins/test_tags.py#L225) — `test_history_action_maps_to_enum`: parametrized over all 6 live transitions.
- [`tests/airflow/test_plugins/test_tags.py:312`](../../tests/airflow/test_plugins/test_tags.py#L312) — `test_bulk_rejects_all_on_any_invalid_transition`: asserts no `UPDATE data_hierarchical_nodes` SQL fires on partial-invalid bulk.
- [`tests/airflow/test_plugins/test_tags.py:342`](../../tests/airflow/test_plugins/test_tags.py#L342) — `test_bulk_duplicate_tag_ids_rejected`: new case for the guard-bypass fix.
- [`tests/airflow/test_plugins/test_tags.py:441`](../../tests/airflow/test_plugins/test_tags.py#L441) — `test_rename_emits_edited_action`: enum action string on rename.