vrr/eigent
2
0
Fork 0
mirror of https://github.com/eigent-ai/eigent.git synced 2026-08-25 08:32:03 +00:00
Code Issues Projects Releases Packages Wiki Activity Actions 3

fix: harden memory v2 lifecycle and sync

Browse source
This commit is contained in:
4pmtong 2026-08-14 00:15:31 +08:00
parent 8e2e442d3a
commit 923cdd3e4a
14 changed files with 841 additions and 133 deletions
Download patch file Download diff file Expand all files Collapse all files

256
backend/app/agent/toolkit/memory_toolkit.py
View file

@ -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"

3
backend/app/controller/memory_controller.py
View file

@ -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
),
}

184
backend/app/lightweight_memory/maintainer.py
View file

@ -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",

56
backend/app/lightweight_memory/service.py
View file

@ -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"):

79
backend/app/run_journal/store.py
View file

@ -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

74
backend/app/run_sync/cloud_sync.py
View file

@ -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:

1
backend/tests/app/agent/factory/test_mcp.py
View file

@ -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"

91
backend/tests/app/lightweight_memory/test_lightweight_memory_service.py
View file

@ -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")

61
backend/tests/app/run_journal/test_memory_v2.py
View file

@ -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",

40
backend/tests/app/run_sync/test_cloud_sync.py
View file

@ -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)

64
src/pages/Agents/Memory.test.tsx
View file

@ -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(<Memory />);
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(<Memory />);

55
src/pages/Agents/Memory.tsx
View file

@ -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<string | null>(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 (
<div className="flex w-full flex-1 flex-col gap-5 px-6 pb-12 pt-8">
@ -225,7 +245,13 @@ export default function Memory() {
<div className="text-body-sm">
<strong className="block">Memory Sync</strong>
<span className="text-xs text-ds-text-neutral-muted-default">
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'}
</span>
</div>
</div>
@ -305,6 +331,13 @@ export default function Memory() {
Show archived
</label>
</div>
<Input
className="mb-3"
value={search}
onChange={(event) => setSearch(event.target.value)}
placeholder="Search Memory"
aria-label="Search Memory"
/>
{error && (
<div className="mb-3 text-body-sm text-red-600">{error}</div>
)}
@ -312,14 +345,14 @@ export default function Memory() {
<div className="py-8 text-center text-body-sm">
Loading Memory…
</div>
) : entries.length === 0 ? (
) : visibleEntries.length === 0 ? (
<div className="py-8 text-center text-body-sm text-ds-text-neutral-muted-default">
No saved Memory for this {scopeType}. That is okay—History
remains available to the Agent.
</div>
) : (
<div className="flex flex-col gap-3">
{entries.map((entry) => (
{visibleEntries.map((entry) => (
<article
key={entry.memory_id}
className={`border-ds-border-neutral-default rounded-xl border p-4 ${entry.deleted_at ? 'opacity-70' : ''}`}
@ -335,6 +368,18 @@ export default function Memory() {
{entry.pinned_by_user && <span>📌 Pinned</span>}
{entry.deleted_at && <span>Archived</span>}
</div>
<details className="mb-2 text-xs text-ds-text-neutral-muted-default">
<summary>Source and provenance</summary>
<div className="mt-1">
Created by {entry.created_by}; trust:{' '}
{TRUST_LABELS[entry.source_trust]}
</div>
{entry.source_refs.length > 0 ? (
<div className="mt-1 break-all">
Sources: {entry.source_refs.join(', ')}
</div>
) : null}
</details>
{editingId === entry.memory_id ? (
<Textarea
value={editingText}

3
src/service/humanInteractionApi.ts
View file

@ -26,7 +26,8 @@ export interface HumanInteractionPayload {
| 'approval'
| 'diff_review'
| 'merge_conflict'
| 'credential_binding';
| 'credential_binding'
| 'memory_change_review';
run_id?: string;
version?: number;
approval_id?: string;

7
src/service/memoryApi.ts
View file

@ -68,6 +68,13 @@ export interface MemoryEntry {
export interface MemoryListResponse {
scope_state: MemoryScopeState;
items: MemoryEntry[];
sync_status?: {
state: 'synced' | 'pending' | 'blocked';
pending_count: number;
blocked_count: number;
last_error: string | null;
last_synced_at: number | null;
};
}
const scopePath = (scopeType: MemoryScopeType, scopeId: string) =>