diff --git a/backend/app/agent/toolkit/memory_toolkit.py b/backend/app/agent/toolkit/memory_toolkit.py index d715843b6..cc0c08640 100644 --- a/backend/app/agent/toolkit/memory_toolkit.py +++ b/backend/app/agent/toolkit/memory_toolkit.py @@ -16,6 +16,9 @@ from __future__ import annotations +import asyncio +import json +import uuid from dataclasses import asdict from camel.toolkits import BaseToolkit, FunctionTool @@ -23,8 +26,16 @@ from camel.toolkits import BaseToolkit, FunctionTool from app.agent.toolkit.abstract_toolkit import AbstractToolkit from app.lightweight_memory import get_lightweight_memory_service from app.run_policy import ToolSafetyClass -from app.run_runtime.tool_checkpoint import declare_tool_safety -from app.service.task import get_task_lock +from app.run_runtime.tool_checkpoint import ( + declare_tool_safety, + get_current_tool_checkpoint, +) +from app.service.task import ( + TASK_LOCK_CLEANUP_SENTINEL, + Action, + ActionAskData, + get_task_lock, +) class MemoryToolkit(BaseToolkit, AbstractToolkit): @@ -89,6 +100,7 @@ class MemoryToolkit(BaseToolkit, AbstractToolkit): }: raise ValueError("invalid Memory source trust") context = self._run_context() + activity_id, decision_id = self._audit_link() result = get_lightweight_memory_service().create_entry( scope_type="project", scope_id=context.project_id, @@ -100,6 +112,8 @@ class MemoryToolkit(BaseToolkit, AbstractToolkit): source_refs=tuple(source_event_ids or ()), actor_id=self.agent_name, run_id=context.run_id, + activity_id=activity_id, + decision_id=decision_id, ) return { "entry": asdict(result.entry) @@ -108,7 +122,7 @@ class MemoryToolkit(BaseToolkit, AbstractToolkit): "scope_state": asdict(result.scope_state), } - def update_project_memory( + async def update_project_memory( self, memory_id: str, expected_version: int, @@ -134,28 +148,46 @@ class MemoryToolkit(BaseToolkit, AbstractToolkit): ) if existing is None or existing.scope_id != context.project_id: raise ValueError("Memory entry is outside the current Project") - if existing.confirmed_by_user or existing.pinned_by_user: - raise PermissionError( - "Confirmed or pinned Memory requires user review" + activity_id, decision_id = self._audit_link() + actor_type = "agent" + actor_id: str | None = self.agent_name + source_trust = "model_inferred" + if ( + existing.created_by != "agent" + or existing.confirmed_by_user + or existing.pinned_by_user + ): + decision_id = await self._request_memory_review( + operation="replace", + existing=existing, + proposed={"kind": kind, "content": content}, + reason=reason, ) + if decision_id is None: + return {"status": "rejected", "entry": asdict(existing)} + actor_type = "user" + actor_id = None + source_trust = "user_confirmed" result = get_lightweight_memory_service().update_entry( memory_id=memory_id, expected_version=expected_version, content=content, kind=kind, - actor_type="agent", + actor_type=actor_type, reason=reason, request_id=( f"agent-update:{context.run_id}:{memory_id}:{expected_version}" ), - source_trust=existing.source_trust, + source_trust=source_trust, source_refs=existing.source_refs, - actor_id=self.agent_name, + actor_id=actor_id, run_id=context.run_id, + activity_id=activity_id, + decision_id=decision_id, ) return {"entry": asdict(result.entry)} - def forget_project_memory( + async def forget_project_memory( self, memory_id: str, expected_version: int, @@ -178,20 +210,115 @@ class MemoryToolkit(BaseToolkit, AbstractToolkit): ) if existing is None or existing.scope_id != context.project_id: raise ValueError("Memory entry is outside the current Project") + activity_id, decision_id = self._audit_link() + actor_type = "agent" + actor_id: str | None = self.agent_name + if ( + existing.created_by != "agent" + or existing.confirmed_by_user + or existing.pinned_by_user + ): + decision_id = await self._request_memory_review( + operation="remove", + existing=existing, + proposed=None, + reason=reason, + ) + if decision_id is None: + return {"status": "rejected", "entry": asdict(existing)} + actor_type = "user" + actor_id = None result = get_lightweight_memory_service().transition_entry( memory_id=memory_id, expected_version=expected_version, operation="remove", - actor_type="agent", + actor_type=actor_type, reason=reason, request_id=( f"agent-remove:{context.run_id}:{memory_id}:{expected_version}" ), - actor_id=self.agent_name, + actor_id=actor_id, run_id=context.run_id, + activity_id=activity_id, + decision_id=decision_id, ) return {"entry": asdict(result.entry)} + async def promote_project_memory( + self, + memory_id: str, + expected_version: int, + target_scope: str, + reason: str, + ) -> dict: + """Propose adopting Project Memory into the current Space or User scope. + + Args: + memory_id: Stable identifier of the Project Memory entry. + expected_version: Version currently visible to the Agent. + target_scope: Destination scope, either ``space`` or ``user``. + reason: Human-readable reason shown in the review card. + """ + + if target_scope not in {"space", "user"}: + raise ValueError("target_scope must be space or user") + context = self._run_context() + service = get_lightweight_memory_service() + existing = service.journal.get_memory_entry(memory_id) + if ( + existing is None + or existing.scope_type != "project" + or existing.scope_id != context.project_id + ): + raise ValueError("Memory entry is outside the current Project") + if existing.version != expected_version: + raise ValueError("Memory entry version changed") + target_scope_id = ( + context.space_id + if target_scope == "space" + else str(context.user_id) + ) + if not target_scope_id or target_scope_id == "None": + raise ValueError(f"Run has no {target_scope} scope") + decision_id = await self._request_memory_review( + operation="promote", + existing=existing, + proposed={ + "target_scope": target_scope, + "target_scope_id": target_scope_id, + "kind": existing.kind, + "content": existing.content, + }, + reason=reason, + ) + if decision_id is None: + return {"status": "rejected", "entry": asdict(existing)} + activity_id, _ = self._audit_link() + result = service.create_entry( + scope_type=target_scope, + scope_id=target_scope_id, + kind=existing.kind, + content=existing.content, + actor_type="user", + reason=reason, + source_trust="user_confirmed", + source_refs=existing.source_refs, + priority=existing.priority, + sensitivity=existing.sensitivity, + request_id=( + f"memory-promote:{context.run_id}:{memory_id}:" + f"{expected_version}:{target_scope}" + ), + actor_id=None, + run_id=context.run_id, + activity_id=activity_id, + decision_id=decision_id, + ) + return { + "status": "promoted", + "entry": asdict(result.entry) if result.entry else None, + } + def search_project_history( self, query: str, @@ -225,11 +352,12 @@ class MemoryToolkit(BaseToolkit, AbstractToolkit): FunctionTool(self.remember_project_memory), FunctionTool(self.update_project_memory), FunctionTool(self.forget_project_memory), + FunctionTool(self.promote_project_memory), FunctionTool(self.search_project_history), ] - for tool in (tools[0], tools[4]): + for tool in (tools[0], tools[5]): declare_tool_safety(tool, ToolSafetyClass.SAFE_READ) - for tool in tools[1:4]: + for tool in tools[1:5]: declare_tool_safety(tool, ToolSafetyClass.UNSAFE_WRITE) for tool in tools: try: @@ -245,6 +373,106 @@ class MemoryToolkit(BaseToolkit, AbstractToolkit): raise RuntimeError("Memory tools require an admitted RunContext") return context + def _audit_link(self) -> tuple[str | None, str | None]: + checkpoint = get_current_tool_checkpoint() + if checkpoint is None: + return None, None + decisions = get_lightweight_memory_service().journal.list_human_interaction_decisions( + f"approval:{checkpoint.tool_call_id}" + ) + return ( + checkpoint.tool_call_id, + decisions[-1].decision_id if decisions else None, + ) + + async def _request_memory_review( + self, + *, + operation: str, + existing, + proposed: dict | None, + reason: str, + ) -> str | None: + context = self._run_context() + service = get_lightweight_memory_service() + run = service.journal.get_run(context.run_id) + if run is None or run.active_attempt_id is None: + raise RuntimeError("Memory review requires an active RunAttempt") + interaction_id = str( + uuid.uuid5( + uuid.NAMESPACE_URL, + f"eigent:memory-review:{context.run_id}:{existing.memory_id}:" + f"{existing.version}:{operation}:" + f"{json.dumps(proposed, sort_keys=True, separators=(',', ':'))}", + ) + ) + question = ( + f"Allow the Agent to {operation} Memory '{existing.content}'?" + ) + service.journal.create_human_interaction( + interaction_id=interaction_id, + run_id=context.run_id, + attempt_id=run.active_attempt_id, + interaction_type="memory_change_review", + request={ + "title": "Review Memory change", + "question": question, + "agent": self.agent_name, + "memory_change": { + "operation": operation, + "memory_id": existing.memory_id, + "expected_version": existing.version, + "before": asdict(existing), + "after": proposed, + "reason": reason, + }, + }, + response_schema={ + "type": "object", + "properties": {"decision": {"enum": ["approved", "rejected"]}}, + "required": ["decision"], + "additionalProperties": False, + }, + requested_by=f"agent:{self.agent_name}", + ) + try: + from app.run_sync.runtime import notify_default_cloud_sync_worker + + notify_default_cloud_sync_worker() + except Exception: + pass + task_lock = get_task_lock(self.api_task_id) + await task_lock.put_queue( + ActionAskData( + action=Action.ask, + data={ + "question": question, + "title": "Review Memory change", + "agent": self.agent_name, + "interaction_id": interaction_id, + "interaction_type": "memory_change_review", + "run_id": context.run_id, + "version": 0, + "display_arguments": { + "before": asdict(existing), + "after": proposed, + "reason": reason, + }, + }, + ) + ) + reply = await task_lock.get_human_input(self.agent_name) + if reply == TASK_LOCK_CLEANUP_SENTINEL: + raise asyncio.CancelledError("Memory review interrupted") + if str(reply).casefold() != "approved": + return None + decisions = service.journal.list_human_interaction_decisions( + interaction_id + ) + if not decisions: + raise RuntimeError("Memory review decision was not persisted") + return decisions[-1].decision_id + @classmethod def toolkit_name(cls) -> str: return "Memory Toolkit" diff --git a/backend/app/controller/memory_controller.py b/backend/app/controller/memory_controller.py index 2aaadcfae..4b769845c 100644 --- a/backend/app/controller/memory_controller.py +++ b/backend/app/controller/memory_controller.py @@ -164,6 +164,9 @@ async def list_memory_entries( return { "scope_state": asdict(state), "items": [asdict(item) for item in entries], + "sync_status": service.journal.get_memory_sync_status( + scope_type, scope_id + ), } diff --git a/backend/app/lightweight_memory/maintainer.py b/backend/app/lightweight_memory/maintainer.py index 307f00bf7..eab9d433f 100644 --- a/backend/app/lightweight_memory/maintainer.py +++ b/backend/app/lightweight_memory/maintainer.py @@ -155,92 +155,99 @@ class IncrementalMemoryMaintainer: f"{state.revision}" ), ).scope_state - after = state.processed_through_watermark - page = self._service.search_history( - project_id=project_id, - after_cursor=after, - limit=100, - byte_budget=256 * 1024, - token_budget=16384, - ) - if parse_project_cursor(page.next_cursor) == parse_project_cursor( - after - ): - return state - active = self._service.list_entries("project", project_id) - proposals = self._extractor.extract( - active_memory=active, - history_delta=page.items, - )[:3] - projected_tokens = state.current_token_count - bounded_proposals: list[ProposedMemoryMutation] = [] - for proposal in proposals: - proposal_tokens = count_tokens(proposal.content) - if ( - projected_tokens + proposal_tokens - > state.token_limit * 0.9 - ): - continue - bounded_proposals.append(proposal) - projected_tokens += proposal_tokens - proposals = tuple(bounded_proposals) - cursor_from = after or format_project_cursor(0) - if not proposals: - identity = hashlib.sha256( - ( - f"{self._extractor.version}|{project_id}|" - f"{cursor_from}|{page.next_cursor}|noop" - ).encode() - ).hexdigest() - self._service.journal.apply_memory_mutation( - mutation_id=f"mut_{identity[:32]}", - idempotency_key=f"memory-extract-noop:{identity}", - operation="noop", - scope_type="project", - scope_id=project_id, - memory_id=None, - actor_type="extractor", - reason=( - "incremental extraction found no durable Memory " - f"in {cursor_from}..{page.next_cursor}" - ), - source_refs=tuple( - item.event_id for item in page.items[:32] - ), + # One terminal trigger may represent a very long Run. Process a + # bounded number of pages instead of silently stopping after the + # first 100 events. The scheduler queues another bounded pass when + # history still remains. + for _ in range(10): + after = state.processed_through_watermark + page = self._service.search_history( + project_id=project_id, + after_cursor=after, + limit=100, + byte_budget=256 * 1024, + token_budget=16384, ) - for index, proposal in enumerate(proposals): - identity = hashlib.sha256( - ( - f"{self._extractor.version}|{project_id}|" - f"{cursor_from}|" - f"{page.next_cursor}|{index}|{proposal.kind}|" - f"{proposal.content}" - ).encode() - ).hexdigest() - self._service.create_entry( - scope_type="project", - scope_id=project_id, - kind=proposal.kind, - content=proposal.content, - actor_type="extractor", - reason=( - "incremental extraction " - f"{cursor_from}..{page.next_cursor}" - ), - source_trust=proposal.source_trust, - source_refs=proposal.source_event_ids, - sensitivity=proposal.sensitivity, - request_id=f"memory-extract:{identity}", + if parse_project_cursor( + page.next_cursor + ) == parse_project_cursor(after): + return state + active = self._service.list_entries("project", project_id) + proposals = self._extractor.extract( + active_memory=active, + history_delta=page.items, + )[:3] + projected_tokens = state.current_token_count + bounded_proposals: list[ProposedMemoryMutation] = [] + for proposal in proposals: + proposal_tokens = count_tokens(proposal.content) + if ( + projected_tokens + proposal_tokens + > state.token_limit * 0.9 + ): + continue + bounded_proposals.append(proposal) + projected_tokens += proposal_tokens + proposals = tuple(bounded_proposals) + cursor_from = after or format_project_cursor(0) + if not proposals: + identity = hashlib.sha256( + ( + f"{self._extractor.version}|{project_id}|" + f"{cursor_from}|{page.next_cursor}|noop" + ).encode() + ).hexdigest() + self._service.journal.apply_memory_mutation( + mutation_id=f"mut_{identity[:32]}", + idempotency_key=f"memory-extract-noop:{identity}", + operation="noop", + scope_type="project", + scope_id=project_id, + memory_id=None, + actor_type="extractor", + reason=( + "incremental extraction found no durable Memory " + f"in {cursor_from}..{page.next_cursor}" + ), + source_refs=tuple( + item.event_id for item in page.items[:32] + ), + ) + for index, proposal in enumerate(proposals): + identity = hashlib.sha256( + ( + f"{self._extractor.version}|{project_id}|" + f"{cursor_from}|{page.next_cursor}|{index}|" + f"{proposal.kind}|{proposal.content}" + ).encode() + ).hexdigest() + self._service.create_entry( + scope_type="project", + scope_id=project_id, + kind=proposal.kind, + content=proposal.content, + actor_type="extractor", + reason=( + "incremental extraction " + f"{cursor_from}..{page.next_cursor}" + ), + source_trust=proposal.source_trust, + source_refs=proposal.source_event_ids, + sensitivity=proposal.sensitivity, + request_id=f"memory-extract:{identity}", + ) + current = self._service.scope("project", project_id) + state = self._service.journal.record_memory_maintenance_result( + "project", + project_id, + expected_revision=current.revision, + processed_through_watermark=page.next_cursor, + watermark_kind="journal_cursor", + extractor_version=self._extractor.version, ) - current = self._service.scope("project", project_id) - return self._service.journal.record_memory_maintenance_result( - "project", - project_id, - expected_revision=current.revision, - processed_through_watermark=page.next_cursor, - watermark_kind="journal_cursor", - extractor_version=self._extractor.version, - ) + if page.complete: + return state + return state except Exception as exc: current = self._service.scope("project", project_id) try: @@ -274,7 +281,14 @@ def schedule_project_memory_maintenance(project_id: str) -> None: def _done(completed: Future) -> None: _FUTURES.discard(completed) try: - completed.result() + state = completed.result() + if ( + parse_project_cursor(state.processed_through_watermark) + < get_lightweight_memory_service().journal.get_project_history_cursor( + project_id + ) + ): + schedule_project_memory_maintenance(project_id) except Exception: logger.exception( "Incremental Memory maintenance failed", diff --git a/backend/app/lightweight_memory/service.py b/backend/app/lightweight_memory/service.py index c08a7b150..42401d133 100644 --- a/backend/app/lightweight_memory/service.py +++ b/backend/app/lightweight_memory/service.py @@ -135,7 +135,9 @@ class LightweightMemoryService: actor_id: str | None = None, run_id: str | None = None, activity_id: str | None = None, + decision_id: str | None = None, ) -> MemoryMutationResult: + source_refs = tuple(dict.fromkeys(source_refs)) self._assert_mutation_policy( scope_type=scope_type, kind=kind, @@ -143,6 +145,13 @@ class LightweightMemoryService: actor_type=actor_type, source_trust=source_trust, ) + if actor_type == "agent": + self._assert_agent_provenance( + scope_type=scope_type, + scope_id=scope_id, + source_trust=source_trust, + source_refs=source_refs, + ) state = self.scope(scope_type, scope_id) if ( actor_type in {"agent", "extractor"} @@ -177,6 +186,7 @@ class LightweightMemoryService: source_refs=source_refs, run_id=run_id, activity_id=activity_id, + decision_id=decision_id, ) def consolidate_scope( @@ -274,9 +284,16 @@ class LightweightMemoryService: actor_id: str | None = None, run_id: str | None = None, activity_id: str | None = None, + decision_id: str | None = None, ) -> MemoryMutationResult: existing = self._require_entry(memory_id) - trust = source_trust or existing.source_trust + # Rewritten model text is a new claim. It must not inherit a stronger + # user/tool provenance merely because it replaced an older entry. + trust = ( + "model_inferred" + if actor_type == "agent" + else source_trust or existing.source_trust + ) self._assert_mutation_policy( scope_type=existing.scope_type, kind=kind, @@ -305,6 +322,7 @@ class LightweightMemoryService: source_refs=source_refs, run_id=run_id, activity_id=activity_id, + decision_id=decision_id, ) def transition_entry( @@ -319,10 +337,12 @@ class LightweightMemoryService: actor_id: str | None = None, run_id: str | None = None, activity_id: str | None = None, + decision_id: str | None = None, ) -> MemoryMutationResult: existing = self._require_entry(memory_id) if actor_type == "agent" and ( existing.scope_type != "project" + or existing.created_by != "agent" or existing.confirmed_by_user or existing.pinned_by_user or operation in {"confirm", "pin"} @@ -344,6 +364,7 @@ class LightweightMemoryService: reason=reason, run_id=run_id, activity_id=activity_id, + decision_id=decision_id, ) def search_memory( @@ -533,6 +554,39 @@ class LightweightMemoryService: raise KeyError(memory_id) return entry + def _assert_agent_provenance( + self, + *, + scope_type: str, + scope_id: str, + source_trust: str, + source_refs: tuple[str, ...], + ) -> None: + """Prevent model-authored Memory from laundering source authority.""" + + if source_trust != "user_asserted": + return + if scope_type != "project" or not source_refs: + raise PermissionError( + "Agent user_asserted Memory requires cited user History events" + ) + events = self._journal.get_events_by_id(source_refs) + valid = len(events) == len(source_refs) + for event in events: + run = self._journal.get_run(event.run_id) + if ( + event.event_type != "user.message" + or run is None + or run.project_id != scope_id + ): + valid = False + break + if not valid: + raise PermissionError( + "Agent user_asserted Memory citations must be user.message " + "events from the same Project" + ) + def event_source_trust(event_type: str) -> str: if event_type == "user.message" or event_type.startswith("legacy.human"): diff --git a/backend/app/run_journal/store.py b/backend/app/run_journal/store.py index 75db09284..94b4ed6f5 100644 --- a/backend/app/run_journal/store.py +++ b/backend/app/run_journal/store.py @@ -7706,6 +7706,29 @@ class SQLiteRunJournal: rows = self._connection.execute(query, parameters).fetchall() return [self._event_from_row(row) for row in rows] + def get_events_by_id( + self, event_ids: tuple[str, ...] | list[str] + ) -> list[CommittedRunEvent]: + """Return canonical events in caller order for provenance checks.""" + + identifiers = tuple(dict.fromkeys(event_ids)) + if not identifiers: + return [] + with self._lock: + rows = self._connection.execute( + """ + SELECT event_id, run_id, sequence, run_version, event_type, + payload_json, legacy_step, created_at + FROM run_events + WHERE event_id IN (SELECT value FROM json_each(?)) + """, + (json.dumps(identifiers, separators=(",", ":")),), + ).fetchall() + by_id = {row["event_id"]: self._event_from_row(row) for row in rows} + return [ + by_id[event_id] for event_id in identifiers if event_id in by_id + ] + def get_project_history_cursor(self, project_id: str) -> int: """Return the last committed Project History cursor.""" @@ -8251,7 +8274,7 @@ class SQLiteRunJournal: confirmed_by_user, created_by, source_trust, sensitivity, source_refs_json, usage_count, created_at, updated_at - ) VALUES (?, ?, ?, ?, ?, ?, 1, ?, 0, 0, ?, ?, ?, ?, 0, ?, ?) + ) VALUES (?, ?, ?, ?, ?, ?, 1, ?, 0, ?, ?, ?, ?, ?, 0, ?, ?) """, ( memory_id, @@ -8261,6 +8284,7 @@ class SQLiteRunJournal: content.strip() if content else content, priority, token_count, + int(actor_type == "user"), created_by, source_trust, sensitivity, @@ -8326,6 +8350,8 @@ class SQLiteRunJournal: SET kind = ?, content = ?, priority = ?, token_count = ?, created_by = ?, source_trust = ?, sensitivity = ?, source_refs_json = ?, + confirmed_by_user = CASE + WHEN ? THEN 1 ELSE confirmed_by_user END, version = version + 1, updated_at = ? WHERE memory_id = ? AND version = ? """, @@ -8338,6 +8364,7 @@ class SQLiteRunJournal: resolved_trust, sensitivity, replacement_refs, + int(actor_type == "user"), timestamp, memory_id, expected_version, @@ -9624,6 +9651,7 @@ class SQLiteRunJournal: "diff_review", "merge_conflict", "credential_binding", + "memory_change_review", }: if interaction_type == "approval": raise ValueError( @@ -12383,6 +12411,40 @@ class SQLiteRunJournal: ) return snapshots + def get_memory_sync_status( + self, scope_type: str, scope_id: str + ) -> dict[str, Any]: + """Return truthful local delivery state for the Memory Center.""" + + with self._lock: + row = self._connection.execute( + """ + SELECT + SUM(CASE WHEN status IN ('pending', 'sending') + THEN 1 ELSE 0 END) AS pending_count, + SUM(CASE WHEN status = 'dead_letter' + THEN 1 ELSE 0 END) AS blocked_count, + MAX(CASE WHEN status != 'sent' THEN last_error END) + AS last_error, + MAX(CASE WHEN status = 'sent' THEN updated_at END) + AS last_synced_at + FROM memory_mutation_outbox + WHERE scope_type = ? AND scope_id = ? + """, + (scope_type, scope_id), + ).fetchone() + pending = int(row["pending_count"] or 0) + blocked = int(row["blocked_count"] or 0) + return { + "state": ( + "blocked" if blocked else "pending" if pending else "synced" + ), + "pending_count": pending, + "blocked_count": blocked, + "last_error": row["last_error"], + "last_synced_at": row["last_synced_at"], + } + def has_legacy_memory_import( self, source_path: str, source_checksum: str ) -> bool: @@ -12448,6 +12510,7 @@ class SQLiteRunJournal: max_scopes: int = 4, batch_size: int = 100, lease_seconds: float = 30.0, + eligible_scopes: set[tuple[str, str]] | None = None, ) -> list[MemoryMutationSyncBatch]: if max_scopes < 1 or batch_size < 1 or lease_seconds <= 0: raise ValueError("Memory outbox claim limits must be positive") @@ -12480,11 +12543,21 @@ class SQLiteRunJournal: ) GROUP BY outbox.scope_type, outbox.scope_id ORDER BY head_scope_revision, outbox.scope_type, outbox.scope_id - LIMIT ? """, - (timestamp, max_scopes), + (timestamp,), ).fetchall() for candidate in candidates: + scope_key = ( + str(candidate["scope_type"]), + str(candidate["scope_id"]), + ) + if ( + eligible_scopes is not None + and scope_key not in eligible_scopes + ): + continue + if len(batches) >= max_scopes: + break state = connection.execute( """ SELECT * FROM memory_scope_state diff --git a/backend/app/run_sync/cloud_sync.py b/backend/app/run_sync/cloud_sync.py index 46a40f33c..b8e4b05f5 100644 --- a/backend/app/run_sync/cloud_sync.py +++ b/backend/app/run_sync/cloud_sync.py @@ -331,7 +331,12 @@ class HttpRunEventSyncTransport: configuration: CloudSyncConfiguration, payload: dict[str, Any], ) -> dict[str, Any]: - await self._ensure_device(configuration) + if str(payload.get("scope_type")) == "project": + await self._ensure_device_and_route( + configuration, str(payload["scope_id"]) + ) + else: + await self._ensure_device(configuration) return await self._json_request( "POST", f"{self._sync_base(configuration)}/memory/mutations:ingest", @@ -344,7 +349,12 @@ class HttpRunEventSyncTransport: configuration: CloudSyncConfiguration, payload: dict[str, Any], ) -> dict[str, Any]: - await self._ensure_device(configuration) + if str(payload.get("scope_type")) == "project": + await self._ensure_device_and_route( + configuration, str(payload["scope_id"]) + ) + else: + await self._ensure_device(configuration) return await self._json_request( "PUT", f"{self._sync_base(configuration)}/memory/snapshot", @@ -388,6 +398,7 @@ class CloudSyncWorker: self._bootstrap_attempt_count = 0 self._bootstrap_next_attempt_at = 0.0 self._memory_snapshot_revisions: dict[tuple[str, str], int] = {} + self._memory_snapshot_verified_at: dict[tuple[str, str], float] = {} def configure(self, configuration: CloudSyncConfiguration) -> None: if configuration != self._configuration: @@ -395,6 +406,7 @@ class CloudSyncWorker: self._bootstrap_attempt_count = 0 self._bootstrap_next_attempt_at = 0.0 self._memory_snapshot_revisions.clear() + self._memory_snapshot_verified_at.clear() self._configuration = configuration self.notify() @@ -429,23 +441,18 @@ class CloudSyncWorker: # Keep the flag set so the normal poll loop retries. logger.exception("Cloud Run history bootstrap failed") memory_count = 0 - memory_snapshot_ready = False + memory_snapshot_ready: set[tuple[str, str]] = set() if not self._bootstrap_pending: - try: + memory_snapshot_ready = ( await self._sync_memory_snapshots_if_changed(configuration) - memory_snapshot_ready = True - except asyncio.CancelledError: - raise - except Exception: - # Memory replication is independent of Run durability. A - # snapshot failure delays only the Memory lane. - logger.exception("Cloud Memory snapshot sync failed") + ) if memory_snapshot_ready: memory_batches = await asyncio.to_thread( self._journal.claim_ready_memory_mutation_batches, max_scopes=self._max_parallel_runs, batch_size=self._batch_size, lease_seconds=self._lease_seconds, + eligible_scopes=memory_snapshot_ready, ) if memory_batches: memory_results = await asyncio.gather( @@ -616,17 +623,24 @@ class CloudSyncWorker: async def _sync_memory_snapshots_if_changed( self, configuration: CloudSyncConfiguration, - ) -> None: + ) -> set[tuple[str, str]]: put_snapshot = getattr(self._transport, "put_memory_snapshot", None) if not callable(put_snapshot): - return + return set() snapshots = await asyncio.to_thread( self._journal.list_memory_sync_snapshots ) + ready: set[tuple[str, str]] = set() + now = time.monotonic() for snapshot in snapshots: key = (str(snapshot["scope_type"]), str(snapshot["scope_id"])) revision = int(snapshot["revision"]) - if self._memory_snapshot_revisions.get(key) == revision: + if ( + self._memory_snapshot_revisions.get(key) == revision + and now - self._memory_snapshot_verified_at.get(key, 0.0) + < 30.0 + ): + ready.add(key) continue payload = { "scope_type": key[0], @@ -635,17 +649,31 @@ class CloudSyncWorker: "source_revision": revision, "entries": snapshot["entries"], } - response = await put_snapshot(configuration, payload) - if ( - response.get("scope_type") != key[0] - or response.get("scope_id") != key[1] - or not isinstance(response.get("source_revision"), int) - or int(response["source_revision"]) < revision - ): - raise RunEventSyncProtocolError( - "Memory snapshot response does not acknowledge the source revision" + try: + response = await put_snapshot(configuration, payload) + if ( + response.get("scope_type") != key[0] + or response.get("scope_id") != key[1] + or not isinstance(response.get("source_revision"), int) + or int(response["source_revision"]) < revision + ): + raise RunEventSyncProtocolError( + "Memory snapshot response does not acknowledge the " + "source revision" + ) + except asyncio.CancelledError: + raise + except Exception: + # Scope isolation is deliberate: one malformed or stale scope + # must not stop unrelated Memory outboxes from draining. + logger.exception( + "Cloud Memory snapshot sync failed for %s/%s", *key ) + continue self._memory_snapshot_revisions[key] = revision + self._memory_snapshot_verified_at[key] = now + ready.add(key) + return ready @staticmethod def _timestamp(value: Any) -> float: diff --git a/backend/tests/app/agent/factory/test_mcp.py b/backend/tests/app/agent/factory/test_mcp.py index d650cb239..89d2667ee 100644 --- a/backend/tests/app/agent/factory/test_mcp.py +++ b/backend/tests/app/agent/factory/test_mcp.py @@ -82,6 +82,7 @@ async def test_mcp_agent_creation(sample_chat_data): "remember_project_memory", "update_project_memory", "forget_project_memory", + "promote_project_memory", "search_project_history", } assert call_args[0][1] == "MCP system prompt" diff --git a/backend/tests/app/lightweight_memory/test_lightweight_memory_service.py b/backend/tests/app/lightweight_memory/test_lightweight_memory_service.py index 2cdfbe5bf..3114e83ef 100644 --- a/backend/tests/app/lightweight_memory/test_lightweight_memory_service.py +++ b/backend/tests/app/lightweight_memory/test_lightweight_memory_service.py @@ -117,6 +117,79 @@ def test_user_confirmed_source_cannot_be_laundered_by_agent(service): ) +def test_agent_user_asserted_memory_requires_same_project_user_event(service): + journal = service.journal + journal.ensure_run(run_id="run-1", project_id="project-1") + journal.append_event( + "run-1", + RunEventDraft( + event_id="assistant-1", + event_type="assistant.final", + payload={"content": "The user said this."}, + ), + ) + + with pytest.raises(PermissionError, match="user.message"): + service.create_entry( + scope_type="project", + scope_id="project-1", + kind="fact", + content="The user said this.", + actor_type="agent", + reason="invalid provenance", + source_trust="user_asserted", + source_refs=("assistant-1",), + ) + + +def test_agent_cannot_delete_user_memory_or_keep_its_trust_on_rewrite(service): + user_entry = service.create_entry( + scope_type="project", + scope_id="project-1", + kind="fact", + content="User-owned fact.", + actor_type="user", + reason="user authored", + source_trust="user_confirmed", + request_id="user-memory", + ).entry + assert user_entry is not None and user_entry.confirmed_by_user is True + + with pytest.raises(PermissionError, match="unconfirmed Project"): + service.transition_entry( + memory_id=user_entry.memory_id, + expected_version=user_entry.version, + operation="remove", + actor_type="agent", + reason="agent tried to forget user Memory", + request_id="agent-delete-user-memory", + ) + + inferred = service.create_entry( + scope_type="project", + scope_id="project-1", + kind="fact", + content="Initial inference.", + actor_type="agent", + reason="initial model inference", + source_trust="model_inferred", + request_id="agent-memory", + ).entry + assert inferred is not None + rewritten = service.update_entry( + memory_id=inferred.memory_id, + expected_version=inferred.version, + content="Replacement model text.", + kind="fact", + actor_type="agent", + reason="rewrite", + request_id="agent-rewrite", + source_trust="user_asserted", + ).entry + assert rewritten is not None + assert rewritten.source_trust == "model_inferred" + + def test_search_memory_respects_total_budget_and_scope_specificity(service): for scope_type, scope_id, content in ( ("user", "user-1", "Use concise answers."), @@ -193,6 +266,24 @@ def test_incremental_maintainer_advances_cursor_and_is_idempotent(service): assert entries[0].source_trust == "user_asserted" +def test_incremental_maintainer_consumes_more_than_one_history_page(service): + journal = service.journal + journal.ensure_run(run_id="long-run", project_id="project-1") + for index in range(150): + journal.append_event( + "long-run", + RunEventDraft( + event_id=f"tool-{index}", + event_type="tool.completed", + payload={"content": f"observation {index}"}, + ), + ) + + state = IncrementalMemoryMaintainer(service).process_project("project-1") + + assert state.processed_through_watermark == "sqlite-project-v1:150" + + def test_incremental_maintainer_records_noop_before_advancing_cursor(service): journal = service.journal journal.ensure_run(run_id="run-1", project_id="project-1") diff --git a/backend/tests/app/run_journal/test_memory_v2.py b/backend/tests/app/run_journal/test_memory_v2.py index 4b9f176f8..55fa25634 100644 --- a/backend/tests/app/run_journal/test_memory_v2.py +++ b/backend/tests/app/run_journal/test_memory_v2.py @@ -419,6 +419,67 @@ def test_memory_outbox_fifo_uses_scope_revision_not_timestamp_or_id(journal): ] +def test_memory_outbox_claim_can_isolate_snapshot_verified_scopes(journal): + for scope_type, scope_id in ( + ("project", "project-1"), + ("user", "user-1"), + ): + journal.apply_memory_mutation( + mutation_id=f"mutation-{scope_type}", + idempotency_key=f"request-{scope_type}", + operation="add", + scope_type=scope_type, + scope_id=scope_id, + memory_id=f"memory-{scope_type}", + actor_type="user", + reason="Create", + content=scope_type, + kind="fact", + token_count=1, + created_by="user", + source_trust="user_confirmed", + now=1, + ) + + batches = journal.claim_ready_memory_mutation_batches( + now=2, + eligible_scopes={("user", "user-1")}, + ) + + assert [(batch.scope_type, batch.scope_id) for batch in batches] == [ + ("user", "user-1") + ] + + +def test_memory_sync_status_reports_pending_and_sent_truthfully(journal): + journal.apply_memory_mutation( + mutation_id="mutation-status", + idempotency_key="request-status", + operation="add", + scope_type="project", + scope_id="project-1", + memory_id="memory-status", + actor_type="user", + reason="Create", + content="Status", + kind="fact", + token_count=1, + created_by="user", + source_trust="user_confirmed", + now=1, + ) + assert journal.get_memory_sync_status("project", "project-1")["state"] == ( + "pending" + ) + batch = journal.claim_ready_memory_mutation_batches(now=2)[0] + journal.mark_memory_mutation_batch_sent(batch, now=3) + + status = journal.get_memory_sync_status("project", "project-1") + + assert status["state"] == "synced" + assert status["last_synced_at"] == 3 + + def test_memory_outbox_splits_across_scope_setting_revision_gap(journal): first = journal.apply_memory_mutation( mutation_id="mutation-add", diff --git a/backend/tests/app/run_sync/test_cloud_sync.py b/backend/tests/app/run_sync/test_cloud_sync.py index 1dab49db3..fb93c5167 100644 --- a/backend/tests/app/run_sync/test_cloud_sync.py +++ b/backend/tests/app/run_sync/test_cloud_sync.py @@ -43,6 +43,7 @@ class FakeTransport: self.project_events: dict[str, list[dict[str, Any]]] = {} self.memory_snapshots: list[dict[str, Any]] = [] self.memory_payloads: list[dict[str, Any]] = [] + self.memory_snapshot_failures: set[tuple[str, str]] = set() async def ingest(self, configuration, payload): self.payloads.append(payload) @@ -107,6 +108,10 @@ class FakeTransport: async def put_memory_snapshot(self, configuration, payload): self.memory_snapshots.append(payload) + if (payload["scope_type"], payload["scope_id"]) in ( + self.memory_snapshot_failures + ): + raise RuntimeError("malformed Memory scope") return { "scope_type": payload["scope_type"], "scope_id": payload["scope_id"], @@ -224,6 +229,41 @@ async def test_worker_syncs_full_memory_snapshot_and_independent_outbox( await worker.close() +@pytest.mark.asyncio +async def test_bad_memory_snapshot_does_not_block_an_unrelated_scope(journal): + for scope_type, scope_id, suffix in ( + ("project", "project-1", "project"), + ("user", "user-1", "user"), + ): + journal.apply_memory_mutation( + mutation_id=f"mutation-{suffix}", + idempotency_key=f"request-{suffix}", + operation="add", + scope_type=scope_type, + scope_id=scope_id, + memory_id=f"memory-{suffix}", + actor_type="user", + reason="Created in Memory Center", + content=f"{suffix} preference", + kind="preference", + token_count=3, + created_by="user", + source_trust="user_confirmed", + ) + transport = FakeTransport() + transport.memory_snapshot_failures.add(("project", "project-1")) + worker = _worker(journal, transport) + + assert await worker.drain_once() == 1 + + assert [item["scope_id"] for item in transport.memory_payloads] == [ + "user-1" + ] + remaining = journal.claim_ready_memory_mutation_batches(now=float("inf")) + assert [batch.scope_id for batch in remaining] == ["project-1"] + await worker.close() + + @pytest.mark.asyncio async def test_worker_redacts_approval_arguments_and_local_targets(journal): journal.ensure_run(run_id="run-approval", project_id="project-1", now=1) diff --git a/src/pages/Agents/Memory.test.tsx b/src/pages/Agents/Memory.test.tsx index 3902478ae..43ff568c2 100644 --- a/src/pages/Agents/Memory.test.tsx +++ b/src/pages/Agents/Memory.test.tsx @@ -78,7 +78,17 @@ const scopeState = { describe('Memory Center', () => { beforeEach(() => { vi.clearAllMocks(); - api.list.mockResolvedValue({ scope_state: scopeState, items: [] }); + api.list.mockResolvedValue({ + scope_state: scopeState, + items: [], + sync_status: { + state: 'synced', + pending_count: 0, + blocked_count: 0, + last_error: null, + last_synced_at: 1, + }, + }); api.create.mockResolvedValue({}); api.consolidate.mockResolvedValue({}); }); @@ -94,6 +104,58 @@ describe('Memory Center', () => { expect(api.consolidate).toHaveBeenCalledWith('project', 'project-1'); }); + it('ignores a stale scope response after the user switches scope', async () => { + let resolveProject!: (value: unknown) => void; + api.list.mockImplementation((scopeType: string) => { + if (scopeType === 'project') { + return new Promise((resolve) => { + resolveProject = resolve; + }); + } + return Promise.resolve({ + scope_state: { + ...scopeState, + scope_type: 'space', + scope_id: 'space-1', + }, + items: [ + { + memory_id: 'space-memory', + scope_type: 'space', + scope_id: 'space-1', + kind: 'fact', + content: 'Space response', + priority: 'normal', + version: 1, + token_count: 2, + pinned_by_user: false, + confirmed_by_user: true, + created_by: 'user', + source_trust: 'user_confirmed', + sensitivity: 'normal', + source_refs: [], + deleted_at: null, + created_at: 1, + updated_at: 1, + }, + ], + }); + }); + const user = userEvent.setup(); + render(); + + await user.click(screen.getByRole('button', { name: 'Space' })); + expect(await screen.findByText('Space response')).toBeInTheDocument(); + resolveProject({ + scope_state: scopeState, + items: [{ content: 'Stale project response' }], + }); + + await waitFor(() => + expect(screen.queryByText('Stale project response')).toBeNull() + ); + }); + it('explains the History boundary and creates editable Memory', async () => { const user = userEvent.setup(); render(); diff --git a/src/pages/Agents/Memory.tsx b/src/pages/Agents/Memory.tsx index 0d82657e5..e49077e60 100644 --- a/src/pages/Agents/Memory.tsx +++ b/src/pages/Agents/Memory.tsx @@ -53,7 +53,7 @@ import { Save, Trash2, } from 'lucide-react'; -import { useCallback, useEffect, useMemo, useState } from 'react'; +import { useCallback, useEffect, useMemo, useRef, useState } from 'react'; const KINDS: MemoryKind[] = [ 'fact', @@ -88,6 +88,11 @@ export default function Memory() { const [editingId, setEditingId] = useState(null); const [editingText, setEditingText] = useState(''); const [showArchived, setShowArchived] = useState(false); + const [search, setSearch] = useState(''); + const [syncStatus, setSyncStatus] = useState< + 'synced' | 'pending' | 'blocked' | 'unknown' + >('unknown'); + const requestGeneration = useRef(0); const scopeIds = useMemo( () => ({ @@ -100,9 +105,11 @@ export default function Memory() { const scopeId = scopeIds[scopeType]; const reload = useCallback(async () => { + const generation = ++requestGeneration.current; if (!scopeId) { setEntries([]); setScopeState(null); + setSyncStatus('unknown'); return; } setLoading(true); @@ -113,12 +120,15 @@ export default function Memory() { scopeId, showArchived ); + if (generation !== requestGeneration.current) return; setEntries(response.items); setScopeState(response.scope_state); + setSyncStatus(response.sync_status?.state ?? 'unknown'); } catch (caught) { + if (generation !== requestGeneration.current) return; setError(caught instanceof Error ? caught.message : String(caught)); } finally { - setLoading(false); + if (generation === requestGeneration.current) setLoading(false); } }, [scopeId, scopeType, showArchived]); @@ -157,6 +167,16 @@ export default function Memory() { ) ) : 0; + const visibleEntries = useMemo(() => { + const needle = search.trim().toLocaleLowerCase(); + if (!needle) return entries; + return entries.filter( + (entry) => + entry.content.toLocaleLowerCase().includes(needle) || + entry.kind.includes(needle) || + TRUST_LABELS[entry.source_trust].toLocaleLowerCase().includes(needle) + ); + }, [entries, search]); return (
@@ -225,7 +245,13 @@ export default function Memory() {
Memory Sync - Synced to your Eigent account for use across devices + {syncStatus === 'synced' + ? 'Synced to your Eigent account' + : syncStatus === 'pending' + ? 'Waiting to sync automatically' + : syncStatus === 'blocked' + ? 'Sync needs attention; local Memory is safe' + : 'Sync status is not available yet'}
@@ -305,6 +331,13 @@ export default function Memory() { Show archived + setSearch(event.target.value)} + placeholder="Search Memory" + aria-label="Search Memory" + /> {error && (
{error}
)} @@ -312,14 +345,14 @@ export default function Memory() {
Loading Memory…
- ) : entries.length === 0 ? ( + ) : visibleEntries.length === 0 ? (
No saved Memory for this {scopeType}. That is okay—History remains available to the Agent.
) : (
- {entries.map((entry) => ( + {visibleEntries.map((entry) => (
📌 Pinned} {entry.deleted_at && Archived}
+
+ Source and provenance +
+ Created by {entry.created_by}; trust:{' '} + {TRUST_LABELS[entry.source_trust]} +
+ {entry.source_refs.length > 0 ? ( +
+ Sources: {entry.source_refs.join(', ')} +
+ ) : null} +
{editingId === entry.memory_id ? (