feat: add Object.create_if_absent atomic insert

CHANGELOG.md

@@ -7,6 +7,19 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased]
 
+## [0.0.20] - 2026-09-18
+
+### Added
+
+- **`Database.insert_if_absent` / `Object.create_if_absent`.** Atomic
+  insert-if-absent on primary key `id` across SQLite (`INSERT OR IGNORE`),
+  Postgres (`ON CONFLICT DO NOTHING RETURNING`), MongoDB (`insert_one` +
+  `DuplicateKeyError`), JsonDB (path lock), and DynamoDB
+  (`attribute_not_exists(id)`). Returns `InsertIfAbsentResult(record, created)`
+  / `(entity, created)` without calling through upsert `save()`. Wrappers
+  (`CachingDatabase`, `ObservableDatabase`) forward correctly.
+  `DeferredSaveMixin.flush()` runs only when `created=True`.
+
 ## [0.0.19] - 2026-09-14
 
 ### Removed

SPEC.md

@@ -102,7 +102,7 @@ AttributeMixin + pydantic.BaseModel
 
 `jvspatial/core/entities/object.py`. Provides:
 - `id`, `entity`, `type_code` fields
-- CRUD methods: `create()`, `get()`, `find()`, `save()`, `delete()`, `count()`, `export()`
+- CRUD methods: `create()`, `create_if_absent()`, `get()`, `find()`, `save()`, `delete()`, `count()`, `export()`
 - Context lookup: `set_context()`, `get_context()` (default via `get_default_context()`)
 - Collection mapping via `get_collection_name()` → `{n: node, e: edge, o: object, w: walker}`
 
@@ -211,6 +211,7 @@ When `DeferredSaveMixin` is mixed into an entity *and* `deferred_saves_globally_
 | Method | Required | Description |
 |---|---|---|
 | `save(collection, data)` | Yes | Insert-or-replace by ID; returns saved record |
+| `insert_if_absent(collection, data, *, conflict_target="id")` | Yes (built-ins) | Atomic insert-if-absent on primary key `id` only (v1); returns `InsertIfAbsentResult(record, created)`. Never updates. Default ABC raises `NotImplementedError` after validating args. |
 | `get(collection, id)` | Yes | Fetch by ID or `None` |
 | `delete(collection, id)` | Yes | Idempotent delete by ID |
 | `find(collection, query, *, limit, sort)` | Yes | Mongo-style query; returns list. Ordering contract below |

docs/md/entity-reference.md

@@ -21,6 +21,8 @@ class Object(BaseModel):
     @classmethod
     async def create(cls, **kwargs) -> "Object"
     @classmethod
+    async def create_if_absent(cls, **kwargs) -> tuple["Object", bool]
+    @classmethod
     async def find(cls, query: Optional[dict] = None, **filters) -> List["Object"]
     @classmethod
     async def find_one(cls, query: Optional[dict] = None, **filters) -> Optional["Object"]
@@ -30,6 +32,12 @@ class Object(BaseModel):
     async def export() -> dict
 ```
 
+**`create_if_absent`** — persist only when no row exists for the given `id`.
+Returns `(entity, created)`. On conflict, rehydrates the **stored** winner
+(never merges proposed fields). Does not call through upsert `save()`.
+Requires a deterministic `id` for idempotency; auto-generated ids always
+create. See `Database.insert_if_absent` / `InsertIfAbsentResult`.
+
 **Convenience Methods**
 
 - `await Object.count()` → count all objects of that type

docs/md/stability.md

@@ -17,15 +17,18 @@ bump (post-1.0) or a minor bump (pre-1.0) and must be called out in
 These names are exported from `jvspatial/__init__.py`'s `__all__` and
 are the canonical import path:
 
-- **Core entities.** `Object`, `Node`, `Edge`, `Walker`, `Root`,
-  `GraphContext`.
+- **Core entities.** `Object` (including `Object.create_if_absent`),
+  `Node`, `Edge`, `Walker`, `Root`, `GraphContext`.
 - **Decorators.** `attribute`, `endpoint`.
 - **Server / config.** `Server`, `ServerConfig`.
-- **Database.** `Database`, `create_database`. The
+- **Database.** `Database`, `create_database`, `InsertIfAbsentResult`. The
   `Database.supports_transactions` capability flag is part of this
   surface, as are the bulk methods `Database.find_many` and
