feat(event-log): persist domain events

Author: outlndrr

CHANGELOG.md

@@ -35,6 +35,7 @@
 - added CommandBus idempotency ack persistence with a 24-hour duplicate window.
 - added opt-in `LUMMY_COMMAND_BUS_ENABLED` server flag for HTTP write routes.
 - widened the internal EventBus payload to shared domain-event envelopes.
+- added an internal domain event log writer subscribed to EventBus.
 
 ### Added
 - Gleam project bootstrap for `lummy_agent`

docs/refactors/0004-review-event-log-writer.md

@@ -0,0 +1,25 @@
+# Review — Step 5.2 EventLog writer
+
+Scope: `EventBus.subscribe_all`, `eventlog/event_log.gleam`, `domain_event_log` migration/repository, app supervisor wiring, tests.
+
+## Findings
+
+- No blocking correctness issues found in the landed slice.
+- Existing legacy `event_log` table is still used for session replay/SSE backlog, so the new append-only domain event table is named `domain_event_log`. This avoids destructive rename work in this phase.
+- `domain_event_log` uses `(aggregate_id, sequence)` primary key and `INSERT OR IGNORE`, so duplicate publish/retry does not double-write.
+- EventLog writer subscribes with `subscribe_all_named`; existing session-specific subscribers still receive only their aggregate events.
+- Writer is supervised after EventBus in app supervisor.
+
+## Follow-up
+
+- Writer message type is the EventBus subscription message, so it has no explicit `Shutdown` call surface. Acceptable for supervised internal subscriber now; revisit if tests need direct stop semantics.
+- Bounded mailbox/backpressure logging is still deferred per task note.
+- `domain_event_log` currently exposes count/append only. Replay/query APIs belong with later projection/backfill phases.
+
+## Validation
+
+- `gleam format --check && gleam check` — pass
+- `bash scripts/with-zig-env.sh gleam test` — pass, 188 tests
+- `cd shared && gleam format --check && gleam check && gleam test` — pass, 11 tests
+- `cd ui && gleam format --check && gleam check && gleam test` — pass, 13 tests
+- guard grep for `should.equal`, `io.debug`, `todo` — pass

docs/refactors/0004-tasks.md

@@ -251,13 +251,17 @@ Migrate pure domain types out of `src/` into `shared/`. Use re-export shims to k
 
 ### 5.2 EventLog table + writer actor
 
-- [ ] Migration: `event_log` table (see plan §8 schema).
-- [ ] Migration: index on `(aggregate_id, sequence)` PK + `occurred_at`.
-- [ ] New module `agent/eventlog/event_log.gleam`:
-  - [ ] Subscribes to EventBus on init.
-  - [ ] On received envelope, encodes payload via codec from §3.3, appends to table.
+- [x] Migration: `event_log` table (see plan §8 schema).
+  - Implemented as additive `domain_event_log` because legacy session replay already owns `event_log`.
+- [x] Migration: index on `(aggregate_id, sequence)` PK + `occurred_at`.
+  - `(aggregate_id, sequence)` primary key plus `domain_event_log_occurred_at_idx`.
+- [x] New module `agent/eventlog/event_log.gleam`:
+  - [x] Subscribes to EventBus on init.
+    - EventBus now has `subscribe_all_named` for event-log fan-out.
+  - [x] On received envelope, encodes payload via codec from §3.3, appends to table.
   - [ ] Bounded mailbox: above water mark, log warning, drop projection writes (NEVER drop event_log writes).
-- [ ] Supervised under app supervisor, after StorageOwner and EventBus.
+    - Deferred: writer is one DB insert per event; add mailbox telemetry before projections fan out.
+- [x] Supervised under app supervisor, after StorageOwner and EventBus.
 
 ### 5.3 Backfill historical events
 
@@ -504,7 +508,7 @@ For each major step, fill in:
 | 4.6 | | 2026-04-29 | | Added enabled-vs-direct HTTP write route parity coverage. |
 | 4.7 | | | | |
 | 5.1 | | 2026-04-29 | | EventBus widened to shared DomainEvent envelopes; SSE ignores non-session events. |
