feat: add Jev-aware scheduling

README.MD

@@ -65,7 +65,7 @@ That gives you intelligent routing without putting an LLM call in the `Publish()
 - **Dead-letter queue** with replay support and panic recovery.
 - **Redis Streams and RabbitMQ EventStore options** for cross-process delivery.
 - **Transactions** with local buffering, synchronous commit, and optional Rules/Cache/Jev event selection.
-- **Scheduling** with absolute and relative timers.
+- **Scheduling** with absolute/relative timers and optional Rules/Cache/Jev event selection.
 - **Metrics** for published, processed, and failed events.
 - **Pluggable decision cache** through the `DecisionCache` interface.
 
@@ -322,6 +322,7 @@ The repository includes runnable examples for both the classic event-bus path an
 | [`routing_cache`](examples/routing_cache) | Reusing a previous decision so the fallback selector is called only once |
 | [`routing_jev`](examples/routing_jev) | Full `rules → cache → Jev → GoEventBus` routing with OpenRouter |
 | [`transaction_jev`](examples/transaction_jev) | Rules/Cache/Jev event selection buffered until transaction commit |
+| [`schedule_jev`](examples/schedule_jev) | Rules/Cache/Jev event selection before delayed scheduling |
 | [`hello_world`](examples/hello_world) | Minimal direct `Subscribe → Publish` flow |
 | [`middleware`](examples/middleware) | Handler middleware and lifecycle behavior |
 | [`goroutines-subscribe-publisher`](examples/goroutines-subscribe-publisher) | Concurrent producers and publishing |
@@ -340,6 +341,7 @@ go run ./examples/routing_rules
 go run ./examples/routing_cache
 OPENROUTER_API_KEY=... go run ./examples/routing_jev
 OPENROUTER_API_KEY=... go run ./examples/transaction_jev
+OPENROUTER_API_KEY=... go run ./examples/schedule_jev
 ```
 
 See [`examples/README.md`](examples/README.md) for a compact guide to all examples.
@@ -870,7 +872,7 @@ shared EventStore ring.
 
 # Scheduling
 
-Schedule an event at a time:
+Schedule a known event at a specific time:
 
 ```go
 timer := store.Schedule(
@@ -890,9 +892,85 @@ timer := store.ScheduleAfter(
 )
 ```
 
-The returned `*time.Timer` may be stopped before it fires.
+## Intelligent scheduling
 
-Past times and non-positive durations are handled immediately.
+Use the same Rules / Cache / Jev selector pipeline when the event type must be
+chosen before it is scheduled.
+
+Schedule at an absolute time:
+
+```go
+decision, timer, err := store.DecideAndSchedule(
+    ctx,
+    time.Now().Add(10*time.Second),
+    selector,
+    state,
+    GoEventBus.Event{
+        ID:   "evt-scheduled-1",
+        Data: payload,
+    },
+    candidates,
+)
+if err != nil {
+    log.Fatal(err)
+}
+
+fmt.Println("scheduled:", decision.Choice)
+```
+
+Or after a duration:
+
+```go
+decision, timer, err := store.DecideAndScheduleAfter(
+    ctx,
+    30*time.Second,
+    selector,
+    state,
+    GoEventBus.Event{
+        ID:   "evt-scheduled-2",
+        Data: payload,
+    },
+    candidates,
+)
+if err != nil {
+    log.Fatal(err)
+}
+```
+
+The selector is evaluated **when scheduling is requested**, not when the timer
+fires:
+
+```text
+state
+  ↓
+Rules
+  ↓
+Cache
+  ↓
+Jev
+  ↓
+selected event
+  ↓
+timer
+  ↓
+Subscribe → Publish → handler
+```
+
+That means selection errors are returned immediately and no timer is created.
+Once scheduling succeeds, the timer contains an already-resolved event, so a
+later change in rules, cache contents, or Jev output does not change what will
+fire.
+
+`DecideAndSchedule` and `DecideAndScheduleAfter` work with
+`RuleCacheSelector`, `JevSelector`, or any custom `EventSelector`.
+
+The returned `*time.Timer` may be stopped before it fires. Past times and
+non-positive durations preserve the existing immediate-execution behavior and
+return a nil timer.
+
+Scheduling remains a **local EventStore operation**. It does not publish the
+scheduled event to Redis/RabbitMQ even when the parent store has a broker
+configured.
 
 ---
 
@@ -975,7 +1053,8 @@ Important methods:
 | `Use` | Register middleware |
 | `OnBefore` / `OnAfter` / `OnError` | Register lifecycle hooks |
 | `Metrics` | Read event counters |
-| `Schedule` / `ScheduleAfter` | Schedule future events |
+| `Schedule` / `ScheduleAfter` | Schedule known events |
+| `DecideAndSchedule` / `DecideAndScheduleAfter` | Select with Rules/Cache/Jev, then schedule the resolved event |
 | `BeginTransaction` | Start a local transaction buffer; use `tx.DecideAndPublish` for Rules/Cache/Jev selection |
 | `Drain` / `Close` | Stop accepting events and finish in-flight work |
 

examples/README.md

