[Budget/Notification] atomic accounting lifecycle에 threshold 알림 연결

AGENTS.md

@@ -50,6 +50,7 @@ Never describe a roadmap item as an implemented or published capability.
 - `token-pilot-notification`은 알림 이벤트 발행과 중복 방지 로직만 담당한다.
 - 실제 메일/Slack/Webhook 전송은 사용자 애플리케이션의 `BudgetNotificationHandler` 구현체가 담당한다.
 - 라이브러리 내부에서 SMTP 설정이나 외부 메일 서비스를 기본 흐름으로 포함하지 않는다.
+- 알림 전달은 process-local best-effort이며 durable outbox나 재시작 후 replay를 제공하지 않는다.
 
 ## Architecture Decision: Framework Independence and Observability
 
@@ -75,7 +76,7 @@ Token Pilot의 제품 포지션은 framework-independent Java LLM control and ac
 | `token-pilot-spring-ai` | Basic implementation complete | Spring AI 2.0.0 `UsageExtractor`, `LedgerAdvisor`, pricing snapshot resolution, response usage recording, reconciliation decisions, and legacy provider-boundary BLOCK enforcement |
 | `token-pilot-micrometer` | Basic implementation complete | `MetricsOptions`, tag whitelist, and metric metadata exist; metric ownership must be narrowed |
 | `token-pilot-budget` | Atomic reservation and reconciliation implemented | Typed monthly keys, Clock/ZoneId windows, safe-upper-bound reservations, commit/release/write-off lifecycle, pending reconciliation liability, estimate/actual token and cost deltas, duplicate callback protection, and framework-independent best-effort accounting events implemented; candidate production and durable stores remain |
-| `token-pilot-notification` | Basic implementation complete | Event API and deduplication exist; not yet connected to the full advisor/budget lifecycle |
+| `token-pilot-notification` | Atomic accounting integration implemented | Commit, reconciliation-required, late reconciliation, and reservation BLOCK results produce process-local deduplicated threshold events with isolated handlers and a sanitized error hook; durable delivery remains |
 | `token-pilot-autoconfigure` | Basic implementation complete | Bean registration, property binding, pricing/budget/notification wiring, and `ChatClientBuilderCustomizer` implemented |
 | `token-pilot-starter` | Basic implementation complete | Thin final user entrypoint that brings runtime modules together |
 | `token-pilot-sample-app` | Basic E2E complete | Direct ledger metrics, budget, and fake Spring AI advisor E2E implemented |
@@ -216,6 +217,7 @@ Autoconfigure tests should use `ApplicationContextRunner` and verify:
 - Spring AI classpath registers `UsageExtractor`, `LedgerAdvisor`, and `ChatClientBuilderCustomizer`
 - notification beans do not register by default
 - notification beans register when notification is enabled and `BudgetNotificationHandler` bean exists
+- the default budget store receives the notification service as an accounting listener
 
 ## Notification Contract
 