-| 5.2 | | | | |
+| 5.2 | | 2026-04-29 | | Added domain event log table/repository and EventLog writer subscriber. |
 | 5.3 | | | | |
 | 5.4 | | | | |
 | 5.5 | | | | |

src/lummy_agent/app/supervisor.gleam

@@ -2,6 +2,7 @@ import gleam/erlang/process
 import gleam/otp/actor
 import gleam/otp/static_supervisor as supervisor
 import lummy_agent/app/config as app_config
+import lummy_agent/eventlog/event_log
 import lummy_agent/observability/metrics
 import lummy_agent/observability/otel
 import lummy_agent/run/run_supervisor
@@ -44,6 +45,11 @@ pub fn start(
       session_manager_name: session_manager_name,
       repositories: repositories,
     )
+  let event_log_start_arg =
+    event_log.StartArg(
+      event_bus_name: event_bus_name,
+      repositories: repositories,
+    )
 
   let http_dependencies =
     http_server.Dependencies(
@@ -59,6 +65,7 @@ pub fn start(
     config.storage.database_path,
   ))
   |> supervisor.add(event_bus.supervised(event_bus_name))
+  |> supervisor.add(event_log.supervised(event_log_start_arg))
   |> supervisor.add(session_supervisor.supervised(session_supervisor_name))
   |> supervisor.add(run_supervisor.supervised(run_supervisor_name))
   |> supervisor.add(session_manager.supervised(

src/lummy_agent/eventlog/event_log.gleam

@@ -0,0 +1,55 @@
+import gleam/erlang/process
+import gleam/otp/actor
+import gleam/otp/supervision
+import lummy_agent/domain/error as domain_error
+import lummy_agent/runtime/event_bus
+import lummy_agent/storage/repository
+
+pub type StartArg {
+  StartArg(
+    event_bus_name: event_bus.Name,
+    repositories: repository.Repositories,
+  )
+}
+
+pub type Handle =
+  process.Subject(Message)
+
+pub type Message =
+  event_bus.SubscriptionMessage
+
+type State {
+  State(repositories: repository.Repositories)
+}
+
+pub fn start(start_arg: StartArg) -> actor.StartResult(process.Subject(Message)) {
+  actor.new_with_initialiser(1000, fn(subject) {
+    event_bus.subscribe_all_named(start_arg.event_bus_name, subject)
+    Ok(
+      actor.initialised(State(repositories: start_arg.repositories))
+      |> actor.returning(subject),
+    )
+  })
+  |> actor.on_message(handle_message)
+  |> actor.start
+}
+
+pub fn supervised(start_arg: StartArg) {
+  supervision.worker(fn() { start(start_arg) })
+}
+
+fn handle_message(state: State, message: Message) -> actor.Next(State, Message) {
+  case message {
+    event_bus.Published(envelope) -> {
+      let _ = append_event(state.repositories.domain_event_log, envelope)
+      actor.continue(state)
+    }
+  }
+}
+
+fn append_event(
+  repo: repository.DomainEventLogRepository,
+  envelope,
+) -> Result(Nil, domain_error.DomainError) {
+  repository.append_domain_event_log(repo, envelope)
+}

src/lummy_agent/runtime/event_bus.gleam

@@ -16,6 +16,10 @@ pub type Message {
     subscriber: process.Subject(SubscriptionMessage),
     reply_with: process.Subject(Nil),
   )
+  SubscribeAll(
+    subscriber: process.Subject(SubscriptionMessage),
+    reply_with: process.Subject(Nil),
+  )
   Unsubscribe(
     session_id: domain_id.SessionId,
     subscriber: process.Subject(SubscriptionMessage),
@@ -32,11 +36,14 @@ pub type Name =
   process.Name(Message)
 
 type State {
-  State(subscriptions: Dict(String, List(process.Subject(SubscriptionMessage))))
+  State(
+    subscriptions: Dict(String, List(process.Subject(SubscriptionMessage))),
+    all_subscribers: List(process.Subject(SubscriptionMessage)),
+  )
 }
 
 pub fn start(name: Name) -> actor.StartResult(Handle) {
-  actor.new(State(subscriptions: dict.new()))
+  actor.new(State(subscriptions: dict.new(), all_subscribers: []))
   |> actor.named(name)
   |> actor.on_message(handle_message)
   |> actor.start
@@ -60,6 +67,15 @@ pub fn subscribe_named(
   })
 }
 
+pub fn subscribe_all_named(
+  name: Name,
+  subscriber: process.Subject(SubscriptionMessage),
+) -> Nil {
+  actor.call(named_handle(name), waiting: 5000, sending: fn(reply) {
+    SubscribeAll(subscriber:, reply_with: reply)
+  })
+}
+
 pub fn unsubscribe_named(
   name: Name,
   session_id: domain_id.SessionId,
@@ -88,7 +104,20 @@ fn handle_message(state: State, message: Message) -> actor.Next(State, Message)
       }
       actor.send(reply_with, Nil)
       actor.continue(
-        State(subscriptions: dict.insert(state.subscriptions, key, subscribers)),
+        State(
+          ..state,
+          subscriptions: dict.insert(state.subscriptions, key, subscribers),
+        ),
+      )
+    }
+
+    SubscribeAll(subscriber, reply_with) -> {
+      actor.send(reply_with, Nil)
+      actor.continue(
+        State(
+          ..state,
+          all_subscribers: ensure_subscriber(state.all_subscribers, subscriber),
+        ),
       )
     }
 
@@ -100,7 +129,10 @@ fn handle_message(state: State, message: Message) -> actor.Next(State, Message)
       }
       actor.send(reply_with, Nil)
       actor.continue(
-        State(subscriptions: dict.insert(state.subscriptions, key, subscribers)),
+        State(
+          ..state,
+          subscriptions: dict.insert(state.subscriptions, key, subscribers),
+        ),
       )
     }
 
