Route Python SQL writes through context-owned Internal ID allocator

README.md

@@ -265,6 +265,26 @@ repository's local runtime. No external database server or generator is needed
 to run the retained generated example. Use only the example's test database;
 reset/repeated-run behavior is documented in its README.
 
+## Customize internal ID allocation
+
+By default, the SQL provider uses its own durable per-entity ID space. To use
+an application allocator, install an asynchronous `InternalIdAllocator` on the
+Context before saving entities:
+
+```python
+context.with_internal_id_allocator(my_allocator)
+```
+
+The allocator implements `async next_id(entity: str) -> int` and
+`async ensure_id_floor(entity: str, floor: int) -> None`. The latter reserves
+explicitly supplied IDs so future automatic IDs cannot collide. Direct INSERT,
+nested batch and graph Save use this one Context capability. An installed
+allocator failure stops the write; it never silently falls back to the SQL
+provider. Implementations shared across processes must provide durable,
+atomic allocation and floor advancement. See the runnable
+[Internal ID example](examples/internal-id/README.md). The old synchronous
+`generate_id`/`next_id` context calls are unsupported; use the async capability.
+
 ## Optional standard OpenTelemetry module
 
 The local standard-module changes are not yet released. Install the optional

examples/internal-id/README.md

@@ -0,0 +1,15 @@
+# Context-owned Internal ID example
+
+This SQLite example installs a typed, asynchronous `InternalIdAllocator` on
+`UserContext`. Direct, nested-batch, and graph saves use the same allocator;
+explicit IDs advance its floor. An allocator failure rejects the write rather
+than silently falling back to the provider's ID space.
+
+```bash
+PYTHONPATH=src python examples/internal-id/main.py
+```
+
+The in-memory allocator here is for one-process demonstration only. Production
+allocators shared by multiple processes must reserve IDs and explicit-ID floors
+durably and atomically. Gaps after failed operations are acceptable; reuse is
+not.

examples/internal-id/main.py

@@ -0,0 +1,83 @@
+"""Run a small end-to-end Internal ID capability example."""
+
+import asyncio
+from pathlib import Path
+from tempfile import TemporaryDirectory
+
+import aiosqlite
+
+from teaql.core.entity import EntityKey
+from teaql.core.meta import EntityDescriptor, PropertyDescriptor
+from teaql.core.mutation import InsertCommand, MutationRequest
+from teaql.core.value import DataType
+from teaql.provider.sqlite import SimpleSchemaProvider, create_sqlite_service
+from teaql.runtime import RuntimeModule
+
+
+class ExampleAllocator:
+    def __init__(self):
+        self._level = 99
+        self._lock = asyncio.Lock()
+
+    async def next_id(self, entity: str) -> int:
+        assert entity == "School"
+        async with self._lock:
+            self._level += 1
+            return self._level
+
+    async def ensure_id_floor(self, entity: str, floor: int) -> None:
+        assert entity == "School"
+        async with self._lock:
+            self._level = max(self._level, floor)
+
+
+def school_insert(name: str, identity: int | None = None) -> MutationRequest:
+    command = InsertCommand.new("School").value("name", name)
+    if identity is not None:
+        command.value("id", identity)
+    return MutationRequest(command, comment="create school with custom internal ID")
+
+
+async def run() -> None:
+    with TemporaryDirectory() as directory:
+        path = Path(directory) / "internal-id.db"
+        school = (EntityDescriptor("School").table_name("school_data")
+                  .property(PropertyDescriptor("id", DataType.I64).is_id())
+                  .property(PropertyDescriptor("version", DataType.I64).is_version())
+                  .property(PropertyDescriptor("name", DataType.Text)))
+        provider = SimpleSchemaProvider()
+        provider.register_entity(school)
+        service = create_sqlite_service(str(path), provider)
+        context = (RuntimeModule.new().entity(school).into_context()
+                   .with_schema_provider(service)
+                   .insert_resource("dataService", service)
+                   .with_internal_id_allocator(ExampleAllocator()))
+        context.disable_sql_log()
+        await context.ensure_schema()
+
+        await service.mutate(context, school_insert("Direct"))
+        await service.mutate(context, MutationRequest.Batch([
+            school_insert("Batch"),
+            MutationRequest.Batch([school_insert("Nested")], comment="nested schools"),
+        ], comment="batch schools"))
+        await service.mutate(context, school_insert("Explicit", 500))
+
+        async def save_graph(graph):
+            scope = graph.scope(EntityKey("School", 501))
+            return await graph.transaction.mutate(
+                graph.context,
+                graph.request(InsertCommand.new("School").value("name", "Graph"), scope),
+            )
+
+        await context.execute_graph_save(save_graph, comment="save school graph")
+        async with aiosqlite.connect(path) as db:
+            async with db.execute("SELECT id FROM school_data ORDER BY id") as rows:
+                assert [row[0] for row in await rows.fetchall()] == [100, 101, 102, 500, 501]
+            async with db.execute("SELECT COUNT(*) FROM teaql_id_space") as rows:
+                assert (await rows.fetchone())[0] == 0
+        await service.close()
+    print("PASS Python context-owned Internal ID example")
+
+
+if __name__ == "__main__":
+    asyncio.run(run())

