Skip to content

← Back to Roadmap Index

Phase 2: Queue Management ✅ COMPLETED

Goal: Implement core business logic for queue operations.

Status: ✅ Completed on 2026-01-19 (Commit: 52eac4c)

Tasks

  1. PulseQueue Class (src/reeve/pulse/queue.py) ✅
  2. Implemented async SQLAlchemy session management
  3. Implemented all methods:
    • schedule_pulse() - Create new pulse
    • get_due_pulses() - Query pending pulses with priority ordering
    • get_upcoming_pulses() - List future pulses
    • get_pulse() - Retrieve pulse by ID
    • mark_processing() - Transition to processing
    • mark_completed() - Mark success
    • mark_failed() - Handle failures + retry logic with exponential backoff
    • cancel_pulse() - Cancel pending
    • reschedule_pulse() - Change time
    • initialize() - Initialize database schema
    • close() - Clean up resources

  4. Priority ordering using SQLAlchemy CASE statement

  5. Unit Tests (tests/test_pulse_queue.py) ✅

  6. Test database: sqlite+aiosqlite:///:memory:

  7. 29 comprehensive unit tests covering:

    • All CRUD operations
    • Priority ordering (CRITICAL → HIGH → NORMAL → LOW → DEFERRED)
    • Time-based FIFO within same priority
    • Retry logic with exponential backoff (2^retry_count minutes)
    • Concurrent operations
    • Edge cases (nonexistent pulses, invalid states, etc.)
    • Timezone awareness

  8. Configuration (src/reeve/utils/config.py) ✅

  9. Implemented ReeveConfig class with environment variable support
  10. Path expansion for ~ and $VAR
  11. Database URL handling (async and sync modes)

  12. Singleton pattern with get_config() and reload_config()

  13. Enhanced Database Models (src/reeve/pulse/models.py) ✅

  14. Added TZDateTime custom type for proper timezone handling with SQLite
  15. Updated to SQLAlchemy 2.0 style (orm.declarative_base)

  16. All datetime fields preserve timezone information

  17. Validation Tests (tests/test_phase2_validation.py) ✅

  18. Integration test from roadmap validation example
  19. End-to-end workflow validation

Deliverables

  • ✅ Fully functional PulseQueue class
  • ✅ 100% test coverage for queue operations (33/33 tests passed)
  • ✅ Configuration management
  • ✅ Timezone-aware datetime handling
  • ✅ Code formatted with black and isort

Validation

# All tests passing
uv run pytest tests/ -v
# Result: 33 passed (3 Phase 1 + 1 Phase 2 integration + 29 Phase 2 unit tests)

Demo

Queue Operations:

# Run the demo script
uv run python demos/phase2_queue_demo.py

# Expected output:
# ✓ Initialized PulseQueue
# ✓ Scheduled pulse #1: "High priority task" (due in 5 seconds, priority: HIGH)
# ✓ Scheduled pulse #2: "Normal maintenance" (due in 10 seconds, priority: NORMAL)
# ✓ Scheduled pulse #3: "Low priority cleanup" (due in 15 seconds, priority: LOW)
#
# Upcoming pulses (3):
# 🔔 #1 - HIGH - in 5s - "High priority task"
# ⏰ #2 - NORMAL - in 10s - "Normal maintenance"
# 📋 #3 - LOW - in 15s - "Low priority cleanup"
#
# ✓ Marked pulse #1 as PROCESSING
# ✓ Marked pulse #1 as COMPLETED
#
# ✓ Simulating failure for pulse #2...
# ✓ Marked pulse #2 as FAILED (retry_count=1, will retry in 2 minutes)
#
# ✓ Cancelled pulse #3
#
# Final status:
# - Pulse #1: COMPLETED ✅
# - Pulse #2: PENDING (scheduled for retry in ~2 minutes) 🔄
# - Pulse #3: CANCELLED ❌
#
# ✓ Phase 2 Demo Complete!

Previous: Phase 1: Foundation

Next: Phase 3: MCP Servers