@@ -111,6 +143,7 @@ fn handle_message(state: State, message: Message) -> actor.Next(State, Message)
         Error(_) -> []
       }
       subscribers
+      |> list.append(state.all_subscribers)
       |> list.each(fn(subscriber) {
         actor.send(subscriber, Published(envelope))
       })

src/lummy_agent/storage/migration.gleam

@@ -9,11 +9,36 @@ pub fn all() -> List(Migration) {
     Migration(version: 1, name: "0001_initial", sql: initial_sql()),
     Migration(version: 2, name: "0002_session_pinned_at", sql: pinned_at_sql()),
     Migration(version: 3, name: "0003_command_acks", sql: command_acks_sql()),
+    Migration(
+      version: 4,
+      name: "0004_domain_event_log",
+      sql: domain_event_log_sql(),
+    ),
   ]
 }
 
 pub fn latest_version() -> Int {
-  3
+  4
+}
+
+fn domain_event_log_sql() -> String {
+  string.join(
+    [
+      "CREATE TABLE IF NOT EXISTS domain_event_log (",
+      "  aggregate_id TEXT NOT NULL,",
+      "  sequence INTEGER NOT NULL,",
+      "  occurred_at INTEGER NOT NULL,",
+      "  causation_id TEXT,",
+      "  correlation_id TEXT,",
+      "  event_name TEXT NOT NULL,",
+      "  payload TEXT NOT NULL,",
+      "  PRIMARY KEY (aggregate_id, sequence)",
+      ");",
+      "CREATE INDEX IF NOT EXISTS domain_event_log_occurred_at_idx",
+      "  ON domain_event_log(occurred_at);",
+    ],
+    with: "\n",
+  )
 }
 
 fn command_acks_sql() -> String {

src/lummy_agent/storage/repository.gleam

@@ -6,6 +6,7 @@ import lummy_agent/domain/run
 import lummy_agent/domain/session
 import lummy_agent/domain/tool_call
 import lummy_agent/session/event as session_event
+import lummy_agent_shared/event/envelope as shared_envelope
 
 pub type SessionRepository {
   SessionRepository(
@@ -88,6 +89,13 @@ pub type CommandAckRepository {
   )
 }
 
+pub type DomainEventLogRepository {
+  DomainEventLogRepository(
+    append: fn(shared_envelope.EventEnvelope) -> Result(Nil, error.DomainError),
+    count: fn() -> Result(Int, error.DomainError),
+  )
+}
+
 pub type Repositories {
   Repositories(
     sessions: SessionRepository,
@@ -97,6 +105,7 @@ pub type Repositories {
     events: EventRepository,
     commits: CommitRepository,
     command_acks: CommandAckRepository,
+    domain_event_log: DomainEventLogRepository,
   )
 }
 
@@ -252,6 +261,19 @@ pub fn insert_command_ack(
   repo.insert(command_id, idempotency_key, accepted_at, accepted_at_epoch)
 }
 
+pub fn append_domain_event_log(
+  repo: DomainEventLogRepository,
+  envelope: shared_envelope.EventEnvelope,
+) -> Result(Nil, error.DomainError) {
+  repo.append(envelope)
+}
+
+pub fn count_domain_event_log(
+  repo: DomainEventLogRepository,
+) -> Result(Int, error.DomainError) {
+  repo.count()
+}
+
 pub fn list_events_for_session(
   repo: EventRepository,
   session_id: id.SessionId,

src/lummy_agent/storage/sqlite.gleam

@@ -16,6 +16,7 @@ import lummy_agent/storage/migration
 import lummy_agent/storage/repository
 import lummy_agent/storage/sqlite_commit as commit_db
 import lummy_agent/storage/sqlite_connection.{type Connection}
+import lummy_agent/storage/sqlite_domain_event_log as domain_event_log_db
 import lummy_agent/storage/sqlite_event as event_db
 import lummy_agent/storage/sqlite_lowlevel as lowlevel
 import lummy_agent/storage/sqlite_message as message_db
@@ -96,6 +97,10 @@ pub type OwnerMessage {
     operation: fn(Connection) -> StorageResult(option.Option(String)),
     reply_with: process.Subject(StorageResult(option.Option(String))),
   )
+  RunInt(
+    operation: fn(Connection) -> StorageResult(Int),
+    reply_with: process.Subject(StorageResult(Int)),
+  )
   Shutdown(reply_with: process.Subject(Nil))
 }
 
@@ -364,6 +369,20 @@ pub fn owner_repositories(name: Name) -> repository.Repositories {
         })
       },
     ),
