Separate outbox worker store boundary

README.md

@@ -675,27 +675,32 @@ let message = OutboxMessage::encode_for_entity(
 A separate process claims and publishes pending messages:
 
 ```rust
-use sourced_rust::{LogPublisher, OutboxRepositoryExt, OutboxWorker};
+use sourced_rust::{ClaimOutboxMessages, LogPublisher, OutboxClaimRef, OutboxStore, OutboxWorker};
 use std::time::Duration;
 
 let repo = HashMapRepository::new();
+let outbox = repo.outbox_store();
 let worker_id = "worker-1";
 let mut worker = OutboxWorker::new(LogPublisher::new())
     .with_worker_id(worker_id)
     .with_max_attempts(3);
 
-let claimed = repo.claim_outbox_messages(worker_id, 100, Duration::from_secs(30))?;
+let mut claimed = outbox.claim(ClaimOutboxMessages::new(worker_id, 100, Duration::from_secs(30)))?;
+let claims = claimed
+    .iter()
+    .map(OutboxClaimRef::from_message)
+    .collect::<Result<Vec<_>, _>>()?;
 
-for mut message in claimed {
-    let result = worker.process_message(&mut message)?;
+for (message, claim) in claimed.iter_mut().zip(claims.iter()) {
+    let result = worker.process_message(message)?;
     if result.completed {
-        repo.complete_outbox_message_for_worker(message.id(), worker_id)?;
+        outbox.complete(claim)?;
     } else if result.released || result.failed {
         let error = match message.last_error.as_deref() {
             Some(error) => error,
             None => "publish failed",
         };
-        repo.record_outbox_publish_failure(message.id(), worker_id, error, 3)?;
+        outbox.record_failure(claim, error, 3)?;
     }
 }
 ```

docs/async-repositories.md

@@ -20,8 +20,9 @@ persistence should override it with an explicit durable name through
   `AsyncRelationalReadModelQueryStore` mirror the current document and
   relational read-model surfaces for async adapters.
 - `AsyncSnapshotStore` keys snapshots by full stream identity.
-- `AsyncOutboxRepositoryExt` exposes async worker operations for durable outbox
-  implementations.
+- `AsyncOutboxStore` exposes async claim/update operations for durable outbox
+  table stores. Aggregate repositories commit outbox rows transactionally, but
+  workers do not hydrate outbox messages through aggregate repositories.
 
 Async methods use an `_async` suffix where a synchronous method with the same
 name already exists. This keeps `HashMapRepository`, `InMemoryReadModelStore`,

docs/read-models.md

@@ -256,6 +256,21 @@ schema changes should be generated or user-authored migrations plus
 verification; normal repository construction and command handling should not
 silently sync production schemas.
 
+SQL repositories expose the same lifecycle through neutral table-schema APIs so
+read models and operational tables, such as the outbox table, use one metadata
+path:
+
+```rust
+let mut registry = TableSchemaRegistry::new();
+registry.register_schema(sourced_rust::outbox_message_schema())?;
+
+let artifacts = repo.generate_table_migration_artifacts(&registry)?;
+let bootstrap = repo.bootstrap_table_schema_for_dev(&registry).await?;
+```
+
+Use generated artifacts as migration input for production tooling such as Atlas.
+Reserve `bootstrap_table_schema_for_dev` for tests and local development.
+
 ## Bomberman And Document Views
 
 Bomberman `BoardView` is intentionally a document-row read model. It stores a

src/bus/mod.rs

@@ -34,7 +34,8 @@
 //!
 //! // 2. Worker drains outbox and publishes via bus
 //! let bus = Bus::new(kafka_publisher, kafka_subscriber);
-//! for msg in repo.claim_outbox_messages(...) {
+//! let outbox = repo.outbox_store();
+//! for msg in outbox.claim(...) {
 //!     let event = Event::new(msg.id(), &msg.event_type, msg.payload.clone());
 //!     bus.publish(event)?;
 //! }

src/hashmap_repo/mod.rs

@@ -1,3 +1,3 @@
 mod repository;
 
-pub use repository::HashMapRepository;
+pub use repository::{HashMapOutboxStore, HashMapRepository};

src/hashmap_repo/repository.rs

@@ -38,6 +38,12 @@ pub struct HashMapRepository {
     snapshot_store: InMemorySnapshotStore,
 }
 
+/// In-memory outbox table handle.
+#[derive(Clone)]
+pub struct HashMapOutboxStore {
+    pub(crate) storage: Arc<RwLock<HashMap<String, OutboxMessage>>>,
+}
+
 impl Default for HashMapRepository {
     fn default() -> Self {
         Self::new()
@@ -55,10 +61,18 @@ impl HashMapRepository {
         }
     }
 
-    pub(crate) fn outbox_store(&self) -> &RwLock<HashMap<String, OutboxMessage>> {
+    #[cfg(test)]
+    pub(crate) fn outbox_storage(&self) -> &RwLock<HashMap<String, OutboxMessage>> {
         self.outbox_store.as_ref()
     }
 
+    /// Access the in-memory outbox table handle.
+    pub fn outbox_store(&self) -> HashMapOutboxStore {
+        HashMapOutboxStore {
+            storage: Arc::clone(&self.outbox_store),
+        }
+    }
+
     /// Access the embedded read model store directly.
     pub fn model_store(&self) -> &InMemoryReadModelStore {
         &self.model_store
@@ -375,6 +389,8 @@ fn reject_duplicate_async_streams(streams: &[AsyncStreamWrite<'_>]) -> Result<()
 fn reject_duplicate_outbox_messages(messages: &[OutboxMessage]) -> Result<(), RepositoryError> {
     let mut seen = HashSet::with_capacity(messages.len());
     for message in messages {
+        crate::outbox::validate_outbox_message_table_write(message)
+            .map_err(|err| RepositoryError::Model(err.to_string()))?;
         let id = message.id();
         if id.trim().is_empty() {
             return Err(RepositoryError::Model(

src/lib.rs

@@ -26,6 +26,7 @@ pub mod snapshot;
 pub mod sqlite_repo;
 #[cfg(any(feature = "postgres", feature = "sqlite"))]
 mod sqlx_repo;
+pub mod table;
 
 // Re-export entity types at crate root for convenience
 pub use entity::{
@@ -38,11 +39,11 @@ pub type SourcedResult<T = ()> = std::result::Result<T, EventRecordError>;
 
 // Re-export repository traits at crate root for convenience
 pub use repository::{
-    AsyncCommitBatch, AsyncGetStream, AsyncOutboxRepositoryExt, AsyncReadModelSessionStore,
-    AsyncReadModelStore, AsyncRelationalReadModelQueryStore, AsyncRepository, AsyncSnapshotStore,
-    AsyncSnapshotWrite, AsyncStreamWrite, AsyncTransactionalCommit, Commit, CommitBatch, Get,
-    GetMany, GetOne, Gettable, PreparedEventAppend, Repository, RepositoryError, SnapshotWrite,
-    StreamIdentity, TransactionalCommit,
+    AsyncCommitBatch, AsyncGetStream, AsyncReadModelSessionStore, AsyncReadModelStore,
+    AsyncRelationalReadModelQueryStore, AsyncRepository, AsyncSnapshotStore, AsyncSnapshotWrite,
+    AsyncStreamWrite, AsyncTransactionalCommit, Commit, CommitBatch, Get, GetMany, GetOne,
+    Gettable, PreparedEventAppend, Repository, RepositoryError, SnapshotWrite, StreamIdentity,
+    TransactionalCommit,
 };
 
 // Re-export aggregate types at crate root for convenience
@@ -51,31 +52,35 @@ pub use aggregate::{
     AsyncAggregateRepository, CommitAggregate, GetAggregate, GetAllAggregates, RepositoryExt,
 };
 
-pub use hashmap_repo::HashMapRepository;
+pub use hashmap_repo::{HashMapOutboxStore, HashMapRepository};
 #[cfg(feature = "postgres")]
-pub use postgres_repo::PostgresRepository;
+pub use postgres_repo::{PostgresOutboxStore, PostgresRepository};
 #[cfg(feature = "sqlite")]
-pub use sqlite_repo::SqliteRepository;
+pub use sqlite_repo::{SqliteOutboxStore, SqliteRepository};
 
 // Re-export lock traits and types at crate root for convenience
 pub use lock::{InMemoryLock, InMemoryLockManager, Lock, LockError, LockManager};
 
 // Outbox: commit concerns (aggregate + outbox in one commit)
 pub use outbox::{
-    AsyncOutboxCommit, OutboxCommit, OutboxCommitExt, OutboxMessage, OutboxMessageStatus,
+    outbox_message_insert_plan, outbox_message_key, outbox_message_row_values,
+    outbox_message_schema, AsyncOutboxCommit, OutboxCommit, OutboxCommitExt, OutboxMessage,
+    OutboxMessageStatus, OUTBOX_MESSAGES_TABLE,
 };
 
 // Outbox Worker: drain and publish concerns
 pub use outbox_worker::{
+    AsyncOutboxStore,
+    ClaimOutboxMessages,
     // Worker
     DrainResult,
     // Publishers
     LogPublisher,
     LogPublisherError,
+    OutboxClaimRef,
     OutboxPublishFailureAction,
     OutboxPublisher,
-    // Repository extension for claiming/completing messages
-    OutboxRepositoryExt,
+    OutboxStore,
     OutboxWorker,
     ProcessOneResult,
 };
@@ -123,6 +128,17 @@ pub use read_model::{
     DEFAULT_READ_MODEL_VERSION_COLUMN,
 };
 
+// Neutral table/row primitives shared by read models and operational tables.
+pub use table::{
+    generate_table_migration_artifacts, table_schema_bootstrap_result, table_schema_statements,
+    DeleteTableRowMutation, PatchTableRowMutation, TableAdapterCapabilities, TableColumn,
+    TableCommitOutcome, TableDocumentMutation, TableIndex, TableMigrationArtifact, TableModel,
+    TableMutation, TableRowMutation, TableSchema, TableSchemaAdapter,
+    TableSchemaAdapterCapabilities, TableSchemaBootstrap, TableSchemaIssue, TableSchemaIssueKind,
+    TableSchemaRegistry, TableSchemaRegistryExt, TableSchemaVerification, TableSqlDialect,
+    TableSqlSchemaAdapter, TableStoreError, TableWritePlan, DEFAULT_TABLE_VERSION_COLUMN,
+};
+
 // CommitBuilder: transactional batches of read models, outbox, and aggregates
 pub use commit_builder::{
     CommitBuilder, CommitBuilderExt, ReadModelSessionCommitExt, StagedCommitBuilder,

src/outbox/commit.rs

@@ -87,7 +87,7 @@ mod tests {
     use super::*;
     use crate::{
         impl_aggregate, AggregateBuilder, CommitBatch, Entity, EventRecord, HashMapRepository,
-        OutboxRepositoryExt, TransactionalCommit,
+        OutboxStore, TransactionalCommit,
     };
     use std::cell::RefCell;
 
@@ -145,7 +145,7 @@ mod tests {
 
         repo.outbox(event).commit(&mut aggregate).unwrap();
 
-        let pending = repo.repo().outbox_messages_pending().unwrap();
+        let pending = repo.repo().outbox_store().pending().unwrap();
         assert_eq!(pending.len(), 1);
         assert_eq!(pending[0].id(), "msg-1");
     }

src/outbox/mod.rs

@@ -39,9 +39,15 @@
 
 mod commit;
 mod message;
+mod table;
 
 // Outbox message record
 pub use message::{OutboxMessage, OutboxMessageStatus};
+pub(crate) use table::validate_outbox_message_table_write;
+pub use table::{
+    outbox_message_insert_plan, outbox_message_key, outbox_message_row_values,
+    outbox_message_schema, OUTBOX_MESSAGES_TABLE,
+};
 
 // Commit helpers
 pub use commit::{AsyncOutboxCommit, OutboxCommit, OutboxCommitExt};