fix(api): Phase 7b.1 — upsert name-collision + /readiness proxy gap

backend/app/db/lifespan.py

@@ -208,6 +208,61 @@ async def upsert_runtime_printer(
     name: str = f"{model} ({host})"
 
     existing = await session.get(Printer, printer_id)
+    if existing is None:
+        # Phase 7b.1: handle upgrade path — an old Printer row (e.g. from
+        # Phase 7a with a random uuid4 id) may share the same name that the
+        # new deterministic UUIDv5 wants.  Migrate its id in-place via raw
+        # SQL UPDATEs so historical FK references (jobs, printer_state,
+        # printer_status_cache) are rewritten rather than cascade-deleted.
+        import json
+
+        from sqlalchemy import text
+
+        from app.repositories.printers import get_by_name
+
+        existing_by_name = await get_by_name(session, name)
+        if existing_by_name is not None:
+            old_id = existing_by_name.id
+            # SQLite stores UUIDs as 32-char hex strings (no dashes); raw SQL
+            # bind params must match that format for WHERE/SET to match rows.
+            new_hex = str(printer_id).replace("-", "")
+            old_hex = str(old_id).replace("-", "")
+            # Rewrite FK columns in dependent tables first
+            await session.execute(
+                text("UPDATE printer_status_cache SET printer_id = :new WHERE printer_id = :old"),
+                {"new": new_hex, "old": old_hex},
+            )
+            await session.execute(
+                text("UPDATE printer_state SET printer_id = :new WHERE printer_id = :old"),
+                {"new": new_hex, "old": old_hex},
+            )
+            await session.execute(
+                text("UPDATE jobs SET printer_id = :new WHERE printer_id = :old"),
+                {"new": new_hex, "old": old_hex},
+            )
+            # Migrate the printer row's PK in-place (SQLite allows UPDATE on PK)
+            await session.execute(
+                text(
+                    "UPDATE printers"
+                    " SET id = :new, name = :name, connection = :conn,"
+                    " enabled = 1 WHERE id = :old"
+                ),
+                {
+                    "new": new_hex,
+                    "old": old_hex,
+                    "name": name,
+                    "conn": json.dumps(connection),
+                },
+            )
+            await session.flush()
+            # Expunge the stale ORM object and expire the identity map so
+            # raw SQL UPDATEs are visible on the next ORM read.
+            session.expunge(existing_by_name)
+            session.expire_all()
+            # The row now exists with the new id — treat as existing so the
+            # INSERT branch below is skipped.
+            existing = await session.get(Printer, printer_id)
+
     if existing is not None:
         existing.name = name
         existing.connection = connection

backend/tests/integration/db/test_lifespan_printer_upsert.py

@@ -91,3 +91,132 @@ async def test_upsert_returns_none_when_no_env_printer(async_session_empty):
     assert result_id is None
     result = await async_session_empty.execute(select(Printer))
     assert len(list(result.scalars())) == 0
+
+
+async def test_upsert_handles_existing_row_with_same_name_different_id(
+    async_session_empty,
+):
+    """Phase 7b.1 regression test for issue #76:
+    An old Printer row (with a random uuid4 id) from Phase 7a has the
+    same NAME the new deterministic UUIDv5 wants. upsert_runtime_printer
+    must replace the old row, not crash with UNIQUE constraint failed.
+    """
+    from uuid import uuid4
+
+    # Settings configured for PT-P750W on 192.0.2.50:9100 — matches the
+    # deterministic UUIDv5 the test below expects to land in the DB.
+    settings = _settings_with_pt750w()
+    expected_id = derive_printer_id(_PT750W_MODEL, _PT750W_HOST, _PT750W_PORT)
+
+    # Seed the DB with the SAME name but a different (random) id, mimicking
+    # the Phase 7a row that triggered the production crash.
+    old_id = uuid4()
+    assert old_id != expected_id
+    async_session_empty.add(
+        Printer(
+            id=old_id,
+            name=f"{_PT750W_MODEL} ({_PT750W_HOST})",  # same name upsert_runtime_printer computes
+            model="pt-p750w",
+            backend="ptouch",
+            connection={"host": _PT750W_HOST, "port": _PT750W_PORT},
+            enabled=True,
+        )
+    )
+    await async_session_empty.flush()
+
+    # Now call upsert — it must NOT raise IntegrityError
+    returned_id = await upsert_runtime_printer(async_session_empty, settings)
+
+    assert returned_id == expected_id
+
+    # Exactly one row should remain — the new one with the deterministic id
+    result = await async_session_empty.execute(select(Printer))
+    rows = list(result.scalars())
+    assert len(rows) == 1
+    assert rows[0].id == expected_id
+    assert rows[0].id != old_id
+
+
+async def test_upsert_preserves_dependent_rows_during_id_migration(
+    async_session_empty,
+):
+    """Phase 7b.1 round 2: when a name-collision triggers id-migration, ALL
+    dependent rows (jobs, printer_state, printer_status_cache) must have
+    their FK references rewritten to the new deterministic UUIDv5 — not
+    cascade-deleted (which would lose historical print jobs)."""
+    from datetime import UTC, datetime
+    from uuid import uuid4
+
+    from app.models.job import Job, JobState
+    from app.models.printer import Printer
+    from app.models.printer_state import PrinterState
+    from app.models.printer_status_cache import PrinterStatusCache
+
+    settings = _settings_with_pt750w()
+    expected_id = derive_printer_id(_PT750W_MODEL, _PT750W_HOST, _PT750W_PORT)
+
+    # Seed: old Printer row with random uuid4 + dependent rows
+    old_id = uuid4()
+    assert old_id != expected_id
+    async_session_empty.add(
+        Printer(
+            id=old_id,
+            name=f"{_PT750W_MODEL} ({_PT750W_HOST})",
+            model="pt-p750w",
+            backend="ptouch",
+            connection={"host": _PT750W_HOST, "port": _PT750W_PORT},
+            enabled=True,
+        )
+    )
+    await async_session_empty.flush()
+
+    # Dependent rows that would trigger FOREIGN KEY constraint failure on DELETE
+    async_session_empty.add(
+        PrinterState(
+            printer_id=old_id,
+            paused=False,
+            updated_at=datetime.now(UTC),
+        )
+    )
+    async_session_empty.add(
+        PrinterStatusCache(
+            printer_id=old_id,
+            captured_at=datetime.now(UTC),
+            parsed={"online": False, "last_error": "stale data"},
+        )
+    )
+    async_session_empty.add(
+        Job(
+            id=uuid4(),
+            printer_id=old_id,
+            template_key="label/address",
+            state=JobState.DONE.value,
+        )
+    )
+    await async_session_empty.flush()
+
+    # NOW call upsert — must NOT raise IntegrityError, and must preserve dependent rows
+    returned_id = await upsert_runtime_printer(async_session_empty, settings)
+    assert returned_id == expected_id
+
+    # The Printer row migrated id (old row gone, new row exists with new id)
+    result = await async_session_empty.execute(select(Printer))
+    rows = list(result.scalars())
+    assert len(rows) == 1
+    assert rows[0].id == expected_id
+
+    # Dependent rows have FK rewritten to new id (not deleted)
+    result = await async_session_empty.execute(select(PrinterState))
+    states = list(result.scalars())
+    assert len(states) == 1
+    assert states[0].printer_id == expected_id
+
+    result = await async_session_empty.execute(select(PrinterStatusCache))
+    caches = list(result.scalars())
+    assert len(caches) == 1
+    assert caches[0].printer_id == expected_id
+
+    result = await async_session_empty.execute(select(Job))
+    jobs = list(result.scalars())
+    assert len(jobs) == 1
+    assert jobs[0].printer_id == expected_id

frontend/cmd/server/main.go

@@ -149,6 +149,7 @@ func newRouter(ph *handlers.PageHandler, prx http.Handler, staticSubFS fs.FS) *c
 	r.Handle("/docs", prx)
 	r.Handle("/openapi.json", prx)
 	r.Handle("/redoc", prx)
+	r.Handle("/readiness", prx)
 
 	return r
 }

frontend/cmd/server/main_test.go

@@ -336,6 +336,9 @@ func TestProxyMountsBackendDocRoutes(t *testing.T) {
 		case "/redoc":
 			w.Header().Set("Content-Type", "text/html")
 			fmt.Fprint(w, "ReDoc")
+		case "/readiness":
+			w.Header().Set("Content-Type", "application/json")
+			fmt.Fprint(w, `{"status":"ready","checks":{}}`)
 		default:
 			http.NotFound(w, r)
 		}
@@ -359,6 +362,7 @@ func TestProxyMountsBackendDocRoutes(t *testing.T) {
 		"/docs":         "Swagger UI",
 		"/openapi.json": `"openapi":"3.1.0"`,
 		"/redoc":        "ReDoc",
+		"/readiness":    `"status":"ready"`,
 	} {
 		path, want := path, want // capture loop variables
 		t.Run(path, func(t *testing.T) {