86: Add AssetAZ Journal Service with durable PDE event journal and replayable consumers

2026-10-04 19:42:21 +02:00
parent f1dd5ed9b1
commit 035a01c48c
34 changed files with 1172 additions and 0 deletions
@@ -6,3 +6,6 @@ FORMAT_VERSION=1
 [AUTO_START]
 nenjim.registry.service
 nenjim.journal.service
+
+# Optional AssetAZ fact journal (separate from the mandatory Nenjim artifact journal).
+# assetaz.journal.service
@@ -0,0 +1,2 @@
+schema: spec-driven
+created: 2026-09-09
@@ -0,0 +1,32 @@
+## Context
+
+See proposal.md. The authoritative baseline is Gitea 0.1-dev at f1dd5ed. Nenjim separates public component APIs from impl.ref processes; the Hub owns explicit construction and registry registration. The artifact JournalService is unrelated.
+
+## Goals / Non-Goals
+
+Implement the two capability contracts. Keep primitive PDE parsing below event codecs and record integrity. Do not extend the v1 scope in the proposal.
+
+## Decisions
+
+- Place events in com.r35157.assetaz.events, service API in services.journal, codecs under services.journal.codec and implementations under impl.ref. PDE belongs to com.r35157.libs.pde. Codec components expose read identity, domain type and whether they encode current writes; historical read-only codecs can coexist without implicit highest-version selection.
+- The Hub registers explicit codec instances and offers assetaz.journal.service through its process constructor catalogue. The journal resolves the registry catalogue at startup, validates duplicate identities/current encoders, and uses a single lifetime file channel and exclusive file lock. No service restart API is added.
+- One journal monitor serializes durable append, registration and queue publication. Each subscription has its own monitor, FIFO and worker. Every publication carries a sequence, including irrelevant records, so workers advance only after examining the ordered stream. Consumer code never runs under the journal monitor. An empty queue observed under the subscription monitor selects the ONLINE cutover; subsequent publications queue behind it while the same worker calls ONLINE.
+- Historical readers scan from the beginning to validate chain state but invoke callbacks only from replayFromSequence through captured registrationHead. This avoids adding indexes or snapshots. The reader closes before pending live records drain.
+- PDE uses the published little-endian integer/length type table; UUID conversion alone is big endian at codec level. Record hashes operate on original bounded ranges, not canonicalized representations.
+- JournalFormatVersionChanged has common event fields followed by a target version. Only format 1 is implemented; unsupported targets fail before append and during loading. Future parsers can switch after the ordinary event without a new physical record type.
+- Barriers are interruptible (InterruptedException alongside DegradedException); no timeout or recovery API is introduced.
+
+## Risks / Trade-offs
+
+- Slow consumers can accumulate FIFO entries → intentional independent v1 workers; no unrequested tuning or overflow policy.
+- Callbacks cannot be rolled back by the journal → document consumer atomic update responsibility and preserve safe cursor on failure.
+- Uncertain force/write can leave a complete or partial record → permanently fail closed and validate the authoritative file next JVM; no automatic repair.
+- Hash chains do not prevent a writer recomputing history → external anchors remain out of scope.
+
+## Migration Plan
+
+New optional process, using data/assetaz/journal/events.pde. No migration of the Nenjim artifact journal or existing service histories. Add the process ID to the example configuration; existing user configurations remain opt-in. No commits or publication in this task.
+
+## ValueTags
+
+The owner name uses the existing Name tag (String). No shared Detag definitions are changed. Proposed future tags, pending separate approval, are AssetAZJournalSequence (parent long, Java long) for authoritative cursors, AssetAZEventType (parent String, Java String) for persistent short identifiers, AssetAZEventVersion and AssetAZJournalFormatVersion (parent int, Java int) to distinguish schema and envelope versions, and AssetOwnerId and AssetAZCorrelationId (parent UUID, Java UUID) to distinguish owner identity from workflow grouping. This implementation keeps the issue's primitive/UUID signatures where these tags do not yet exist.
@@ -0,0 +1,28 @@
+## Why
+
+AssetAZ needs an authoritative durable event history from which services can reconstruct their models and establish causal freshness. Gitea issue #86 defines a deliberately strict first version.
+
+## What Changes
+
+- Add an append-only PDE journal with SHA-256 integrity, durable sequence commitment and fail-closed writes.
+- Add domain events and registry-provided positional event codecs, independent of Java class naming.
+- Add independent virtual-thread subscriptions, replay, exact filtering, progress barriers and isolated degradation.
+- Add reusable bounded PDE primitive encoding and decoding.
+
+## Capabilities
+
+### New Capabilities
+- `assetaz-journal`: Persistence, event schemas, replay and consumer consistency contracts.
+- `pde-encoding`: Minimal bounded PDE primitives and object framing.
+
+### Modified Capabilities
+
+None.
+
+## Impact
+
+New AssetAZ API and reference implementation packages, a reusable PDE library, and explicit codec/process entries in the existing Hub composition root. No new dependencies. The Nenjim artifact journal is separate and unchanged.
+
+## Non-goals
+
+No snapshots, repair, truncation, retries, consumer recovery, unregister, restart support, adaptive limits, pools, generalized serialization, reflection discovery, migration, PDL, signatures, causation IDs or unrelated thread changes.
@@ -0,0 +1,68 @@
+## Purpose
+
+Define the authoritative AssetAZ fact journal and independent consumer reconstruction and causal freshness contracts.
+
+## ADDED Requirements
+
+### Requirement: Durable authoritative ordering
+The journal SHALL be append-only and serialize appends, assigning contiguous sequences from 1 independently of occurredAt. Fields 2..6 SHALL be encoded once, hashed, and copied into the final record. Only successful FileChannel.force(true) SHALL commit the candidate head, publish and return its sequence. Missing append codecs and oversized records MUST fail before any append. An uncertain write SHALL permanently fail closed in that JVM without retries, repair or truncation.
+
+#### Scenario: Failed force
+- **WHEN** persistence may have left an uncertain tail
+- **THEN** the candidate is not committed or published and all later appends fail.
+
+### Requirement: Record integrity and bounded readers
+Each record SHALL be exactly one top-level PDE OBJECT containing checksum, previousChecksum, sequence, stable short eventType, eventVersion and OBJECT payload in that order. Both hashes SHALL be 32 raw bytes. SHA-256 SHALL cover the original persisted fields 2..6; the first previousChecksum SHALL be SHA-256(empty). Each independent reader SHALL reuse one fixed 1024-byte buffer. The entire record including header MUST be at most 1024 bytes; reading SHALL reject declared excess or truncation immediately after framing and before payload decoding. Startup SHALL validate the entire history, checksums, chain, contiguous sequences, PDE, known codecs and exact schemas before succeeding.
+
+#### Scenario: Corrupt history
+- **WHEN** a checksum, chain, sequence, schema, type/version or framing cannot be verified
+- **THEN** startup fails with the rejection layer identified; no record is skipped.
+
+### Requirement: Event and codec contract
+Events SHALL represent facts, not commands. occurredAt SHALL be semantic time; note and correlationId are nullable. Correlation groups independently meaningful workflow events without atomicity or expected count; inseparable changes belong in one event. Codecs SHALL be registry components resolved by stable eventType and persistent version, never constructed in the journal. Domain class names MUST NOT encode persistent versions. Journal integrity, PDE validity and event schema validity SHALL remain separate layers.
+
+#### Scenario: Historical codec
+- **WHEN** a known historical persistent version is read
+- **THEN** its codec yields the current domain event without changing the domain type name.
+
+### Requirement: Owner payload
+AssetOwnerRegistered version 1 SHALL contain exactly seven positional values: signed long epoch seconds, nano-of-second, nullable UTF-8 note, nullable UUID correlationId, required UUID assetOwnerId, required textual assetOwnerType, required UTF-8 name. UUIDs SHALL be exactly 16 bytes, most-significant then least-significant long, both big endian. Nanos MUST be 0..999999999; owner type MUST be NATURAL_PERSON or LEGAL_PERSON, never ordinal. Required values, Instant range and exact field count SHALL be validated by the codec. No additional owner metadata SHALL be added.
+
+#### Scenario: Extra payload value
+- **WHEN** an otherwise valid owner payload has an eighth value
+- **THEN** schema validation rejects it.
+
+### Requirement: Format changes
+New journals SHALL implicitly start in format 1 with no header. JournalFormatVersionChanged SHALL be an ordinary event in the same envelope and hash chain, with a positive target version and common event fields. A change applies to subsequent records; returning to an earlier supported version SHALL be representable. Unsupported formats MUST fail closed; v1 implements only format 1.
+
+#### Scenario: Chained format event
+- **WHEN** a format-1 change to format 1 is appended
+- **THEN** it is checksummed and replayed normally; unsupported target formats are rejected before append.
+
+### Requirement: Registration and independent replay
+Registration SHALL atomically capture registrationHead, validate 1 <= replayFromSequence <= registrationHead + 1, defensively copy a non-null filter without null members and reject duplicate consumer identity (==). Empty filters SHALL mean all; other filters SHALL match exact runtime class only. Registration SHALL return without awaiting replay. Each subscription SHALL own a virtual-thread worker and independent reader, replay through registrationHead, close the reader and drain subsequent FIFO publications. Degraded registrations remain registered. registrationHead MUST NOT be public API.
+
+#### Scenario: Late subscriber
+- **WHEN** another consumer registers while an existing one processes events
+- **THEN** only the new consumer replays; both receive their own ordered stream without gaps or duplicates.
+
+### Requirement: Progress and barriers
+processedThrough SHALL mean all records through the cursor were successfully handled or examined as irrelevant, starting at replayFromSequence - 1. A barrier SHALL return whenever the safe cursor meets its minimum, including DEGRADED; otherwise wait while CATCHING_UP/ONLINE or throw DegradedException when DEGRADED. Progress and degradation SHALL wake waiters. The barrier MUST NOT imply a historical snapshot.
+
+#### Scenario: Filtered progress
+- **WHEN** a consumer last handled 1000 and examined irrelevant records through 1100
+- **THEN** its cursor is 1100 and a barrier for 1100 succeeds.
+
+### Requirement: States and atomic cutover
+Subscriptions SHALL begin CATCHING_UP, without requiring an initial state callback. After replay and pending events through an atomic cutover, the worker SHALL request ONLINE. Successful return establishes readiness to act; the callback should establish readiness without irreversible actions. Publication concurrent with cutover SHALL be processed before ONLINE or as the first normal event after ONLINE, with no gap. All callbacks for one consumer SHALL be sequential.
+
+#### Scenario: Append during readiness callback
+- **WHEN** a new event arrives at the chosen cutover
+- **THEN** it is queued and processed next after readiness succeeds.
+
+### Requirement: Isolated degradation and atomic models
+Callback or ONLINE transition failure SHALL retain the last safe cursor, mark only that subscription DEGRADED, stop delivery, wake barriers, and log the exception plus consumer identity, sequence and state. Consumers SHALL atomically apply each event or leave their externally visible model unchanged. Replay SHALL rebuild state without repeating trades, transfers or notifications. No OFFLINE, unregister or recovery SHALL be implemented.
+
+#### Scenario: Consumer throws
+- **WHEN** one subscriber fails applying sequence N
+- **THEN** it retains its safe cursor, future barriers beyond that cursor fail, and other subscribers continue.
@@ -0,0 +1,26 @@
+## Purpose
+
+Provide reusable primitive binary encoding and bounded structural decoding for positional AssetAZ journal data.
+
+## ADDED Requirements
+
+### Requirement: Primitive representation
+The PDE layer SHALL implement Jenkov PDE integers, UTF-8, bytes, typed null and length-delimited OBJECT framing only. Integer magnitudes and lengths use little endian; negative integers encode -(value + 1). It MUST reject wrong types, malformed lengths, invalid UTF-8 and values outside the requested Java int/long range.
+
+#### Scenario: Integer extremes
+- **WHEN** signed long extrema and nullable strings/bytes are encoded
+- **THEN** they round-trip without loss and null is distinguished from empty.
+
+### Requirement: Bounded decoding
+Every decoder SHALL operate on byte[], offset and length; child objects SHALL have independent logical bounds. requireEnd SHALL reject unconsumed values using the logical limit, never backing-array size. Framing SHALL expose total object size before payload decoding.
+
+#### Scenario: Reused backing storage
+- **WHEN** a short record follows a long record in one buffer
+- **THEN** stale bytes are invisible and truncated children are rejected.
+
+### Requirement: Abstraction boundary
+PDE SHALL know nothing about UUID, Instant, events, sequence numbers or checksums. It MUST NOT implement object mapping, reflection, annotation schemas or PDL.
+
+#### Scenario: Domain payload decoding
+- **WHEN** a UUID byte value is decoded
+- **THEN** PDE validates its binary framing while the codec validates its 16-byte schema.
@@ -0,0 +1,23 @@
+## 1. Encoding and contracts
+
+- [x] 1.1 Implement bounded PDE primitives and framing.
+- [x] 1.2 Add event/service contracts and registry codecs with exact schemas.
+
+## 2. Journal and consumers
+
+- [x] 2.1 Implement record hashing, bounded readers, startup validation and durable fail-closed append.
+- [x] 2.2 Implement independent virtual-thread replay, FIFO delivery, cutover and barriers.
+- [x] 2.3 Wire passive codecs and journal process in the Hub.
+
+## 3. Verification
+
+- [x] 3.1 Run build and existing tests; verify issue invariants and diagnostics.
+- [x] 3.2 Validate OpenSpec and review unstaged changes against scope.
+
+## Validation results
+
+- Normal Gradle build passed, including all four existing tests (zero failures/errors/skips).
+- 78 temporary checks outside the repository passed: PDE ranges and bounds, exact schemas, raw hashes, full startup validation, 1024/1025-byte boundary, replay/live ordering, filtering, safe cursors, barriers, isolated degradation and uncertain force failure before publication.
+- OpenSpec strict validation and git diff whitespace checks passed. No repository tests were added, following project policy.
+- The fresh checkout omitted ignored conf/detag.conf; build validation used a local symlink to the existing configuration without changing its definitions, and a workspace-local Gradle cache.
+- All changes remain unstaged and uncommitted on the fetched Gitea baseline.
@@ -0,0 +1,69 @@
+# assetaz-journal Specification
+
+## Purpose
+Define the authoritative AssetAZ fact journal and independent consumer reconstruction and causal freshness contracts.
+
+## Requirements
+
+### Requirement: Durable authoritative ordering
+The journal SHALL be append-only and serialize appends, assigning contiguous sequences from 1 independently of occurredAt. Fields 2..6 SHALL be encoded once, hashed, and copied into the final record. Only successful FileChannel.force(true) SHALL commit the candidate head, publish and return its sequence. Missing append codecs and oversized records MUST fail before any append. An uncertain write SHALL permanently fail closed in that JVM without retries, repair or truncation.
+
+#### Scenario: Failed force
+- **WHEN** persistence may have left an uncertain tail
+- **THEN** the candidate is not committed or published and all later appends fail.
+
+### Requirement: Record integrity and bounded readers
+Each record SHALL be exactly one top-level PDE OBJECT containing checksum, previousChecksum, sequence, stable short eventType, eventVersion and OBJECT payload in that order. Both hashes SHALL be 32 raw bytes. SHA-256 SHALL cover the original persisted fields 2..6; the first previousChecksum SHALL be SHA-256(empty). Each independent reader SHALL reuse one fixed 1024-byte buffer. The entire record including header MUST be at most 1024 bytes; reading SHALL reject declared excess or truncation immediately after framing and before payload decoding. Startup SHALL validate the entire history, checksums, chain, contiguous sequences, PDE, known codecs and exact schemas before succeeding.
+
+#### Scenario: Corrupt history
+- **WHEN** a checksum, chain, sequence, schema, type/version or framing cannot be verified
+- **THEN** startup fails with the rejection layer identified; no record is skipped.
+
+### Requirement: Event and codec contract
+Events SHALL represent facts, not commands. occurredAt SHALL be semantic time; note and correlationId are nullable. Correlation groups independently meaningful workflow events without atomicity or expected count; inseparable changes belong in one event. Codecs SHALL be registry components resolved by stable eventType and persistent version, never constructed in the journal. Domain class names MUST NOT encode persistent versions. Journal integrity, PDE validity and event schema validity SHALL remain separate layers.
+
+#### Scenario: Historical codec
+- **WHEN** a known historical persistent version is read
+- **THEN** its codec yields the current domain event without changing the domain type name.
+
+### Requirement: Owner payload
+AssetOwnerRegistered version 1 SHALL contain exactly seven positional values: signed long epoch seconds, nano-of-second, nullable UTF-8 note, nullable UUID correlationId, required UUID assetOwnerId, required textual assetOwnerType, required UTF-8 name. UUIDs SHALL be exactly 16 bytes, most-significant then least-significant long, both big endian. Nanos MUST be 0..999999999; owner type MUST be NATURAL_PERSON or LEGAL_PERSON, never ordinal. Required values, Instant range and exact field count SHALL be validated by the codec. No additional owner metadata SHALL be added.
+
+#### Scenario: Extra payload value
+- **WHEN** an otherwise valid owner payload has an eighth value
+- **THEN** schema validation rejects it.
+
+### Requirement: Format changes
+New journals SHALL implicitly start in format 1 with no header. JournalFormatVersionChanged SHALL be an ordinary event in the same envelope and hash chain, with a positive target version and common event fields. A change applies to subsequent records; returning to an earlier supported version SHALL be representable. Unsupported formats MUST fail closed; v1 implements only format 1.
+
+#### Scenario: Chained format event
+- **WHEN** a format-1 change to format 1 is appended
+- **THEN** it is checksummed and replayed normally; unsupported target formats are rejected before append.
+
+### Requirement: Registration and independent replay
+Registration SHALL atomically capture registrationHead, validate 1 <= replayFromSequence <= registrationHead + 1, defensively copy a non-null filter without null members and reject duplicate consumer identity (==). Empty filters SHALL mean all; other filters SHALL match exact runtime class only. Registration SHALL return without awaiting replay. Each subscription SHALL own a virtual-thread worker and independent reader, replay through registrationHead, close the reader and drain subsequent FIFO publications. Degraded registrations remain registered. registrationHead MUST NOT be public API.
+
+#### Scenario: Late subscriber
+- **WHEN** another consumer registers while an existing one processes events
+- **THEN** only the new consumer replays; both receive their own ordered stream without gaps or duplicates.
+
+### Requirement: Progress and barriers
+processedThrough SHALL mean all records through the cursor were successfully handled or examined as irrelevant, starting at replayFromSequence - 1. A barrier SHALL return whenever the safe cursor meets its minimum, including DEGRADED; otherwise wait while CATCHING_UP/ONLINE or throw DegradedException when DEGRADED. Progress and degradation SHALL wake waiters. The barrier MUST NOT imply a historical snapshot.
+
+#### Scenario: Filtered progress
+- **WHEN** a consumer last handled 1000 and examined irrelevant records through 1100
+- **THEN** its cursor is 1100 and a barrier for 1100 succeeds.
+
+### Requirement: States and atomic cutover
+Subscriptions SHALL begin CATCHING_UP, without requiring an initial state callback. After replay and pending events through an atomic cutover, the worker SHALL request ONLINE. Successful return establishes readiness to act; the callback should establish readiness without irreversible actions. Publication concurrent with cutover SHALL be processed before ONLINE or as the first normal event after ONLINE, with no gap. All callbacks for one consumer SHALL be sequential.
+
+#### Scenario: Append during readiness callback
+- **WHEN** a new event arrives at the chosen cutover
+- **THEN** it is queued and processed next after readiness succeeds.
+
+### Requirement: Isolated degradation and atomic models
+Callback or ONLINE transition failure SHALL retain the last safe cursor, mark only that subscription DEGRADED, stop delivery, wake barriers, and log the exception plus consumer identity, sequence and state. Consumers SHALL atomically apply each event or leave their externally visible model unchanged. Replay SHALL rebuild state without repeating trades, transfers or notifications. No OFFLINE, unregister or recovery SHALL be implemented.
+
+#### Scenario: Consumer throws
+- **WHEN** one subscriber fails applying sequence N
+- **THEN** it retains its safe cursor, future barriers beyond that cursor fail, and other subscribers continue.
@@ -0,0 +1,27 @@
+# pde-encoding Specification
+
+## Purpose
+Provide reusable primitive binary encoding and bounded structural decoding for positional AssetAZ journal data.
+
+## Requirements
+
+### Requirement: Primitive representation
+The PDE layer SHALL implement Jenkov PDE integers, UTF-8, bytes, typed null and length-delimited OBJECT framing only. Integer magnitudes and lengths use little endian; negative integers encode -(value + 1). It MUST reject wrong types, malformed lengths, invalid UTF-8 and values outside the requested Java int/long range.
+
+#### Scenario: Integer extremes
+- **WHEN** signed long extrema and nullable strings/bytes are encoded
+- **THEN** they round-trip without loss and null is distinguished from empty.
+
+### Requirement: Bounded decoding
+Every decoder SHALL operate on byte[], offset and length; child objects SHALL have independent logical bounds. requireEnd SHALL reject unconsumed values using the logical limit, never backing-array size. Framing SHALL expose total object size before payload decoding.
+
+#### Scenario: Reused backing storage
+- **WHEN** a short record follows a long record in one buffer
+- **THEN** stale bytes are invisible and truncated children are rejected.
+
+### Requirement: Abstraction boundary
+PDE SHALL know nothing about UUID, Instant, events, sequence numbers or checksums. It MUST NOT implement object mapping, reflection, annotation schemas or PDL.
+
+#### Scenario: Domain payload decoding
+- **WHEN** a UUID byte value is decoded
+- **THEN** PDE validates its binary framing while the codec validates its 16-byte schema.
@@ -0,0 +1,16 @@
+package com.r35157.assetaz.events;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import java.time.Instant;
+import java.util.UUID;
+
+/** A fact/state change, never a command. Sequence, not semantic time, orders facts.
+ * Replay rebuilds state without repeating external side effects. Inseparable changes
+ * belong in one event; correlation groups independent facts, without atomicity.
+ */
+public interface AssetAZEvent {
+    @NotNull Instant occurredAt();
+    @Nullable String note();
+    @Nullable UUID correlationId();
+}
@@ -0,0 +1,15 @@
+package com.r35157.assetaz.events;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import java.time.Instant;
+import java.util.UUID;
+
+public record AssetOwnerRegistered(
+        @NotNull Instant occurredAt,
+        @NotNull UUID assetOwnerId,
+        @NotNull AssetOwnerType assetOwnerType,
+        @NotNull ΩNameΩ name,
+        @Nullable String note,
+        @Nullable UUID correlationId
+) implements AssetAZEvent { }
@@ -0,0 +1,4 @@
+package com.r35157.assetaz.events;
+
+/** Juridical owner category. Persistent values are textual, never enum ordinals. */
+public enum AssetOwnerType { NATURAL_PERSON, LEGAL_PERSON }
@@ -0,0 +1,14 @@
+package com.r35157.assetaz.events;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import java.time.Instant;
+import java.util.UUID;
+
+/** Ordinary chained fact; target format takes effect for subsequent records. */
+public record JournalFormatVersionChanged(
+        @NotNull Instant occurredAt,
+        int journalFormatVersion,
+        @Nullable String note,
+        @Nullable UUID correlationId
+) implements AssetAZEvent { }
@@ -0,0 +1,16 @@
+package com.r35157.assetaz.services.journal;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import org.jetbrains.annotations.NotNull;
+
+/**
+ * Sequential callbacks on one subscription worker, identical for replay and live facts.
+ * Each event must update the externally visible model atomically, or leave it unchanged
+ * on failure. Rebuild state only during catch-up; never replay historical side effects.
+ */
+public interface AssetAZJournalConsumer {
+    void onEvent(long sequence, @NotNull AssetAZEvent event) throws Exception;
+    /** Successful ONLINE return establishes readiness to act. This callback should
+     * establish readiness without itself performing irreversible domain actions. */
+    void requestState(@NotNull AssetAZJournalConsumerState state) throws Exception;
+}
@@ -0,0 +1,3 @@
+package com.r35157.assetaz.services.journal;
+
+public enum AssetAZJournalConsumerState { CATCHING_UP, ONLINE, DEGRADED }
@@ -0,0 +1,19 @@
+package com.r35157.assetaz.services.journal;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import com.r35157.nenjim.component.NenjimComponent;
+import org.jetbrains.annotations.NotNull;
+import java.util.Set;
+
+/** Authoritative append-only fact journal. Lifecycle belongs to the Nenjim process view. */
+public interface AssetAZJournalService extends NenjimComponent {
+    /** Returns the committed global sequence only after durable persistence. */
+    long append(@NotNull AssetAZEvent event);
+    /** Returns without waiting for replay. Scan starts inclusively at replayFromSequence,
+     * in [1, registration head + 1]. Empty filters mean all; nonempty filters match
+     * exact runtime classes. The same consumer object cannot register twice. */
+    @NotNull AssetAZJournalSubscription registerConsumer(
+            long replayFromSequence,
+            @NotNull Set<Class<? extends AssetAZEvent>> eventTypes,
+            @NotNull AssetAZJournalConsumer consumer);
+}
@@ -0,0 +1,14 @@
+package com.r35157.assetaz.services.journal;
+
+import com.r35157.assetaz.services.journal.exception.DegradedException;
+import org.jetbrains.annotations.NotNull;
+
+/** A permanent registration; v1 has no unregister or degraded recovery. */
+public interface AssetAZJournalSubscription {
+    /** All records through this cursor were handled or examined as irrelevant. */
+    long getProcessedThroughSequence();
+    @NotNull AssetAZJournalConsumerState getState();
+    /** Minimum freshness, not a historical snapshot. Succeeds even when degraded
+     * if the safe cursor already meets the minimum; otherwise degradation throws. */
+    void awaitProcessedThrough(long requiredSequence) throws DegradedException, InterruptedException;
+}
@@ -0,0 +1,20 @@
+package com.r35157.assetaz.services.journal.codec;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import com.r35157.libs.pde.PDEEncoder;
+import com.r35157.libs.pde.PDEDecoder;
+import com.r35157.nenjim.component.NenjimComponent;
+import org.jetbrains.annotations.NotNull;
+
+/** One persistent schema mapped to a current domain type, obtained through Registry.
+ * Implementations must be thread-safe. Only one codec per domain type is a writer;
+ * historical read-only versions may coexist. No reflection discovery is involved.
+ */
+public interface AssetAZEventCodec extends NenjimComponent {
+    @NotNull String eventType();
+    int eventVersion();
+    @NotNull Class<? extends AssetAZEvent> eventClass();
+    boolean supportsEncoding();
+    void encode(@NotNull AssetAZEvent event, @NotNull PDEEncoder payload);
+    @NotNull AssetAZEvent decode(@NotNull PDEDecoder payload);
+}
@@ -0,0 +1,7 @@
+package com.r35157.assetaz.services.journal.codec;
+
+/** Domain schema rejection, distinct from PDE validity and record integrity. */
+public final class EventSchemaException extends IllegalArgumentException {
+    public EventSchemaException(String message) { super("event/schema validation failure: " + message); }
+    public EventSchemaException(String message, Throwable cause) { super("event/schema validation failure: " + message, cause); }
+}
@@ -0,0 +1,40 @@
+package com.r35157.assetaz.services.journal.codec.impl.ref;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import com.r35157.assetaz.events.AssetOwnerRegistered;
+import com.r35157.assetaz.events.AssetOwnerType;
+import com.r35157.assetaz.services.journal.codec.AssetAZEventCodec;
+import com.r35157.assetaz.services.journal.codec.EventSchemaException;
+import com.r35157.libs.pde.PDEDecoder;
+import com.r35157.libs.pde.PDEEncoder;
+import org.jetbrains.annotations.NotNull;
+import java.time.Instant;
+import java.util.UUID;
+
+/** Persistent AssetOwnerRegistered schema 1, with exactly seven positional values. */
+public final class AssetOwnerRegisteredCodec implements AssetAZEventCodec {
+    @Override public @NotNull String eventType() { return "AssetOwnerRegistered"; }
+    @Override public int eventVersion() { return 1; }
+    @Override public @NotNull Class<AssetOwnerRegistered> eventClass() { return AssetOwnerRegistered.class; }
+    @Override public boolean supportsEncoding() { return true; }
+    @Override public void encode(@NotNull AssetAZEvent event, @NotNull PDEEncoder out) {
+        if (!(event instanceof AssetOwnerRegistered owner)) throw new EventSchemaException("expected AssetOwnerRegistered");
+        EventFields.writeCommon(owner, out);
+        out.writeBytes(EventFields.uuid(EventFields.required(owner.assetOwnerId(), "assetOwnerId")));
+        out.writeUtf8(EventFields.required(owner.assetOwnerType(), "assetOwnerType").name());
+        out.writeUtf8(EventFields.required(owner.name(), "name"));
+    }
+    @Override public @NotNull AssetOwnerRegistered decode(@NotNull PDEDecoder in) {
+        Instant occurredAt = EventFields.instant(in);
+        String note = in.readNullableUtf8();
+        UUID correlationId = EventFields.uuid(in.readNullableBytes());
+        UUID ownerId = EventFields.required(EventFields.uuid(in.readNullableBytes()), "assetOwnerId");
+        String type = EventFields.required(in.readNullableUtf8(), "assetOwnerType");
+        AssetOwnerType ownerType;
+        try { ownerType = AssetOwnerType.valueOf(type); }
+        catch (IllegalArgumentException failure) { throw new EventSchemaException("unknown AssetOwnerType " + type, failure); }
+        ΩNameΩ name = EventFields.required(in.readNullableUtf8(), "name");
+        EventFields.end(in);
+        return new AssetOwnerRegistered(occurredAt, ownerId, ownerType, name, note, correlationId);
+    }
+}
@@ -0,0 +1,48 @@
+package com.r35157.assetaz.services.journal.codec.impl.ref;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import com.r35157.assetaz.services.journal.codec.EventSchemaException;
+import com.r35157.libs.pde.PDEDecoder;
+import com.r35157.libs.pde.PDEEncoder;
+import com.r35157.libs.pde.PDEException;
+import java.nio.ByteBuffer;
+import java.time.DateTimeException;
+import java.time.Instant;
+import java.util.UUID;
+
+/** Codec-level domain conversions. PDE itself has no UUID or Instant semantics. */
+final class EventFields {
+    private EventFields() { }
+    static void writeCommon(AssetAZEvent event, PDEEncoder out) {
+        required(event.occurredAt(), "occurredAt");
+        out.writeInteger(event.occurredAt().getEpochSecond());
+        out.writeInteger(event.occurredAt().getNano());
+        out.writeNullableUtf8(event.note());
+        out.writeNullableBytes(uuid(event.correlationId()));
+    }
+    static Instant instant(PDEDecoder in) {
+        long seconds = in.readLong();
+        int nanos = in.readInt();
+        if (nanos < 0 || nanos > 999999999) throw new EventSchemaException("invalid nano-of-second");
+        try { return Instant.ofEpochSecond(seconds, nanos); }
+        catch (DateTimeException failure) { throw new EventSchemaException("invalid occurredAt", failure); }
+    }
+    static byte[] uuid(UUID value) {
+        return value == null ? null : ByteBuffer.allocate(16)
+                .putLong(value.getMostSignificantBits()).putLong(value.getLeastSignificantBits()).array();
+    }
+    static UUID uuid(byte[] bytes) {
+        if (bytes == null) return null;
+        if (bytes.length != 16) throw new EventSchemaException("UUID must contain exactly 16 bytes");
+        ByteBuffer buffer = ByteBuffer.wrap(bytes);
+        return new UUID(buffer.getLong(), buffer.getLong());
+    }
+    static <T> T required(T value, String name) {
+        if (value == null) throw new EventSchemaException(name + " is required");
+        return value;
+    }
+    static void end(PDEDecoder in) {
+        try { in.requireEnd(); }
+        catch (PDEException failure) { throw new EventSchemaException("unexpected trailing payload values", failure); }
+    }
+}
@@ -0,0 +1,36 @@
+package com.r35157.assetaz.services.journal.codec.impl.ref;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import com.r35157.assetaz.events.JournalFormatVersionChanged;
+import com.r35157.assetaz.services.journal.codec.AssetAZEventCodec;
+import com.r35157.assetaz.services.journal.codec.EventSchemaException;
+import com.r35157.libs.pde.PDEDecoder;
+import com.r35157.libs.pde.PDEEncoder;
+import org.jetbrains.annotations.NotNull;
+import java.time.Instant;
+import java.util.UUID;
+
+/** Common event fields followed by the positive target journal format. */
+public final class JournalFormatVersionChangedCodec implements AssetAZEventCodec {
+    @Override public @NotNull String eventType() { return "JournalFormatVersionChanged"; }
+    @Override public int eventVersion() { return 1; }
+    @Override public @NotNull Class<JournalFormatVersionChanged> eventClass() { return JournalFormatVersionChanged.class; }
+    @Override public boolean supportsEncoding() { return true; }
+    @Override public void encode(@NotNull AssetAZEvent event, @NotNull PDEEncoder out) {
+        if (!(event instanceof JournalFormatVersionChanged changed)) throw new EventSchemaException("expected JournalFormatVersionChanged");
+        EventFields.writeCommon(changed, out);
+        out.writeInteger(positive(changed.journalFormatVersion()));
+    }
+    @Override public @NotNull JournalFormatVersionChanged decode(@NotNull PDEDecoder in) {
+        Instant occurredAt = EventFields.instant(in);
+        String note = in.readNullableUtf8();
+        UUID correlation = EventFields.uuid(in.readNullableBytes());
+        int version = positive(in.readInt());
+        EventFields.end(in);
+        return new JournalFormatVersionChanged(occurredAt, version, note, correlation);
+    }
+    private static int positive(int version) {
+        if (version < 1) throw new EventSchemaException("journal format must be positive");
+        return version;
+    }
+}
@@ -0,0 +1,6 @@
+package com.r35157.assetaz.services.journal.exception;
+
+/** Requested freshness exceeds this consumer's last safely processed sequence. */
+public final class DegradedException extends Exception {
+    public DegradedException(String message, Throwable cause) { super(message, cause); }
+}
@@ -0,0 +1,7 @@
+package com.r35157.assetaz.services.journal.exception;
+
+/** Authoritative history could not be verified or durably appended. */
+public final class JournalException extends IllegalStateException {
+    public JournalException(String message) { super(message); }
+    public JournalException(String message, Throwable cause) { super(message, cause); }
+}
@@ -0,0 +1,102 @@
+package com.r35157.assetaz.services.journal.impl.ref;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import com.r35157.assetaz.services.journal.AssetAZJournalConsumer;
+import com.r35157.assetaz.services.journal.AssetAZJournalService;
+import com.r35157.assetaz.services.journal.AssetAZJournalSubscription;
+import com.r35157.assetaz.services.journal.exception.JournalException;
+import com.r35157.nenjim.process.NenjimProcess;
+import com.r35157.nenjim.service.registry.NenjimRegistryService;
+import org.jetbrains.annotations.NotNull;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.channels.FileLock;
+import java.nio.file.Files;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.util.IdentityHashMap;
+import java.util.Objects;
+import java.util.Set;
+
+/** Hub-managed lifetime process; public API has no restart/recovery operations. */
+public final class AssetAZJournalServiceImpl implements AssetAZJournalService, NenjimProcess {
+    private final NenjimRegistryService registry;
+    private final Path path;
+    private final IdentityHashMap<AssetAZJournalConsumer, JournalSubscription> subscriptions = new IdentityHashMap<>();
+    private boolean startAttempted;
+    private boolean started;
+    private Throwable writeFailure;
+    private FileChannel channel;
+    private FileLock fileLock;
+    private EventCodecs codecs;
+    private long head;
+    private byte[] checksum;
+
+    public AssetAZJournalServiceImpl(@NotNull NenjimRegistryService registry, @NotNull Path path) {
+        this.registry = Objects.requireNonNull(registry, "registry");
+        this.path = Objects.requireNonNull(path, "path").toAbsolutePath().normalize();
+    }
+    @Override public synchronized void start() throws IOException {
+        if (startAttempted) throw new IllegalStateException("AssetAZ journal start already attempted");
+        startAttempted = true;
+        codecs = new EventCodecs(registry);
+        Files.createDirectories(path.getParent());
+        FileChannel opened = FileChannel.open(path, StandardOpenOption.CREATE, StandardOpenOption.READ, StandardOpenOption.WRITE);
+        try {
+            fileLock = opened.tryLock();
+            if (fileLock == null) throw new JournalException("AssetAZ journal is already owned by another process");
+            try (JournalRecords.Reader reader = new JournalRecords.Reader(path, codecs)) {
+                while (reader.next() != null) { /* Complete authoritative validation. */ }
+                head = reader.sequence();
+                checksum = reader.checksum();
+            }
+            opened.position(opened.size());
+            channel = opened;
+            started = true;
+        } catch (IOException | RuntimeException | Error failure) {
+            try { opened.close(); } catch (IOException closeFailure) { failure.addSuppressed(closeFailure); }
+            throw failure;
+        }
+    }
+    @Override public synchronized long append(@NotNull AssetAZEvent event) {
+        requireStarted();
+        if (writeFailure != null) throw new JournalException("AssetAZ journal failed closed after uncertain write", writeFailure);
+        Objects.requireNonNull(event, "event");
+        if (head == Long.MAX_VALUE) throw new JournalException("Journal sequence exhausted");
+        long candidate = head + 1;
+        JournalRecords.Encoded encoded = JournalRecords.encode(candidate, checksum, event, codecs);
+        try {
+            ByteBuffer bytes = ByteBuffer.wrap(encoded.bytes());
+            while (bytes.hasRemaining()) channel.write(bytes);
+            channel.force(true);
+        } catch (IOException | RuntimeException | Error failure) {
+            // Retain the file lock and failed state for this JVM. Never retry/truncate.
+            writeFailure = failure;
+            throw new JournalException("Uncertain append at candidate sequence " + candidate + "; journal failed closed", failure);
+        }
+        head = candidate;
+        checksum = encoded.checksum();
+        JournalRecords.Record record = new JournalRecords.Record(candidate, event, checksum);
+        for (JournalSubscription subscription : subscriptions.values()) subscription.publish(record);
+        return candidate;
+    }
+    @Override public synchronized @NotNull AssetAZJournalSubscription registerConsumer(
+            long replayFromSequence, @NotNull Set<Class<? extends AssetAZEvent>> eventTypes,
+            @NotNull AssetAZJournalConsumer consumer) {
+        requireStarted();
+        Objects.requireNonNull(consumer, "consumer");
+        Set<Class<? extends AssetAZEvent>> types = Set.copyOf(Objects.requireNonNull(eventTypes, "eventTypes"));
+        if (replayFromSequence < 1 || (replayFromSequence > head && replayFromSequence - 1 != head)) {
+            throw new IllegalArgumentException("replayFromSequence must be in [1, registration head + 1]");
+        }
+        if (subscriptions.containsKey(consumer)) throw new IllegalArgumentException("Consumer instance already registered");
+        JournalSubscription subscription = new JournalSubscription(path, codecs, replayFromSequence, head, types, consumer);
+        subscriptions.put(consumer, subscription);
+        subscription.start();
+        return subscription;
+    }
+    private void requireStarted() {
+        if (!started) throw new IllegalStateException("AssetAZ journal has not successfully started");
+    }
+}
@@ -0,0 +1,42 @@
+package com.r35157.assetaz.services.journal.impl.ref;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import com.r35157.assetaz.services.journal.codec.AssetAZEventCodec;
+import com.r35157.assetaz.services.journal.exception.JournalException;
+import com.r35157.nenjim.service.registry.NenjimRegistryService;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+
+/** Immutable catalogue snapshot; no concrete codec construction or reflection. */
+final class EventCodecs {
+    private record Key(String type, int version) { }
+    private final Map<Key, AssetAZEventCodec> readers = new HashMap<>();
+    private final Map<Class<? extends AssetAZEvent>, AssetAZEventCodec> writers = new HashMap<>();
+    EventCodecs(NenjimRegistryService registry) {
+        for (var id : registry.getComponentIds(AssetAZEventCodec.class)) {
+            AssetAZEventCodec codec = Objects.requireNonNull(registry.getComponent(id, AssetAZEventCodec.class), "registered codec");
+            String type = Objects.requireNonNull(codec.eventType(), "eventType");
+            if (!type.matches("[A-Za-z][A-Za-z0-9_]*") || codec.eventVersion() < 1) {
+                throw new JournalException("Invalid stable codec identity: " + id);
+            }
+            Objects.requireNonNull(codec.eventClass(), "eventClass");
+            if (readers.putIfAbsent(new Key(type, codec.eventVersion()), codec) != null) {
+                throw new JournalException("Duplicate event codec: " + type + "/" + codec.eventVersion());
+            }
+            if (codec.supportsEncoding() && writers.putIfAbsent(codec.eventClass(), codec) != null) {
+                throw new JournalException("Duplicate current encoder: " + codec.eventClass().getName());
+            }
+        }
+    }
+    AssetAZEventCodec reader(String type, int version) {
+        AssetAZEventCodec codec = readers.get(new Key(type, version));
+        if (codec == null) throw new JournalException("unknown event codec: " + type + "/" + version);
+        return codec;
+    }
+    AssetAZEventCodec writer(AssetAZEvent event) {
+        AssetAZEventCodec codec = writers.get(event.getClass());
+        if (codec == null) throw new JournalException("unknown event codec for append: " + event.getClass().getName());
+        return codec;
+    }
+}
@@ -0,0 +1,135 @@
+package com.r35157.assetaz.services.journal.impl.ref;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import com.r35157.assetaz.events.JournalFormatVersionChanged;
+import com.r35157.assetaz.services.journal.codec.AssetAZEventCodec;
+import com.r35157.assetaz.services.journal.codec.EventSchemaException;
+import com.r35157.assetaz.services.journal.exception.JournalException;
+import com.r35157.libs.pde.impl.ref.PDEDecoderImpl;
+import com.r35157.libs.pde.impl.ref.PDEEncoderImpl;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.nio.channels.FileChannel;
+import java.nio.file.Path;
+import java.nio.file.StandardOpenOption;
+import java.security.MessageDigest;
+import java.security.NoSuchAlgorithmException;
+import java.util.Arrays;
+
+/** Record integrity owns hashes, chain, sequence and file boundaries, never schemas. */
+final class JournalRecords {
+    private static final int MAX_RECORD_SIZE_BYTES = 1024;
+    private JournalRecords() { }
+    record Record(long sequence, AssetAZEvent event, byte[] checksum) { }
+    record Encoded(byte[] bytes, byte[] checksum) { }
+
+    static byte[] initialChecksum() { return digest(new byte[0], 0, 0); }
+    static Encoded encode(long sequence, byte[] previous, AssetAZEvent event, EventCodecs codecs) {
+        requireSupportedFormat(event);
+        AssetAZEventCodec codec = codecs.writer(event);
+        PDEEncoderImpl body = new PDEEncoderImpl();
+        body.writeBytes(previous);
+        body.writeInteger(sequence);
+        body.writeUtf8(codec.eventType());
+        body.writeInteger(codec.eventVersion());
+        body.writeObject(payload -> codec.encode(event, payload));
+        byte[] bodyBytes = body.toByteArray();
+        byte[] checksum = digest(bodyBytes, 0, bodyBytes.length);
+        PDEEncoderImpl checksumField = new PDEEncoderImpl();
+        checksumField.writeBytes(checksum);
+        byte[] prefix = checksumField.toByteArray();
+        byte[] fields = Arrays.copyOf(prefix, Math.addExact(prefix.length, bodyBytes.length));
+        System.arraycopy(bodyBytes, 0, fields, prefix.length, bodyBytes.length);
+        PDEEncoderImpl record = new PDEEncoderImpl();
+        record.writeEncodedObject(fields);
+        byte[] bytes = record.toByteArray();
+        if (bytes.length > MAX_RECORD_SIZE_BYTES) throw integrity("record exceeds 1024 bytes");
+        return new Encoded(bytes, checksum);
+    }
+    static void requireSupportedFormat(AssetAZEvent event) {
+        if (event instanceof JournalFormatVersionChanged changed && changed.journalFormatVersion() != 1) {
+            throw new JournalException("Unsupported journal format: " + changed.journalFormatVersion());
+        }
+    }
+    private static byte[] digest(byte[] bytes, int offset, int length) {
+        try {
+            MessageDigest digest = MessageDigest.getInstance("SHA-256");
+            digest.update(bytes, offset, length);
+            return digest.digest();
+        } catch (NoSuchAlgorithmException impossible) { throw new AssertionError(impossible); }
+    }
+    private static JournalException integrity(String message) {
+        return new JournalException("record integrity failure: " + message);
+    }
+
+    /** Each instance owns exactly one fixed record buffer, reused without clearing. */
+    static final class Reader implements AutoCloseable {
+        private final FileChannel channel;
+        private final EventCodecs codecs;
+        private final byte[] buffer = new byte[MAX_RECORD_SIZE_BYTES];
+        private long offset;
+        private long sequence;
+        private byte[] previous = initialChecksum();
+
+        Reader(Path path, EventCodecs codecs) throws IOException {
+            channel = FileChannel.open(path, StandardOpenOption.READ);
+            this.codecs = codecs;
+        }
+        long sequence() { return sequence; }
+        byte[] checksum() { return previous; }
+        Record next() throws IOException {
+            long remaining = channel.size() - offset;
+            if (remaining == 0) return null;
+            try {
+                if (remaining < 1) throw integrity("file shrank during reading");
+                readFully(0, 1);
+                int headerSize = 1 + PDEDecoderImpl.objectLengthWidth(buffer[0] & 255);
+                if (headerSize > remaining) throw integrity("truncated OBJECT header");
+                readFully(1, headerSize - 1);
+                long size = PDEDecoderImpl.objectSize(buffer, 0, headerSize);
+                if (size > MAX_RECORD_SIZE_BYTES) throw integrity("declared record exceeds 1024 bytes");
+                if (size > remaining) throw integrity("declared record extends beyond journal file");
+                readFully(headerSize, (int) size - headerSize);
+                PDEDecoderImpl outer = new PDEDecoderImpl(buffer, 0, (int) size);
+                PDEDecoderImpl fields = outer.readObject();
+                outer.requireEnd();
+                byte[] checksum = fields.readBytes();
+                if (checksum.length != 32) throw integrity("checksum must be 32 bytes");
+                byte[] calculated = digest(buffer, fields.position(), fields.limit() - fields.position());
+                if (!MessageDigest.isEqual(checksum, calculated)) throw integrity("checksum mismatch");
+                byte[] predecessor = fields.readBytes();
+                if (predecessor.length != 32 || !MessageDigest.isEqual(previous, predecessor)) {
+                    throw integrity("previousChecksum mismatch");
+                }
+                long nextSequence = fields.readLong();
+                if (sequence == Long.MAX_VALUE || nextSequence != sequence + 1) throw integrity("sequence discontinuity");
+                String type = fields.readUtf8();
+                int version = fields.readInt();
+                PDEDecoderImpl payload = fields.readObject();
+                fields.requireEnd();
+                AssetAZEventCodec codec = codecs.reader(type, version);
+                AssetAZEvent event = codec.decode(payload);
+                // The codec owns its schema, but cannot leave an envelope payload unread.
+                payload.requireEnd();
+                if (event == null || event.getClass() != codec.eventClass()) {
+                    throw new EventSchemaException("codec returned wrong domain type");
+                }
+                requireSupportedFormat(event);
+                offset += size;
+                sequence = nextSequence;
+                previous = checksum;
+                return new Record(sequence, event, checksum);
+            } catch (RuntimeException failure) {
+                throw new JournalException("Journal byte offset " + offset + ", expected sequence "
+                        + (sequence + 1) + ": " + failure.getMessage(), failure);
+            }
+        }
+        private void readFully(int start, int length) throws IOException {
+            ByteBuffer target = ByteBuffer.wrap(buffer, start, length);
+            while (target.hasRemaining()) {
+                if (channel.read(target) < 0) throw integrity("truncated record");
+            }
+        }
+        @Override public void close() throws IOException { channel.close(); }
+    }
+}
@@ -0,0 +1,111 @@
+package com.r35157.assetaz.services.journal.impl.ref;
+
+import com.r35157.assetaz.events.AssetAZEvent;
+import com.r35157.assetaz.services.journal.AssetAZJournalConsumer;
+import com.r35157.assetaz.services.journal.AssetAZJournalConsumerState;
+import com.r35157.assetaz.services.journal.AssetAZJournalSubscription;
+import com.r35157.assetaz.services.journal.exception.DegradedException;
+import com.r35157.assetaz.services.journal.exception.JournalException;
+import org.jetbrains.annotations.NotNull;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import java.nio.file.Path;
+import java.util.ArrayDeque;
+import java.util.Set;
+import static com.r35157.assetaz.services.journal.AssetAZJournalConsumerState.*;
+
+/** One independent, permanently registered sequential worker. */
+final class JournalSubscription implements AssetAZJournalSubscription {
+    private static final Logger log = LoggerFactory.getLogger(JournalSubscription.class);
+    private final Path path;
+    private final EventCodecs codecs;
+    private final long replayFrom;
+    private final long registrationHead;
+    private final Set<Class<? extends AssetAZEvent>> types;
+    private final AssetAZJournalConsumer consumer;
+    private final ArrayDeque<JournalRecords.Record> pending = new ArrayDeque<>();
+    private long processedThrough;
+    private AssetAZJournalConsumerState state = CATCHING_UP;
+    private Throwable failure;
+
+    JournalSubscription(Path path, EventCodecs codecs, long replayFrom, long registrationHead,
+            Set<Class<? extends AssetAZEvent>> types, AssetAZJournalConsumer consumer) {
+        this.path = path; this.codecs = codecs; this.replayFrom = replayFrom;
+        this.registrationHead = registrationHead; this.types = types; this.consumer = consumer;
+        processedThrough = replayFrom - 1;
+    }
+    void start() {
+        Thread.ofVirtual().name("assetaz-journal-" + identity()).start(this::run);
+    }
+    synchronized void publish(JournalRecords.Record record) {
+        if (state == DEGRADED) return;
+        pending.addLast(record);
+        notifyAll();
+    }
+    @Override public synchronized long getProcessedThroughSequence() { return processedThrough; }
+    @Override public synchronized @NotNull AssetAZJournalConsumerState getState() { return state; }
+    @Override public synchronized void awaitProcessedThrough(long requiredSequence)
+            throws DegradedException, InterruptedException {
+        while (processedThrough < requiredSequence) {
+            if (state == DEGRADED) throw new DegradedException(
+                    "Consumer " + identity() + " is degraded through " + processedThrough
+                            + "; required " + requiredSequence, failure);
+            wait();
+        }
+    }
+    private void run() {
+        long currentSequence = replayFrom;
+        try {
+            if (replayFrom <= registrationHead) {
+                try (JournalRecords.Reader reader = new JournalRecords.Reader(path, codecs)) {
+                    while (reader.sequence() < registrationHead) {
+                        JournalRecords.Record record = reader.next();
+                        if (record == null) throw new JournalException("record integrity failure: replay ended before registration head");
+                        currentSequence = record.sequence();
+                        if (record.sequence() >= replayFrom) process(record);
+                    }
+                }
+            }
+            boolean online = false;
+            while (true) {
+                JournalRecords.Record record;
+                synchronized (this) {
+                    while (online && pending.isEmpty()) wait();
+                    record = pending.pollFirst();
+                    // Empty queue under the publication monitor chooses the cutover.
+                    // Arrivals after this point queue as normal post-ONLINE events.
+                }
+                if (record == null) {
+                    currentSequence = getProcessedThroughSequence();
+                    consumer.requestState(ONLINE);
+                    synchronized (this) { state = ONLINE; notifyAll(); }
+                    online = true;
+                } else {
+                    currentSequence = record.sequence();
+                    process(record);
+                }
+            }
+        } catch (Throwable exception) {
+            if (exception instanceof InterruptedException) Thread.currentThread().interrupt();
+            degrade(exception, currentSequence);
+        }
+    }
+    private void process(JournalRecords.Record record) throws Exception {
+        if (types.isEmpty() || types.contains(record.event().getClass())) {
+            consumer.onEvent(record.sequence(), record.event());
+        }
+        synchronized (this) { processedThrough = record.sequence(); notifyAll(); }
+    }
+    private void degrade(Throwable exception, long sequence) {
+        AssetAZJournalConsumerState oldState;
+        synchronized (this) {
+            oldState = state; failure = exception; state = DEGRADED;
+            pending.clear(); notifyAll();
+        }
+        log.error("AssetAZ journal consumer {} failed at sequence {}, state {}, safe cursor {}",
+                identity(), sequence, oldState, getProcessedThroughSequence(), exception);
+    }
+    private String identity() {
+        return consumer.getClass().getName() + "@" + Integer.toHexString(System.identityHashCode(consumer));
+    }
+}
@@ -0,0 +1,16 @@
+package com.r35157.libs.pde;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+
+/** Decoder restricted to a logical byte range, including each child object. */
+public interface PDEDecoder {
+    int readInt();
+    long readLong();
+    @NotNull String readUtf8();
+    @Nullable String readNullableUtf8();
+    byte @NotNull [] readBytes();
+    byte @Nullable [] readNullableBytes();
+    @NotNull PDEDecoder readObject();
+    void requireEnd();
+}
@@ -0,0 +1,15 @@
+package com.r35157.libs.pde;
+
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import java.util.function.Consumer;
+
+/** Primitive PDE writer; domain conversions belong to codecs. */
+public interface PDEEncoder {
+    void writeInteger(long value);
+    void writeUtf8(@NotNull String value);
+    void writeNullableUtf8(@Nullable String value);
+    void writeBytes(byte @NotNull [] value);
+    void writeNullableBytes(byte @Nullable [] value);
+    void writeObject(@NotNull Consumer<PDEEncoder> writer);
+}
@@ -0,0 +1,7 @@
+package com.r35157.libs.pde;
+
+/** Invalid PDE type, range, encoding or framing (not a domain schema error). */
+public final class PDEException extends IllegalArgumentException {
+    public PDEException(String message) { super("PDE decoding/framing failure: " + message); }
+    public PDEException(String message, Throwable cause) { super("PDE decoding/framing failure: " + message, cause); }
+}
@@ -0,0 +1,115 @@
+package com.r35157.libs.pde.impl.ref;
+
+import com.r35157.libs.pde.PDEDecoder;
+import com.r35157.libs.pde.PDEException;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import java.nio.ByteBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.StandardCharsets;
+import java.util.Arrays;
+import java.util.Objects;
+
+public final class PDEDecoderImpl implements PDEDecoder {
+    private final byte[] bytes;
+    private final int limit;
+    private int position;
+
+    public PDEDecoderImpl(byte @NotNull [] bytes, int offset, int length) {
+        Objects.checkFromIndexSize(offset, length, bytes.length);
+        this.bytes = bytes; position = offset; limit = offset + length;
+    }
+    /** Absolute backing-array offset, for hashing original encoded ranges. */
+    public int position() { return position; }
+    public int limit() { return limit; }
+    @Override public int readInt() {
+        long value = readLong();
+        if (value < Integer.MIN_VALUE || value > Integer.MAX_VALUE) throw error("integer outside int range");
+        return (int) value;
+    }
+    @Override public long readLong() {
+        int type = next();
+        if (type < 4 || type > 19) throw error("expected non-null integer");
+        boolean negative = type >= 12;
+        int width = type - (negative ? 11 : 3);
+        long magnitude = unsigned(width);
+        if (magnitude < 0) throw error("integer outside long range");
+        return negative ? -1 - magnitude : magnitude;
+    }
+    @Override public @NotNull String readUtf8() {
+        String result = readNullableUtf8();
+        if (result == null) throw error("required UTF-8 is null");
+        return result;
+    }
+    @Override public @Nullable String readNullableUtf8() {
+        int length = fieldLength(48, 49, 65);
+        if (length < 0) return null;
+        require(length);
+        try {
+            String result = StandardCharsets.UTF_8.newDecoder().decode(ByteBuffer.wrap(bytes, position, length)).toString();
+            position += length;
+            return result;
+        } catch (CharacterCodingException failure) { throw new PDEException("invalid UTF-8", failure); }
+    }
+    @Override public byte @NotNull [] readBytes() {
+        byte[] result = readNullableBytes();
+        if (result == null) throw error("required bytes are null");
+        return result;
+    }
+    @Override public byte @Nullable [] readNullableBytes() {
+        int length = fieldLength(23, 24, 40);
+        if (length < 0) return null;
+        require(length);
+        byte[] result = Arrays.copyOfRange(bytes, position, position + length);
+        position += length;
+        return result;
+    }
+    @Override public @NotNull PDEDecoderImpl readObject() {
+        int width = objectLengthWidth(next());
+        int length = length(width);
+        require(length);
+        PDEDecoderImpl child = new PDEDecoderImpl(bytes, position, length);
+        position += length;
+        return child;
+    }
+    @Override public void requireEnd() {
+        if (position != limit) throw error("unexpected trailing values");
+    }
+    /** Number of following length bytes for a non-null OBJECT type. */
+    public static int objectLengthWidth(int type) {
+        if (type < 120 || type > 127) throw new PDEException("expected non-null OBJECT");
+        return type - 119;
+    }
+    /** Inspect just an OBJECT header; the body need not be present yet. */
+    public static long objectSize(byte @NotNull [] bytes, int offset, int headerLength) {
+        PDEDecoderImpl header = new PDEDecoderImpl(bytes, offset, headerLength);
+        int width = objectLengthWidth(header.next());
+        long size = header.unsigned(width);
+        header.requireEnd();
+        if (size < 0 || size > Long.MAX_VALUE - headerLength) throw error("OBJECT length overflow");
+        return size + headerLength;
+    }
+    private int fieldLength(int nullType, int shortBase, int lengthBase) {
+        int type = next();
+        if (type == nullType) return -1;
+        if (type >= shortBase && type <= shortBase + 15) return type - shortBase;
+        if (type >= lengthBase && type <= lengthBase + 7) return length(type - lengthBase + 1);
+        throw error("unexpected primitive type " + type);
+    }
+    private int length(int width) {
+        long value = unsigned(width);
+        if (value < 0 || value > Integer.MAX_VALUE) throw error("encoded length outside supported range");
+        return (int) value;
+    }
+    private long unsigned(int width) {
+        require(width);
+        long result = 0;
+        for (int i = 0; i < width; i++) result |= (long) next() << (8 * i);
+        return result;
+    }
+    private int next() { require(1); return bytes[position++] & 255; }
+    private void require(int length) {
+        if (length < 0 || length > limit - position) throw error("truncated value");
+    }
+    private static PDEException error(String message) { return new PDEException(message); }
+}
@@ -0,0 +1,72 @@
+package com.r35157.libs.pde.impl.ref;
+
+import com.r35157.libs.pde.PDEEncoder;
+import com.r35157.libs.pde.PDEException;
+import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
+import java.io.ByteArrayOutputStream;
+import java.nio.CharBuffer;
+import java.nio.charset.CharacterCodingException;
+import java.nio.charset.StandardCharsets;
+import java.util.Objects;
+import java.util.function.Consumer;
+
+/** Minimal Jenkov PDE: little-endian magnitudes/lengths, typed nulls and objects. */
+public final class PDEEncoderImpl implements PDEEncoder {
+    private final ByteArrayOutputStream output = new ByteArrayOutputStream();
+
+    @Override public void writeInteger(long value) {
+        long magnitude = value < 0 ? -(value + 1) : value;
+        int width = width(magnitude);
+        output.write((value < 0 ? 11 : 3) + width);
+        littleEndian(magnitude, width);
+    }
+    @Override public void writeUtf8(@NotNull String value) {
+        Objects.requireNonNull(value, "value");
+        try {
+            var encoded = StandardCharsets.UTF_8.newEncoder().encode(CharBuffer.wrap(value));
+            byte[] bytes = new byte[encoded.remaining()];
+            encoded.get(bytes);
+            field(49, 65, bytes);
+        } catch (CharacterCodingException failure) {
+            throw new PDEException("invalid UTF-8 input", failure);
+        }
+    }
+    @Override public void writeNullableUtf8(@Nullable String value) {
+        if (value == null) output.write(48); else writeUtf8(value);
+    }
+    @Override public void writeBytes(byte @NotNull [] value) {
+        field(24, 40, Objects.requireNonNull(value, "value"));
+    }
+    @Override public void writeNullableBytes(byte @Nullable [] value) {
+        if (value == null) output.write(23); else writeBytes(value);
+    }
+    @Override public void writeObject(@NotNull Consumer<PDEEncoder> writer) {
+        PDEEncoderImpl child = new PDEEncoderImpl();
+        Objects.requireNonNull(writer, "writer").accept(child);
+        writeEncodedObject(child.toByteArray());
+    }
+    /** Wrap already encoded PDE fields without semantic re-encoding. */
+    public void writeEncodedObject(byte @NotNull [] fields) {
+        int width = width(fields.length);
+        output.write(119 + width);
+        littleEndian(fields.length, width);
+        output.writeBytes(fields);
+    }
+    public byte @NotNull [] toByteArray() { return output.toByteArray(); }
+    private void field(int shortBase, int lengthBase, byte[] value) {
+        if (value.length <= 15) output.write(shortBase + value.length);
+        else {
+            int width = width(value.length);
+            output.write(lengthBase + width - 1);
+            littleEndian(value.length, width);
+        }
+        output.writeBytes(value);
+    }
+    private static int width(long value) {
+        return Math.max(1, (64 - Long.numberOfLeadingZeros(value) + 7) / 8);
+    }
+    private void littleEndian(long value, int width) {
+        for (int i = 0; i < width; i++) { output.write((int) value & 255); value >>>= 8; }
+    }
+}
@@ -1,5 +1,8 @@
 package com.r35157.nenjim.hubd.impl.ref;