scripts/verify-examples.sh

@@ -2,7 +2,7 @@
 set -euo pipefail
 
 repo="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
-expected=(business-clock business-id conformance facet-trace mutation-policy object-storage opaque-entity-reference order-management query-policy school-management task_board trace-chain)
+expected=(business-clock business-id conformance facet-trace internal-id mutation-policy object-storage opaque-entity-reference order-management query-policy school-management task_board trace-chain)
 mapfile -t actual < <(find "$repo/examples" -mindepth 1 -maxdepth 1 -type d -printf '%f\n' | sort)
 if [[ "${actual[*]}" != "${expected[*]}" ]]; then
   echo "example inventory changed; update scripts/verify-examples.sh: ${actual[*]}" >&2
@@ -29,6 +29,7 @@ PYTHONPATH="$repo/src" python -m unittest discover -s "$repo/examples/school-man
 PYTHONPATH="$repo/src" python "$repo/examples/mutation-policy/main.py"
 PYTHONPATH="$repo/src" python "$repo/examples/business-clock/main.py"
 PYTHONPATH="$repo/src" python "$repo/examples/business-id/main.py"
+PYTHONPATH="$repo/src" python "$repo/examples/internal-id/main.py"
 PYTHONPATH="$repo/src" python "$repo/examples/query-policy/main.py"
 PYTHONPATH="$repo/src" python "$repo/examples/opaque-entity-reference/main.py"
 order_management_tmp="$(mktemp -d)"

src/teaql/runtime/__init__.py

@@ -12,6 +12,7 @@
 from .store import DataStore
 from .audit import RawAuditEvent, SafeAuditEvent, MutationAuditKind
 from .business_clock import BusinessClock, FixedBusinessClock, SystemBusinessClock