-  `Database.bulk_save`. Adapters not overriding the bulk methods
-  fall through to the (slower) default serial implementations.
+  `Database.bulk_save`, and `Database.insert_if_absent` (v1:
+  `conflict_target="id"` only). Adapters not overriding the bulk methods
+  fall through to the (slower) default serial implementations. Custom
+  adapters must implement `insert_if_absent` or callers hit
+  `NotImplementedError`.
 - **Cache.** `create_cache`.
 - **Mixins.** `DeferredSaveMixin`, `deferred_saves_globally_allowed`,
   `flush_deferred_entities`.

docs/superpowers/plans/2026-09-18-create-if-absent.md

@@ -0,0 +1,48 @@
+# Object.create_if_absent Implementation Plan
+
+> **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking.
+
+**Goal:** Ship portable atomic `Database.insert_if_absent` + `Object.create_if_absent` without changing upsert `save()`.
+
+**Architecture:** New ABC result type + method; each adapter implements insert-or-return-existing on primary `id`; Object layer exports, calls DB, rehydrates on conflict. Never route through `save()`.
+
+**Tech Stack:** Python 3.9+, aiosqlite, asyncpg, motor, pytest-asyncio.
+
+**Spec:** `docs/superpowers/specs/2026-09-18-create-if-absent-design.md`
+
+---
+
+### Task 1: ABC + SQLite + Object + tests
+
+**Files:**
+- Modify: `jvspatial/db/database.py`
+- Modify: `jvspatial/db/sqlite.py`
+- Modify: `jvspatial/core/entities/object.py`
+- Modify: `jvspatial/core/context.py` (if needed for cache)
+- Create: `tests/core/test_object_create_if_absent.py`
+- Create: `tests/db/test_sqlite_insert_if_absent.py`
+
+- [ ] RED/GREEN Object + SQLite concurrent same-id
+- [ ] Commit
+
+### Task 2: Postgres, Mongo, JsonDB, DynamoDB, wrappers
+
+**Files:**
+- Modify: `jvspatial/db/postgres.py` (+ transaction save path if separate)
+- Modify: `jvspatial/db/mongodb.py`
+- Modify: `jvspatial/db/jsondb.py`
+- Modify: `jvspatial/db/dynamodb.py` (if present)
+- Modify: `jvspatial/db/_cache.py`, `jvspatial/db/_observable.py`
+- Extend integration tests
+
+- [ ] RED/GREEN per adapter
+- [ ] Commit
+
+### Task 3: Docs, version, changelog, stability, PR
+
+**Files:**
+- Modify: `CHANGELOG.md`, `jvspatial/version.py`, `docs/md/stability.md`, `docs/md/entity-reference.md`, `SPEC.md` if needed
+- Export `__all__`
+
+- [ ] Docs + 0.0.20 bump
+- [ ] Push + `gh pr create`

docs/superpowers/specs/2026-09-18-create-if-absent-design.md