+import com.r35157.assetaz.services.journal.impl.ref.AssetAZJournalServiceImpl;
+import com.r35157.assetaz.services.journal.codec.impl.ref.AssetOwnerRegisteredCodec;
+import com.r35157.assetaz.services.journal.codec.impl.ref.JournalFormatVersionChangedCodec;
 import com.fanitas.evelyn.core.Evelyn;
 import com.fanitas.evelyn.core.impl.ref.EvelynImpl;
 import com.fanitas.evelyn.service.burner.EvelynIOUBurnerService;
@@ -152,6 +155,13 @@ public final class NenjimHubImpl implements NenjimHub {

    private void initializeComponents(
            NenjimRegistryServiceManager registryServiceManager) {
+        registryServiceManager.registerComponent(
+                new NenjimComponentId("assetaz.journal.codec.asset-owner-registered.v1"),
+                new AssetOwnerRegisteredCodec());
+        registryServiceManager.registerComponent(
+                new NenjimComponentId("assetaz.journal.codec.journal-format-version-changed.v1"),
+                new JournalFormatVersionChangedCodec());
+
        objectCache = new ObjectCacheImpl();
        registryServiceManager.registerComponent(OBJECT_CACHE_ID, objectCache);

@@ -371,6 +381,8 @@ public final class NenjimHubImpl implements NenjimHub {
    createProcessConstructorCatalogue() {
        return Map.ofEntries(
                Map.entry(JOURNAL_SERVICE_ID, this::constructJournalProcess),
+                Map.entry(ASSETAZ_JOURNAL_SERVICE_ID, () -> new AssetAZJournalServiceImpl(
+                        registryService, Path.of("data/assetaz/journal/events.pde"))),
                Map.entry(TICKER_SERVICE_ID, this::constructTickerProcess),
                Map.entry(JUPITER_PERPS_ALARM_ID, this::constructJupiterPerpsAlarmProcess),
                Map.entry(EVELYN_SERVICE_PROD_ID, this::constructEvelynProdProcess),
@@ -446,6 +458,8 @@ public final class NenjimHubImpl implements NenjimHub {
            new NenjimComponentId("nenjim.registry.service");
    private static final NenjimComponentId JOURNAL_SERVICE_ID =
            new NenjimComponentId("nenjim.journal.service");
+    private static final NenjimComponentId ASSETAZ_JOURNAL_SERVICE_ID =
+            new NenjimComponentId("assetaz.journal.service");
    private static final Path JOURNAL_ROOT =
            Path.of("data/nenjim/journals");
    private static final NenjimComponentId OBJECT_CACHE_ID =