test: add edit failure-path and cross-agent PATCH rejection tests (Refs PR #633, BountyHub #508)

memanto/app/routes/memory.py

@@ -17,6 +17,7 @@
 from memanto.app.clients.backend import get_active_llm_model
 from memanto.app.clients.moorcheh import get_moorcheh_client
 from memanto.app.config import settings
+from memanto.app.constants import VALID_MEMORY_TYPES
 from memanto.app.core import MemoryRecord
 from memanto.app.models import (
     AnswerRequest,
@@ -120,6 +121,18 @@ class RecallRecentRequest(BaseModel):
     type: list[str] | None = Field(default=None, description="Memory type filters")
 
 
+class MemoryEditRequest(BaseModel):
+    title: str | None = Field(default=None, max_length=100)
+    content: str | None = Field(default=None, max_length=10000)
+    type: str | None = None
+    confidence: float | None = Field(default=None, ge=0.0, le=1.0)
+    tags: list[str] | None = None
+    source: str | None = None
+
+    def to_updates(self) -> dict[str, object]:
+        return self.model_dump(exclude_none=True)
+
+
 @router.post("/{agent_id}/remember")
 async def remember(
     agent_id: str,
@@ -299,6 +312,90 @@ async def batch_remember(
         raise map_error_to_http_exception(e)
 
 
+@router.patch("/{agent_id}/memories/{memory_id}")
+async def edit_memory(
+    agent_id: str,
+    memory_id: str,
+    request: MemoryEditRequest = Body(...),
+    session: Session = Depends(get_current_session),
+    client=Depends(get_moorcheh_client),
+):
+    """
+    Update one memory in the active agent's namespace (Session-based).
+
+    Requires:
+    - X-Session-Token: {session_token}
+
+    The session must be for the specified agent_id.
+    """
+    if session.agent_id != agent_id:
+        raise map_error_to_http_exception(
+            Exception(
+                f"Session is for agent '{session.agent_id}', cannot access '{agent_id}'"
+            )
+        )
+
+    updates = request.to_updates()
+    if not updates:
+        raise HTTPException(
+            status_code=400,
+            detail="Provide at least one field to update.",
+        )
+
+    if "content" in updates:
+        content = updates["content"]
+        if content is None or not str(content).strip():
+            raise HTTPException(
+                status_code=400,
+                detail="Memory content must be a non-empty string.",
+            )
+        CostGuard.validate_text_length(str(content), "Memory content")
+    if "confidence" in updates:
+        confidence = updates["confidence"]
+        try:
+            confidence_value = float(confidence)  # type: ignore[arg-type]
+        except (TypeError, ValueError):
+            raise HTTPException(
+                status_code=400,
+                detail=f"Confidence must be a number between 0.0 and 1.0, got {confidence!r}.",
+            )
+        if not 0.0 <= confidence_value <= 1.0:
+            raise HTTPException(
+                status_code=400,
+                detail=f"Confidence must be between 0.0 and 1.0, got {confidence_value}.",
+            )
+    if "type" in updates and updates["type"] not in VALID_MEMORY_TYPES:
+        raise HTTPException(
+            status_code=400,
+            detail=(
+                f"Invalid memory_type '{updates['type']}'. "
+                f"Must be one of: {', '.join(sorted(VALID_MEMORY_TYPES))}."
+            ),
+        )
+
+    try:
+        write_service = MemoryWriteService(client)
+        result = await asyncio.to_thread(
+            write_service.update_memory, memory_id, session.namespace, updates
+        )
+        return {
+            "agent_id": agent_id,
+            "session_id": session.session_id,
+            "namespace": session.namespace,
+            "memory_id": memory_id,
+            "status": result.get("status", "updated"),
+            "action": result.get("action", "updated"),
+            "updated_fields": result.get("updated_fields", list(updates.keys())),
+        }
+
+    except Exception as e:
+        if "not found" in str(e).lower():
+            raise HTTPException(
+                status_code=404, detail=f"Memory '{memory_id}' was not found."
+            )
+        raise map_error_to_http_exception(e)
+
+
 @router.post("/{agent_id}/upload-file")
 async def upload_file(
     agent_id: str,

memanto/cli/client/direct_client.py

@@ -729,6 +729,81 @@ def batch_remember(
 
         return result
 
+    def update_memory(
+        self, agent_id: str, memory_id: str, updates: dict[str, Any]
+    ) -> dict[str, Any]:
+        """
+        Update a single memory in the active agent namespace.
+
+        Args:
+            agent_id: Target agent.
+            memory_id: Memory document ID to update.
+            updates: Fields to update.
+
+        Returns:
+            Dict with update result metadata.
+
+        Raises:
+            ValueError: If no update fields are provided.
+        """
+        session = self._get_validated_session_for_agent(agent_id)
+        if not updates:
+            raise ValueError("Provide at least one field to update")
+        _ALLOWED_UPDATE_FIELDS = {
+            "title", "content", "type", "confidence", "tags", "source",
+        }
+        unknown_fields = set(updates) - _ALLOWED_UPDATE_FIELDS
+        if unknown_fields:
+            raise ValueError(
+                f"Unknown update fields: {', '.join(sorted(unknown_fields))}. "
+                f"Allowed fields: {', '.join(sorted(_ALLOWED_UPDATE_FIELDS))}."
+            )
+        if "content" in updates:
+            content = updates["content"]
+            if content is None or not str(content).strip():
+                raise ValueError("Memory content must be a non-empty string")
+            if len(str(content)) > _MAX_CONTENT_LENGTH:
+                raise ValueError(
+                    f"Memory content exceeds {_MAX_CONTENT_LENGTH} characters"
+                )
+        if "title" in updates:
+            title = updates["title"]
+            if title is not None and len(str(title)) > _MAX_TITLE_LENGTH:
+                raise ValueError(
+                    f"Memory title exceeds {_MAX_TITLE_LENGTH} characters"
+                )
+        if "type" in updates:
+            memory_type = updates["type"]
+            if memory_type not in _VALID_MEMORY_TYPES:
+                raise ValueError(
+                    f"Invalid memory_type '{memory_type}'. "
+                    f"Must be one of: {', '.join(sorted(_VALID_MEMORY_TYPES))}"
+                )
+        if "confidence" in updates:
+            try:
+                confidence_value = float(updates["confidence"])  # type: ignore[arg-type]
+            except (TypeError, ValueError):
+                raise ValueError(
+                    f"Confidence must be a number between 0.0 and 1.0, got {updates['confidence']!r}"
+                )
+            if not 0.0 <= confidence_value <= 1.0:
+                raise ValueError(
+                    f"Confidence must be between 0.0 and 1.0, got {confidence_value}"
+                )
+
+        result = self._get_write_service().update_memory(
+            memory_id, session.namespace, updates
+        )
+
+        return {
+            "agent_id": agent_id,
+            "namespace": session.namespace,
+            "memory_id": memory_id,
+            "status": result.get("status", "updated"),
+            "action": result.get("action", "updated"),
+            "updated_fields": result.get("updated_fields", list(updates.keys())),
+        }
+
     def recall(
         self,
         agent_id: str,

memanto/cli/client/sdk_client.py

@@ -566,6 +566,40 @@ def batch_remember(
 
         return result
 
+    def update_memory(
+        self, agent_id: str, memory_id: str, updates: dict[str, Any]
+    ) -> dict[str, Any]:
+        """
+        Update a single memory in the active agent namespace.
+
+        Args:
+            agent_id: Target agent.
+            memory_id: Memory document ID to update.
+            updates: Fields to update.
+
+        Returns:
+            Dict with update result metadata.
+
+        Raises:
+            ValueError: If no update fields are provided.
+        """
+        session = self._get_validated_session_for_agent(agent_id)
+        if not updates:
+            raise ValueError("Provide at least one field to update")
+
+        result = self._get_write_service().update_memory(
+            memory_id, session.namespace, updates
+        )
+
+        return {
+            "agent_id": agent_id,
+            "namespace": session.namespace,
+            "memory_id": memory_id,
+            "status": result.get("status", "updated"),
+            "action": result.get("action", "updated"),
+            "updated_fields": result.get("updated_fields", list(updates.keys())),
+        }
+
     def upload_file(self, agent_id: str, file_path: str) -> dict[str, Any]:
         """
         Upload a file directly to the agent's memory namespace.

memanto/cli/commands/memory.py

@@ -172,6 +172,70 @@ def remember(
         _error(f"Failed to store memory: {e}")
 
 
+@app.command()
+def edit(
+    memory_id: str = typer.Argument(..., help="Memory ID to update"),
+    title: str | None = typer.Option(None, "--title", help="New memory title"),
+    content: str | None = typer.Option(None, "--content", help="New memory content"),
+    memory_type: str | None = typer.Option(
+        None, "--type", "-t", help="New memory type"
+    ),
+    confidence: float | None = typer.Option(
+        None, "--confidence", "-c", help="New confidence score (0.0-1.0)"
+    ),
+    tags: str | None = typer.Option(None, "--tags", help="New comma-separated tags"),
+    source: str | None = typer.Option(None, "--source", "-s", help="New memory source"),
+):
+    """Update fields on an existing memory for the active agent."""
+    start = time.perf_counter()
+    active_agent_id, active_session_token = config_manager.get_active_session()
+
+    if not active_agent_id or not active_session_token:
+        _error(
+            "No active agent.", hint="Run 'memanto agent activate <agent-id>' first."
+        )
+
+    updates: dict[str, object] = {}
+    if title is not None:
+        updates["title"] = title
+    if content is not None:
+        updates["content"] = content
+    if memory_type is not None:
+        updates["type"] = memory_type
+    if confidence is not None:
+        updates["confidence"] = confidence
+    if tags is not None:
+        updates["tags"] = [t.strip() for t in tags.split(",") if t.strip()]
+    if source is not None:
+        updates["source"] = source
+
+    if not updates:
+        _error(
+            "No update fields provided.",
+            hint="Pass at least one of: --title, --content, --type, --confidence, --tags, --source.",
+        )
+
+    client = get_client()
+
+    try:
+        with console.status("[cyan]Updating memory...", spinner="dots"):
+            result = client.update_memory(
+                agent_id=active_agent_id,
+                memory_id=memory_id,
+                updates=updates,
+            )
+        elapsed = time.perf_counter() - start
+
+        updated_fields = ", ".join(result.get("updated_fields", updates.keys()))
+        console.print("[green]Memory updated successfully![/green]")
+        console.print(f"[dim]Memory ID: {result.get('memory_id', memory_id)}[/dim]")
+        console.print(f"[dim]Updated fields: {updated_fields}[/dim]")
+        console.print(f"[dim]Completed in {elapsed:.2f}s[/dim]")
+
+    except Exception as e:
+        _error(f"Failed to update memory: {e}")
+
+
 @app.command()
 def upload(
     file_path: str = typer.Argument(..., help="Path to the file to upload"),