From 90b1dc7ee3aed5ef187f28355a53ae8195a327c5 Mon Sep 17 00:00:00 2001 From: Jaixii Date: Mon, 20 Jul 2026 22:11:55 -0400 Subject: [PATCH 1/2] chore: restore soft-close edge dedup, drop dead billing dispatch, dedup test fixture MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - service.py merge_workspaces(): replace the hard DELETE on colliding live edges with the bi-temporal soft-close used elsewhere in the codebase (Store._deduplicate_live_edges): merge weight/provenance/valid_from/ ingested_at and live edge_supports onto the surviving edge, then close the losing edge with valid_to+expired_at and a canonical_deduplicated_into provenance marker instead of deleting it. Ported from the validated closed-PR #30 (codex/commercial-v1-ga) implementation. Drops the now-orphaned _dumps import (its only call site was the removed hard-delete path). - tests/test_workspace_ops.py: update test_merge_deduplicates_colliding_live_edges to assert soft-close semantics (losing edge still present with valid_to/expired_at set and the canonical_deduplicated_into marker pointing at the survivor) instead of asserting the row is gone. - billing.py: remove polar_webhook()'s unreachable second refund/revoke dispatch (order.refunded / subscription.revoked / subscription.updated status==revoked) — the _REVOKING_EVENTS guard earlier in the function always returns first, so this code never ran. Also removes the now-fully-unused _revoke_refunded_order, _revoke_subscription_event, _polar_subscription_id and _polar_order_id (confirmed zero other callers/uses repo-wide via grep before deleting). - tests/test_billing.py: remove the duplicate def _registry_rows() (two byte-identical definitions; Python silently kept only the second). Validated: ruff check . clean; pytest tests/ 1605 passed, 14 skipped, 0 failed (identical to unmodified origin/main in this environment, confirming no regressions). --- engraphis/billing.py | 99 ----------------------------- engraphis/service.py | 120 +++++++++++++++++------------------- tests/test_billing.py | 13 ---- tests/test_workspace_ops.py | 14 ++++- 4 files changed, 69 insertions(+), 177 deletions(-) diff --git a/engraphis/billing.py b/engraphis/billing.py index 3c76c52..0c52409 100644 --- a/engraphis/billing.py +++ b/engraphis/billing.py @@ -454,29 +454,6 @@ def _finalize_webhook(delivery_id: str, fulfillment_id: str, finally: conn.close() -def _polar_subscription_id(data: dict, *, object_is_subscription: bool = False) -> str: - """Extract a Polar subscription id from direct or nested event data.""" - from engraphis.inspector.webhooks import _extract_subscription_id - sub_id = _extract_subscription_id(data, object_is_subscription=object_is_subscription) - if sub_id: - return sub_id - order = data.get("order") or {} - if isinstance(order, dict): - return _extract_subscription_id(order) - return "" - - -def _polar_order_id(data: dict) -> str: - """Extract a Polar order id from direct or nested event data.""" - from engraphis.inspector.webhooks import _extract_order_id - order_id = _extract_order_id(data) - if order_id: - return order_id - order = data.get("order") or {} - if isinstance(order, dict): - return _extract_order_id(order) - return "" - def _release_claims(*claim_ids: str) -> None: """Best-effort rollback used only while returning a retryable failure.""" @@ -509,74 +486,6 @@ def _order_id(data: dict) -> str: return "" -def _revoke_refunded_order(data: dict, webhook_id: str) -> JSONResponse: - """Refunds return the money, so revoke the affected key(s) immediately.""" - subscription_id = _polar_subscription_id(data) - order_id = _polar_order_id(data) - if not subscription_id and not order_id: - return JSONResponse({"status": "ignored", "reason": "missing refund target", - "type": "order.refunded"}, status_code=202) - - delivery_claim = "dlv:" + webhook_id - claim_state = claim_webhook(delivery_claim) - if claim_state == "fulfilled": - logger.info("polar webhook: duplicate refund delivery %s ignored", webhook_id) - return JSONResponse({"status": "duplicate", "revoked": 0}, status_code=202) - if claim_state == "in_flight": - return JSONResponse({"status": "processing", "revoked": 0}, status_code=503) - - try: - from engraphis.inspector.license_registry import ( - revoke_by_order, revoke_by_subscription) - if order_id: - revoked = revoke_by_order(order_id) - target = {"order_id": order_id} - else: - revoked = revoke_by_subscription(subscription_id) - target = {"subscription_id": subscription_id} - except Exception: # noqa: BLE001 — force Polar to retry if durable revoke failed - release_webhook(delivery_claim) - logger.exception("polar webhook: refund revocation failed") - return JSONResponse({"error": "revocation failed"}, status_code=503) - - complete_webhook(delivery_claim) - logger.warning("polar webhook: refund revoked %d license key(s) for %s", - revoked, target) - return JSONResponse({"status": "revoked", "reason": "refund", - "revoked": revoked, **target}, status_code=202) - - -def _revoke_subscription_event(data: dict, webhook_id: str, *, - reason: str) -> JSONResponse: - """Definitive subscription revocation: access should end now, not at expiry.""" - subscription_id = _polar_subscription_id(data, object_is_subscription=True) - if not subscription_id: - return JSONResponse({"status": "ignored", "reason": "missing subscription id", - "type": "subscription.revoked"}, status_code=202) - - delivery_claim = "dlv:" + webhook_id - claim_state = claim_webhook(delivery_claim) - if claim_state == "fulfilled": - logger.info("polar webhook: duplicate revocation delivery %s ignored", webhook_id) - return JSONResponse({"status": "duplicate", "revoked": 0}, status_code=202) - if claim_state == "in_flight": - return JSONResponse({"status": "processing", "revoked": 0}, status_code=503) - - try: - from engraphis.inspector.license_registry import revoke_by_subscription - revoked = revoke_by_subscription(subscription_id) - except Exception: # noqa: BLE001 — force Polar to retry if durable revoke failed - release_webhook(delivery_claim) - logger.exception("polar webhook: subscription revocation failed") - return JSONResponse({"error": "revocation failed"}, status_code=503) - - complete_webhook(delivery_claim) - logger.warning("polar webhook: %s revoked %d license key(s) for subscription %s", - reason, revoked, subscription_id) - return JSONResponse({"status": "revoked", "reason": reason, "revoked": revoked, - "subscription_id": subscription_id}, status_code=202) - - def _event_modified_at(data: dict) -> Optional[float]: """Epoch of the event object's own last-modification time, if the payload carries one (Polar sends ``modified_at`` on Subscription objects). Used to reject an @@ -866,14 +775,9 @@ async def polar_webhook(request: Request): # and unrelated updates cannot spam replacement keys. # A non-trial subscription.created is a no-op: its paid key comes from order.paid, so # a canceled trial can never keep Pro — the short trial key just expires. - if event_type == "order.refunded": - return _revoke_refunded_order(data, webhook_id) if event_type in ("subscription.canceled", "subscription.cancelled"): return JSONResponse({"status": "ignored", "reason": "paid period honored", "type": event_type}, status_code=202) - if event_type == "subscription.revoked": - return _revoke_subscription_event(data, webhook_id, - reason="subscription_revoked") pending_seat_baseline = None # (sub_id, seats, event_ts); persisted after key issuance seat_lock_claim = "" if event_type == "order.paid": @@ -898,9 +802,6 @@ async def polar_webhook(request: Request): elif event_type == "subscription.updated": status = event_status sub_id = str(data.get("id") or "").strip()[:128] - if status == "revoked": - return _revoke_subscription_event(data, webhook_id, - reason="subscription_revoked") if status != "active" or not sub_id: return JSONResponse({"status": "ignored", "reason": "not an active " "subscription", "type": event_type}, status_code=202) diff --git a/engraphis/service.py b/engraphis/service.py index 15a3bf2..28dd4ad 100644 --- a/engraphis/service.py +++ b/engraphis/service.py @@ -40,12 +40,7 @@ from engraphis.core.graph_layers import normalize_graph_layer from engraphis.core.ids import new_id as make_id from engraphis.core.interfaces import Edge, GraphLayer, MemoryType, Node, Scope, SearchFilter -from engraphis.core.store import ( - _dumps, - _loads, - _merge_edge_provenance, - normalize_entity_name, -) +from engraphis.core.store import _loads, _merge_edge_provenance, normalize_entity_name from engraphis.graphdata import build_graph_payload, empty_graph # ── validation limits (memory-poisoning / resource-exhaustion guards) ────────── @@ -2312,9 +2307,9 @@ def _new_repo(old_repo_id): ) # 3) Edges: relabel workspace/repo, remapping any entity ids folded in step 2. - # When a remapped source edge would collide with an existing live edge - # on the unique index (workspace_id, [repo_id,] src, dst, relation, layer), - # merge its evidence rows into the survivor and drop the duplicate. + # When a live source edge collides with an existing live target edge (same + # src/dst/relation/layer/repo), merge metadata instead of violating the + # partial unique index — mirrors Store._deduplicate_live_edges(). src_edges = [dict(x) for x in c.execute( "SELECT id, repo_id, src, dst, relation, layer, weight, provenance, " "valid_from, ingested_at, valid_to, expired_at " @@ -2324,66 +2319,67 @@ def _new_repo(old_repo_id): new_dst = entity_remap.get(ed["dst"], ed["dst"]) new_repo = _new_repo(ed["repo_id"]) is_live = ed["valid_to"] is None and ed["expired_at"] is None + target = None if is_live: - if new_repo is None: - collision = c.execute( - "SELECT id, weight, valid_from, ingested_at, provenance FROM edges " - "WHERE workspace_id=? AND repo_id IS NULL " - "AND src=? AND dst=? AND relation=? AND layer=? " + # Check for a live target edge with the same identity. + if new_repo is not None: + target = c.execute( + "SELECT id, weight, provenance, valid_from, ingested_at " + "FROM edges WHERE workspace_id=? AND repo_id=? AND src=? " + "AND dst=? AND relation=? AND layer=? " "AND valid_to IS NULL AND expired_at IS NULL LIMIT 1", - (wid_dst, new_src, new_dst, ed["relation"], ed["layer"]), + (wid_dst, new_repo, new_src, new_dst, + ed["relation"], ed["layer"]), ).fetchone() else: - collision = c.execute( - "SELECT id, weight, valid_from, ingested_at, provenance FROM edges " - "WHERE workspace_id=? AND repo_id=? " + target = c.execute( + "SELECT id, weight, provenance, valid_from, ingested_at " + "FROM edges WHERE workspace_id=? AND repo_id IS NULL " "AND src=? AND dst=? AND relation=? AND layer=? " "AND valid_to IS NULL AND expired_at IS NULL LIMIT 1", - (wid_dst, new_repo, new_src, new_dst, ed["relation"], ed["layer"]), + (wid_dst, new_src, new_dst, + ed["relation"], ed["layer"]), ).fetchone() - if collision: - # Fold the source relation into the surviving live target edge, - # mirroring Store._deduplicate_live_edges(): keep the stronger - # weight and merge provenance so a higher manual weight or unique - # audit context on the source is never silently discarded. - merged_provenance = _merge_edge_provenance( - [_loads(collision["provenance"], {}), - _loads(ed["provenance"], {})], - merged_ids=[ed["id"]], - ) - # Keep the earliest temporal anchor so the surviving relation is - # never reported as newer than it truly is, mirroring - # Store._deduplicate_live_edges(). - valid_values = [ - float(v) for v in (collision["valid_from"], ed["valid_from"]) - if v is not None - ] - ingested_values = [ - float(v) for v in (collision["ingested_at"], ed["ingested_at"]) - if v is not None - ] - c.execute( - "UPDATE edges SET weight=?, valid_from=?, ingested_at=?, " - "provenance=? WHERE id=?", - ( - max(float(collision["weight"] or 0.0), - float(ed["weight"] or 0.0)), - min(valid_values) if valid_values else None, - min(ingested_values) if ingested_values else None, - _dumps(merged_provenance), collision["id"], - ), - ) - # Re-point evidence rows onto the survivor, skipping duplicates. - c.execute( - "UPDATE OR IGNORE edge_supports SET edge_id=? WHERE edge_id=?", - (collision["id"], ed["id"]), - ) - c.execute("DELETE FROM edge_supports WHERE edge_id=?", (ed["id"],)) - c.execute("DELETE FROM edges WHERE id=?", (ed["id"],)) - continue - c.execute( - "UPDATE edges SET workspace_id=?, repo_id=?, src=?, dst=? WHERE id=?", - (wid_dst, new_repo, new_src, new_dst, ed["id"])) + if target: + # Merge: keep target as survivor, retire source edge. + closed_at = time.time() + src_prov = _loads(ed["provenance"], {}) + tgt_prov = _loads(target["provenance"], {}) + merged_prov = _merge_edge_provenance( + [tgt_prov, src_prov], merged_ids=[ed["id"]]) + merged_weight = max( + float(ed["weight"] or 0.0), + float(target["weight"] or 0.0)) + valid_vals = [v for v in (target["valid_from"], ed["valid_from"]) + if v is not None] + ingested_vals = [v for v in + (target["ingested_at"], ed["ingested_at"]) + if v is not None] + c.execute( + "UPDATE edges SET weight=?, provenance=?, " + "valid_from=?, ingested_at=? WHERE id=?", + (merged_weight, + json.dumps(merged_prov, ensure_ascii=False), + min(valid_vals) if valid_vals else None, + min(ingested_vals) if ingested_vals else None, + target["id"])) + # Move live edge_supports from source to target. + c.execute( + "UPDATE edge_supports SET edge_id=? WHERE edge_id=? " + "AND valid_to IS NULL AND expired_at IS NULL", + (target["id"], ed["id"])) + # Bi-temporally close the source edge. + src_prov["canonical_deduplicated_into"] = target["id"] + c.execute( + "UPDATE edges SET valid_to=?, expired_at=?, " + "provenance=? WHERE id=?", + (closed_at, closed_at, + json.dumps(src_prov, ensure_ascii=False), ed["id"])) + else: + c.execute( + "UPDATE edges SET workspace_id=?, repo_id=?, src=?, dst=? " + "WHERE id=?", + (wid_dst, new_repo, new_src, new_dst, ed["id"])) # 4) Memories / sessions / events: relabel workspace/repo per distinct repo_id # bucket (ids, content and history are untouched). diff --git a/tests/test_billing.py b/tests/test_billing.py index 98de675..5d354cc 100644 --- a/tests/test_billing.py +++ b/tests/test_billing.py @@ -748,19 +748,6 @@ def _registry_rows(): conn.close() - -def _registry_rows(): - from engraphis.inspector import license_registry as reg - conn = reg.connect() - try: - rows = conn.execute( - "SELECT key_id, status, subscription_id, order_id FROM issued_licenses " - "ORDER BY created_at").fetchall() - return [dict(row) for row in rows] - finally: - conn.close() - - def test_trial_subscription_issues_short_lived_key(monkeypatch): from engraphis.inspector import webhooks as WH monkeypatch.setenv("ENGRAPHIS_VENDOR_SIGNING_KEY", VENDOR_SEED) diff --git a/tests/test_workspace_ops.py b/tests/test_workspace_ops.py index 9b68f4f..cabae75 100644 --- a/tests/test_workspace_ops.py +++ b/tests/test_workspace_ops.py @@ -327,7 +327,8 @@ def test_merge_deduplicates_colliding_live_edges(): """When both workspaces hold the same live workspace-level relation (X related Y), step 2 folds the source entities onto the target IDs. Without dedup, the edge relabel produces two rows matching the v4 live-edge unique index and the merge - rolls back. The fix merges evidence into the survivor and drops the duplicate.""" + rolls back. The fix merges evidence into the survivor and soft-closes the + duplicate (bi-temporal history preserved, never hard-deleted).""" svc = _svc() c = svc.store.conn svc.create_workspace("a") @@ -371,8 +372,15 @@ def test_merge_deduplicates_colliding_live_edges(): (survivor,))} assert supports == {"mem_a", "mem_b"} - # The duplicate edge row is gone. - assert c.execute("SELECT 1 FROM edges WHERE id='edge_from_a'").fetchone() is None + # The duplicate edge is soft-closed, not deleted: the row survives with + # valid_to/expired_at set and a canonical_deduplicated_into provenance marker. + dup = c.execute( + "SELECT valid_to, expired_at, provenance FROM edges WHERE id='edge_from_a'" + ).fetchone() + assert dup is not None + assert dup["valid_to"] is not None + assert dup["expired_at"] is not None + assert json.loads(dup["provenance"])["canonical_deduplicated_into"] == survivor # ── copy_workspace ─────────────────────────────────────────────────────────── From 581ab4d93984e20e46b54f4a5e477387a54d3dca Mon Sep 17 00:00:00 2001 From: Jaixii Date: Mon, 20 Jul 2026 22:33:45 -0400 Subject: [PATCH 2/2] fix: avoid IntegrityError when moving colliding edge_supports in merge_workspaces The soft-close support move (UPDATE edge_supports SET edge_id=...) was a plain UPDATE. idx_edge_support_live_unique is a partial unique index on (edge_id, memory_id, source_kind) WHERE live, so if the surviving target edge already had a live support for the same memory (same memory_id + source_kind), retargeting the source edge's identical live support onto it violated the index and raised sqlite3.IntegrityError, rolling back the whole merge transaction. Main's prior hard-delete code guarded against this with UPDATE OR IGNORE; the soft-close port dropped that safeguard. Fix: use UPDATE OR IGNORE for the move (matches main's original convention), then soft-close (valid_to+expired_at) whatever live support OR IGNORE left behind on the source edge, so nothing stays live on a now-closed edge -- mirroring how Store._deduplicate_live_edges() closes retired supports. Added test_merge_deduplicates_colliding_live_edges_with_colliding_support to tests/test_workspace_ops.py: both workspaces' colliding edges are backed by the same memory id, so the support move itself collides. Verified this test fails with the exact predicted IntegrityError against the pre-fix code and passes against the fix. --- engraphis/service.py | 16 +++++++-- tests/test_workspace_ops.py | 70 +++++++++++++++++++++++++++++++++++++ 2 files changed, 84 insertions(+), 2 deletions(-) diff --git a/engraphis/service.py b/engraphis/service.py index 28dd4ad..fb58b96 100644 --- a/engraphis/service.py +++ b/engraphis/service.py @@ -2363,11 +2363,23 @@ def _new_repo(old_repo_id): min(valid_vals) if valid_vals else None, min(ingested_vals) if ingested_vals else None, target["id"])) - # Move live edge_supports from source to target. + # Move live edge_supports from source to target, skipping any that + # would collide with an identical live support already on the + # survivor (idx_edge_support_live_unique is a partial unique index + # on (edge_id, memory_id, source_kind) WHERE live) — a plain UPDATE + # would raise IntegrityError and roll back the whole merge. c.execute( - "UPDATE edge_supports SET edge_id=? WHERE edge_id=? " + "UPDATE OR IGNORE edge_supports SET edge_id=? WHERE edge_id=? " "AND valid_to IS NULL AND expired_at IS NULL", (target["id"], ed["id"])) + # Whatever OR IGNORE left behind is still live but attached to an + # edge that's about to close — soft-close it too instead of leaving + # orphaned live evidence on a non-live edge, mirroring how + # Store._deduplicate_live_edges() closes retired supports. + c.execute( + "UPDATE edge_supports SET valid_to=?, expired_at=? " + "WHERE edge_id=? AND valid_to IS NULL AND expired_at IS NULL", + (closed_at, closed_at, ed["id"])) # Bi-temporally close the source edge. src_prov["canonical_deduplicated_into"] = target["id"] c.execute( diff --git a/tests/test_workspace_ops.py b/tests/test_workspace_ops.py index cabae75..b4b6623 100644 --- a/tests/test_workspace_ops.py +++ b/tests/test_workspace_ops.py @@ -383,6 +383,76 @@ def test_merge_deduplicates_colliding_live_edges(): assert json.loads(dup["provenance"])["canonical_deduplicated_into"] == survivor +def test_merge_deduplicates_colliding_live_edges_with_colliding_support(): + """When the two colliding live edges are ALSO supported by the same memory id + (same memory_id + source_kind), moving the source edge's live support onto the + survivor collides on idx_edge_support_live_unique(edge_id, memory_id, source_kind) + -- the survivor already has its own live row for that exact memory. A plain + UPDATE would raise IntegrityError and roll back the whole merge; the fix uses + UPDATE OR IGNORE for the move and then soft-closes whatever support OR IGNORE + left behind, so nothing is left live on a now-closed edge.""" + svc = _svc() + c = svc.store.conn + svc.create_workspace("a") + svc.create_workspace("b") + wid_a = _wsid(svc, "a") + wid_b = _wsid(svc, "b") + svc.store.upsert_entity(Node(id="ent_a_alpha", name="Alpha", ntype="concept", workspace_id=wid_a)) + svc.store.upsert_entity(Node(id="ent_a_beta", name="Beta", ntype="concept", workspace_id=wid_a)) + svc.store.upsert_entity(Node(id="ent_b_alpha", name="Alpha", ntype="concept", workspace_id=wid_b)) + svc.store.upsert_entity(Node(id="ent_b_beta", name="Beta", ntype="concept", workspace_id=wid_b)) + # Both edges are backed by the SAME memory id, with no source/source_kind in + # provenance so both supports land on the default "legacy_unknown" source_kind -- + # the target already holds a live (edge_id, memory_id, source_kind) triple + # identical to what moving the source's support would try to create. + svc.store.upsert_edge(Edge( + id="edge_from_a", src="ent_a_alpha", dst="ent_a_beta", relation="related", + workspace_id=wid_a, + provenance={"memory_id": "mem_shared", "memory_ids": ["mem_shared"]}, + )) + svc.store.upsert_edge(Edge( + id="edge_from_b", src="ent_b_alpha", dst="ent_b_beta", relation="related", + workspace_id=wid_b, + provenance={"memory_id": "mem_shared", "memory_ids": ["mem_shared"]}, + )) + + # Must not raise IntegrityError on idx_edge_support_live_unique. + out = svc.merge_workspaces("a", "b") + assert out["target"] == "b" + + # Exactly one live edge survives, with exactly one live support for the shared + # memory -- not duplicated, not dropped. + live = [dict(r) for r in c.execute( + "SELECT id FROM edges WHERE workspace_id=? AND valid_to IS NULL AND expired_at IS NULL", + (wid_b,))] + assert len(live) == 1 + survivor = live[0]["id"] + live_supports = [r["memory_id"] for r in c.execute( + "SELECT memory_id FROM edge_supports WHERE edge_id=? " + "AND valid_to IS NULL AND expired_at IS NULL", + (survivor,))] + assert live_supports == ["mem_shared"] + + # The retired source edge is soft-closed with the canonical marker, same as the + # non-colliding-support case. + dup = c.execute( + "SELECT valid_to, expired_at, provenance FROM edges WHERE id='edge_from_a'" + ).fetchone() + assert dup is not None + assert dup["valid_to"] is not None + assert dup["expired_at"] is not None + assert json.loads(dup["provenance"])["canonical_deduplicated_into"] == survivor + + # Its own copy of the colliding support -- the one UPDATE OR IGNORE could not + # move -- is soft-closed too, not left live and orphaned on a closed edge. + dup_supports = [dict(r) for r in c.execute( + "SELECT valid_to, expired_at FROM edge_supports WHERE edge_id='edge_from_a'" + )] + assert len(dup_supports) == 1 + assert dup_supports[0]["valid_to"] is not None + assert dup_supports[0]["expired_at"] is not None + + # ── copy_workspace ─────────────────────────────────────────────────────────── def test_copy_auto_names_and_leaves_source_untouched(): svc = _svc()