File size: 2,742 Bytes
f02fdcc | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 | """
Working Context Memory — temporary state for the active task.
Holds structured text, latent state references, subgoal trees, and
immediate execution context. Enforces a maximum slot limit defined
by memory_max_working_context. Oldest items are evicted first.
"""
from collections import OrderedDict
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional
from .models import MemoryItem, MemoryTier
from src.config import settings
class WorkingContextMemory:
def __init__(self):
self._slots: OrderedDict[str, MemoryItem] = OrderedDict()
self._subgoals: List[Dict[str, Any]] = []
# ── Context slots ─────────────────────────────────────────────────
def set(self, item: MemoryItem) -> MemoryItem:
item.tier = MemoryTier.WORKING
if item.item_id in self._slots:
self._slots.move_to_end(item.item_id)
self._slots[item.item_id] = item
self._evict_if_full()
return item
def get(self, item_id: str) -> Optional[MemoryItem]:
return self._slots.get(item_id)
def all(self) -> List[MemoryItem]:
return list(self._slots.values())
def clear(self) -> None:
self._slots.clear()
self._subgoals.clear()
# ── Subgoal tree ──────────────────────────────────────────────────
def push_subgoal(self, goal: str, metadata: Optional[Dict[str, Any]] = None) -> None:
self._subgoals.append({
"goal": goal,
"status": "pending",
"created_at": datetime.now(timezone.utc).isoformat(),
"metadata": metadata or {},
})
def complete_subgoal(self, goal: str) -> None:
for sg in self._subgoals:
if sg["goal"] == goal and sg["status"] == "pending":
sg["status"] = "completed"
break
def pending_subgoals(self) -> List[Dict[str, Any]]:
return [sg for sg in self._subgoals if sg["status"] == "pending"]
def subgoal_tree(self) -> List[Dict[str, Any]]:
return list(self._subgoals)
# ── Eviction ──────────────────────────────────────────────────────
def _evict_if_full(self) -> None:
while len(self._slots) > settings.memory_max_working_context:
self._slots.popitem(last=False) # evict oldest
@property
def utilization(self) -> float:
return len(self._slots) / settings.memory_max_working_context
|