@@ -73,6 +73,22 @@ Flow:
 state -> rules/cache/Jev -> selected event -> transaction buffer -> Commit -> handler
 ```
 
+### `schedule_jev`
+
+Uses Rules / Cache / Jev to resolve the event type before starting a timer.
+The decision is made immediately; the selected event is dispatched only when
+the timer fires.
+
+```bash
+OPENROUTER_API_KEY=... go run ./examples/schedule_jev
+```
+
+Flow:
+
+```text
+state -> rules/cache/Jev -> selected event -> timer -> Subscribe -> Publish -> handler
+```
+
 ## Core event bus
 
 | Example | Purpose |
@@ -89,6 +105,7 @@ state -> rules/cache/Jev -> selected event -> transaction buffer -> Commit -> ha
 | `redis` | Redis Streams configured directly on `NewEventStore` |
 | `rabbitmq` | RabbitMQ configured directly on `NewEventStore` |
 | `transaction_jev` | Rules/Cache/Jev event selection buffered until transaction commit |
+| `schedule_jev` | Rules/Cache/Jev event selection before delayed scheduling |
 
 ## Broker-backed stores
 

examples/schedule_jev/main.go

@@ -0,0 +1,88 @@
+package main
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"os"
+	"time"
+
+	GoEventBus "github.com/Protocol-Lattice/GoEventBus"
+)
+
+func main() {
+	apiKey := os.Getenv("OPENROUTER_API_KEY")
+	if apiKey == "" {
+		fmt.Println("set OPENROUTER_API_KEY to run this example")
+		return
+	}
+
+	ctx := context.Background()
+	dispatcher := GoEventBus.Dispatcher{}
+	dispatcher.Register("order.created", printHandler)
+	dispatcher.Register("invoice.created", printHandler)
+
+	store := GoEventBus.NewEventStore(&dispatcher, 16, GoEventBus.DropOldest)
+	defer func() { _ = store.Close(context.Background()) }()
+
+	selector := &GoEventBus.RuleCacheSelector{
+		Cache: GoEventBus.NewMemoryDecisionCache(5 * time.Minute),
+		Fallback: &GoEventBus.JevSelector{
+			APIKey: apiKey,
+		},
+	}
+
+	candidates := []GoEventBus.EventCandidate{
+		{
+			Key:         "order_created",
+			Projection:  "order.created",
+			Description: "A new order should be created",
+		},
+		{
+			Key:         "invoice_created",
+			Projection:  "invoice.created",
+			Description: "An invoice should be created for an existing order",
+		},
+	}
+
+	decision, timer, err := store.DecideAndScheduleAfter(
+		ctx,
+		2*time.Second,
+		selector,
+		map[string]any{
+			"message": "Create an invoice for order 42 in two seconds",
+		},
+		GoEventBus.Event{
+			ID:   "evt-scheduled-1",
+			Data: map[string]any{"order_id": "42"},
+		},
+		candidates,
+	)
+	if err != nil {
+		log.Fatal(err)
+	}
+
+	fmt.Printf(
+		"scheduled choice=%s confidence=%.2f\n",
+		decision.Choice,
+		decision.Confidence,
+	)
+
+	if timer == nil {
+		return
+	}
+
+	select {
+	case <-time.After(3 * time.Second):
+	case <-ctx.Done():
+		log.Fatal(ctx.Err())
+	}
+}
+
+func printHandler(
+	_ context.Context,
+	ev GoEventBus.Event,
+) (GoEventBus.Result, error) {
+	fmt.Printf("handled projection=%v event=%s\n", ev.Projection, ev.ID)
+	return GoEventBus.Result{Message: "ok"}, nil
+}

jev.go

@@ -13,8 +13,8 @@ import (
 )
 
 var (
-	// ErrNilEventSelector is returned when DecideAndSubscribe is called without
-	// an EventSelector.
+	// ErrNilEventSelector is returned when a selector-aware operation is called
+	// without an EventSelector.
 	ErrNilEventSelector = errors.New("goeventbus: event selector is nil")
 	// ErrNoEventCandidates is returned when there are no event types to choose from.
 	ErrNoEventCandidates = errors.New("goeventbus: no event candidates")

scheduling_jev.go

@@ -0,0 +1,51 @@
+package GoEventBus
+
+import (
+	"context"
+	"time"
+)
+
+// DecideAndSchedule asks selector to choose an event type immediately, assigns
+// the chosen projection, and schedules the resolved event for t.
+//
+// Selection happens when DecideAndSchedule is called, not when the timer fires.
+// This makes routing failures visible immediately and guarantees that the timer
+// always represents one already-resolved event. The scheduled event follows the
+// existing local Schedule semantics.
+func (es *EventStore) DecideAndSchedule(
+	ctx context.Context,
+	t time.Time,
+	selector EventSelector,
+	state any,
+	event Event,
+	candidates []EventCandidate,
+) (EventDecision, *time.Timer, error) {
+	decision, selected, err := selectEvent(ctx, selector, state, event, candidates)
+	if err != nil {
+		return EventDecision{}, nil, err
+	}
+
+	return decision, es.Schedule(ctx, t, selected), nil
+}
+
+// DecideAndScheduleAfter asks selector to choose an event type immediately,
+// assigns the chosen projection, and schedules the resolved event after d.
+//
+// Selection happens when DecideAndScheduleAfter is called. A non-positive
+// duration preserves ScheduleAfter's immediate-execution behavior and returns
+// a nil timer.
+func (es *EventStore) DecideAndScheduleAfter(
+	ctx context.Context,
+	d time.Duration,
+	selector EventSelector,
+	state any,
+	event Event,
+	candidates []EventCandidate,
+) (EventDecision, *time.Timer, error) {
+	decision, selected, err := selectEvent(ctx, selector, state, event, candidates)
+	if err != nil {
+		return EventDecision{}, nil, err
+	}
+
+	return decision, es.ScheduleAfter(ctx, d, selected), nil
+}