@@ -233,9 +235,13 @@ class MailBudgetNotificationHandler implements BudgetNotificationHandler {
 
 - `BudgetNotificationHandler` 빈이 없으면 `BudgetNotificationService`는 등록되지 않는다 (no-op).
 - `token-pilot.notification.enabled=true` 설정 시에만 notification 빈이 등록된다.
-- 알림 중복 방지는 evaluator가 확정한 `BudgetKey(policyId, targetType, targetId, window)`로 처리된다.
-- 같은 window 안에서는 낮거나 같은 threshold 재발송이 방지된다.
+- 알림은 legacy evaluator 호출이 아니라 적용된 commit/reconciliation-required/late reconciliation과 원자적 reservation BLOCK 결과를 소비한다.
+- 알림 중복 방지는 `BudgetKey(policyId, targetType, targetId, window) + BudgetThreshold`로 처리된다.
+- 같은 window 안에서는 낮거나 같은 threshold 재발송이 방지되고 duplicate accounting callback은 같은 reservation ID로 다시 누적되지 않는다.
 - 새 window에서는 50/80/100% 알림이 다시 가능하다.
+- dedup 상태는 in-memory store 인스턴스 생명주기 동안 보존되며 TTL, 재시작 후 replay, exactly-once delivery를 제공하지 않는다.
+- handler 실패는 다음 handler, 회계 결과, provider 응답을 바꾸지 않으며 bounded/sanitized `BudgetNotificationErrorHook`으로만 관찰한다.
+- custom notification store는 atomic lifecycle을 위해 `AtomicNotificationStateStore`를 구현해야 하며, legacy `NotificationStateStore`만 등록하면 자동 설정이 명확히 실패한다.
 
 ## Recommended Configuration Shape
 
@@ -402,6 +408,11 @@ Stage and deploy a Central release:
 
 ## Update History
 
+### 2026-08-24
+
+- Connected budget notifications to applied commit, reconciliation-required, late reconciliation, and atomic reservation BLOCK results instead of legacy evaluator calls.
+- Added process-local atomic threshold deduplication, multi-handler failure isolation, sanitized error observation, and Spring Boot listener wiring.
+
 ### 2026-08-22
 
 - Restricted usage-based reservation reconciliation to provider-reported or provider-derived usage so local and heuristic estimates cannot be committed as actual spend.

token-pilot-autoconfigure/src/main/java/io/tokenpilot/autoconfigure/TokenPilotAutoConfiguration.java

@@ -3,6 +3,7 @@
 import io.micrometer.core.instrument.MeterRegistry;
 import io.tokenpilot.budget.BudgetEvaluator;
 import io.tokenpilot.budget.BudgetStateStore;
+import io.tokenpilot.budget.ReservationAccountingListener;
 import io.tokenpilot.budget.internal.LedgerBudgetComponents;
 import io.tokenpilot.core.CostCalculator;
 import io.tokenpilot.core.LedgerListener;
@@ -13,6 +14,12 @@
 import io.tokenpilot.core.domain.MissingPricingPolicy;
 import io.tokenpilot.core.internal.LedgerComponents;
 import io.tokenpilot.micrometer.internal.LedgerMicrometerComponents;
+import io.tokenpilot.notification.AtomicNotificationStateStore;
+import io.tokenpilot.notification.BudgetNotificationErrorHook;
+import io.tokenpilot.notification.BudgetNotificationHandler;
+import io.tokenpilot.notification.BudgetNotificationService;
+import io.tokenpilot.notification.InMemoryNotificationStateStore;
+import io.tokenpilot.notification.NotificationStateStore;
 import io.tokenpilot.springai.LedgerAdvisor;
 import io.tokenpilot.springai.UsageExtractor;
 import io.tokenpilot.springai.internal.LedgerSpringAiComponents;
@@ -25,10 +32,6 @@
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
-import io.tokenpilot.notification.BudgetNotificationHandler;
-import io.tokenpilot.notification.BudgetNotificationService;
-import io.tokenpilot.notification.InMemoryNotificationStateStore;
-import io.tokenpilot.notification.NotificationStateStore;
 
 import java.time.Clock;
 
@@ -176,8 +179,12 @@ public LedgerListener microCostMetricsPublisher(MeterRegistry meterRegistry, Tok
     @ConditionalOnMissingBean
     @ConditionalOnClass(LedgerBudgetComponents.class)
     @ConditionalOnProperty(prefix = "token-pilot.budget", name = "enabled", havingValue = "true")
-    public BudgetStateStore budgetStateStore() {
-        return LedgerBudgetComponents.inMemoryBudgetStateStore();
+    public BudgetStateStore budgetStateStore(
+        ObjectProvider<ReservationAccountingListener> accountingListeners
+    ) {
+        return LedgerBudgetComponents.inMemoryBudgetStateStore(
+            accountingListeners.orderedStream().toList()
+        );
     }
 
     /**
@@ -205,9 +212,9 @@ public BudgetEvaluator budgetEvaluator(
      * - token-pilot.notification.enabled=true 일 때만 등록
      */
     @Bean
-    @ConditionalOnMissingBean
+    @ConditionalOnMissingBean(NotificationStateStore.class)
     @ConditionalOnProperty(prefix = "token-pilot.notification", name = "enabled", havingValue = "true")
-    public NotificationStateStore notificationStateStore() {
+    public AtomicNotificationStateStore notificationStateStore() {
         return new InMemoryNotificationStateStore();
     }
 
@@ -222,9 +229,21 @@ public NotificationStateStore notificationStateStore() {
     @ConditionalOnBean(BudgetNotificationHandler.class)
     @ConditionalOnProperty(prefix = "token-pilot.notification", name = "enabled", havingValue = "true")
     public BudgetNotificationService budgetNotificationService(
-        BudgetNotificationHandler handler,
-        NotificationStateStore notificationStateStore
+        ObjectProvider<BudgetNotificationHandler> handlers,
+        AtomicNotificationStateStore notificationStateStore,
+        ObjectProvider<BudgetStateStore> budgetStateStore,
+        TokenPilotProperties properties,
+        ObjectProvider<BudgetNotificationErrorHook> errorHook
     ) {
-        return new BudgetNotificationService(handler, notificationStateStore);
+        var policy = properties.toBudgetPolicy();
+        return new BudgetNotificationService(
+            handlers.orderedStream().toList(),
+            notificationStateStore,
+            key -> budgetStateStore.getObject().snapshot(
+                key,
+                policy.monthlyLimit()
+            ),
+            errorHook.getIfAvailable(BudgetNotificationErrorHook::noOp)
+        );
     }
 }

token-pilot-autoconfigure/src/test/java/io/tokenpilot/autoconfigure/TokenPilotAutoConfigurationTest.java

@@ -5,10 +5,13 @@
 import io.tokenpilot.budget.BudgetDecision;
 import io.tokenpilot.budget.BudgetEvaluator;
 import io.tokenpilot.budget.BudgetKey;
+import io.tokenpilot.budget.BudgetReservationResult;
 import io.tokenpilot.budget.BudgetState;
 import io.tokenpilot.budget.BudgetStateStore;
 import io.tokenpilot.budget.BudgetThreshold;
 import io.tokenpilot.budget.BudgetWindow;
+import io.tokenpilot.budget.ReservationAccountingListener;
+import io.tokenpilot.budget.ReservationStatus;
 import io.tokenpilot.core.CostCalculator;
 import io.tokenpilot.core.LedgerManager;
 import io.tokenpilot.core.PricingEvaluator;
@@ -22,8 +25,11 @@
 import io.tokenpilot.core.domain.TokenType;
 import io.tokenpilot.core.domain.TokenUsage;
 import io.tokenpilot.core.exception.MissingPricingException;
+import io.tokenpilot.notification.AtomicNotificationStateStore;
 import io.tokenpilot.notification.BudgetNotificationHandler;
+import io.tokenpilot.notification.BudgetNotificationEvent;
 import io.tokenpilot.notification.BudgetNotificationService;
+import io.tokenpilot.notification.BudgetNotificationSource;
 import io.tokenpilot.notification.NotificationStateStore;
 import io.tokenpilot.springai.LedgerAdvisor;
 import io.tokenpilot.springai.UsageExtractor;
@@ -47,8 +53,10 @@
 import java.time.Instant;
 import java.time.ZoneOffset;
 import java.util.Currency;
+import java.util.List;
 import java.util.Map;
 import java.util.Optional;
+import java.util.concurrent.CopyOnWriteArrayList;
 import java.util.stream.Stream;
 
 import static io.tokenpilot.core.domain.TokenType.COMPLETION;
@@ -428,12 +436,73 @@ void shouldRegisterNotificationServiceWhenEnabledAndHandlerExists() {
             .withPropertyValues("token-pilot.notification.enabled=true")
             .run(context -> {
                 assertThat(context).hasSingleBean(NotificationStateStore.class);
+                assertThat(context).hasSingleBean(AtomicNotificationStateStore.class);
                 assertThat(context).hasSingleBean(BudgetNotificationService.class);
+                assertThat(context).hasSingleBean(ReservationAccountingListener.class);
+                assertThat(context.getBean(ReservationAccountingListener.class))
+                    .isSameAs(context.getBean(BudgetNotificationService.class));
                 assertThat(context.getBean(TokenPilotProperties.class).getNotification().isEnabled())
                     .isTrue();
             });
     }
 
+    @Test
+    @DisplayName("legacy custom notification store는 atomic lifecycle을 조용히 무시하지 않아야 한다")
+    void shouldFailFastForLegacyCustomNotificationStore() {
+        this.contextRunner
+            .withUserConfiguration(
+                FakeBudgetNotificationHandlerConfiguration.class,
+                LegacyNotificationStateStoreConfiguration.class
+            )
+            .withPropertyValues("token-pilot.notification.enabled=true")
+            .run(context -> {
+                assertThat(context).hasFailed();
+                assertThat(context.getStartupFailure())
+                    .hasMessageContaining("AtomicNotificationStateStore");
+            });
+    }
+
+    @Test
+    @DisplayName("원자적 budget BLOCK 결과가 자동 설정된 notification handler에 전달되어야 한다")
+    void shouldConnectAtomicBudgetBlockToNotificationHandler() {
+        this.contextRunner
+            .withUserConfiguration(FakeBudgetNotificationHandlerConfiguration.class)
+            .withPropertyValues(
+                "token-pilot.budget.enabled=true",
+                "token-pilot.notification.enabled=true"
+            )
+            .run(context -> {
+                BudgetStateStore stateStore = context.getBean(BudgetStateStore.class);
+                TokenPilotProperties properties = context.getBean(TokenPilotProperties.class);
+                Cost limit = properties.toBudgetPolicy().monthlyLimit();
+                BudgetKey key = new BudgetKey(
+                    "budget-policy",
+                    "tenant",
+                    "tenant-a",
+                    BudgetWindow.parse("2026-08")
+                );
+                RecordingBudgetNotificationHandler handler =
+                    (RecordingBudgetNotificationHandler) context.getBean(
+                        BudgetNotificationHandler.class
+                    );
+
+                BudgetReservationResult result = stateStore.checkAndReserve(
+                    key,
+                    limit,
+                    limit,
+                    "request-1"
+                );
+
+                assertThat(result.status()).isEqualTo(ReservationStatus.BLOCKED);
+                assertThat(handler.events())
+                    .extracting(BudgetNotificationEvent::threshold)
+                    .containsExactly(BudgetThreshold.EXCEEDED);
+                assertThat(handler.events())
+                    .extracting(BudgetNotificationEvent::source)
+                    .containsExactly(BudgetNotificationSource.RESERVATION_BLOCK);
+            });
+    }
+
     private static Stream<Arguments> providePricingConfigs() {
         return Stream.of(
             argumentSet(
@@ -565,7 +634,46 @@ public MeterRegistry meterRegistry() {
     static class FakeBudgetNotificationHandlerConfiguration {
         @Bean
         public BudgetNotificationHandler budgetNotificationHandler() {
-            return event -> {};
+            return new RecordingBudgetNotificationHandler();
+        }
+    }
+
+    @Configuration(proxyBeanMethods = false)
+    static class LegacyNotificationStateStoreConfiguration {
+        @Bean
+        public NotificationStateStore legacyNotificationStateStore() {
+            return new NotificationStateStore() {
+                private BudgetThreshold threshold = BudgetThreshold.NONE;
+
+                @Override
+                public BudgetThreshold getLastNotifiedThreshold(BudgetKey key) {
+                    return threshold;
+                }
+
+                @Override
+                public void updateLastNotifiedThreshold(
+                    BudgetKey key,
+                    BudgetThreshold threshold
+                ) {
+                    this.threshold = threshold;
+                }
+            };
+        }
+    }
+
+    static class RecordingBudgetNotificationHandler
+            implements BudgetNotificationHandler {
+
+        private final List<BudgetNotificationEvent> events =
+            new CopyOnWriteArrayList<>();
+
+        @Override
+        public void handle(BudgetNotificationEvent event) {
+            events.add(event);
+        }
+
+        List<BudgetNotificationEvent> events() {
+            return List.copyOf(events);
         }
     }
 }

token-pilot-budget/src/main/java/io/tokenpilot/budget/ReservationAccountingListener.java

@@ -1,8 +1,38 @@
 package io.tokenpilot.budget;
 
-/** 예약 정산 이벤트를 수신하는 framework-independent 계약입니다. */
+/**
+ * 예약 정산과 admission 차단 결과를 수신하는 framework-independent 계약입니다.
+ *
+ * <p>회계 변경은 새로 적용된 commit/reconciliation에만 전달됩니다. 예약 결과 callback은
+ * {@link ReservationStatus#BLOCKED} 결과에만 사용되며 기존 단일 추상 메서드를 유지하므로
+ * 기존 lambda listener와 source 호환됩니다.</p>
+ */
 @FunctionalInterface
 public interface ReservationAccountingListener {
 
     void onCommitted(ReservationAccountingEvent event);
+
+    /**
+     * 적용된 accounting transition과 같은 linearization point의 bucket snapshot을 전달합니다.
+     * 기존 listener는 {@link #onCommitted(ReservationAccountingEvent)}로 위임됩니다.
+     */
+    default void onAccountingApplied(
+            ReservationAccountingEvent event,
+            BudgetSnapshot snapshot
+    ) {
+        onCommitted(event);
+    }
+
+    /** actual 미확정 estimate가 pending liability로 이동한 결과를 전달합니다. */
+    default void onReconciliationRequired(
+            ReservationReconciliationRequiredEvent event
+    ) {
+    }
+
+    /** 원자적 admission에서 상태 변경 없이 차단된 결과를 전달합니다. */
+    default void onReservationBlocked(
+            BudgetReservationRequest request,
+            BudgetReservationResult result
+    ) {
+    }
 }

token-pilot-budget/src/main/java/io/tokenpilot/budget/ReservationReconciliationRequiredEvent.java

@@ -0,0 +1,42 @@
+package io.tokenpilot.budget;
+
+import java.util.Objects;
+
+/** actual usage를 확보하지 못해 pending liability로 이동한 원자적 회계 결과입니다. */
+public record ReservationReconciliationRequiredEvent(
+        ReservationId reservationId,
+        BudgetKey budgetKey,
+        ReservationTransition transition,
+        ReservationAccountingReason reason,
+        BudgetSnapshot snapshot
+) {
+
+    public ReservationReconciliationRequiredEvent {
+        Objects.requireNonNull(
+                reservationId,
+                "reservationId must not be null"
+        );
+        Objects.requireNonNull(budgetKey, "budgetKey must not be null");
+        Objects.requireNonNull(transition, "transition must not be null");
+        Objects.requireNonNull(reason, "reason must not be null");
+        Objects.requireNonNull(snapshot, "snapshot must not be null");
+        if (!transition.status().isApplied()
+                || transition.resultingState()
+                != ReservationState.RECONCILIATION_REQUIRED) {
+            throw new IllegalArgumentException(
+                    "event requires an applied RECONCILIATION_REQUIRED transition"
+            );
+        }
+        if (reason != ReservationAccountingReason.ACTUAL_USAGE_UNAVAILABLE
+                && reason != ReservationAccountingReason.CALLBACK_TIMED_OUT) {
+            throw new IllegalArgumentException(
+                    "reason must require later actual reconciliation"
+            );
+        }
+        if (!budgetKey.equals(snapshot.key())) {
+            throw new IllegalArgumentException(
+                    "event and snapshot must use the same budget key"
+            );
+        }
+    }
+}