Skip to content

chore(ga): restore merge soft-close, drop dead refund path, dedup tests #34

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
Coding-Dev-Tools merged 2 commits into from
Jul 21, 2026
Merged

chore(ga): restore merge soft-close, drop dead refund path, dedup tests #34

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
90b1dc7
chore: restore soft-close edge dedup, drop dead billing dispatch, ded…
Coding-Dev-Tools Jul 21, 2026
581ab4d
fix: avoid IntegrityError when moving colliding edge_supports in merg…
Coding-Dev-Tools Jul 21, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
99 changes: 0 additions & 99 deletions engraphis/billing.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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."""
Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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":
Expand All @@ -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)
Expand Down
132 changes: 70 additions & 62 deletions engraphis/service.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) ──────────
Expand Down Expand Up @@ -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 "
Expand All @@ -2324,66 +2319,79 @@ 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, 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 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(
"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).
Expand Down
13 changes: 0 additions & 13 deletions tests/test_billing.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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)
Expand Down
Loading