feat(memory): #62-3 Working Memory Implementation #89

Closed
opened 2026-01-05 00:13:56 +00:00 by cardosofelipe · 0 comments

Parent Issue

  • #62 Agent Memory System
  • Depends on: #62-1, #62-2

Overview

Implement the Working Memory (short-term memory) component with Redis primary storage and in-memory fallback.

Components

  • backend/app/services/memory/working/memory.py - WorkingMemory class
  • backend/app/services/memory/working/storage.py - Redis + in-memory backend

API Design

class WorkingMemory:
    async def set(self, key: str, value: Any, ttl_seconds: int | None = None) -> None
    async def get(self, key: str, default: Any = None) -> Any
    async def delete(self, key: str) -> bool
    async def exists(self, key: str) -> bool
    async def list_keys(self, pattern: str = "*") -> list[str]
    async def get_all(self) -> dict[str, Any]
    async def clear(self) -> int
    async def set_task_state(self, state: TaskState) -> None
    async def get_task_state(self) -> TaskState | None
    async def append_scratchpad(self, content: str) -> None
    async def get_scratchpad(self) -> list[str]
    async def create_checkpoint(self) -> str  # Returns checkpoint ID
    async def restore_checkpoint(self, checkpoint_id: str) -> None

Tasks

  • Implement WorkingMemory class
  • Implement Redis storage backend with connection pooling
  • Implement in-memory fallback when Redis unavailable
  • Session-scoped containers with automatic cleanup
  • Variable storage (get/set/delete)
  • Task state tracking (current step, status, progress)
  • Scratchpad for reasoning steps
  • Configurable capacity limits
  • TTL-based expiration
  • Checkpoint/snapshot support for recovery
  • Unit tests (>90% coverage)
  • Integration tests with Redis

Acceptance Criteria

  • get/set operations < 5ms P95
  • TTL expiration works correctly
  • Graceful fallback when Redis unavailable
  • Checkpoints can be created and restored
  • make validate-all passes

Priority

P0 - Core functionality

Labels

phase-2, mcp, backend, agents

## Parent Issue - #62 Agent Memory System - Depends on: #62-1, #62-2 ## Overview Implement the Working Memory (short-term memory) component with Redis primary storage and in-memory fallback. ## Components - `backend/app/services/memory/working/memory.py` - WorkingMemory class - `backend/app/services/memory/working/storage.py` - Redis + in-memory backend ## API Design ```python class WorkingMemory: async def set(self, key: str, value: Any, ttl_seconds: int | None = None) -> None async def get(self, key: str, default: Any = None) -> Any async def delete(self, key: str) -> bool async def exists(self, key: str) -> bool async def list_keys(self, pattern: str = "*") -> list[str] async def get_all(self) -> dict[str, Any] async def clear(self) -> int async def set_task_state(self, state: TaskState) -> None async def get_task_state(self) -> TaskState | None async def append_scratchpad(self, content: str) -> None async def get_scratchpad(self) -> list[str] async def create_checkpoint(self) -> str # Returns checkpoint ID async def restore_checkpoint(self, checkpoint_id: str) -> None ``` ## Tasks - [ ] Implement WorkingMemory class - [ ] Implement Redis storage backend with connection pooling - [ ] Implement in-memory fallback when Redis unavailable - [ ] Session-scoped containers with automatic cleanup - [ ] Variable storage (get/set/delete) - [ ] Task state tracking (current step, status, progress) - [ ] Scratchpad for reasoning steps - [ ] Configurable capacity limits - [ ] TTL-based expiration - [ ] Checkpoint/snapshot support for recovery - [ ] Unit tests (>90% coverage) - [ ] Integration tests with Redis ## Acceptance Criteria - [ ] get/set operations < 5ms P95 - [ ] TTL expiration works correctly - [ ] Graceful fallback when Redis unavailable - [ ] Checkpoints can be created and restored - [ ] `make validate-all` passes ## Priority P0 - Core functionality ## Labels `phase-2`, `mcp`, `backend`, `agents`
Sign in to join this conversation.