@@ -0,0 +1,134 @@
+# Design: `Object.create_if_absent` / `Database.insert_if_absent`
+
+**Date:** 2026-09-18
+**Status:** Proposed
+**Target version:** 0.0.20 (minor, pre-1.0)
+**Motivation:** Durable idempotency records need atomic create-or-return-existing semantics. Today every `Database.save` path is upsert (`INSERT OR REPLACE` / `ON CONFLICT DO UPDATE` / Mongo `replace_one(upsert=True)`), so concurrent writers can overwrite peers and SQLite unique collisions can delete the wrong row via `OR REPLACE`.
+
+## Goal
+
+Ship a portable, atomic **insert-if-absent** primitive:
+
+1. Persist a new record only when no conflict exists on the target key.
+2. Never update or replace an existing row.
+3. Return the **stored** winner plus a boolean `created` flag.
+4. Work identically (semantic contract) across SQLite, Postgres, MongoDB, JsonDB, and DynamoDB.
+
+Integral Core will use this for `QueryResultSet` and similar receipt/idempotency Objects after jvspatial ships.
+
+## Non-goals
+
+- Changing `save()` / `create()` upsert semantics.
+- Business-key `ON CONFLICT` on arbitrary unique indexes in v1 (phase 2).
+- Automatic migration of webhook `get_or_create_idempotency_key` (follow-up once unique index exists).
+- Cross-document multi-key transactions on backends without transactions.
+
+## Public contract
+
+### `InsertIfAbsentResult`
+
+```python
+@dataclass(frozen=True)
+class InsertIfAbsentResult:
+    record: Dict[str, Any]  # stored document (existing or newly inserted)
+    created: bool
+```
+
+### `Database.insert_if_absent`
+
+```python
+async def insert_if_absent(
+    self,
+    collection: str,
+    data: Dict[str, Any],
+    *,
+    conflict_target: str = "id",
+) -> InsertIfAbsentResult:
+    ...
+```
+
+**v1 rules:**
+
+- `conflict_target` must be `"id"` (primary key). Other values → `ValueError`.
+- Missing / empty `id` in `data` → `ValueError` (same as existing Postgres payload split).
+- On insert success: `created=True`, return the inserted record (normalized as other writes).
+- On conflict: `created=False`, **load and return the existing stored record unchanged** (no merge of proposed fields).
+- Must not call through `save()`.
+
+### `Object.create_if_absent`
+
+```python
+@classmethod
+async def create_if_absent(cls, **kwargs) -> tuple["Object", bool]:
+    """Create and persist only if absent.
+
+    Idempotent when the caller supplies a deterministic ``id``.
+    Unlike ``create()`` / ``save()``, never updates an existing row.
+    """
+```
+
+**Flow:**
+
+1. `obj = cls(**kwargs)` (auto-id if omitted — same as `create`).
+2. `await context.ensure_indexes(cls)`.
+3. Export record; `await database.insert_if_absent(collection, record)`.
+4. If `created=False`, rehydrate entity from stored record (not the proposed in-memory instance).
+5. Attach `_graph_context`; update entity cache consistently with `save`.
+6. If `DeferredSaveMixin` and `created`, `await flush()` (mirror `create()`).
+7. Return `(entity, created)`.
+
+Optional kwargs **not** in v1: `raise_on_conflict`. Callers inspect `created`.
+
+## Adapter semantics
+
+| Backend | Insert path | Conflict path |
+|---------|-------------|---------------|
+| **Postgres** | `INSERT ... ON CONFLICT (id) DO NOTHING RETURNING data`; if no row returned, `SELECT` by id | Return stored row, `created=False` |
+| **SQLite** | `INSERT OR IGNORE INTO records ...`; if `changes()==0`, `SELECT` | **Never** `INSERT OR REPLACE` |
+| **MongoDB** | `insert_one`; catch `DuplicateKeyError` → `find_one` | Return stored doc |
+| **JsonDB** | Under path lock: if file exists → read; else write | Same-id atomic via lock; no secondary unique enforcement |
+| **DynamoDB** | `PutItem` with `attribute_not_exists(id)` | Conditional check fail → `GetItem` |
+
+Wrappers (`CachingDatabase`, `ObservableDatabase`) must override or forward `insert_if_absent` so cache/metrics stay correct on both created and existing paths.
+
+Default ABC implementation: raise `NotImplementedError` with a clear message, **or** a documented non-atomic find-then-insert only if PRD requires a default — prefer requiring every built-in adapter to implement (PRD §8).
+
+## Error handling
+
+- `ValueError` — missing id, unsupported `conflict_target`.
+- `DuplicateEntityError` — **not** raised by default in v1 (reserved for optional `raise_on_conflict=True` later).
+- Adapter-specific DB errors propagate as today (`DatabaseError` subclasses where applicable).
+
+## Concurrency guarantees
+
+- **Same deterministic `id`:** at most one insert wins; losers observe `created=False` and the winner's stored payload.
+- **Different ids, same business key:** not atomic in v1. Callers that need business-key uniqueness must declare a unique index **and** wait for phase 2 `conflict_target` extension, or encode the business key into a deterministic `id` (Integral pattern).
+
+## Testing requirements
+
+- Unit/integration: create → absent inserts; second call same id returns existing, no field overwrite.
+- Concurrent: N parallel `create_if_absent` with same id → exactly one `created=True`, all return identical stored payload.
+- Regression: `save()` remains upsert (`test_save_is_upsert` unchanged).
+- SQLite: prove unique secondary collision via this API does **not** delete rows (contrast `OR REPLACE` hazard).
+- Object-layer: rehydrate on miss; DeferredSaveMixin flush only when created.
+- Adapters covered: SQLite (always), Postgres (CI postgres job), MongoDB/JsonDB as existing suite patterns allow.
+
+## Docs / stability / release
+
+- Document in `docs/md/entity-reference.md`, `SPEC.md` persistence section, `docs/md/stability.md` (public: `Object.create_if_absent`, `Database.insert_if_absent`, `InsertIfAbsentResult`).
+- `CHANGELOG.md` under `[Unreleased]` → shipped as `[0.0.20]`.
+- Bump `jvspatial/version.py` to `0.0.20` when merging to main per VERSIONING workflow.
+- Export new names from public `__all__` where appropriate.
+
+## Phase 2 (out of this PR)
+
+- `conflict_target` naming unique indexes / compound keys from `get_indexes()`.
+- Migrate webhook idempotency helper off find-then-save.
+- Optional `raise_on_conflict: bool = False`.
+
+## Success criteria
+
+1. Concurrent identical-id creates never overwrite stored data on SQLite and Postgres.
+2. `save()` behavior unchanged.
+3. Public API documented and exported.
+4. Integral can replace race-prone QueryResultSet create with `create_if_absent` in a follow-up bump.