+from .internal_id import InternalIdAllocator
 from .entity_reference import (
     AeadEntityReferenceCodec,
     EntityReferenceClaims,
@@ -57,7 +58,7 @@
 from .wire_fields import NormalizedWireInput, WireEntityMetadata, WireFieldMetadata, WireInputError, create_wire_entity_metadata, encode_wire_output, normalize_wire_input, retain_submitted_paths
 from teaql.core.entity import EntityKey, EntityChangeSet, EntityRoot
 
-__all__ = ["WireFieldMetadata", "WireEntityMetadata", "NormalizedWireInput", "WireInputError", "create_wire_entity_metadata", "normalize_wire_input", "encode_wire_output", "retain_submitted_paths", "EntityKey", "EntityChangeSet", "EntityRoot", "ContextEntityRef", "ContextRootError", "CheckException", "CheckResult", "I18nCatalog", "JsonFieldNamingProfile", "Locale", "ObjectLocation", "UnsupportedLocaleError", "UserContext", "TeaqlRuntime", "SqlLogEntry", "SqlLogOperation", "DiagnosticSqlLogSink", "TextDiagnosticSqlLogSink", "ServiceRuntimeFromEnv", "RuntimeModule", "DataStore", "RawAuditEvent", "SafeAuditEvent", "MutationAuditKind", "BusinessClock", "FixedBusinessClock", "SystemBusinessClock", "AeadEntityReferenceCodec", "EntityReferenceClaims", "EntityReferenceCodec", "EntityReferenceTokenError", "ENTITY_REFERENCE_AAD", "UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT", "UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT", "BUSINESS_ID_PERMUTATION_V1_ALPHABET", "BUSINESS_ID_PERMUTATION_V1_DOMAIN_SIZE", "BUSINESS_ID_PERMUTATION_V1_MAX_SEQUENCE", "BUSINESS_ID_PERMUTATION_V1_WIDTH", "DefaultBusinessIdProfileFactory", "DefaultBusinessIdService", "InMemoryBusinessIdAllocator", "PermutedDailyBusinessIdProfile", "StaticBusinessIdKeyProvider", "encode_business_id_permutation_v1", "ContextTools", "ExecutableHttpTool", "HTTP_TOOL", "HttpIntentPhase", "HttpTool", "HttpToolProvider", "ToolDeniedError", "ToolError", "ToolPolicy", "ToolRisk", "Tools", "ToolToken", "ToolUnavailableError"]
+__all__ = ["WireFieldMetadata", "WireEntityMetadata", "NormalizedWireInput", "WireInputError", "create_wire_entity_metadata", "normalize_wire_input", "encode_wire_output", "retain_submitted_paths", "EntityKey", "EntityChangeSet", "EntityRoot", "ContextEntityRef", "ContextRootError", "CheckException", "CheckResult", "I18nCatalog", "JsonFieldNamingProfile", "Locale", "ObjectLocation", "UnsupportedLocaleError", "UserContext", "TeaqlRuntime", "SqlLogEntry", "SqlLogOperation", "DiagnosticSqlLogSink", "TextDiagnosticSqlLogSink", "ServiceRuntimeFromEnv", "RuntimeModule", "DataStore", "RawAuditEvent", "SafeAuditEvent", "MutationAuditKind", "BusinessClock", "FixedBusinessClock", "SystemBusinessClock", "InternalIdAllocator", "AeadEntityReferenceCodec", "EntityReferenceClaims", "EntityReferenceCodec", "EntityReferenceTokenError", "ENTITY_REFERENCE_AAD", "UNSAFE_RAW_ENTITY_REFERENCES_ACKNOWLEDGEMENT", "UNSAFE_RAW_ENTITY_REFERENCES_ENVIRONMENT", "BUSINESS_ID_PERMUTATION_V1_ALPHABET", "BUSINESS_ID_PERMUTATION_V1_DOMAIN_SIZE", "BUSINESS_ID_PERMUTATION_V1_MAX_SEQUENCE", "BUSINESS_ID_PERMUTATION_V1_WIDTH", "DefaultBusinessIdProfileFactory", "DefaultBusinessIdService", "InMemoryBusinessIdAllocator", "PermutedDailyBusinessIdProfile", "StaticBusinessIdKeyProvider", "encode_business_id_permutation_v1", "ContextTools", "ExecutableHttpTool", "HTTP_TOOL", "HttpIntentPhase", "HttpTool", "HttpToolProvider", "ToolDeniedError", "ToolError", "ToolPolicy", "ToolRisk", "Tools", "ToolToken", "ToolUnavailableError"]
 
 __all__ += [
     "GraphMutationSession", "GraphCommittedError",

src/teaql/runtime/context.py

@@ -8,6 +8,7 @@
 from datetime import date, datetime, timedelta, timezone
 
 from .business_clock import BusinessClock, SystemBusinessClock
+from .internal_id import InternalIdAllocator, validate_internal_id_allocator
 from .entity_reference import (
     EntityReferenceClaims,
     EntityReferenceCodec,
@@ -123,6 +124,7 @@ def __init__(self):
         self._checked_mutations = set()
         self._business_clock: BusinessClock = SystemBusinessClock()
         self._entity_reference_codec: Optional[EntityReferenceCodec] = None
+        self._internal_id_allocator: Optional[InternalIdAllocator] = None
         from .mutation_policy import MutationPolicyRuntimeState
         self._mutation_policy = MutationPolicyRuntimeState()
 
@@ -548,12 +550,20 @@ def configure_audit_policy(self, entity: str, mask_fields: List[str],
             descriptor.audit_value_max_len(value_max_len)
         return self
 
-    def with_internal_id_generator(self, gen: Any) -> 'UserContext':
-        self.insert_resource("internal_id_generator", gen)
+    def with_internal_id_allocator(self, allocator: InternalIdAllocator) -> 'UserContext':
+        """Install ID allocation independently of the query/mutation provider."""
+        self._internal_id_allocator = validate_internal_id_allocator(allocator)
         return self
 
-    def set_internal_id_generator(self, gen: Any):
-        self.insert_resource("internal_id_generator", gen)
+    def internal_id_allocator(self) -> Optional[InternalIdAllocator]:
+        return self._internal_id_allocator
+
+    def with_internal_id_generator(self, gen: InternalIdAllocator) -> 'UserContext':
+        """Compatibility spelling; synchronous generate_id providers are unsupported."""
+        return self.with_internal_id_allocator(gen)
+
+    def set_internal_id_generator(self, gen: InternalIdAllocator):
+        self.with_internal_id_allocator(gen)
 
     def with_business_id_profile_factory(self, factory: Any) -> 'UserContext':
         if factory is None or not callable(getattr(factory, "create", None)):
@@ -621,17 +631,10 @@ def i18n_catalog(self):
         return self.get_resource("i18n_catalog") or I18nCatalog.builtin()
 
     def generate_id(self, entity: str) -> Optional[int]:
-        gen = self.get_resource("internal_id_generator")
-        if gen and hasattr(gen, "generate_id"):
-            return gen.generate_id(entity)
-        return None
+        raise RuntimeError("synchronous generate_id is unsupported; use an async InternalIdAllocator")
 
     def next_id(self, entity: str) -> int:
-        gen_id = self.generate_id(entity)
-        if gen_id is not None:
-            return gen_id
-        # Simple fallback
-        return int(datetime.now().timestamp() * 1000)
+        raise RuntimeError("synchronous next_id is unsupported; use an async InternalIdAllocator")
 
     def entity(self, name: str) -> Optional[Any]:
         if self._metadata and hasattr(self._metadata, "entity"):

src/teaql/runtime/internal_id.py

@@ -0,0 +1,26 @@
+"""Trusted, context-owned internal ID allocation capability."""
+
+from inspect import iscoroutinefunction
+from typing import Protocol, runtime_checkable
+
+
+@runtime_checkable
+class InternalIdAllocator(Protocol):
+    """Allocate unique per-entity IDs and reserve explicit-ID floors.
+
+    Implementations used across processes must persist the floor. A failed
+    transaction may leave gaps, but an allocated ID must never be reused.
+    """
+
+    async def next_id(self, entity: str) -> int: ...
+
+    async def ensure_id_floor(self, entity: str, floor: int) -> None: ...
+
+
+def validate_internal_id_allocator(allocator: InternalIdAllocator) -> InternalIdAllocator:
+    """Reject incomplete/synchronous capabilities before provider execution."""
+    if not isinstance(allocator, InternalIdAllocator):
+        raise TypeError("internal ID allocator must provide next_id() and ensure_id_floor()")
+    if not iscoroutinefunction(allocator.next_id) or not iscoroutinefunction(allocator.ensure_id_floor):
+        raise TypeError("internal ID allocator methods must be async")
+    return allocator

src/teaql/sql/executor.py

@@ -1030,12 +1030,14 @@ async def _mutate(self, context: 'UserContext', request: MutationRequest, privac
             None,
         )
         if isinstance(req_data, InsertCommand) and id_prop is not None:
+            allocator = context.internal_id_allocator() if context is not None else None
+            allocator = allocator if allocator is not None else self
             if id_prop.name not in req_data.values:
                 req_data.values[id_prop.name] = Value.from_any(
-                    await self.next_id(req_data.entity)
+                    await allocator.next_id(req_data.entity)
                 )
             else:
-                await self.ensure_id_floor(
+                await allocator.ensure_id_floor(
                     req_data.entity, int(req_data.values[id_prop.name].val))
             version_prop = next((
                 prop for prop in getattr(entity_desc, "properties", [])