diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedEventStore.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedEventStore.java
new file mode 100644
index 00000000..f91dada1
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedEventStore.java
@@ -0,0 +1,269 @@
+package io.teaql.core.sqlite;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+
+import java.nio.charset.StandardCharsets;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.HexFormat;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.function.LongFunction;
+
+/**
+ * SQLite durability boundary for already-committed logical mutation facts.
+ *
+ * The caller must install the schema before use. Append and apply require the
+ * same active SQL transaction as business writes. This class deliberately does
+ * not infer mutations from a ledger, replay business logic, or expose a remote
+ * replication endpoint. Projection writers are trusted infrastructure code.
+ */
+public final class SqliteCommittedEventStore {
+ private static final ObjectMapper JSON = new ObjectMapper();
+ private final SqlExecutionAdapter sql;
+
+ public SqliteCommittedEventStore(SqlExecutionAdapter sql) {
+ this.sql = Objects.requireNonNull(sql, "sql");
+ }
+
+ public record OutboxRecord(String sourceRuntimeId, long sourceSequence,
+ String changeSetId, String modelHash, String payloadSha256, String payloadJson) {
+ @Override public String toString() {
+ return "OutboxRecord[sourceRuntimeId=" + sourceRuntimeId
+ + ", sourceSequence=" + sourceSequence + ", payload=]";
+ }
+ }
+
+ public record InboxOutcome(boolean duplicate, long checkpoint) {
+ public static InboxOutcome applied(long checkpoint) { return new InboxOutcome(false, checkpoint); }
+ public static InboxOutcome duplicate(long checkpoint) { return new InboxOutcome(true, checkpoint); }
+ }
+
+ @FunctionalInterface
+ interface TrustedProjectionWriter {
+ void apply(SqlExecutionAdapter transaction, JsonNode committedFacts);
+ }
+
+ public static final class CommittedEventException extends RuntimeException {
+ private final String code;
+
+ CommittedEventException(String code, String message) {
+ super(message);
+ this.code = code;
+ }
+
+ public String code() { return code; }
+ }
+
+ public void ensureSchema() {
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_sequence ("
+ + "source_runtime_id TEXT PRIMARY KEY, "
+ + "current_sequence INTEGER NOT NULL CHECK(current_sequence >= 0))");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_outbox ("
+ + "source_runtime_id TEXT NOT NULL, "
+ + "source_sequence INTEGER NOT NULL CHECK(source_sequence > 0), "
+ + "change_set_id TEXT NOT NULL, model_hash TEXT NOT NULL, "
+ + "payload_sha256 TEXT NOT NULL, payload_json TEXT NOT NULL, "
+ + "PRIMARY KEY(source_runtime_id, source_sequence), "
+ + "UNIQUE(source_runtime_id, change_set_id))");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_inbox ("
+ + "source_runtime_id TEXT NOT NULL, change_set_id TEXT NOT NULL, "
+ + "source_sequence INTEGER NOT NULL CHECK(source_sequence > 0), "
+ + "payload_sha256 TEXT NOT NULL, PRIMARY KEY(source_runtime_id, change_set_id))");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_applied_payload ("
+ + "source_runtime_id TEXT NOT NULL, source_sequence INTEGER NOT NULL CHECK(source_sequence > 0), "
+ + "change_set_id TEXT NOT NULL, payload_json TEXT NOT NULL, "
+ + "PRIMARY KEY(source_runtime_id, source_sequence), "
+ + "UNIQUE(source_runtime_id, change_set_id))");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_checkpoint ("
+ + "source_runtime_id TEXT PRIMARY KEY, "
+ + "source_sequence INTEGER NOT NULL CHECK(source_sequence >= 0), "
+ + "model_hash TEXT NOT NULL)");
+ sql.execute("CREATE TABLE IF NOT EXISTS teaql_event_delivery_cursor ("
+ + "source_runtime_id TEXT NOT NULL, target_id TEXT NOT NULL, "
+ + "acked_sequence INTEGER NOT NULL CHECK(acked_sequence >= 0), "
+ + "model_hash TEXT NOT NULL, PRIMARY KEY(source_runtime_id, target_id))");
+ }
+
+ /** Bounded publisher read of one durable, ordered source stream. */
+ public List outboxAfter(String sourceRuntimeId, long after, int limit) {
+ if (blank(sourceRuntimeId) || after < 0 || limit < 1 || limit > 1000) {
+ throw invalid("invalid source, position, or read limit");
+ }
+ return sql.queryForList("SELECT source_runtime_id, source_sequence, change_set_id, "
+ + "model_hash, payload_sha256, payload_json FROM teaql_event_outbox "
+ + "WHERE source_runtime_id = ? AND source_sequence > ? "
+ + "ORDER BY source_sequence ASC LIMIT ?", new Object[] {sourceRuntimeId, after, limit})
+ .stream().map(row -> new OutboxRecord((String) row.get("source_runtime_id"),
+ number(row.get("source_sequence")), (String) row.get("change_set_id"),
+ (String) row.get("model_hash"), (String) row.get("payload_sha256"),
+ (String) row.get("payload_json"))).toList();
+ }
+
+ /** Allocate and append within the caller's already-open business transaction. */
+ public OutboxRecord append(String sourceRuntimeId, String changeSetId, String modelHash,
+ LongFunction payloadBuilder) {
+ requireTransaction();
+ if (blank(sourceRuntimeId) || blank(changeSetId) || blank(modelHash) || payloadBuilder == null) {
+ throw invalid("missing committed-event identity or payload builder");
+ }
+ return savepoint("teaql_event_append", () -> {
+ sql.update("INSERT INTO teaql_event_sequence(source_runtime_id, current_sequence) "
+ + "VALUES (?, 0) ON CONFLICT(source_runtime_id) DO NOTHING",
+ new Object[] {sourceRuntimeId});
+ int updated = sql.update("UPDATE teaql_event_sequence SET current_sequence = current_sequence + 1 "
+ + "WHERE source_runtime_id = ? AND current_sequence < 9223372036854775807",
+ new Object[] {sourceRuntimeId});
+ if (updated != 1) throw new CommittedEventException("SEQUENCE_EXHAUSTED", "source sequence exhausted");
+ long sequence = number(sql.queryForList("SELECT current_sequence FROM teaql_event_sequence "
+ + "WHERE source_runtime_id = ?", new Object[] {sourceRuntimeId}).get(0).get("current_sequence"));
+ String payloadJson = payloadBuilder.apply(sequence);
+ JsonNode payload = parse(payloadJson);
+ validateHeader(payload, sourceRuntimeId, sequence, changeSetId, modelHash);
+ String digest = digest(payloadJson);
+ sql.update("INSERT INTO teaql_event_outbox "
+ + "(source_runtime_id, source_sequence, change_set_id, model_hash, payload_sha256, payload_json) "
+ + "VALUES (?, ?, ?, ?, ?, ?)",
+ new Object[] {sourceRuntimeId, sequence, changeSetId, modelHash, digest, payloadJson});
+ return new OutboxRecord(sourceRuntimeId, sequence, changeSetId, modelHash, digest, payloadJson);
+ });
+ }
+
+ /**
+ * Apply one set through a restricted projection writer. The writer must not
+ * call normal entity save, Checker/Fix, ID generation or business hooks.
+ */
+ InboxOutcome apply(OutboxRecord record, String expectedModelHash,
+ TrustedProjectionWriter writer) {
+ requireTransaction();
+ if (record == null || blank(expectedModelHash) || writer == null) {
+ throw invalid("missing committed event, model hash, or projection writer");
+ }
+ if (!expectedModelHash.equals(record.modelHash())) {
+ throw new CommittedEventException("MODEL_MISMATCH", "committed-event model hash mismatch");
+ }
+ JsonNode payload = parse(record.payloadJson());
+ validateHeader(payload, record.sourceRuntimeId(), record.sourceSequence(),
+ record.changeSetId(), record.modelHash());
+ if (!digest(record.payloadJson()).equals(record.payloadSha256())) {
+ throw invalid("payload digest mismatch");
+ }
+ List> checkpoints = sql.queryForList(
+ "SELECT source_sequence, model_hash FROM teaql_event_checkpoint WHERE source_runtime_id = ?",
+ new Object[] {record.sourceRuntimeId()});
+ long last = checkpoints.isEmpty() ? 0 : number(checkpoints.get(0).get("source_sequence"));
+ if (!checkpoints.isEmpty() && !expectedModelHash.equals(checkpoints.get(0).get("model_hash"))) {
+ throw new CommittedEventException("MODEL_MISMATCH", "committed-event checkpoint model mismatch");
+ }
+ List> prior = sql.queryForList(
+ "SELECT source_sequence, payload_sha256 FROM teaql_event_inbox "
+ + "WHERE source_runtime_id = ? AND change_set_id = ?",
+ new Object[] {record.sourceRuntimeId(), record.changeSetId()});
+ if (!prior.isEmpty()) {
+ Map seen = prior.get(0);
+ if (number(seen.get("source_sequence")) == record.sourceSequence()
+ && record.payloadSha256().equals(seen.get("payload_sha256"))) {
+ List> retained = sql.queryForList(
+ "SELECT source_sequence, payload_json FROM teaql_event_applied_payload "
+ + "WHERE source_runtime_id = ? AND change_set_id = ?",
+ new Object[] {record.sourceRuntimeId(), record.changeSetId()});
+ if (retained.size() != 1
+ || number(retained.get(0).get("source_sequence")) != record.sourceSequence()
+ || !record.payloadSha256().equals(digest((String) retained.get(0).get("payload_json")))) {
+ throw invalid("prior Inbox identity lacks retained source evidence");
+ }
+ if (last < record.sourceSequence()) {
+ throw invalid("prior Inbox identity is ahead of target checkpoint");
+ }
+ return InboxOutcome.duplicate(record.sourceSequence());
+ }
+ throw new CommittedEventException("DUPLICATE_CONFLICT", "committed-event identity has conflicting content");
+ }
+ if (last == Long.MAX_VALUE) {
+ throw new CommittedEventException("SEQUENCE_EXHAUSTED", "source sequence exhausted");
+ }
+ if (record.sourceSequence() != last + 1) {
+ throw new CommittedEventException("SEQUENCE_GAP",
+ "committed-event sequence gap: expected " + (last + 1) + ", received " + record.sourceSequence());
+ }
+ return savepoint("teaql_event_apply", () -> {
+ writer.apply(sql, payload);
+ sql.update("INSERT INTO teaql_event_applied_payload "
+ + "(source_runtime_id, source_sequence, change_set_id, payload_json) VALUES (?, ?, ?, ?)",
+ new Object[] {record.sourceRuntimeId(), record.sourceSequence(),
+ record.changeSetId(), record.payloadJson()});
+ sql.update("INSERT INTO teaql_event_inbox "
+ + "(source_runtime_id, change_set_id, source_sequence, payload_sha256) VALUES (?, ?, ?, ?)",
+ new Object[] {record.sourceRuntimeId(), record.changeSetId(),
+ record.sourceSequence(), record.payloadSha256()});
+ sql.update("INSERT INTO teaql_event_checkpoint (source_runtime_id, source_sequence, model_hash) "
+ + "VALUES (?, ?, ?) ON CONFLICT(source_runtime_id) DO UPDATE SET "
+ + "source_sequence = excluded.source_sequence, model_hash = excluded.model_hash",
+ new Object[] {record.sourceRuntimeId(), record.sourceSequence(), record.modelHash()});
+ return InboxOutcome.applied(record.sourceSequence());
+ });
+ }
+
+ private static JsonNode parse(String payloadJson) {
+ if (payloadJson == null) throw invalid("missing payload");
+ try {
+ return JSON.readTree(payloadJson);
+ } catch (Exception ignored) {
+ throw invalid("invalid payload JSON");
+ }
+ }
+
+ private static void validateHeader(JsonNode payload, String source, long sequence,
+ String changeSetId, String modelHash) {
+ if (blank(source) || sequence < 1 || blank(changeSetId) || blank(modelHash)
+ || payload == null || payload.path("protocol_version").asInt(-1) != 1
+ || !source.equals(payload.path("source_runtime_id").asText())
+ || !Long.toString(sequence).equals(payload.path("source_sequence").asText())
+ || !changeSetId.equals(payload.path("change_set_id").asText())
+ || !modelHash.equals(payload.path("source_model_hash").asText())
+ || !payload.path("mutations").isArray() || payload.path("mutations").isEmpty()) {
+ throw invalid("required v1 header or mutations mismatch");
+ }
+ }
+
+ private T savepoint(String name, java.util.function.Supplier action) {
+ sql.execute("SAVEPOINT " + name);
+ try {
+ T value = action.get();
+ sql.execute("RELEASE SAVEPOINT " + name);
+ return value;
+ } catch (RuntimeException | Error failure) {
+ try {
+ sql.execute("ROLLBACK TO SAVEPOINT " + name);
+ sql.execute("RELEASE SAVEPOINT " + name);
+ } catch (RuntimeException rollbackFailure) {
+ // A failed savepoint rollback must not leave partial rows committable.
+ try { sql.execute("ROLLBACK"); } catch (RuntimeException ignored) { }
+ throw new CommittedEventException("ROLLBACK_FAILED", "committed-event rollback failed");
+ }
+ throw failure;
+ }
+ }
+
+ private void requireTransaction() {
+ if (!sql.hasActiveTransaction()) throw invalid("an active SQL transaction is required");
+ }
+
+ private static long number(Object value) { return ((Number) value).longValue(); }
+ private static boolean blank(String value) { return value == null || value.isBlank(); }
+ private static CommittedEventException invalid(String reason) {
+ return new CommittedEventException("INVALID_ENVELOPE", "invalid committed-event envelope: " + reason);
+ }
+ private static String digest(String payload) {
+ try {
+ byte[] hash = MessageDigest.getInstance("SHA-256")
+ .digest(payload.getBytes(StandardCharsets.UTF_8));
+ return HexFormat.of().formatHex(hash);
+ } catch (NoSuchAlgorithmException impossible) {
+ throw new AssertionError("JDK SHA-256 is unavailable", impossible);
+ }
+ }
+}
diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedMutationRecorder.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedMutationRecorder.java
new file mode 100644
index 00000000..1e45db2f
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedMutationRecorder.java
@@ -0,0 +1,111 @@
+package io.teaql.core.sqlite;
+
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import io.teaql.core.GeneratedSchemaBootstrap;
+import io.teaql.core.CommittedEventSchemaContributor;
+import io.teaql.core.MutationIntent;
+import io.teaql.core.SchemaExecutor;
+import io.teaql.core.TraceNode;
+import io.teaql.core.UserContext;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+import io.teaql.runtime.CommittedMutationFact;
+import io.teaql.runtime.CommittedMutationRecorder;
+
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.UUID;
+
+/** SQLite source adapter that records one graph-save result inside its write transaction. */
+public final class SqliteCommittedMutationRecorder
+ implements CommittedMutationRecorder, CommittedEventSchemaContributor {
+ private static final ObjectMapper JSON = new ObjectMapper();
+ private final SqlExecutionAdapter adapter;
+ private final SqliteCommittedEventStore events;
+ private final String sourceRuntimeId;
+ private final String modelHash;
+
+ public SqliteCommittedMutationRecorder(SqlExecutionAdapter adapter,
+ String sourceRuntimeId, String modelHash) {
+ this.adapter = Objects.requireNonNull(adapter, "adapter");
+ this.events = new SqliteCommittedEventStore(adapter);
+ if (sourceRuntimeId == null || sourceRuntimeId.isBlank()
+ || modelHash == null || !modelHash.matches("sha256:[0-9a-f]{64}")) {
+ throw new IllegalArgumentException("A source runtime ID and accepted SHA-256 model hash are required");
+ }
+ this.sourceRuntimeId = sourceRuntimeId;
+ this.modelHash = modelHash;
+ }
+
+ @Override
+ public void ensureSchema(UserContext context, SchemaExecutor.Invocation invocation) {
+ SchemaExecutor.Invocation.requireContextOwned(invocation);
+ Objects.requireNonNull(context, "context");
+ events.ensureSchema();
+ }
+ public SqliteCommittedEventStore eventStore() { return events; }
+
+ @Override
+ public void record(UserContext context, Object transactionResource,
+ MutationIntent intent, List facts) {
+ Objects.requireNonNull(context, "context");
+ Objects.requireNonNull(intent, "intent");
+ if (transactionResource != adapter || !adapter.hasActiveTransaction()) {
+ throw new IllegalStateException("Committed graph capture requires the source provider transaction");
+ }
+ if (facts == null || facts.isEmpty()) {
+ throw new IllegalArgumentException("Committed graph capture requires at least one fact");
+ }
+ String changeSetId = UUID.randomUUID().toString();
+ String committedAt = Instant.now().toString();
+ String actor = context.getAttribute(GeneratedSchemaBootstrap.AUDIT_ACTOR_ATTRIBUTE, String.class);
+ events.append(sourceRuntimeId, changeSetId, modelHash, sequence -> {
+ Map envelope = new LinkedHashMap<>();
+ envelope.put("protocol_version", 1);
+ envelope.put("change_set_id", changeSetId);
+ envelope.put("source_runtime_id", sourceRuntimeId);
+ envelope.put("source_sequence", Long.toString(sequence));
+ envelope.put("source_model_hash", modelHash);
+ envelope.put("source_transaction_id", changeSetId);
+ envelope.put("committed_at", committedAt);
+ Map audit = new LinkedHashMap<>();
+ audit.put("actor_ref", actor);
+ audit.put("comment", intent.comment());
+ // MutationIntent currently has no separate business purpose field.
+ audit.put("purpose", "replicate committed graph");
+ envelope.put("audit", audit);
+ List> mutations = new ArrayList<>(facts.size());
+ for (CommittedMutationFact fact : facts) {
+ Map mutation = new LinkedHashMap<>();
+ mutation.put("entity_type", fact.entityType());
+ mutation.put("entity_id", fact.entityId());
+ mutation.put("operation", fact.operation());
+ mutation.put("base_version", fact.baseVersion());
+ mutation.put("committed_version", fact.committedVersion());
+ mutation.put("fields", fact.fields());
+ List> chain = new ArrayList<>();
+ for (TraceNode node : fact.traceChain()) {
+ Map step = new LinkedHashMap<>();
+ step.put("kind", node.getKind().name());
+ step.put("name", Objects.toString(node.getName(), ""));
+ step.put("entity_id", node.getEntityId() == null ? null : node.getEntityId().toString());
+ step.put("comment", Objects.toString(node.getComment(), ""));
+ chain.add(step);
+ }
+ mutation.put("trace_chain", chain);
+ mutations.add(mutation);
+ }
+ envelope.put("mutations", mutations);
+ try {
+ return JSON.writeValueAsString(envelope);
+ } catch (JsonProcessingException failure) {
+ // Do not attach values or serializer causes to public errors.
+ throw new IllegalStateException("Committed mutation payload encoding failed");
+ }
+ });
+ }
+}
diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedOutboxPublisher.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedOutboxPublisher.java
new file mode 100644
index 00000000..82b18a64
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteCommittedOutboxPublisher.java
@@ -0,0 +1,118 @@
+package io.teaql.core.sqlite;
+
+import io.teaql.core.sqlite.SqliteCommittedEventStore.CommittedEventException;
+import io.teaql.core.sqlite.SqliteCommittedEventStore.InboxOutcome;
+import io.teaql.core.sqlite.SqliteCommittedEventStore.OutboxRecord;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+
+/**
+ * Bounded ordered publication from one durable SQLite Outbox stream. The
+ * supplied sink must own its authentication and network retry budget; this
+ * publisher never treats a callback as proof of remote identity.
+ */
+public final class SqliteCommittedOutboxPublisher {
+ @FunctionalInterface
+ public interface DeliverySink {
+ InboxOutcome deliver(OutboxRecord record);
+ }
+
+ public record DeliveryStatus(long ackedSequence, long sourceHighWatermark, long backlog) { }
+ public record BatchOutcome(int acknowledged, DeliveryStatus status) { }
+
+ private final SqlExecutionAdapter sql;
+ private final SqliteCommittedEventStore events;
+ private final String sourceRuntimeId;
+ private final String targetId;
+ private final String modelHash;
+
+ public SqliteCommittedOutboxPublisher(SqlExecutionAdapter sql, String sourceRuntimeId,
+ String targetId, String modelHash) {
+ this.sql = Objects.requireNonNull(sql, "sql");
+ if (sourceRuntimeId == null || sourceRuntimeId.isBlank()
+ || targetId == null || targetId.isBlank()
+ || modelHash == null || !modelHash.matches("sha256:[0-9a-f]{64}")) {
+ throw failure("INVALID_ENVELOPE", "delivery source, target or model identity is invalid");
+ }
+ this.sourceRuntimeId = sourceRuntimeId;
+ this.targetId = targetId;
+ this.modelHash = modelHash;
+ this.events = new SqliteCommittedEventStore(sql);
+ }
+
+ public DeliveryStatus status() {
+ List> rows = sql.queryForList(
+ "SELECT acked_sequence, model_hash FROM teaql_event_delivery_cursor "
+ + "WHERE source_runtime_id = ? AND target_id = ?",
+ new Object[] {sourceRuntimeId, targetId});
+ long acked = 0;
+ if (!rows.isEmpty()) {
+ if (!modelHash.equals(rows.get(0).get("model_hash"))) {
+ throw failure("MODEL_MISMATCH", "delivery cursor model hash mismatch");
+ }
+ acked = number(rows.get(0).get("acked_sequence"));
+ }
+ List> sourceRows = sql.queryForList(
+ "SELECT current_sequence FROM teaql_event_sequence WHERE source_runtime_id = ?",
+ new Object[] {sourceRuntimeId});
+ long highWatermark = sourceRows.isEmpty() ? 0
+ : number(sourceRows.get(0).get("current_sequence"));
+ if (highWatermark < acked) {
+ throw failure("INVALID_ENVELOPE", "delivery cursor is ahead of source sequence");
+ }
+ return new DeliveryStatus(acked, highWatermark, highWatermark - acked);
+ }
+
+ /**
+ * One attempt per event, at most {@code limit} events. The source cursor
+ * advances only after an acknowledgement for the exact sequence. If the
+ * target applied an event but the response is lost, a later call/restart
+ * re-delivers it and the target Inbox supplies a duplicate acknowledgement.
+ */
+ public BatchOutcome publishBatch(int limit, DeliverySink sink) {
+ Objects.requireNonNull(sink, "sink");
+ DeliveryStatus before = status();
+ List records = events.outboxAfter(sourceRuntimeId, before.ackedSequence(), limit);
+ if (records.isEmpty() && before.backlog() > 0) {
+ throw failure("SEQUENCE_GAP", "source Outbox has an unacknowledged gap");
+ }
+ long acked = before.ackedSequence();
+ int count = 0;
+ for (OutboxRecord record : records) {
+ if (acked == Long.MAX_VALUE) {
+ throw failure("SEQUENCE_EXHAUSTED", "source sequence exhausted");
+ }
+ if (record.sourceSequence() != acked + 1) {
+ throw failure("SEQUENCE_GAP", "source Outbox sequence is not contiguous");
+ }
+ if (!modelHash.equals(record.modelHash())) {
+ throw failure("MODEL_MISMATCH", "source Outbox model hash mismatch");
+ }
+ InboxOutcome outcome = sink.deliver(record);
+ if (outcome == null || outcome.checkpoint() != record.sourceSequence()) {
+ throw failure("ACK_MISMATCH", "target acknowledgement does not match delivered sequence");
+ }
+ int changed = sql.update("INSERT INTO teaql_event_delivery_cursor "
+ + "(source_runtime_id, target_id, acked_sequence, model_hash) VALUES (?, ?, ?, ?) "
+ + "ON CONFLICT(source_runtime_id, target_id) DO UPDATE SET "
+ + "acked_sequence = excluded.acked_sequence "
+ + "WHERE teaql_event_delivery_cursor.acked_sequence = ? "
+ + "AND teaql_event_delivery_cursor.model_hash = excluded.model_hash",
+ new Object[] {sourceRuntimeId, targetId, record.sourceSequence(), modelHash, acked});
+ if (changed != 1) {
+ throw failure("CURSOR_CONFLICT", "delivery cursor changed concurrently");
+ }
+ acked = record.sourceSequence();
+ count++;
+ }
+ return new BatchOutcome(count, status());
+ }
+
+ private static long number(Object value) { return ((Number) value).longValue(); }
+ private static CommittedEventException failure(String code, String message) {
+ return new CommittedEventException(code, "committed-event " + message);
+ }
+}
diff --git a/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteModelCheckedCommittedTarget.java b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteModelCheckedCommittedTarget.java
new file mode 100644
index 00000000..941e3761
--- /dev/null
+++ b/teaql-sqlite/src/main/java/io/teaql/core/sqlite/SqliteModelCheckedCommittedTarget.java
@@ -0,0 +1,417 @@
+package io.teaql.core.sqlite;
+
+import com.fasterxml.jackson.databind.JsonNode;
+import io.teaql.core.UserContext;
+import io.teaql.core.meta.EntityDescriptor;
+import io.teaql.core.meta.EntityMetaFactory;
+import io.teaql.core.meta.PropertyDescriptor;
+import io.teaql.core.meta.Relation;
+import io.teaql.core.sql.SQLColumn;
+import io.teaql.core.sql.portable.SQLPropertyUtil;
+import io.teaql.dataservice.sql.SqlExecutionAdapter;
+
+import java.math.BigDecimal;
+import java.time.OffsetDateTime;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Map;
+import java.util.Objects;
+import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
+import java.util.regex.Matcher;
+import java.util.regex.Pattern;
+
+/**
+ * Startup-owned, one-source SQLite target for committed model facts. This is
+ * infrastructure, not a business-facing Mutation or remote query endpoint.
+ */
+public final class SqliteModelCheckedCommittedTarget {
+ private static final Set TRACE_KINDS = Set.of(
+ "OPERATION", "REQUEST", "RELATION", "ENTITY", "PROVIDER", "SQL",
+ "COMMENT", "PURPOSE", "AUDIT_REASON");
+ private static final Pattern DECLARED_DECIMAL = Pattern.compile(
+ "(?i)\\s*(?:NUMERIC|DECIMAL)\\s*\\(\\s*([0-9]+)\\s*,\\s*([0-9]+)\\s*\\)\\s*");
+ private final SqlExecutionAdapter sql;
+ private final SqliteCommittedEventStore events;
+ private final String sourceRuntimeId;
+ private final String modelHash;
+
+ public SqliteModelCheckedCommittedTarget(SqlExecutionAdapter sql,
+ String sourceRuntimeId, String modelHash) {
+ this.sql = Objects.requireNonNull(sql, "sql");
+ if (sourceRuntimeId == null || sourceRuntimeId.isBlank()
+ || modelHash == null || !modelHash.matches("sha256:[0-9a-f]{64}")) {
+ throw invalid("target source identity or SHA-256 model hash is invalid");
+ }
+ this.sourceRuntimeId = sourceRuntimeId;
+ this.modelHash = modelHash;
+ this.events = new SqliteCommittedEventStore(sql);
+ }
+
+ public void ensureSchema() { events.ensureSchema(); }
+
+ /** One graph, its Inbox identity, original audit payload and checkpoint commit together. */
+ public SqliteCommittedEventStore.InboxOutcome apply(
+ UserContext context, SqliteCommittedEventStore.OutboxRecord record) {
+ Objects.requireNonNull(context, "context");
+ if (record == null) throw invalid("committed event is missing");
+ if (!sourceRuntimeId.equals(record.sourceRuntimeId())) {
+ throw new SqliteCommittedEventStore.CommittedEventException(
+ "UNAUTHORIZED_SOURCE", "committed-event source is not authorized");
+ }
+ // Resolve from this Context, never from the legacy process-global fallback.
+ EntityMetaFactory metadata = EntityMetaFactory.requireFrom(context);
+ AtomicReference result = new AtomicReference<>();
+ sql.executeInTransaction(() -> result.set(events.apply(record, modelHash,
+ (transaction, payload) -> applyGraph(transaction, metadata, payload))));
+ return result.get();
+ }
+
+ private static void applyGraph(SqlExecutionAdapter sql, EntityMetaFactory metadata, JsonNode payload) {
+ JsonNode audit = payload.path("audit");
+ if (!audit.isObject() || !nonblank(audit.path("comment"))
+ || !nonblank(audit.path("purpose"))
+ || !nonblank(payload.path("source_transaction_id"))) {
+ throw invalid("source transaction or audit intent is missing");
+ }
+ try { OffsetDateTime.parse(payload.path("committed_at").asText()); }
+ catch (RuntimeException failure) { throw invalid("commit time is invalid"); }
+ List mutations = new ArrayList<>();
+ Set seen = new HashSet<>();
+ for (JsonNode fact : payload.path("mutations")) {
+ Mutation mutation = decode(metadata, fact);
+ if (!seen.add(mutation.entityType() + ":" + mutation.id())) {
+ throw invalid("one set mutates the same entity twice");
+ }
+ mutations.add(mutation);
+ }
+ // Decode every fact before touching domain rows. The surrounding
+ // savepoint also rolls back a later FK or readback failure.
+ for (Mutation mutation : mutations) applyOne(sql, mutation);
+ }
+
+ private static Mutation decode(EntityMetaFactory metadata, JsonNode fact) {
+ if (!fact.isObject() || !nonblank(fact.path("entity_type"))) {
+ throw invalid("mutation entity type is missing");
+ }
+ validateTraceChain(fact.path("trace_chain"));
+ String entityType = fact.path("entity_type").textValue();
+ EntityDescriptor entity = metadata.allEntityDescriptors().stream()
+ .filter(candidate -> entityType.equals(candidate.getType())).findFirst()
+ .orElseThrow(() -> invalid("unknown entity type"));
+ // Multi-table inheritance needs an explicit target mapping. Reject it
+ // instead of silently writing only the leaf table.
+ if (entity.getParent() != null || entity.hasChildren()) {
+ throw invalid("multi-table entity target mapping is not supported");
+ }
+ PropertyDescriptor idProperty = entity.findIdProperty();
+ PropertyDescriptor versionProperty = entity.findVersionProperty();
+ if (idProperty == null || versionProperty == null) {
+ throw invalid("entity has no ID or version property");
+ }
+ Mapping idMapping = mapping(idProperty);
+ Mapping versionMapping = mapping(versionProperty);
+ if (!idMapping.table().equals(versionMapping.table())) {
+ throw invalid("identity and version span multiple tables");
+ }
+ long id = positiveId(fact.path("entity_id"));
+ String operation = fact.path("operation").asText("");
+ if (!List.of("create", "update", "delete", "recover").contains(operation)) {
+ throw invalid("unknown mutation operation");
+ }
+ JsonNode baseNode = fact.path("base_version");
+ Long base = "create".equals(operation) && baseNode.isNull()
+ ? null : signedVersion(baseNode);
+ long committed = signedVersion(fact.path("committed_version"));
+ boolean valid = switch (operation) {
+ case "create" -> base == null && committed == 1;
+ case "update" -> base != null && base > 0 && base < Long.MAX_VALUE
+ && committed == base + 1;
+ case "delete" -> base != null && base > 0 && base < Long.MAX_VALUE
+ && committed == -(base + 1);
+ case "recover" -> base != null && base < 0 && base != Long.MIN_VALUE
+ && -base < Long.MAX_VALUE && committed == -base + 1;
+ default -> false;
+ };
+ if (!valid) throw invalid("invalid version transition");
+ JsonNode supplied = fact.path("fields");
+ if (!supplied.isObject()) throw invalid("mutation fields are missing");
+ if ("delete".equals(operation) && !supplied.isEmpty()) {
+ throw invalid("delete must be a tombstone");
+ }
+ List ownProperties = entity.getOwnProperties();
+ supplied.fieldNames().forEachRemaining(name -> {
+ PropertyDescriptor property = entity.findProperty(name);
+ if (property == null || !ownProperties.contains(property)
+ || property.isId() || property.isVersion()) {
+ throw invalid("unknown or reserved model field");
+ }
+ });
+ List fields = new ArrayList<>();
+ for (PropertyDescriptor property : ownProperties) {
+ if (property.isId() || property.isVersion()) continue;
+ JsonNode value = supplied.get(property.getName());
+ if (value == null) {
+ if ("create".equals(operation)) throw invalid("create omitted a modeled field");
+ continue;
+ }
+ Mapping column = mapping(property);
+ if (!idMapping.table().equals(column.table())) {
+ throw invalid("mutation field spans multiple tables");
+ }
+ Class type = property.getType() == null ? null : property.getType().javaType();
+ fields.add(new Field(column.column(), decodeValue(property, value),
+ type == BigDecimal.class || type == java.math.BigInteger.class));
+ }
+ return new Mutation(entityType, idMapping.table(), idMapping.column(),
+ versionMapping.column(), id, base, committed, operation, fields);
+ }
+
+ private static void validateTraceChain(JsonNode chain) {
+ if (!chain.isArray()) throw invalid("mutation trace chain is missing");
+ for (JsonNode node : chain) {
+ JsonNode id = node.path("entity_id");
+ if (!node.isObject() || node.size() != 4
+ || !TRACE_KINDS.contains(node.path("kind").asText(""))
+ || !node.path("name").isTextual()
+ || !node.path("comment").isTextual()
+ || !(id.isNull() || id.isTextual() && id.textValue().matches("[1-9][0-9]*"))) {
+ throw invalid("trace node has wrong shape");
+ }
+ }
+ }
+
+ private static Object decodeValue(PropertyDescriptor property, JsonNode value) {
+ if (value.isNull()) {
+ if (property.getBoolean("required", false)) throw invalid("non-null field is null");
+ return null;
+ }
+ if (property instanceof Relation relation) {
+ if (!value.isObject() || value.size() != 2 || relation.getReverseProperty() == null
+ || relation.getReverseProperty().getOwner() == null
+ || !relation.getReverseProperty().getOwner().getType()
+ .equals(value.path("entity_type").asText())) {
+ throw invalid("reference target does not match model");
+ }
+ return positiveId(value.path("entity_id"));
+ }
+ Class type = property.getType() == null ? null : property.getType().javaType();
+ if (type == java.time.LocalDate.class || type == java.sql.Date.class) {
+ if (!value.isTextual()) throw invalid("Date field has wrong type");
+ String text = value.textValue();
+ try {
+ if (!java.time.LocalDate.parse(text).toString().equals(text)) {
+ throw invalid("Date field has wrong type");
+ }
+ } catch (java.time.format.DateTimeParseException failure) {
+ throw invalid("Date field has wrong type");
+ }
+ return text;
+ }
+ if (type == java.time.Instant.class || type == java.util.Date.class
+ || type == java.sql.Timestamp.class) {
+ if (!value.isTextual() || !value.textValue().matches("-?(0|[1-9][0-9]*)")
+ || value.textValue().equals("-0")) {
+ throw invalid("Timestamp field has wrong type");
+ }
+ try { return Long.parseLong(value.textValue()); }
+ catch (NumberFormatException failure) { throw invalid("Timestamp field has wrong type"); }
+ }
+ if (type == java.time.LocalDateTime.class) {
+ throw invalid("zone-free LocalDateTime has no committed instant mapping");
+ }
+ if (type == String.class || type == java.util.UUID.class || type != null && type.isEnum()) {
+ if (!value.isTextual()) throw invalid("text field has wrong type");
+ return value.textValue();
+ }
+ if (type == Boolean.class || type == boolean.class) {
+ if (!value.isBoolean()) throw invalid("Boolean field has wrong type");
+ return value.booleanValue() ? 1 : 0;
+ }
+ if (type == Long.class || type == long.class || type == Integer.class || type == int.class
+ || type == Short.class || type == short.class || type == Byte.class || type == byte.class) {
+ if (!value.isTextual() || !value.textValue().matches("-?(0|[1-9][0-9]*)")
+ || value.textValue().equals("-0")) {
+ throw invalid("integer field has wrong type");
+ }
+ long number;
+ try { number = Long.parseLong(value.textValue()); }
+ catch (NumberFormatException failure) { throw invalid("integer field overflows"); }
+ if (type == Integer.class || type == int.class) {
+ if (number < Integer.MIN_VALUE || number > Integer.MAX_VALUE) throw invalid("integer overflows field");
+ } else if (type == Short.class || type == short.class) {
+ if (number < Short.MIN_VALUE || number > Short.MAX_VALUE) throw invalid("integer overflows field");
+ } else if (type == Byte.class || type == byte.class) {
+ if (number < Byte.MIN_VALUE || number > Byte.MAX_VALUE) throw invalid("integer overflows field");
+ }
+ return number;
+ }
+ if (type == BigDecimal.class || type == java.math.BigInteger.class) {
+ if (!value.isTextual()) throw invalid("Decimal field has wrong type");
+ try {
+ if (type == java.math.BigInteger.class) {
+ if (!value.textValue().matches("-?(0|[1-9][0-9]*)")
+ || value.textValue().equals("-0")) {
+ throw invalid("integer field has wrong type");
+ }
+ return new java.math.BigInteger(value.textValue()).toString();
+ }
+ BigDecimal decimal = new BigDecimal(value.textValue());
+ if (!fitsDeclaredDecimal(property, decimal)) {
+ throw invalid("Decimal lacks or exceeds modeled precision/scale");
+ }
+ return decimal.toPlainString();
+ }
+ catch (NumberFormatException failure) { throw invalid("Decimal field is invalid"); }
+ }
+ if (type == Double.class || type == double.class || type == Float.class || type == float.class) {
+ if (!value.isNumber() || !Double.isFinite(value.doubleValue())) {
+ throw invalid("floating field has wrong type");
+ }
+ return value.doubleValue();
+ }
+ if (type == byte[].class && value.isObject() && value.size() == 1
+ && value.path("$binary_base64").isTextual()) {
+ try { return java.util.Base64.getDecoder().decode(value.path("$binary_base64").textValue()); }
+ catch (IllegalArgumentException failure) { throw invalid("binary field is invalid"); }
+ }
+ throw invalid("unsupported modeled field type");
+ }
+
+ private static void applyOne(SqlExecutionAdapter sql, Mutation mutation) {
+ String table = quoted(mutation.table());
+ String idColumn = quoted(mutation.idColumn());
+ String versionColumn = quoted(mutation.versionColumn());
+ List> current = sql.queryForList("SELECT " + versionColumn
+ + " AS committed_version FROM " + table + " WHERE " + idColumn + " = ?",
+ new Object[] {mutation.id()});
+ Long actualVersion = current.isEmpty() ? null : ((Number) current.get(0).get("committed_version")).longValue();
+ if (!Objects.equals(actualVersion, mutation.baseVersion())) {
+ throw new SqliteCommittedEventStore.CommittedEventException("VERSION_CONFLICT",
+ "committed-event predecessor version mismatch");
+ }
+ List