jvspatial/__init__.py

@@ -74,7 +74,7 @@
 )
 
 # Simplified database and cache
-from .db import Database, create_database
+from .db import Database, InsertIfAbsentResult, create_database
 from .db.work_claim import claim_record, delete_claimed_record, release_claim
 
 # Observability primitives
@@ -140,6 +140,7 @@
     "Server",
     # Database & Cache
     "Database",
+    "InsertIfAbsentResult",
     "create_database",
     "create_cache",
     # Observability

jvspatial/core/entities/object.py

@@ -191,6 +191,109 @@ async def create(cls: Type["Object"], **kwargs: Any) -> "Object":
             await flush_fn()
         return obj
 
+    @classmethod
+    async def create_if_absent(
+        cls: Type["Object"], **kwargs: Any
+    ) -> tuple["Object", bool]:
+        """Create and persist only if absent for the given ``id``.
+
+        Idempotent when the caller supplies a deterministic ``id``.
+        Unlike :meth:`create` / :meth:`save`, never updates an existing row.
+        Returns ``(entity, created)`` where ``entity`` is the stored winner
+        (rehydrated from the database when ``created`` is ``False``).
+
+        Does not call through :meth:`save`. For
+        :class:`~jvspatial.core.mixins.DeferredSaveMixin` types, ``flush()``
+        runs only when ``created`` is ``True`` (mirrors :meth:`create`'s
+        end-of-batch clear when the insert won).
+        """
+        obj = cls(**kwargs)
+        context = await obj.get_context()
+        await context.ensure_indexes(cls)
+
+        # Build the persistence record the same way GraphContext.save does,
+        # but route through insert_if_absent instead of save.
+        if hasattr(obj, "type_code"):
+            type_code = getattr(obj, "type_code", "")
+            if type_code in ("n", "e", "w"):
+                record = await obj.export()
+                if "entity" not in record:
+                    record["entity"] = obj.entity
+            else:
+                record = await obj.export()
+                entity_id = getattr(obj, "id", None)
+                if entity_id:
+                    id_parts = entity_id.split(".")
+                    entity_name_resolver = getattr(cls, "_entity_name", None)
+                    expected_entity_name = (
+                        entity_name_resolver()
+                        if callable(entity_name_resolver)
+                        else cls.__name__
+                    )
+                    if (
+                        len(id_parts) != 3
+                        or id_parts[0] != obj.type_code
+                        or id_parts[1] != expected_entity_name
+                    ):
+                        new_id = generate_id(obj.type_code, expected_entity_name)
+                        object.__setattr__(obj, "id", new_id)
+                        record["id"] = new_id
+        else:
+            record = await obj.export()
+            from jvspatial.utils.serialization import serialize_datetime
+
+            record = serialize_datetime(record)
+            if "entity" not in record and hasattr(obj, "entity"):
+                record["entity"] = obj.entity
+
+        from jvspatial.utils.normalization import (
+            is_text_normalization_enabled,
+            normalize_data,
+        )
+
+        if is_text_normalization_enabled():
+            record = normalize_data(record)
+
+        if hasattr(obj, "get_collection_name"):
+            collection = obj.get_collection_name()
+        else:
+            type_code = obj.type_code
+            if type_code == "n":
+                collection = "node"
+            elif type_code == "e":
+                collection = "edge"
+            else:
+                collection = type_code.lower()
+
+        is_node = hasattr(obj, "type_code") and getattr(obj, "type_code", "") == "n"
+        if is_node:
+            record.pop("edges", None)
+
+        result = await context.database.insert_if_absent(collection, record)
+
+        if result.created:
+            entity: Object = obj
+            await context._add_to_cache(entity.id, entity)
+            flush_fn = getattr(entity, "flush", None)
+            if (
+                flush_fn is not None
+                and callable(flush_fn)
+                and inspect.iscoroutinefunction(flush_fn)
+            ):
+                await flush_fn()
+            return entity, True
+
+        entity_opt = await context._deserialize_entity(cls, result.record)
+        if entity_opt is None:
+            raise RuntimeError(
+                f"create_if_absent could not rehydrate {cls.__name__} "
+                f"from stored record id={result.record.get('id')!r}"
+            )
+        entity = entity_opt
+        entity._graph_context = context
+        await context._add_to_cache(entity.id, entity)
+        return entity, False
+
     async def update(
         self: "Object",
         properties: Dict[str, Any],