""" AgentOS – Agent memory service. Provides append-only lo...

Prompt

""" AgentOS – Agent memory service. Provides append-only long-term knowledge storage with explicit retention policies, provenance metadata, and soft-deletion support. Scoped globally (no tenant/workspace concept) with optional linkage back to the task that produced an entry. The LLM conversation window is NEVER used as the authoritative source of state; compact restorable summaries are generated from external storage. """ from __future__ import annotations import json import uuid from datetime import datetime, timezone from typing import Any, Dict, List, Optional from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from core.config import settings from core.logging_config import get_logger from core.storage import memory_key, upload_text from models.memory import AgentMemoryEntry logger = get_logger(__name__) class MemoryService: """Manages globally-scoped long-term memory entries.""" def __init__(self, db: AsyncSession) -> None: self._db = db async def store_entry( self, category: str, title: str, content: str, source_task_id: Optional[uuid.UUID] = None, source_type: Optional[str] = None, source_reference: Optional[str] = None, confidence: Optional[float] = None, structured_data: Optional[Dict[str, Any]] = None, retention_policy: str = "permanent", expires_at: Optional[datetime] = None, ) -> AgentMemoryEntry: """ Persist a new memory entry to the knowledge store. Uses append-only semantics; existing entries are never modified. """ entry = AgentMemoryEntry( id=uuid.uuid4(), source_task_id=source_task_id, category=category, title=title, content=content, source_type=source_type, source_reference=source_reference, confidence=confidence, structured_data=structured_data, retention_policy=retention_policy, expires_at=expires_at, ) self._db.add(entry) await self._db.flush() # Upload a copy to object storage for long-term durable archival obj_key = memory_key("global", str(entry.id)) payload = { "id": str(entry.id), "category": category, "title": title, "content": content, "source_type": source_type, "source_reference": source_reference, "confidence": confidence, "structured_data": structured_data, "retention_policy": retention_policy, "created_at": entry.created_at.isoformat() if entry.created_at else None, } await upload_text( bucket=settings.MINIO_ARTIFACTS_BUCKET, key=obj_key, text=json.dumps(payload, default=str, indent=2), content_type="application/json", ) logger.info( "Memory entry stored", entry_id=str(entry.id), category=category, title=title[:60], ) return entry async def retrieve_entries( self, category: Optional[str] = None, limit: int = 50, offset: int = 0, ) -> List[AgentMemoryEntry]: """ Retrieve non-deleted memory entries. Optionally filter by category. Results are sorted newest-first. """ from sqlalchemy import desc stmt = ( select(AgentMemoryEntry) .where(AgentMemoryEntry.is_deleted.is_(False)) .order_by(desc(AgentMemoryEntry.created_at)) .limit(limit) .offset(offset) ) if category: stmt = stmt.where(AgentMemoryEntry.category == category) result = await self._db.execute(stmt) return list(result.scalars().all()) async def soft_delete_entry(self, entry_id: uuid.UUID) -> bool: """ Soft-delete a memory entry. Returns True if the entry was found and deleted; False if not found. """ result = await self._db.execute( select(AgentMemoryEntry).where( AgentMemoryEntry.id == entry_id, AgentMemoryEntry.is_deleted.is_(False), ) ) entry: Optional[AgentMemoryEntry] = result.scalar_one_or_none() if entry is None: return False entry.is_deleted = True entry.deleted_at = datetime.now(tz=timezone.utc) await self._db.flush() logger.info("Memory entry deleted", entry_id=str(entry_id)) return True async def generate_context_summary( self, task_id: Optional[uuid.UUID] = None, max_entries: int = 20, ) -> str: """ Generate a compact, restorable state summary for model context. This summary is derived from external storage, not the LLM context window. The full raw logs and artifacts are retained externally. """ entries = await self.retrieve_entries(limit=max_entries) if not entries: return "No relevant memory entries found." lines = ["## Agent Memory Summary", ""] for entry in entries: lines.append( f"- [{entry.category}] **{entry.title}**: " f"{entry.content[:200]}" + ("..." if len(entry.content) > 200 else "") ) if entry.confidence is not None: lines[-1] += f" (confidence: {entry.confidence:.2f})" if entry.source_reference: lines[-1] += f" [source: {entry.source_reference[:80]}]" lines.append("") lines.append( f"*{len(entries)} entries shown; full history available in external storage.*" ) return "\n".join(lines) Send back the complete code with all the fixes. Fix each of the listed errors one by one, making sure to actually correct them so that there are 0 errors remaining. Keep the original imports, since the files exist. Write out every single character; do not abbreviate anything. Fix every error. There must be exactly one file. Do not write anything else; just output the complete code, and it must not contain any comments. Never, under any circumstances, use simplified, substitute, dummy, simulated, or fake code. Write the entire file as complete, unabridged, production-ready code in a single code block. It must be 100% error-free, a complete, error-free file, and must be submitted as a downloadable file. These requirements are mandatory and must be strictly adhered to. If no list of errors is provided, you must find all the errors and fix them. If there were comments in the original code, delete them. And most importantly: YOU MUST NEVER SIMPLIFY!