feat(core): wire journal backfill and audit cleanup as system changes

core/flamingock-core/src/main/java/io/flamingock/internal/core/builder/AbstractChangeRunnerBuilder.java

@@ -58,6 +58,7 @@
 import io.flamingock.internal.core.builder.runner.RunnerBuilder;
 import io.flamingock.internal.core.builder.runner.RunnerFactory;
 import io.flamingock.internal.core.change.filter.ChangeFilter;
+import io.flamingock.internal.core.system.CoreSystemChangeFilter;
 import io.flamingock.internal.util.CollectionUtil;
 import io.flamingock.internal.util.Property;
 import io.flamingock.internal.util.id.RunnerId;
@@ -279,6 +280,11 @@ private LoadedPipeline loadPipeline(FlamingockMetadata flamingockMetadata) {
                 .flatMap(List::stream)
                 .collect(Collectors.toList());
 
+        // Keeps the journal-backfill / audit-cleanup system changes out of the runtime pipeline while
+        // JOURNAL_EVENTS is disabled - see CoreSystemChangeFilter for why this can't be done where they're
+        // contributed (compile time, a different process from where the flag actually lives).
+        changeFiltersFromPlugins.add(CoreSystemChangeFilter.instance());
+
         return LoadedPipeline.builder()
                 .addFilters(changeFiltersFromPlugins)
                 .addPreviewPipeline(flamingockMetadata.getPipeline())

core/flamingock-core/src/main/java/io/flamingock/internal/core/journal/JournalEventSequencerFactory.java

@@ -15,18 +15,35 @@
  */
 package io.flamingock.internal.core.journal;
 
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * Owns one sequencer per stream for this factory's lifetime. Persistence instances using the same
+ * stream share confirmed sequence progress. The stage lock still serializes event writes;
+ * sharing a sequencer does not make the allocation/write/confirmation cycle concurrent-safe.
+ */
 public class JournalEventSequencerFactory {
 
     private final JournalEventReader journalEventReader;
+    private final Map<String, JournalEventSequencer> sequencersByStream = new ConcurrentHashMap<>();
 
     public JournalEventSequencerFactory(JournalEventReader journalEventReader) {
         this.journalEventReader = journalEventReader;
     }
 
+    /**
+     * Returns the shared sequencer for a stream, seeded from its durable tail on first access.
+     *
+     * @param streamId stream whose sequence positions are allocated
+     * @return the same sequencer for every request for this stream within this factory
+     */
     public JournalEventSequencer forStream(String streamId) {
-        long initialSequence = journalEventReader.getLastEventByStream(streamId)
-                .map(e -> e.getStreamSequence() + 1)
-                .orElse(1L);
-        return new JournalEventSequencer(streamId, initialSequence);
+        return sequencersByStream.computeIfAbsent(streamId, id -> {
+            long initialSequence = journalEventReader.getLastEventByStream(id)
+                    .map(e -> e.getStreamSequence() + 1)
+                    .orElse(1L);
+            return new JournalEventSequencer(id, initialSequence);
+        });
     }
 }

core/flamingock-core/src/main/java/io/flamingock/internal/core/system/AuditCleanupChange.java

@@ -28,32 +28,16 @@
  * The system change that collapses the audit store to one record per change, once the journal holds the
  * history that the audit store used to carry.
  *
- * <h2>NOT YET WIRED — read this before assuming it runs</h2>
- * This class is <b>deliberately not contributed to any pipeline.</b> Nothing registers it as a system
- * change, so it never appears in a stage and never executes. It exists so that the per-store
- * {@code AuditCompactor} implementations have a consumer ready and reviewed, and so the eventual wiring is
- * a small, obvious change rather than a large one.
- * <p>
- * Two things must happen before it can be wired, in this order:
- * <ol>
- *   <li><b>The journal backfill system change must exist.</b> Compaction <em>deletes</em> the audit ledger;
- *       the backfill is what first preserves it as {@code JournalEvent}s. If this change ever runs before
- *       that one has, the change history is gone — not in the audit store, not in the journal, and
- *       unrecoverable. The ordering guarantee is real once both exist (the system stage sorts by
- *       {@code order} and runs before every other stage), but it cannot protect against a backfill that
- *       does not exist.</li>
- *   <li><b>The target-system problem must be settled.</b> Every change is resolved against a target system
- *       ({@code ChangeProcessStrategyFactory} → {@code TargetSystemManager#getTargetSystem}, which throws
- *       on an unknown id), and there is no default or null-object target system. Audit cleanup operates on
- *       the audit store, not on a target system, so it has no id to supply. The Mongock importer sidesteps
- *       this by taking one from {@code @MongockSupport(targetSystem = ...)}; there is no equivalent here.</li>
- * </ol>
- * When wiring, append a system change to {@code FlamingockAnnotationProcessorPlugin#findAnnotatedChanges()}
- * — mirroring {@code MongockAnnotationProcessorPlugin#getImporterChange()} — with
- * {@code setSystem(true)} and an {@code order} after the backfill's. Reserved sequence: Mongock import
- * {@code 00100}, journal backfill {@code 00200}, audit cleanup {@code 00300}. Those orders are compared as
- * plain strings and nothing enforces uniqueness, so the backfill's author has to be told which value is
- * taken.
+ * <h2>Wiring</h2>
+ * Contributed unconditionally by {@code CoreSystemChangeProvider}, consumed by
+ * {@code FlamingockAnnotationProcessorPlugin#findAnnotatedChanges()} — mirroring
+ * {@code MongockAnnotationProcessorPlugin#getImporterChange()} — with {@code setSystem(true)} and
+ * {@code order = "00300"}, after the journal backfill's {@code "00200"} (Mongock import is
+ * {@code "00100"}). The system stage sorts by {@code order} and runs before every other stage, so the
+ * backfill is guaranteed to run first and this change never races it. Its {@link AuditCompactor} dependency
+ * is already registered by {@code CommunityChangeRunnerBuilder#contributeEditionDependencies}. The
+ * target-system problem this javadoc used to describe is resolved: audit-store-only system changes resolve
+ * against the reserved {@code PipelineHelper#AUDIT_STORE_TARGET_SYSTEM} alias.
  *
  * <h2>Why it fails instead of skipping when the journal is off</h2>
  * With {@code Features.JOURNAL_EVENTS} disabled no journal event has ever been written, so the audit ledger
@@ -67,6 +51,8 @@
  */
 public class AuditCleanupChange {
 
+    public static final String CHANGE_ID = "compact-audit-store";
+
     private static final Logger logger = FlamingockLoggerFactory.getLogger("AuditCleanup");
 
     /**

core/flamingock-core/src/main/java/io/flamingock/internal/core/system/CoreSystemChangeFilter.java

@@ -0,0 +1,57 @@
+/*
+ * Copyright 2026 Flamingock (https://www.flamingock.io)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.flamingock.internal.core.system;
+
+import io.flamingock.internal.common.core.feature.Features;
+import io.flamingock.internal.core.change.filter.ChangeFilter;
+import io.flamingock.internal.core.change.loaded.AbstractLoadedChange;
+import io.flamingock.internal.util.FeatureFlag;
+
+/**
+ * Excludes the journal-backfill and audit-cleanup system changes from the runtime pipeline while
+ * {@link Features#JOURNAL_EVENTS} is disabled.
+ * <p>
+ * They are contributed unconditionally at compile time - {@code CoreSystemChangeProvider} has no way to
+ * see the runtime flag, since it runs inside javac, a different process. This filter is the runtime-side
+ * half: it runs in the same process where {@link FeatureFlag} state actually lives, so it can keep them
+ * out of the executable pipeline while the flag is off.
+ * <p>
+ * Excluding rather than letting them run and throw (their own fail-fast behavior, kept for when they are
+ * genuinely in the pipeline but something is otherwise wrong) matters because of what "ran and threw" does
+ * to the audit store: nothing, since the failure is thrown before any audit write - but letting every
+ * application startup reach that throw would turn an internal, off-by-default flag into a hard failure for
+ * every application, which defeats the point of it being off by default. Excluded here, the change simply
+ * never attempts to run and never gets an audit record, so the first startup after the flag is finally
+ * enabled sees it for the first time and runs it then - never "skipped but recorded forever".
+ */
+public final class CoreSystemChangeFilter implements ChangeFilter {
+
+    private static final CoreSystemChangeFilter INSTANCE = new CoreSystemChangeFilter();
+
+    public static CoreSystemChangeFilter instance() {
+        return INSTANCE;
+    }
+
+    private CoreSystemChangeFilter() {
+    }
+
+    @Override
+    public boolean filter(AbstractLoadedChange change) {
+        String id = change.getId();
+        boolean gated = JournalBackfillChange.CHANGE_ID.equals(id) || AuditCleanupChange.CHANGE_ID.equals(id);
+        return !gated || FeatureFlag.isEnabled(Features.JOURNAL_EVENTS);
+    }
+}

core/flamingock-core/src/main/java/io/flamingock/internal/core/system/JournalBackfillChange.java

@@ -17,6 +17,7 @@
 
 import io.flamingock.api.annotations.NonLockGuarded;
 import io.flamingock.internal.common.core.audit.AuditEntry;
+import io.flamingock.internal.common.core.audit.AuditPersistence;
 import io.flamingock.internal.common.core.audit.AuditPersistenceFactory;
 import io.flamingock.internal.common.core.error.FlamingockException;
 import io.flamingock.internal.common.core.feature.Features;
@@ -33,15 +34,20 @@
 
 /**
  * Copies historical audit entries into local journal streams without modifying audit history.
- * Registration and execution order are separate concerns; this change only retains its identity to
- * exclude its own audit records.
+ * This change only retains its identity to exclude its own audit records.
+ * <p>
+ * Contributed unconditionally by {@code CoreSystemChangeProvider}, consumed by
+ * {@code FlamingockAnnotationProcessorPlugin#findAnnotatedChanges()} with {@code setSystem(true)} and
+ * {@code order = "00200"} — after the Mongock importer's {@code "00100"} and before audit cleanup's
+ * {@code "00300"}. Resolved against the reserved {@code PipelineHelper#AUDIT_STORE_TARGET_SYSTEM} target
+ * system, since this change operates on the audit store itself.
  */
 public class JournalBackfillChange {
     public static final String CHANGE_ID = "backfill-audit-history-to-journal";
 
     private static final Logger logger = FlamingockLoggerFactory.getLogger("JournalBackfill");
 
-    private final Map<String, JournalHistoryAppender> journalPersistenceByStage = new HashMap<>();
+    private final Map<String, AuditPersistence> persistenceByStage = new HashMap<>();
     private AuditPersistenceFactory<?> persistenceFactory;
 
     /**
@@ -62,11 +68,17 @@ public void journalBackfill(@NonLockGuarded AuditPersistenceFactory<?> persisten
 
         logger.info("Backfilling audit history into journal streams");
         this.persistenceFactory = persistenceFactory;
-        for (AuditEntry auditEntry : persistenceFactory.get(SYSTEM_STAGE_ID).getAuditHistory()) {
+        // Routed through the same cache as every per-entry lookup below - not just the general
+        // "one less instantiation" saving the review comment on PR #982 flagged, but required for
+        // correctness on DynamoDB: a persistence instance owns a journal-sequence baseline, and a second,
+        // independently-built instance for this same stage starts from a stale baseline, handing out a
+        // sequence number a later write (e.g. the next system change's own STARTED entry) has already
+        // consumed - a hard DynamoDB transaction conflict, not a mere inefficiency.
+        for (AuditEntry auditEntry : getPersistence(SYSTEM_STAGE_ID).getAuditHistory()) {
             if (CHANGE_ID.equals(auditEntry.getChangeId())) {
                 continue;
             }
-            JournalHistoryAppender persistence = getJournalPersistence(auditEntry);
+            JournalHistoryAppender persistence = (JournalHistoryAppender) getPersistence(auditEntry.getStageId());
             Result result = persistence.appendEventFrom(auditEntry);
             if (!(result instanceof Result.Ok)) {
                 Throwable cause = result instanceof Result.Error ? ((Result.Error) result).getError() : null;
@@ -76,8 +88,7 @@ public void journalBackfill(@NonLockGuarded AuditPersistenceFactory<?> persisten
         logger.info("Audit history backfilled into journal streams");
     }
 
-    private JournalHistoryAppender getJournalPersistence(AuditEntry auditEntry) {
-        return journalPersistenceByStage.computeIfAbsent(auditEntry.getStageId(),
-                stageId -> (JournalHistoryAppender) persistenceFactory.get(stageId));
+    private AuditPersistence getPersistence(String stageId) {
+        return persistenceByStage.computeIfAbsent(stageId, persistenceFactory::get);
     }
 }

core/flamingock-core/src/test/java/io/flamingock/internal/core/journal/JournalEventSequencerFactoryTest.java

@@ -0,0 +1,95 @@
+/*
+ * Copyright 2026 Flamingock (https://www.flamingock.io)
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package io.flamingock.internal.core.journal;
+
+import io.flamingock.internal.common.core.audit.AuditEntry;
+import io.flamingock.internal.common.core.journal.JournalEvent;
+import io.flamingock.internal.common.core.journal.JournalEventType;
+import org.junit.jupiter.api.DisplayName;
+import org.junit.jupiter.api.Test;
+
+import java.time.Instant;
+import java.util.Optional;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotSame;
+import static org.junit.jupiter.api.Assertions.assertSame;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.when;
+
+class JournalEventSequencerFactoryTest {
+
+    private final JournalEventReader reader = mock(JournalEventReader.class);
+    private final AuditEntry payload = mock(AuditEntry.class);
+    private final JournalEventSequencerFactory factory = new JournalEventSequencerFactory(reader);
+
+    @Test
+    @DisplayName("shares confirmed sequence progress between consumers of the same stream")
+    void sharesSequenceProgressBetweenConsumers() {
+        when(reader.getLastEventByStream("stage-1")).thenReturn(Optional.empty());
+        JournalEventSequencer outer = factory.forStream("stage-1");
+        assertEquals(1L, outer.newEvent(payload).getStreamSequence());
+        outer.confirm();
+
+        when(reader.getLastEventByStream("stage-1")).thenReturn(Optional.of(event("stage-1", 1L)));
+        JournalEventSequencer inner = factory.forStream("stage-1");
+        assertEquals(2L, inner.newEvent(payload).getStreamSequence());
+        inner.confirm();
+
+        assertEquals(3L, outer.newEvent(payload).getStreamSequence());
+        outer.confirm();
+        assertSame(outer, inner);
+        assertSame(outer, factory.forStream("stage-1"));
+        assertEquals(4L, factory.forStream("stage-1").newEvent(payload).getStreamSequence());
+    }
+
+    @Test
+    @DisplayName("seeds independent streams from their own durable tails")
+    void seedsIndependentStreamsFromDurableTails() {
+        when(reader.getLastEventByStream("stage-1")).thenReturn(Optional.of(event("stage-1", 7L)));
+        when(reader.getLastEventByStream("stage-2")).thenReturn(Optional.empty());
+        JournalEventSequencer first = factory.forStream("stage-1");
+        JournalEventSequencer second = factory.forStream("stage-2");
+
+        assertNotSame(first, second);
+        JournalEvent<AuditEntry> firstEvent = first.newEvent(payload);
+        assertEquals("stage-1", firstEvent.getStreamId());
+        assertEquals(8L, firstEvent.getStreamSequence());
+        first.confirm();
+        JournalEvent<AuditEntry> secondEvent = second.newEvent(payload);
+        assertEquals("stage-2", secondEvent.getStreamId());
+        assertEquals(1L, secondEvent.getStreamSequence());
+        assertEquals(9L, first.newEvent(payload).getStreamSequence());
+    }
+
+    @Test
+    @DisplayName("reuses an unconfirmed position through another consumer")
+    void reusesUnconfirmedPosition() {
+        when(reader.getLastEventByStream("stage-1")).thenReturn(Optional.empty());
+        JournalEventSequencer first = factory.forStream("stage-1");
+        assertEquals(1L, first.newEvent(payload).getStreamSequence());
+
+        JournalEventSequencer second = factory.forStream("stage-1");
+        assertEquals(1L, second.newEvent(payload).getStreamSequence());
+        second.confirm();
+        assertEquals(2L, first.newEvent(payload).getStreamSequence());
+    }
+
+    private JournalEvent<AuditEntry> event(String streamId, long sequence) {
+        return new JournalEvent<>("event-" + sequence, "key-" + sequence, JournalEventType.CHANGE_STATE,
+                streamId, sequence, Instant.now(), payload);
+    }
+}

core/flamingock-processor/src/main/java/io/flamingock/core/processor/FlamingockAnnotationProcessorPlugin.java

@@ -16,6 +16,7 @@
 package io.flamingock.core.processor;
 
 import io.flamingock.api.annotations.Change;
+import io.flamingock.core.processor.system.CoreSystemChangeProvider;
 import io.flamingock.internal.common.core.processor.AnnotationProcessorPlugin;
 import io.flamingock.internal.common.core.util.LoggerPreProcessor;
 import io.flamingock.internal.common.core.processor.ChangeDiscoverer;
@@ -26,6 +27,7 @@
 import javax.lang.model.element.ElementKind;
 import javax.lang.model.element.TypeElement;
 import java.util.Collection;
+import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.stream.Collectors;
@@ -44,13 +46,19 @@ public void initialize(RoundEnvironment roundEnv, LoggerPreProcessor logger) {
     @Override
     public Collection<CodePreviewChange> findAnnotatedChanges() {
         logger.info("Searching for code-based changes (Java classes annotated with @Change annotation)");
-        return roundEnv.getElementsAnnotatedWith(Change.class)
+        List<CodePreviewChange> changes = roundEnv.getElementsAnnotatedWith(Change.class)
                 .stream()
                 .filter(e -> e.getKind() == ElementKind.CLASS)
                 .map(e -> (TypeElement) e)
                 .map(this::buildCodePreviewChange)
                 .filter(Objects::nonNull)
                 .collect(Collectors.toList());
+
+        // Unconditional, unlike the Mongock importer: every audit store needs its history preserved to the
+        // journal and then compacted, regardless of which optional modules are present.
+        changes.addAll(CoreSystemChangeProvider.getSystemChanges());
+
+        return changes;
     }
 
     private CodePreviewChange buildCodePreviewChange(TypeElement typeElement) {