feat(persistence): event-driven snapshot checkpointing with policy safety net (#2005)

[arnabnandy7]

Summary

Transitions process snapshot checkpointing from an invocation-only model (save() / update()) to an event-driven primary architecture, backed by repository policy checks as a safety net and CAS deduplication.

Problem & Motivation

Previously, durable snapshots were only saved synchronously during PersistentAgentProcessRepository.save() and update(). This meant process state transitions triggered asynchronously or within agent turn execution (such as entering wait states, completing, or terminating) were not checkpointed until or unless an explicit repository write occurred.

What's Changed

• Event-Driven Checkpointing (Primary Layer): PersistentAgentProcessRepository now implements AgenticEventListener, reactively checkpointing snapshots on key lifecycle events (AgentProcessWaitingEvent, AgentProcessCompletedEvent, AgentProcessFailedEvent, ProcessKilledEvent, and AgentProcessTerminatedEvent).
• Lifecycle Events:
  • AgentProcessTerminatedEvent: emitted whenever a process transitions to TERMINATED (via early termination signals, policies, turn termination, or explicit termination).
  • AgentProcessRestoredEvent: emitted after snapshot reconstitution in PersistentAgentProcessRepository.restore().
• Policy Safety Net: Retained checkpointIfNeeded() in save() and update() to ensure processes are persisted even if lifecycle events are bypassed or deferred.
• CAS Deduplication: Deduplicates duplicate writes between the event listener and the repository safety net using optimistic versioning (current.version >= nextVersion).
• Platform Wiring & Docs: Registered agentProcessCheckpointListener bean in AgentPlatformConfiguration, exposed SPI accessor AgentProcessPersistence.checkpointListener(), and updated reference documentation and architecture guides.

Verification

mvn spotless:check -pl embabel-agent-api

mvn test -pl embabel-agent-api "-Dtest=PersistentAgentProcessRepositoryTest,SimpleAgentProcessTest,AgentProcessPersistenceTest,AgentProcessPersistenceWiringTest"

Closes #2005

[igordayen]

@arnabnandy7

From Claude:

The InterruptedException path is the one to verify. _status.set(TERMINATED) is a direct atomic write, not
through setStatus(). runAction() then returns ActionStatus(FAILED). Whether tick() then overwrites _status
back to FAILED (via direct set) or respects the existing TERMINATED (via CAS from RUNNING) determines
whether:

• The process ends up in the wrong terminal state (FAILED instead of TERMINATED), OR
• The when(status) block fires AgentProcessTerminatedEvent a second time

This needs a targeted test: process in RUNNING, action throws InterruptedException, verify status is
TERMINATED and event fires exactly once.

---

How the two layers interact — and what happens when an event is missing

The composite listener wiring is correct: agentProcessCheckpointListener returns AgenticEventListener,
Spring's List collects it; the @primary eventListener bean fans out to it. Events DO
reach the checkpoint listener.

The safety net (doSave/doUpdate) fires when the framework writes to the repository. But there is no
guaranteed doUpdate() after a process terminates — especially for terminateAgent(WAITING) called externally.
In that case:

• The event fires, checkpoint() is called, snapshot is written ✓
• No doUpdate() follows (process is done, nothing calls update) — safety net never fires

So the safety net is not actually a safety net for termination paths; it only catches events missed during
active execution ticks. The naming "safety net" is misleading — it's really "checkpoint on tick". The
docs/README say "ensures persistence even if lifecycle events are not published" — that's only true for
events that happen to coincide with a repository save/update.

---

Issues to raise:

1. checkpointIfNeeded is now dead code — it just delegates to checkpoint(). Either remove it and call
   checkpoint() directly in doSave/doUpdate, or rename checkpoint() back to checkpointIfNeeded() and keep it
   private. Currently checkpoint() is package-visible (no internal modifier) for test access, but it's on an
   internal class — make it an internal fun checkpoint() to be explicit.
2. checkpoint() visibility — public on an internal class. Should be internal.
3. Thread safety gap — the read-compute-CAS pattern is not synchronized. The CAS deduplication handles
   idempotent concurrent writes (both writing the same state), but not divergent concurrent writes (event layer
   writing state A, safety net writing state B, both read version=N). One will be silently dropped via the
   current. version >= nextVersion check with no log at WARN. The debug log on deduplication says "already at
   version X" but doesn't capture that the dropped write may have been a different state.
4. E2E integration test missing — the JCache test exercises the policy/repository path. No E2E test exercises
   the event path: process enters WAITING, verify event fired checkpoint without explicit doSave, restore
   process, verify AgentProcessRestoredEvent.
5. InterruptedException double-fire — needs a targeted test: action interrupted → status TERMINATED → event
   fires exactly once.
6. Logging gap in onProcessEvent — nothing is logged when the event-driven checkpoint fires (only deduplication is
   logged at DEBUG). A single logger.debug("Checkpointing process {} on {}", agentProcess.id,
   event::class.simpleName) at the entry of onProcessEvent arms or before checkpoint() call would make the event
   path observable.
7. AgentProcessTerminatedEvent on terminateAgent(TERMINATED/KILLED/FAILED) branch — current code logs
   "already in terminal state, ignoring." No event fired — correct, but worth noting the comment in the code is
   clear about this.

PR requires very careful development — as it represents a core platform component — and e2e testing, thanks

[igordayen]

Compiling more....

Intent:

• Event fires → checkpoint immediately, inline with the state transition (primary)
• Event missing (bug, not-yet-implemented event, future code path) → doUpdate() fires after the tick →
  checkpoint as fallback

The safety net genuinely covers the missing-event case for ALL states, not just non-terminal ones.

But the deduplication implementation is broken for the normal sequential case.

When both layers fire in sequence (which is the common case, not a race):

1. Event → checkpoint() → reads version=0, writes version=1 ✓
2. Framework calls doUpdate() → checkpoint() → reads version=1, computes nextVersion=2, writes version=2 with
   expectedVersion=1 → succeeds, no conflict detected

The CAS exception never fires because versions differ (1 vs 2). So two identical snapshots are written every
time. The deduplication only guards against two threads computing the same nextVersion concurrently — not the
sequential event→doUpdate flow.

The fix: before serializing, compare whether the snapshot's stored version already reflects the current
process state version. If yes, skip. That way, the fallback does nothing when the event already handled it,
and does real work only when the event was missing.

The design is right. The deduplication implementation needs to match the design.

Two concepts, both lightweight:

1. Process state version
   Every meaningful mutation on AgentProcess (status change, blackboard write) increments an AtomicLong
   stateVersion on the process. This is the authoritative "how much has happened" counter.

2. Last persisted version
   PersistentAgentProcessRepository maintains a ConcurrentHashMap<processId, Long> — the stateVersion at which
   the last successful snapshot was written.

---

checkpoint() becomes:

if lastPersistedVersion[processId] >= process.stateVersion → skip (already done)
serialize + CAS write to store
lastPersistedVersion[processId] = process.stateVersion

That's it. No pre-write store read needed to detect duplicates. No CAS exception dance. Works identically for
concurrent races and sequential event→doUpdate flows.

---

How the two layers interact:

• Event fires → checkpoint() → skip check fails (not yet persisted) → writes snapshot → records stateVersion
• doUpdate() fires → checkpoint() → skip check passes (already at this stateVersion) → returns immediately
• Event missed → doUpdate() fires → skip check fails (nothing recorded yet) → writes snapshot → records
  stateVersion

The fallback genuinely does real work only when the event was absent.

---

Where lastPersistedVersion lives:
In the repository (not on the process). Doesn't pollute the domain model. Cleared naturally when a process is
evicted from the runtime repository. On restore, starts fresh — the first checkpoint after restore always
writes.

@arnabnandy7 - please confirm understanding before proceeding, thanks

[arnabnandy7]

@igordayen Confirming my understanding of both your review here and your latest direction on #2005.

The current CAS check does not deduplicate the normal sequential event -> doUpdate() flow: the second checkpoint reads the newly stored version and successfully writes another version of the same state. It also cannot safely treat every concurrent version conflict as an identical-state duplicate. I understand the state-version approach you suggested to address that, but your latest issue comment supersedes that approach by removing the need for two checkpointing layers.

I'll follow the revised scope in a separate PR linked to #1988:

• Retain the existing checkpointIfNeeded() path in doSave() / doUpdate().
• Add repository.update(this) after the immediate transition to TERMINATED for a non-RUNNING process, closing the path that has no subsequent run-cycle update.
• Keep AgentProcessTerminatedEvent for observability/cache invalidation, without checkpoint listener wiring or deduplication machinery.
• Add focused persistence coverage for external immediate termination and verify that an interrupted action leaves the process TERMINATED and emits the termination event exactly once.

Once that replacement PR is available, I'll link it here and close this PR as superseded. I've also acknowledged the revised scope on #2005: #2005 (comment)

[arnabnandy7]

Closing this PR as superceded to #2010

[igordayen]

• Add focused persistence coverage for external immediate termination and verify that an interrupted action leaves the process TERMINATED and emits the termination event exactly once.

@arnabnandy7 termination/kill APIs: process A can terminate process B. Please refer to the user guide. Thanks