+    domain_event_log: repository.DomainEventLogRepository(
+      append: fn(envelope) {
+        call_owner_nil(name, fn(connection) {
+          tx.with_transaction_connection(connection, fn(connection) {
+            domain_event_log_db.append_event(connection, envelope)
+          })
+        })
+      },
+      count: fn() {
+        call_owner_int(name, fn(connection) {
+          domain_event_log_db.count_events(connection)
+        })
+      },
+    ),
   )
 }
 
@@ -432,6 +451,11 @@ fn handle_owner_message(
       actor.continue(state)
     }
 
+    RunInt(operation, reply_with) -> {
+      actor.send(reply_with, operation(state.connection))
+      actor.continue(state)
+    }
+
     Shutdown(reply_with) -> {
       sqlite_connection.close(state.connection)
       actor.send(reply_with, Nil)
@@ -596,6 +620,15 @@ fn call_owner_optional_string(
   })
 }
 
+fn call_owner_int(
+  name: Name,
+  operation: fn(Connection) -> StorageResult(Int),
+) -> StorageResult(Int) {
+  actor.call(named_handle(name), waiting: owner_timeout_ms, sending: fn(reply) {
+    RunInt(operation: operation, reply_with: reply)
+  })
+}
+
 fn find_recent_command_ack_connection(
   connection: Connection,
   idempotency_key: String,